puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-10-30 13:13:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(server): stream terminal replay from the worker

Browse Source 
Move ANSI generation and replay/live ordering to the worker while preserving
existing ANSI clients through the daemon's single-frame restore adapter.
Keep 1,000 rows of retention and hard cut the unreleased cell transport.

Pace replay writes so a healthy large restore cannot fill the shared socket.
If the bounded live tail overflows, end every viewer of that terminal with
an explicit error; a fresh attach restores current state and the PTY survives.

Credit Manav Dhindsa (DrakenKor), #6139, and Lennart Schoch, #6743, for the
prior worker-side ANSI and scrollback work informing this change.
This commit is contained in:
Mohamed Boudra
2026-10-30 09:17:42 +01:00
parent a8e0effb32
commit 9ba63a30fa
21 changed files with 892 additions and 489 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/session.ts
-2
View File
@@ -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({
packages/server/src/server/worktree-bootstrap.test.ts
+2
View File
@@ -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,
packages/server/src/server/worktree-bootstrap.ts
+3 -14
View File
@@ -269,9 +269,9 @@ export function buildWorktreeSetupDetail(input: {
}
async function waitForTerminalBootstrapReadiness(
terminal: Pick<TerminalSession, "getState" | "subscribe">,
terminal: Pick<TerminalSession, "hasOutput" | "subscribe">,
): Promise<void> {
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<TerminalSession["getState"]>): 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;
packages/server/src/terminal/terminal-capture.ts
+2 -2
View File
@@ -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<LocalTerminalSession, "getState">,
options: CaptureTerminalLinesOptions = {},
): CaptureTerminalLinesResult {
const state = terminal.getState();
packages/server/src/terminal/terminal-manager.ts
+20 -13
View File
@@ -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<string, string> }): void;
validateTerminalActivityToken(terminalId: string, token: string): "valid" | "unknown" | "invalid";
getTerminal(id: string): TerminalSession | undefined;
getTerminalState(
replayTerminal(
id: string,
options?: TerminalStateSnapshotOptions,
): Promise<TerminalStateSnapshot | null>;
options: TerminalReplayOptions | undefined,
sink: TerminalReplaySink,
): Promise<TerminalReplayMetadata | null>;
setTerminalTitle(id: string, title: string): boolean;
setTerminalActivity(id: string, state: TerminalActivityState): Promise<boolean>;
clearTerminalAttention(id: string): Promise<boolean>;
@@ -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<TerminalManager["createTerminal"]>[0],
): Promise<LocalTerminalSession>;
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<string, string>;
paseoCliPath?: string | null;
}): Promise<TerminalSession> {
}): Promise<LocalTerminalSession> {
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<TerminalStateSnapshot | null> {
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 {
packages/server/src/terminal/terminal-replay.ts
+261
View File
@@ -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<TerminalReplayEvent, TerminalReplayMetadata> {
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<void>;
resolve: () => void;
reject: (error: unknown) => void;
exit?: Extract<TerminalWorkerToParentMessage, { type: "terminalExit" }>;
}
const replayTails = new Map<string, ReplayTail>();
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<TerminalWorkerRequest, { type: "replayTerminal" }>,
): Promise<void> {
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<void>((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<void> {
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<TerminalWorkerRequest, { type: "replayTerminal" }>,
): Promise<void> {
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(() => {});
}
},
};
}
packages/server/src/terminal/terminal-restore.test.ts
+1 -21
View File
@@ -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"),
}),
);
packages/server/src/terminal/terminal-restore.ts
+4 -20
View File
@@ -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,
});
}
packages/server/src/terminal/terminal-session-controller.test.ts
+116 -168
View File
@@ -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<ReplayFixture | null>,
): 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<T>(): {
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<TerminalStateSnapshot | null>();
const snapshot = deferred<ReplayFixture | null>();
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<void> {
await new Promise((resolve) => setTimeout(resolve, 30));
}
describe("terminal-session-controller wrap-flag gating", () => {
function setup(clientSupportsWrapReflow?: () => boolean): {
controller: ReturnType<typeof createController>;
getTerminalState: ReturnType<typeof vi.fn>;
} {
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<TerminalStateSnapshot>({ 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<typeof createController>): Promise<void> {
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<TerminalStateSnapshot>({ 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<TerminalStateSnapshot | null>, backedUp = false) {
function exitFixture(readReplay: () => Promise<ReplayFixture | null>, backedUp = false) {
const outputs = new Set<(message: ServerMessage) => void>();
const exits = new Set<Parameters<TerminalSession["onExit"]>[0]>();
const terminal: TerminalSession = {
@@ -934,8 +845,7 @@ function exitFixture(readSnapshot: () => Promise<TerminalStateSnapshot | null>,
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<TerminalStateSnapshot | null>,
};
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<TerminalStateSnapshot | null>();
const snapshot = deferred<ReplayFixture | null>();
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<TerminalStateSnapshot | null>();
const snapshot = deferred<ReplayFixture | null>();
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<TerminalStateSnapshot | null>();
const snapshot = deferred<ReplayFixture | null>();
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<TerminalStateSnapshot | null>();
const snapshot = deferred<ReplayFixture | null>();
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");
},
);
packages/server/src/terminal/terminal-session-controller.ts
+20 -41
View File
@@ -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<readonly TerminalWorkspaceRef[]>;
listTerminalWorkspaceRoots?: () => Promise<readonly string[]>;
// 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<readonly TerminalWorkspaceRef[]>;
private readonly listTerminalWorkspaceRoots: () => Promise<readonly string[]>;
private readonly clientSupportsWrapReflow: (source: object) => boolean;
private readonly getClientBufferedAmount: (source: object) => number | null;
private readonly subscribedDirectories = new Map<string, TerminalDirectorySubscription>();
@@ -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<SnapshotSendResult> {
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
packages/server/src/terminal/terminal-worker-client.test.ts
+43
View File
@@ -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<void>((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<import("node:net").Socket>((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<void>((resolve) => server.close(() => resolve()));
}
});
packages/server/src/terminal/terminal-worker-frames.ts
+12
View File
@@ -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<void>;
close(): void;
}
@@ -91,6 +92,17 @@ export function attachFramedSocket(
}
return socket.write(frame, (error) => callback?.(error ?? null));
},
sendAndWait(message) {
return new Promise<void>((resolve, reject) => {
this.send(message, (error) => {
if (error) {
reject(error);
return;
}
resolve();
});
});
},
close() {
socket.end(() => socket.destroy());
},
packages/server/src/terminal/terminal-worker-pool.ts
+3 -3
View File
@@ -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<TerminalStateSnapshot | null> {
return routed(id).getTerminalState(id, stateOptions);
async replayTerminal(id, replayOptions, sink) {
return routed(id).replayTerminal(id, replayOptions, sink);
},
setTerminalTitle(id, title) {
packages/server/src/terminal/terminal-worker-process.ts
+22 -42
View File
@@ -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<string, TerminalSession>();
const sessionsById = new Map<string, LocalTerminalSession>();
const unsubscribeByTerminalId = new Map<string, Array<() => void>>();
const outputCoalescerByTerminalId = new Map<string, TerminalOutputCoalescer>();
// 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<void> {
async function recordTerminalProgram(session: LocalTerminalSession): Promise<void> {
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();
}
});
packages/server/src/terminal/terminal-worker-protocol.test.ts
+3
View File
@@ -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"],
packages/server/src/terminal/terminal-worker-protocol.ts
+17 -18
View File
@@ -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<ServerMessage, { type: "snapshot" }>;
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;
packages/server/src/terminal/terminal.ts
+10 -2
View File
@@ -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,
packages/server/src/terminal/worker-terminal-manager.test.ts
+310 -97
View File
@@ -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<void>((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<void>((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);
packages/server/src/terminal/worker-terminal-manager.ts
+31 -40
View File
@@ -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<string, TerminalReplaySink>();
const pendingRequests = new Map<string, PendingRequest>();
const inFlight = new Set<Promise<void>>();
const recordsById = new Map<string, WorkerTerminalRecord>();
@@ -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<unknown> {
if (reconnecting && workerDisconnected) return reconnecting.then(() => sendRequest(input));
function sendRequest(
input: TerminalWorkerRequestInput,
sink?: TerminalReplaySink,
): Promise<unknown> {
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<TerminalStateSnapshot | null> {
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<TerminalReplayMetadata | null> {
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 {