fix(terminal): preserve current modes and default ANSI restores

Keep capture-time input modes local to the restoring viewer so replay completion cannot rewind newer shared modes. Publish initial modes after terminal creation and request explicit full restores for default SDK observations on ANSI hosts.

Derive worker test ownership from terminal records and apply the short replay deadline only after program setup.

docs/protocol-compatibility.md

@@ -56,7 +56,9 @@ without requesting streams. Application caches and which agents are visible rema
 
 `terminal_chunked_replay` is SDK-owned delivery behavior. The daemon sends new binary opcodes only to a source
 advertising the capability. The SDK keeps the established subscribe request against older
-ANSI hosts and normalizes their single Restore frame into begin/data/end. JSON snapshot
+ANSI hosts and normalizes their single Restore frame into begin/data/end. `observeTerminal` defaults
+to an explicit full snapshot: omitting restore on an ANSI-era host selects its discarded JSON path.
+Explicit caller restore modes and history counts pass through unchanged. JSON snapshot
 wire declarations remain accepted, but JSON snapshot frames have no rendering runtime.
 Terminals require `terminal-restore-modes`; older hosts show an update message without attaching.
 

docs/terminal-performance.md

@@ -25,7 +25,7 @@ Terminal frames share the daemon main event loop with all agent traffic. The `ev
 - **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.
+- **The input-mode tracker runs once per process boundary, not per hop.** The worker owns the authoritative tracker; the daemon caches current modes from ordered worker events and adoption. A replay response carries capture-time modes for its viewer; it must not overwrite the shared cache after newer live output. Apply that viewer's preamble before its buffered live bytes. 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.
 - **Plugin daemon sessions report IPC queue bytes.** Their virtual socket increments `bufferedAmount` before `process.send` and decrements it only in the send callback. Text and binary frames share that ordered queue, so the normal snapshot catch-up and physical high-water gates remain valid for server-side plugin SDK traffic.
 - **Client output writes are not serialized per frame.** The emulator runtime drains contiguous plain writes straight into xterm (which buffers internally). Only barrier ops (`clear`, `snapshot`, `suppressInput` writes) wait — behind a zero-length sentinel write — so resets can't interleave with in-flight output.

packages/client/src/daemon-client.test.ts

@@ -8486,3 +8486,89 @@ test("closes before admission when a protected-capable daemon is reached in lega
   expect(daemonReceived).toHaveLength(1);
   expect(JSON.parse(daemonReceived[0] as string)).toMatchObject({ type: "hello" });
 });
+
+test.each([false, true])(
+  "default terminal observation requests ANSI replay (chunked=%s)",
+  async (chunked) => {
+    const transport = createMockTransport();
+    const client = new DaemonClient({
+      url: "ws://test",
+      clientId: "clsk_unit_test",
+      logger: createMockLogger(),
+      reconnect: { enabled: false },
+      transportFactory: () => transport.transport,
+    });
+    clients.push(client);
+    const connecting = client.connect();
+    transport.triggerOpen({
+      features: { ownedSubscriptions: true, "terminal-restore-modes": true },
+    });
+    await connecting;
+    const events: import("./terminal-stream-router").TerminalStreamEvent[] = [];
+    const observation = client.observeTerminal("default-terminal", (event) => events.push(event));
+    await vi.waitFor(() =>
+      expect(parseSentFrame(transport.sent.at(-1)).type).toBe("subscribe_terminal_request"),
+    );
+    const request = parseSentFrame(transport.sent.at(-1));
+    expect(request.restore).toEqual({ mode: "full-snapshot" });
+    transport.triggerMessage(
+      wrapSessionMessage({
+        type: "subscribe_terminal_response",
+        payload: {
+          terminalId: "default-terminal",
+          requestId: request.requestId,
+          subscriptionId: "default-subscription",
+          slot: 2,
+          error: null,
+        },
+      }),
+    );
+    await observation.ready;
+    const data = new TextEncoder().encode("\x1b[31mretained 界🙂\x1b[0m");
+    const metadata = { rows: 24, cols: 80, retainedHistoryLines: 1000 };
+    if (chunked) {
+      transport.triggerMessage(
+        encodeTerminalStreamFrame({
+          opcode: TerminalStreamOpcode.ReplayBegin,
+          slot: 2,
+          payload: JSON.stringify(metadata),
+        }),
+      );
+      transport.triggerMessage(
+        encodeTerminalStreamFrame({
+          opcode: TerminalStreamOpcode.ReplayData,
+          slot: 2,
+          payload: data,
+        }),
+      );
+      transport.triggerMessage(
+        encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.ReplayEnd, slot: 2 }),
+      );
+    } else {
+      // An ANSI-era host selects its existing Restore path for this explicit request.
+      transport.triggerMessage(
+        encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Restore, slot: 2, payload: data }),
+      );
+    }
+    expect(events).toEqual([
+      {
+        terminalId: "default-terminal",
+        subscriptionId: "default-subscription",
+        type: "replay",
+        replay: chunked ? { type: "begin", ...metadata } : { type: "begin" },
+      },
+      {
+        terminalId: "default-terminal",
+        subscriptionId: "default-subscription",
+        type: "replay",
+        replay: { type: "data", data },
+      },
+      {
+        terminalId: "default-terminal",
+        subscriptionId: "default-subscription",
+        type: "replay",
+        replay: { type: "end" },
+      },
+    ]);
+  },
+);

packages/client/src/daemon-client.ts

@@ -6414,7 +6414,8 @@ export class DaemonClient {
   }
 
   /**
-   * Replay begin/data/end and live output arrive in receive. Subscribe to the returned handle
+   * Defaults to a full ANSI replay. Replay begin/data/end and live output arrive in receive.
+   * Subscribe to the returned handle
    * for terminal_stream_exit: payload.error means observation failure, not PTY exit.
    * Either outcome detaches this slot and releases the handle automatically.
    */
@@ -6433,7 +6434,7 @@ export class DaemonClient {
       {
         type: "subscribe_terminal_request",
         terminalId,
-        ...(options?.restore ? { restore: options.restore } : {}),
+        restore: options?.restore ?? { mode: "full-snapshot" },
       },
       options,
     );

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

@@ -1931,3 +1931,67 @@ test.each([400, 3000])(
   },
   30000,
 );
+
+test.each([false, true])(
+  "default SDK observation replays a terminal without waiting for output (quietScreen=%s)",
+  async (quietScreen) => {
+    const cwd = tmpCwd();
+    tempDirs.push(cwd);
+    const created = await createTerminalInWorkspace(ctx.client, {
+      cwd,
+      options: {
+        command: process.execPath,
+        args: [
+          "-e",
+          `${quietScreen ? 'process.stdout.write("QUIET-SCREEN\\r\\n");' : ""} process.stdin.on("data", data => process.stdout.write("LIVE:" + data)); setInterval(() => {}, 1000);`,
+        ],
+      },
+    });
+    const id = created.terminal!.id;
+    if (quietScreen)
+      await waitForCondition(
+        async () => (await ctx.client.captureTerminal(id)).lines.join("").includes("QUIET-SCREEN"),
+        10000,
+      );
+    const events: import("@getpaseo/client").TerminalStreamEvent[] = [];
+    const observation = ctx.client.observeTerminal(id, (event) => events.push(event));
+    try {
+      await observation.ready;
+      await waitForCondition(
+        () => events.some((event) => event.type === "replay" && event.replay.type === "end"),
+        10000,
+      );
+      const replay = events.flatMap((event) => (event.type === "replay" ? [event.replay] : []));
+      const begin = replay[0];
+      if (begin?.type !== "begin" || !begin.rows || !begin.cols)
+        throw new Error("Missing replay begin geometry");
+      expect(replay.at(-1)).toEqual({ type: "end" });
+      const terminal = new HeadlessTerminal({
+        rows: begin.rows,
+        cols: begin.cols,
+        allowProposedApi: true,
+      });
+      try {
+        for (const event of replay)
+          if (event.type === "data")
+            await new Promise<void>((resolve) => terminal.write(event.data, resolve));
+        const text = extractStateText(readHeadlessState(terminal));
+        expect(text).toBe(quietScreen ? "QUIET-SCREEN" : "");
+      } finally {
+        terminal.dispose();
+      }
+      ctx.client.sendTerminalInput(id, { type: "input", data: "next\r" });
+      await waitForCondition(
+        () =>
+          events.some(
+            (event) =>
+              event.type === "output" && new TextDecoder().decode(event.data).includes("LIVE:next"),
+          ),
+        10000,
+      );
+    } finally {
+      await observation.release();
+    }
+  },
+  30000,
+);

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

@@ -23,10 +23,10 @@ interface ReplayFixture {
   data: Uint8Array;
   metadata: TerminalReplayMetadata;
 }
-function terminalReplay(text: string, revision: number): ReplayFixture {
+function terminalReplay(text: string, revision: number, replayPreamble = ""): ReplayFixture {
   return {
     data: Buffer.from(text),
-    metadata: { rows: 1, cols: 80, retainedHistoryLines: 0, revision, replayPreamble: "" },
+    metadata: { rows: 1, cols: 80, retainedHistoryLines: 0, revision, replayPreamble },
   };
 }
 function replayFixture(
@@ -168,7 +168,7 @@ describe("terminal-session-controller restore", () => {
     expect(binaryFrames).toEqual([]);
 
     terminalListener?.({ type: "output", data: "restore-after\n", revision: 2 });
-    snapshot.resolve(terminalReplay("restore-before", 1));
+    snapshot.resolve(terminalReplay("restore-before", 1, "\x1b[?1h\x1b[?2004h"));
     await snapshot.promise;
     await new Promise((resolve) => setTimeout(resolve, 20));
 

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

@@ -34,6 +34,7 @@ import {
   resolveTerminalSubscriptionSnapshotMode,
   type TerminalRestoreOptions,
 } from "./terminal-restore.js";
+import type { TerminalReplayMetadata } from "./terminal-replay.js";
 import type { TerminalSession } from "./terminal.js";
 import type { TerminalManager, TerminalsChangedEvent } from "./terminal-manager.js";
 import { applyTerminalSize } from "./terminal-size-ownership.js";
@@ -76,10 +77,9 @@ interface ActiveTerminalStream {
   outputCoalescer: TerminalOutputCoalescer;
 }
 
-interface SnapshotSendResult {
-  shouldContinue: boolean;
-  replayRevision?: number;
-}
+type SnapshotSendResult =
+  | { shouldContinue: false }
+  | { shouldContinue: true; replay: TerminalReplayMetadata | null };
 
 export interface TerminalSessionControllerOptions {
   terminalManager: TerminalManager | null;
@@ -1041,7 +1041,7 @@ export class TerminalSessionController {
         return;
       }
       activeStream.snapshotOutput = undefined;
-      this.replayTerminalOutputAfterSnapshot(activeStream, terminal, snapshotResult.replayRevision);
+      this.replayTerminalOutputAfterSnapshot(activeStream, terminal, snapshotResult.replay);
       activeStream.needsSnapshot = false;
       activeStream.outputBytesSinceSnapshot = 0;
     } catch (error) {
@@ -1090,7 +1090,7 @@ export class TerminalSessionController {
         ? { scrollbackLines: restore.scrollbackLines }
         : resolveTerminalRestoreSnapshotOptions(restore);
     if (snapshotOptions === null) {
-      return { shouldContinue: true };
+      return { shouldContinue: true, replay: null };
     }
 
     const deliver = createTerminalRestoreDelivery({
@@ -1118,15 +1118,17 @@ export class TerminalSessionController {
     // The restore frame went out-of-band; keep the replay that follows on the
     // coalescer's trailing path so it doesn't flush back-to-back with it.
     activeStream.outputCoalescer.markFlushed();
-    return { shouldContinue: true, replayRevision: snapshot.revision };
+    return { shouldContinue: true, replay: snapshot };
   }
 
   private replayTerminalOutputAfterSnapshot(
     activeStream: ActiveTerminalStream,
     terminal: TerminalSession,
-    replayRevision: number | undefined,
+    replay: TerminalReplayMetadata | null,
   ): void {
-    const replayPreamble = terminal.getReplayPreamble();
+    // A captured viewer must start from its cut, then apply the buffered mode changes.
+    // Live-only subscriptions have no capture and start from the current shared modes.
+    const replayPreamble = replay === null ? terminal.getReplayPreamble() : replay.replayPreamble;
     if (replayPreamble.length > 0) {
       activeStream.outputCoalescer.handle(replayPreamble);
     }
@@ -1137,11 +1139,7 @@ export class TerminalSessionController {
       activeStream.bufferedOutputs.length,
     );
     for (const output of bufferedOutputs) {
-      if (
-        replayRevision !== undefined &&
-        output.revision !== undefined &&
-        output.revision <= replayRevision
-      ) {
+      if (replay !== null && output.revision !== undefined && output.revision <= replay.revision) {
         continue;
       }
       activeStream.outputCoalescer.handle(output.data);

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

@@ -475,6 +475,14 @@ function runWorker(argv: string[]): void {
         terminal: toTerminalInfo(session),
         state,
       });
+      // Startup output/readiness may have arrived while recording the program,
+      // before the daemon knew this terminal. Publish its current modes after creation.
+      sendToController({
+        type: "terminalMessage",
+        terminalId: session.id,
+        message: { type: "snapshotReady" },
+        replayPreamble: session.getReplayPreamble(),
+      });
       sendToController({
         type: "response",
         requestId: message.requestId,

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

@@ -1,3 +1,6 @@
+import { TerminalInputModeTracker } from "@getpaseo/protocol/terminal-input-mode";
+import { encodeTerminalKeyInput } from "@getpaseo/protocol/terminal-key-input";
+import type { SubscribeTerminalRequest } from "@getpaseo/protocol/messages";
 import { TerminalSessionController } from "./terminal-session-controller.js";
 import { SessionDelivery } from "../server/session/owned-subscriptions/index.js";
 import {
@@ -385,27 +388,36 @@ it.each(["current", "draining"] as const)(
   "keeps the peer usable when the %s worker dies",
   async (target) => {
     const paseoHome = createTemporaryHome();
-    await launchDetachedWorkerWithTerminal(paseoHome);
-    const old = await launchDetachedWorkerWithTerminal(paseoHome);
+    const seeds = [
+      await launchDetachedWorkerWithTerminal(paseoHome),
+      await launchDetachedWorkerWithTerminal(paseoHome),
+    ];
+    // Exercise an existing seed adopted as current; launch age does not imply draining ownership.
+    const [firstAdopted] = await workerRecords(paseoHome);
+    const seed = seeds.find((candidate) => candidate.workerPid === firstAdopted!.pid)!;
     manager = await startRealWorkerManager({ paseoHome });
     const added = await manager.createTerminal({
       workspaceId: "ws-peer",
-      cwd: old.cwd,
+      cwd: seed.cwd,
       ...nodeTerminalCommand("setInterval(() => {}, 1000)"),
     });
-    const current = (await workerRecords(paseoHome)).find(
-      (record) => record.pid !== old.workerPid,
+    const records = await workerRecords(paseoHome);
+    const current = records.find((record) =>
+      record.terminals.some((terminal) => terminal.id === added.id),
     )!;
-    const deadId = target === "current" ? added.id : old.terminalId;
-    const healthyId = target === "current" ? old.terminalId : added.id;
+    const draining = records.find((record) => record.pid !== current.pid)!;
+    const drainingId = draining.terminals[0]!.id;
+    const deadId = target === "current" ? added.id : drainingId;
+    const healthyId = target === "current" ? drainingId : added.id;
     const healthy = manager.getTerminal(healthyId);
-    process.kill(target === "current" ? current.pid : old.workerPid, "SIGKILL");
+    expect(healthy).toBeDefined();
+    process.kill(target === "current" ? current.pid : draining.pid, "SIGKILL");
     await waitForCondition(() => !manager!.getTerminal(deadId), 10000);
     expect(manager.getTerminal(healthyId)).toBe(healthy);
     expect(await readReplay(manager, healthyId)).not.toBeNull();
     await manager.createTerminal({
       workspaceId: "ws-peer",
-      cwd: old.cwd,
+      cwd: seed.cwd,
       ...nodeTerminalCommand("setInterval(() => {}, 1000)"),
     });
     expect((await workerRecords(paseoHome)).length).toBeLessThanOrEqual(2);
@@ -546,7 +558,7 @@ it("pulls fresh terminal state from the worker authority", async () => {
 // preamble-caching contract is verified on Linux/macOS; the daemon's input-mode
 // handling runs identically on every platform once the escape is observed.
 it.skipIf(isPlatform("win32"))(
-  "caches the input-mode replay preamble from the worker after getTerminalState",
+  "caches input modes from worker output independently of replay capture",
   async () => {
     const cwd = mkdtempSync(join(tmpdir(), "worker-terminal-manager-preamble-"));
     temporaryDirs.push(cwd);
@@ -564,11 +576,10 @@ it.skipIf(isPlatform("win32"))(
       }),
     );
 
-    await waitForCondition(async () => {
-      const snapshot = await readReplay(manager!, session.id);
-      return snapshot !== null && session.getReplayPreamble() === "\x1b[=1;1u";
-    }, 10000);
-
+    await waitForCondition(() => session.getReplayPreamble() === "\x1b[=1;1u", 10000);
+    // Initial shared modes must be ready without a replay response repairing the cache.
+    expect(session.getReplayPreamble()).toBe("\x1b[=1;1u");
+    expect(await readReplay(manager!, session.id)).not.toBeNull();
     expect(session.getReplayPreamble()).toBe("\x1b[=1;1u");
   },
 );
@@ -855,6 +866,7 @@ it("never adopts a live worker's terminals after a kill cleanup it could not fin
 // the shim; the PATH is restored as soon as it is up.
 async function startWorkerWithSlowProcessTable(
   delaySeconds: number,
+  options: { paseoHome?: string; requestTimeoutMs?: number } = {},
 ): Promise<DaemonTerminalManager> {
   const realPs = (process.env.PATH ?? "")
     .split(delimiter)
@@ -868,7 +880,7 @@ async function startWorkerWithSlowProcessTable(
   const originalPath = process.env.PATH;
   process.env.PATH = `${shimDir}${delimiter}${originalPath ?? ""}`;
   try {
-    return await startRealWorkerManager();
+    return await startRealWorkerManager(options);
   } finally {
     process.env.PATH = originalPath;
   }
@@ -1982,13 +1994,20 @@ function terminalViewers(
     },
     started: () => started,
     finish,
-    async attach(source: object, requestId: string) {
+    async attach(
+      source: object,
+      requestId: string,
+      restore: NonNullable<SubscribeTerminalRequest["restore"]> = {
+        mode: "visible-snapshot",
+        scrollbackLines: 1000,
+      },
+    ) {
       delivery.attach(source, true);
       const request = {
         type: "subscribe_terminal_request" as const,
         terminalId,
         requestId,
-        restore: { mode: "visible-snapshot" as const, scrollbackLines: 1000 },
+        restore,
       };
       await delivery.request(source, request, () => controller.dispatch(request, delivery));
     },
@@ -2113,20 +2132,28 @@ it("keeps concurrent viewer replays and live output ordered behind slow app deli
 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: "slow-delivery",
-        ...nodeTerminalCommand(
-          'process.stdout.write("READY\\r\\n"); process.stdin.on("data", data => process.stdout.write("ECHO:" + data)); setInterval(() => {}, 1000);',
-        ),
-      }),
-    );
+    const paseoHome = createTemporaryHome();
+    // Slow program identification belongs to setup, not the replay request deadline.
+    // The POSIX shim deterministically exceeds 800 ms; Windows keeps its real OS path.
+    manager = isPlatform("win32")
+      ? await startRealWorkerManager({ paseoHome })
+      : await startWorkerWithSlowProcessTable(1, { paseoHome });
+    const created = await manager.createTerminal({
+      cwd: paseoHome,
+      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.join("").includes("READY"),
+      async () => (await manager!.captureTerminal(created.id)).lines.join("").includes("READY"),
       10000,
     );
+    await manager.detach();
+    manager = await startRealWorkerManager({ paseoHome, requestTimeoutMs: 800 });
+    const session = manager.getTerminal(created.id);
+    if (!session) throw new Error("Replay fixture did not adopt its prepared terminal");
+    trackTerminal(session);
     const viewers = terminalViewers(manager, session.id, chunked);
     const phone = {};
     viewers.block(phone);
@@ -2208,3 +2235,108 @@ it.each(["replacement", "socket-overflow"] as const)(
   },
   30000,
 );
+
+it.each([true, false])(
+  "keeps live modes after a replay captured enabled=%s while restoring its own modes first",
+  async (enabled) => {
+    const on = "\x1b[?1h\x1b[?2004h";
+    const off = "\x1b[?1l\x1b[?2004l";
+    const captured = enabled ? on : "";
+    const latest = enabled ? "" : on;
+    const live = (enabled ? off : on) + "MODE-LIVE";
+    const worker = new FakeTerminalWorker();
+    manager = createWorkerTerminalManager({ worker, requestTimeoutMs: 1000 });
+    worker.emitWorkerMessage({
+      type: "terminalCreated",
+      terminal: {
+        id: "mode-terminal",
+        name: "Terminal",
+        cwd: "/tmp",
+        workspaceId: "modes",
+        activity: null,
+      },
+      state: createTerminalState(),
+    });
+    worker.emitWorkerMessage({
+      type: "terminalMessage",
+      terminalId: "mode-terminal",
+      replayPreamble: captured,
+      message: { type: "snapshotReady", revision: 1 },
+    });
+    const session = manager.getTerminal("mode-terminal")!;
+    const viewers = terminalViewers(manager, session.id);
+    const restoring = {},
+      following = {};
+    try {
+      await viewers.attach(restoring, "restoring");
+      await waitForCondition(
+        () => worker.sentMessages.some((message) => message.type === "replayTerminal"),
+        1000,
+      );
+      const request = worker.sentMessages.find((message) => message.type === "replayTerminal")!;
+      const metadata = {
+        rows: 30,
+        cols: 120,
+        retainedHistoryLines: 10,
+        revision: 1,
+        replayPreamble: captured,
+      };
+      worker.emitWorkerMessage({
+        type: "terminalReplay",
+        requestId: request.requestId,
+        event: { type: "begin", metadata },
+      });
+      worker.emitWorkerMessage({
+        type: "terminalReplay",
+        requestId: request.requestId,
+        event: { type: "data", data: Buffer.from("CAPTURE-A") },
+      });
+      worker.emitWorkerMessage({
+        type: "terminalMessage",
+        terminalId: session.id,
+        replayPreamble: latest,
+        message: { type: "output", data: live, revision: 2 },
+      });
+      expect(session.getReplayPreamble()).toBe(latest);
+      worker.emitWorkerMessage({
+        type: "terminalReplay",
+        requestId: request.requestId,
+        event: { type: "data", data: Buffer.from("CAPTURE-B") },
+      });
+      worker.emitWorkerMessage({
+        type: "terminalReplay",
+        requestId: request.requestId,
+        event: { type: "end" },
+      });
+      worker.emitWorkerMessage({
+        type: "response",
+        requestId: request.requestId,
+        ok: true,
+        result: metadata,
+      });
+      await waitForCondition(() => viewers.text(restoring).includes("MODE-LIVE"), 1000);
+      expect(session.getReplayPreamble()).toBe(latest);
+      expect(viewers.text(restoring)).toBe(captured + live);
+      const sequence = viewers.frames
+        .filter((row) => row.source === restoring)
+        .map((row) => row.frame.opcode);
+      expect(sequence.indexOf(TerminalStreamOpcode.Output)).toBeGreaterThan(
+        sequence.indexOf(TerminalStreamOpcode.ReplayEnd),
+      );
+      const modes = new TerminalInputModeTracker();
+      modes.feed(session.getReplayPreamble());
+      expect(encodeTerminalKeyInput({ key: "ArrowUp" }, { inputMode: modes.getState() })).toBe(
+        enabled ? "\x1b[A" : "\x1bOA",
+      );
+      expect(modes.getState().bracketedPaste).toBe(!enabled);
+      await viewers.attach(following, "live-following", { mode: "live" });
+      await waitForCondition(() => latest === "" || viewers.text(following).length > 0, 1000);
+      expect(viewers.text(following)).toBe(latest);
+      expect(
+        worker.sentMessages.filter((message) => message.type === "replayTerminal"),
+      ).toHaveLength(1);
+    } finally {
+      await viewers.delivery.close();
+    }
+  },
+);

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

@@ -885,8 +885,6 @@ export function createWorkerTerminalManager(
         { type: "replayTerminal", terminalId: id, options },
         sink,
       )) as TerminalReplayMetadata | null;
-      const record = recordsById.get(id);
-      if (record && metadata) record.replayPreamble = metadata.replayPreamble;
       return metadata;
     },