fix(terminal): bound backpressure recovery history

2026-10-30 21:23:04 +00:00 · 2026-10-30 18:22:20 +01:00
parent 4ce30bd90a
commit bc86241857
7 changed files with 289 additions and 16 deletions
@@ -27,7 +27,7 @@ Terminal frames share the daemon main event loop with all agent traffic. The `ev
 - **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 commits its authoritative tracker with xterm parsing before publishing the corresponding revision; 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.
+- **Snapshot catch-up is backpressure-gated.** After more than 256 KiB of output, a stream catches up only when its transport reports more than 4 MiB queued, or exposes no backpressure reading. A draining client streams continuously. Pressure recovery sends at most 500 recent history rows plus the screen, preserving smaller explicit demands including zero; even a live-only subscription needs this resynchronization. It does not change the subscription's configured demand: later attach/reconnect restores still request the full configured history. The recent window avoids repeatedly serializing all retained history for a slow viewer.
 - **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.
 - **Find preserves the inspected xterm viewport, including at the live bottom.** Keep its output anchor inside the existing batched write path. Serializing output to preserve a search result would add terminal latency.
@@ -34,7 +34,7 @@ import {
 import { mkdtemp, rm } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
-import { getAvailablePort } from "./helpers/network.ts";
+import { createServer } from "node:net";

 if (process.platform === "win32") {
  console.log("Skipping terminal worker upgrade CLI test on Windows (POSIX shell terminals).");
@@ -49,7 +49,6 @@ const paseoHome = join(root, "home");
 const cliHome = join(root, "os-home");
 const workDir = join(root, "work");
 for (const dir of [paseoHome, cliHome, workDir]) mkdirSync(dir, { recursive: true });
-const port = await getAvailablePort();
 const currentProtocolVersion = 5;
 const zsh = ["/bin/zsh", "/usr/bin/zsh"].find((candidate) => existsSync(candidate)) ?? null;

@@ -336,7 +335,18 @@ async function terminalIds(installation: Installation): Promise<string[]> {
 }

 async function startDaemon(installation: Installation, extraEnv: Record<string, string> = {}) {
-  await paseo(installation, ["daemon", "start"], extraEnv);
+  const result = await paseo(installation, ["daemon", "start", "--json"], extraEnv);
+  lastListen = (JSON.parse(result.stdout) as { listen: string }).listen;
+  assert.match(lastListen, /^127\.0\.0\.1:[1-9]\d*$/, "start publishes the actual bound port");
+  if (competingListener.listening) {
+    const competitor = competingListener.address();
+    assert(competitor && typeof competitor !== "string");
+    assert.notEqual(
+      Number(lastListen.split(":").at(-1)),
+      competitor.port,
+      "each daemon binds its own endpoint while the released port belongs to a peer",
+    );
+  }
 }

 async function stopDaemon(installation: Installation, keepTerminals = false) {
@@ -371,10 +381,14 @@ const a = createInstallation("A", "0.98.0");
 const b = createInstallation("B", "0.99.0");
 let restoreA: (() => void) | null = null;
 let passed = false;
+let lastListen = "";
+const competingListener = createServer();

 try {
  for (const [field, value] of [
-    ["daemon.listen", `127.0.0.1:${port}`],
+    // Let the listening daemon reserve its port atomically. Clients discover the
+    // actual endpoint from this home's PID record after every start.
+    ["daemon.listen", "127.0.0.1:0"],
    ["daemon.relay.enabled", "false"],
    ["features.dictation.enabled", "false"],
    ["features.voiceMode.enabled", "false"],
@@ -433,6 +447,15 @@ try {
  restoreA = null;
  console.log("✓ closed and cleaned up\n");

+  // A released endpoint belongs to nobody. Reproduce another process taking it
+  // between starts; worker adoption is keyed by the home, not by a stable TCP port.
+  const releasedPort = Number(lastListen.split(":").at(-1));
+  await new Promise<void>((resolve, reject) => {
+    competingListener.once("error", reject);
+    competingListener.listen(releasedPort, "127.0.0.1", resolve);
+  });
+  console.log(`Competing listener owns released port ${releasedPort}`);
+
  // There is no supported previous wire at protocol 5. Future previous-version classification
  // is covered by terminal-worker-protocol.test.ts; do not simulate an unreleased drain adapter.
  for (const protocolVersion of [0, 1, 2, 3, 4]) {
@@ -514,6 +537,17 @@ try {
      }
    }
  }
+  if (competingListener.listening) {
+    await new Promise<void>((resolve, reject) =>
+      competingListener.close((error) => {
+        if (error) {
+          reject(error);
+          return;
+        }
+        resolve();
+      }),
+    );
+  }
  restoreA?.();
  for (const installation of [b, a]) {
    await run(installation, ["daemon", "stop", "--home", paseoHome, "--force"]);
@@ -26,12 +26,14 @@ describe("terminal restore policy", () => {
    ).toEqual({ scrollbackLines: 500 });
  });

-  test("promotes live restore to visible restore after output overflow", () => {
+  test("bounds pressure recovery for every restore mode", () => {
    expect(resolveRestoreAfterOutputOverflow({ mode: "live" })).toEqual({
      mode: "visible-snapshot",
+      scrollbackLines: 500,
    });
    expect(resolveRestoreAfterOutputOverflow({ mode: "full-snapshot" })).toEqual({
-      mode: "full-snapshot",
+      mode: "visible-snapshot",
+      scrollbackLines: 500,
    });
  });

@@ -23,11 +23,14 @@ export type TerminalRestoreOptions = NonNullable<SubscribeTerminalRequest["resto

 export function resolveRestoreAfterOutputOverflow(
  restore: TerminalRestoreOptions | undefined,
-): TerminalRestoreOptions | undefined {
-  if (restore?.mode === "live") {
-    return { mode: "visible-snapshot" };
-  }
-  return restore;
+): TerminalRestoreOptions {
+  return {
+    mode: "visible-snapshot",
+    scrollbackLines: Math.min(
+      restore?.scrollbackLines ?? MAX_VISIBLE_RESTORE_SCROLLBACK_LINES,
+      MAX_VISIBLE_RESTORE_SCROLLBACK_LINES,
+    ),
+  };
 }

 export function resolveTerminalRestoreSnapshotOptions(
@@ -1,3 +1,5 @@
+import type { TerminalReplayOptions } from "./terminal-replay.js";
+import type { TerminalRestoreOptions } from "./terminal-restore.js";
 import type { TerminalReplayMetadata } from "./terminal-replay.js";
 import { describe, expect, test, vi } from "vitest";
 import pino from "pino";
@@ -682,10 +684,13 @@ describe("terminal-session-controller backpressure snapshot fallback", () => {
    getClientBufferedAmount: () => number | null,
    chunked = false,
    pace?: () => Promise<void>,
+    restore?: TerminalRestoreOptions,
  ): Promise<{
    messages: SessionOutboundMessage[];
    pushOutput: (data: string) => void;
    frames: TerminalStreamFrame[];
+    replayOptions: Array<TerminalReplayOptions | undefined>;
+    ready: () => void;
  }> {
    let terminalListener: ((message: ServerMessage) => void) | null = null;
    const terminal: TerminalSession = {
@@ -712,12 +717,16 @@ describe("terminal-session-controller backpressure snapshot fallback", () => {
      kill: vi.fn(),
      killAndWait: vi.fn(),
    };
+    const replayOptions: Array<TerminalReplayOptions | undefined> = [];
    const terminalManager: TerminalManager = {
      getTerminals: vi.fn(),
      createTerminal: vi.fn(),
      registerCwdEnv: vi.fn(),
      getTerminal: vi.fn(() => terminal),
-      replayTerminal: replayFixture(async () => terminalReplay("live", 1)),
+      replayTerminal: (id, options, sink) => {
+        replayOptions.push(options);
+        return replayFixture(async () => terminalReplay("live", 1))(id, options, sink);
+      },
      setTerminalTitle: vi.fn(),
      killTerminal: vi.fn(),
      killTerminalAndWait: vi.fn(),
@@ -757,6 +766,7 @@ describe("terminal-session-controller backpressure snapshot fallback", () => {
      type: "subscribe_terminal_request",
      terminalId: "term-1",
      requestId: "req-1",
+      restore,
    });
    await waitForCoalescerFlush();
    // Drop the initial subscribe snapshot frame so each test only sees frames
@@ -764,12 +774,44 @@ describe("terminal-session-controller backpressure snapshot fallback", () => {
    frames.length = 0;

    return {
+      replayOptions,
+      ready: () => terminalListener?.({ type: "snapshotReady" }),
      pushOutput: (data) => terminalListener?.({ type: "output", data, revision: 2 }),
      frames,
      messages,
    };
  }

+  test.each([false, true])(
+    "bounds recovery without changing attach demand (chunked=%s)",
+    async (chunked) => {
+      for (const [restore, history] of [
+        [{ mode: "visible-snapshot", scrollbackLines: 10000 }, 500],
+        [{ mode: "full-snapshot" }, 500],
+        [{ mode: "visible-snapshot", scrollbackLines: 5 }, 5],
+        [{ mode: "visible-snapshot", scrollbackLines: 0 }, 0],
+        [{ mode: "full-snapshot", scrollbackLines: 0 }, 0],
+        [{ mode: "live", scrollbackLines: 0 }, 0],
+        [{ mode: "live" }, 500],
+        [undefined, 500],
+      ] as const) {
+        const fixture = await setup(() => 8 * 1024 * 1024, chunked, undefined, restore);
+        const initial = fixture.replayOptions.slice();
+        fixture.pushOutput("pressure".repeat(40000));
+        await waitForCoalescerFlush();
+        expect(fixture.replayOptions.at(-1)).toEqual({ scrollbackLines: history });
+        expect(fixture.replayOptions).toHaveLength(initial.length + 1);
+        fixture.ready(); // Worker adoption must use the original demand, even on the same stream.
+        await waitForCoalescerFlush();
+        expect(fixture.replayOptions.slice(initial.length + 1)).toEqual(initial);
+        fixture.frames.length = 0;
+        fixture.pushOutput("after-catch-up");
+        await waitForCoalescerFlush();
+        expect(fixture.frames.map((frame) => frame.opcode)).toEqual([TerminalStreamOpcode.Output]);
+      }
+    },
+  );
+
  test("streams all output without a snapshot when the client keeps up", async () => {
    const { pushOutput, frames } = await setup(() => 0);

@@ -929,7 +929,6 @@ export class TerminalSessionController {
            // The snapshot replaces this batch only after it succeeds. If the
            // terminal disappears during the read, completion still owns these bytes.
            activeStream.snapshotOutput = payload;
-            activeStream.restore = resolveRestoreAfterOutputOverflow(activeStream.restore);
            activeStream.needsSnapshot = true;
            void this.trySendSnapshot(activeStream);
            return;
@@ -1029,7 +1028,12 @@ export class TerminalSessionController {
    activeStream.outputCoalescer.flush();
    activeStream.snapshotInFlight = true;
    try {
-      const restore = activeStream.restore;
+      // A pressure catch-up replaces the skipped batch with a recent window.
+      // Keep the subscription demand intact for later adoption/full restores.
+      const restore =
+        activeStream.snapshotOutput !== undefined
+          ? resolveRestoreAfterOutputOverflow(activeStream.restore)
+          : activeStream.restore;
      const snapshotResult = restore
        ? await this.emitRestoreSnapshot(activeStream, terminalManager, restore)
        : await this.emitRestoreSnapshot(activeStream, terminalManager, { mode: "full-snapshot" });
@@ -1958,6 +1958,7 @@ function terminalViewers(
  terminalManager: DaemonTerminalManager,
  terminalId: string,
  chunked = true,
+  getClientBufferedAmount?: (source: object) => number | null,
 ) {
  const frames: Array<{ source: object; frame: TerminalStreamFrame }> = [];
  const messages: Array<{ source: object; message: SessionOutboundMessage }> = [];
@@ -1990,6 +1991,7 @@ function terminalViewers(
    },
    hasBinaryChannel: () => true,
    supportsChunkedReplay: () => chunked,
+    getClientBufferedAmount,
    isPathWithinRoot: () => true,
    sessionLogger: pino({ level: "silent" }),
  });
@@ -2354,6 +2356,8 @@ it.each([3000, 12050])(
  "retains ten thousand history rows and restores exact demands after %s printed rows",
  async (count) => {
    manager = await startRealWorkerManager();
+    const gate = createTemporaryHome();
+    const releaseNewline = join(gate, "newline");
    const session = trackTerminal(
      await manager.createTerminal({
        cwd: createTemporaryHome(),
@@ -2361,7 +2365,14 @@ it.each([3000, 12050])(
        rows: 4,
        cols: 80,
        ...nodeTerminalCommand(
-          `for(let i=0;i<${count};i++) process.stdout.write("\\x1b[31mrow-"+String(i).padStart(5,"0")+" 界é🙂\\x1b[0m\\r\\n");setInterval(()=>{},1000)`,
+          `const fs=require("node:fs");
+          const watcher=fs.watch(${JSON.stringify(gate)},()=>{
+            if(fs.existsSync(${JSON.stringify(releaseNewline)})){
+              watcher.close();process.stdout.write("\\r\\nROWS-COMPLETE");
+            }
+          });
+          for(let i=0;i<${count};i++) process.stdout.write("\\x1b[31mrow-"+String(i).padStart(5,"0")+" 界é🙂\\x1b[0m"+(i===${count}-1?"":"\\r\\n"));
+          setInterval(()=>{},1000)`,
        ),
      }),
    );
@@ -2372,6 +2383,14 @@ it.each([3000, 12050])(
          .includes("row-" + String(count - 1).padStart(5, "0")),
      15000,
    );
+    // A visible final label can precede its trailing CRLF. Hold that transition
+    // deliberately, then observe content on the following row before capture.
+    writeFileSync(releaseNewline, "release");
+    await waitForCondition(
+      async () =>
+        (await manager!.captureTerminal(session.id, { start: -1 })).lines[0] === "ROWS-COMPLETE",
+      15000,
+    );
    for (const history of [0, 200, 1000, 10000]) {
      const chunks: Uint8Array[] = [];
      const metadata = await manager.replayTerminal(
@@ -2399,6 +2418,8 @@ it.each([3000, 12050])(
        const labels = Array.from({ length: buffer.length }, (_, i) =>
          buffer.getLine(i)!.translateToString(true),
        ).filter((line) => line.includes("row-"));
+        expect({ row: buffer.cursorY, col: buffer.cursorX }).toEqual({ row: 3, col: 13 });
+        expect(buffer.getLine(buffer.length - 1)!.translateToString(true)).toBe("ROWS-COMPLETE");
        expect(labels).toHaveLength(Math.min(count, history + 3));
        expect(labels[0]).toContain("row-" + String(count - labels.length).padStart(5, "0"));
        expect(labels.at(-1)).toContain("row-" + String(count - 1).padStart(5, "0") + " 界é🙂");
@@ -2409,3 +2430,170 @@ it.each([3000, 12050])(
  },
  30000,
 );
+
+it.each([false, true])(
+  "shrinks real ten-thousand-row pressure recovery and preserves reattach (chunked=%s)",
+  async (chunked) => {
+    manager = await startRealWorkerManager();
+    const session = trackTerminal(
+      await manager.createTerminal({
+        cwd: createTemporaryHome(),
+        workspaceId: "pressure-history",
+        rows: 4,
+        cols: 120,
+        ...nodeTerminalCommand(`
+      process.stdin.setRawMode(true);
+      let row = 0;
+      function print(count) {
+        for(let i=0;i<count;i++,row++) process.stdout.write("\\x1b[31mrow-" + String(row).padStart(5,"0") + " 界é🙂 " + "log ".repeat(20) + "\\x1b[0m\\r\\n");
+      }
+      print(12050);
+      process.stdout.write("READY\\r\\n");
+      process.stdin.on("data", data => {
+        if(data.toString().includes("BURST")) { print(4000); process.stdout.write("BURST-DONE\\r\\n"); }
+        else process.stdout.write("ECHO:" + data.toString() + "\\r\\n");
+      });
+      setInterval(()=>{},1000);
+    `),
+      }),
+    );
+    await waitForCondition(
+      async () =>
+        (await manager!.captureTerminal(session.id, { start: -5 })).lines
+          .join("")
+          .includes("READY"),
+      15000,
+    );
+    const desktop = {},
+      phone = {},
+      fresh = {};
+    let pressure = false;
+    const viewers = terminalViewers(manager, session.id, chunked, (source) =>
+      source === phone && pressure ? 8 * 1024 * 1024 : 0,
+    );
+    const restore = { mode: "full-snapshot" as const, scrollbackLines: 10000 };
+    function payloads(source: object): Buffer[] {
+      const result: Buffer[] = [];
+      let chunks: Uint8Array[] = [];
+      for (const { source: owner, frame } of viewers.frames) {
+        if (owner !== source) continue;
+        if (frame.opcode === TerminalStreamOpcode.Restore) result.push(Buffer.from(frame.payload));
+        if (frame.opcode === TerminalStreamOpcode.ReplayBegin) chunks = [];
+        if (frame.opcode === TerminalStreamOpcode.ReplayData) chunks.push(frame.payload);
+        if (frame.opcode === TerminalStreamOpcode.ReplayEnd) result.push(Buffer.concat(chunks));
+      }
+      return result;
+    }
+    async function parse(payload: Uint8Array) {
+      const receiver = new xterm.Terminal({
+        rows: 4,
+        cols: 120,
+        scrollback: 10000,
+        allowProposedApi: true,
+      });
+      try {
+        await new Promise<void>((resolve) => receiver.write(payload, resolve));
+        const buffer = receiver.buffer.active;
+        return {
+          history: buffer.baseY,
+          text: Array.from({ length: buffer.length }, (_, i) =>
+            buffer.getLine(i)!.translateToString(true),
+          ).join("\n"),
+        };
+      } finally {
+        receiver.dispose();
+      }
+    }
+    try {
+      await viewers.attach(desktop, "healthy", restore);
+      await viewers.attach(phone, "pressured", restore);
+      await waitForCondition(
+        () => payloads(phone).length === 1 && payloads(desktop).length === 1,
+        15000,
+      );
+      const initial = payloads(phone)[0]!;
+      expect((await parse(initial)).history).toBe(10000);
+      pressure = true;
+      session.send({ type: "input", data: "BURST" });
+      await waitForCondition(() => payloads(phone).length >= 2, 15000);
+      pressure = false;
+      // The final burst marker may belong to the replay cut or to its live tail.
+      // Read the same replay/output exchange a renderer sees rather than assuming
+      // worker parsing and daemon request scheduling cut the burst at one point.
+      async function currentView(source: object) {
+        const frames = viewers.frames
+          .filter((row) => row.source === source)
+          .map((row) => row.frame);
+        const start = frames.findLastIndex(
+          (frame) =>
+            frame.opcode === TerminalStreamOpcode.Restore ||
+            frame.opcode === TerminalStreamOpcode.ReplayBegin,
+        );
+        return parse(
+          Buffer.concat(
+            frames
+              .slice(start)
+              .filter(
+                (frame) =>
+                  frame.opcode === TerminalStreamOpcode.Restore ||
+                  frame.opcode === TerminalStreamOpcode.ReplayData ||
+                  frame.opcode === TerminalStreamOpcode.Output,
+              )
+              .map((frame) => frame.payload),
+          ),
+        );
+      }
+      await waitForCondition(
+        async () =>
+          (await currentView(phone)).text.includes("BURST-DONE") &&
+          (await currentView(desktop)).text.includes("BURST-DONE"),
+        15000,
+      );
+      const recovery = payloads(phone)[1]!;
+      const recovered = await parse(recovery);
+      expect(recovered.history).toBe(500);
+      expect(recovered.text).toContain("界é🙂");
+      for (const source of [phone, desktop]) {
+        const indexes = [...(await currentView(source)).text.matchAll(/row-(\d+)/g)].map((match) =>
+          Number(match[1]),
+        );
+        expect(indexes.at(-1)).toBe(16049);
+        expect(indexes).toEqual(Array.from({ length: indexes.length }, (_, i) => indexes[0]! + i));
+      }
+      expect(recovery.byteLength).toBeLessThan(initial.byteLength / 10);
+      expect(recovery.byteLength).toBeLessThan(256 * 1024);
+      expect(payloads(desktop)).toHaveLength(1); // The healthy peer keeps its full replay and live stream.
+      session.send({ type: "input", data: "AFTER-CATCH-UP" });
+      await waitForCondition(
+        () =>
+          viewers.text(phone).includes("ECHO:AFTER-CATCH-UP") &&
+          viewers.text(desktop).includes("ECHO:AFTER-CATCH-UP"),
+        10000,
+      );
+      expect(viewers.text(phone).match(/ECHO:AFTER-CATCH-UP/g)).toHaveLength(1);
+      expect(viewers.text(desktop).match(/ECHO:AFTER-CATCH-UP/g)).toHaveLength(1);
+      await viewers.delivery.detach(phone);
+      await viewers.attach(fresh, "reattach", restore);
+      await waitForCondition(() => payloads(fresh).length === 1, 15000);
+      const reattached = await parse(payloads(fresh)[0]!);
+      expect(reattached.history).toBe(10000);
+      expect(reattached.text).toContain("row-16049");
+      expect(reattached.text).toContain("ECHO:AFTER-CATCH-UP");
+      expect(viewers.ended(desktop)).toHaveLength(0);
+      expect(viewers.ended(phone)).toHaveLength(0);
+      process.stdout.write(
+        JSON.stringify({
+          chunked,
+          initialBytes: initial.byteLength,
+          recoveryBytes: recovery.byteLength,
+          reattachBytes: payloads(fresh)[0]!.byteLength,
+          retainedHistory: reattached.history,
+          recoveryHistory: recovered.history,
+        }) + "\n",
+      );
+    } finally {
+      await viewers.delivery.close();
+    }
+  },
+  45000,
+);