feat(terminal): stream negotiated ANSI replay to clients

Restore configured client history up to host retention without collecting capable replay in the daemon. Preserve single-frame ANSI delivery for older clients and normalize older hosts at the SDK boundary.
2026-10-31 05:32:58 +00:00 · 2026-10-30 09:56:54 +01:00
parent 9ba63a30fa
commit bfe2b10cef
44 changed files with 787 additions and 499 deletions
@@ -7,7 +7,6 @@ import { Terminal as HeadlessTerminal, type IBufferCell, type IBufferLine } from
 import { expect, test } from "vitest";

 import type { TerminalCell, TerminalState } from "@getpaseo/protocol/messages";
-import { renderTerminalSnapshotToAnsi } from "@getpaseo/protocol/terminal-snapshot";
 import type { TerminalStreamEvent } from "@getpaseo/client/internal/daemon-client";
 import { DaemonClient } from "../test-utils/daemon-client.js";
 import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
@@ -46,7 +45,7 @@ test("byte-stream headless terminal matches daemon state after high-output attac
      terminalId: terminal.id,
      headless: liveHeadless,
    });
-    const initialSnapshot = waitForTerminalStreamEvent(client, terminal.id, "snapshot");
+    const initialSnapshot = waitForReplayEnd(client, terminal.id);

    const liveSubscription = await subscribeTerminal(client, terminal.id);
    await initialSnapshot;
@@ -60,13 +59,13 @@ test("byte-stream headless terminal matches daemon state after high-output attac
      text: BYTE_DONE_MARKER,
      rows: BYTE_TEST_SIZE.rows,
    });
-    const daemonSnapshot = await readTerminalSnapshot({
+    const daemonSnapshot = await readReplayViewport({
      port: daemon.port,
      terminalId: terminal.id,
    });
    await liveHeadless.flush();

-    expect(liveHeadless.visibleLines()).toEqual(visibleTerminalLines(daemonSnapshot));
+    expect(liveHeadless.visibleLines()).toEqual(daemonSnapshot.lines);
    expect(liveHeadless.cursor()).toEqual(daemonSnapshot.cursor);
    expect(liveHeadless.visibleLines()).toEqual(capturedViewport.lines);

@@ -79,7 +78,7 @@ test("byte-stream headless terminal matches daemon state after high-output attac
      terminalId: terminal.id,
      headless: restoreHeadless,
    });
-    const restoreFrame = waitForTerminalStreamEvent(client, terminal.id, "restore");
+    const restoreFrame = waitForReplayEnd(client, terminal.id);

    await subscribeTerminal(client, terminal.id, {
      restore: { mode: "visible-snapshot", scrollbackLines: 200, size: BYTE_TEST_SIZE },
@@ -116,12 +115,10 @@ class ClientHeadlessTerminal {
  }

  applyEvent(event: TerminalStreamEvent): void {
-    if (event.type === "snapshot") {
-      this.enqueueText(renderTerminalSnapshotToAnsi(event.state), { resetDecoder: true });
-      return;
-    }
-    if (event.type === "restore") {
-      this.enqueueText(this.decoder.decode(event.data, { stream: false }), { resetDecoder: true });
+    if (event.type === "replay") {
+      if (event.replay.type === "begin") this.enqueueText("\x1bc", { resetDecoder: true });
+      if (event.replay.type === "data")
+        this.enqueueText(this.decoder.decode(event.replay.data, { stream: true }));
      return;
    }
    this.enqueueText(this.decoder.decode(event.data, { stream: true }));
@@ -187,35 +184,32 @@ async function subscribeTerminal(
  return response.subscription;
 }

-async function readTerminalSnapshot(input: {
-  port: number;
-  terminalId: string;
-}): Promise<TerminalState> {
-  const client = new DaemonClient({
-    url: `ws://127.0.0.1:${input.port}/ws`,
-    appVersion: "0.1.96",
-    clientType: "mobile",
-  });
+async function readReplayViewport(input: { port: number; terminalId: string }): Promise<{
+  lines: string[];
+  cursor: TerminalState["cursor"];
+}> {
+  const client = new DaemonClient({ url: `ws://127.0.0.1:${input.port}/ws`, clientType: "mobile" });
+  const headless = new ClientHeadlessTerminal(BYTE_TEST_SIZE);
+  const stop = pipeTerminalStreamToHeadless({ client, terminalId: input.terminalId, headless });
  try {
    await client.connect();
-    const snapshot = waitForTerminalStreamEvent(client, input.terminalId, "snapshot");
+    const completed = waitForReplayEnd(client, input.terminalId);
    await subscribeTerminal(client, input.terminalId);
-    const event = await snapshot;
-    return event.state;
+    await completed;
+    await headless.flush();
+    return { lines: headless.visibleLines(), cursor: headless.cursor() };
  } finally {
+    stop();
    await client.close();
  }
 }

-function waitForTerminalStreamEvent<TType extends TerminalStreamEvent["type"]>(
-  client: DaemonClient,
-  terminalId: string,
-  type: TType,
-): Promise<Extract<TerminalStreamEvent, { type: TType }>> {
-  return client.waitForTerminalStreamEvent(
-    (event) => event.terminalId === terminalId && event.type === type,
-    10_000,
-  ) as Promise<Extract<TerminalStreamEvent, { type: TType }>>;
+async function waitForReplayEnd(client: DaemonClient, terminalId: string): Promise<void> {
+  await client.waitForTerminalStreamEvent(
+    (event) =>
+      event.terminalId === terminalId && event.type === "replay" && event.replay.type === "end",
+    10000,
+  );
 }

 async function createPackageLockTerminalCwd(): Promise<PackageLockTerminalCwd> {
@@ -287,24 +281,16 @@ async function waitForCondition<T>(
  throw new Error(`Timed out after ${timeoutMs}ms waiting for terminal byte parity`);
 }

-function visibleTerminalLines(state: TerminalState): string[] {
-  return state.grid.map(terminalRowText);
-}
-
-function terminalRowText(row: TerminalState["grid"][number]): string {
-  return row
-    .map((cell) => cell.char)
-    .join("")
-    .trimEnd();
-}
-
 function bufferLineText(line: IBufferLine | undefined, reusableCell: IBufferCell): string {
  const cells: TerminalCell[] = [];
  for (let col = 0; col < BYTE_TEST_SIZE.cols; col += 1) {
    const cell = line?.getCell(col, reusableCell);
    cells.push({ char: cell?.getChars() || " " });
  }
-  return terminalRowText(cells);
+  return cells
+    .map((cell) => cell.char)
+    .join("")
+    .trimEnd();
 }

 function packageLockStreamScript(): string {
@@ -1,3 +1,4 @@
+import { Terminal as HeadlessTerminal } from "@xterm/headless";
 import { afterEach, beforeEach, describe, expect, test } from "vitest";
 import { mkdtempSync, rmSync } from "fs";
 import { tmpdir } from "os";
@@ -80,6 +81,24 @@ async function waitForCondition(
  throw new Error(`Timed out after ${timeoutMs}ms waiting for condition`);
 }

+function readHeadlessState(terminal: HeadlessTerminal): TerminalState {
+  const buffer = terminal.buffer.active;
+  const lines: TerminalState["grid"] = [];
+  for (let row = 0; row < buffer.length; row++) {
+    const cells: TerminalState["grid"][number] = [];
+    for (let col = 0; col < terminal.cols; col++)
+      cells.push({ char: buffer.getLine(row)?.getCell(col)?.getChars() ?? "" });
+    lines.push(cells);
+  }
+  return {
+    rows: terminal.rows,
+    cols: terminal.cols,
+    grid: lines.slice(buffer.baseY),
+    scrollback: lines.slice(0, buffer.baseY),
+    cursor: { row: buffer.cursorY, col: buffer.cursorX },
+  };
+}
+
 async function waitForTerminalSnapshot(
  client: DaemonClient,
  terminalId: string,
@@ -92,16 +111,34 @@ async function waitForTerminalSnapshot(
      reject(new Error(`Timeout waiting for terminal snapshot (${timeout}ms)`));
    }, timeout);

+    let terminal: HeadlessTerminal | null = null;
+    let writes = Promise.resolve();
    const unsubscribe = client.onTerminalStreamEvent((event) => {
-      if (event.terminalId !== terminalId || event.type !== "snapshot") {
-        return;
+      if (event.terminalId !== terminalId || event.type !== "replay") return;
+      const replay = event.replay;
+      if (replay.type === "begin") {
+        terminal?.dispose();
+        terminal = new HeadlessTerminal({
+          rows: replay.rows ?? 24,
+          cols: replay.cols ?? 80,
+          scrollback: 1000,
+          allowProposedApi: true,
+        });
+      } else if (replay.type === "data") {
+        const active = terminal!;
+        writes = writes.then(() => new Promise<void>((done) => active.write(replay.data, done)));
+      } else {
+        void writes.then(() => {
+          const active = terminal!;
+          const state = readHeadlessState(active);
+          if (!predicate(state)) return null;
+          clearTimeout(timeoutHandle);
+          unsubscribe();
+          active.dispose();
+          resolve(state);
+          return null;
+        });
      }
-      if (!predicate(event.state)) {
-        return;
-      }
-      clearTimeout(timeoutHandle);
-      unsubscribe();
-      resolve(event.state);
    });
  });
 }
@@ -1836,3 +1873,68 @@ describe("list terminals across directories", () => {
    ]);
  });
 });
+
+test.each([400, 3000])(
+  "chunked SDK replay retains requested history for %s printed lines",
+  async (count) => {
+    const cwd = tmpCwd();
+    tempDirs.push(cwd);
+    const created = await createTerminalInWorkspace(ctx.client, {
+      cwd,
+      options: {
+        command: "/bin/sh",
+        args: [
+          "-lc",
+          `i=1; while [ $i -le ${count} ]; do printf '\\033[31mreplay-%04d界🙂 payload-abcdefghijklmnopqrstuvwxyz0123456789\\033[0m\\n' $i; i=$((i+1)); done; sleep 30`,
+        ],
+      },
+    });
+    const id = created.terminal!.id;
+    await waitForCondition(
+      async () =>
+        (await ctx.client.captureTerminal(id)).lines
+          .join("\n")
+          .includes(`replay-${String(count).padStart(4, "0")}`),
+      10000,
+    );
+    const events: import("@getpaseo/client").TerminalStreamEvent[] = [];
+    let end!: () => void;
+    const completed = new Promise<void>((resolve) => {
+      end = resolve;
+    });
+    const subscription = ctx.client.observeTerminal(
+      id,
+      (event) => {
+        events.push(event);
+        if (event.type === "replay" && event.replay.type === "end") end();
+      },
+      { restore: { mode: "visible-snapshot", scrollbackLines: 10000 } },
+    );
+    try {
+      await subscription.ready;
+      await completed;
+      const replay = events.flatMap((event) => (event.type === "replay" ? [event.replay] : []));
+      expect(replay[0]).toMatchObject({
+        type: "begin",
+        retainedHistoryLines: count === 3000 ? 1000 : expect.any(Number),
+      });
+      expect(replay.at(-1)).toEqual({ type: "end" });
+      const chunks = replay.flatMap((event) => (event.type === "data" ? [event.data] : []));
+      expect(chunks.every((chunk) => chunk.length <= 65536)).toBe(true);
+      const text = Buffer.concat(chunks).toString("utf8");
+      const begin = replay[0];
+      if (begin?.type !== "begin" || begin.rows === undefined)
+        throw new Error("Missing replay dimensions");
+      expect(text.match(/replay-\d{4}/g)).toHaveLength(Math.min(count, 1000 + begin.rows - 1));
+      expect(text).toContain(`replay-${String(count).padStart(4, "0")}界🙂`);
+      if (count === 400) expect(text).toContain("replay-0001界🙂");
+      else {
+        expect(text).not.toContain("replay-0001");
+        expect(chunks.length).toBeGreaterThan(1);
+      }
+    } finally {
+      await subscription.release();
+    }
+  },
+  30000,
+);
@@ -1129,6 +1129,8 @@ export class Session {
      terminalManager,
      emit: (msg) => this.emit(msg),
      hasBinaryChannel: () => this.onBinaryMessage !== null,
+      supportsChunkedReplay: (source) =>
+        this.supportsForSource(CLIENT_CAPS.terminalChunkedReplay, source),
      isPathWithinRoot: (rootPath, candidatePath) => this.isPathWithinRoot(rootPath, candidatePath),
      sessionLogger: this.sessionLogger,
      listTerminalWorkspaceRefs: () => this.listActiveWorkspaceRefs(),
@@ -2016,6 +2016,7 @@ export class VoiceAssistantWebSocketServer {
        resumeOnStartupSetting: true,
        // COMPAT(terminalRestoreModes): added in v0.1.81, remove gate after 2026-11-23.
        "terminal-restore-modes": true,
+        terminalChunkedReplay: true,
        // COMPAT(terminalInputModeReplay): added in v0.2.6, remove gate after 2027-02-02.
        "terminal-input-mode-replay": true,
        // COMPAT(terminalSizeOwnership): added in v0.2.6, remove gate after 2027-02-02.
@@ -21,7 +21,7 @@ export type TerminalReplayEvent =
  | { type: "begin"; metadata: TerminalReplayMetadata }
  | { type: "data"; data: Uint8Array }
  | { type: "end" };
-export type TerminalReplaySink = (event: TerminalReplayEvent) => void;
+export type TerminalReplaySink = (event: TerminalReplayEvent) => void | Promise<void>;
 export const TERMINAL_REPLAY_CHUNK_BYTES = 64 * 1024;

 export function* createTerminalReplay(
@@ -45,15 +45,15 @@ export function* createTerminalReplay(
  return metadata;
 }

-export function replayTerminal(
+export async function replayTerminal(
  session: LocalTerminalSession,
  options: TerminalReplayOptions | undefined,
  sink: TerminalReplaySink,
-): TerminalReplayMetadata {
+): Promise<TerminalReplayMetadata> {
  const replay = createTerminalReplay(session, options);
  let next = replay.next();
  while (!next.done) {
-    sink(next.value);
+    await sink(next.value);
    next = replay.next();
  }
  return next.value;
@@ -67,6 +67,7 @@ interface TerminalReplayPorts {
 /** Owns the worker's replay/live cut, bounded live tail and controller lifetime. */
 export function createTerminalReplayController(ports: TerminalReplayPorts) {
  let controller: FramedSocket | null = null;
+  const acknowledgements = new Map<string, { resolve(): void; reject(error: Error): void }>();
  // Only the live tail of a replaying terminal is held. Replay chunks use the socket's
  // write completion directly; they never enter this buffer or fill the shared socket.
  interface ReplayTail {
@@ -142,11 +143,18 @@ export function createTerminalReplayController(ports: TerminalReplayPorts) {
      while (!next.done) {
        if (tail.overflow)
          throw new Error("Terminal output exceeded the replay buffer; attach again");
+        const acknowledged = new Promise<void>((resolve, reject) => {
+          acknowledgements.set(message.requestId, { resolve, reject });
+        });
+        void acknowledged.catch(() => {});
+        // Register before sending: even a synchronous controller cannot race the acknowledgement.
        await connection.sendAndWait({
          type: "terminalReplay",
          requestId: message.requestId,
          event: next.value,
        });
+        await acknowledged;
+        acknowledgements.delete(message.requestId);
        next = replay.next();
      }
      released = true;
@@ -161,6 +169,8 @@ export function createTerminalReplayController(ports: TerminalReplayPorts) {
      tail.failure = error;
      if (!released) await releaseReplayTail(session, tail);
      throw error;
+    } finally {
+      acknowledgements.delete(message.requestId);
    }
  }

@@ -230,6 +240,11 @@ export function createTerminalReplayController(ports: TerminalReplayPorts) {
  }

  return {
+    acknowledge(requestId: string, error?: string): void {
+      const pending = acknowledgements.get(requestId);
+      if (error) pending?.reject(new Error(error));
+      else pending?.resolve();
+    },
    setConnection(connection: FramedSocket | null): void {
      const previous = controller;
      controller = connection;
@@ -237,6 +252,9 @@ export function createTerminalReplayController(ports: TerminalReplayPorts) {
      // Old writes settle on their original socket; new output and replay belong
      // exclusively to the replacement controller.
      replayTails.clear();
+      for (const pending of acknowledgements.values())
+        pending.reject(new Error("Terminal replay controller disconnected"));
+      acknowledgements.clear();
    },
    send: sendToController,
    async replay(
@@ -3,7 +3,7 @@ import {
  TerminalStreamOpcode,
  encodeTerminalStreamFrame,
 } from "@getpaseo/protocol/binary-frames/index";
-import type { TerminalReplayOptions } from "./terminal-replay.js";
+import type { TerminalReplaySink, TerminalReplayOptions } from "./terminal-replay.js";

 export const MAX_TERMINAL_OUTPUT_FRAME_BYTES = 256 * 1024;

@@ -64,3 +64,37 @@ function resolveVisibleRestoreScrollbackLines(value: number | undefined): number
  }
  return Math.min(Math.max(0, value), MAX_VISIBLE_RESTORE_SCROLLBACK_LINES);
 }
+
+/** The daemon's only old-app adapter. Capable delivery never assembles the replay. */
+export function createTerminalRestoreDelivery(input: {
+  slot: number;
+  chunked: boolean;
+  send(frame: Uint8Array): Promise<void>;
+}): TerminalReplaySink {
+  // COMPAT(terminalSingleRestore): added after v0.14.0; remove after 2027-04-30 when client floor supports chunked replay.
+  const legacyChunks: Uint8Array[] = [];
+  return async (event) => {
+    if (!input.chunked) {
+      if (event.type === "data") legacyChunks.push(event.data);
+      if (event.type === "end")
+        await input.send(
+          encodeTerminalRestoreFrame({
+            slot: input.slot,
+            data: Buffer.concat(legacyChunks),
+          }),
+        );
+      return;
+    }
+    const opcodes = {
+      begin: TerminalStreamOpcode.ReplayBegin,
+      data: TerminalStreamOpcode.ReplayData,
+      end: TerminalStreamOpcode.ReplayEnd,
+    };
+    let payload: string | Uint8Array | undefined;
+    if (event.type === "begin") payload = JSON.stringify(event.metadata);
+    if (event.type === "data") payload = event.data;
+    await input.send(
+      encodeTerminalStreamFrame({ slot: input.slot, opcode: opcodes[event.type], payload }),
+    );
+  };
+}
@@ -35,9 +35,9 @@ function replayFixture(
  return async (_id, _options, sink) => {
    const replay = await read();
    if (!replay) return null;
-    sink({ type: "begin", metadata: replay.metadata });
-    sink({ type: "data", data: replay.data });
-    sink({ type: "end" });
+    await sink({ type: "begin", metadata: replay.metadata });
+    await sink({ type: "data", data: replay.data });
+    await sink({ type: "end" });
    return replay.metadata;
  };
 }
@@ -28,7 +28,7 @@ import { TerminalOutputCoalescer } from "./terminal-output-coalescer.js";
 import {
  MAX_CLIENT_BUFFERED_BYTES,
  MAX_TERMINAL_OUTPUT_FRAME_BYTES,
-  encodeTerminalRestoreFrame,
+  createTerminalRestoreDelivery,
  resolveRestoreAfterOutputOverflow,
  resolveTerminalRestoreSnapshotOptions,
  resolveTerminalSubscriptionSnapshotMode,
@@ -70,6 +70,7 @@ interface ActiveTerminalStream {
  readyRevision?: number;
  restore?: TerminalRestoreOptions;
  bufferedOutputs: BufferedTerminalOutput[];
+  bufferedOutputBytes: number;
  outputBytesSinceSnapshot: number;
  outputCoalescer: TerminalOutputCoalescer;
 }
@@ -83,6 +84,7 @@ export interface TerminalSessionControllerOptions {
  terminalManager: TerminalManager | null;
  emit: (msg: SessionOutboundMessage) => void;
  hasBinaryChannel: () => boolean;
+  supportsChunkedReplay?: (source: object) => boolean;
  isPathWithinRoot: (rootPath: string, candidatePath: string) => boolean;
  sessionLogger: pino.Logger;
  listTerminalWorkspaceRefs?: () => Promise<readonly TerminalWorkspaceRef[]>;
@@ -146,7 +148,7 @@ export class TerminalSessionController {
  private readonly activeStreams = new Map<number, ActiveTerminalStream>();
  private nextSlot = 0;

-  constructor(options: TerminalSessionControllerOptions) {
+  constructor(private readonly options: TerminalSessionControllerOptions) {
    this.terminalManager = options.terminalManager;
    this.emit = options.emit;
    this.hasBinaryChannel = options.hasBinaryChannel;
@@ -899,6 +901,7 @@ export class TerminalSessionController {
      readyRevision: undefined,
      restore: options?.restore,
      bufferedOutputs: [],
+      bufferedOutputBytes: 0,
      outputBytesSinceSnapshot: 0,
      outputCoalescer: new TerminalOutputCoalescer({
        timers: { setTimeout, clearTimeout },
@@ -971,6 +974,19 @@ export class TerminalSessionController {
          return;
        }
        if (activeStream.needsSnapshot || activeStream.snapshotInFlight) {
+          activeStream.bufferedOutputBytes += Buffer.byteLength(message.data);
+          if (activeStream.bufferedOutputBytes > MAX_CLIENT_BUFFERED_BYTES) {
+            activeStream.owner.emit({
+              type: "terminal_stream_exit",
+              payload: {
+                terminalId: activeStream.terminalId,
+                error: "Terminal replay live tail exceeded the buffer; attach again",
+              },
+            });
+            this.detachRegistration(activeStream);
+            void activeStream.owner.release();
+            return;
+          }
          activeStream.bufferedOutputs.push({
            data: message.data,
            revision: message.revision,
@@ -1075,19 +1091,45 @@ export class TerminalSessionController {
    terminalManager: TerminalManager,
    restore: TerminalRestoreOptions,
  ): Promise<SnapshotSendResult> {
-    const snapshotOptions = resolveTerminalRestoreSnapshotOptions(restore);
+    const chunked = this.options.supportsChunkedReplay?.(activeStream.owner.source) === true;
+    const snapshotOptions =
+      chunked && restore.mode !== "live"
+        ? { scrollbackLines: restore.scrollbackLines }
+        : resolveTerminalRestoreSnapshotOptions(restore);
    if (snapshotOptions === null) {
      return { shouldContinue: true };
    }

-    const chunks: Uint8Array[] = [];
-    // COMPAT(terminalSingleRestore): ANSI clients through v0.14 collect one restore;
-    // remove after the supported client floor implements chunked replay.
+    const deliver = createTerminalRestoreDelivery({
+      slot: activeStream.slot,
+      chunked,
+      send: (frame) => activeStream.owner.emitBinaryAndWait(frame),
+    });
    const snapshot = await terminalManager.replayTerminal(
      activeStream.terminalId,
      snapshotOptions,
-      (event) => {
-        if (event.type === "data") chunks.push(event.data);
+      async (event) => {
+        if (this.activeStreams.get(activeStream.slot) !== activeStream) return;
+        const buffered = this.options.getClientBufferedAmount?.(activeStream.owner.source);
+        if (
+          chunked &&
+          buffered !== null &&
+          buffered !== undefined &&
+          buffered > MAX_CLIENT_BUFFERED_BYTES
+        ) {
+          activeStream.owner.emit({
+            type: "terminal_stream_exit",
+            payload: {
+              terminalId: activeStream.terminalId,
+              error: "Terminal replay transport overflow; attach again",
+            },
+          });
+          this.detachRegistration(activeStream);
+          void activeStream.owner.release();
+
+          return;
+        }
+        await deliver(event);
      },
    );
    if (this.activeStreams.get(activeStream.slot) !== activeStream) {
@@ -1098,12 +1140,6 @@ export class TerminalSessionController {
      return { shouldContinue: false };
    }

-    activeStream.owner.emitBinary(
-      encodeTerminalRestoreFrame({
-        slot: activeStream.slot,
-        data: Buffer.concat(chunks),
-      }),
-    );
    // The restore frame went out-of-band; keep the replay that follows on the
    // coalescer's trailing path so it doesn't flush back-to-back with it.
    activeStream.outputCoalescer.markFlushed();
@@ -1120,6 +1156,7 @@ export class TerminalSessionController {
      activeStream.outputCoalescer.handle(replayPreamble);
    }

+    activeStream.bufferedOutputBytes = 0;
    const bufferedOutputs = activeStream.bufferedOutputs.splice(
      0,
      activeStream.bufferedOutputs.length,
@@ -552,6 +552,11 @@ function runWorker(argv: string[]): void {
        return;
      }

+      case "replayAcknowledged": {
+        replayController.acknowledge(message.requestId, message.error);
+        return;
+      }
+
      case "replayTerminal": {
        await replayController.replay(message);
        return;
@@ -4,6 +4,7 @@ import { classifyTerminalWorkerProtocol } from "./terminal-worker-protocol.js";
 describe("classifyTerminalWorkerProtocol", () => {
  it("rejects the unreleased cell protocol without a draining adapter", () => {
    expect(classifyTerminalWorkerProtocol(1, 2)).toBe("obsolete");
+    expect(classifyTerminalWorkerProtocol(2, 3)).toBe("obsolete");
  });
  it.each([
    [5, "current"],
@@ -43,7 +43,7 @@ export interface WorkerCreateTerminalOptions {
 // `shutdown`. A daemon drives a worker of its own version, drains one a version older (no new
 // terminals; it exits after its last terminal), shuts down anything older, and leaves a newer
 // worker running untouched. See docs/terminal-worker.md.
-export const TERMINAL_WORKER_PROTOCOL_VERSION = 2;
+export const TERMINAL_WORKER_PROTOCOL_VERSION = 3;

 // Test-only: the protocol version a worker reports, so tests can stand up older and newer
 // workers from one build.
@@ -59,8 +59,8 @@ export function classifyTerminalWorkerProtocol(
  // A version this daemon cannot read comes from a Paseo it does not know: treat it as newer.
  if (typeof reported !== "number" || !Number.isSafeInteger(reported)) return "newer";
  if (reported === current) return "current";
-  // Protocol 1 was unreleased and transported cells; it has no draining adapter.
-  if (reported === current - 1 && reported >= 2) return "previous";
+  // Protocols 1 and 2 were unreleased; they have no draining adapter.
+  if (reported === current - 1 && reported >= 3) return "previous";
  return reported < current ? "obsolete" : "newer";
 }

@@ -122,6 +122,11 @@ export type TerminalWorkerRequest =
      terminalId: string;
      options?: WorkerKillAndWaitOptions;
    }
+  | {
+      type: "replayAcknowledged";
+      requestId: string;
+      error?: string;
+    }
  | {
      type: "replayTerminal";
      requestId: string;
@@ -1570,18 +1570,18 @@ it("hard cuts the unreleased cell transport, closing its terminals", async () =>
  expect(manager.getTerminal(old.terminalId)).toBeUndefined();
  const workers = await workerRecords(paseoHome);
  expect(workers).toHaveLength(1);
-  expect(workers[0]!.protocolVersion).toBe(2);
+  expect(workers[0]!.protocolVersion).toBe(3);
 });

 it("leaves a worker from a newer Paseo running, reports it incompatible, and closes it on a kill stop", async () => {
  const paseoHome = createTemporaryHome();
-  const newer = await leaveWorkerWithTerminal(paseoHome, 3);
+  const newer = await leaveWorkerWithTerminal(paseoHome, 4);

  manager = await startRealWorkerManager({ paseoHome });

  const unavailable = manager.getUnavailableError?.();
  expect(unavailable).toBeInstanceOf(TerminalWorkerIncompatibleError);
-  expect(unavailable?.message).toContain("worker protocol 3");
+  expect(unavailable?.message).toContain("worker protocol 4");
  await expect(manager.getTerminals(newer.cwd)).rejects.toBeInstanceOf(
    TerminalWorkerIncompatibleError,
  );
@@ -1799,9 +1799,12 @@ it("completes a replay larger than the socket budget while a sibling stays usabl
    if (event.type === "output") seen.push(event.data);
  });
  let bytes = 0;
-  const metadata = await manager.replayTerminal(terminal.id, undefined, (event) => {
+  const metadata = await manager.replayTerminal(terminal.id, undefined, async (event) => {
    if (event.type === "data") bytes += event.data.byteLength;
-    if (event.type === "begin") sibling.send({ type: "input", data: "go\r" });
+    if (event.type === "begin") {
+      sibling.send({ type: "input", data: "go\r" });
+      await waitForCondition(() => seen.join("").includes("SIBLING-ALIVE"), 10000);
+    }
  });
  expect(bytes).toBeGreaterThan(4 * 1024 * 1024);
  expect(metadata?.cols).toBe(1024);
@@ -1840,6 +1843,7 @@ it("completes a replay larger than the socket budget while a sibling stays usabl
    if (message.type === "helloAck") {
      framed.send({ type: "replayTerminal", terminalId: terminal.id, requestId: "stalled-replay" });
    } else if (message.type === "terminalReplay") {
+      framed.send({ type: "replayAcknowledged", requestId: message.requestId });
      if (message.event.type === "begin") {
        began = true;
        socket.pause();
@@ -1853,7 +1857,7 @@ it("completes a replay larger than the socket budget while a sibling stays usabl
      completed = message.ok;
  });
  try {
-    framed.send({ type: "hello", secret: record!.secret, protocolVersion: 2 });
+    framed.send({ type: "hello", secret: record!.secret, protocolVersion: 3 });
    await waitForCondition(() => began, 10000);
    framed.send({
      type: "send",
@@ -1903,6 +1907,8 @@ it("completes a replay larger than the socket budget while a sibling stays usabl
        terminalId: terminal.id,
        requestId: "overflow-replay",
      });
+    if (message.type === "terminalReplay")
+      stalledFrames.send({ type: "replayAcknowledged", requestId: message.requestId });
    if (message.type === "terminalReplay" && message.event.type === "begin") {
      overflowBegan = true;
      stalled.pause();
@@ -1911,7 +1917,7 @@ it("completes a replay larger than the socket budget while a sibling stays usabl
      overflowError = message.error;
  });
  try {
-    stalledFrames.send({ type: "hello", secret: record!.secret, protocolVersion: 2 });
+    stalledFrames.send({ type: "hello", secret: record!.secret, protocolVersion: 3 });
    await waitForCondition(() => overflowBegan, 10000);
    stalledFrames.send({
      type: "send",
@@ -599,7 +599,22 @@ export function createWorkerTerminalManager(
  function attachWorker(): void {
    worker.on("message", (message: TerminalWorkerToParentMessage) => {
      if (message.type === "terminalReplay") {
-        replaySinks.get(message.requestId)?.(message.event);
+        const sink = replaySinks.get(message.requestId);
+        void Promise.resolve()
+          .then(() => sink?.(message.event))
+          .then(
+            () =>
+              worker.send({ type: "replayAcknowledged", requestId: message.requestId }, () => {}),
+            (error: unknown) =>
+              worker.send(
+                {
+                  type: "replayAcknowledged",
+                  requestId: message.requestId,
+                  error: error instanceof Error ? error.message : String(error),
+                },
+                () => {},
+              ),
+          );
        return;
      }
      if (isResponse(message)) {
@@ -678,6 +693,11 @@ export function createWorkerTerminalManager(
    const request = new Promise((resolve, reject) => {
      const timeout = setTimeout(() => {
        pendingRequests.delete(requestId);
+        if (sink)
+          worker.send(
+            { type: "replayAcknowledged", requestId, error: "Terminal replay delivery timed out" },
+            () => {},
+          );
        reject(new Error(`Terminal worker request timed out: ${input.type}`));
      }, requestTimeoutMs);
      pendingRequests.set(requestId, { resolve, reject, timeout });