From 72b67f48e3a9a6fff5b1619a0dcc196e043480de Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 13 Jun 2026 13:38:58 +0800 Subject: [PATCH] Eliminate spiky terminal lag under load (#1500) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Eliminate spiky terminal lag under load Terminal output stuttered during heavy output and when the daemon was busy. Two compounding causes, both confirmed by measurement: - Every 256KB of output, the daemon dropped pending frames and re-sent a full cell-grid snapshot (cloned across IPC, stringified to JSON) — a 10MB build did this ~40 times. The snapshot fallback is now gated on real client backpressure (ws bufferedAmount), so a client that keeps draining streams continuously and never pays the snapshot tax. This also removes the periodic GC hitch the churn caused. - Per-chunk overhead on the shared event loop: one IPC message per pty chunk, a duplicate input-mode regex scan on the daemon main loop, a double JSON.stringify per outbound message, and 16KB string realloc per chunk. Output now coalesces in the worker before IPC (leading-edge so keystrokes still echo immediately), the duplicate scan is gone, and the client feeds xterm back-to-back instead of one render tick a frame. Adds eventLoopDelay percentiles to ws_runtime_metrics for main-loop stall visibility, plus a reproducible Node benchmark (no port 6767). Measured: echo p50 7.7ms to 2.3ms; a 2MB burst now streams fully with zero snapshots; loop-stall spikes 100-173ms to 2ms. * Guard trace-level check against partial logger stubs The isLevelEnabled gate added for the emit() perf fix crashed every session test that injects a hand-rolled logger stub (10 of them omit isLevelEnabled). Real pino loggers always provide it; optional-chaining keeps the perf gate intact in production and no-ops on stubs. * Make recent-output exit-summary test deterministic The added test self-exited the child immediately after writing 3000 lines, then asserted the newest line was in the exit summary — but the summary reads the headless xterm buffer, which parses writes asynchronously, so a loaded CI runner saw a stale buffer (line-2707 not line-2999). Keep the process alive and poll the parsed buffer until the final line lands before killing, removing the race. * Skip ConPTY-fragile worker terminal tests on Windows The coalescing and input-mode-preamble tests assert byte-contiguous PTY output and an exact kitty-escape round-trip. Windows ConPTY injects repaint sequences between writes and normalizes the escape, so both time out there while passing on Linux/macOS. Gate them with skipIf(win32), matching the existing terminal-test convention for PTY-sensitive cases. * Apply terminal barriers immediately when no writes are pending A barrier op (snapshot/restore/clear) only needs the sentinel-write gate to wait out plain writes still parsing in xterm's buffer. When none are ungated — at mount, or right after another barrier — the sentinel cost a wasted parse cycle, adding latency to the first snapshot/restore on the hot first-paint path. Track ungated writes with a flag and skip the sentinel when there's nothing to gate. * Fix mobile startup saved-host e2e * Unskip Windows worker terminal coverage * Tighten worker coalescing assertion * Reset terminal ungated writes on unmount * Fix terminal regressions found in adversarial review Three issues this PR's terminal changes introduced: - Worker snapshot could duplicate coalesced output. getTerminalState snapshotted without flushing the worker output coalescer, so a batch spanning the snapshot point carried a revision past it and the controller's dedup couldn't drop it — the client saw the bytes twice. Flush before snapshotting. - Relay clients lost backpressure protection. The snapshot fallback was gated on bufferedAmount, which the multiplexed relay socket reports as absent; that read as 0 ("keeping up") so a slow relay client never caught up via a snapshot. Distinguish "no signal" (null) from 0 and keep the unconditional byte-threshold fallback for signal-less transports, preserving the pre-change relay behavior. - Recent-output buffer was no longer a hard cap. A single chunk larger than the limit was retained whole; slice its tail. Documents the live-restore preamble gap (unreachable: no client sends restore mode "live"). * Skip input-mode preamble test on Windows ConPTY CI confirmed Windows ConPTY normalizes away the kitty keyboard escape the child writes, so it never reaches the worker's input-mode tracker and the preamble stays empty — the test times out. ConPTY can't exercise this contract; the coalescing test (file-gated, line-oriented) stays on all platforms, only the preamble assertion is gated to Linux/macOS. --- CLAUDE.md | 1 + docs/terminal-performance.md | 40 + packages/app/e2e/helpers/startup-dsl.ts | 24 +- .../runtime/terminal-emulator-runtime.test.ts | 197 ++++- .../runtime/terminal-emulator-runtime.ts | 98 +++ packages/server/src/server/session.ts | 24 +- .../server/src/server/websocket-server.ts | 40 + .../terminal-output-coalescer.test.ts | 81 +- .../src/terminal/terminal-output-coalescer.ts | 22 + .../server/src/terminal/terminal-restore.ts | 7 + .../terminal-session-controller.test.ts | 136 ++++ .../terminal/terminal-session-controller.ts | 33 +- .../src/terminal/terminal-worker-process.ts | 58 +- .../src/terminal/terminal-worker-protocol.ts | 2 + packages/server/src/terminal/terminal.test.ts | 44 + packages/server/src/terminal/terminal.ts | 47 +- .../terminal/worker-terminal-manager.test.ts | 130 ++- .../src/terminal/worker-terminal-manager.ts | 28 +- scripts/benchmark-terminal-latency.ts | 751 ++++++++++++++++++ 19 files changed, 1693 insertions(+), 70 deletions(-) create mode 100644 docs/terminal-performance.md create mode 100644 scripts/benchmark-terminal-latency.ts diff --git a/CLAUDE.md b/CLAUDE.md index 697bac7bf..26c82593f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -39,6 +39,7 @@ At the start of non-trivial work, list `docs/` and skim anything relevant to the | [docs/service-proxy.md](docs/service-proxy.md) | Service proxy: exposing workspace scripts at public URLs, DNS setup, reverse proxy config | | [docs/development.md](docs/development.md) | Dev server, build sync gotchas, CLI reference, agent state, Playwright MCP | | [docs/rpc-namespacing.md](docs/rpc-namespacing.md) | WebSocket RPC naming convention — dotted namespaces and `.request`/`.response` pairs | +| [docs/terminal-performance.md](docs/terminal-performance.md) | Terminal latency pipeline, coalescing/backpressure invariants, benchmark + perf spec usage | | [docs/testing.md](docs/testing.md) | TDD workflow, determinism, real dependencies over mocks, test organization | | [docs/mobile-testing.md](docs/mobile-testing.md) | Maestro and mobile test workflows | | [docs/ad-hoc-daemon-testing.md](docs/ad-hoc-daemon-testing.md) | Isolated in-process daemon test harness | diff --git a/docs/terminal-performance.md b/docs/terminal-performance.md new file mode 100644 index 000000000..287855aa6 --- /dev/null +++ b/docs/terminal-performance.md @@ -0,0 +1,40 @@ +# Terminal performance + +How terminal output stays low-latency, what the invariants are, and how to measure before/after any change to the pipeline. Read this before touching anything under `packages/server/src/terminal/` or `packages/app/src/terminal/runtime/`. + +## The pipeline + +``` +pty (node-pty, forked worker process) + → headless xterm parse (worker, snapshot fidelity) + → TerminalOutputCoalescer (worker, ≤1 IPC message per 5ms per terminal) + → process.send IPC → daemon main process + → TerminalOutputCoalescer (per client stream, terminal-session-controller.ts) + → binary ws frame (2-byte header + raw bytes) + → client decode (daemon-client.ts) → stream router → emulator runtime + → xterm.write (back-to-back; xterm batches internally) +``` + +Terminal frames share the daemon main event loop with all agent traffic. The `eventLoopDelay` block in the `ws_runtime_metrics` log line (every 30s in `daemon.log`) is the ground truth for "the daemon is busy" — p99/max there directly bound worst-case terminal frame delay. + +## Invariants (the easy-to-break ones) + +- **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 `process.send` 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. +- **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 `getTerminalState` 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. +- **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. + +## Measuring + +- **Node-only benchmark (fast iteration, server pipeline):** `npx tsx scripts/benchmark-terminal-latency.ts`. Boots an isolated daemon (fresh `PASEO_HOME`, random port — never 6767), measures echo latency percentiles, burst jitter, and snapshot counts under ramped mock-agent load. Writes JSON to `/tmp/paseo-terminal-bench/`. Healthy numbers (2026-06): echo p50 ~2.3ms, p95 ~3.3ms, a 2MB burst fully streamed with `snap=0`. +- **Browser perf specs (user-perceived path):** gated behind `PASEO_TERMINAL_PERF_E2E=1` — + `packages/app/e2e/terminal-performance.spec.ts` and `packages/app/e2e/terminal-keystroke-stress.spec.ts` (per-stage keydown→xterm-commit breakdown under mock-agent load). Healthy: keydown→commit p50 ~18ms under 600-key burst. +- **Production:** grep `daemon.log` for `ws_runtime_metrics` and read `eventLoopDelay` + `bufferedAmount`. + +## Known remaining contention (follow-up candidates) + +- A single large `agent_stream` message (e.g. a 250KB diff payload) measurably delays terminal echo (~100ms-class dips) — cost is split between daemon serialization and app-side parse/render on the shared browser main thread. +- Relay-attached clients pay pure-JS tweetnacl encryption + base64 per frame on the daemon main loop (`packages/relay/src/encrypted-channel.ts`). +- `sendToClient` re-stringifies session messages per socket; only matters for multi-socket connections. diff --git a/packages/app/e2e/helpers/startup-dsl.ts b/packages/app/e2e/helpers/startup-dsl.ts index 60e8893da..c48e51cf9 100644 --- a/packages/app/e2e/helpers/startup-dsl.ts +++ b/packages/app/e2e/helpers/startup-dsl.ts @@ -1,3 +1,4 @@ +import type { Route } from "@playwright/test"; import { expect, type Page } from "../fixtures"; import { buildCreateAgentPreferences, buildSeededHost } from "./daemon-registry"; import { wsRoutePatternForPort } from "./daemon-port"; @@ -6,6 +7,7 @@ const DISABLE_DEFAULT_SEED_ONCE_KEY = "@paseo:e2e-disable-default-seed-once"; const SEED_NONCE_KEY = "@paseo:e2e-seed-nonce"; const REGISTRY_KEY = "@paseo:daemon-registry"; const E2E_KEY = "@paseo:e2e"; +const STORAGE_SEED_HTML = "storage seed"; interface SavedHostInput { serverId: string; @@ -88,9 +90,9 @@ class StartupScenario { return; } - // Let the shared fixture create its seed nonce, then opt out of that seed for - // the next navigation so this scenario owns the stored host registry. - await this.page.goto("/"); + // Create same-origin storage without booting the app. If the default host runtime + // starts first, it can race this scenario's registry override during CI. + await openStorageSeedDocument(this.page); const nowIso = new Date().toISOString(); const registry = this.savedHosts.map((host) => buildStoredHost({ @@ -132,6 +134,21 @@ class StartupScenario { } } +async function openStorageSeedDocument(page: Page): Promise { + await page.route( + "**/*", + async (route: Route) => { + await route.fulfill({ + status: 200, + contentType: "text/html", + body: STORAGE_SEED_HTML, + }); + }, + { times: 1 }, + ); + await page.goto("/"); +} + class StartupAssertions { private readonly page: Page; @@ -140,6 +157,7 @@ class StartupAssertions { } async expectsSavedHostShell(input: { label: string }): Promise { + await this.page.getByRole("button", { name: "Open menu", exact: true }).click(); await expect(this.page.getByText(input.label, { exact: true })).toBeVisible({ timeout: 15_000, }); diff --git a/packages/app/src/terminal/runtime/terminal-emulator-runtime.test.ts b/packages/app/src/terminal/runtime/terminal-emulator-runtime.test.ts index 4cf9c6708..965e3912b 100644 --- a/packages/app/src/terminal/runtime/terminal-emulator-runtime.test.ts +++ b/packages/app/src/terminal/runtime/terminal-emulator-runtime.test.ts @@ -104,13 +104,34 @@ function createRuntimeWithTerminal(): { writeTexts: string[]; } { const runtime = new TerminalEmulatorRuntime(); + const terminalState = attachStubTerminal(runtime); + + return { + runtime, + ...terminalState, + }; +} + +function attachStubTerminal(runtime: TerminalEmulatorRuntime): { + terminal: StubTerminal & { + resetCalls: number; + }; + writeCallbacks: Array<() => void>; + writeTexts: string[]; +} { const writeCallbacks: Array<() => void> = []; const writeTexts: string[] = []; let resetCalls = 0; const terminal: StubTerminal & { resetCalls: number } = { write: (data: string | Uint8Array, callback?: () => void) => { - writeTexts.push(decodeTerminalOutput(data)); + const text = decodeTerminalOutput(data); + // The runtime submits a zero-length sentinel write to gate barrier ops behind the + // drained write run. Keep its callback (the gate resolves through it) but exclude the + // empty payload from writeTexts so assertions read like real terminal output. + if (text.length > 0) { + writeTexts.push(text); + } if (callback) { writeCallbacks.push(callback); } @@ -131,7 +152,6 @@ function createRuntimeWithTerminal(): { (runtime as unknown as { terminal: StubTerminal }).terminal = terminal; return { - runtime, terminal, writeCallbacks, writeTexts, @@ -164,7 +184,7 @@ describe("terminal-emulator-runtime", () => { vi.useRealTimers(); }); - it("processes write and clear operations in strict order", () => { + it("drains contiguous plain writes without waiting for each commit, gating a clear behind them", () => { const { runtime, terminal, writeCallbacks, writeTexts } = createRuntimeWithTerminal(); const committed: string[] = []; @@ -174,43 +194,64 @@ describe("terminal-emulator-runtime", () => { committed.push("first"); }, }); - runtime.clear({ - onCommitted: () => { - committed.push("clear"); - }, - }); runtime.write({ data: terminalOutput("second"), onCommitted: () => { committed.push("second"); }, }); + runtime.clear({ + onCommitted: () => { + committed.push("clear"); + }, + }); + runtime.write({ + data: terminalOutput("third"), + onCommitted: () => { + committed.push("third"); + }, + }); - expect(writeTexts).toEqual(["first"]); + // Both plain writes are submitted back-to-back; neither waited on the other's callback. + // The clear and the write after it stay queued behind the barrier gate (sentinel write). + expect(writeTexts).toEqual(["first", "second"]); expect(terminal.resetCalls).toBe(0); expect(committed).toEqual([]); + // writeCallbacks: [0]=first, [1]=second, [2]=barrier-gate sentinel. writeCallbacks[0]?.(); - - expect(committed).toEqual(["first", "clear"]); - expect(terminal.resetCalls).toBe(1); - expect(writeTexts).toEqual(["first", "second"]); - writeCallbacks[1]?.(); - expect(committed).toEqual(["first", "clear", "second"]); + expect(committed).toEqual(["first", "second"]); + expect(terminal.resetCalls).toBe(0); + + // Resolving the sentinel releases the clear, which resets and then drains "third". + writeCallbacks[2]?.(); + expect(committed).toEqual(["first", "second", "clear"]); + expect(terminal.resetCalls).toBe(1); + expect(writeTexts).toEqual(["first", "second", "third"]); + + // "third" still commits through its own write callback. + writeCallbacks[3]?.(); + expect(committed).toEqual(["first", "second", "clear", "third"]); }); - it("falls back to timeout commit when xterm write callback does not fire", () => { + it("falls back to timeout commit for a barrier op when the gate sentinel never fires", () => { vi.useFakeTimers(); - const { runtime } = createRuntimeWithTerminal(); + const { runtime, terminal } = createRuntimeWithTerminal(); const onCommitted = vi.fn(); - runtime.write({ + // restoreOutput is a barrier; if xterm never commits the gate sentinel, the barrier still + // applies (and commits) after the 5s safety timeout instead of stalling forever. + runtime.restoreOutput({ data: terminalOutput("stuck"), onCommitted, }); expect(onCommitted).not.toHaveBeenCalled(); + expect(terminal.resetCalls).toBe(0); + vi.advanceTimersByTime(5_000); + // The barrier now runs (write submitted) but its own commit still waits on xterm; advance + // again to fire the barrier's own write timeout. vi.advanceTimersByTime(5_000); expect(onCommitted).toHaveBeenCalledTimes(1); }); @@ -239,6 +280,12 @@ describe("terminal-emulator-runtime", () => { }, }, }); + // The plain write reports kitty flags synchronously during drain; the snapshot resets + // them only after its barrier gate (the sentinel write callback) resolves. + expect(inputModeChanges).toEqual([{ kittyKeyboardFlags: 7, win32InputMode: false }]); + + // The plain write carries no onCommitted, so it registers no callback; writeCallbacks[0] + // is the barrier gate sentinel. writeCallbacks[0]?.(); expect(inputModeChanges).toEqual([ @@ -247,8 +294,8 @@ describe("terminal-emulator-runtime", () => { ]); }); - it("ignores stale duplicate write callbacks from a previous operation", () => { - const { runtime, writeCallbacks } = createRuntimeWithTerminal(); + it("commits each drained plain write through its own xterm callback", () => { + const { runtime, writeTexts, writeCallbacks } = createRuntimeWithTerminal(); const committed: string[] = []; runtime.write({ @@ -264,14 +311,89 @@ describe("terminal-emulator-runtime", () => { }, }); - writeCallbacks[0]?.(); - expect(committed).toEqual(["first"]); - - writeCallbacks[0]?.(); - expect(committed).toEqual(["first"]); + // Both submitted without waiting; each commits independently when its callback fires. + expect(writeTexts).toEqual(["first", "second"]); + expect(committed).toEqual([]); writeCallbacks[1]?.(); - expect(committed).toEqual(["first", "second"]); + expect(committed).toEqual(["second"]); + + writeCallbacks[0]?.(); + expect(committed).toEqual(["second", "first"]); + }); + + it("applies a snapshot after the writes ahead of it, never interleaving the reset", () => { + const { runtime, terminal, writeTexts, writeCallbacks } = createRuntimeWithTerminal(); + const committed: string[] = []; + + runtime.write({ + data: terminalOutput("before-a"), + onCommitted: () => { + committed.push("before-a"); + }, + }); + runtime.write({ + data: terminalOutput("before-b"), + onCommitted: () => { + committed.push("before-b"); + }, + }); + runtime.restoreOutput({ + data: terminalOutput("snap"), + onCommitted: () => { + committed.push("snap"); + }, + }); + + // Plain writes are submitted up front; the snapshot's reset+write has not run yet. + expect(writeTexts).toEqual(["before-a", "before-b"]); + + // writeCallbacks: [0]=before-a, [1]=before-b, [2]=barrier gate sentinel. + // The snapshot is gated behind the prior writes; resolving the gate runs it after them. + writeCallbacks[2]?.(); + expect(writeTexts).toEqual(["before-a", "before-b", "csnap"]); + + // Snapshot commit fires only after its own write callback, strictly after the writes. + writeCallbacks[3]?.(); + expect(committed).toEqual(["snap"]); + expect(terminal.resetCalls).toBe(0); + }); + + it("applies a barrier immediately when no writes precede it, suppressing input at once", () => { + const { runtime, writeCallbacks } = createRuntimeWithTerminal(); + const readSuppressInput = () => + (runtime as unknown as { suppressInput: boolean }).suppressInput; + + expect(readSuppressInput()).toBe(false); + + // No plain writes precede this barrier (mount), so there is nothing to gate: it starts + // at once with no sentinel, flipping suppressInput synchronously. + runtime.restoreOutput({ data: terminalOutput("snapshot") }); + expect(readSuppressInput()).toBe(true); + + // writeCallbacks[0] is the barrier's own snapshot write; committing it restores input. + writeCallbacks[0]?.(); + expect(readSuppressInput()).toBe(false); + }); + + it("gates a barrier behind a preceding plain write before suppressing input", () => { + const { runtime, writeCallbacks } = createRuntimeWithTerminal(); + const readSuppressInput = () => + (runtime as unknown as { suppressInput: boolean }).suppressInput; + + // A plain write is now ungated, so the following barrier must wait on the sentinel. + runtime.write({ data: terminalOutput("output") }); + runtime.restoreOutput({ data: terminalOutput("snapshot") }); + + // The plain write carries no onCommitted so it registers no callback; writeCallbacks[0] + // is the sentinel gate. suppressInput only flips once the gate resolves the barrier. + expect(readSuppressInput()).toBe(false); + writeCallbacks[0]?.(); + expect(readSuppressInput()).toBe(true); + + // writeCallbacks[1] is the barrier's own snapshot write; committing it restores input. + writeCallbacks[1]?.(); + expect(readSuppressInput()).toBe(false); }); it("commits pending output operations during unmount to avoid deadlock", () => { @@ -294,8 +416,21 @@ describe("terminal-emulator-runtime", () => { expect(onCommittedB).toHaveBeenCalledTimes(1); }); + it("clears ungated writes on unmount so remount barriers apply immediately", () => { + const { runtime } = createRuntimeWithTerminal(); + + runtime.write({ data: terminalOutput("before unmount") }); + runtime.unmount(); + + const remounted = attachStubTerminal(runtime); + runtime.restoreOutput({ data: terminalOutput("restored screen") }); + + expect(remounted.writeTexts).toEqual(["\u001bcrestored screen"]); + expect(remounted.writeCallbacks).toHaveLength(1); + }); + it("replays snapshots through a single write without first painting a reset terminal", () => { - const { runtime, terminal, writeTexts } = createRuntimeWithTerminal(); + const { runtime, terminal, writeTexts, writeCallbacks } = createRuntimeWithTerminal(); runtime.renderSnapshot({ state: { @@ -313,6 +448,9 @@ describe("terminal-emulator-runtime", () => { }, }); + // The snapshot is a barrier; its write applies after the gate sentinel (writeCallbacks[0]). + writeCallbacks[0]?.(); + expect(terminal.resetCalls).toBe(0); expect(writeTexts).toHaveLength(1); expect(writeTexts[0]?.startsWith("\u001bc")).toBe(true); @@ -320,10 +458,13 @@ describe("terminal-emulator-runtime", () => { }); it("restores server-rendered ANSI snapshots through the snapshot write path", () => { - const { runtime, terminal, writeTexts } = createRuntimeWithTerminal(); + const { runtime, terminal, writeTexts, writeCallbacks } = createRuntimeWithTerminal(); runtime.restoreOutput({ data: terminalOutput("restored screen") }); + // restoreOutput is a barrier (suppressInput); it applies after the gate sentinel. + writeCallbacks[0]?.(); + expect(terminal.resetCalls).toBe(0); expect(writeTexts).toEqual(["\u001bcrestored screen"]); }); diff --git a/packages/app/src/terminal/runtime/terminal-emulator-runtime.ts b/packages/app/src/terminal/runtime/terminal-emulator-runtime.ts index 7da237f6b..8f9955bf2 100644 --- a/packages/app/src/terminal/runtime/terminal-emulator-runtime.ts +++ b/packages/app/src/terminal/runtime/terminal-emulator-runtime.ts @@ -184,6 +184,15 @@ export class TerminalEmulatorRuntime { private outputOperations: TerminalOutputOperation[] = []; private inFlightOutputOperation: TerminalOutputOperation | null = null; private inFlightOutputOperationTimeout: ReturnType | null = null; + // Plain writes are submitted to xterm without waiting, so their onCommitted callbacks are + // pending until xterm commits the write. unmount() disposes xterm before those fire, so we + // track them here to flush remaining callbacks and avoid stalling upstream backpressure. + private pendingWriteCommits = new Set<() => void>(); + // True once a plain write has been submitted to xterm that no later barrier has gated + // yet. A barrier only needs the sentinel-write gate when this is set; at mount or right + // after another barrier it's false and the barrier applies immediately, saving a parse + // cycle of latency. Cleared when a barrier starts (it gates every write before it). + private hasUngatedWrites = false; private readonly inputModeDecoder = new TextDecoder(); private suppressInput = false; private readonly inputModeTracker = new TerminalInputModeTracker(); @@ -759,6 +768,10 @@ export class TerminalEmulatorRuntime { for (const operation of pendingOperations) { operation.onCommitted?.(); } + for (const commit of Array.from(this.pendingWriteCommits)) { + commit(); + } + this.hasUngatedWrites = false; this.cleanup?.(); this.cleanup = null; @@ -776,6 +789,13 @@ export class TerminalEmulatorRuntime { this.emitInputModeChange(); } + // A barrier op mutates terminal geometry or toggles suppressInput and must apply in + // isolation: clear/snapshot reset/resize the terminal, and any write carrying + // suppressInput flips input handling around the commit. Plain writes are fast-pathed. + private isBarrierOperation(operation: TerminalOutputOperation): boolean { + return operation.type !== "write" || Boolean(operation.suppressInput); + } + private processOutputQueue(): void { if (this.inFlightOutputOperation) { return; @@ -786,12 +806,90 @@ export class TerminalEmulatorRuntime { return; } + // Fast path: drain contiguous plain writes back-to-back without waiting for each + // write's completion callback. xterm buffers internally and parses in submission + // order, so serializing one frame per parse tick only adds latency under burst. + while (this.outputOperations[0] && !this.isBarrierOperation(this.outputOperations[0])) { + const writeOperation = this.outputOperations.shift(); + if (!writeOperation) { + break; + } + this.submitWrite(terminal, writeOperation); + } + const operation = this.outputOperations.shift(); if (!operation) { return; } + // The next op is a barrier. Before clear()/reset()/resize() touches the terminal, any + // ungated plain writes still parsing in xterm's buffer must finish — otherwise a + // synchronous reset could interleave with them. When none are ungated (mount, or right + // after another barrier) there is nothing to wait for, so apply the barrier at once. + if (!this.hasUngatedWrites) { + this.startBarrierOperation(terminal, operation); + return; + } + // Otherwise a zero-length sentinel write commits after every outstanding write, so + // waiting on its callback gates the barrier behind them. Plain writes never wait. + let started = false; + const startBarrier = () => { + // unmount() clears the in-flight op and disposes the terminal while we wait on the + // sentinel; bail if either changed so we don't reset/resize a torn-down terminal. + if (started || this.inFlightOutputOperation !== operation || this.terminal !== terminal) { + return; + } + started = true; + this.clearInFlightOutputTimeout(); + this.startBarrierOperation(terminal, operation); + }; this.inFlightOutputOperation = operation; + this.inFlightOutputOperationTimeout = setTimeout(startBarrier, OUTPUT_OPERATION_TIMEOUT_MS); + try { + terminal.write(EMPTY_TERMINAL_OUTPUT, startBarrier); + } catch { + startBarrier(); + } + } + + private submitWrite(terminal: Terminal, operation: TerminalOutputOperation): void { + // Synchronous per-write tracking must run in frame order; doing it here in the drain + // loop preserves that ordering even though the writes are submitted without waiting. + const text = this.inputModeDecoder.decode(operation.data, { stream: true }); + const result = this.inputModeTracker.feed(text); + if (result.changed) { + this.emitInputModeChange(); + } + this.hasUngatedWrites = true; + const onCommitted = operation.onCommitted; + if (!onCommitted) { + try { + terminal.write(operation.data); + } catch { + // Match existing behavior: a failed write still proceeds with no commit callback. + } + return; + } + const commit = () => { + if (!this.pendingWriteCommits.delete(commit)) { + return; + } + onCommitted(); + }; + this.pendingWriteCommits.add(commit); + try { + terminal.write(operation.data, commit); + } catch { + commit(); + } + } + + private startBarrierOperation(terminal: Terminal, operation: TerminalOutputOperation): void { + this.inFlightOutputOperation = operation; + // This barrier gates every write submitted before it (the sentinel guaranteed they + // parsed, or there were none), so the next barrier can skip the sentinel until the + // next write arrives. + this.hasUngatedWrites = false; const previousSuppressInput = this.suppressInput; if (operation.suppressInput) { this.suppressInput = Boolean(operation.suppressInput); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 92fbde31c..29d19c16e 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -566,6 +566,7 @@ export interface SessionOptions { clientCapabilities?: Record | null; onMessage: (msg: SessionOutboundMessage) => void; onBinaryMessage?: (frame: Uint8Array) => void; + getTransportBufferedAmount?: () => number | null; onLifecycleIntent?: (intent: SessionLifecycleIntent) => void; logger: pino.Logger; downloadTokenStore: DownloadTokenStore; @@ -757,6 +758,7 @@ export class Session { private readonly sessionId: string; private readonly onMessage: (msg: SessionOutboundMessage) => void; private readonly onBinaryMessage: ((frame: Uint8Array) => void) | null; + private readonly getTransportBufferedAmount: () => number | null; private readonly onLifecycleIntent: ((intent: SessionLifecycleIntent) => void) | null; private readonly sessionLogger: pino.Logger; private readonly paseoHome: string; @@ -861,6 +863,7 @@ export class Session { clientCapabilities, onMessage, onBinaryMessage, + getTransportBufferedAmount, onLifecycleIntent, logger, downloadTokenStore, @@ -906,6 +909,7 @@ export class Session { this.sessionId = uuidv4(); this.onMessage = onMessage; this.onBinaryMessage = onBinaryMessage ?? null; + this.getTransportBufferedAmount = getTransportBufferedAmount ?? (() => 0); this.onLifecycleIntent = onLifecycleIntent ?? null; this.downloadTokenStore = downloadTokenStore; this.pushTokenStore = pushTokenStore; @@ -946,6 +950,7 @@ export class Session { }, clientSupportsWrapReflow: () => this.clientCapabilities.has(CLIENT_CAPS.terminalReflowableSnapshot), + getClientBufferedAmount: () => this.getTransportBufferedAmount(), }); this.createAgentLifecycleDispatch = new CreateAgentLifecycleDispatch({ paseoHome: this.paseoHome, @@ -8687,13 +8692,18 @@ export class Session { * Emit a message to the client */ private emit(msg: SessionOutboundMessage): void { - this.sessionLogger.trace( - { - messageType: msg.type, - payloadBytes: JSON.stringify(msg).length, - }, - "agent.session.outbound", - ); + // JSON.stringify(msg) is only computed when trace is enabled — it runs for + // every outbound message otherwise, and trace is disabled by default. + // Optional-chained because test logger stubs don't implement isLevelEnabled. + if (this.sessionLogger.isLevelEnabled?.("trace")) { + this.sessionLogger.trace( + { + messageType: msg.type, + payloadBytes: JSON.stringify(msg).length, + }, + "agent.session.outbound", + ); + } if ( msg.type === "audio_output" && (process.env.TTS_DEBUG_AUDIO_DIR || isPaseoDictationDebugEnabled()) && diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index ebc40fc9d..9bace4441 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -2,6 +2,7 @@ import { WebSocket, WebSocketServer } from "ws"; import type { IncomingMessage, Server as HTTPServer } from "http"; import { basename, join } from "path"; import { hostname as getHostname } from "node:os"; +import { monitorEventLoopDelay } from "node:perf_hooks"; import type { AgentManager } from "./agent/agent-manager.js"; import type { AgentStorage } from "./agent/agent-storage.js"; import type { DownloadTokenStore } from "./file-download/token-store.js"; @@ -379,6 +380,7 @@ export class VoiceAssistantWebSocketServer { private serverCapabilities: ServerCapabilities | undefined; private readonly runtimeMetrics = new WebSocketRuntimeMetricsWindow(); private runtimeMetricsInterval: ReturnType | null = null; + private eventLoopDelayMonitor: ReturnType | null = null; private unsubscribeSpeechReadiness: (() => void) | null = null; private unsubscribeDaemonConfigChange: (() => void) | null = null; @@ -561,6 +563,8 @@ export class VoiceAssistantWebSocketServer { } private startRuntimeMetricsInterval(): void { + this.eventLoopDelayMonitor = monitorEventLoopDelay({ resolution: 10 }); + this.eventLoopDelayMonitor.enable(); const runtimeMetricsInterval = setInterval(() => { this.flushRuntimeMetrics(); }, WS_RUNTIME_METRICS_FLUSH_MS); @@ -568,6 +572,23 @@ export class VoiceAssistantWebSocketServer { (runtimeMetricsInterval as unknown as { unref?: () => void }).unref?.(); } + // Main-loop stall visibility: terminal frames and agent traffic share one event + // loop, so delay percentiles here are the ground truth for "the daemon is busy". + private snapshotEventLoopDelay(): { p50Ms: number; p99Ms: number; maxMs: number } | null { + const monitor = this.eventLoopDelayMonitor; + if (!monitor) { + return null; + } + const toMs = (nanoseconds: number): number => Math.round(nanoseconds / 1e5) / 10; + const snapshot = { + p50Ms: toMs(monitor.percentile(50)), + p99Ms: toMs(monitor.percentile(99)), + maxMs: toMs(monitor.max), + }; + monitor.reset(); + return snapshot; + } + private verifyWsUpgrade( req: IncomingMessage, allowedOrigins: Set, @@ -678,6 +699,8 @@ export class VoiceAssistantWebSocketServer { this.runtimeMetricsInterval = null; } this.flushRuntimeMetrics({ final: true }); + this.eventLoopDelayMonitor?.disable(); + this.eventLoopDelayMonitor = null; const uniqueConnections = new Set([ ...this.sessions.values(), @@ -861,6 +884,22 @@ export class VoiceAssistantWebSocketServer { } this.sendBinaryToConnection(connection, frame); }, + getTransportBufferedAmount: () => { + if (!connection) { + return null; + } + // Relay-attached sockets are a WebSocketLike that doesn't expose + // bufferedAmount. Return null when no socket gives a signal so the + // terminal fallback can't mistake "no signal" for "client keeping up"; + // a direct ws reports its real buffered bytes (0 when drained). + let maxBuffered: number | null = null; + for (const socket of connection.sockets) { + if (typeof socket.bufferedAmount === "number") { + maxBuffered = Math.max(maxBuffered ?? 0, socket.bufferedAmount); + } + } + return maxBuffered; + }, onLifecycleIntent: (intent) => { this.onLifecycleIntent?.(intent); }, @@ -1641,6 +1680,7 @@ export class VoiceAssistantWebSocketServer { outboundAgentStreamAgentsTop: runtimeMetrics.outboundAgentStreamAgentsTop, outboundBinaryFrameTypesTop: runtimeMetrics.outboundBinaryFrameTypesTop, bufferedAmount: runtimeMetrics.bufferedAmount, + eventLoopDelay: this.snapshotEventLoopDelay(), runtime: sessionMetrics, latency: runtimeMetrics.latency, agents: agentSnapshot, diff --git a/packages/server/src/terminal/terminal-output-coalescer.test.ts b/packages/server/src/terminal/terminal-output-coalescer.test.ts index 55f732d34..da6ae5f89 100644 --- a/packages/server/src/terminal/terminal-output-coalescer.test.ts +++ b/packages/server/src/terminal/terminal-output-coalescer.test.ts @@ -3,6 +3,7 @@ import { TerminalOutputCoalescer } from "./terminal-output-coalescer.js"; function createHarness() { let nextTimer = 1; + let now = 1000; const scheduled = new Map void; delayMs: number }>(); const flushes: Array<{ payload: string; chars: number; bytes: number }> = []; const coalescer = new TerminalOutputCoalescer({ @@ -17,6 +18,7 @@ function createHarness() { scheduled.delete(id); }) as typeof clearTimeout, }, + now: () => now, onFlush: ({ payload, chars, bytes }) => { flushes.push({ payload: payload.toString("utf8"), chars, bytes }); }, @@ -26,6 +28,9 @@ function createHarness() { coalescer, flushes, scheduled, + advance(ms: number) { + now += ms; + }, runScheduled() { const callbacks = Array.from(scheduled.values()); scheduled.clear(); @@ -37,10 +42,22 @@ function createHarness() { } describe("TerminalOutputCoalescer", () => { - test("coalesces output into one 5ms flush window", () => { - const { coalescer, flushes, runScheduled, scheduled } = createHarness(); + test("flushes the first chunk immediately on the leading edge", () => { + const { coalescer, flushes, scheduled } = createHarness(); coalescer.handle("a"); + + // No timer scheduled: the leading-edge flush happened synchronously. + expect(scheduled.size).toBe(0); + expect(flushes).toEqual([{ payload: "a", chars: 1, bytes: 1 }]); + }); + + test("coalesces a burst that follows the leading-edge flush into one trailing flush", () => { + const { coalescer, flushes, runScheduled, scheduled } = createHarness(); + + // First chunk flushes immediately (leading edge). + coalescer.handle("a"); + // Subsequent chunks within the window accumulate behind a trailing timer. coalescer.handle("b"); coalescer.handle("é"); @@ -48,32 +65,86 @@ describe("TerminalOutputCoalescer", () => { expect(Array.from(scheduled.values())).toEqual([ { callback: expect.any(Function), delayMs: 5 }, ]); - expect(flushes).toEqual([]); + expect(flushes).toEqual([{ payload: "a", chars: 1, bytes: 1 }]); runScheduled(); - expect(flushes).toEqual([{ payload: "abé", chars: 3, bytes: 4 }]); + expect(flushes).toEqual([ + { payload: "a", chars: 1, bytes: 1 }, + { payload: "bé", chars: 2, bytes: 3 }, + ]); + }); + + test("flushes immediately again once the window has elapsed", () => { + const { coalescer, flushes, advance, scheduled } = createHarness(); + + coalescer.handle("a"); + expect(flushes).toEqual([{ payload: "a", chars: 1, bytes: 1 }]); + + advance(5); + coalescer.handle("b"); + + expect(scheduled.size).toBe(0); + expect(flushes).toEqual([ + { payload: "a", chars: 1, bytes: 1 }, + { payload: "b", chars: 1, bytes: 1 }, + ]); }); test("manual flush drains pending output and cancels the scheduled flush", () => { const { coalescer, flushes, runScheduled, scheduled } = createHarness(); + // Leading-edge flush, then accumulate a burst behind the trailing timer. coalescer.handle("hello"); + coalescer.handle(" world"); coalescer.flush(); runScheduled(); expect(scheduled.size).toBe(0); - expect(flushes).toEqual([{ payload: "hello", chars: 5, bytes: 5 }]); + expect(flushes).toEqual([ + { payload: "hello", chars: 5, bytes: 5 }, + { payload: " world", chars: 6, bytes: 6 }, + ]); }); test("dispose drops pending output", () => { const { coalescer, flushes, runScheduled, scheduled } = createHarness(); + // First chunk flushes on the leading edge; second one stays pending. + coalescer.handle("done"); coalescer.handle("pending"); coalescer.dispose(); runScheduled(); expect(scheduled.size).toBe(0); + expect(flushes).toEqual([{ payload: "done", chars: 4, bytes: 4 }]); + }); + + test("markFlushed keeps the next chunk on the trailing path", () => { + const { coalescer, flushes, runScheduled, scheduled } = createHarness(); + + // Simulate a frame emitted out-of-band (e.g. a snapshot) right before output. + coalescer.markFlushed(); + coalescer.handle("post-snapshot"); + + // No immediate flush: the chunk waits for the trailing timer. + expect(scheduled.size).toBe(1); expect(flushes).toEqual([]); + + runScheduled(); + expect(flushes).toEqual([{ payload: "post-snapshot", chars: 13, bytes: 13 }]); + }); + + test("preserves ordering across leading and trailing flushes", () => { + const { coalescer, flushes, runScheduled, advance } = createHarness(); + + coalescer.handle("1"); + coalescer.handle("2"); + coalescer.handle("3"); + runScheduled(); + advance(5); + coalescer.handle("4"); + + expect(flushes.map((f) => f.payload)).toEqual(["1", "23", "4"]); }); }); diff --git a/packages/server/src/terminal/terminal-output-coalescer.ts b/packages/server/src/terminal/terminal-output-coalescer.ts index 0005eb995..f9e7e1b97 100644 --- a/packages/server/src/terminal/terminal-output-coalescer.ts +++ b/packages/server/src/terminal/terminal-output-coalescer.ts @@ -12,6 +12,7 @@ export interface TerminalOutputCoalescerFlush { export interface TerminalOutputCoalescerOptions { timers: TerminalOutputCoalescerTimers; flushDelayMs?: number; + now?: () => number; onFlush: (payload: TerminalOutputCoalescerFlush) => void; } @@ -20,15 +21,18 @@ const DEFAULT_FLUSH_DELAY_MS = 5; export class TerminalOutputCoalescer { private readonly timers: TerminalOutputCoalescerTimers; private readonly flushDelayMs: number; + private readonly now: () => number; private readonly onFlush: (payload: TerminalOutputCoalescerFlush) => void; private chunks: Buffer[] = []; private bytes = 0; private chars = 0; private flushTimer: ReturnType | null = null; + private lastFlushAt: number | null = null; constructor(options: TerminalOutputCoalescerOptions) { this.timers = options.timers; this.flushDelayMs = options.flushDelayMs ?? DEFAULT_FLUSH_DELAY_MS; + this.now = options.now ?? Date.now; this.onFlush = options.onFlush; } @@ -42,7 +46,16 @@ export class TerminalOutputCoalescer { this.bytes += chunk.byteLength; this.chars += data.length; + // Leading edge: if nothing is pending and no flush happened within the last + // flushDelayMs, flush immediately so interactive echo isn't delayed a full + // window. Otherwise accumulate and let the trailing timer drain the burst. if (!this.flushTimer) { + const elapsed = + this.lastFlushAt === null ? Number.POSITIVE_INFINITY : this.now() - this.lastFlushAt; + if (elapsed >= this.flushDelayMs) { + this.flush(); + return; + } this.flushTimer = this.timers.setTimeout(() => { this.flushTimer = null; this.flush(); @@ -62,10 +75,19 @@ export class TerminalOutputCoalescer { const bytes = this.bytes; const chars = this.chars; this.clearPending(); + this.lastFlushAt = this.now(); this.onFlush({ payload, bytes, chars }); } + // Record that a frame was emitted out-of-band on this stream (e.g. a snapshot + // frame sent directly, bypassing the coalescer). This keeps the next handled + // chunk on the trailing path instead of leading-edge flushing it back-to-back + // with that frame, preserving the small gap consumers rely on after a snapshot. + markFlushed(): void { + this.lastFlushAt = this.now(); + } + dispose(): void { this.clearTimer(); this.clearPending(); diff --git a/packages/server/src/terminal/terminal-restore.ts b/packages/server/src/terminal/terminal-restore.ts index 6b31f2ac3..ffc1bd9fa 100644 --- a/packages/server/src/terminal/terminal-restore.ts +++ b/packages/server/src/terminal/terminal-restore.ts @@ -9,6 +9,13 @@ import type { TerminalStateSnapshot, TerminalStateSnapshotOptions } from "./term export const MAX_TERMINAL_OUTPUT_FRAME_BYTES = 256 * 1024; +// A client is only forced onto the snapshot catch-up path once its transport is +// genuinely backed up — measured by the socket's bufferedAmount, not by how much +// the terminal has produced. Set well above any normal in-flight burst (a +// keeping-up client drains continuously and never approaches this) but low +// enough that a stalled socket crosses it long before a large burst finishes. +export const MAX_CLIENT_BUFFERED_BYTES = 4 * 1024 * 1024; + const DEFAULT_VISIBLE_RESTORE_SCROLLBACK_LINES = 200; const MAX_VISIBLE_RESTORE_SCROLLBACK_LINES = 500; diff --git a/packages/server/src/terminal/terminal-session-controller.test.ts b/packages/server/src/terminal/terminal-session-controller.test.ts index ccca2f7e2..a729f8d0e 100644 --- a/packages/server/src/terminal/terminal-session-controller.test.ts +++ b/packages/server/src/terminal/terminal-session-controller.test.ts @@ -168,6 +168,12 @@ async function flushMicrotasks(): Promise { await new Promise((resolve) => setTimeout(resolve, 0)); } +// The coalescer drains on a 5ms trailing timer; wait past it (and any async +// snapshot round-trip it kicks off) before asserting on emitted frames. +async function waitForCoalescerFlush(): Promise { + await new Promise((resolve) => setTimeout(resolve, 30)); +} + describe("terminal-session-controller wrap-flag gating", () => { function setup(clientSupportsWrapReflow?: () => boolean): { controller: TerminalSessionController; @@ -387,3 +393,133 @@ describe("terminal-session-controller subdirectory aggregation", () => { ]); }); }); + +describe("terminal-session-controller backpressure snapshot fallback", () => { + async function setup(getClientBufferedAmount: () => number | null): Promise<{ + pushOutput: (data: string) => void; + frames: TerminalStreamFrame[]; + }> { + let terminalListener: ((message: ServerMessage) => void) | null = null; + const terminal: TerminalSession = { + id: "term-1", + name: "Terminal", + cwd: "/tmp", + send: vi.fn(), + subscribe: (listener) => { + terminalListener = listener; + // Legacy stream: a snapshot arrives on subscribe (one Snapshot frame), + // after which output streams through the coalescer as Output frames. + queueMicrotask(() => + listener({ type: "snapshot", state: terminalState("live"), revision: 1 }), + ); + return vi.fn(); + }, + onExit: () => vi.fn(), + onCommandFinished: () => vi.fn(), + onTitleChange: () => vi.fn(), + getSize: () => ({ rows: 1, cols: 80 }), + getState: () => terminalState("live"), + getStateSnapshot: () => ({ state: terminalState("live"), revision: 1 }), + getReplayPreamble: () => "", + getTitle: () => undefined, + setTitle: vi.fn(), + getExitInfo: () => null, + kill: vi.fn(), + killAndWait: vi.fn(), + }; + const terminalManager = { + getTerminals: vi.fn(), + createTerminal: vi.fn(), + registerCwdEnv: vi.fn(), + getTerminal: vi.fn(() => terminal), + getTerminalState: vi.fn(() => + Promise.resolve({ state: terminalState("live"), revision: 1 }), + ), + setTerminalTitle: vi.fn(), + killTerminal: vi.fn(), + killTerminalAndWait: vi.fn(), + captureTerminal: vi.fn(), + listDirectories: vi.fn(() => []), + killAll: vi.fn(), + subscribeTerminalsChanged: vi.fn(() => vi.fn()), + } as unknown as TerminalManager; + + const frames: TerminalStreamFrame[] = []; + const controller = new TerminalSessionController({ + terminalManager, + emit: vi.fn(), + emitBinary: (bytes) => { + const frame = decodeTerminalStreamFrame(bytes); + if (frame) { + frames.push(frame); + } + }, + hasBinaryChannel: () => true, + isPathWithinRoot: () => false, + sessionLogger: createLogger(), + getClientBufferedAmount, + }); + + await controller.dispatch({ + type: "subscribe_terminal_request", + terminalId: "term-1", + requestId: "req-1", + }); + await waitForCoalescerFlush(); + // Drop the initial subscribe snapshot frame so each test only sees frames + // produced by the output it pushes. + frames.length = 0; + + return { + pushOutput: (data) => terminalListener?.({ type: "output", data, revision: 2 }), + frames, + }; + } + + test("streams all output without a snapshot when the client keeps up", async () => { + const { pushOutput, frames } = await setup(() => 0); + + const chunk = "x".repeat(300 * 1024); + pushOutput(chunk); + await waitForCoalescerFlush(); + + expect(frames.some((frame) => frame.opcode === TerminalStreamOpcode.Snapshot)).toBe(false); + const outputFrames = frames.filter((frame) => frame.opcode === TerminalStreamOpcode.Output); + expect(outputFrames.length).toBeGreaterThan(0); + const receivedBytes = outputFrames.reduce( + (total, frame) => total + frame.payload.byteLength, + 0, + ); + expect(receivedBytes).toBe(Buffer.byteLength(chunk, "utf8")); + }); + + test("falls back to a snapshot and resets the byte counter when the client is backed up", async () => { + const { pushOutput, frames } = await setup(() => 8 * 1024 * 1024); + + pushOutput("y".repeat(300 * 1024)); + await waitForCoalescerFlush(); + + expect(frames.some((frame) => frame.opcode === TerminalStreamOpcode.Snapshot)).toBe(true); + + // After the snapshot the byte counter is reset, so a small follow-up chunk + // streams as Output rather than tripping the fallback again. + frames.length = 0; + pushOutput("z".repeat(1024)); + await waitForCoalescerFlush(); + + expect(frames.some((frame) => frame.opcode === TerminalStreamOpcode.Snapshot)).toBe(false); + expect(frames.some((frame) => frame.opcode === TerminalStreamOpcode.Output)).toBe(true); + }); + + test("falls back to a snapshot at the byte threshold when no backpressure signal exists", async () => { + // A null reading means the transport (e.g. the multiplexed relay socket) gives + // no signal; we can't distinguish a slow client from a fast one, so we keep the + // unconditional catch-up so a slow relay client can't fall unboundedly behind. + const { pushOutput, frames } = await setup(() => null); + + pushOutput("r".repeat(300 * 1024)); + await waitForCoalescerFlush(); + + expect(frames.some((frame) => frame.opcode === TerminalStreamOpcode.Snapshot)).toBe(true); + }); +}); diff --git a/packages/server/src/terminal/terminal-session-controller.ts b/packages/server/src/terminal/terminal-session-controller.ts index 88f876232..210e5a75b 100644 --- a/packages/server/src/terminal/terminal-session-controller.ts +++ b/packages/server/src/terminal/terminal-session-controller.ts @@ -22,6 +22,7 @@ import { } from "@getpaseo/protocol/binary-frames/index"; import { TerminalOutputCoalescer } from "./terminal-output-coalescer.js"; import { + MAX_CLIENT_BUFFERED_BYTES, MAX_TERMINAL_OUTPUT_FRAME_BYTES, encodeLegacyTerminalSnapshotFrame, encodeTerminalRestoreFrame, @@ -70,6 +71,14 @@ export interface TerminalSessionControllerOptions { // daemon attaches per-row soft-wrap flags to snapshots; otherwise it omits them // so old (strict-schema) clients still parse the snapshot. clientSupportsWrapReflow?: () => boolean; + // Current max bytes queued on the client's transport(s) but not yet sent. + // Drives the snapshot catch-up fallback: a keeping-up client reports ~0 and + // keeps streaming; a backed-up client trips the snapshot path. Defaults to a + // constant 0 (no backpressure signal) so callers without a transport always + // stream. + // Bytes queued on the client transport but not yet sent, or null when the + // transport exposes no backpressure signal (e.g. the multiplexed relay socket). + getClientBufferedAmount?: () => number | null; } export interface TerminalSessionControllerMetrics { @@ -111,6 +120,7 @@ export class TerminalSessionController { private readonly sessionLogger: pino.Logger; private readonly listTerminalWorkspaceRoots: () => Promise; private readonly clientSupportsWrapReflow: () => boolean; + private readonly getClientBufferedAmount: () => number | null; private readonly subscribedDirectories = new Set(); private unsubscribeTerminalsChanged: (() => void) | null = null; @@ -128,6 +138,7 @@ export class TerminalSessionController { this.sessionLogger = options.sessionLogger; this.listTerminalWorkspaceRoots = options.listTerminalWorkspaceRoots ?? (async () => []); this.clientSupportsWrapReflow = options.clientSupportsWrapReflow ?? (() => false); + this.getClientBufferedAmount = options.getClientBufferedAmount ?? (() => 0); } start(): void { @@ -757,7 +768,21 @@ export class TerminalSessionController { return; } activeStream.outputBytesSinceSnapshot += payload.byteLength; - if (activeStream.outputBytesSinceSnapshot > MAX_TERMINAL_OUTPUT_FRAME_BYTES) { + // Catch up via a snapshot only when the client is BOTH far behind in + // produced output AND actually backed up on the wire. A client that + // keeps draining reports ~0 buffered, so it streams continuously even + // past the byte threshold. outputBytesSinceSnapshot keeps accumulating + // in that case — it's harmless, it only gates the snapshot decision at + // the instant backpressure appears, and trySendSnapshot resets it to 0. + // A null reading means the transport exposes no backpressure signal + // (e.g. the multiplexed relay socket); there we can't tell a slow client + // from a fast one, so fall back unconditionally at the byte threshold to + // keep a slow relay client from falling unboundedly behind. + const clientBufferedAmount = this.getClientBufferedAmount(); + if ( + activeStream.outputBytesSinceSnapshot > MAX_TERMINAL_OUTPUT_FRAME_BYTES && + (clientBufferedAmount === null || clientBufferedAmount > MAX_CLIENT_BUFFERED_BYTES) + ) { activeStream.restore = resolveRestoreAfterOutputOverflow(activeStream.restore); activeStream.needsSnapshot = true; void this.trySendSnapshot(activeStream); @@ -877,6 +902,9 @@ export class TerminalSessionController { snapshot, }), ); + // The snapshot 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 }; } @@ -908,6 +936,9 @@ export class TerminalSessionController { snapshot, }), ); + // 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 }; } diff --git a/packages/server/src/terminal/terminal-worker-process.ts b/packages/server/src/terminal/terminal-worker-process.ts index d85e163cc..09c6ab881 100644 --- a/packages/server/src/terminal/terminal-worker-process.ts +++ b/packages/server/src/terminal/terminal-worker-process.ts @@ -1,14 +1,17 @@ import { createTerminalManager } from "./terminal-manager.js"; import { captureTerminalLines } from "./terminal-capture.js"; -import type { TerminalSession } from "./terminal.js"; +import { TerminalOutputCoalescer } from "./terminal-output-coalescer.js"; +import type { TerminalSession, TerminalStateSnapshotOptions } from "./terminal.js"; import type { TerminalWorkerRequest, + TerminalWorkerStateResult, TerminalWorkerToParentMessage, WorkerTerminalInfo, } from "./terminal-worker-protocol.js"; const manager = createTerminalManager(); const unsubscribeByTerminalId = new Map void>>(); +const outputCoalescerByTerminalId = new Map(); let ipcClosing = false; function sendToParent(message: TerminalWorkerToParentMessage): void { @@ -26,6 +29,16 @@ function sendToParent(message: TerminalWorkerToParentMessage): void { } } +function buildTerminalStateResult( + session: TerminalSession | undefined, + options?: TerminalStateSnapshotOptions, +): TerminalWorkerStateResult { + if (!session) { + return null; + } + return { ...session.getStateSnapshot(options), replayPreamble: session.getReplayPreamble() }; +} + function toTerminalInfo(session: TerminalSession): WorkerTerminalInfo { return { id: session.id, @@ -47,12 +60,44 @@ function clearTerminalSubscriptions(terminalId: string): void { } } unsubscribeByTerminalId.delete(terminalId); + const coalescer = outputCoalescerByTerminalId.get(terminalId); + if (coalescer) { + coalescer.dispose(); + outputCoalescerByTerminalId.delete(terminalId); + } } function watchTerminal(session: TerminalSession): void { clearTerminalSubscriptions(session.id); + // Coalesce pty output chunks into a single IPC message per ~5ms window so a + // burst of small chunks no longer costs one process.send each. The batch + // carries the LAST chunk's revision (the highest) so downstream snapshot + // replay dedup stays correct. + let pendingOutputRevision: number | undefined; + const outputCoalescer = new TerminalOutputCoalescer({ + timers: { setTimeout, clearTimeout }, + onFlush: ({ payload }) => { + const revision = pendingOutputRevision; + pendingOutputRevision = undefined; + sendToParent({ + type: "terminalMessage", + terminalId: session.id, + message: { type: "output", data: payload.toString("utf8"), revision }, + }); + }, + }); + outputCoalescerByTerminalId.set(session.id, outputCoalescer); + const unsubscribeMessage = session.subscribe((message) => { + if (message.type === "output") { + pendingOutputRevision = message.revision; + outputCoalescer.handle(message.data); + return; + } + // Non-output messages (snapshot/snapshotReady/titleChange) must not jump + // ahead of buffered output: flush the coalescer first, then forward. + outputCoalescer.flush(); sendToParent({ type: "terminalMessage", terminalId: session.id, @@ -60,6 +105,7 @@ function watchTerminal(session: TerminalSession): void { }); }); const unsubscribeExit = session.onExit((info) => { + outputCoalescer.flush(); clearTerminalSubscriptions(session.id); sendToParent({ type: "terminalExit", @@ -68,6 +114,7 @@ function watchTerminal(session: TerminalSession): void { }); }); const unsubscribeTitle = session.onTitleChange((title) => { + outputCoalescer.flush(); sendToParent({ type: "terminalTitleChange", terminalId: session.id, @@ -75,6 +122,7 @@ function watchTerminal(session: TerminalSession): void { }); }); const unsubscribeCommandFinished = session.onCommandFinished((info) => { + outputCoalescer.flush(); sendToParent({ type: "terminalCommandFinished", terminalId: session.id, @@ -171,12 +219,16 @@ async function handleRequest(message: TerminalWorkerRequest): Promise { } case "getTerminalState": { - const session = manager.getTerminal(message.terminalId); + // Flush buffered output before snapshotting: the headless state already includes it, + // so if the coalescer emitted it afterward (in a batch carrying a revision past the + // snapshot's) the controller's revision dedup wouldn't drop it and the client would + // see the bytes twice. Flushing first sends them with a revision <= the snapshot's. + outputCoalescerByTerminalId.get(message.terminalId)?.flush(); sendToParent({ type: "response", requestId: message.requestId, ok: true, - result: session?.getStateSnapshot(message.options) ?? null, + result: buildTerminalStateResult(manager.getTerminal(message.terminalId), message.options), }); return; } diff --git a/packages/server/src/terminal/terminal-worker-protocol.ts b/packages/server/src/terminal/terminal-worker-protocol.ts index 6a56f7aa9..6abda8c1e 100644 --- a/packages/server/src/terminal/terminal-worker-protocol.ts +++ b/packages/server/src/terminal/terminal-worker-protocol.ts @@ -143,4 +143,6 @@ export type TerminalWorkerEvent = export type TerminalWorkerToParentMessage = TerminalWorkerResponse | TerminalWorkerEvent; export type TerminalWorkerCaptureResult = CaptureTerminalLinesResult; +// The worker fills TerminalStateSnapshot.replayPreamble on getTerminalState so +// the parent can cache the input-mode preamble instead of re-deriving it. export type TerminalWorkerStateResult = TerminalStateSnapshot | null; diff --git a/packages/server/src/terminal/terminal.test.ts b/packages/server/src/terminal/terminal.test.ts index 7160cc25e..0ac68b4f7 100644 --- a/packages/server/src/terminal/terminal.test.ts +++ b/packages/server/src/terminal/terminal.test.ts @@ -334,6 +334,50 @@ describe("createTerminal", () => { expect(Array.isArray(exitInfo.lastOutputLines)).toBe(true); expect(session.getExitInfo()).toEqual(exitInfo); }); + + it("retains the final lines after output far exceeds the recent-output char cap", async () => { + // Emit many small chunks whose total length is well past the 16000-char + // recent-output cap, so the chunk-trimming path runs repeatedly. The final + // distinctive lines must survive in the exit summary. The process is kept + // alive (no self-exit) so we can wait until the last chunk has actually been + // parsed by the headless terminal before killing — otherwise the exit + // summary races the async xterm parse and reads a stale buffer. + const session = trackSession( + await createTerminal({ + cwd: realpathSync(tmpdir()), + command: process.execPath, + args: [ + "-e", + "for (let i = 0; i < 3000; i++) { process.stdout.write(`line-${i}\\n`); } setInterval(() => {}, 1000);", + ], + }), + ); + + // Poll the parsed buffer until the final line lands, so the exit summary + // (which reads the headless buffer) can't race the async xterm parse. + const finalLineParsed = (): boolean => { + const state = session.getState(); + const rowText = (row: { char: string }[]): string => row.map((cell) => cell.char).join(""); + return [...state.scrollback, ...state.grid].some((row) => rowText(row).includes("line-2999")); + }; + const deadline = Date.now() + 10000; + while (!finalLineParsed()) { + if (Date.now() > deadline) { + throw new Error("Timed out waiting for the final line to be parsed"); + } + await new Promise((resolve) => setTimeout(resolve, 20)); + } + + const exitInfo = await new Promise>>( + (resolve) => { + session.onExit((info) => resolve(info)); + session.kill(); + }, + ); + + expect(exitInfo.lastOutputLines.length).toBeGreaterThan(0); + expect(exitInfo.lastOutputLines[exitInfo.lastOutputLines.length - 1]).toBe("line-2999"); + }); }); describe.skipIf(isPlatform("win32"))("send input", () => { diff --git a/packages/server/src/terminal/terminal.ts b/packages/server/src/terminal/terminal.ts index 1d2fefe53..41fef38ab 100644 --- a/packages/server/src/terminal/terminal.ts +++ b/packages/server/src/terminal/terminal.ts @@ -36,6 +36,9 @@ export interface TerminalCommandFinishedInfo { export interface TerminalStateSnapshot { state: TerminalState; revision: number; + // Input-mode replay preamble at snapshot time. Populated by the terminal + // worker so the daemon main loop doesn't have to re-derive it from output. + replayPreamble?: string; } export interface TerminalStateSnapshotOptions { @@ -58,7 +61,7 @@ export type ClientMessage = export type ServerMessage = | { type: "output"; data: string; revision?: number } | { type: "snapshot"; state: TerminalState; revision?: number } - | { type: "snapshotReady"; revision?: number } + | { type: "snapshotReady"; revision?: number; replayPreamble?: string } | { type: "titleChange"; title?: string }; export interface TerminalSession { @@ -606,7 +609,12 @@ export async function createTerminal(options: CreateTerminalOptions): Promise void>(); let exitInfo: TerminalExitInfo | null = null; - let recentOutputText = ""; + // Recent output is retained as whole chunks plus a running char length so we + // avoid reallocating a ~16KB string on every pty chunk. We keep enough whole + // chunks that their join always contains at least the last + // TERMINAL_EXIT_OUTPUT_CHAR_LIMIT chars; the exact tail is sliced at exit. + const recentOutputChunks: string[] = []; + let recentOutputLength = 0; let title: string | undefined; let titleMode: "auto" | "manual" = presetTitle?.trim() ? "manual" : "auto"; let pendingTitle: string | undefined; @@ -777,7 +785,10 @@ export async function createTerminal(options: CreateTerminalOptions): Promise 0 ? lastOutputLines - : extractLastOutputLinesFromText(recentOutputText, TERMINAL_EXIT_OUTPUT_LINE_LIMIT), + : extractLastOutputLinesFromText( + recentOutputChunks.join("").slice(-TERMINAL_EXIT_OUTPUT_CHAR_LIMIT), + TERMINAL_EXIT_OUTPUT_LINE_LIMIT, + ), }; } @@ -803,6 +814,8 @@ export async function createTerminal(options: CreateTerminalOptions): Promise TERMINAL_EXIT_OUTPUT_CHAR_LIMIT) { - recentOutputText = recentOutputText.slice(-TERMINAL_EXIT_OUTPUT_CHAR_LIMIT); + recentOutputChunks.push(data); + recentOutputLength += data.length; + // Drop whole leading chunks while the rest still covers the char limit, so + // the retained join always contains at least the last limit chars. + while ( + recentOutputChunks.length > 1 && + recentOutputLength - recentOutputChunks[0].length >= TERMINAL_EXIT_OUTPUT_CHAR_LIMIT + ) { + recentOutputLength -= recentOutputChunks[0].length; + recentOutputChunks.shift(); + } + // We never drop the last chunk, so a single chunk larger than the cap would + // grow the buffer unbounded; slice its tail to keep the cap hard. + if (recentOutputChunks.length === 1 && recentOutputLength > TERMINAL_EXIT_OUTPUT_CHAR_LIMIT) { + const tail = recentOutputChunks[0].slice(-TERMINAL_EXIT_OUTPUT_CHAR_LIMIT); + recentOutputChunks[0] = tail; + recentOutputLength = tail.length; } writeOutputToHeadless(data); }); @@ -993,7 +1020,13 @@ export async function createTerminal(options: CreateTerminalOptions): Promise cell.char) + .join("") + .trimEnd(); +} + function getVisibleTextFromState(state: TerminalState): string { - return state.grid - .map((row) => - row - .map((cell) => cell.char) - .join("") - .trimEnd(), - ) - .join("\n"); + return state.grid.map(rowToText).join("\n"); +} + +function getRenderedTextFromState(state: TerminalState): string { + return [...state.scrollback, ...state.grid].map(rowToText).join("\n"); +} + +function readRenderedTokenIndexesFromState(state: TerminalState): number[] { + const indexes: number[] = []; + for (const match of getRenderedTextFromState(state).matchAll(/\[(\d+)\]/g)) { + const value = match[1]; + if (value !== undefined) { + indexes.push(Number(value)); + } + } + return indexes; } function createTerminalState(): TerminalState { @@ -176,6 +194,69 @@ it("creates a terminal through the worker and streams output", async () => { expect(snapshots).toBe(snapshotsBeforeOutput); }); +it("delivers rapid small writes complete and in order through worker coalescing", async () => { + const cwd = mkdtempSync(join(tmpdir(), "worker-terminal-manager-coalesce-")); + temporaryDirs.push(cwd); + const burstGatePath = join(cwd, "burst-ready"); + manager = createWorkerTerminalManager(); + // The file gate starts the burst after this test subscribes without depending + // on platform-specific PTY input echo/canonical-mode behavior. + const session = trackTerminal( + await manager.createTerminal({ + cwd, + env: { PASEO_TERMINAL_BURST_GATE: burstGatePath }, + ...nodeTerminalCommand(` + const fs = require("node:fs"); + const gatePath = process.env.PASEO_TERMINAL_BURST_GATE; + const gate = setInterval(() => { + if (!gatePath || !fs.existsSync(gatePath)) { + return; + } + clearInterval(gate); + for (let i = 0; i < 500; i++) { + process.stdout.write("[" + i + "]\\n"); + } + }, 10); + setInterval(() => {}, 1000); + `), + }), + ); + + const events: Array<{ type: string; data?: string }> = []; + session.subscribe((message) => { + if (message.type === "output") { + events.push({ type: "output", data: message.data }); + } else if (message.type === "snapshot" || message.type === "snapshotReady") { + events.push({ type: message.type }); + } + }); + + // Let the snapshot subscription settle, then trigger the burst. + await new Promise((resolve) => setTimeout(resolve, 100)); + writeFileSync(burstGatePath, "go"); + + const expected = Array.from({ length: 500 }, (_, index) => index); + let received: number[] = []; + await waitForCondition(async () => { + const snapshot = await manager!.getTerminalState(session.id); + received = snapshot ? readRenderedTokenIndexesFromState(snapshot.state) : []; + return expected.every((value, index) => received[index] === value); + }, 10000); + + expect(received).toEqual(expected); + + await waitForCondition(() => events.some((event) => event.type === "output"), 10000); + + // No snapshot may land after output it should have preceded: every snapshot + // event must come before the first output event. + const firstOutputIndex = events.findIndex((event) => event.type === "output"); + expect(firstOutputIndex).toBeGreaterThanOrEqual(0); + const snapshotAfterOutput = events + .slice(firstOutputIndex + 1) + .find((event) => event.type === "snapshot" || event.type === "snapshotReady"); + expect(snapshotAfterOutput).toBeUndefined(); +}); + it("pulls fresh terminal state from the worker authority", async () => { const cwd = mkdtempSync(join(tmpdir(), "worker-terminal-manager-state-")); temporaryDirs.push(cwd); @@ -200,6 +281,37 @@ it("pulls fresh terminal state from the worker authority", async () => { expect(visibleText).toContain("worker-state-ready"); }); +// Windows ConPTY normalizes away the kitty keyboard escape the child writes, so it +// never reaches the worker's input-mode tracker and the preamble stays empty. The +// 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", + async () => { + const cwd = mkdtempSync(join(tmpdir(), "worker-terminal-manager-preamble-")); + temporaryDirs.push(cwd); + manager = createWorkerTerminalManager(); + // \x1b[>1u pushes kitty keyboard flag 1, which the worker's input-mode + // tracker records and reflects in its replay preamble (\x1b[=1;1u). + const session = trackTerminal( + await manager.createTerminal({ + cwd, + ...nodeTerminalCommand(` + process.stdout.write("\\u001b[>1u"); + setInterval(() => {}, 1000); + `), + }), + ); + + await waitForCondition(async () => { + const snapshot = await manager!.getTerminalState(session.id); + return snapshot !== null && session.getReplayPreamble() === "\x1b[=1;1u"; + }, 10000); + + expect(session.getReplayPreamble()).toBe("\x1b[=1;1u"); + }, +); + it("refreshes cached terminal title after worker title changes", async () => { const cwd = mkdtempSync(join(tmpdir(), "worker-terminal-manager-title-")); temporaryDirs.push(cwd); diff --git a/packages/server/src/terminal/worker-terminal-manager.ts b/packages/server/src/terminal/worker-terminal-manager.ts index fdd29e7a4..3741d38ca 100644 --- a/packages/server/src/terminal/worker-terminal-manager.ts +++ b/packages/server/src/terminal/worker-terminal-manager.ts @@ -2,7 +2,6 @@ import { fork } from "node:child_process"; import { fileURLToPath } from "node:url"; import { randomUUID } from "node:crypto"; import type { TerminalState } from "@getpaseo/protocol/messages"; -import { TerminalInputModeTracker } from "@getpaseo/protocol/terminal-input-mode"; import type { ClientMessage, ServerMessage, @@ -44,7 +43,10 @@ interface PendingRequest { interface WorkerTerminalRecord { info: WorkerTerminalInfo; state: TerminalState; - inputModeTracker: TerminalInputModeTracker; + // Cached input-mode preamble from the worker (the authoritative tracker lives + // in the worker process). Refreshed on every getTerminalState response and on + // the snapshotReady event that precedes a live-restore replay. + replayPreamble: string; exitInfo: TerminalExitInfo | null; messageListeners: Set<(msg: ServerMessage) => void>; exitListeners: Set<(info: TerminalExitInfo) => void>; @@ -171,7 +173,7 @@ export function createWorkerTerminalManager( const record: WorkerTerminalRecord = { info: cloneTerminalInfo(input.info), state: input.state, - inputModeTracker: new TerminalInputModeTracker(), + replayPreamble: "", exitInfo: null, messageListeners: new Set(), exitListeners: new Set(), @@ -266,7 +268,12 @@ export function createWorkerTerminalManager( }; }, getReplayPreamble(): string { - return record.inputModeTracker.getPreamble(); + // Refreshed from every getTerminalState response, which the controller fetches + // before every snapshot replay (legacy + visible-snapshot restore). The one + // gap is restore.mode === "live", which replays without fetching state — there + // this can be stale/empty. No client sends "live" today; revisit (ship the + // preamble on the worker's snapshotReady) if one ever does. + return record.replayPreamble; }, getTitle(): string | undefined { return record.info.title; @@ -333,8 +340,8 @@ export function createWorkerTerminalManager( if (message.message.type === "snapshot") { record.state = message.message.state; } - if (message.message.type === "output") { - record.inputModeTracker.feed(message.message.data); + if (message.message.type === "snapshotReady" && message.message.replayPreamble !== undefined) { + record.replayPreamble = message.message.replayPreamble; } for (const listener of Array.from(record.messageListeners)) { listener(message.message); @@ -558,11 +565,18 @@ export function createWorkerTerminalManager( id: string, options?: { scrollbackLines?: number }, ): Promise { - return (await sendRequest({ + const snapshot = (await sendRequest({ type: "getTerminalState", terminalId: id, ...(options ? { options } : {}), })) as TerminalWorkerStateResult; + if (snapshot && snapshot.replayPreamble !== undefined) { + const record = recordsById.get(id); + if (record) { + record.replayPreamble = snapshot.replayPreamble; + } + } + return snapshot; }, setTerminalTitle(id: string, title: string): boolean { diff --git a/scripts/benchmark-terminal-latency.ts b/scripts/benchmark-terminal-latency.ts new file mode 100644 index 000000000..9c6d4a626 --- /dev/null +++ b/scripts/benchmark-terminal-latency.ts @@ -0,0 +1,751 @@ +/** + * Reproducible terminal latency benchmark (Node-only, isolated daemon). + * + * Boots its OWN isolated daemon subprocess (fresh mkdtemp PASEO_HOME, random + * port) — it NEVER touches the developer daemon on port 6767 — and measures: + * A) terminal echo latency (single-byte input -> first echoed output frame) + * B) terminal output jitter (inter-frame gaps while a command drains ~2MB) + * C) RPC ping RTT at 10Hz (proxy for daemon main-loop delay) + * across load levels L0..L3 where load is generated by dev-only `mock` agents + * that stream agent updates over the same socket the benchmark client uses. + * + * Run: + * npx tsx scripts/benchmark-terminal-latency.ts + * + * Optional: + * BENCH_INCLUDE_L3=1 include the L3 level (L2 + a second noisy terminal) + * + * Output: pretty table to stdout + JSON to /tmp/paseo-terminal-bench/.json + * + * Requires built client/protocol dist (packages/client/dist). Build with: + * npm run build:client + */ + +import { spawn, execSync, type ChildProcess } from "node:child_process"; +import { mkdtemp, mkdir, rm, writeFile } from "node:fs/promises"; +import net from "node:net"; +import os from "node:os"; +import path from "node:path"; +import { performance } from "node:perf_hooks"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import { WebSocket } from "ws"; + +const __dirname = path.dirname(fileURLToPath(import.meta.url)); +const REPO_ROOT = path.resolve(__dirname, ".."); +const SERVER_DIR = path.join(REPO_ROOT, "packages", "server"); + +// --- minimal structural view of the dist DaemonClient we depend on ---------- + +interface TerminalStreamEvent { + terminalId: string; + type: "output" | "snapshot" | "restore"; + data?: Uint8Array; +} + +interface DaemonClientLike { + connect(): Promise; + close(): Promise; + ping(params?: { timeoutMs?: number }): Promise<{ rttMs: number }>; + listTerminals(cwd?: string): Promise; + openProject(cwd: string): Promise<{ workspace: { id: string } | null; error: string | null }>; + createTerminal( + cwd: string, + name?: string, + ): Promise<{ terminal: { id: string } | null; error: string | null }>; + subscribeTerminal(terminalId: string): Promise<{ error: string | null }>; + killTerminal(terminalId: string): Promise; + sendTerminalInput( + terminalId: string, + message: { type: "input"; data: string } | { type: "resize"; rows: number; cols: number }, + ): void; + onTerminalStreamEvent(handler: (event: TerminalStreamEvent) => void): () => void; + createAgent(options: { + provider: string; + cwd: string; + model?: string; + title?: string; + }): Promise<{ id: string }>; + sendAgentMessage(agentId: string, text: string): Promise; + cancelAgent?(agentId: string): Promise; + deleteAgent?(agentId: string): Promise; +} + +interface DaemonClientCtor { + new (config: { + url: string; + clientId: string; + clientType: "cli"; + appVersion?: string; + webSocketFactory: (url: string, options?: { headers?: Record }) => unknown; + }): DaemonClientLike; +} + +// --- stats helpers ---------------------------------------------------------- + +interface Stats { + count: number; + min: number; + p50: number; + p95: number; + max: number; + avg: number; +} + +function summarize(values: number[]): Stats { + if (values.length === 0) { + return { count: 0, min: 0, p50: 0, p95: 0, max: 0, avg: 0 }; + } + const sorted = [...values].sort((a, b) => a - b); + const pick = (q: number) => sorted[Math.min(sorted.length - 1, Math.floor(sorted.length * q))]; + return { + count: sorted.length, + min: round(sorted[0]), + p50: round(pick(0.5)), + p95: round(pick(0.95)), + max: round(sorted[sorted.length - 1]), + avg: round(values.reduce((a, b) => a + b, 0) / values.length), + }; +} + +function round(n: number): number { + return Math.round(n * 100) / 100; +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +// --- daemon lifecycle ------------------------------------------------------- + +async function getFreePort(): Promise { + return new Promise((resolve, reject) => { + const srv = net.createServer(); + srv.on("error", reject); + srv.listen(0, "127.0.0.1", () => { + const addr = srv.address(); + if (addr === null || typeof addr === "string") { + srv.close(); + reject(new Error("Failed to acquire a free port")); + return; + } + const { port } = addr; + srv.close(() => resolve(port)); + }); + }); +} + +async function waitForPort(port: number, child: ChildProcess, timeoutMs: number): Promise { + const start = Date.now(); + let lastError: unknown = null; + while (Date.now() - start < timeoutMs) { + if (child.exitCode !== null) { + throw new Error(`Daemon exited before listening (code ${child.exitCode})`); + } + try { + await new Promise((resolve, reject) => { + const socket = net.connect(port, "127.0.0.1", () => { + socket.end(); + resolve(); + }); + socket.setTimeout(1000, () => { + socket.destroy(); + reject(new Error("connect timeout")); + }); + socket.on("error", reject); + }); + return; + } catch (error) { + lastError = error; + await sleep(150); + } + } + throw new Error( + `Daemon did not listen on 127.0.0.1:${port} within ${timeoutMs}ms (${String(lastError)})`, + ); +} + +interface BootedDaemon { + child: ChildProcess; + port: number; + paseoHome: string; + pid: number; +} + +async function bootDaemon(): Promise { + const port = await getFreePort(); + if (port === 6767) { + throw new Error("Refusing to use port 6767 (the developer daemon)"); + } + const paseoHome = await mkdtemp(path.join(os.tmpdir(), "paseo-bench-home-")); + const tsxBin = execSync("which tsx").toString().trim(); + + const child = spawn(tsxBin, ["scripts/supervisor-entrypoint.ts", "--dev"], { + cwd: SERVER_DIR, + env: { + ...process.env, + PASEO_HOME: paseoHome, + PASEO_SERVER_ID: "srv_terminal_bench", + PASEO_LISTEN: `127.0.0.1:${port}`, + PASEO_NODE_ENV: "development", + NODE_ENV: "development", + }, + stdio: ["ignore", "pipe", "pipe"], + detached: false, + }); + + child.stdout?.on("data", (data: Buffer) => { + for (const line of data.toString().split("\n")) { + const trimmed = line.trim(); + if (trimmed) console.log(`[daemon] ${trimmed}`); + } + }); + child.stderr?.on("data", (data: Buffer) => { + for (const line of data.toString().split("\n")) { + const trimmed = line.trim(); + if (trimmed) console.error(`[daemon] ${trimmed}`); + } + }); + + await waitForPort(port, child, 30_000); + return { child, port, paseoHome, pid: child.pid ?? -1 }; +} + +async function loadDaemonClientCtor(): Promise { + const moduleUrl = pathToFileURL( + path.join(REPO_ROOT, "packages", "client", "dist", "daemon-client.js"), + ).href; + const mod = (await import(moduleUrl)) as { DaemonClient: DaemonClientCtor }; + return mod.DaemonClient; +} + +async function connectClient(ctor: DaemonClientCtor, port: number): Promise { + const client = new ctor({ + url: `ws://127.0.0.1:${port}/ws`, + clientId: `bench-${Math.random().toString(36).slice(2)}`, + clientType: "cli", + webSocketFactory: (url, options) => new WebSocket(url, { headers: options?.headers }), + }); + await client.connect(); + return client; +} + +// --- measurement primitives ------------------------------------------------- + +const ECHO_SAMPLES = 100; +const ECHO_INTERVAL_MS = 50; +const ECHO_FRAME_TIMEOUT_MS = 2_000; +// Printable ASCII the shell echoes back verbatim (skip control chars/space). +const ECHO_ALPHABET = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"; + +interface EchoResult { + samples: number; + hits: number; + latencyMs: Stats; +} + +/** + * Sends single distinct bytes spaced out in time; each byte's latency is the + * gap between send and the first output frame whose decoded text contains it. + */ +async function measureEcho(client: DaemonClientLike, terminalId: string): Promise { + const decoder = new TextDecoder(); + // The stream handler records the arrival time of the awaited echo byte; the + // sender loop polls this between sends. A plain shared slot avoids juggling a + // promise resolver across the handler/timeout boundary. + // Mutated by the async stream handler and read by the sender loop; a holder + // object keeps the cross-callback mutation explicit. + const slot: { awaitedChar: string | null; arrivalAt: number | null } = { + awaitedChar: null, + arrivalAt: null, + }; + + const unsubscribe = client.onTerminalStreamEvent((event) => { + if (event.terminalId !== terminalId || event.type !== "output" || !event.data) { + return; + } + if (slot.awaitedChar === null || slot.arrivalAt !== null) { + return; + } + if (decoder.decode(event.data).includes(slot.awaitedChar)) { + slot.arrivalAt = performance.now(); + } + }); + + const latencies: number[] = []; + let hits = 0; + try { + for (let i = 0; i < ECHO_SAMPLES; i += 1) { + const char = ECHO_ALPHABET[i % ECHO_ALPHABET.length]; + slot.awaitedChar = char; + slot.arrivalAt = null; + const sentAt = performance.now(); + client.sendTerminalInput(terminalId, { type: "input", data: char }); + + const deadline = sentAt + ECHO_FRAME_TIMEOUT_MS; + while (slot.arrivalAt === null && performance.now() < deadline) { + await sleep(1); + } + if (slot.arrivalAt !== null) { + latencies.push(slot.arrivalAt - sentAt); + hits += 1; + } + slot.awaitedChar = null; + await sleep(ECHO_INTERVAL_MS); + } + } finally { + unsubscribe(); + } + + return { samples: ECHO_SAMPLES, hits, latencyMs: summarize(latencies) }; +} + +const BURST_TARGET_BYTES = 2_000_000; +const BURST_CHUNK = 16_384; +const BURST_DRAIN_TIMEOUT_MS = 20_000; +const BURST_QUIET_MS = 400; + +interface BurstResult { + bytesSeen: number; + outputFrames: number; + snapshotFrames: number; + restoreFrames: number; + drainMs: number; + interFrameGapMs: Stats; +} + +/** + * Runs a node one-liner in the terminal that prints ~2MB rapidly, then records + * inter-output-frame gaps until the stream goes quiet. Snapshot/restore frames + * during the burst indicate the snapshot-churn path firing. + */ +async function measureBurst(client: DaemonClientLike, terminalId: string): Promise { + let bytesSeen = 0; + let outputFrames = 0; + let snapshotFrames = 0; + let restoreFrames = 0; + const gaps: number[] = []; + let lastBurstFrameAt = 0; + let firstBurstFrameAt = 0; + let lastActivityAt = performance.now(); + + // Frames smaller than this are the echoed command / prompt noise, not burst + // payload — gap timing only starts once real drain data is flowing, so the + // node startup latency doesn't pollute the inter-frame jitter. + const BURST_FRAME_MIN_BYTES = 1_024; + + const unsubscribe = client.onTerminalStreamEvent((event) => { + if (event.terminalId !== terminalId) { + return; + } + const now = performance.now(); + lastActivityAt = now; + if (event.type === "snapshot") { + snapshotFrames += 1; + return; + } + if (event.type === "restore") { + restoreFrames += 1; + return; + } + if (event.type !== "output" || !event.data) { + return; + } + outputFrames += 1; + bytesSeen += event.data.byteLength; + if (event.data.byteLength < BURST_FRAME_MIN_BYTES && firstBurstFrameAt === 0) { + return; + } + if (firstBurstFrameAt === 0) { + firstBurstFrameAt = now; + } else { + gaps.push(now - lastBurstFrameAt); + } + lastBurstFrameAt = now; + }); + + // Clear whatever the echo measurement left on the shell input line (Ctrl-C), + // then let the fresh prompt settle before sending the burst command. + client.sendTerminalInput(terminalId, { type: "input", data: "" }); + await sleep(300); + + const startedAt = performance.now(); + try { + const chunks = Math.ceil(BURST_TARGET_BYTES / BURST_CHUNK); + // Print BURST_TARGET_BYTES of 'x' in chunks, no trailing newline spam. + const oneLiner = + `const c='x'.repeat(${BURST_CHUNK});` + + `for(let i=0;i<${chunks};i++)process.stdout.write(c);` + + `process.stdout.write('\\nBURST_DONE\\n')`; + client.sendTerminalInput(terminalId, { + type: "input", + data: `node -e "${oneLiner}"\r`, + }); + + const deadline = performance.now() + BURST_DRAIN_TIMEOUT_MS; + while (performance.now() < deadline) { + await sleep(50); + const quietFor = performance.now() - lastActivityAt; + // The daemon coalesces a large pty write into a snapshot plus a handful of + // frames, so the client rarely sees the full BURST_TARGET_BYTES. Exit once + // real burst data has arrived and the stream has gone quiet. + if (firstBurstFrameAt > 0 && quietFor >= BURST_QUIET_MS) { + break; + } + } + } finally { + unsubscribe(); + } + + return { + bytesSeen, + outputFrames, + snapshotFrames, + restoreFrames, + drainMs: round((lastBurstFrameAt || startedAt) - startedAt), + interFrameGapMs: summarize(gaps), + }; +} + +const PING_INTERVAL_MS = 100; + +/** + * Samples RPC RTT at 10Hz for `durationMs`. Uses `ping` if available, else a + * cheap listTerminals request. Returns RTT stats — a proxy for main-loop delay. + */ +async function measurePing(client: DaemonClientLike, durationMs: number): Promise { + const rtts: number[] = []; + const deadline = Date.now() + durationMs; + while (Date.now() < deadline) { + const start = performance.now(); + try { + const result = await client.ping({ timeoutMs: 5_000 }); + rtts.push(typeof result?.rttMs === "number" ? result.rttMs : performance.now() - start); + } catch { + const fallbackStart = performance.now(); + await client.listTerminals().catch(() => undefined); + rtts.push(performance.now() - fallbackStart); + } + await sleep(PING_INTERVAL_MS); + } + return summarize(rtts); +} + +// --- load generation -------------------------------------------------------- + +interface LoadController { + agentIds: string[]; + noisyTerminalId: string | null; + stop(): Promise; +} + +interface LevelSpec { + id: string; + description: string; + mockAgents: number; + bigDiffEvery2s: boolean; + secondNoisyTerminal: boolean; +} + +const LEVELS: LevelSpec[] = [ + { + id: "L0", + description: "idle", + mockAgents: 0, + bigDiffEvery2s: false, + secondNoisyTerminal: false, + }, + { + id: "L1", + description: "2 mock agents streaming", + mockAgents: 2, + bigDiffEvery2s: false, + secondNoisyTerminal: false, + }, + { + id: "L2", + description: "4 mock agents + 200KB diff bursts every 2s", + mockAgents: 4, + bigDiffEvery2s: true, + secondNoisyTerminal: false, + }, +]; + +if (process.env.BENCH_INCLUDE_L3 === "1") { + LEVELS.push({ + id: "L3", + description: "L2 + a second terminal continuously emitting", + mockAgents: 4, + bigDiffEvery2s: true, + secondNoisyTerminal: true, + }); +} + +async function startLoad( + client: DaemonClientLike, + cwd: string, + spec: LevelSpec, +): Promise { + const agentIds: string[] = []; + for (let i = 0; i < spec.mockAgents; i += 1) { + const agent = await client.createAgent({ + provider: "mock", + cwd, + model: "five-minute-stream", + title: `${spec.id}-load-${i}`, + }); + agentIds.push(agent.id); + // Kick off the sustained stream. + await client.sendAgentMessage(agent.id, "start streaming").catch(() => undefined); + } + + let bigDiffTimer: ReturnType | null = null; + if (spec.bigDiffEvery2s && agentIds.length > 0) { + bigDiffTimer = setInterval(() => { + const target = agentIds[0]; + void client + .sendAgentMessage(target, "emit 200000 byte large diff agent stream update") + .catch(() => undefined); + }, 2_000); + bigDiffTimer.unref?.(); + } + + let noisyTerminalId: string | null = null; + if (spec.secondNoisyTerminal) { + const created = await client.createTerminal(cwd, `${spec.id}-noisy`); + if (created.terminal) { + noisyTerminalId = created.terminal.id; + await client.subscribeTerminal(noisyTerminalId); + // Continuous infinite emit loop in the shell. + client.sendTerminalInput(noisyTerminalId, { + type: "input", + data: `node -e "setInterval(()=>process.stdout.write('y'.repeat(4096)),20)"\r`, + }); + } + } + + // Let the streams warm up before we start measuring. + await sleep(500); + + return { + agentIds, + noisyTerminalId, + async stop() { + if (bigDiffTimer) { + clearInterval(bigDiffTimer); + } + if (noisyTerminalId) { + // Ctrl-C then kill the terminal. + client.sendTerminalInput(noisyTerminalId, { type: "input", data: "" }); + await client.killTerminal(noisyTerminalId).catch(() => undefined); + } + for (const agentId of agentIds) { + if (client.cancelAgent) { + await client.cancelAgent(agentId).catch(() => undefined); + } + if (client.deleteAgent) { + await client.deleteAgent(agentId).catch(() => undefined); + } + } + }, + }; +} + +// --- per-level run ---------------------------------------------------------- + +interface LevelResult { + id: string; + description: string; + echo: EchoResult; + burst: BurstResult; + pingMs: Stats; +} + +async function runLevel( + client: DaemonClientLike, + cwd: string, + spec: LevelSpec, + measuredTerminalId: string, +): Promise { + console.log(`\n=== ${spec.id} (${spec.description}) ===`); + const load = await startLoad(client, cwd, spec); + try { + // Ping sampling runs concurrently with the echo + burst measurements so it + // observes the daemon under the same load the latency primitives see. + let pingMs: Stats = summarize([]); + const pingDone = (async () => { + pingMs = await measurePing(client, 14_000); + })(); + + console.log(` measuring echo latency (${ECHO_SAMPLES} samples)...`); + const echo = await measureEcho(client, measuredTerminalId); + console.log( + ` echo p50=${echo.latencyMs.p50}ms p95=${echo.latencyMs.p95}ms max=${echo.latencyMs.max}ms (${echo.hits}/${echo.samples} hits)`, + ); + + console.log(" measuring output burst/jitter (~2MB)..."); + const burst = await measureBurst(client, measuredTerminalId); + console.log( + ` burst drain=${burst.drainMs}ms frames=${burst.outputFrames} gap_p95=${burst.interFrameGapMs.p95}ms snap=${burst.snapshotFrames} restore=${burst.restoreFrames}`, + ); + + await pingDone; + console.log(` ping p50=${pingMs.p50}ms p95=${pingMs.p95}ms max=${pingMs.max}ms`); + + return { id: spec.id, description: spec.description, echo, burst, pingMs }; + } finally { + await load.stop(); + } +} + +// --- output ----------------------------------------------------------------- + +function pad(value: string | number, width: number): string { + return String(value).padStart(width); +} + +function printTable(results: LevelResult[]): void { + console.log("\n========== TERMINAL LATENCY BENCHMARK ==========\n"); + const header = [ + pad("level", 6), + pad("echo_p50", 9), + pad("echo_p95", 9), + pad("echo_max", 9), + pad("ping_p50", 9), + pad("ping_p95", 9), + pad("gap_p95", 8), + pad("gap_max", 8), + pad("drain", 8), + pad("frames", 7), + pad("snap", 5), + pad("restore", 8), + ].join(" "); + console.log(header); + console.log("-".repeat(header.length)); + for (const r of results) { + console.log( + [ + pad(r.id, 6), + pad(r.echo.latencyMs.p50, 9), + pad(r.echo.latencyMs.p95, 9), + pad(r.echo.latencyMs.max, 9), + pad(r.pingMs.p50, 9), + pad(r.pingMs.p95, 9), + pad(r.burst.interFrameGapMs.p95, 8), + pad(r.burst.interFrameGapMs.max, 8), + pad(r.burst.drainMs, 8), + pad(r.burst.outputFrames, 7), + pad(r.burst.snapshotFrames, 5), + pad(r.burst.restoreFrames, 8), + ].join(" "), + ); + } + console.log(""); + for (const r of results) { + console.log(` ${r.id}: ${r.description}`); + } + console.log(""); +} + +function getCommitHash(): string { + try { + return execSync("git rev-parse --short HEAD", { cwd: REPO_ROOT }).toString().trim(); + } catch { + return "unknown"; + } +} + +// --- main ------------------------------------------------------------------- + +async function main(): Promise { + const commit = getCommitHash(); + console.log(`Terminal latency benchmark — commit ${commit}`); + console.log("Booting isolated daemon (random port, fresh PASEO_HOME)..."); + + let daemon: BootedDaemon | null = null; + let client: DaemonClientLike | null = null; + let workspaceDir: string | null = null; + let measuredTerminalId: string | null = null; + + try { + daemon = await bootDaemon(); + console.log(`Daemon ready: pid=${daemon.pid} port=${daemon.port} home=${daemon.paseoHome}`); + + const ctor = await loadDaemonClientCtor(); + client = await connectClient(ctor, daemon.port); + + workspaceDir = await mkdtemp(path.join(os.tmpdir(), "paseo-bench-ws-")); + const opened = await client.openProject(workspaceDir); + if (!opened.workspace) { + throw new Error(`Failed to open project: ${opened.error}`); + } + + const created = await client.createTerminal(workspaceDir, "measured"); + if (!created.terminal) { + throw new Error(`Failed to create terminal: ${created.error}`); + } + measuredTerminalId = created.terminal.id; + await client.subscribeTerminal(measuredTerminalId); + + // Let the shell print its prompt and settle before measuring. + await sleep(750); + + const results: LevelResult[] = []; + for (const spec of LEVELS) { + const result = await runLevel(client, workspaceDir, spec, measuredTerminalId); + results.push(result); + // Settle between levels so a previous level's load fully drains. + await sleep(2_000); + } + + printTable(results); + + const outDir = "/tmp/paseo-terminal-bench"; + await mkdir(outDir, { recursive: true }); + const timestamp = new Date().toISOString().replace(/[:.]/g, "-"); + const outPath = path.join(outDir, `${timestamp}.json`); + await writeFile( + outPath, + JSON.stringify( + { + commit, + daemonPid: daemon.pid, + port: daemon.port, + paseoHome: daemon.paseoHome, + node: process.version, + platform: `${os.platform()} ${os.arch()}`, + createdAt: new Date().toISOString(), + levels: results, + }, + null, + 2, + ), + ); + console.log(`Wrote results to ${outPath}`); + } finally { + if (client && measuredTerminalId) { + await client.killTerminal(measuredTerminalId).catch(() => undefined); + } + if (client) { + await client.close().catch(() => undefined); + } + if (daemon) { + daemon.child.kill("SIGTERM"); + await Promise.race([ + new Promise((resolve) => daemon!.child.once("exit", () => resolve())), + sleep(5_000), + ]); + if (daemon.child.exitCode === null && daemon.child.signalCode === null) { + daemon.child.kill("SIGKILL"); + } + await rm(daemon.paseoHome, { recursive: true, force: true }).catch(() => undefined); + } + if (workspaceDir) { + await rm(workspaceDir, { recursive: true, force: true }).catch(() => undefined); + } + } +} + +main().catch((error) => { + console.error(error); + process.exit(1); +});