fix(terminal): preserve replay across slow delivery and resizing

Keep app network delivery outside worker capture deadlines and hold captured geometry until parsing commits. Reconnect readiness is independent of the capture revision used for live-output deduplication.

Remove redundant replay acknowledgements and hard cut the unreleased worker contract at protocol 5 while preserving bounded transports.

docs/terminal-performance.md

@@ -22,7 +22,8 @@ Terminal frames share the daemon main event loop with all agent traffic. The `ev
 - **Replay is incremental across the daemon and SDK.** The worker retains 1,000 history rows. Capable clients request their configured history; the worker bounds it by retained history. Older apps retain their own 200-row request. Explicit SDK history demand is unchanged against older ANSI hosts, which apply their existing 500-row ceiling. Begin resets once, chunks keep a streaming UTF-8 decoder/parser, and each renderer suppresses generated replies while parsing replay. Never concatenate capable replay chunks to feed the renderer.
 - **Coalescers are leading+trailing throttles.** The first chunk after an idle window flushes immediately (synchronously); only sustained bursts wait for the trailing timer. Reverting to trailing-only adds a full window (~5ms) to every keystroke echo.
 - **Output coalescing happens in the worker, before IPC.** One socket frame per pty chunk was a main-loop flood under build output. Non-output messages (snapshot/snapshotReady/titleChange/exit) must flush the coalescer first so ordering is preserved.
-- **Replay delivery is isolated per viewer.** Live worker output never waits for a replay acknowledgement. Each daemon stream owns its bounded live tail and revision deduplication; a slow or failed replay cannot pause another viewer of the same terminal.
+- **Replay delivery is isolated per viewer.** Live worker output never waits for app delivery. Each daemon stream owns its bounded live tail and revision deduplication; a slow or failed replay cannot pause another viewer of the same terminal.
+- **Replay geometry belongs to the parser queue.** Native measurements and web fits must leave the captured dimensions intact through every chunk's commit. Apply the latest measured size after replay, before live bytes; clear, replacement and disposal must release that geometry ownership. A layout effect must not resize the renderer directly during replay.
 - **Coalesced output carries the LAST chunk's revision.** Snapshot replay dedup (`replayTerminalOutputAfterSnapshot`) skips buffered output with `revision <= replayRevision`; a merged batch with a lower revision would be wrongly skipped (lost output).
 - **The input-mode tracker runs once per process boundary, not per hop.** The worker owns the authoritative tracker; the daemon caches the replay preamble from replay responses and `snapshotReady` messages. Do not reintroduce a per-chunk `feed()` on the daemon main loop.
 - **Snapshot catch-up is backpressure-gated.** A stream falls back to a full snapshot only when `outputBytesSinceSnapshot > MAX_TERMINAL_OUTPUT_FRAME_BYTES` (256KB) **and** the client transport reports `bufferedAmount > MAX_CLIENT_BUFFERED_BYTES` (4MB). A client that keeps draining streams continuously, no matter how much output is produced. Before this gate existed, every 256KB of build output dropped a frame and forced a full JSON cell-grid snapshot (~200k objects across IPC) — the historical source of spiky lag and GC hitches.

docs/terminal-worker.md

@@ -40,7 +40,7 @@ The worker writes `$PASEO_HOME/terminal-worker/<pid>.json` (owner-only) with its
 
 Never more than two workers run. A new worker launches only when none is current, none is incompatible and fewer than two run. The count comes from the records, so an unreachable or unidentified worker counts, and so does one a cleanup could not confirm gone. When two run, the daemon starts with no current worker, and the first terminal created after one of them exits launches it.
 
-**The frozen envelope never changes across versions:** the frame format (4-byte length + `v8.serialize`), `hello`/`helloAck`, the `listTerminals` info fields and `shutdown`. That is what lets any daemon close any worker: a kill stop sends `shutdown` to a newer worker it cannot otherwise drive. Protocol 4 hard cuts unreleased protocols 1 (cell transport), 2 (socket-paced replay without consumption acknowledgements) and 3 (terminal-wide replay tails): those workers are obsolete and close. For future versions, keep a previous-version adapter so the draining worker can still be driven, and drop the adapter for the version before that.
+**The frozen envelope never changes across versions:** the frame format (4-byte length + `v8.serialize`), `hello`/`helloAck`, the `listTerminals` info fields and `shutdown`. That is what lets any daemon close any worker: a kill stop sends `shutdown` to a newer worker it cannot otherwise drive. Protocol 5 hard cuts unreleased protocols 1 through 4, including the acknowledgement-based replay contract: those workers are obsolete and close. For future versions, keep a previous-version adapter so the draining worker can still be driven, and drop the adapter for the version before that.
 
 Tests stand up other versions with `PASEO_TERMINAL_WORKER_PROTOCOL_OVERRIDE`, read by the worker. It is test-only.
 
@@ -48,7 +48,11 @@ Tests stand up other versions with `PASEO_TERMINAL_WORKER_PROTOCOL_OVERRIDE`, re
 
 The worker keeps the headless xterm model and generates ANSI. A replay request emits begin metadata (dimensions, retained history, revision and input modes), at most 64 KiB per data chunk, and end. Live output keeps flowing independently during replay. Capture is synchronous with PTY callbacks; flush the output coalescer before the cut so a later batch never includes bytes already in replay. Retained-history metadata counts the worker history before the request limit.
 
-Replay waits for the daemon to acknowledge each event after its sink finishes before sending the next chunk. The acknowledgement belongs to the replay request, so a slow viewer does not pause the shared worker socket or block sibling terminals. PTYs keep parsing while writes wait. Each daemon viewer holds at most 4 MiB of its own live tail while replay is in flight, then drops revisions already present in the worker capture. Tail overflow, sink rejection and the existing 10-second worker request deadline fail only that viewer; healthy viewers of the same terminal keep streaming. The worker has no terminal-wide replay tail. A timed-out request rejects its outstanding acknowledgement; a late sink completion cannot acknowledge a request on a replacement connection. Live traffic that fills the shared socket beyond 4 MiB disconnects the stalled controller; the daemon reconnects and requests fresh replay. The SDK advertises `terminal_chunked_replay` only because it decodes begin/data/end. Capable clients receive these events directly; the daemon collects one Restore frame only for older ANSI clients. The SDK normalizes an older host's single Restore into the same lifecycle. Capture stays a worker request; bootstrap readiness uses `hasOutput`, not a transported grid.
+The worker paces replay writes on its local socket; the daemon synchronously queues each event into the application's bounded transport. App network delivery does not extend the worker's request deadline. There is no per-chunk acknowledgement or second replay delivery queue in the daemon. The physical application transport owns its 64 MiB high-water close, so a backed-up app cannot hold worker capture open. Each daemon viewer still holds at most 4 MiB of live output during capture, then drops revisions already present in the worker replay. Capture failure and tail overflow affect only that viewer; PTYs and healthy viewers keep running.
+
+Live traffic that fills the shared worker socket beyond 4 MiB disconnects the stalled controller. Reconnection adopts the surviving sessions and requests fresh replay for existing streams. Readiness does not require a revision: adoption announces readiness without one, while the replay capture supplies the revision used for deduplication.
+
+The SDK advertises `terminal_chunked_replay` only because it decodes begin/data/end. Capable clients receive these events directly; the daemon collects one Restore frame only for older ANSI clients. The SDK normalizes an older host's single Restore into the same lifecycle. Capture stays a worker request; bootstrap readiness uses `hasOutput`, not a transported grid.
 
 ## Running from its own copy
 

packages/app/src/components/terminal-emulator-native-grid.native.tsx

@@ -379,8 +379,7 @@ function NativeTerminalEmulator({
       hasNotifiedMeasuredSizeRef.current = true;
       terminalSizeRef.current = policyResult.measuredSize;
       if (policyResult.measuredSizeChanged) {
-        terminalRef.current?.resize(policyResult.measuredSize);
-        schedulePaint();
+        outputDrainRef.current?.resize();
       }
       callbacksRef.current.onResize?.({
         rows: policyResult.measuredSize.rows,
@@ -389,7 +388,7 @@ function NativeTerminalEmulator({
         forceClaim: policyResult.resizeClaim?.force,
       });
     },
-    [schedulePaint],
+    [],
   );
 
   const resetTerminal = useCallback(() => {

packages/app/src/terminal/native-renderer/headless-terminal-output-drain.test.ts

@@ -461,6 +461,9 @@ test.each(["clear", "replacement", "dispose"] as const)(
       drain.replay({ type: "begin", rows: 30, cols: 120 });
       drain.replay({ type: "data", data: new TextEncoder().encode("old") });
       await writing;
+      local.cols = 60;
+      drain.resize();
+      expect(terminal.getState().cols).toBe(120);
       drain.replay({ type: "end" });
       if (action === "clear") drain.clear();
       else if (action === "dispose") {
@@ -482,3 +485,36 @@ test.each(["clear", "replacement", "dispose"] as const)(
     }
   },
 );
+
+test("holds captured geometry across mid-replay measurement changes", async () => {
+  const source = createNativeHeadlessTerminal({ rows: 4, cols: 8, scrollbackLines: 100 });
+  await source.write("ABCDEFGH\r\n12345678\r\nabcdefgh\r\nxyz\x1b[2;8H");
+  const ansi = new TextEncoder().encode(renderTerminalSnapshotToAnsi(source.getState()));
+  const terminal = createNativeHeadlessTerminal({ rows: 4, cols: 8, scrollbackLines: 100 });
+  const expected = createNativeHeadlessTerminal({ rows: 4, cols: 8, scrollbackLines: 100 });
+  await expected.write(new TextDecoder().decode(ansi));
+  expected.resize({ rows: 4, cols: 4 });
+  let measured = { rows: 4, cols: 8 };
+  const drain = createNativeTerminalOutputDrain({
+    write: (chunk) => terminal.write(chunk),
+    reset: (size) => {
+      if (size) terminal.resize(size);
+      terminal.reset();
+    },
+    restoreSize: () => terminal.resize(measured),
+    getViewportState: () => terminal.getViewportState(),
+    onPaint: () => {},
+  });
+  try {
+    drain.replay({ type: "begin", rows: 4, cols: 8 });
+    await drain.flush();
+    measured = { rows: 4, cols: 4 };
+    drain.resize();
+    drain.replay({ type: "data", data: ansi });
+    drain.replay({ type: "end" });
+    await drain.flush();
+    expect(terminal.getState()).toEqual(expected.getState());
+  } finally {
+    drain.dispose();
+  }
+});

packages/app/src/terminal/native-renderer/headless-terminal-output-drain.ts

@@ -14,6 +14,7 @@ export interface NativeTerminalParseSample {
 export interface NativeTerminalOutputDrain {
   replay(event: import("@getpaseo/client").TerminalReplayEvent): void;
   enqueueText(text: string): void;
+  resize(): void;
   clear(): void;
   flush(): Promise<void>;
   dispose(): void;
@@ -75,6 +76,7 @@ export function createNativeTerminalOutputDrain(
   let drainScheduled = false;
   let draining = false;
   let disposed = false;
+  let replaying = false;
   let paintFrame: number | null = null;
   let generation = 0;
   const replayDecoder = new TextDecoder();
@@ -138,7 +140,10 @@ export function createNativeTerminalOutputDrain(
         if (operation.type !== "write") {
           operations.shift();
           if (operation.type === "reset") options.reset(operation.size);
-          else options.restoreSize();
+          else {
+            options.restoreSize();
+            replaying = false;
+          }
           requestPaint();
           continue;
         }
@@ -172,6 +177,7 @@ export function createNativeTerminalOutputDrain(
     replay(event): void {
       if (disposed) return;
       if (event.type === "begin") {
+        replaying = true;
         generation++;
         replayDecoder.decode();
         operations = [
@@ -199,6 +205,12 @@ export function createNativeTerminalOutputDrain(
       scheduleDrain();
     },
 
+    resize(): void {
+      if (disposed || replaying) return;
+      operations.push({ type: "restore-size" });
+      scheduleDrain();
+    },
+
     clear(): void {
       if (disposed) return;
       generation += 1;

packages/app/src/terminal/runtime/terminal-emulator-runtime.browser.test.ts

@@ -781,7 +781,9 @@ describe("incremental replay lifecycle", () => {
     "recovers live input after interrupted replay and %s",
     async (cleanup) => {
       const mounted = createTerminalHost({ width: 720, height: 360 });
-      mounted.runtime.replay({ type: "begin" });
+      mounted.runtime.replay({ type: "begin", rows: 30, cols: 120 });
+      mounted.root.style.width = "360px";
+      mounted.runtime.resize({ shouldClaim: false });
       mounted.runtime.replay({ type: "data", data: terminalOutput("interrupted") });
       if (cleanup === "clear")
         await new Promise<void>((resolve) => mounted.runtime.clear({ onCommitted: resolve }));
@@ -799,6 +801,7 @@ describe("incremental replay lifecycle", () => {
       );
       getBrowserTerminal().input("user-input", true);
       expect(mounted.inputs).toContain("user-input");
+      expect(getBrowserTerminal().cols).toBeLessThan(60);
     },
   );
 
@@ -836,3 +839,31 @@ it("restores fitted web geometry after replay before live output", async () => {
   mounted.runtime.resize({ forceClaim: true, shouldClaim: true });
   expect(getBrowserTerminal().cols).toBe(latestSize(mounted.sizes).cols);
 });
+
+it("defers layout fit until all replay chunks commit", async () => {
+  const mounted = createTerminalHost({ width: 720, height: 360 });
+  await settleMountRefits();
+  mounted.runtime.replay({ type: "begin", rows: 30, cols: 120 });
+  await new Promise<void>((resolve) =>
+    mounted.runtime.write({
+      data: terminalOutput("first"),
+      suppressInput: true,
+      onCommitted: resolve,
+    }),
+  );
+  expect(getBrowserTerminal().cols).toBe(120);
+  mounted.root.style.width = "360px";
+  mounted.runtime.resize({ shouldClaim: true, forceClaim: true, forceRefresh: true });
+  await nextFrame();
+  expect(getBrowserTerminal().cols).toBe(120);
+  mounted.runtime.replay({ type: "data", data: terminalOutput("\x1b[2;110Hsecond") });
+  mounted.runtime.replay({ type: "end" });
+  await new Promise<void>((resolve) =>
+    mounted.runtime.write({ data: terminalOutput("LIVE"), onCommitted: resolve }),
+  );
+  expect(getBrowserTerminal().cols).toBe(latestSize(mounted.sizes).cols);
+  expect(getBrowserTerminal().cols).toBeLessThan(60);
+  expect(latestSize(mounted.sizes)).toMatchObject({ shouldClaim: true, forceClaim: true });
+  mounted.runtime.resize({ shouldClaim: true, forceClaim: true });
+  expect(getBrowserTerminal().cols).toBe(latestSize(mounted.sizes).cols);
+});

packages/app/src/terminal/runtime/terminal-emulator-runtime.ts

@@ -277,6 +277,7 @@ export class TerminalEmulatorRuntime {
     },
   };
 
+  private replayResize: TerminalResizeRequest | null = null;
   private fitAndEmitResize: ((input?: TerminalResizeRequest) => void) | null = null;
   private lastSize: { rows: number; cols: number } | null = null;
   private cleanup: (() => void) | null = null;
@@ -649,6 +650,7 @@ export class TerminalEmulatorRuntime {
     window.__paseoTerminal = terminal;
 
     const fitAndEmitResize = (resizeInput?: TerminalResizeRequest): void => {
+      if (this.deferReplayResize(resizeInput)) return;
       const forceRefresh = resizeInput?.forceRefresh ?? false;
       const shouldClaim = resizeInput?.shouldClaim ?? true;
       const forceClaim = resizeInput?.forceClaim ?? false;
@@ -893,6 +895,7 @@ export class TerminalEmulatorRuntime {
 
   replay(event: import("@getpaseo/client").TerminalReplayEvent): void {
     if (event.type === "begin") {
+      this.replayResize ??= { shouldClaim: false };
       this.restoreOutput({ data: EMPTY_TERMINAL_OUTPUT, rows: event.rows, cols: event.cols });
     } else if (event.type === "data") {
       this.write({ data: event.data, suppressInput: true });
@@ -919,6 +922,16 @@ export class TerminalEmulatorRuntime {
     this.processOutputQueue();
   }
 
+  private deferReplayResize(input?: TerminalResizeRequest): boolean {
+    if (!this.replayResize) return false;
+    this.replayResize = {
+      forceRefresh: this.replayResize.forceRefresh || input?.forceRefresh,
+      shouldClaim: this.replayResize.shouldClaim || (input?.shouldClaim ?? true),
+      forceClaim: this.replayResize.forceClaim || input?.forceClaim,
+    };
+    return true;
+  }
+
   resize(input?: TerminalResizeRequest): void {
     this.fitAndEmitResize?.(input);
   }
@@ -1038,6 +1051,7 @@ export class TerminalEmulatorRuntime {
       commit();
     }
     this.hasUngatedWrites = false;
+    this.replayResize = null;
 
     this.cleanup?.();
     this.cleanup = null;
@@ -1172,7 +1186,9 @@ export class TerminalEmulatorRuntime {
       // commit callback before fitting, or that flush consumes the next write.
       queueMicrotask(() => {
         if (this.inFlightOutputOperation !== operation || this.terminal !== terminal) return;
-        this.fitAndEmitResize?.({ shouldClaim: false });
+        const resize = this.replayResize ?? { shouldClaim: false };
+        this.replayResize = null;
+        this.fitAndEmitResize?.(resize);
         finalizeOperation(operation);
       });
       return;
@@ -1183,11 +1199,22 @@ export class TerminalEmulatorRuntime {
       this.inputModeTracker.reset();
       this.emitInputModeChange();
       terminal.reset();
-      finalizeOperation(operation);
+      if (!this.replayResize) {
+        finalizeOperation(operation);
+        return;
+      }
+      queueMicrotask(() => {
+        if (this.inFlightOutputOperation !== operation || this.terminal !== terminal) return;
+        const resize = this.replayResize ?? { shouldClaim: false };
+        this.replayResize = null;
+        this.fitAndEmitResize?.(resize);
+        finalizeOperation(operation);
+      });
       return;
     }
 
     if (operation.type === "snapshot") {
+      this.replayResize ??= { shouldClaim: false };
       this.inputModeDecoder.decode();
       this.inputModeTracker.reset();
       this.emitInputModeChange();

packages/server/src/terminal/terminal-replay.ts

@@ -21,7 +21,7 @@ export type TerminalReplayEvent =
   | { type: "begin"; metadata: TerminalReplayMetadata }
   | { type: "data"; data: Uint8Array }
   | { type: "end" };
-export type TerminalReplaySink = (event: TerminalReplayEvent) => void | Promise<void>;
+export type TerminalReplaySink = (event: TerminalReplayEvent) => void;
 export const TERMINAL_REPLAY_CHUNK_BYTES = 64 * 1024;
 
 export function* createTerminalReplay(
@@ -53,7 +53,7 @@ export async function replayTerminal(
   const replay = createTerminalReplay(session, options);
   let next = replay.next();
   while (!next.done) {
-    await sink(next.value);
+    sink(next.value);
     next = replay.next();
   }
   return next.value;
@@ -67,7 +67,6 @@ interface TerminalReplayPorts {
 /** Worker-owned ANSI capture and paced delivery; live observations never wait for a replay. */
 export function createTerminalReplayController(ports: TerminalReplayPorts) {
   let controller: FramedSocket | null = null;
-  const acknowledgements = new Map<string, { resolve(): void; reject(error: Error): void }>();
 
   function sendToController(message: TerminalWorkerToParentMessage): void {
     // A stalled physical controller is different from a stalled viewer: all its
@@ -98,47 +97,29 @@ export function createTerminalReplayController(ports: TerminalReplayPorts) {
     // its own live tail and discards revisions already included in this cut.
     const replay = createTerminalReplay(session, message.options);
     let next = replay.next();
-    try {
-      while (!next.done) {
-        if (controller !== connection) throw new Error("Terminal replay controller disconnected");
-        const acknowledged = new Promise<void>((resolve, reject) => {
-          acknowledgements.set(message.requestId, { resolve, reject });
-        });
-        void acknowledged.catch(() => {});
-        await connection.sendAndWait({
-          type: "terminalReplay",
-          requestId: message.requestId,
-          event: next.value,
-        });
-        await acknowledged;
-        acknowledgements.delete(message.requestId);
-        next = replay.next();
-      }
+    while (!next.done) {
+      if (controller !== connection) throw new Error("Terminal replay controller disconnected");
       await connection.sendAndWait({
-        type: "response",
+        type: "terminalReplay",
         requestId: message.requestId,
-        ok: true,
-        result: next.value,
+        event: next.value,
       });
-    } finally {
-      acknowledgements.delete(message.requestId);
+      next = replay.next();
     }
+    await connection.sendAndWait({
+      type: "response",
+      requestId: message.requestId,
+      ok: true,
+      result: next.value,
+    });
   }
 
   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 {
       if (controller === connection) return;
       const previous = controller;
       controller = connection;
       previous?.socket.destroy();
-      for (const pending of acknowledgements.values())
-        pending.reject(new Error("Terminal replay controller disconnected"));
-      acknowledgements.clear();
     },
     send: sendToController,
     async replay(

packages/server/src/terminal/terminal-restore.ts

@@ -69,15 +69,15 @@ function resolveVisibleRestoreScrollbackLines(value: number | undefined): number
 export function createTerminalRestoreDelivery(input: {
   slot: number;
   chunked: boolean;
-  send(frame: Uint8Array): Promise<void>;
+  send(frame: Uint8Array): 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) => {
+  return (event) => {
     if (!input.chunked) {
       if (event.type === "data") legacyChunks.push(event.data);
       if (event.type === "end")
-        await input.send(
+        input.send(
           encodeTerminalRestoreFrame({
             slot: input.slot,
             data: Buffer.concat(legacyChunks),
@@ -93,7 +93,7 @@ export function createTerminalRestoreDelivery(input: {
     let payload: string | Uint8Array | undefined;
     if (event.type === "begin") payload = JSON.stringify(event.metadata);
     if (event.type === "data") payload = event.data;
-    await input.send(
+    input.send(
       encodeTerminalStreamFrame({ slot: input.slot, opcode: opcodes[event.type], payload }),
     );
   };

packages/server/src/terminal/terminal-session-controller.test.ts

@@ -831,7 +831,7 @@ describe("terminal-session-controller backpressure snapshot fallback", () => {
 
     expect(frames.some((frame) => frame.opcode === TerminalStreamOpcode.Restore)).toBe(true);
   });
-  test("paces capable catch-up while the direct socket drains instead of ending the stream", async () => {
+  test("queues capable catch-up without waiting for the direct socket to drain", async () => {
     let buffered = 0;
     let release!: () => void;
     const gate = new Promise<void>((resolve) => {
@@ -846,7 +846,11 @@ describe("terminal-session-controller backpressure snapshot fallback", () => {
     pushOutput("y".repeat(300 * 1024));
     await waitForCoalescerFlush();
     expect(messages.filter((message) => message.type === "terminal_stream_exit")).toEqual([]);
-    expect(frames.map((frame) => frame.opcode)).toEqual([TerminalStreamOpcode.ReplayBegin]);
+    expect(frames.map((frame) => frame.opcode)).toEqual([
+      TerminalStreamOpcode.ReplayBegin,
+      TerminalStreamOpcode.ReplayData,
+      TerminalStreamOpcode.ReplayEnd,
+    ]);
     buffered = 0;
     release();
     await waitForCoalescerFlush();
@@ -1292,3 +1296,51 @@ test.each([false, true])(
     }
   },
 );
+
+test("bounds a capturing viewer tail without ending a healthy viewer", async () => {
+  const pending = deferred<ReplayFixture | null>();
+  let reads = 0;
+  const f = exitFixture(() =>
+    ++reads === 2 ? pending.promise : Promise.resolve(terminalReplay("retained", 0)),
+  );
+  const desktop = {},
+    phone = {};
+  try {
+    await f.subscribe(desktop, true, "desktop");
+    await waitForCoalescerFlush();
+    await f.subscribe(phone, true, "phone");
+    f.output("x".repeat(4 * 1024 * 1024 + 1), 1);
+    await waitForCoalescerFlush();
+    expect(
+      f.frames.filter(
+        (row) => row.source === phone && row.message?.type === "terminal_stream_exit",
+      ),
+    ).toHaveLength(1);
+    expect(
+      f.frames.filter(
+        (row) => row.source === desktop && row.message?.type === "terminal_stream_exit",
+      ),
+    ).toHaveLength(0);
+    expect(
+      f.frames.some(
+        (row) => row.source === desktop && row.binary?.opcode === TerminalStreamOpcode.Output,
+      ),
+    ).toBe(true);
+    pending.resolve(terminalReplay("phone", 1));
+    await waitForCoalescerFlush();
+    f.output("AFTER-OVERFLOW", 2);
+    await waitForCoalescerFlush();
+    expect(
+      f.frames.some(
+        (row) =>
+          row.source === desktop &&
+          row.binary?.opcode === TerminalStreamOpcode.Output &&
+          Buffer.from(row.binary.payload).toString().includes("AFTER-OVERFLOW"),
+      ),
+    ).toBe(true);
+    expect(f.delivery.registrationCount).toBe(1);
+  } finally {
+    pending.resolve(null);
+    await f.delivery.close();
+  }
+});

packages/server/src/terminal/terminal-session-controller.ts

@@ -67,7 +67,8 @@ interface ActiveTerminalStream {
   needsSnapshot: boolean;
   snapshotInFlight: boolean;
   retrySnapshotErrors: boolean;
-  readyRevision?: number;
+  // Readiness is independent of revision: reconnect adoption has no revision.
+  snapshotReady: boolean;
   restore?: TerminalRestoreOptions;
   bufferedOutputs: BufferedTerminalOutput[];
   bufferedOutputBytes: number;
@@ -898,7 +899,7 @@ export class TerminalSessionController {
       snapshotInFlight: false,
       retrySnapshotErrors: options.retrySnapshotErrors,
       exiting: false,
-      readyRevision: undefined,
+      snapshotReady: false,
       restore: options?.restore,
       bufferedOutputs: [],
       bufferedOutputBytes: 0,
@@ -953,7 +954,7 @@ export class TerminalSessionController {
           return;
         }
         if (message.type === "snapshot" || message.type === "snapshotReady") {
-          activeStream.readyRevision = message.revision;
+          activeStream.snapshotReady = true;
           activeStream.outputCoalescer.flush();
           activeStream.needsSnapshot = true;
           void this.trySendSnapshot(activeStream);
@@ -1025,7 +1026,7 @@ export class TerminalSessionController {
       this.detachStream(activeStream.terminalId, { emitExit: true });
       return;
     }
-    if (activeStream.restore && activeStream.readyRevision === undefined) {
+    if (activeStream.restore && !activeStream.snapshotReady) {
       return;
     }
 
@@ -1095,14 +1096,15 @@ export class TerminalSessionController {
     const deliver = createTerminalRestoreDelivery({
       slot: activeStream.slot,
       chunked,
-      send: (frame) => activeStream.owner.emitBinaryAndWait(frame),
+      // Network delivery belongs to the bounded application transport, not the worker request.
+      send: (frame) => activeStream.owner.emitBinary(frame),
     });
     const snapshot = await terminalManager.replayTerminal(
       activeStream.terminalId,
       snapshotOptions,
-      async (event) => {
+      (event) => {
         if (this.activeStreams.get(activeStream.slot) !== activeStream) return;
-        await deliver(event);
+        deliver(event);
       },
     );
     if (this.activeStreams.get(activeStream.slot) !== activeStream) {

packages/server/src/terminal/terminal-worker-process.ts

@@ -552,11 +552,6 @@ function runWorker(argv: string[]): void {
         return;
       }
 
-      case "replayAcknowledged": {
-        replayController.acknowledge(message.requestId, message.error);
-        return;
-      }
-
       case "replayTerminal": {
         await replayController.replay(message);
         return;

packages/server/src/terminal/terminal-worker-protocol.test.ts

@@ -6,6 +6,7 @@ describe("classifyTerminalWorkerProtocol", () => {
     expect(classifyTerminalWorkerProtocol(1, 2)).toBe("obsolete");
     expect(classifyTerminalWorkerProtocol(2, 3)).toBe("obsolete");
     expect(classifyTerminalWorkerProtocol(3, 4)).toBe("obsolete");
+    expect(classifyTerminalWorkerProtocol(4, 5)).toBe("obsolete");
   });
   it.each([
     [6, "current"],

packages/server/src/terminal/terminal-worker-protocol.ts

@@ -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 = 4;
+export const TERMINAL_WORKER_PROTOCOL_VERSION = 5;
 
 // 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";
-  // Protocols 1, 2 and 3 were unreleased; they have no draining adapter.
-  if (reported === current - 1 && reported >= 4) return "previous";
+  // Protocols 1 through 4 were unreleased; they have no draining adapter.
+  if (reported === current - 1 && reported >= 5) return "previous";
   return reported < current ? "obsolete" : "newer";
 }
 
@@ -122,11 +122,6 @@ export type TerminalWorkerRequest =
       terminalId: string;
       options?: WorkerKillAndWaitOptions;
     }
-  | {
-      type: "replayAcknowledged";
-      requestId: string;
-      error?: string;
-    }
   | {
       type: "replayTerminal";
       requestId: string;

packages/server/src/terminal/worker-terminal-manager.test.ts

@@ -1569,31 +1569,34 @@ async function leaveWorkerWithTerminal(
   };
 }
 
-it("hard cuts the unreleased cell transport, closing its terminals", async () => {
-  const paseoHome = createTemporaryHome();
-  const old = await leaveWorkerWithTerminal(paseoHome, 1);
+it.each([1, 4])(
+  "hard cuts unreleased worker protocol %s, closing its terminals",
+  async (version) => {
+    const paseoHome = createTemporaryHome();
+    const old = await leaveWorkerWithTerminal(paseoHome, version);
 
-  manager = await startRealWorkerManager({ paseoHome });
+    manager = await startRealWorkerManager({ paseoHome });
 
-  await waitForCondition(
-    () => !isProcessAlive(old.workerPid) && !isProcessAlive(old.programPid),
-    10000,
-  );
-  expect(manager.getTerminal(old.terminalId)).toBeUndefined();
-  const workers = await workerRecords(paseoHome);
-  expect(workers).toHaveLength(1);
-  expect(workers[0]!.protocolVersion).toBe(4);
-});
+    await waitForCondition(
+      () => !isProcessAlive(old.workerPid) && !isProcessAlive(old.programPid),
+      10000,
+    );
+    expect(manager.getTerminal(old.terminalId)).toBeUndefined();
+    const workers = await workerRecords(paseoHome);
+    expect(workers).toHaveLength(1);
+    expect(workers[0]!.protocolVersion).toBe(5);
+  },
+);
 
 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, 5);
+  const newer = await leaveWorkerWithTerminal(paseoHome, 6);
 
   manager = await startRealWorkerManager({ paseoHome });
 
   const unavailable = manager.getUnavailableError?.();
   expect(unavailable).toBeInstanceOf(TerminalWorkerIncompatibleError);
-  expect(unavailable?.message).toContain("worker protocol 5");
+  expect(unavailable?.message).toContain("worker protocol 6");
   await expect(manager.getTerminals(newer.cwd)).rejects.toBeInstanceOf(
     TerminalWorkerIncompatibleError,
   );
@@ -1812,11 +1815,10 @@ 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, async (event) => {
+  const metadata = await manager.replayTerminal(terminal.id, undefined, (event) => {
     if (event.type === "data") bytes += event.data.byteLength;
     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);
@@ -1841,6 +1843,7 @@ it("completes a replay larger than the socket budget while a sibling stays usabl
   let ended = false;
   let completed = false;
   let resumedBytes = 0;
+  let cancelledReplayBegan = false;
   let existingLiveOutput = "";
   const replayOrder: string[] = [];
   const framed = attachFramedSocket(socket, (value) => {
@@ -1856,7 +1859,13 @@ 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.requestId === "cancelled-replay") {
+        if (message.event.type === "begin") {
+          cancelledReplayBegan = true;
+          socket.pause();
+        }
+        return;
+      }
       if (message.event.type === "begin") {
         began = true;
         socket.pause();
@@ -1870,7 +1879,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: 4 });
+    framed.send({ type: "hello", secret: record!.secret, protocolVersion: 5 });
     await waitForCondition(() => began, 10000);
     framed.send({
       type: "send",
@@ -1898,12 +1907,19 @@ it("completes a replay larger than the socket budget while a sibling stays usabl
     await waitForCondition(() => existingLiveOutput.includes("SAME-TERMINAL-LIVE"), 10000);
     expect(existingLiveOutput.match(/SAME-TERMINAL-LIVE/g)).toHaveLength(1);
     expect(replayOrder.indexOf("live")).toBeLessThan(replayOrder.indexOf("end"));
+    framed.send({ type: "replayTerminal", terminalId: terminal.id, requestId: "cancelled-replay" });
+    await waitForCondition(() => cancelledReplayBegan, 10000);
+    // Replace a controller while its large replay is socket-paced. The original
+    // request must never leak frames or completion onto the replacement connection.
+    manager = await startRealWorkerManager({ paseoHome });
+    socket.resume();
+    await waitForCondition(() => socket.destroyed, 10000);
+    expect((await readReplay(manager, terminal.id))?.state.cols).toBe(1024);
   } finally {
     socket.destroy();
   }
-  manager = await startRealWorkerManager({ paseoHome });
   expect(
-    (await manager.captureTerminal(terminal.id)).lines.some((line) => line.includes("READY-WIDE")),
+    (await manager!.captureTerminal(terminal.id)).lines.some((line) => line.includes("READY-WIDE")),
   ).toBe(true);
   const exitingReplay = await manager.replayTerminal(
     terminal.id,
@@ -1917,7 +1933,11 @@ it("completes a replay larger than the socket budget while a sibling stays usabl
   expect(manager.getTerminal(sibling.id)?.id).toBe(sibling.id);
 }, 60000);
 
-function terminalViewers(terminalManager: DaemonTerminalManager, terminalId: string) {
+function terminalViewers(
+  terminalManager: DaemonTerminalManager,
+  terminalId: string,
+  chunked = true,
+) {
   const frames: Array<{ source: object; frame: TerminalStreamFrame }> = [];
   const messages: Array<{ source: object; message: SessionOutboundMessage }> = [];
   let blockedSource: object | null = null;
@@ -1933,7 +1953,10 @@ function terminalViewers(terminalManager: DaemonTerminalManager, terminalId: str
     async (source, bytes) => {
       const frame = decodeTerminalStreamFrame(bytes)!;
       frames.push({ source, frame });
-      if (source === blockedSource && frame.opcode === TerminalStreamOpcode.ReplayBegin) {
+      if (
+        source === blockedSource &&
+        frame.opcode === (chunked ? TerminalStreamOpcode.ReplayBegin : TerminalStreamOpcode.Restore)
+      ) {
         started = true;
         await block;
       }
@@ -1945,7 +1968,7 @@ function terminalViewers(terminalManager: DaemonTerminalManager, terminalId: str
       delivery.reply(message);
     },
     hasBinaryChannel: () => true,
-    supportsChunkedReplay: () => true,
+    supportsChunkedReplay: () => chunked,
     isPathWithinRoot: () => true,
     sessionLogger: pino({ level: "silent" }),
   });
@@ -1988,7 +2011,7 @@ function terminalViewers(terminalManager: DaemonTerminalManager, terminalId: str
   };
 }
 
-it("isolates a rejected replay sink while another viewer of the same terminal stays live", async () => {
+it("isolates a failed app transport while another viewer of the same terminal stays live", async () => {
   manager = await startRealWorkerManager();
   const session = trackTerminal(
     await manager.createTerminal({
@@ -2011,6 +2034,11 @@ it("isolates a rejected replay sink while another viewer of the same terminal st
   try {
     await viewers.attach(desktop, "desktop");
     await waitForCondition(() => viewers.replayed(desktop), 10000);
+    await expect(
+      manager.replayTerminal(session.id, undefined, () => {
+        throw new Error("Consumer cancelled replay");
+      }),
+    ).rejects.toThrow("Consumer cancelled replay");
     viewers.block(phone);
     await viewers.attach(phone, "phone");
     await waitForCondition(viewers.started, 10000);
@@ -2019,10 +2047,8 @@ it("isolates a rejected replay sink while another viewer of the same terminal st
     expect(viewers.ended(desktop)).toHaveLength(0);
     expect(viewers.ended(phone)).toHaveLength(0);
     viewers.finish(new Error("Phone socket closed"));
-    await waitForCondition(() => viewers.ended(phone).length === 1, 10000);
-    expect(viewers.ended(phone)[0]?.message).toMatchObject({
-      payload: { error: "Phone socket closed" },
-    });
+    await viewers.delivery.detach(phone);
+    expect(viewers.ended(desktop)).toHaveLength(0);
     session.send({ type: "input", data: "AFTER-FAILURE\r" });
     await waitForCondition(() => viewers.text(desktop).includes("ECHO:AFTER-FAILURE"), 10000);
     expect(viewers.ended(desktop)).toHaveLength(0);
@@ -2035,141 +2061,150 @@ it("isolates a rejected replay sink while another viewer of the same terminal st
   }
 }, 20000);
 
-it.each(["overflow", "timeout", "complete"] as const)(
-  "keeps healthy same-terminal viewers live during replay %s",
-  async (outcome) => {
-    manager = await startRealWorkerManager({
-      requestTimeoutMs: outcome === "timeout" ? 2000 : 10000,
-    });
+it("keeps concurrent viewer replays and live output ordered behind slow app delivery", async () => {
+  manager = await startRealWorkerManager();
+  const session = trackTerminal(
+    await manager.createTerminal({
+      cwd: createTemporaryHome(),
+      workspaceId: "concurrent-delivery",
+      ...nodeTerminalCommand(
+        'process.stdout.write("READY\\r\\n"); process.stdin.on("data", data => process.stdout.write("ECHO:" + data)); setInterval(() => {}, 1000);',
+      ),
+    }),
+  );
+  await waitForCondition(
+    async () => (await manager!.captureTerminal(session.id)).lines.join("").includes("READY"),
+    10000,
+  );
+  const viewers = terminalViewers(manager, session.id);
+  const desktop = {},
+    phone = {},
+    concurrent = {};
+  try {
+    await viewers.attach(desktop, "desktop");
+    await waitForCondition(() => viewers.replayed(desktop), 10000);
+    viewers.block(phone);
+    await viewers.attach(phone, "phone");
+    await waitForCondition(viewers.started, 10000);
+    await viewers.attach(concurrent, "concurrent");
+    await waitForCondition(() => viewers.replayed(concurrent) && viewers.replayed(phone), 10000);
+    session.send({ type: "input", data: "DURING-DELIVERY\r" });
+    await waitForCondition(
+      () =>
+        viewers.text(phone).includes("ECHO:DURING-DELIVERY") &&
+        viewers.text(desktop).includes("ECHO:DURING-DELIVERY"),
+      3000,
+    );
+    expect(viewers.text(phone).match(/ECHO:DURING-DELIVERY/g)).toHaveLength(1);
+    const order = viewers.frames
+      .filter((row) => row.source === phone)
+      .map((row) => row.frame.opcode);
+    expect(order.indexOf(TerminalStreamOpcode.Output)).toBeGreaterThan(
+      order.indexOf(TerminalStreamOpcode.ReplayEnd),
+    );
+    expect(viewers.ended(phone)).toHaveLength(0);
+    expect(viewers.ended(desktop)).toHaveLength(0);
+  } finally {
+    viewers.finish();
+    await viewers.delivery.close();
+  }
+}, 20000);
+
+it.each([true, false])(
+  "queues replay independently of app delivery past the worker deadline (chunked=%s)",
+  async (chunked) => {
+    manager = await startRealWorkerManager({ requestTimeoutMs: 800 });
     const session = trackTerminal(
       await manager.createTerminal({
         cwd: createTemporaryHome(),
-        workspaceId: "per-viewer-tail",
-        ...nodeTerminalCommand(`
-        process.stdout.write("READY\\r\\n");
-        process.stdin.on("data", data => {
-          if (data.toString().includes("burst")) process.stdout.write("x".repeat(5 * 1024 * 1024) + "\\r\\nBURST-DONE\\r\\n");
-          else process.stdout.write("ECHO:" + data);
-        });
-        setInterval(() => {}, 1000);
-      `),
+        workspaceId: "slow-delivery",
+        ...nodeTerminalCommand(
+          'process.stdout.write("READY\\r\\n"); process.stdin.on("data", data => process.stdout.write("ECHO:" + data)); setInterval(() => {}, 1000);',
+        ),
       }),
     );
     await waitForCondition(
-      async () =>
-        (await manager!.captureTerminal(session.id)).lines.some((line) => line.includes("READY")),
+      async () => (await manager!.captureTerminal(session.id)).lines.join("").includes("READY"),
       10000,
     );
-    const viewers = terminalViewers(manager, session.id);
-    const desktop = {},
-      phone = {},
-      concurrent = {};
+    const viewers = terminalViewers(manager, session.id, chunked);
+    const phone = {};
+    viewers.block(phone);
     try {
-      await viewers.attach(desktop, "desktop");
-      await waitForCondition(() => viewers.replayed(desktop), 10000);
-      viewers.block(phone);
-      await viewers.attach(phone, "phone");
+      await viewers.attach(phone, "slow-phone");
       await waitForCondition(viewers.started, 10000);
-      session.send({ type: "input", data: "DURING-REPLAY\r" });
-      await waitForCondition(() => viewers.text(desktop).includes("ECHO:DURING-REPLAY"), 1500);
-      if (outcome === "complete") {
-        await viewers.attach(concurrent, "concurrent");
-        await waitForCondition(() => viewers.replayed(concurrent), 1500);
-        expect(viewers.replayed(phone)).toBe(false);
-        viewers.finish();
-        await waitForCondition(() => viewers.text(phone).includes("ECHO:DURING-REPLAY"), 10000);
-        expect(viewers.text(phone).match(/ECHO:DURING-REPLAY/g)).toHaveLength(1);
-        const order = viewers.frames
-          .filter((row) => row.source === phone)
-          .map((row) => row.frame.opcode);
-        expect(order.indexOf(TerminalStreamOpcode.Output)).toBeGreaterThan(
-          order.indexOf(TerminalStreamOpcode.ReplayEnd),
-        );
-        expect(viewers.ended(phone)).toHaveLength(0);
-      } else {
-        if (outcome === "overflow") session.send({ type: "input", data: "burst\r" });
-        await waitForCondition(() => viewers.ended(phone).length === 1, 10000);
-        const error =
-          outcome === "overflow"
-            ? "Terminal replay live tail exceeded the buffer; attach again"
-            : "Terminal worker request timed out: replayTerminal";
-        expect(viewers.ended(phone)[0]?.message).toMatchObject({ payload: { error } });
-        viewers.finish(); // A late sink completion must not acknowledge another request.
-        await viewers.attach(concurrent, "fresh");
-        await waitForCondition(() => viewers.replayed(concurrent), 10000);
-      }
-      session.send({ type: "input", data: "AFTER-REPLAY\r" });
-      await waitForCondition(() => viewers.text(desktop).includes("ECHO:AFTER-REPLAY"), 10000);
-      expect(viewers.ended(desktop)).toHaveLength(0);
-      expect(manager.getTerminal(session.id)).toBe(session);
+      await new Promise((resolve) => setTimeout(resolve, 1100));
+      expect(viewers.ended(phone)).toHaveLength(0);
+      expect((await manager.captureTerminal(session.id)).lines.join("")).toContain("READY");
+      session.send({ type: "input", data: "AFTER-DEADLINE\r" });
+      await waitForCondition(() => viewers.text(phone).includes("ECHO:AFTER-DEADLINE"), 3000);
     } finally {
       viewers.finish();
       await viewers.delivery.close();
     }
   },
-  25000,
+  20000,
 );
 
-it("cancels a replay on controller replacement and ignores its late sink completion", async () => {
-  const paseoHome = createTemporaryHome();
-  manager = await startRealWorkerManager({ paseoHome });
-  const session = trackTerminal(
-    await manager.createTerminal({
-      cwd: paseoHome,
-      workspaceId: "replay-replacement",
-      ...nodeTerminalCommand('process.stdout.write("READY\\r\\n"); setInterval(() => {}, 1000);'),
-    }),
-  );
-  await waitForCondition(
-    async () =>
-      (await manager!.captureTerminal(session.id)).lines.some((line) => line.includes("READY")),
-    10000,
-  );
-  let releaseOld!: () => void;
-  const oldGate = new Promise<void>((resolve) => {
-    releaseOld = resolve;
-  });
-  let began = false;
-  const oldReplay = manager.replayTerminal(session.id, undefined, async (event) => {
-    if (event.type === "begin") {
-      began = true;
-      await oldGate;
-    }
-  });
-  const failure = oldReplay.catch((error) => error as Error);
-  let releaseNew!: () => void;
-  const newGate = new Promise<void>((resolve) => {
-    releaseNew = resolve;
-  });
-  try {
-    await waitForCondition(() => began, 10000);
-    const [record] = await workerRecords(paseoHome);
-    const replacement = await connectTerminalWorker({
-      socketPath: record!.socketPath,
-      secret: record!.secret,
-    });
-    await waitForCondition(() => !replacement.connected, 10000);
-    expect(await failure).toBeInstanceOf(Error);
-    let newBegan = false;
-    let ended = false;
-    const freshReplay = manager.replayTerminal(session.id, undefined, async (event) => {
-      if (event.type === "begin") {
-        newBegan = true;
-        await newGate;
+it.each(["replacement", "socket-overflow"] as const)(
+  "recovers an attached restore viewer after worker %s",
+  async (loss) => {
+    const paseoHome = createTemporaryHome();
+    const drained = join(paseoHome, "burst-drained");
+    manager = await startRealWorkerManager({ paseoHome });
+    const session = trackTerminal(
+      await manager.createTerminal({
+        cwd: paseoHome,
+        workspaceId: "attached-reconnect",
+        ...nodeTerminalCommand(
+          `process.stdout.write("RETAINED\\r\\n"); process.stdin.on("data", data => {
+            if (data.toString().includes("BURST")) process.stdout.write("x".repeat(12 * 1024 * 1024) + "\\r\\nRETAINED-AFTER-LOSS\\r\\n", () => require("node:fs").writeFileSync(${JSON.stringify(drained)}, "done"));
+            else process.stdout.write("ECHO:" + data);
+          }); setInterval(() => {}, 1000);`,
+        ),
+      }),
+    );
+    await waitForCondition(
+      async () => (await manager!.captureTerminal(session.id)).lines.join("").includes("RETAINED"),
+      10000,
+    );
+    const viewers = terminalViewers(manager, session.id);
+    const desktop = {};
+    try {
+      await viewers.attach(desktop, "desktop");
+      await waitForCondition(() => viewers.replayed(desktop), 10000);
+      viewers.frames.length = 0;
+      const [record] = await workerRecords(paseoHome);
+      if (loss === "replacement") {
+        const competitor = await connectTerminalWorker({
+          socketPath: record!.socketPath,
+          secret: record!.secret,
+        });
+        await waitForCondition(() => !competitor.connected, 10000);
+      } else {
+        session.send({ type: "input", data: "BURST\r" });
+        // Stall only the daemon event loop. The detached worker and PTY keep running,
+        // filling the bounded worker socket and forcing its disconnect/re-adoption path.
+        const stall = new Int32Array(new SharedArrayBuffer(4));
+        const deadline = Date.now() + 10000;
+        while (!existsSync(drained) && Date.now() < deadline) Atomics.wait(stall, 0, 0, 25);
+        expect(existsSync(drained)).toBe(true);
       }
-      if (event.type === "end") ended = true;
-    });
-    await waitForCondition(() => newBegan, 10000);
-    releaseOld();
-    // A real worker request crosses the replacement socket while its replay remains blocked.
-    expect((await manager.captureTerminal(session.id)).lines.join("")).toContain("READY");
-    expect(ended).toBe(false);
-    releaseNew();
-    expect(await freshReplay).not.toBeNull();
-    expect(ended).toBe(true);
-    expect(manager.getTerminal(session.id)).toBe(session);
-  } finally {
-    releaseOld();
-    releaseNew();
-  }
-}, 20000);
+      await waitForCondition(() => viewers.replayed(desktop), 10000);
+      const replay = viewers.frames
+        .filter((row) => row.frame.opcode === TerminalStreamOpcode.ReplayData)
+        .map((row) => Buffer.from(row.frame.payload).toString("utf8"))
+        .join("");
+      expect(replay).toContain("RETAINED");
+      if (loss === "socket-overflow") expect(replay).toContain("RETAINED-AFTER-LOSS");
+      expect((await workerRecords(paseoHome)).map((worker) => worker.pid)).toEqual([record!.pid]);
+      session.send({ type: "input", data: "RECOVERED\r" });
+      await waitForCondition(() => viewers.text(desktop).includes("ECHO:RECOVERED"), 3000);
+      expect(viewers.ended(desktop)).toHaveLength(0);
+      expect(manager.getTerminal(session.id)).toBe(session);
+    } finally {
+      await viewers.delivery.close();
+    }
+  },
+  30000,
+);

packages/server/src/terminal/worker-terminal-manager.ts

@@ -600,27 +600,18 @@ export function createWorkerTerminalManager(
     const connection = worker;
     connection.on("message", (message: TerminalWorkerToParentMessage) => {
       if (message.type === "terminalReplay") {
-        const sink = replaySinks.get(message.requestId);
-        void Promise.resolve()
-          .then(() => sink?.(message.event))
-          .then(
-            () =>
-              pendingRequests.has(message.requestId) &&
-              connection.send(
-                { type: "replayAcknowledged", requestId: message.requestId },
-                () => {},
-              ),
-            (error: unknown) =>
-              pendingRequests.has(message.requestId) &&
-              connection.send(
-                {
-                  type: "replayAcknowledged",
-                  requestId: message.requestId,
-                  error: error instanceof Error ? error.message : String(error),
-                },
-                () => {},
-              ),
-          );
+        // Consumption queues into the application's bounded transport synchronously.
+        // Never wait for app network delivery or accumulate a second replay queue here.
+        if (worker !== connection || !pendingRequests.has(message.requestId)) return;
+        try {
+          replaySinks.get(message.requestId)?.(message.event);
+        } catch (error) {
+          const pending = pendingRequests.get(message.requestId)!;
+          clearTimeout(pending.timeout);
+          pendingRequests.delete(message.requestId);
+          replaySinks.delete(message.requestId);
+          pending.reject(error instanceof Error ? error : new Error(String(error)));
+        }
         return;
       }
       if (isResponse(message)) {
@@ -701,11 +692,6 @@ export function createWorkerTerminalManager(
     const request = new Promise((resolve, reject) => {
       const timeout = setTimeout(() => {
         pendingRequests.delete(requestId);
-        if (sink)
-          connection.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 });