diff --git a/docs/terminal-performance.md b/docs/terminal-performance.md index 3bb7c4d71c..80dfceea05 100644 --- a/docs/terminal-performance.md +++ b/docs/terminal-performance.md @@ -6,9 +6,9 @@ How terminal output stays low-latency, what the invariants are, and how to measu ``` 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 + → headless xterm parse (worker, ANSI replay fidelity) + → TerminalOutputCoalescer (worker, ≤1 socket frame per 5ms per terminal) + → framed local socket → 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 @@ -20,9 +20,9 @@ Terminal frames share the daemon main event loop with all agent traffic. The `ev ## 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. +- **Output coalescing happens in the worker, before IPC.** One socket frame per pty chunk was a main-loop flood under build output. Non-output messages (snapshot/snapshotReady/titleChange/exit) must flush the coalescer first so ordering is preserved. - **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. +- **The input-mode tracker runs once per process boundary, not per hop.** The worker owns the authoritative tracker; the daemon caches the replay preamble from replay responses and `snapshotReady` messages. Do not reintroduce a per-chunk `feed()` on the daemon main loop. - **Snapshot catch-up is backpressure-gated.** A stream falls back to a full snapshot only when `outputBytesSinceSnapshot > MAX_TERMINAL_OUTPUT_FRAME_BYTES` (256KB) **and** the client transport reports `bufferedAmount > MAX_CLIENT_BUFFERED_BYTES` (4MB). A client that keeps draining streams continuously, no matter how much output is produced. Before this gate existed, every 256KB of build output dropped a frame and forced a full JSON cell-grid snapshot (~200k objects across IPC) — the historical source of spiky lag and GC hitches. - **Plugin daemon sessions report IPC queue bytes.** Their virtual socket increments `bufferedAmount` before `process.send` and decrements it only in the send callback. Text and binary frames share that ordered queue, so the normal snapshot catch-up and physical high-water gates remain valid for server-side plugin SDK traffic. - **Client output writes are not serialized per frame.** The emulator runtime drains contiguous plain writes straight into xterm (which buffers internally). Only barrier ops (`clear`, `snapshot`, `suppressInput` writes) wait — behind a zero-length sentinel write — so resets can't interleave with in-flight output. diff --git a/docs/terminal-worker.md b/docs/terminal-worker.md index 991fc5d644..aedd6053ef 100644 --- a/docs/terminal-worker.md +++ b/docs/terminal-worker.md @@ -40,10 +40,16 @@ The worker writes `$PASEO_HOME/terminal-worker/.json` (owner-only) with its Never more than two workers run. A new worker launches only when none is current, none is incompatible and fewer than two run. The count comes from the records, so an unreachable or unidentified worker counts, and so does one a cleanup could not confirm gone. When two run, the daemon starts with no current worker, and the first terminal created after one of them exits launches it. -**The frozen envelope never changes across versions:** the frame format (4-byte length + `v8.serialize`), `hello`/`helloAck`, the `listTerminals` info fields and `shutdown`. That is what lets any daemon close any worker: a kill stop sends `shutdown` to a newer worker it cannot otherwise drive. When you bump the version, keep a previous-version adapter so the draining worker can still be driven, and drop the adapter for the version before that. +**The frozen envelope never changes across versions:** the frame format (4-byte length + `v8.serialize`), `hello`/`helloAck`, the `listTerminals` info fields and `shutdown`. That is what lets any daemon close any worker: a kill stop sends `shutdown` to a newer worker it cannot otherwise drive. Protocol 2 hard cuts the unreleased protocol 1 cell transport: protocol 1 workers are obsolete and close. For future versions, keep a previous-version adapter so the draining worker can still be driven, and drop the adapter for the version before that. Tests stand up other versions with `PASEO_TERMINAL_WORKER_PROTOCOL_OVERRIDE`, read by the worker. It is test-only. +## Display replay + +The worker keeps the headless xterm model and generates ANSI. A replay request emits begin metadata (dimensions, retained history, revision and input modes), at most 64 KiB per data chunk, and end before later live output. Capture is synchronous with PTY callbacks; flush the output coalescer before the cut so a later batch never includes bytes already in replay. Retained-history metadata counts the worker history before the request limit. + +Replay waits for each socket write to complete before sending the next chunk. PTYs keep parsing while writes wait. The replaying terminal holds at most 4 MiB of live output until replay ends; exceeding that bound fails the replay and ends every display stream of that terminal with an error requiring a fresh attach. Existing viewers share the worker live stream, so they must also stop before bytes resume after a discarded tail. Sibling terminals continue streaming. Live traffic that fills the shared socket beyond 4 MiB disconnects the stalled controller; the daemon reconnects and requests fresh replay. Existing ANSI clients receive one collected Restore frame from the daemon. Capture stays a worker request; bootstrap readiness uses `hasOutput`, not a transported grid. + ## Running from its own copy A worker outlives the installation that started it. An npm or desktop update replaces that installation's files, and the next terminal the old worker creates would load node-pty's `spawn-helper` and the zsh integration from paths that no longer exist. So a built daemon does not launch the worker from its install: diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index cca0b4f3fa..1dee3857b8 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -1132,8 +1132,6 @@ export class Session { isPathWithinRoot: (rootPath, candidatePath) => this.isPathWithinRoot(rootPath, candidatePath), sessionLogger: this.sessionLogger, listTerminalWorkspaceRefs: () => this.listActiveWorkspaceRefs(), - clientSupportsWrapReflow: (source) => - this.supportsForSource(CLIENT_CAPS.terminalReflowableSnapshot, source), getClientBufferedAmount: (source) => this.getTransportBufferedAmount(source), }); this.agentDirectory = new AgentDirectory({ diff --git a/packages/server/src/server/worktree-bootstrap.test.ts b/packages/server/src/server/worktree-bootstrap.test.ts index 96da6e648b..b26c9e91f4 100644 --- a/packages/server/src/server/worktree-bootstrap.test.ts +++ b/packages/server/src/server/worktree-bootstrap.test.ts @@ -209,6 +209,7 @@ describe("runWorktreeAutoTerminals", () => { getSize: () => ({ rows: 0, cols: 0 }), getTitle: () => undefined, getExitInfo: () => null, + hasOutput: () => false, getState: () => ({ rows: 0, cols: 0, @@ -312,6 +313,7 @@ describe("runWorktreeAutoTerminals", () => { } }; }, + hasOutput: () => true, getState: () => ({ rows: 1, cols: 1, diff --git a/packages/server/src/server/worktree-bootstrap.ts b/packages/server/src/server/worktree-bootstrap.ts index d656738bd4..1e1ba58f8e 100644 --- a/packages/server/src/server/worktree-bootstrap.ts +++ b/packages/server/src/server/worktree-bootstrap.ts @@ -269,9 +269,9 @@ export function buildWorktreeSetupDetail(input: { } async function waitForTerminalBootstrapReadiness( - terminal: Pick, + terminal: Pick, ): Promise { - if (terminalHasOutput(terminal.getState())) { + if (terminal.hasOutput()) { return; } @@ -304,7 +304,7 @@ async function waitForTerminalBootstrapReadiness( finish(); }); - if (terminalHasOutput(terminal.getState())) { + if (terminal.hasOutput()) { finish(); return; } @@ -313,17 +313,6 @@ async function waitForTerminalBootstrapReadiness( }); } -function terminalHasOutput(state: ReturnType): boolean { - for (const row of [...state.scrollback, ...state.grid]) { - for (const cell of row) { - if (cell.char.trim().length > 0) { - return true; - } - } - } - return false; -} - export async function runWorktreeAutoTerminals(options: { workspaceId: string; worktree: WorktreeConfig; diff --git a/packages/server/src/terminal/terminal-capture.ts b/packages/server/src/terminal/terminal-capture.ts index e22d3b6f69..9d7aed9622 100644 --- a/packages/server/src/terminal/terminal-capture.ts +++ b/packages/server/src/terminal/terminal-capture.ts @@ -1,6 +1,6 @@ import stripAnsi from "strip-ansi"; import type { TerminalCell } from "@getpaseo/protocol/messages"; -import type { TerminalSession } from "./terminal.js"; +import type { LocalTerminalSession } from "./terminal.js"; export interface CaptureTerminalLinesOptions { start?: number; @@ -46,7 +46,7 @@ function resolveCaptureLineIndex( } export function captureTerminalLines( - terminal: TerminalSession, + terminal: Pick, options: CaptureTerminalLinesOptions = {}, ): CaptureTerminalLinesResult { const state = terminal.getState(); diff --git a/packages/server/src/terminal/terminal-manager.ts b/packages/server/src/terminal/terminal-manager.ts index 86b4cdb662..80a1975f5b 100644 --- a/packages/server/src/terminal/terminal-manager.ts +++ b/packages/server/src/terminal/terminal-manager.ts @@ -1,10 +1,14 @@ +import { + replayTerminal, + type TerminalReplaySink, + type TerminalReplayMetadata, + type TerminalReplayOptions, +} from "./terminal-replay.js"; import { createTerminal, type LocalTerminalSession, type TerminalActivityTransition, type TerminalSession, - type TerminalStateSnapshot, - type TerminalStateSnapshotOptions, } from "./terminal.js"; import { captureTerminalLines, type CaptureTerminalLinesResult } from "./terminal-capture.js"; import { randomBytes, randomUUID } from "node:crypto"; @@ -76,10 +80,11 @@ export interface TerminalManager { registerCwdEnv(options: { cwd: string; env: Record }): void; validateTerminalActivityToken(terminalId: string, token: string): "valid" | "unknown" | "invalid"; getTerminal(id: string): TerminalSession | undefined; - getTerminalState( + replayTerminal( id: string, - options?: TerminalStateSnapshotOptions, - ): Promise; + options: TerminalReplayOptions | undefined, + sink: TerminalReplaySink, + ): Promise; setTerminalTitle(id: string, title: string): boolean; setTerminalActivity(id: string, state: TerminalActivityState): Promise; clearTerminalAttention(id: string): Promise; @@ -107,6 +112,10 @@ export interface TerminalManagerOptions { /** The in-process manager the terminal worker runs: it owns each PTY, so it knows its pid. */ export interface LocalTerminalManager extends TerminalManager { + createTerminal( + options: Parameters[0], + ): Promise; + getTerminal(id: string): LocalTerminalSession | undefined; getTerminalProcessId(id: string): number | null; getTerminalActivityToken(id: string): string | null; } @@ -337,7 +346,7 @@ export function createTerminalManager( activityUrl?: string | null; baseEnv?: Record; paseoCliPath?: string | null; - }): Promise { + }): Promise { assertAbsolutePath(options.cwd); const terminals = terminalsByCwd.get(options.cwd) ?? []; @@ -357,7 +366,7 @@ export function createTerminalManager( ...(terminalActivityUrl ? { PASEO_TERMINAL_ACTIVITY_URL: terminalActivityUrl } : {}), }; terminalActivityTokenById.set(terminalId, activityToken); - let session: TerminalSession; + let session: LocalTerminalSession; try { session = registerSession( await createTerminal({ @@ -404,7 +413,7 @@ export function createTerminalManager( return expected === token ? "valid" : "invalid"; }, - getTerminal(id: string): TerminalSession | undefined { + getTerminal(id: string): LocalTerminalSession | undefined { return terminalsById.get(id); }, @@ -416,11 +425,9 @@ export function createTerminalManager( return terminalActivityTokenById.get(id) ?? null; }, - async getTerminalState( - id: string, - options?: TerminalStateSnapshotOptions, - ): Promise { - return terminalsById.get(id)?.getStateSnapshot(options) ?? null; + async replayTerminal(id, options, sink) { + const session = terminalsById.get(id); + return session ? replayTerminal(session, options, sink) : null; }, setTerminalTitle(id: string, title: string): boolean { diff --git a/packages/server/src/terminal/terminal-replay.ts b/packages/server/src/terminal/terminal-replay.ts new file mode 100644 index 0000000000..eda3d0d519 --- /dev/null +++ b/packages/server/src/terminal/terminal-replay.ts @@ -0,0 +1,261 @@ +import type { FramedSocket } from "./terminal-worker-frames.js"; +import type { + TerminalWorkerRequest, + TerminalWorkerToParentMessage, +} from "./terminal-worker-protocol.js"; +import { renderTerminalSnapshotToAnsi } from "@getpaseo/protocol/terminal-snapshot"; +import type { LocalTerminalSession } from "./terminal.js"; + +export interface TerminalReplayOptions { + scrollbackLines?: number; +} + +export interface TerminalReplayMetadata { + rows: number; + cols: number; + retainedHistoryLines: number; + revision: number; + replayPreamble: string; +} +export type TerminalReplayEvent = + | { type: "begin"; metadata: TerminalReplayMetadata } + | { type: "data"; data: Uint8Array } + | { type: "end" }; +export type TerminalReplaySink = (event: TerminalReplayEvent) => void; +export const TERMINAL_REPLAY_CHUNK_BYTES = 64 * 1024; + +export function* createTerminalReplay( + session: LocalTerminalSession, + options: TerminalReplayOptions | undefined, +): Generator { + const snapshot = session.getStateSnapshot({ ...options, includeWrapFlags: true }); + const metadata = { + rows: snapshot.state.rows, + cols: snapshot.state.cols, + retainedHistoryLines: session.getRetainedHistoryLines(), + revision: snapshot.revision, + replayPreamble: session.getReplayPreamble(), + }; + const data = Buffer.from(renderTerminalSnapshotToAnsi(snapshot.state), "utf8"); + yield { type: "begin", metadata }; + for (let offset = 0; offset < data.length; offset += TERMINAL_REPLAY_CHUNK_BYTES) { + yield { type: "data", data: data.subarray(offset, offset + TERMINAL_REPLAY_CHUNK_BYTES) }; + } + yield { type: "end" }; + return metadata; +} + +export function replayTerminal( + session: LocalTerminalSession, + options: TerminalReplayOptions | undefined, + sink: TerminalReplaySink, +): TerminalReplayMetadata { + const replay = createTerminalReplay(session, options); + let next = replay.next(); + while (!next.done) { + sink(next.value); + next = replay.next(); + } + return next.value; +} + +interface TerminalReplayPorts { + getSession(id: string): LocalTerminalSession | undefined; + flushOutput(id: string): void; +} + +/** Owns the worker's replay/live cut, bounded live tail and controller lifetime. */ +export function createTerminalReplayController(ports: TerminalReplayPorts) { + let controller: FramedSocket | null = null; + // Only the live tail of a replaying terminal is held. Replay chunks use the socket's + // write completion directly; they never enter this buffer or fill the shared socket. + interface ReplayTail { + readers: number; + bytes: number; + connection: FramedSocket; + output: TerminalWorkerToParentMessage[]; + overflow: boolean; + failure: unknown; + completed: Promise; + resolve: () => void; + reject: (error: unknown) => void; + exit?: Extract; + } + const replayTails = new Map(); + const MAX_REPLAY_TAIL_BYTES = 4 * 1024 * 1024; + + function sendToController(message: TerminalWorkerToParentMessage): void { + if ( + message.type === "terminalExit" || + (message.type === "terminalMessage" && message.message.type === "output") + ) { + const tail = replayTails.get(message.terminalId); + if (tail) { + if (message.type === "terminalExit") { + tail.exit = message; + return; + } + if (!tail.overflow && message.message.type === "output") { + tail.bytes += Buffer.byteLength(message.message.data); + if (tail.bytes > MAX_REPLAY_TAIL_BYTES) { + tail.overflow = true; + tail.output = []; + } else tail.output.push(message); + } + return; + } + } + // With no controller the event is dropped; the headless terminal keeps the screen, and the + // next daemon reads it through listTerminals. + if (controller && controller.socket.writableLength > 4 * 1024 * 1024) { + controller.socket.destroy(); + return; + } + controller?.send(message); + } + + async function deliverReplay( + message: Extract, + ): Promise { + const connection = controller; + if (!connection) return; + ports.flushOutput(message.terminalId); + const session = ports.getSession(message.terminalId); + if (!session) { + await connection.sendAndWait({ + type: "response", + requestId: message.requestId, + ok: true, + result: null, + }); + return; + } + const existingTail = replayTails.get(session.id); + const tail = + existingTail?.connection === connection ? existingTail : createReplayTail(connection); + tail.readers++; + replayTails.set(session.id, tail); + let released = false; + try { + const replay = createTerminalReplay(session, message.options); + let next = replay.next(); + while (!next.done) { + if (tail.overflow) + throw new Error("Terminal output exceeded the replay buffer; attach again"); + await connection.sendAndWait({ + type: "terminalReplay", + requestId: message.requestId, + event: next.value, + }); + next = replay.next(); + } + released = true; + await releaseReplayTail(session, tail); + await connection.sendAndWait({ + type: "response", + requestId: message.requestId, + ok: true, + result: next.value, + }); + } catch (error) { + tail.failure = error; + if (!released) await releaseReplayTail(session, tail); + throw error; + } + } + + function createReplayTail(connection: FramedSocket): ReplayTail { + let resolve!: () => void; + let reject!: (error: unknown) => void; + const completed = new Promise((onResolve, onReject) => { + resolve = onResolve; + reject = onReject; + }); + // A write can fail before every reader reaches the shared completion barrier. + void completed.catch(() => {}); + return { + connection, + readers: 0, + bytes: 0, + output: [], + overflow: false, + failure: null, + completed, + resolve, + reject, + }; + } + + async function releaseReplayTail(session: LocalTerminalSession, tail: ReplayTail): Promise { + if (--tail.readers === 0) { + try { + while (!tail.failure && !tail.overflow && tail.readers === 0 && tail.output.length > 0) { + if (controller !== tail.connection) throw new Error("Terminal worker controller changed"); + const output = tail.output.shift()!; + if (output.type === "terminalMessage" && output.message.type === "output") + tail.bytes -= Buffer.byteLength(output.message.data); + await tail.connection.sendAndWait(output); + } + } catch (error) { + tail.failure = error; + } finally { + finishReplayTail(session, tail); + } + } + await tail.completed; + } + + function finishReplayTail(session: LocalTerminalSession, tail: ReplayTail): void { + if (tail.readers !== 0) return; + if (replayTails.get(session.id) === tail) replayTails.delete(session.id); + if ((tail.overflow || tail.failure) && controller === tail.connection) { + // The live stream is shared by every viewer of this terminal. Once its tail + // is discarded none of those viewers may continue from the old screen. + sendToController({ + type: "terminalMessage", + terminalId: session.id, + message: { + type: "streamError", + error: "Terminal output exceeded the replay buffer; attach again", + }, + replayPreamble: session.getReplayPreamble(), + }); + } + if (tail.exit && controller === tail.connection) sendToController(tail.exit); + if (tail.failure || tail.overflow) + tail.reject( + tail.failure ?? new Error("Terminal output exceeded the replay buffer; attach again"), + ); + else tail.resolve(); + } + + return { + setConnection(connection: FramedSocket | null): void { + const previous = controller; + controller = connection; + if (previous !== connection) previous?.socket.destroy(); + // Old writes settle on their original socket; new output and replay belong + // exclusively to the replacement controller. + replayTails.clear(); + }, + send: sendToController, + async replay( + request: Extract, + ): Promise { + const connection = controller; + if (!connection) return; + try { + await deliverReplay(request); + } catch (error) { + await connection + .sendAndWait({ + type: "response", + requestId: request.requestId, + ok: false, + error: error instanceof Error ? error.message : String(error), + }) + .catch(() => {}); + } + }, + }; +} diff --git a/packages/server/src/terminal/terminal-restore.test.ts b/packages/server/src/terminal/terminal-restore.test.ts index 7255cba1a5..adf29cb726 100644 --- a/packages/server/src/terminal/terminal-restore.test.ts +++ b/packages/server/src/terminal/terminal-restore.test.ts @@ -4,7 +4,6 @@ import { TerminalStreamOpcode, decodeTerminalStreamFrame, } from "@getpaseo/protocol/binary-frames/index"; -import type { TerminalCell, TerminalState } from "@getpaseo/protocol/messages"; import { encodeTerminalRestoreFrame, resolveRestoreAfterOutputOverflow, @@ -12,22 +11,6 @@ import { resolveTerminalSubscriptionSnapshotMode, } from "./terminal-restore.js"; -function terminalRow(text: string, cols = 80): TerminalCell[] { - return Array.from({ length: cols }, (_, index) => ({ - char: text[index] ?? " ", - })); -} - -function terminalState(text: string): TerminalState { - return { - rows: 1, - cols: 80, - grid: [terminalRow(text)], - scrollback: [], - cursor: { row: 0, col: text.length }, - }; -} - describe("terminal restore policy", () => { test("uses ready snapshots only for restore-aware subscriptions", () => { expect(resolveTerminalSubscriptionSnapshotMode(undefined)).toBe("state"); @@ -61,10 +44,7 @@ describe("terminal restore policy", () => { const frame = decodeTerminalStreamFrame( encodeTerminalRestoreFrame({ slot: 4, - snapshot: { - state: terminalState("restored"), - revision: 1, - }, + data: new TextEncoder().encode("restored"), }), ); diff --git a/packages/server/src/terminal/terminal-restore.ts b/packages/server/src/terminal/terminal-restore.ts index ffc1bd9faa..c16360db50 100644 --- a/packages/server/src/terminal/terminal-restore.ts +++ b/packages/server/src/terminal/terminal-restore.ts @@ -1,11 +1,9 @@ import type { SubscribeTerminalRequest } from "@getpaseo/protocol/messages"; import { TerminalStreamOpcode, - encodeTerminalSnapshotPayload, encodeTerminalStreamFrame, } from "@getpaseo/protocol/binary-frames/index"; -import { renderTerminalSnapshotToAnsi } from "@getpaseo/protocol/terminal-snapshot"; -import type { TerminalStateSnapshot, TerminalStateSnapshotOptions } from "./terminal.js"; +import type { TerminalReplayOptions } from "./terminal-replay.js"; export const MAX_TERMINAL_OUTPUT_FRAME_BYTES = 256 * 1024; @@ -40,7 +38,7 @@ export function resolveRestoreAfterOutputOverflow( export function resolveTerminalRestoreSnapshotOptions( restore: TerminalRestoreOptions, -): TerminalStateSnapshotOptions | null | undefined { +): TerminalReplayOptions | null | undefined { if (restore.mode === "live") { return null; } @@ -52,25 +50,11 @@ export function resolveTerminalRestoreSnapshotOptions( return undefined; } -export function encodeLegacyTerminalSnapshotFrame(input: { - slot: number; - snapshot: TerminalStateSnapshot; -}): Uint8Array { - return encodeTerminalStreamFrame({ - opcode: TerminalStreamOpcode.Snapshot, - slot: input.slot, - payload: encodeTerminalSnapshotPayload(input.snapshot.state), - }); -} - -export function encodeTerminalRestoreFrame(input: { - slot: number; - snapshot: TerminalStateSnapshot; -}): Uint8Array { +export function encodeTerminalRestoreFrame(input: { slot: number; data: Uint8Array }): Uint8Array { return encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Restore, slot: input.slot, - payload: renderTerminalSnapshotToAnsi(input.snapshot.state), + payload: input.data, }); } diff --git a/packages/server/src/terminal/terminal-session-controller.test.ts b/packages/server/src/terminal/terminal-session-controller.test.ts index 27f69cb44d..6042db805e 100644 --- a/packages/server/src/terminal/terminal-session-controller.test.ts +++ b/packages/server/src/terminal/terminal-session-controller.test.ts @@ -1,3 +1,4 @@ +import type { TerminalReplayMetadata } from "./terminal-replay.js"; import { describe, expect, test, vi } from "vitest"; import pino from "pino"; @@ -7,8 +8,7 @@ import { decodeTerminalStreamFrame, type TerminalStreamFrame, } from "@getpaseo/protocol/binary-frames/index"; -import type { TerminalCell, TerminalState } from "@getpaseo/protocol/messages"; -import type { ServerMessage, TerminalSession, TerminalStateSnapshot } from "./terminal.js"; +import type { ServerMessage, TerminalSession } from "./terminal.js"; import { TerminalSessionController, type TerminalSessionControllerOptions, @@ -19,6 +19,29 @@ import { isSameOrDescendantPath } from "../server/path-utils.js"; import { PluginSessionSocket } from "../server/plugins/session-socket.js"; import { TerminalWorkerIncompatibleError } from "./terminal-worker-pool.js"; +interface ReplayFixture { + data: Uint8Array; + metadata: TerminalReplayMetadata; +} +function terminalReplay(text: string, revision: number): ReplayFixture { + return { + data: Buffer.from(text), + metadata: { rows: 1, cols: 80, retainedHistoryLines: 0, revision, replayPreamble: "" }, + }; +} +function replayFixture( + read: () => Promise, +): TerminalManager["replayTerminal"] { + return async (_id, _options, sink) => { + const replay = await read(); + if (!replay) return null; + sink({ type: "begin", metadata: replay.metadata }); + sink({ type: "data", data: replay.data }); + sink({ type: "end" }); + return replay.metadata; + }; +} + function createController( options: TerminalSessionControllerOptions & { emitBinary(frame: Uint8Array): void }, ) { @@ -59,22 +82,6 @@ function deferred(): { return { promise, resolve, reject }; } -function terminalRow(text: string, cols = 80): TerminalCell[] { - return Array.from({ length: cols }, (_, index) => ({ - char: text[index] ?? " ", - })); -} - -function terminalState(text: string): TerminalState { - return { - rows: 1, - cols: 80, - grid: [terminalRow(text)], - scrollback: [], - cursor: { row: 0, col: text.length }, - }; -} - function createLogger(): pino.Logger { return { warn: vi.fn(), @@ -85,7 +92,7 @@ function createLogger(): pino.Logger { describe("terminal-session-controller restore", () => { test("delivers output produced while restore is in flight after the restore frame", async () => { let terminalListener: ((message: ServerMessage) => void) | null = null; - const snapshot = deferred(); + const snapshot = deferred(); const binaryFrames: TerminalStreamFrame[] = []; const outboundMessages: SessionOutboundMessage[] = []; const terminal: TerminalSession = { @@ -104,8 +111,7 @@ describe("terminal-session-controller restore", () => { onTitleChange: () => vi.fn(), onActivityChange: () => vi.fn(), getSize: () => ({ rows: 1, cols: 80 }), - getState: () => terminalState("restore-before"), - getStateSnapshot: () => ({ state: terminalState("restore-before"), revision: 1 }), + hasOutput: () => true, getReplayPreamble: () => "\x1b[?1h\x1b[?2004h", getTitle: () => undefined, getActivity: () => null, @@ -121,7 +127,7 @@ describe("terminal-session-controller restore", () => { registerCwdEnv: vi.fn(), validateTerminalActivityToken: vi.fn(() => "unknown"), getTerminal: vi.fn(() => terminal), - getTerminalState: vi.fn(() => snapshot.promise), + replayTerminal: replayFixture(() => snapshot.promise), setTerminalTitle: vi.fn(), setTerminalActivity: vi.fn(), killTerminal: vi.fn(), @@ -157,10 +163,10 @@ describe("terminal-session-controller restore", () => { }, }); await Promise.resolve(); - expect(terminalManager.getTerminalState).toHaveBeenCalledTimes(1); + expect(binaryFrames).toEqual([]); terminalListener?.({ type: "output", data: "restore-after\n", revision: 2 }); - snapshot.resolve({ state: terminalState("restore-before"), revision: 1 }); + snapshot.resolve(terminalReplay("restore-before", 1)); await snapshot.promise; await new Promise((resolve) => setTimeout(resolve, 20)); @@ -202,8 +208,7 @@ function listSession(input: { onTitleChange: () => vi.fn(), onActivityChange: () => vi.fn(), getSize: () => ({ rows: 1, cols: 80 }), - getState: () => terminalState(""), - getStateSnapshot: () => ({ state: terminalState(""), revision: 0 }), + hasOutput: () => true, getReplayPreamble: () => "", getTitle: () => undefined, getActivity: () => null, @@ -236,7 +241,7 @@ describe("terminal-session-controller legacy terminal creation", () => { registerCwdEnv: vi.fn(), validateTerminalActivityToken: vi.fn(() => "unknown"), getTerminal: vi.fn(), - getTerminalState: vi.fn(), + replayTerminal: async () => null, setTerminalTitle: vi.fn(), setTerminalActivity: vi.fn(), clearTerminalAttention: vi.fn(), @@ -311,7 +316,7 @@ describe("terminal-session-controller legacy terminal creation", () => { registerCwdEnv: vi.fn(), validateTerminalActivityToken: vi.fn(() => "unknown"), getTerminal: vi.fn(), - getTerminalState: vi.fn(), + replayTerminal: async () => null, setTerminalTitle: vi.fn(), setTerminalActivity: vi.fn(), clearTerminalAttention: vi.fn(), @@ -362,7 +367,7 @@ describe("terminal-session-controller with terminals held by a newer Paseo", () registerCwdEnv: vi.fn(), validateTerminalActivityToken: vi.fn(() => "unknown"), getTerminal: vi.fn(), - getTerminalState: vi.fn(), + replayTerminal: async () => null, setTerminalTitle: vi.fn(), setTerminalActivity: vi.fn(), clearTerminalAttention: vi.fn(), @@ -428,99 +433,6 @@ async function waitForCoalescerFlush(): Promise { await new Promise((resolve) => setTimeout(resolve, 30)); } -describe("terminal-session-controller wrap-flag gating", () => { - function setup(clientSupportsWrapReflow?: () => boolean): { - controller: ReturnType; - getTerminalState: ReturnType; - } { - const terminal: TerminalSession = { - id: "term-1", - name: "Terminal", - cwd: "/tmp", - workspaceId: "ws-test", - send: vi.fn(), - subscribe: (listener) => { - queueMicrotask(() => listener({ type: "snapshotReady", revision: 1 })); - return vi.fn(); - }, - onExit: () => vi.fn(), - onCommandFinished: () => vi.fn(), - onTitleChange: () => vi.fn(), - onActivityChange: () => vi.fn(), - getSize: () => ({ rows: 1, cols: 80 }), - getState: () => terminalState("hello"), - getStateSnapshot: () => ({ state: terminalState("hello"), revision: 1 }), - getReplayPreamble: () => "", - getTitle: () => undefined, - getActivity: () => null, - setActivity: vi.fn(), - setTitle: vi.fn(), - getExitInfo: () => null, - kill: vi.fn(), - killAndWait: vi.fn(), - }; - const getTerminalState = vi.fn(() => - Promise.resolve({ state: terminalState("hello"), revision: 1 }), - ); - const terminalManager = { - getTerminals: vi.fn(), - createTerminal: vi.fn(), - registerCwdEnv: vi.fn(), - validateTerminalActivityToken: vi.fn(() => "unknown"), - getTerminal: vi.fn(() => terminal), - getTerminalState, - setTerminalTitle: vi.fn(), - setTerminalActivity: vi.fn(), - killTerminal: vi.fn(), - killTerminalAndWait: vi.fn(), - captureTerminal: vi.fn(), - listDirectories: vi.fn(() => []), - killAll: vi.fn(), - subscribeTerminalsChanged: vi.fn(() => vi.fn()), - subscribeTerminalActivity: vi.fn(() => vi.fn()), - subscribeTerminalWorkspaceContributionChanged: vi.fn(() => vi.fn()), - } as unknown as TerminalManager; - const controller = createController({ - terminalManager, - emit: vi.fn(), - emitBinary: vi.fn(), - hasBinaryChannel: () => true, - isPathWithinRoot: () => false, - sessionLogger: createLogger(), - ...(clientSupportsWrapReflow ? { clientSupportsWrapReflow } : {}), - }); - return { controller, getTerminalState }; - } - - async function subscribe(controller: ReturnType): Promise { - await controller.dispatch({ - type: "subscribe_terminal_request", - terminalId: "term-1", - requestId: "req-1", - restore: { mode: "visible-snapshot", scrollbackLines: 200 }, - }); - await flushMicrotasks(); - } - - test("requests wrap flags when the client supports reflowable snapshots", async () => { - const { controller, getTerminalState } = setup(() => true); - await subscribe(controller); - expect(getTerminalState).toHaveBeenCalledWith( - "term-1", - expect.objectContaining({ includeWrapFlags: true }), - ); - }); - - test("omits wrap flags when the client does not advertise support", async () => { - const { controller, getTerminalState } = setup(); - await subscribe(controller); - expect(getTerminalState).toHaveBeenCalledWith( - "term-1", - expect.objectContaining({ includeWrapFlags: false }), - ); - }); -}); - describe("terminal-session-controller subdirectory aggregation", () => { test("delivers a subdirectory change to a root subscriber as an aggregated, root-keyed snapshot", async () => { const rootCwd = "/work/repo"; @@ -541,7 +453,7 @@ describe("terminal-session-controller subdirectory aggregation", () => { registerCwdEnv: vi.fn(), validateTerminalActivityToken: vi.fn(() => "unknown"), getTerminal: vi.fn(), - getTerminalState: vi.fn(), + replayTerminal: async () => null, setTerminalTitle: vi.fn(), setTerminalActivity: vi.fn(), killTerminal: vi.fn(), @@ -609,7 +521,7 @@ describe("terminal-session-controller subdirectory aggregation", () => { registerCwdEnv: vi.fn(), validateTerminalActivityToken: vi.fn(() => "unknown"), getTerminal: vi.fn(), - getTerminalState: vi.fn(), + replayTerminal: async () => null, setTerminalTitle: vi.fn(), setTerminalActivity: vi.fn(), killTerminal: vi.fn(), @@ -701,7 +613,7 @@ describe("terminal-session-controller workspace-scoped subscriptions", () => { registerCwdEnv: vi.fn(), validateTerminalActivityToken: vi.fn(() => "unknown"), getTerminal: vi.fn(), - getTerminalState: vi.fn(), + replayTerminal: async () => null, setTerminalTitle: vi.fn(), setTerminalActivity: vi.fn(), killTerminal: vi.fn(), @@ -777,19 +689,15 @@ describe("terminal-session-controller backpressure snapshot fallback", () => { 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 }), - ); + // Readiness starts ANSI replay; later bytes use Output frames. + queueMicrotask(() => listener({ type: "snapshotReady", 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 }), + hasOutput: () => true, getReplayPreamble: () => "", getTitle: () => undefined, setTitle: vi.fn(), @@ -797,14 +705,12 @@ describe("terminal-session-controller backpressure snapshot fallback", () => { kill: vi.fn(), killAndWait: vi.fn(), }; - const terminalManager = { + const terminalManager: TerminalManager = { getTerminals: vi.fn(), createTerminal: vi.fn(), registerCwdEnv: vi.fn(), getTerminal: vi.fn(() => terminal), - getTerminalState: vi.fn(() => - Promise.resolve({ state: terminalState("live"), revision: 1 }), - ), + replayTerminal: replayFixture(async () => terminalReplay("live", 1)), setTerminalTitle: vi.fn(), killTerminal: vi.fn(), killTerminalAndWait: vi.fn(), @@ -812,7 +718,12 @@ describe("terminal-session-controller backpressure snapshot fallback", () => { listDirectories: vi.fn(() => []), killAll: vi.fn(), subscribeTerminalsChanged: vi.fn(() => vi.fn()), - } as unknown as TerminalManager; + validateTerminalActivityToken: () => "unknown", + setTerminalActivity: async () => true, + clearTerminalAttention: async () => false, + subscribeTerminalActivity: () => () => {}, + subscribeTerminalWorkspaceContributionChanged: () => () => {}, + }; const frames: TerminalStreamFrame[] = []; const controller = createController({ @@ -853,7 +764,7 @@ describe("terminal-session-controller backpressure snapshot fallback", () => { pushOutput(chunk); await waitForCoalescerFlush(); - expect(frames.some((frame) => frame.opcode === TerminalStreamOpcode.Snapshot)).toBe(false); + expect(frames.some((frame) => frame.opcode === TerminalStreamOpcode.Restore)).toBe(false); const outputFrames = frames.filter((frame) => frame.opcode === TerminalStreamOpcode.Output); expect(outputFrames.length).toBeGreaterThan(0); const receivedBytes = outputFrames.reduce( @@ -869,7 +780,7 @@ describe("terminal-session-controller backpressure snapshot fallback", () => { pushOutput("y".repeat(300 * 1024)); await waitForCoalescerFlush(); - expect(frames.some((frame) => frame.opcode === TerminalStreamOpcode.Snapshot)).toBe(true); + expect(frames.some((frame) => frame.opcode === TerminalStreamOpcode.Restore)).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. @@ -877,7 +788,7 @@ describe("terminal-session-controller backpressure snapshot fallback", () => { pushOutput("z".repeat(1024)); await waitForCoalescerFlush(); - expect(frames.some((frame) => frame.opcode === TerminalStreamOpcode.Snapshot)).toBe(false); + expect(frames.some((frame) => frame.opcode === TerminalStreamOpcode.Restore)).toBe(false); expect(frames.some((frame) => frame.opcode === TerminalStreamOpcode.Output)).toBe(true); }); @@ -893,7 +804,7 @@ describe("terminal-session-controller backpressure snapshot fallback", () => { pushOutput("p".repeat(300 * 1024)); await waitForCoalescerFlush(); - expect(frames.some((frame) => frame.opcode === TerminalStreamOpcode.Snapshot)).toBe(true); + expect(frames.some((frame) => frame.opcode === TerminalStreamOpcode.Restore)).toBe(true); }); test("falls back to a snapshot at the byte threshold when no backpressure signal exists", async () => { @@ -905,11 +816,11 @@ describe("terminal-session-controller backpressure snapshot fallback", () => { pushOutput("r".repeat(300 * 1024)); await waitForCoalescerFlush(); - expect(frames.some((frame) => frame.opcode === TerminalStreamOpcode.Snapshot)).toBe(true); + expect(frames.some((frame) => frame.opcode === TerminalStreamOpcode.Restore)).toBe(true); }); }); -function exitFixture(readSnapshot: () => Promise, backedUp = false) { +function exitFixture(readReplay: () => Promise, backedUp = false) { const outputs = new Set<(message: ServerMessage) => void>(); const exits = new Set[0]>(); const terminal: TerminalSession = { @@ -934,8 +845,7 @@ function exitFixture(readSnapshot: () => Promise, onTitleChange: () => () => {}, onActivityChange: () => () => {}, getSize: () => ({ rows: 1, cols: 80 }), - getState: () => terminalState("snapshot"), - getStateSnapshot: () => ({ state: terminalState("snapshot"), revision: 0 }), + hasOutput: () => true, getReplayPreamble: () => "", getTitle: () => undefined, getActivity: () => null, @@ -948,7 +858,7 @@ function exitFixture(readSnapshot: () => Promise, }; const manager: TerminalManager = { getTerminal: () => terminal, - getTerminalState: readSnapshot, + replayTerminal: replayFixture(readReplay), getTerminals: async () => [terminal], createTerminal: async () => terminal, registerCwdEnv() {}, @@ -1026,7 +936,7 @@ describe("terminal output before natural exit", () => { "flushes each observer's final bytes before exit (modern=%s)", async (modern) => { vi.useFakeTimers(); - const f = exitFixture(async () => ({ state: terminalState("snapshot"), revision: 0 })); + const f = exitFixture(async () => terminalReplay("snapshot", 0)); try { const source = {}; const owners = [ @@ -1048,7 +958,7 @@ describe("terminal output before natural exit", () => { row.message.payload.subscriptionId === owner.id)), ); expect(ordered.map((row) => row.binary?.opcode ?? row.message?.type)).toEqual([ - TerminalStreamOpcode.Snapshot, + TerminalStreamOpcode.Restore, TerminalStreamOpcode.Output, TerminalStreamOpcode.Output, "terminal_stream_exit", @@ -1068,7 +978,7 @@ describe("terminal output before natural exit", () => { test("waits for bootstrap snapshot and flushes buffered bytes without starting backpressure work", async () => { vi.useFakeTimers(); - const snapshot = deferred(); + const snapshot = deferred(); let reads = 0; const f = exitFixture(() => { reads++; @@ -1081,10 +991,10 @@ describe("terminal output before natural exit", () => { f.exit(); await vi.advanceTimersByTimeAsync(0); expect(f.frames.map((row) => row.message?.type)).toEqual(["subscribe_terminal_response"]); - snapshot.resolve({ state: terminalState("snapshot"), revision: 0 }); + snapshot.resolve(terminalReplay("snapshot", 0)); await vi.advanceTimersByTimeAsync(0); expect(f.frames.slice(1).map((row) => row.binary?.opcode ?? row.message?.type)).toEqual([ - TerminalStreamOpcode.Snapshot, + TerminalStreamOpcode.Restore, TerminalStreamOpcode.Output, "terminal_stream_exit", ]); @@ -1104,11 +1014,10 @@ describe("terminal output before natural exit", () => { "finishes an in-flight backpressure snapshot (available=%s)", async (available) => { vi.useFakeTimers(); - const snapshot = deferred(); + const snapshot = deferred(); let reads = 0; const f = exitFixture( - async () => - ++reads === 1 ? { state: terminalState("initial"), revision: 0 } : snapshot.promise, + async () => (++reads === 1 ? terminalReplay("initial", 0) : snapshot.promise), true, ); try { @@ -1122,10 +1031,7 @@ describe("terminal output before natural exit", () => { await vi.advanceTimersByTimeAsync(0); expect(f.frames.some((row) => row.message?.type === "terminal_stream_exit")).toBe(false); if (available === "rejected") snapshot.reject(new Error("Snapshot failed during exit")); - else - snapshot.resolve( - available ? { state: terminalState("pressure-snapshot"), revision: 1 } : null, - ); + else snapshot.resolve(available ? terminalReplay("pressure-snapshot", 1) : null); await vi.advanceTimersByTimeAsync(0); const output = f.frames .filter((row) => row.binary?.opcode === TerminalStreamOpcode.Output) @@ -1145,7 +1051,7 @@ describe("terminal output before natural exit", () => { test("release during exit's pending snapshot suppresses late delivery", async () => { vi.useFakeTimers(); - const snapshot = deferred(); + const snapshot = deferred(); const f = exitFixture(() => snapshot.promise); const source = {}; try { @@ -1158,7 +1064,7 @@ describe("terminal output before natural exit", () => { subscriptionId: owner.id!, }; const release = f.delivery.request(source, request, () => f.delivery.release(owner.id!)); - snapshot.resolve({ state: terminalState("snapshot"), revision: 0 }); + snapshot.resolve(terminalReplay("snapshot", 0)); await release; await vi.advanceTimersByTimeAsync(0); expect(f.frames.map((row) => row.message?.type)).toEqual(["subscribe_terminal_response"]); @@ -1173,7 +1079,7 @@ describe("terminal output before natural exit", () => { test("explicit release discards only that observer's trailing output", async () => { vi.useFakeTimers(); - const f = exitFixture(async () => ({ state: terminalState("snapshot"), revision: 0 })); + const f = exitFixture(async () => terminalReplay("snapshot", 0)); const source = {}; try { const a = await f.subscribe(source, true, "a"); @@ -1221,8 +1127,8 @@ describe("terminal snapshot failure", () => { let reads = 0; const f = exitFixture(async () => { if (++reads === (phase === "backpressure" ? 2 : 1)) - throw new Error("Terminal worker request timed out: getTerminalState"); - return { state: terminalState("healthy"), revision: 0 }; + throw new Error("Terminal worker request timed out: replayTerminal"); + return terminalReplay("healthy", 0); }, phase === "backpressure"); try { const source = {}; @@ -1250,7 +1156,7 @@ describe("terminal snapshot failure", () => { payload: { terminalId: "exit-terminal", subscriptionId: failed.id, - error: "Terminal worker request timed out: getTerminalState", + error: "Terminal worker request timed out: replayTerminal", }, }, ]); @@ -1281,7 +1187,7 @@ describe("terminal snapshot failure", () => { ); test("release during pending rejection suppresses the failure and completes cleanup", async () => { - const snapshot = deferred(); + const snapshot = deferred(); const f = exitFixture(() => snapshot.promise); const source = {}; try { @@ -1293,7 +1199,7 @@ describe("terminal snapshot failure", () => { subscriptionId: owner.id!, }; const release = f.delivery.request(source, request, () => f.delivery.release(owner.id!)); - snapshot.reject(new Error("Terminal worker request timed out: getTerminalState")); + snapshot.reject(new Error("Terminal worker request timed out: replayTerminal")); await release; expect(f.frames.map((row) => row.message?.type)).toEqual(["subscribe_terminal_response"]); expect(f.delivery.registrationCount).toBe(0); @@ -1313,7 +1219,7 @@ test.each([false, true])( let reads = 0; const f = exitFixture(async () => { if (++reads === (backpressure ? 2 : 1)) throw new Error("Snapshot unavailable"); - return { state: terminalState("recovered"), revision: 0 }; + return terminalReplay("recovered", 0); }, backpressure); try { await f.subscribe({}, false, "legacy-recovery"); @@ -1327,7 +1233,7 @@ test.each([false, true])( for (const listener of f.outputs) listener({ type: "snapshotReady", revision: 2 }); await vi.advanceTimersByTimeAsync(5); expect(reads).toBe(backpressure ? 3 : 2); - expect(f.frames.some((row) => row.binary?.opcode === TerminalStreamOpcode.Snapshot)).toBe( + expect(f.frames.some((row) => row.binary?.opcode === TerminalStreamOpcode.Restore)).toBe( true, ); f.exit(); @@ -1347,3 +1253,45 @@ test.each([false, true])( } }, ); + +test.each([true, false])( + "ends every viewer of an interrupted terminal and permits a fresh attach (modern=%s)", + async (modern) => { + const f = exitFixture(async () => terminalReplay("current screen", 2)); + const source = {}; + const owners = [ + await f.subscribe(source, modern, "existing"), + await f.subscribe(modern ? source : {}, modern, "joining"), + ]; + await waitForCoalescerFlush(); + f.frames.length = 0; + const error = "Terminal output exceeded the replay buffer; attach again"; + for (const listener of Array.from(f.outputs)) listener({ type: "streamError", error }); + await flushMicrotasks(); + expect(f.frames.map((row) => row.message?.type)).toEqual([ + "terminal_stream_exit", + "terminal_stream_exit", + ]); + for (const owner of owners) + expect(f.frames).toContainEqual({ + source: owner.source, + message: { + type: "terminal_stream_exit", + payload: { + terminalId: "exit-terminal", + error, + ...(modern ? { subscriptionId: owner.id } : {}), + }, + }, + }); + expect(f.outputs.size).toBe(0); + expect(f.exits.size).toBe(0); + f.output("must not render against stale screen", 3); + expect(f.frames.length).toBe(2); + await f.subscribe({}, modern, "recovered"); + await waitForCoalescerFlush(); + const restores = f.frames.flatMap((row) => (row.binary ? [row.binary] : [])); + expect(restores.map((frame) => frame.opcode)).toEqual([TerminalStreamOpcode.Restore]); + expect(new TextDecoder().decode(restores[0]!.payload)).toBe("current screen"); + }, +); diff --git a/packages/server/src/terminal/terminal-session-controller.ts b/packages/server/src/terminal/terminal-session-controller.ts index e41186fdb1..b53570cfb4 100644 --- a/packages/server/src/terminal/terminal-session-controller.ts +++ b/packages/server/src/terminal/terminal-session-controller.ts @@ -28,7 +28,6 @@ import { TerminalOutputCoalescer } from "./terminal-output-coalescer.js"; import { MAX_CLIENT_BUFFERED_BYTES, MAX_TERMINAL_OUTPUT_FRAME_BYTES, - encodeLegacyTerminalSnapshotFrame, encodeTerminalRestoreFrame, resolveRestoreAfterOutputOverflow, resolveTerminalRestoreSnapshotOptions, @@ -88,11 +87,6 @@ export interface TerminalSessionControllerOptions { sessionLogger: pino.Logger; listTerminalWorkspaceRefs?: () => Promise; listTerminalWorkspaceRoots?: () => Promise; - // Whether the connected client can reflow restored snapshots. When true the - // daemon attaches per-row soft-wrap flags to snapshots; otherwise it omits them - // so old (strict-schema) clients still parse the snapshot. - clientSupportsWrapReflow?: (source: object) => 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 @@ -145,7 +139,6 @@ export class TerminalSessionController { private readonly sessionLogger: pino.Logger; private readonly listTerminalWorkspaceRefs: () => Promise; private readonly listTerminalWorkspaceRoots: () => Promise; - private readonly clientSupportsWrapReflow: (source: object) => boolean; private readonly getClientBufferedAmount: (source: object) => number | null; private readonly subscribedDirectories = new Map(); @@ -163,7 +156,6 @@ export class TerminalSessionController { this.listTerminalWorkspaceRoots = options.listTerminalWorkspaceRoots ?? (async () => (await this.listTerminalWorkspaceRefs()).map((workspace) => workspace.cwd)); - this.clientSupportsWrapReflow = options.clientSupportsWrapReflow ?? (() => false); this.getClientBufferedAmount = options.getClientBufferedAmount ?? (() => 0); } @@ -957,6 +949,14 @@ export class TerminalSessionController { if (this.activeStreams.get(slot) !== activeStream) { return; } + if (message.type === "streamError") { + activeStream.owner.emit({ + type: "terminal_stream_exit", + payload: { terminalId: activeStream.terminalId, error: message.error }, + }); + void activeStream.owner.release(); + return; + } if (message.type === "snapshot" || message.type === "snapshotReady") { activeStream.readyRevision = message.revision; activeStream.outputCoalescer.flush(); @@ -1027,7 +1027,7 @@ export class TerminalSessionController { const restore = activeStream.restore; const snapshotResult = restore ? await this.emitRestoreSnapshot(activeStream, terminalManager, restore) - : await this.emitLegacySnapshot(activeStream, terminalManager); + : await this.emitRestoreSnapshot(activeStream, terminalManager, { mode: "full-snapshot" }); if (!snapshotResult.shouldContinue) { return; } @@ -1070,33 +1070,6 @@ export class TerminalSessionController { } } - private async emitLegacySnapshot( - activeStream: ActiveTerminalStream, - terminalManager: TerminalManager, - ): Promise { - const snapshot = await terminalManager.getTerminalState(activeStream.terminalId, { - includeWrapFlags: this.clientSupportsWrapReflow(activeStream.owner.source), - }); - if (this.activeStreams.get(activeStream.slot) !== activeStream) { - return { shouldContinue: false }; - } - if (!snapshot) { - this.detachStream(activeStream.terminalId, { emitExit: true }); - return { shouldContinue: false }; - } - - activeStream.owner.emitBinary( - encodeLegacyTerminalSnapshotFrame({ - slot: activeStream.slot, - 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 }; - } - private async emitRestoreSnapshot( activeStream: ActiveTerminalStream, terminalManager: TerminalManager, @@ -1107,10 +1080,16 @@ export class TerminalSessionController { return { shouldContinue: true }; } - const snapshot = await terminalManager.getTerminalState(activeStream.terminalId, { - ...snapshotOptions, - includeWrapFlags: this.clientSupportsWrapReflow(activeStream.owner.source), - }); + const chunks: Uint8Array[] = []; + // COMPAT(terminalSingleRestore): ANSI clients through v0.14 collect one restore; + // remove after the supported client floor implements chunked replay. + const snapshot = await terminalManager.replayTerminal( + activeStream.terminalId, + snapshotOptions, + (event) => { + if (event.type === "data") chunks.push(event.data); + }, + ); if (this.activeStreams.get(activeStream.slot) !== activeStream) { return { shouldContinue: false }; } @@ -1122,7 +1101,7 @@ export class TerminalSessionController { activeStream.owner.emitBinary( encodeTerminalRestoreFrame({ slot: activeStream.slot, - snapshot, + data: Buffer.concat(chunks), }), ); // The restore frame went out-of-band; keep the replay that follows on the diff --git a/packages/server/src/terminal/terminal-worker-client.test.ts b/packages/server/src/terminal/terminal-worker-client.test.ts index a363ab500d..01a4dd91d3 100644 --- a/packages/server/src/terminal/terminal-worker-client.test.ts +++ b/packages/server/src/terminal/terminal-worker-client.test.ts @@ -72,3 +72,46 @@ describe("terminal worker authentication", () => { await waitUntil(() => !existsSync(recordPath), 10_000); }); }); + +it("bounds a stalled framed writer to one chunk and resumes in order", async () => { + const { createServer, createConnection } = await import("node:net"); + const { attachFramedSocket } = await import("./terminal-worker-frames.js"); + const server = createServer(); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("Missing test address"); + const accepted = new Promise((resolve) => + server.once("connection", resolve), + ); + const reader = createConnection(address.port, "127.0.0.1"); + reader.pause(); + const writer = await accepted; + const framed = attachFramedSocket(writer, () => {}); + let completed = 0; + let peakQueuedBytes = 0; + const received: number[] = []; + attachFramedSocket(reader, (value) => received.push((value as { index: number }).index)); + const sending = (async () => { + for (let index = 0; index < 512; index++) { + const pending = framed.sendAndWait({ index, data: Buffer.alloc(64 * 1024) }); + peakQueuedBytes = Math.max(peakQueuedBytes, writer.writableLength); + await pending; + completed++; + } + })(); + try { + await waitUntil(() => writer.writableLength > 0, 10000); + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(completed).toBeLessThan(512); + expect(peakQueuedBytes).toBeLessThan(65 * 1024); + reader.resume(); + await sending; + await waitUntil(() => received.length === 512, 10000); + expect(received).toEqual(Array.from({ length: 512 }, (_, index) => index)); + expect(writer.destroyed).toBe(false); + } finally { + reader.destroy(); + writer.destroy(); + await new Promise((resolve) => server.close(() => resolve())); + } +}); diff --git a/packages/server/src/terminal/terminal-worker-frames.ts b/packages/server/src/terminal/terminal-worker-frames.ts index f0b990e239..897e91826f 100644 --- a/packages/server/src/terminal/terminal-worker-frames.ts +++ b/packages/server/src/terminal/terminal-worker-frames.ts @@ -50,6 +50,7 @@ export class FrameDecoder { export interface FramedSocket { readonly socket: Socket; send(message: unknown, callback?: (error: Error | null) => void): boolean; + sendAndWait(message: unknown): Promise; close(): void; } @@ -91,6 +92,17 @@ export function attachFramedSocket( } return socket.write(frame, (error) => callback?.(error ?? null)); }, + sendAndWait(message) { + return new Promise((resolve, reject) => { + this.send(message, (error) => { + if (error) { + reject(error); + return; + } + resolve(); + }); + }); + }, close() { socket.end(() => socket.destroy()); }, diff --git a/packages/server/src/terminal/terminal-worker-pool.ts b/packages/server/src/terminal/terminal-worker-pool.ts index f15ad5b46d..517d34aa33 100644 --- a/packages/server/src/terminal/terminal-worker-pool.ts +++ b/packages/server/src/terminal/terminal-worker-pool.ts @@ -5,7 +5,7 @@ import type { TerminalsChangedListener, TerminalWorkspaceContributionChangedListener, } from "./terminal-manager.js"; -import type { TerminalSession, TerminalStateSnapshot } from "./terminal.js"; +import type { TerminalSession } from "./terminal.js"; import type { WorkerCreateTerminalOptions } from "./terminal-worker-protocol.js"; import type { DaemonTerminalManager } from "./worker-terminal-manager.js"; @@ -217,8 +217,8 @@ export function createTerminalWorkerPool( return undefined; }, - async getTerminalState(id, stateOptions): Promise { - return routed(id).getTerminalState(id, stateOptions); + async replayTerminal(id, replayOptions, sink) { + return routed(id).replayTerminal(id, replayOptions, sink); }, setTerminalTitle(id, title) { diff --git a/packages/server/src/terminal/terminal-worker-process.ts b/packages/server/src/terminal/terminal-worker-process.ts index 6c13ba04d5..7a22caaa02 100644 --- a/packages/server/src/terminal/terminal-worker-process.ts +++ b/packages/server/src/terminal/terminal-worker-process.ts @@ -1,3 +1,4 @@ +import { createTerminalReplayController } from "./terminal-replay.js"; import { spawn } from "node:child_process"; import { randomBytes, timingSafeEqual } from "node:crypto"; import { chmodSync, closeSync, existsSync, openSync, rmSync, writeSync } from "node:fs"; @@ -8,11 +9,7 @@ import { PRIVATE_FILE_MODE } from "../server/private-files.js"; import { createTerminalManager } from "./terminal-manager.js"; import { captureTerminalLines } from "./terminal-capture.js"; import { TerminalOutputCoalescer } from "./terminal-output-coalescer.js"; -import type { - TerminalExitInfo, - TerminalSession, - TerminalStateSnapshotOptions, -} from "./terminal.js"; +import type { TerminalExitInfo, LocalTerminalSession } from "./terminal.js"; import { attachFramedSocket, type FramedSocket } from "./terminal-worker-frames.js"; import { createTerminalWorkerProcessInspector } from "./terminal-worker-processes.js"; import { @@ -30,7 +27,6 @@ import { type TerminalWorkerHelloAck, type TerminalWorkerListResult, type TerminalWorkerRequest, - type TerminalWorkerStateResult, type TerminalWorkerToParentMessage, type WorkerTerminalInfo, } from "./terminal-worker-protocol.js"; @@ -177,7 +173,7 @@ function runWorker(argv: string[]): void { const protocolVersion = reportedProtocolVersion(); const manager = createTerminalManager(); const inspector = createTerminalWorkerProcessInspector(); - const sessionsById = new Map(); + const sessionsById = new Map(); const unsubscribeByTerminalId = new Map void>>(); const outputCoalescerByTerminalId = new Map(); // A terminal is published (terminalCreated) only once its program is recorded. An exit that @@ -211,10 +207,12 @@ function runWorker(argv: string[]): void { reportInFlightTerminalCreateFailure(error); }); + const replayController = createTerminalReplayController({ + getSession: (id) => sessionsById.get(id), + flushOutput: (id) => outputCoalescerByTerminalId.get(id)?.flush(), + }); function sendToController(message: TerminalWorkerToParentMessage): void { - // With no controller the event is dropped; the headless terminal keeps the screen, and the - // next daemon reads it through listTerminals. - controller?.send(message); + replayController.send(message); } function persistRecord(): void { @@ -259,17 +257,11 @@ function runWorker(argv: string[]): void { } } - function buildTerminalStateResult( - session: TerminalSession | undefined, - options?: TerminalStateSnapshotOptions, - ): TerminalWorkerStateResult { - if (!session) { - return null; - } - return { ...session.getStateSnapshot(options), replayPreamble: session.getReplayPreamble() }; + function terminalMetadata(session: LocalTerminalSession) { + return { ...session.getSize(), hasOutput: session.hasOutput(), title: session.getTitle() }; } - function toTerminalInfo(session: TerminalSession): WorkerTerminalInfo { + function toTerminalInfo(session: LocalTerminalSession): WorkerTerminalInfo { return { id: session.id, name: session.name, @@ -316,7 +308,7 @@ function runWorker(argv: string[]): void { } } - function watchTerminal(session: TerminalSession): void { + function watchTerminal(session: LocalTerminalSession): void { clearTerminalSubscriptions(session.id); // Coalesce pty output chunks into a single frame per ~5ms window so a burst of small @@ -340,6 +332,7 @@ function runWorker(argv: string[]): void { const unsubscribeMessage = session.subscribe( (message) => { + if (message.type === "snapshot") return; if (message.type === "output") { pendingOutputRevision = message.revision; outputCoalescer.handle(message.data); @@ -413,7 +406,7 @@ function runWorker(argv: string[]): void { // A terminal enters the record only with its program's birth time, so cleanup can always // tell the program from a later process that reuses its pid. A program that already exited // needs no entry; one whose birth time cannot be read is killed and the create fails. - async function recordTerminalProgram(session: TerminalSession): Promise { + async function recordTerminalProgram(session: LocalTerminalSession): Promise { if (session.getExitInfo()) return; // The manager drops a session as soon as its program exits, so only a live one has a pid. const pid = manager.getTerminalProcessId(session.id); @@ -476,11 +469,11 @@ function runWorker(argv: string[]): void { await session.killAndWait().catch(() => undefined); throw error; } - const initialSnapshot = session.getStateSnapshot(); + const state = terminalMetadata(session); sendToController({ type: "terminalCreated", terminal: toTerminalInfo(session), - state: initialSnapshot.state, + state, }); sendToController({ type: "response", @@ -488,7 +481,7 @@ function runWorker(argv: string[]): void { ok: true, result: { terminal: toTerminalInfo(session), - state: initialSnapshot.state, + state, }, }); const heldExit = exitsHeldDuringCreation.get(session.id); @@ -513,7 +506,7 @@ function runWorker(argv: string[]): void { outputCoalescerByTerminalId.get(session.id)?.flush(); return { terminal: toTerminalInfo(session), - state: session.getStateSnapshot().state, + state: terminalMetadata(session), activityToken: manager.getTerminalActivityToken(session.id), replayPreamble: session.getReplayPreamble(), }; @@ -559,21 +552,8 @@ function runWorker(argv: string[]): void { return; } - case "getTerminalState": { - // 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(); - sendToController({ - type: "response", - requestId: message.requestId, - ok: true, - result: buildTerminalStateResult( - manager.getTerminal(message.terminalId), - message.options, - ), - }); + case "replayTerminal": { + await replayController.replay(message); return; } @@ -659,10 +639,9 @@ function runWorker(argv: string[]): void { } authenticated = true; // One controller at a time: a new authenticated connection replaces the old one. - const previous = controller; controller = framed; + replayController.setConnection(framed); everControlled = true; - previous?.socket.destroy(); const ack: TerminalWorkerHelloAck = { type: "helloAck", pid: process.pid, @@ -688,6 +667,7 @@ function runWorker(argv: string[]): void { clearTimeout(helloTimer); if (controller === framed) { controller = null; + replayController.setConnection(null); exitIfIdle(); } }); diff --git a/packages/server/src/terminal/terminal-worker-protocol.test.ts b/packages/server/src/terminal/terminal-worker-protocol.test.ts index c3810bf9cf..b3e7074b1d 100644 --- a/packages/server/src/terminal/terminal-worker-protocol.test.ts +++ b/packages/server/src/terminal/terminal-worker-protocol.test.ts @@ -2,6 +2,9 @@ import { describe, expect, it } from "vitest"; import { classifyTerminalWorkerProtocol } from "./terminal-worker-protocol.js"; describe("classifyTerminalWorkerProtocol", () => { + it("rejects the unreleased cell protocol without a draining adapter", () => { + expect(classifyTerminalWorkerProtocol(1, 2)).toBe("obsolete"); + }); it.each([ [5, "current"], [4, "previous"], diff --git a/packages/server/src/terminal/terminal-worker-protocol.ts b/packages/server/src/terminal/terminal-worker-protocol.ts index 86fc4a7772..a22b5a6adf 100644 --- a/packages/server/src/terminal/terminal-worker-protocol.ts +++ b/packages/server/src/terminal/terminal-worker-protocol.ts @@ -1,11 +1,11 @@ -import type { - TerminalExitInfo, - ServerMessage, - ClientMessage, - TerminalStateSnapshot, - TerminalStateSnapshotOptions, -} from "./terminal.js"; -import type { TerminalState } from "@getpaseo/protocol/messages"; +import type { TerminalExitInfo, ServerMessage, ClientMessage } from "./terminal.js"; +import type { TerminalReplayEvent, TerminalReplayOptions } from "./terminal-replay.js"; +export interface WorkerTerminalState { + rows: number; + cols: number; + hasOutput: boolean; + title?: string; +} import type { TerminalActivity, TerminalActivityState } from "@getpaseo/protocol/terminal-activity"; import type { CaptureTerminalLinesResult } from "./terminal-capture.js"; @@ -43,7 +43,7 @@ export interface WorkerCreateTerminalOptions { // `shutdown`. A daemon drives a worker of its own version, drains one a version older (no new // terminals; it exits after its last terminal), shuts down anything older, and leaves a newer // worker running untouched. See docs/terminal-worker.md. -export const TERMINAL_WORKER_PROTOCOL_VERSION = 1; +export const TERMINAL_WORKER_PROTOCOL_VERSION = 2; // Test-only: the protocol version a worker reports, so tests can stand up older and newer // workers from one build. @@ -59,7 +59,8 @@ export function classifyTerminalWorkerProtocol( // A version this daemon cannot read comes from a Paseo it does not know: treat it as newer. if (typeof reported !== "number" || !Number.isSafeInteger(reported)) return "newer"; if (reported === current) return "current"; - if (reported === current - 1) return "previous"; + // Protocol 1 was unreleased and transported cells; it has no draining adapter. + if (reported === current - 1 && reported >= 2) return "previous"; return reported < current ? "obsolete" : "newer"; } @@ -77,7 +78,7 @@ export interface TerminalWorkerHelloAck { export interface WorkerListedTerminal { terminal: WorkerTerminalInfo; - state: TerminalState; + state: WorkerTerminalState; activityToken: string | null; replayPreamble: string; } @@ -122,10 +123,10 @@ export type TerminalWorkerRequest = options?: WorkerKillAndWaitOptions; } | { - type: "getTerminalState"; + type: "replayTerminal"; requestId: string; terminalId: string; - options?: TerminalStateSnapshotOptions; + options?: TerminalReplayOptions; } | { type: "captureTerminal"; @@ -166,15 +167,16 @@ export type TerminalWorkerResponse = }; export type TerminalWorkerEvent = + | { type: "terminalReplay"; requestId: string; event: TerminalReplayEvent } | { type: "terminalCreated"; terminal: WorkerTerminalInfo; - state: TerminalState; + state: WorkerTerminalState; } | { type: "terminalMessage"; terminalId: string; - message: ServerMessage; + message: Exclude; replayPreamble: string; } | { @@ -206,6 +208,3 @@ export type TerminalWorkerToParentMessage = TerminalWorkerResponse | TerminalWor export type TerminalWorkerListResult = WorkerListedTerminal[]; 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.ts b/packages/server/src/terminal/terminal.ts index b8cb7ee8cd..f54f0860d0 100644 --- a/packages/server/src/terminal/terminal.ts +++ b/packages/server/src/terminal/terminal.ts @@ -92,6 +92,7 @@ export type ClientMessage = | { type: "mouse"; row: number; col: number; button: number; action: "down" | "up" | "move" }; export type ServerMessage = + | { type: "streamError"; error: string } | { type: "output"; data: string; revision?: number } | { type: "snapshot"; state: TerminalState; revision?: number } | { type: "snapshotReady"; revision?: number; replayPreamble?: string } @@ -114,8 +115,7 @@ export interface TerminalSession { onTitleChange(listener: (title?: string) => void): () => void; onActivityChange(listener: (transition: TerminalActivityTransition) => void): () => void; getSize(): { rows: number; cols: number }; - getState(): TerminalState; - getStateSnapshot(options?: TerminalStateSnapshotOptions): TerminalStateSnapshot; + hasOutput(): boolean; getReplayPreamble(): string; getTitle(): string | undefined; getActivity(): TerminalActivity | null; @@ -130,6 +130,9 @@ export interface TerminalSession { /** A session that owns its PTY in this process, so it knows the program's pid. */ export interface LocalTerminalSession extends TerminalSession { readonly pid: number; + getRetainedHistoryLines(): number; + getState(): TerminalState; + getStateSnapshot(options?: TerminalStateSnapshotOptions): TerminalStateSnapshot; } function parseCommandFinishedOsc(data: string): TerminalCommandFinishedInfo | null { @@ -1606,6 +1609,11 @@ export async function createTerminal( getSize, getState, getStateSnapshot, + getRetainedHistoryLines: () => terminal.buffer.active.baseY, + hasOutput: () => + [...getState().scrollback, ...getState().grid].some((row) => + row.some((cell) => cell.char.trim().length > 0), + ), getReplayPreamble, getTitle, getActivity, diff --git a/packages/server/src/terminal/worker-terminal-manager.test.ts b/packages/server/src/terminal/worker-terminal-manager.test.ts index d953d6cb69..01b113d0ef 100644 --- a/packages/server/src/terminal/worker-terminal-manager.test.ts +++ b/packages/server/src/terminal/worker-terminal-manager.test.ts @@ -1,3 +1,6 @@ +import { createConnection } from "node:net"; +import { attachFramedSocket } from "./terminal-worker-frames.js"; +import xterm from "@xterm/headless"; import { afterEach, expect, it } from "vitest"; import { isPlatform } from "../test-utils/platform.js"; import { EventEmitter } from "node:events"; @@ -77,8 +80,35 @@ async function waitForCondition( throw new Error(`Timed out after ${timeoutMs}ms waiting for condition`); } -function getVisibleText(session: TerminalSession): string { - return getVisibleTextFromState(session.getState()); +async function readReplay(manager: DaemonTerminalManager, id: string) { + const chunks: Uint8Array[] = []; + const metadata = await manager.replayTerminal(id, undefined, (event) => { + if (event.type === "data") chunks.push(event.data); + }); + if (!metadata) return null; + const terminal = new xterm.Terminal({ + rows: metadata.rows, + cols: metadata.cols, + scrollback: 1000, + allowProposedApi: true, + }); + await new Promise((resolve) => terminal.write(Buffer.concat(chunks), resolve)); + const buffer = terminal.buffer.active; + const rows = Array.from({ length: buffer.length }, (_, i) => { + const line = buffer.getLine(i)!; + return Array.from({ length: metadata.cols }, (_cell, col) => ({ + char: line.getCell(col)?.getChars() || " ", + })); + }); + const state: TerminalState = { + rows: metadata.rows, + cols: metadata.cols, + scrollback: rows.slice(0, buffer.baseY), + grid: rows.slice(buffer.baseY), + cursor: { row: buffer.cursorY, col: buffer.cursorX }, + }; + terminal.dispose(); + return { state, revision: metadata.revision }; } function rowToText(row: TerminalRow): string { @@ -277,7 +307,7 @@ it("reconnects a live worker without replacing terminal sessions or processes", await waitForCondition(() => !competitor.connected, 10000); await waitForCondition(async () => { try { - return (await manager!.getTerminalState(terminal.id)) !== null; + return (await readReplay(manager!, terminal.id)) !== null; } catch { return false; } @@ -343,7 +373,8 @@ it.each(["current", "draining"] as const)( "keeps the peer usable when the %s worker dies", async (target) => { const paseoHome = createTemporaryHome(); - const old = await leaveWorkerWithTerminal(paseoHome, 0); + await launchDetachedWorkerWithTerminal(paseoHome); + const old = await launchDetachedWorkerWithTerminal(paseoHome); manager = await startRealWorkerManager({ paseoHome }); const added = await manager.createTerminal({ workspaceId: "ws-peer", @@ -359,7 +390,7 @@ it.each(["current", "draining"] as const)( process.kill(target === "current" ? current.pid : old.workerPid, "SIGKILL"); await waitForCondition(() => !manager!.getTerminal(deadId), 10000); expect(manager.getTerminal(healthyId)).toBe(healthy); - expect(await manager.getTerminalState(healthyId)).not.toBeNull(); + expect(await readReplay(manager, healthyId)).not.toBeNull(); await manager.createTerminal({ workspaceId: "ws-peer", cwd: old.cwd, @@ -401,16 +432,11 @@ it("creates a terminal through the worker and streams output", async () => { session.send({ type: "input", data: "hello\r" }); - await waitForCondition( - () => - messages.join("").includes("worker-output:hello") || - getVisibleText(session).includes("worker-output:hello"), - 10000, - ); + await waitForCondition(() => messages.join("").includes("worker-output:hello"), 10000); await new Promise((resolve) => setTimeout(resolve, 100)); unsubscribe(); - expect(messages.join("") + getVisibleText(session)).toContain("worker-output:hello"); + expect(messages.join("")).toContain("worker-output:hello"); expect(snapshots).toBe(snapshotsBeforeOutput); }); @@ -459,7 +485,7 @@ it("delivers rapid small writes complete and in order through worker coalescing" const expected = Array.from({ length: 500 }, (_, index) => index); let received: number[] = []; await waitForCondition(async () => { - const snapshot = await manager!.getTerminalState(session.id); + const snapshot = await readReplay(manager!, session.id); received = snapshot ? readRenderedTokenIndexesFromState(snapshot.state) : []; return expected.every((value, index) => received[index] === value); }, 10000); @@ -495,7 +521,7 @@ it("pulls fresh terminal state from the worker authority", async () => { let visibleText = ""; await waitForCondition(async () => { - const snapshot = await manager!.getTerminalState(session.id); + const snapshot = await readReplay(manager!, session.id); visibleText = snapshot ? getVisibleTextFromState(snapshot.state) : ""; return visibleText.includes("worker-state-ready"); }, 10000); @@ -527,7 +553,7 @@ it.skipIf(isPlatform("win32"))( ); await waitForCondition(async () => { - const snapshot = await manager!.getTerminalState(session.id); + const snapshot = await readReplay(manager!, session.id); return snapshot !== null && session.getReplayPreamble() === "\x1b[=1;1u"; }, 10000); @@ -552,7 +578,7 @@ it("refreshes cached terminal title after worker title changes", async () => { await waitForCondition(() => session.getTitle() === "Build Output", 10000); - expect(session.getState().title).toBe("Build Output"); + expect(session.getTitle()).toBe("Build Output"); }); it("refreshes cached terminal size after worker resize", async () => { @@ -562,12 +588,12 @@ it("refreshes cached terminal size after worker resize", async () => { const session = trackTerminal(await manager.createTerminal({ cwd, workspaceId: "ws-test" })); session.send({ type: "resize", rows: 10, cols: 40 }); - const snapshot = await manager.getTerminalState(session.id); + const snapshot = await readReplay(manager, session.id); expect(snapshot?.state.rows).toBe(10); expect(snapshot?.state.cols).toBe(40); - expect(session.getState().rows).toBe(10); - expect(session.getState().cols).toBe(40); + expect(session.getSize().rows).toBe(10); + expect(session.getSize().cols).toBe(40); }); it("captures terminal output from the worker authority", async () => { @@ -854,7 +880,7 @@ async function expectCompletedShortCommand( expect(session.getExitInfo()?.exitCode).toBe(0); expect(terminalManager.getTerminal(session.id)).toBeUndefined(); expect(await terminalManager.getTerminals(cwd)).toEqual([]); - expect(await terminalManager.getTerminalState(session.id)).toBeNull(); + expect(await readReplay(terminalManager, session.id)).toBeNull(); } it.skipIf(isPlatform("win32"))( @@ -1531,41 +1557,9 @@ async function leaveWorkerWithTerminal( }; } -it("drains a worker one protocol older: it keeps its terminal, new ones open in a second worker, and it exits after its last", async () => { +it("hard cuts the unreleased cell transport, closing its terminals", async () => { const paseoHome = createTemporaryHome(); - const old = await leaveWorkerWithTerminal(paseoHome, 0); - - manager = await startRealWorkerManager({ paseoHome }); - - expect(manager.getTerminal(old.terminalId)).toBeDefined(); - const workers = await workerRecords(paseoHome); - expect(workers).toHaveLength(2); - const current = workers.find((record) => record.pid !== old.workerPid)!; - expect(current.protocolVersion).toBe(1); - const added = await manager.createTerminal({ - workspaceId: "ws-protocol", - cwd: old.cwd, - ...nodeTerminalCommand("setInterval(() => {}, 1000);"), - }); - trackTerminal(added); - const currentTerminals = ( - await readTerminalWorkerRecord(join(terminalWorkerDirectory(paseoHome), `${current.pid}.json`)) - ).terminals.map((entry) => entry.id); - expect(currentTerminals).toEqual([added.id]); - expect((await manager.getTerminals(old.cwd)).map((session) => session.id).sort()).toEqual( - [old.terminalId, added.id].sort(), - ); - - await manager.killTerminalAndWait(old.terminalId); - - await waitForCondition(() => !isProcessAlive(old.workerPid), 10000); - expect((await workerRecords(paseoHome)).map((record) => record.pid)).toEqual([current.pid]); - expect(manager.getTerminal(added.id)).toBeDefined(); -}); - -it("shuts down a worker two protocol versions old, closing its terminals", async () => { - const paseoHome = createTemporaryHome(); - const old = await leaveWorkerWithTerminal(paseoHome, -1); + const old = await leaveWorkerWithTerminal(paseoHome, 1); manager = await startRealWorkerManager({ paseoHome }); @@ -1576,18 +1570,18 @@ it("shuts down a worker two protocol versions old, closing its terminals", async expect(manager.getTerminal(old.terminalId)).toBeUndefined(); const workers = await workerRecords(paseoHome); expect(workers).toHaveLength(1); - expect(workers[0]!.protocolVersion).toBe(1); + expect(workers[0]!.protocolVersion).toBe(2); }); it("leaves a worker from a newer Paseo running, reports it incompatible, and closes it on a kill stop", async () => { const paseoHome = createTemporaryHome(); - const newer = await leaveWorkerWithTerminal(paseoHome, 2); + const newer = await leaveWorkerWithTerminal(paseoHome, 3); manager = await startRealWorkerManager({ paseoHome }); const unavailable = manager.getUnavailableError?.(); expect(unavailable).toBeInstanceOf(TerminalWorkerIncompatibleError); - expect(unavailable?.message).toContain("worker protocol 2"); + expect(unavailable?.message).toContain("worker protocol 3"); await expect(manager.getTerminals(newer.cwd)).rejects.toBeInstanceOf( TerminalWorkerIncompatibleError, ); @@ -1694,45 +1688,264 @@ it.skipIf(isPlatform("win32"))( }, ); -it("opens new terminals in a new worker once one of two draining workers exits", async () => { - const paseoHome = createTemporaryHome(); - const first = await launchDetachedWorkerWithTerminal(paseoHome, 0); - const second = await launchDetachedWorkerWithTerminal(paseoHome, 0); - - manager = await startRealWorkerManager({ paseoHome }); - - expect(await liveWorkerPids(paseoHome)).toEqual([first.workerPid, second.workerPid].sort()); - await expect( - manager.createTerminal({ workspaceId: "ws-detached", cwd: first.cwd }), - ).rejects.toBeInstanceOf(NoTerminalWorkerAvailableError); - - await manager.killTerminalAndWait(first.terminalId); - await waitForCondition(() => !isProcessAlive(first.workerPid), 10000); - - // Concurrent creates start one worker between them. - const created = await Promise.all( - [0, 1].map(() => - manager!.createTerminal({ - workspaceId: "ws-detached", - cwd: first.cwd, - ...nodeTerminalCommand("setInterval(() => {}, 1000);"), - }), - ), +it("replays bounded ANSI chunks before concurrent live output and reconstructs retained text", async () => { + const cwd = createTemporaryHome(); + manager = await startRealWorkerManager(); + const session = trackTerminal( + await manager.createTerminal({ + cwd, + workspaceId: "ws-replay", + rows: 8, + cols: 120, + ...nodeTerminalCommand(` + for (let i = 0; i < 800; i++) process.stdout.write("\\x1b[31m" + String(i).padStart(4, "0") + ":" + "x".repeat(100) + "\\x1b[0m\\r\\n"); + process.stdin.on("data", () => process.stdout.write("LIVE-AFTER-REPLAY\\r\\n")); + setInterval(() => {}, 1000); + `), + }), ); - created.forEach(trackTerminal); - const workers = await workerRecords(paseoHome); - expect(workers.map((record) => record.pid).sort()).toEqual(await liveWorkerPids(paseoHome)); - expect(workers).toHaveLength(2); - const started = workers.find((record) => record.pid !== second.workerPid)!; - expect(started.protocolVersion).toBe(1); - expect( - ( - await readTerminalWorkerRecord( - join(terminalWorkerDirectory(paseoHome), `${started.pid}.json`), - ) - ).terminals - .map((entry) => entry.id) - .sort(), - ).toEqual(created.map((session) => session.id).sort()); - expect(manager.getTerminal(second.terminalId)).toBeDefined(); + await waitForCondition( + async () => + (await manager!.captureTerminal(session.id)).lines.some((line) => line.startsWith("0799:")), + 10000, + ); + const expected = await manager.captureTerminal(session.id); + const events: string[] = []; + const chunks: Uint8Array[] = []; + let live = ""; + const unsubscribe = session.subscribe( + (event) => { + if (event.type === "output") { + events.push("live"); + live += event.data; + } + }, + { initialSnapshot: "ready" }, + ); + const metadata = await manager.replayTerminal(session.id, undefined, (event) => { + events.push(event.type); + if (event.type === "begin") session.send({ type: "input", data: "go\r" }); + if (event.type === "data") chunks.push(event.data); + }); + await waitForCondition(() => live.includes("LIVE-AFTER-REPLAY"), 10000); + unsubscribe(); + expect(events[0]).toBe("begin"); + expect(events.indexOf("live")).toBeGreaterThan(events.indexOf("end")); + expect(chunks.length).toBeGreaterThan(1); + expect(chunks.every((chunk) => chunk.byteLength <= 64 * 1024)).toBe(true); + expect(metadata?.rows).toBe(8); + expect(metadata?.cols).toBe(120); + const terminal = new xterm.Terminal({ + rows: 8, + cols: 120, + scrollback: 1000, + allowProposedApi: true, + }); + try { + await new Promise((resolve) => terminal.write(Buffer.concat(chunks), resolve)); + const buffer = terminal.buffer.active; + expect( + Array.from({ length: buffer.length }, (_, i) => buffer.getLine(i)!.translateToString(true)), + ).toEqual(expected.lines); + expect(buffer.getLine(0)!.getCell(0)!.getFgColor()).toBe(1); + } finally { + terminal.dispose(); + } }); + +it("completes a replay larger than the socket budget while a sibling stays usable", async () => { + const paseoHome = createTemporaryHome(); + const overflowProgress = join(paseoHome, "overflow-progress"); + manager = await startRealWorkerManager({ paseoHome }); + const terminal = trackTerminal( + await manager.createTerminal({ + cwd: paseoHome, + workspaceId: "ws-wide", + rows: 4, + cols: 1024, + ...nodeTerminalCommand(` + const row = Array.from({length: 1000}, (_, i) => "\\x1b[" + (31 + i % 2) + "mX").join(""); + for (let i = 0; i < 900; i++) process.stdout.write(row + "\\r\\n"); + process.stdout.write("READY-WIDE\\r\\n"); + process.stdin.on("data", data => { if (data.toString().includes("small")) { process.stdout.write("SAME-TERMINAL-LIVE\\r\\n"); return; } process.stdout.write("q".repeat(8 * 1024 * 1024) + "\\r\\nRECOVERED-LIVE\\r\\n", () => require("node:fs").writeFileSync(${JSON.stringify(overflowProgress)}, "done")); }); + setInterval(() => {}, 1000); + `), + }), + ); + const siblingProgress = join(paseoHome, "sibling-progress"); + const sibling = trackTerminal( + await manager.createTerminal({ + cwd: paseoHome, + workspaceId: "ws-wide", + ...nodeTerminalCommand(` + let count = 0; + process.stdin.on("data", () => { + count++; + process.stdout.write((count > 1 ? "z".repeat(1024 * 1024) : "") + "SIBLING-ALIVE\\r\\n", () => require("node:fs").writeFileSync(${JSON.stringify(siblingProgress)}, String(count))); + }); + setInterval(() => {}, 1000); + `), + }), + ); + await waitForCondition( + async () => + (await manager!.captureTerminal(terminal.id)).lines.some((line) => + line.includes("READY-WIDE"), + ), + 20000, + ); + const seen: string[] = []; + const unsubscribe = sibling.subscribe((event) => { + if (event.type === "output") seen.push(event.data); + }); + let bytes = 0; + const metadata = await manager.replayTerminal(terminal.id, undefined, (event) => { + if (event.type === "data") bytes += event.data.byteLength; + if (event.type === "begin") sibling.send({ type: "input", data: "go\r" }); + }); + expect(bytes).toBeGreaterThan(4 * 1024 * 1024); + expect(metadata?.cols).toBe(1024); + await waitForCondition(() => seen.join("").includes("SIBLING-ALIVE"), 10000); + unsubscribe(); + expect(manager.getTerminal(sibling.id)).toBe(sibling); + const truncated = await manager.replayTerminal(terminal.id, { scrollbackLines: 200 }, () => {}); + expect(metadata!.retainedHistoryLines).toBeGreaterThan(200); + expect(truncated!.retainedHistoryLines).toBe(metadata!.retainedHistoryLines); + const concurrent = await Promise.all( + [0, 1].map(() => manager!.replayTerminal(terminal.id, { scrollbackLines: 200 }, () => {})), + ); + expect(concurrent.map((replay) => replay!.retainedHistoryLines)).toEqual([ + metadata!.retainedHistoryLines, + metadata!.retainedHistoryLines, + ]); + await manager.detach(); + const [record] = await workerRecords(paseoHome); + const socket = createConnection(record!.socketPath); + let began = false; + let ended = false; + let completed = false; + let resumedBytes = 0; + let existingLiveOutput = ""; + const replayOrder: string[] = []; + const framed = attachFramedSocket(socket, (value) => { + const message = value as TerminalWorkerToParentMessage | { type: "helloAck" }; + if ( + message.type === "terminalMessage" && + message.terminalId === terminal.id && + message.message.type === "output" + ) { + existingLiveOutput += message.message.data; + replayOrder.push("live"); + } + if (message.type === "helloAck") { + framed.send({ type: "replayTerminal", terminalId: terminal.id, requestId: "stalled-replay" }); + } else if (message.type === "terminalReplay") { + if (message.event.type === "begin") { + began = true; + socket.pause(); + } + if (message.event.type === "data") resumedBytes += message.event.data.byteLength; + if (message.event.type === "end") { + ended = true; + replayOrder.push("end"); + } + } else if (message.type === "response" && message.requestId === "stalled-replay") + completed = message.ok; + }); + try { + framed.send({ type: "hello", secret: record!.secret, protocolVersion: 2 }); + await waitForCondition(() => began, 10000); + framed.send({ + type: "send", + terminalId: terminal.id, + requestId: "same-terminal-input", + message: { type: "input", data: "small\r" }, + }); + framed.send({ + type: "send", + terminalId: sibling.id, + requestId: "sibling-input", + message: { type: "input", data: "again\r" }, + }); + await waitForCondition( + () => existsSync(siblingProgress) && readFileSync(siblingProgress, "utf8") === "2", + 10000, + ); + await new Promise((resolve) => setTimeout(resolve, 150)); + expect(completed).toBe(false); + expect(socket.destroyed).toBe(false); + socket.resume(); + await waitForCondition(() => completed, 10000); + expect(ended).toBe(true); + expect(resumedBytes).toBe(bytes); + await waitForCondition(() => existingLiveOutput.includes("SAME-TERMINAL-LIVE"), 10000); + expect(existingLiveOutput.match(/SAME-TERMINAL-LIVE/g)).toHaveLength(1); + expect(replayOrder.indexOf("live")).toBeGreaterThan(replayOrder.indexOf("end")); + } finally { + socket.destroy(); + } + manager = await startRealWorkerManager({ paseoHome }); + expect( + (await manager.captureTerminal(terminal.id)).lines.some((line) => line.includes("READY-WIDE")), + ).toBe(true); + await manager.detach(); + const stalled = createConnection(record!.socketPath); + let overflowBegan = false; + let overflowError: string | null = null; + const existingViewerEvents: string[] = []; + const stalledFrames = attachFramedSocket(stalled, (value) => { + const message = value as TerminalWorkerToParentMessage | { type: "helloAck" }; + if (message.type === "terminalMessage" && message.terminalId === terminal.id) + existingViewerEvents.push(message.message.type); + if (message.type === "helloAck") + stalledFrames.send({ + type: "replayTerminal", + terminalId: terminal.id, + requestId: "overflow-replay", + }); + if (message.type === "terminalReplay" && message.event.type === "begin") { + overflowBegan = true; + stalled.pause(); + } + if (message.type === "response" && message.requestId === "overflow-replay" && !message.ok) + overflowError = message.error; + }); + try { + stalledFrames.send({ type: "hello", secret: record!.secret, protocolVersion: 2 }); + await waitForCondition(() => overflowBegan, 10000); + stalledFrames.send({ + type: "send", + terminalId: terminal.id, + requestId: "produce-overflow", + message: { type: "input", data: "burst\r" }, + }); + await waitForCondition(() => existsSync(overflowProgress), 10000); + expect(stalled.destroyed).toBe(false); + stalled.resume(); + await waitForCondition(() => overflowError !== null, 10000); + expect(overflowError).toBe("Terminal output exceeded the replay buffer; attach again"); + expect(existingViewerEvents).toContain("streamError"); + } finally { + stalled.destroy(); + } + manager = await startRealWorkerManager({ paseoHome }); + const recovered = await manager.replayTerminal(terminal.id, { scrollbackLines: 200 }, () => {}); + expect(recovered!.cols).toBe(1024); + expect( + (await manager.captureTerminal(terminal.id)).lines.some((line) => + line.includes("RECOVERED-LIVE"), + ), + ).toBe(true); + expect(manager.getTerminal(sibling.id)?.id).toBe(sibling.id); + expect(recovered!.retainedHistoryLines).toBe(1000); + const exitingReplay = await manager.replayTerminal( + terminal.id, + { scrollbackLines: 200 }, + (event) => { + if (event.type === "begin") manager!.killTerminal(terminal.id); + }, + ); + expect(exitingReplay!.cols).toBe(1024); + await waitForCondition(() => manager!.getTerminal(terminal.id) === undefined, 10000); + expect(manager.getTerminal(sibling.id)?.id).toBe(sibling.id); +}, 60000); diff --git a/packages/server/src/terminal/worker-terminal-manager.ts b/packages/server/src/terminal/worker-terminal-manager.ts index c9a6d7186e..9e1b8921a4 100644 --- a/packages/server/src/terminal/worker-terminal-manager.ts +++ b/packages/server/src/terminal/worker-terminal-manager.ts @@ -1,3 +1,4 @@ +import type { TerminalReplaySink, TerminalReplayMetadata } from "./terminal-replay.js"; import { randomBytes, randomUUID } from "node:crypto"; import { access } from "node:fs/promises"; import { setTimeout as delay } from "node:timers/promises"; @@ -5,7 +6,7 @@ import type { Logger } from "pino"; import { createExternalProcessEnv } from "../server/paseo-env.js"; import { assertAbsolutePath, isSameOrDescendantPath } from "../server/path-utils.js"; import { writtenDuringThisBoot } from "../server/pid-lock.js"; -import type { TerminalState } from "@getpaseo/protocol/messages"; +import type { WorkerTerminalState } from "./terminal-worker-protocol.js"; import type { TerminalActivity, TerminalActivityState } from "@getpaseo/protocol/terminal-activity"; import { deriveTerminalActivityStatusBucket } from "@getpaseo/protocol/terminal-activity"; import type { @@ -15,7 +16,6 @@ import type { TerminalCommandFinishedInfo, TerminalExitInfo, TerminalSession, - TerminalStateSnapshot, } from "./terminal.js"; import { resolvePaseoCliExecutablePath } from "./paseo-cli-path.js"; import type { CaptureTerminalLinesResult } from "./terminal-capture.js"; @@ -59,7 +59,6 @@ import type { TerminalWorkerResponse, TerminalWorkerToParentMessage, WorkerCreateTerminalOptions, - TerminalWorkerStateResult, WorkerTerminalInfo, } from "./terminal-worker-protocol.js"; @@ -93,7 +92,7 @@ interface PendingRequest { interface WorkerTerminalRecord { info: RequiredWorkerTerminalInfo; - state: TerminalState; + state: WorkerTerminalState; activity: TerminalActivity | null; // Cached input-mode preamble from the worker (the authoritative tracker lives // in the worker process). Updated with streamed output and snapshot responses. @@ -172,6 +171,7 @@ export function createWorkerTerminalManager( const requestTimeoutMs = managerOptions.requestTimeoutMs ?? REQUEST_TIMEOUT_MS; const resolveCreateEnvironment = managerOptions.resolveCreateEnvironment ?? resolveDaemonTerminalCreateEnvironment; + const replaySinks = new Map(); const pendingRequests = new Map(); const inFlight = new Set>(); const recordsById = new Map(); @@ -244,7 +244,7 @@ export function createWorkerTerminalManager( function registerRecord(input: { info: RequiredWorkerTerminalInfo; - state: TerminalState; + state: WorkerTerminalState; }): TerminalSession { const existing = recordsById.get(input.info.id); if (existing) { @@ -362,19 +362,8 @@ export function createWorkerTerminalManager( cols: record.state.cols, }; }, - getState(): TerminalState { - return record.state; - }, - getStateSnapshot(options?: { scrollbackLines?: number }): TerminalStateSnapshot { - const scrollbackLines = options?.scrollbackLines; - const scrollback = - typeof scrollbackLines === "number" - ? record.state.scrollback.slice(-scrollbackLines) - : record.state.scrollback; - return { - state: { ...record.state, scrollback }, - revision: 0, - }; + hasOutput(): boolean { + return record.state.hasOutput; }, getReplayPreamble(): string { return record.replayPreamble; @@ -442,9 +431,7 @@ export function createWorkerTerminalManager( if (!record) { return; } - if (message.message.type === "snapshot") { - record.state = message.message.state; - } + if (message.message.type === "output") record.state.hasOutput = true; record.replayPreamble = message.replayPreamble; for (const listener of Array.from(record.messageListeners)) { listener(message.message); @@ -611,6 +598,10 @@ export function createWorkerTerminalManager( function attachWorker(): void { worker.on("message", (message: TerminalWorkerToParentMessage) => { + if (message.type === "terminalReplay") { + replaySinks.get(message.requestId)?.(message.event); + return; + } if (isResponse(message)) { const pending = pendingRequests.get(message.requestId); if (!pending) { @@ -672,12 +663,17 @@ export function createWorkerTerminalManager( } attachWorker(); - function sendRequest(input: TerminalWorkerRequestInput): Promise { - if (reconnecting && workerDisconnected) return reconnecting.then(() => sendRequest(input)); + function sendRequest( + input: TerminalWorkerRequestInput, + sink?: TerminalReplaySink, + ): Promise { + if (reconnecting && workerDisconnected) + return reconnecting.then(() => sendRequest(input, sink)); if (workerDisconnected || !worker.connected) { return Promise.reject(new Error("Terminal worker is not running")); } const requestId = randomUUID(); + if (sink) replaySinks.set(requestId, sink); const message = { ...input, requestId } as TerminalWorkerRequest; const request = new Promise((resolve, reject) => { const timeout = setTimeout(() => { @@ -700,7 +696,10 @@ export function createWorkerTerminalManager( () => undefined, ); inFlight.add(settled); - void settled.finally(() => inFlight.delete(settled)); + void settled.finally(() => { + inFlight.delete(settled); + replaySinks.delete(requestId); + }); return request; } @@ -867,22 +866,14 @@ export function createWorkerTerminalManager( return recordsById.get(id)?.session; }, - async getTerminalState( - id: string, - options?: { scrollbackLines?: number }, - ): Promise { - 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; + async replayTerminal(id, options, sink): Promise { + const metadata = (await sendRequest( + { type: "replayTerminal", terminalId: id, options }, + sink, + )) as TerminalReplayMetadata | null; + const record = recordsById.get(id); + if (record && metadata) record.replayPreamble = metadata.replayPreamble; + return metadata; }, setTerminalTitle(id: string, title: string): boolean {