jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Eliminate spiky terminal lag under load (#1500)

Browse Source 
* 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.
This commit is contained in:
Mohamed Boudra
2026-06-13 05:38:58 +00:00
committed by GitHub
parent 442a1b9a5a
commit 72b67f48e3
19 changed files with 1693 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app/e2e/helpers/startup-dsl.ts
+21 -3
View File
@@ -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 = "<!doctype html><html><body>storage seed</body></html>";
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<void> {
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<this> {
await this.page.getByRole("button", { name: "Open menu", exact: true }).click();
await expect(this.page.getByText(input.label, { exact: true })).toBeVisible({
timeout: 15_000,
});
packages/app/src/terminal/runtime/terminal-emulator-runtime.test.ts
+169 -28
View File
@@ -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"]);
});
packages/app/src/terminal/runtime/terminal-emulator-runtime.ts
+98
View File
@@ -184,6 +184,15 @@ export class TerminalEmulatorRuntime {
private outputOperations: TerminalOutputOperation[] = [];
private inFlightOutputOperation: TerminalOutputOperation | null = null;
private inFlightOutputOperationTimeout: ReturnType<typeof setTimeout> | 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);
packages/server/src/server/session.ts
+17 -7
View File
@@ -566,6 +566,7 @@ export interface SessionOptions {
clientCapabilities?: Record<string, unknown> | 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()) &&
packages/server/src/server/websocket-server.ts
+40
View File
@@ -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<typeof setInterval> | null = null;
private eventLoopDelayMonitor: ReturnType<typeof monitorEventLoopDelay> | 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<string>,
@@ -678,6 +699,8 @@ export class VoiceAssistantWebSocketServer {
this.runtimeMetricsInterval = null;
}
this.flushRuntimeMetrics({ final: true });
this.eventLoopDelayMonitor?.disable();
this.eventLoopDelayMonitor = null;
const uniqueConnections = new Set<SessionConnection>([
...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,
packages/server/src/terminal/terminal-output-coalescer.test.ts
+76 -5
View File
@@ -3,6 +3,7 @@ import { TerminalOutputCoalescer } from "./terminal-output-coalescer.js";
function createHarness() {
let nextTimer = 1;
let now = 1000;
const scheduled = new Map<number, { callback: () => 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"]);
});
});
packages/server/src/terminal/terminal-output-coalescer.ts
+22
View File
@@ -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<typeof setTimeout> | 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();
packages/server/src/terminal/terminal-restore.ts
+7
View File
@@ -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;
packages/server/src/terminal/terminal-session-controller.test.ts
+136
View File
@@ -168,6 +168,12 @@ async function flushMicrotasks(): Promise<void> {
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<void> {
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<TerminalStateSnapshot>({ 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);
});
});
packages/server/src/terminal/terminal-session-controller.ts
+32 -1
View File
@@ -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<readonly string[]>;
private readonly clientSupportsWrapReflow: () => boolean;
private readonly getClientBufferedAmount: () => number | null;
private readonly subscribedDirectories = new Set<string>();
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 };
}
packages/server/src/terminal/terminal-worker-process.ts
+55 -3
View File
@@ -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<string, Array<() => void>>();
const outputCoalescerByTerminalId = new Map<string, TerminalOutputCoalescer>();
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<void> {
}
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;
}
packages/server/src/terminal/terminal-worker-protocol.ts
+2
View File
@@ -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;
packages/server/src/terminal/terminal.test.ts
+44
View File
@@ -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<NonNullable<ReturnType<TerminalSession["getExitInfo"]>>>(
(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", () => {
packages/server/src/terminal/terminal.ts
+40 -7
View File
@@ -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<Te
let processExited = false;
const processExitWaiters = new Set<() => 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<Te
lastOutputLines:
lastOutputLines.length > 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<Te
}
disposed = true;
pendingInput = "";
recentOutputChunks.length = 0;
recentOutputLength = 0;
inputModeTracker.reset();
if (inputFlushImmediate) {
clearImmediate(inputFlushImmediate);
@@ -837,9 +850,23 @@ export async function createTerminal(options: CreateTerminalOptions): Promise<Te
for (const response of inputModeUpdate.responses) {
ptyProcess.write(response);
}
recentOutputText = `${recentOutputText}${data}`;
if (recentOutputText.length > 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<Te
if (!disposed && active && listeners.has(subscriptionListener)) {
snapshotDelivered = true;
if (initialSnapshot === "ready") {
listener({ type: "snapshotReady", revision: stateRevision });
// Carry the input-mode preamble so the snapshot-less "ready" path
// (live restore) can replay it without a separate state fetch.
listener({
type: "snapshotReady",
revision: stateRevision,
replayPreamble: getReplayPreamble(),
});
} else {
listener({ type: "snapshot", ...getStateSnapshot() });
}
packages/server/src/terminal/worker-terminal-manager.test.ts
+121 -9
View File
@@ -1,6 +1,7 @@
import { afterEach, expect, it } from "vitest";
import { isPlatform } from "../test-utils/platform.js";
import { EventEmitter } from "node:events";
import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync } from "node:fs";
import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
import { join } from "node:path";
import { tmpdir } from "node:os";
import { createWorkerTerminalManager } from "./worker-terminal-manager.js";
@@ -12,6 +13,8 @@ import type {
TerminalWorkerToParentMessage,
} from "./terminal-worker-protocol.js";
type TerminalRow = TerminalState["grid"][number];
function nodeTerminalCommand(script: string): { command: string; args: string[] } {
return {
command: process.execPath,
@@ -38,15 +41,30 @@ function getVisibleText(session: TerminalSession): string {
return getVisibleTextFromState(session.getState());
}
function rowToText(row: TerminalRow): string {
return row
.map((cell) => 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);
packages/server/src/terminal/worker-terminal-manager.ts
+21 -7
View File
@@ -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<TerminalStateSnapshot | null> {
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 {