fix(terminal): isolate replay failures to each viewer

A terminal-wide replay tail paused healthy viewers and discarded their live output when another viewer failed or timed out. Keep buffering and revision deduplication with each daemon stream, and pin acknowledgements to their original worker connection.

docs/terminal-performance.md

@@ -22,6 +22,7 @@ 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.
 - **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,15 +40,15 @@ 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 3 hard cuts unreleased protocols 1 (cell transport) and 2 (socket-paced replay without consumption acknowledgements): 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 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.
 
 Tests stand up other versions with `PASEO_TERMINAL_WORKER_PROTOCOL_OVERRIDE`, read by the worker. It is test-only.
 
 ## Display replay
 
-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 before later live output. 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.
+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. The replaying terminal holds at most 4 MiB of live output until replay ends; exceeding that bound fails the replay and ends every display stream of that terminal with an error requiring a fresh attach. Existing viewers share the worker live stream, so they must also stop before bytes resume after a discarded tail. Sibling terminals continue streaming. 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.
+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.
 
 ## Running from its own copy
 

packages/server/src/server/agent/mcp-server.test.ts

@@ -1493,7 +1493,6 @@ describe("terminal MCP tools", () => {
         id: "term-1",
         name: "daemon",
         cwd: process.cwd(),
-        getState: vi.fn().mockReturnValue({ scrollback: [], grid: [[]] }),
       }),
       captureTerminal,
     });

packages/server/src/server/daemon-e2e/terminal.e2e.test.ts

@@ -1297,7 +1297,7 @@ test("resize updates server dimensions without sending a live snapshot", async (
 
   await new Promise((resolve) => setTimeout(resolve, 300));
 
-  const state = ctx.daemon.daemon.terminalManager?.getTerminal(terminalId)?.getState();
+  const state = ctx.daemon.daemon.terminalManager?.getTerminal(terminalId)?.getSize();
   expect(state?.rows).toBe(10);
   expect(state?.cols).toBe(40);
 
@@ -1314,7 +1314,7 @@ test("the latest resize from any attached client owns the PTY size", async () =>
   );
 
   const readSize = () => {
-    const state = ctx.daemon.daemon.terminalManager?.getTerminal(terminalId)?.getState();
+    const state = ctx.daemon.daemon.terminalManager?.getTerminal(terminalId)?.getSize();
     return state ? { rows: state.rows, cols: state.cols } : null;
   };
   const expectSize = async (size: { rows: number; cols: number }) => {

packages/server/src/server/script-health-monitor.test.ts

@@ -68,13 +68,7 @@ function createStubTerminalManager(
         subscribe: () => () => {},
         onExit: () => () => {},
         onCommandFinished: () => () => {},
-        getState: () => ({
-          rows: 1,
-          cols: 1,
-          grid: [[{ char: "$" }]],
-          scrollback: [],
-          cursor: { row: 0, col: 0 },
-        }),
+        hasOutput: () => true,
         kill: () => {},
         onTitleChange: () => () => {},
         getSize: () => ({ rows: 1, cols: 1 }),

packages/server/src/server/worktree-bootstrap.test.ts

@@ -210,13 +210,7 @@ describe("runWorktreeAutoTerminals", () => {
             getTitle: () => undefined,
             getExitInfo: () => null,
             hasOutput: () => false,
-            getState: () => ({
-              rows: 0,
-              cols: 0,
-              grid: [],
-              scrollback: [],
-              cursor: { row: 0, col: 0 },
-            }),
+
             kill: () => {},
             killAndWait: async () => {},
           };
@@ -314,13 +308,7 @@ describe("runWorktreeAutoTerminals", () => {
             };
           },
           hasOutput: () => true,
-          getState: () => ({
-            rows: 1,
-            cols: 1,
-            grid: [[{ char: "$" }]],
-            scrollback: [],
-            cursor: { row: 0, col: 0 },
-          }),
+
           kill: () => {},
           onTitleChange: () => () => {},
           onActivityChange: () => () => {},

packages/server/src/server/worktree-session.test.ts

@@ -212,13 +212,7 @@ function createTerminalManagerStub(options?: {
             id: `terminal-${terminals.length + 1}`,
             name: input.name ?? "Terminal",
             cwd: input.cwd,
-            getState: () => ({
-              rows: 1,
-              cols: 1,
-              scrollback: [[{ char: "$" }]],
-              grid: [],
-              cursor: { row: 0, col: 0 },
-            }),
+            hasOutput: () => true,
             subscribe: () => () => {},
             onExit: () => () => {},
             onCommandFinished: () => () => {},

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

@@ -64,50 +64,14 @@ interface TerminalReplayPorts {
   flushOutput(id: string): void;
 }
 
-/** Owns the worker's replay/live cut, bounded live tail and controller lifetime. */
+/** 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 }>();
-  // 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 {
-    readers: number;
-    bytes: number;
-    connection: FramedSocket;
-    output: TerminalWorkerToParentMessage[];
-    overflow: boolean;
-    failure: unknown;
-    completed: Promise<void>;
-    resolve: () => void;
-    reject: (error: unknown) => void;
-    exit?: Extract<TerminalWorkerToParentMessage, { type: "terminalExit" }>;
-  }
-  const replayTails = new Map<string, ReplayTail>();
-  const MAX_REPLAY_TAIL_BYTES = 4 * 1024 * 1024;
 
   function sendToController(message: TerminalWorkerToParentMessage): void {
-    if (
-      message.type === "terminalExit" ||
-      (message.type === "terminalMessage" && message.message.type === "output")
-    ) {
-      const tail = replayTails.get(message.terminalId);
-      if (tail) {
-        if (message.type === "terminalExit") {
-          tail.exit = message;
-          return;
-        }
-        if (!tail.overflow && message.message.type === "output") {
-          tail.bytes += Buffer.byteLength(message.message.data);
-          if (tail.bytes > MAX_REPLAY_TAIL_BYTES) {
-            tail.overflow = true;
-            tail.output = [];
-          } else tail.output.push(message);
-        }
-        return;
-      }
-    }
-    // With no controller the event is dropped; the headless terminal keeps the screen, and the
-    // next daemon reads it through listTerminals.
+    // A stalled physical controller is different from a stalled viewer: all its
+    // observations must reconnect to the surviving worker after this socket closes.
     if (controller && controller.socket.writableLength > 4 * 1024 * 1024) {
       controller.socket.destroy();
       return;
@@ -116,10 +80,9 @@ export function createTerminalReplayController(ports: TerminalReplayPorts) {
   }
 
   async function deliverReplay(
+    connection: FramedSocket,
     message: Extract<TerminalWorkerRequest, { type: "replayTerminal" }>,
   ): Promise<void> {
-    const connection = controller;
-    if (!connection) return;
     ports.flushOutput(message.terminalId);
     const session = ports.getSession(message.terminalId);
     if (!session) {
@@ -131,23 +94,17 @@ export function createTerminalReplayController(ports: TerminalReplayPorts) {
       });
       return;
     }
-    const existingTail = replayTails.get(session.id);
-    const tail =
-      existingTail?.connection === connection ? existingTail : createReplayTail(connection);
-    tail.readers++;
-    replayTails.set(session.id, tail);
-    let released = false;
+    // Flush and capture share one synchronous turn. Each daemon viewer buffers
+    // its own live tail and discards revisions already included in this cut.
+    const replay = createTerminalReplay(session, message.options);
+    let next = replay.next();
     try {
-      const replay = createTerminalReplay(session, message.options);
-      let next = replay.next();
       while (!next.done) {
-        if (tail.overflow)
-          throw new Error("Terminal output exceeded the replay buffer; attach again");
+        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(() => {});
-        // Register before sending: even a synchronous controller cannot race the acknowledgement.
         await connection.sendAndWait({
           type: "terminalReplay",
           requestId: message.requestId,
@@ -157,88 +114,17 @@ export function createTerminalReplayController(ports: TerminalReplayPorts) {
         acknowledgements.delete(message.requestId);
         next = replay.next();
       }
-      released = true;
-      await releaseReplayTail(session, tail);
       await connection.sendAndWait({
         type: "response",
         requestId: message.requestId,
         ok: true,
         result: next.value,
       });
-    } catch (error) {
-      tail.failure = error;
-      if (!released) await releaseReplayTail(session, tail);
-      throw error;
     } finally {
       acknowledgements.delete(message.requestId);
     }
   }
 
-  function createReplayTail(connection: FramedSocket): ReplayTail {
-    let resolve!: () => void;
-    let reject!: (error: unknown) => void;
-    const completed = new Promise<void>((onResolve, onReject) => {
-      resolve = onResolve;
-      reject = onReject;
-    });
-    // A write can fail before every reader reaches the shared completion barrier.
-    void completed.catch(() => {});
-    return {
-      connection,
-      readers: 0,
-      bytes: 0,
-      output: [],
-      overflow: false,
-      failure: null,
-      completed,
-      resolve,
-      reject,
-    };
-  }
-
-  async function releaseReplayTail(session: LocalTerminalSession, tail: ReplayTail): Promise<void> {
-    if (--tail.readers === 0) {
-      try {
-        while (!tail.failure && !tail.overflow && tail.readers === 0 && tail.output.length > 0) {
-          if (controller !== tail.connection) throw new Error("Terminal worker controller changed");
-          const output = tail.output.shift()!;
-          if (output.type === "terminalMessage" && output.message.type === "output")
-            tail.bytes -= Buffer.byteLength(output.message.data);
-          await tail.connection.sendAndWait(output);
-        }
-      } catch (error) {
-        tail.failure = error;
-      } finally {
-        finishReplayTail(session, tail);
-      }
-    }
-    await tail.completed;
-  }
-
-  function finishReplayTail(session: LocalTerminalSession, tail: ReplayTail): void {
-    if (tail.readers !== 0) return;
-    if (replayTails.get(session.id) === tail) replayTails.delete(session.id);
-    if ((tail.overflow || tail.failure) && controller === tail.connection) {
-      // The live stream is shared by every viewer of this terminal. Once its tail
-      // is discarded none of those viewers may continue from the old screen.
-      sendToController({
-        type: "terminalMessage",
-        terminalId: session.id,
-        message: {
-          type: "streamError",
-          error: "Terminal output exceeded the replay buffer; attach again",
-        },
-        replayPreamble: session.getReplayPreamble(),
-      });
-    }
-    if (tail.exit && controller === tail.connection) sendToController(tail.exit);
-    if (tail.failure || tail.overflow)
-      tail.reject(
-        tail.failure ?? new Error("Terminal output exceeded the replay buffer; attach again"),
-      );
-    else tail.resolve();
-  }
-
   return {
     acknowledge(requestId: string, error?: string): void {
       const pending = acknowledgements.get(requestId);
@@ -246,12 +132,10 @@ export function createTerminalReplayController(ports: TerminalReplayPorts) {
       else pending?.resolve();
     },
     setConnection(connection: FramedSocket | null): void {
+      if (controller === connection) return;
       const previous = controller;
       controller = connection;
-      if (previous !== connection) previous?.socket.destroy();
-      // Old writes settle on their original socket; new output and replay belong
-      // exclusively to the replacement controller.
-      replayTails.clear();
+      previous?.socket.destroy();
       for (const pending of acknowledgements.values())
         pending.reject(new Error("Terminal replay controller disconnected"));
       acknowledgements.clear();
@@ -263,7 +147,7 @@ export function createTerminalReplayController(ports: TerminalReplayPorts) {
       const connection = controller;
       if (!connection) return;
       try {
-        await deliverReplay(request);
+        await deliverReplay(connection, request);
       } catch (error) {
         await connection
           .sendAndWait({

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

@@ -1253,45 +1253,3 @@ test.each([false, true])(
     }
   },
 );
-
-test.each([true, false])(
-  "ends every viewer of an interrupted terminal and permits a fresh attach (modern=%s)",
-  async (modern) => {
-    const f = exitFixture(async () => terminalReplay("current screen", 2));
-    const source = {};
-    const owners = [
-      await f.subscribe(source, modern, "existing"),
-      await f.subscribe(modern ? source : {}, modern, "joining"),
-    ];
-    await waitForCoalescerFlush();
-    f.frames.length = 0;
-    const error = "Terminal output exceeded the replay buffer; attach again";
-    for (const listener of Array.from(f.outputs)) listener({ type: "streamError", error });
-    await flushMicrotasks();
-    expect(f.frames.map((row) => row.message?.type)).toEqual([
-      "terminal_stream_exit",
-      "terminal_stream_exit",
-    ]);
-    for (const owner of owners)
-      expect(f.frames).toContainEqual({
-        source: owner.source,
-        message: {
-          type: "terminal_stream_exit",
-          payload: {
-            terminalId: "exit-terminal",
-            error,
-            ...(modern ? { subscriptionId: owner.id } : {}),
-          },
-        },
-      });
-    expect(f.outputs.size).toBe(0);
-    expect(f.exits.size).toBe(0);
-    f.output("must not render against stale screen", 3);
-    expect(f.frames.length).toBe(2);
-    await f.subscribe({}, modern, "recovered");
-    await waitForCoalescerFlush();
-    const restores = f.frames.flatMap((row) => (row.binary ? [row.binary] : []));
-    expect(restores.map((frame) => frame.opcode)).toEqual([TerminalStreamOpcode.Restore]);
-    expect(new TextDecoder().decode(restores[0]!.payload)).toBe("current screen");
-  },
-);

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

@@ -952,14 +952,6 @@ export class TerminalSessionController {
         if (this.activeStreams.get(slot) !== activeStream) {
           return;
         }
-        if (message.type === "streamError") {
-          activeStream.owner.emit({
-            type: "terminal_stream_exit",
-            payload: { terminalId: activeStream.terminalId, error: message.error },
-          });
-          void activeStream.owner.release();
-          return;
-        }
         if (message.type === "snapshot" || message.type === "snapshotReady") {
           activeStream.readyRevision = message.revision;
           activeStream.outputCoalescer.flush();

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

@@ -5,19 +5,20 @@ describe("classifyTerminalWorkerProtocol", () => {
   it("rejects the unreleased cell protocol without a draining adapter", () => {
     expect(classifyTerminalWorkerProtocol(1, 2)).toBe("obsolete");
     expect(classifyTerminalWorkerProtocol(2, 3)).toBe("obsolete");
+    expect(classifyTerminalWorkerProtocol(3, 4)).toBe("obsolete");
   });
   it.each([
-    [5, "current"],
-    [4, "previous"],
-    [3, "obsolete"],
+    [6, "current"],
+    [5, "previous"],
+    [4, "obsolete"],
     [-1, "obsolete"],
-    [6, "newer"],
+    [7, "newer"],
     // A version this daemon cannot read comes from a Paseo it does not know.
     ["6", "newer"],
     [4.5, "newer"],
     [Number.NaN, "newer"],
     [undefined, "newer"],
-  ])("a worker reporting %s to a daemon at 5 is %s", (reported, fit) => {
-    expect(classifyTerminalWorkerProtocol(reported, 5)).toBe(fit);
+  ])("a worker reporting %s to a daemon at 6 is %s", (reported, fit) => {
+    expect(classifyTerminalWorkerProtocol(reported, 6)).toBe(fit);
   });
 });

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 = 3;
+export const TERMINAL_WORKER_PROTOCOL_VERSION = 4;
 
 // 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 and 2 were unreleased; they have no draining adapter.
-  if (reported === current - 1 && reported >= 3) return "previous";
+  // Protocols 1, 2 and 3 were unreleased; they have no draining adapter.
+  if (reported === current - 1 && reported >= 4) return "previous";
   return reported < current ? "obsolete" : "newer";
 }
 

packages/server/src/terminal/terminal.ts

@@ -92,7 +92,6 @@ export type ClientMessage =
   | { type: "mouse"; row: number; col: number; button: number; action: "down" | "up" | "move" };
 
 export type ServerMessage =
-  | { type: "streamError"; error: string }
   | { type: "output"; data: string; revision?: number }
   | { type: "snapshot"; state: TerminalState; revision?: number }
   | { type: "snapshotReady"; revision?: number; replayPreamble?: string }

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

@@ -1,3 +1,11 @@
+import { TerminalSessionController } from "./terminal-session-controller.js";
+import { SessionDelivery } from "../server/session/owned-subscriptions/index.js";
+import {
+  decodeTerminalStreamFrame,
+  TerminalStreamOpcode,
+  type TerminalStreamFrame,
+} from "@getpaseo/protocol/binary-frames/index";
+import type { SessionOutboundMessage } from "../server/messages.js";
 import { createConnection } from "node:net";
 import { attachFramedSocket } from "./terminal-worker-frames.js";
 import xterm from "@xterm/headless";
@@ -174,7 +182,11 @@ const temporaryDirs: string[] = [];
 
 // A real detached worker in a throwaway home; afterEach closes its terminals and the worker.
 async function startRealWorkerManager(
-  options: { getTerminalActivityUrl?: () => string | null; paseoHome?: string } = {},
+  options: {
+    getTerminalActivityUrl?: () => string | null;
+    requestTimeoutMs?: number;
+    paseoHome?: string;
+  } = {},
 ): Promise<DaemonTerminalManager> {
   const { paseoHome = createTemporaryHome(), ...rest } = options;
   return startWorkerTerminalManager({
@@ -1570,18 +1582,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(3);
+  expect(workers[0]!.protocolVersion).toBe(4);
 });
 
 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, 4);
+  const newer = await leaveWorkerWithTerminal(paseoHome, 5);
 
   manager = await startRealWorkerManager({ paseoHome });
 
   const unavailable = manager.getUnavailableError?.();
   expect(unavailable).toBeInstanceOf(TerminalWorkerIncompatibleError);
-  expect(unavailable?.message).toContain("worker protocol 4");
+  expect(unavailable?.message).toContain("worker protocol 5");
   await expect(manager.getTerminals(newer.cwd)).rejects.toBeInstanceOf(
     TerminalWorkerIncompatibleError,
   );
@@ -1730,7 +1742,8 @@ it("replays bounded ANSI chunks before concurrent live output and reconstructs r
   await waitForCondition(() => live.includes("LIVE-AFTER-REPLAY"), 10000);
   unsubscribe();
   expect(events[0]).toBe("begin");
-  expect(events.indexOf("live")).toBeGreaterThan(events.indexOf("end"));
+  expect(events.lastIndexOf("end")).toBeGreaterThan(events.lastIndexOf("data"));
+  expect(live.match(/LIVE-AFTER-REPLAY/g)).toHaveLength(1);
   expect(chunks.length).toBeGreaterThan(1);
   expect(chunks.every((chunk) => chunk.byteLength <= 64 * 1024)).toBe(true);
   expect(metadata?.rows).toBe(8);
@@ -1857,7 +1870,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: 3 });
+    framed.send({ type: "hello", secret: record!.secret, protocolVersion: 4 });
     await waitForCondition(() => began, 10000);
     framed.send({
       type: "send",
@@ -1884,7 +1897,7 @@ it("completes a replay larger than the socket budget while a sibling stays usabl
     expect(resumedBytes).toBe(bytes);
     await waitForCondition(() => existingLiveOutput.includes("SAME-TERMINAL-LIVE"), 10000);
     expect(existingLiveOutput.match(/SAME-TERMINAL-LIVE/g)).toHaveLength(1);
-    expect(replayOrder.indexOf("live")).toBeGreaterThan(replayOrder.indexOf("end"));
+    expect(replayOrder.indexOf("live")).toBeLessThan(replayOrder.indexOf("end"));
   } finally {
     socket.destroy();
   }
@@ -1892,58 +1905,6 @@ it("completes a replay larger than the socket budget while a sibling stays usabl
   expect(
     (await manager.captureTerminal(terminal.id)).lines.some((line) => line.includes("READY-WIDE")),
   ).toBe(true);
-  await manager.detach();
-  const stalled = createConnection(record!.socketPath);
-  let overflowBegan = false;
-  let overflowError: string | null = null;
-  const existingViewerEvents: string[] = [];
-  const stalledFrames = attachFramedSocket(stalled, (value) => {
-    const message = value as TerminalWorkerToParentMessage | { type: "helloAck" };
-    if (message.type === "terminalMessage" && message.terminalId === terminal.id)
-      existingViewerEvents.push(message.message.type);
-    if (message.type === "helloAck")
-      stalledFrames.send({
-        type: "replayTerminal",
-        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();
-    }
-    if (message.type === "response" && message.requestId === "overflow-replay" && !message.ok)
-      overflowError = message.error;
-  });
-  try {
-    stalledFrames.send({ type: "hello", secret: record!.secret, protocolVersion: 3 });
-    await waitForCondition(() => overflowBegan, 10000);
-    stalledFrames.send({
-      type: "send",
-      terminalId: terminal.id,
-      requestId: "produce-overflow",
-      message: { type: "input", data: "burst\r" },
-    });
-    await waitForCondition(() => existsSync(overflowProgress), 10000);
-    expect(stalled.destroyed).toBe(false);
-    stalled.resume();
-    await waitForCondition(() => overflowError !== null, 10000);
-    expect(overflowError).toBe("Terminal output exceeded the replay buffer; attach again");
-    expect(existingViewerEvents).toContain("streamError");
-  } finally {
-    stalled.destroy();
-  }
-  manager = await startRealWorkerManager({ paseoHome });
-  const recovered = await manager.replayTerminal(terminal.id, { scrollbackLines: 200 }, () => {});
-  expect(recovered!.cols).toBe(1024);
-  expect(
-    (await manager.captureTerminal(terminal.id)).lines.some((line) =>
-      line.includes("RECOVERED-LIVE"),
-    ),
-  ).toBe(true);
-  expect(manager.getTerminal(sibling.id)?.id).toBe(sibling.id);
-  expect(recovered!.retainedHistoryLines).toBe(1000);
   const exitingReplay = await manager.replayTerminal(
     terminal.id,
     { scrollbackLines: 200 },
@@ -1955,3 +1916,260 @@ it("completes a replay larger than the socket budget while a sibling stays usabl
   await waitForCondition(() => manager!.getTerminal(terminal.id) === undefined, 10000);
   expect(manager.getTerminal(sibling.id)?.id).toBe(sibling.id);
 }, 60000);
+
+function terminalViewers(terminalManager: DaemonTerminalManager, terminalId: string) {
+  const frames: Array<{ source: object; frame: TerminalStreamFrame }> = [];
+  const messages: Array<{ source: object; message: SessionOutboundMessage }> = [];
+  let blockedSource: object | null = null;
+  let started = false;
+  let finish!: (error?: Error) => void;
+  const block = new Promise<void>((resolve, reject) => {
+    finish = (error) => (error ? reject(error) : resolve());
+  });
+  const delivery = new SessionDelivery(
+    (source, message) => {
+      messages.push({ source, message });
+    },
+    async (source, bytes) => {
+      const frame = decodeTerminalStreamFrame(bytes)!;
+      frames.push({ source, frame });
+      if (source === blockedSource && frame.opcode === TerminalStreamOpcode.ReplayBegin) {
+        started = true;
+        await block;
+      }
+    },
+  );
+  const controller = new TerminalSessionController({
+    terminalManager,
+    emit: (message) => {
+      delivery.reply(message);
+    },
+    hasBinaryChannel: () => true,
+    supportsChunkedReplay: () => true,
+    isPathWithinRoot: () => true,
+    sessionLogger: pino({ level: "silent" }),
+  });
+  return {
+    frames,
+    messages,
+    delivery,
+    controller,
+    block(source: object) {
+      blockedSource = source;
+    },
+    started: () => started,
+    finish,
+    async attach(source: object, requestId: string) {
+      delivery.attach(source, true);
+      const request = {
+        type: "subscribe_terminal_request" as const,
+        terminalId,
+        requestId,
+        restore: { mode: "visible-snapshot" as const, scrollbackLines: 1000 },
+      };
+      await delivery.request(source, request, () => controller.dispatch(request, delivery));
+    },
+    text(source: object) {
+      return frames
+        .filter((row) => row.source === source && row.frame.opcode === TerminalStreamOpcode.Output)
+        .map((row) => Buffer.from(row.frame.payload).toString("utf8"))
+        .join("");
+    },
+    ended(source: object) {
+      return messages.filter(
+        (row) => row.source === source && row.message.type === "terminal_stream_exit",
+      );
+    },
+    replayed(source: object) {
+      return frames.some(
+        (row) => row.source === source && row.frame.opcode === TerminalStreamOpcode.ReplayEnd,
+      );
+    },
+  };
+}
+
+it("isolates a rejected replay sink while another viewer of the same terminal stays live", async () => {
+  manager = await startRealWorkerManager();
+  const session = trackTerminal(
+    await manager.createTerminal({
+      cwd: createTemporaryHome(),
+      workspaceId: "two-viewers",
+      ...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")),
+    10000,
+  );
+  const viewers = terminalViewers(manager, session.id);
+  const desktop = {},
+    phone = {},
+    fresh = {};
+  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);
+    session.send({ type: "input", data: "BEFORE-FAILURE\r" });
+    await waitForCondition(() => viewers.text(desktop).includes("ECHO:BEFORE-FAILURE"), 1500);
+    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" },
+    });
+    session.send({ type: "input", data: "AFTER-FAILURE\r" });
+    await waitForCondition(() => viewers.text(desktop).includes("ECHO:AFTER-FAILURE"), 10000);
+    expect(viewers.ended(desktop)).toHaveLength(0);
+    await viewers.attach(fresh, "fresh");
+    await waitForCondition(() => viewers.replayed(fresh), 10000);
+    expect(manager.getTerminal(session.id)).toBe(session);
+  } finally {
+    viewers.finish();
+    await viewers.delivery.close();
+  }
+}, 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,
+    });
+    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);
+      `),
+      }),
+    );
+    await waitForCondition(
+      async () =>
+        (await manager!.captureTerminal(session.id)).lines.some((line) => line.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);
+      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);
+    } finally {
+      viewers.finish();
+      await viewers.delivery.close();
+    }
+  },
+  25000,
+);
+
+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;
+      }
+      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);

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

@@ -597,16 +597,22 @@ export function createWorkerTerminalManager(
   }
 
   function attachWorker(): void {
-    worker.on("message", (message: TerminalWorkerToParentMessage) => {
+    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(
             () =>
-              worker.send({ type: "replayAcknowledged", requestId: message.requestId }, () => {}),
+              pendingRequests.has(message.requestId) &&
+              connection.send(
+                { type: "replayAcknowledged", requestId: message.requestId },
+                () => {},
+              ),
             (error: unknown) =>
-              worker.send(
+              pendingRequests.has(message.requestId) &&
+              connection.send(
                 {
                   type: "replayAcknowledged",
                   requestId: message.requestId,
@@ -634,14 +640,15 @@ export function createWorkerTerminalManager(
       handleWorkerEvent(message);
     });
 
-    worker.on("close", () => {
+    connection.on("close", () => {
+      if (worker !== connection) return;
       workerDisconnected = true;
       rejectPendingRequests(new Error("Terminal worker connection closed"));
       const reconnect = managerOptions.reconnect;
       if (!detached && reconnect) {
         reconnecting = (async () => {
-          const connection = await reconnect();
-          if (!connection) {
+          const recovered = await reconnect();
+          if (!recovered) {
             for (const terminalId of Array.from(recordsById.keys())) {
               handleTerminalExitEvent({
                 type: "terminalExit",
@@ -652,10 +659,10 @@ export function createWorkerTerminalManager(
             return;
           }
           if (detached) {
-            connection.disconnect();
+            recovered.disconnect();
             return;
           }
-          worker = connection;
+          worker = recovered;
           workerDisconnected = false;
           attachWorker();
           await adoptTerminals();
@@ -687,6 +694,7 @@ export function createWorkerTerminalManager(
     if (workerDisconnected || !worker.connected) {
       return Promise.reject(new Error("Terminal worker is not running"));
     }
+    const connection = worker;
     const requestId = randomUUID();
     if (sink) replaySinks.set(requestId, sink);
     const message = { ...input, requestId } as TerminalWorkerRequest;
@@ -694,14 +702,14 @@ export function createWorkerTerminalManager(
       const timeout = setTimeout(() => {
         pendingRequests.delete(requestId);
         if (sink)
-          worker.send(
+          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 });
-      worker.send(message, (error) => {
+      connection.send(message, (error) => {
         if (!error) {
           return;
         }