diff --git a/packages/app/src/components/terminal-pane.tsx b/packages/app/src/components/terminal-pane.tsx index 6f411a13e..2f29959cc 100644 --- a/packages/app/src/components/terminal-pane.tsx +++ b/packages/app/src/components/terminal-pane.tsx @@ -216,7 +216,7 @@ export function TerminalPane({ if (terminalIdRef.current === exitedTerminalId) { emulatorRef.current?.clear(); } - streamControllerRef.current?.handleStreamExit({ + streamControllerRef.current?.handleTerminalExit({ terminalId: exitedTerminalId, }); setModifiers({ ...EMPTY_MODIFIERS }); diff --git a/packages/app/src/screens/workspace/workspace-screen.tsx b/packages/app/src/screens/workspace/workspace-screen.tsx index 0f67f66fc..d3c5e1c88 100644 --- a/packages/app/src/screens/workspace/workspace-screen.tsx +++ b/packages/app/src/screens/workspace/workspace-screen.tsx @@ -609,17 +609,10 @@ function WorkspaceScreenContent({ serverId, workspaceId }: WorkspaceScreenProps) })); }); - const unsubscribeStreamExit = client.on("terminal_stream_exit", (message) => { - if (message.type !== "terminal_stream_exit") { - return; - } - }); - client.subscribeTerminals({ cwd: normalizedWorkspaceId }); return () => { unsubscribeChanged(); - unsubscribeStreamExit(); client.unsubscribeTerminals({ cwd: normalizedWorkspaceId }); }; }, [client, isConnected, normalizedWorkspaceId, queryClient, terminalsQueryKey]); diff --git a/packages/app/src/terminal/runtime/terminal-stream-controller.test.ts b/packages/app/src/terminal/runtime/terminal-stream-controller.test.ts index 34543b97f..541481b6d 100644 --- a/packages/app/src/terminal/runtime/terminal-stream-controller.test.ts +++ b/packages/app/src/terminal/runtime/terminal-stream-controller.test.ts @@ -6,44 +6,32 @@ import { type TerminalStreamControllerStatus, } from "./terminal-stream-controller"; +type TerminalSnapshot = { + rows: number; + cols: number; + grid: Array>; + scrollback: Array>; + cursor: { row: number; col: number }; +}; + type TerminalStreamEvent = | { terminalId: string; type: "output"; data: Uint8Array } - | { - terminalId: string; - type: "snapshot"; - state: { - rows: number; - cols: number; - grid: Array>; - scrollback: Array>; - cursor: { row: number; col: number }; - }; - }; + | { terminalId: string; type: "snapshot"; state: TerminalSnapshot }; class FakeTerminalStreamClient implements TerminalStreamControllerClient { private readonly listeners = new Set<(event: TerminalStreamEvent) => void>(); public subscribeCalls: string[] = []; public unsubscribeCalls: string[] = []; public resizeCalls: Array<{ terminalId: string; rows: number; cols: number }> = []; - public nextSubscribeResponses: Array<{ - terminalId: string; - state: { - rows: number; - cols: number; - grid: Array>; - scrollback: Array>; - cursor: { row: number; col: number }; - } | null; - error?: string | null; - }> = []; + public nextSubscribeResults: Array<{ terminalId: string; error?: string | null }> = []; async subscribeTerminal(terminalId: string) { this.subscribeCalls.push(terminalId); - const response = this.nextSubscribeResponses.shift(); - if (!response) { - throw new Error("Missing fake subscribe response"); + const result = this.nextSubscribeResults.shift(); + if (!result) { + throw new Error("Missing fake subscribe result"); } - return response; + return result; } unsubscribeTerminal(terminalId: string): void { @@ -54,11 +42,7 @@ class FakeTerminalStreamClient implements TerminalStreamControllerClient { terminalId: string, message: { type: "resize"; rows: number; cols: number }, ): void { - this.resizeCalls.push({ - terminalId, - rows: message.rows, - cols: message.cols, - }); + this.resizeCalls.push({ terminalId, rows: message.rows, cols: message.cols }); } onTerminalStreamEvent(handler: (event: TerminalStreamEvent) => void): () => void { @@ -75,65 +59,40 @@ class FakeTerminalStreamClient implements TerminalStreamControllerClient { } } -function createControllerHarness(input?: { - client?: FakeTerminalStreamClient; -}): { - client: FakeTerminalStreamClient; - outputs: Array<{ terminalId: string; text: string }>; - snapshots: Array<{ terminalId: string; text: string }>; - statuses: TerminalStreamControllerStatus[]; - controller: TerminalStreamController; -} { +function createHarness(input?: { client?: FakeTerminalStreamClient }) { const client = input?.client ?? new FakeTerminalStreamClient(); const outputs: Array<{ terminalId: string; text: string }> = []; const snapshots: Array<{ terminalId: string; text: string }> = []; const statuses: TerminalStreamControllerStatus[] = []; - const controller = new TerminalStreamController({ client, getPreferredSize: () => ({ rows: 24, cols: 80 }), - onOutput: ({ terminalId, text }) => { - outputs.push({ terminalId, text }); + onOutput: (output) => { + outputs.push(output); }, onSnapshot: ({ terminalId, state }) => { snapshots.push({ terminalId, - text: state.grid - .map((row) => row.map((cell) => cell.char).join("")) - .join("\n"), + text: state.grid.map((row) => row.map((cell) => cell.char).join("")).join("\n"), }); }, onStatusChange: (status) => { statuses.push(status); }, - waitForDelay: async () => {}, }); - return { - client, - outputs, - snapshots, - statuses, - controller, - }; + return { client, controller, outputs, snapshots, statuses }; } async function flushAsyncWork(): Promise { await Promise.resolve(); - await new Promise((resolve) => { - setTimeout(() => resolve(), 0); - }); await Promise.resolve(); } describe("terminal-stream-controller", () => { - it("subscribes to a terminal, resizes it, and forwards snapshot/output events", async () => { - const harness = createControllerHarness(); - harness.client.nextSubscribeResponses.push({ - terminalId: "term-1", - state: null, - error: null, - }); + it("subscribes, resizes, and forwards snapshot/output events", async () => { + const harness = createHarness(); + harness.client.nextSubscribeResults.push({ terminalId: "term-1", error: null }); harness.controller.setTerminal({ terminalId: "term-1" }); await flushAsyncWork(); @@ -166,76 +125,50 @@ describe("terminal-stream-controller", () => { }); }); - it("retries retryable subscribe failures and then attaches", async () => { - const harness = createControllerHarness(); - harness.client.nextSubscribeResponses.push({ + it("surfaces subscribe failures without retrying", async () => { + const harness = createHarness(); + harness.client.nextSubscribeResults.push({ terminalId: "term-1", - state: null, error: "network disconnected", }); - harness.client.nextSubscribeResponses.push({ - terminalId: "term-1", - state: null, - error: null, - }); harness.controller.setTerminal({ terminalId: "term-1" }); await flushAsyncWork(); - expect(harness.client.subscribeCalls).toEqual(["term-1", "term-1"]); + expect(harness.client.subscribeCalls).toEqual(["term-1"]); expect(harness.statuses.at(-1)).toEqual({ terminalId: "term-1", isAttaching: false, - error: null, + error: "network disconnected", }); }); - it("reconnects to the selected terminal when the stream exits", async () => { - const harness = createControllerHarness(); - harness.client.nextSubscribeResponses.push({ - terminalId: "term-1", - state: null, - error: null, - }); - harness.client.nextSubscribeResponses.push({ - terminalId: "term-1", - state: null, - error: null, - }); + it("treats terminal exit as final and does not reconnect", async () => { + const harness = createHarness(); + harness.client.nextSubscribeResults.push({ terminalId: "term-1", error: null }); harness.controller.setTerminal({ terminalId: "term-1" }); await flushAsyncWork(); - - harness.controller.handleStreamExit({ terminalId: "term-1" }); + harness.controller.handleTerminalExit({ terminalId: "term-1" }); await flushAsyncWork(); - expect(harness.client.subscribeCalls).toEqual(["term-1", "term-1"]); + expect(harness.client.subscribeCalls).toEqual(["term-1"]); expect(harness.statuses.at(-1)).toEqual({ terminalId: "term-1", isAttaching: false, - error: null, + error: "Terminal exited", }); }); it("unsubscribes when switching terminals and on dispose", async () => { - const harness = createControllerHarness(); - harness.client.nextSubscribeResponses.push({ - terminalId: "term-1", - state: null, - error: null, - }); - harness.client.nextSubscribeResponses.push({ - terminalId: "term-2", - state: null, - error: null, - }); + const harness = createHarness(); + harness.client.nextSubscribeResults.push({ terminalId: "term-1", error: null }); + harness.client.nextSubscribeResults.push({ terminalId: "term-2", error: null }); harness.controller.setTerminal({ terminalId: "term-1" }); await flushAsyncWork(); - harness.controller.setTerminal({ terminalId: "term-2" }); await flushAsyncWork(); - harness.controller.dispose(); expect(harness.client.unsubscribeCalls).toEqual(["term-1", "term-2"]); diff --git a/packages/app/src/terminal/runtime/terminal-stream-controller.ts b/packages/app/src/terminal/runtime/terminal-stream-controller.ts index 7e2f9e025..b5978b64c 100644 --- a/packages/app/src/terminal/runtime/terminal-stream-controller.ts +++ b/packages/app/src/terminal/runtime/terminal-stream-controller.ts @@ -1,15 +1,8 @@ import type { TerminalState } from "@server/shared/messages"; -import { - getTerminalAttachRetryDelayMs, - isTerminalAttachRetryableError, - waitForDuration, - withPromiseTimeout, -} from "@/utils/terminal-attach"; export type TerminalStreamControllerClient = { subscribeTerminal: (terminalId: string) => Promise<{ terminalId: string; - state: TerminalState | null; error?: string | null; }>; unsubscribeTerminal: (terminalId: string) => void; @@ -43,227 +36,116 @@ export type TerminalStreamControllerOptions = { onOutput: (input: { terminalId: string; text: string }) => void; onSnapshot: (input: { terminalId: string; state: TerminalState }) => void; onStatusChange?: (status: TerminalStreamControllerStatus) => void; - maxAttachAttempts?: number; - attachTimeoutMs?: number; - reconnectErrorMessage?: string; - withTimeout?: (input: { - promise: Promise; - timeoutMs: number; - timeoutMessage: string; - }) => Promise; - waitForDelay?: (input: { durationMs: number }) => Promise; - isRetryableError?: (input: { message: string }) => boolean; - getRetryDelayMs?: (input: { attempt: number }) => number; }; -const DEFAULT_ATTACH_MAX_ATTEMPTS = 4; -const DEFAULT_ATTACH_TIMEOUT_MS = 12_000; -const DEFAULT_RECONNECT_ERROR_MESSAGE = "Terminal stream ended. Reconnecting…"; +const TERMINAL_EXITED_ERROR = "Terminal exited"; export class TerminalStreamController { - private readonly unsubscribeStreamEvents: () => void; private readonly decoder = new TextDecoder(); - private selectedTerminalId: string | null = null; - private attachGeneration = 0; - private isDisposed = false; + private readonly unsubscribeStreamEvents: () => void; + private terminalId: string | null = null; + private disposed = false; constructor(private readonly options: TerminalStreamControllerOptions) { this.unsubscribeStreamEvents = this.options.client.onTerminalStreamEvent((event) => { - if (this.isDisposed || event.terminalId !== this.selectedTerminalId) { + if (this.disposed || event.terminalId !== this.terminalId) { return; } if (event.type === "snapshot") { this.decoder.decode(); - this.options.onSnapshot({ - terminalId: event.terminalId, - state: event.state, - }); + this.options.onSnapshot({ terminalId: event.terminalId, state: event.state }); return; } - const text = this.decoder.decode(event.data, { stream: true }); - if (text.length === 0) { - return; + if (text.length > 0) { + this.options.onOutput({ terminalId: event.terminalId, text }); } - this.options.onOutput({ - terminalId: event.terminalId, - text, - }); }); } setTerminal(input: { terminalId: string | null }): void { - if (this.isDisposed) { + if (this.disposed || input.terminalId === this.terminalId) { return; } - const nextTerminalId = input.terminalId; - if (this.selectedTerminalId === nextTerminalId) { - return; - } - - const previousTerminalId = this.selectedTerminalId; - this.selectedTerminalId = nextTerminalId; - this.attachGeneration += 1; - const generation = this.attachGeneration; - + const previousTerminalId = this.terminalId; + this.terminalId = nextTerminalId; this.decoder.decode(); if (previousTerminalId) { this.options.client.unsubscribeTerminal(previousTerminalId); } - if (!nextTerminalId) { - this.updateStatus({ - terminalId: null, - isAttaching: false, - error: null, - }); + this.options.onStatusChange?.({ terminalId: null, isAttaching: false, error: null }); return; } - - this.updateStatus({ - terminalId: nextTerminalId, - isAttaching: true, - error: null, - }); - void this.attachTerminal({ - terminalId: nextTerminalId, - generation, - }); - } - - handleStreamExit(input: { terminalId: string }): void { - if (this.isDisposed || this.selectedTerminalId !== input.terminalId) { - return; - } - - this.attachGeneration += 1; - const generation = this.attachGeneration; - this.decoder.decode(); - this.updateStatus({ - terminalId: input.terminalId, - isAttaching: true, - error: this.options.reconnectErrorMessage ?? DEFAULT_RECONNECT_ERROR_MESSAGE, - }); - void this.attachTerminal({ - terminalId: input.terminalId, - generation, - }); - } - - dispose(): void { - if (this.isDisposed) { - return; - } - this.isDisposed = true; - this.attachGeneration += 1; - this.decoder.decode(); - const selectedTerminalId = this.selectedTerminalId; - this.selectedTerminalId = null; - if (selectedTerminalId) { - this.options.client.unsubscribeTerminal(selectedTerminalId); - } - this.unsubscribeStreamEvents(); - this.updateStatus({ - terminalId: null, - isAttaching: false, - error: null, - }); - } - - private async attachTerminal(input: { terminalId: string; generation: number }): Promise { - const { - maxAttachAttempts = DEFAULT_ATTACH_MAX_ATTEMPTS, - attachTimeoutMs = DEFAULT_ATTACH_TIMEOUT_MS, - withTimeout = withPromiseTimeout, - waitForDelay = waitForDuration, - isRetryableError = isTerminalAttachRetryableError, - getRetryDelayMs = getTerminalAttachRetryDelayMs, - } = this.options; - - let lastErrorMessage = "Unable to subscribe to terminal"; - - for (let attempt = 0; attempt < maxAttachAttempts; attempt += 1) { - if (!this.isAttachGenerationCurrent(input)) { - return; - } - - try { - const payload = await withTimeout({ - promise: this.options.client.subscribeTerminal(input.terminalId), - timeoutMs: attachTimeoutMs, - timeoutMessage: "Timed out subscribing to terminal", - }); - - if (!this.isAttachGenerationCurrent(input)) { - this.options.client.unsubscribeTerminal(input.terminalId); + this.options.onStatusChange?.({ terminalId: nextTerminalId, isAttaching: true, error: null }); + void this.options.client + .subscribeTerminal(nextTerminalId) + .then((payload) => { + if (this.disposed || this.terminalId !== nextTerminalId) { return; } - if (payload.error) { - lastErrorMessage = payload.error; - const hasRemainingAttempts = attempt < maxAttachAttempts - 1; - if (hasRemainingAttempts && isRetryableError({ message: lastErrorMessage })) { - await waitForDelay({ durationMs: getRetryDelayMs({ attempt }) }); - continue; - } - - this.updateStatus({ - terminalId: input.terminalId, + this.terminalId = null; + this.options.onStatusChange?.({ + terminalId: nextTerminalId, isAttaching: false, - error: lastErrorMessage, + error: payload.error, }); return; } - const preferredSize = this.options.getPreferredSize(); if (preferredSize) { - this.options.client.sendTerminalInput(input.terminalId, { + this.options.client.sendTerminalInput(nextTerminalId, { type: "resize", rows: preferredSize.rows, cols: preferredSize.cols, }); } - - this.updateStatus({ - terminalId: input.terminalId, + this.options.onStatusChange?.({ + terminalId: nextTerminalId, isAttaching: false, error: null, }); - return; - } catch (error) { - lastErrorMessage = - error instanceof Error ? error.message : "Unable to subscribe to terminal"; - const hasRemainingAttempts = attempt < maxAttachAttempts - 1; - if (hasRemainingAttempts && isRetryableError({ message: lastErrorMessage })) { - await waitForDelay({ durationMs: getRetryDelayMs({ attempt }) }); - continue; + }) + .catch((error: unknown) => { + if (this.disposed || this.terminalId !== nextTerminalId) { + return; } - - this.updateStatus({ - terminalId: input.terminalId, + this.terminalId = null; + this.options.onStatusChange?.({ + terminalId: nextTerminalId, isAttaching: false, - error: lastErrorMessage, + error: error instanceof Error ? error.message : "Unable to subscribe to terminal", }); - return; - } - } + }); + } - this.updateStatus({ + handleTerminalExit(input: { terminalId: string }): void { + if (this.disposed || input.terminalId !== this.terminalId) { + return; + } + this.decoder.decode(); + this.terminalId = null; + this.options.onStatusChange?.({ terminalId: input.terminalId, isAttaching: false, - error: lastErrorMessage, + error: TERMINAL_EXITED_ERROR, }); } - private isAttachGenerationCurrent(input: { terminalId: string; generation: number }): boolean { - if (this.isDisposed) { - return false; + dispose(): void { + if (this.disposed) { + return; } - return this.attachGeneration === input.generation && this.selectedTerminalId === input.terminalId; - } - - private updateStatus(status: TerminalStreamControllerStatus): void { - this.options.onStatusChange?.(status); + this.disposed = true; + this.decoder.decode(); + const terminalId = this.terminalId; + this.terminalId = null; + if (terminalId) { + this.options.client.unsubscribeTerminal(terminalId); + } + this.unsubscribeStreamEvents(); + this.options.onStatusChange?.({ terminalId: null, isAttaching: false, error: null }); } } diff --git a/packages/app/src/utils/terminal-attach.test.ts b/packages/app/src/utils/terminal-attach.test.ts deleted file mode 100644 index ecdbff4c7..000000000 --- a/packages/app/src/utils/terminal-attach.test.ts +++ /dev/null @@ -1,50 +0,0 @@ -import { describe, expect, it } from "vitest"; - -import { - getTerminalAttachRetryDelayMs, - isTerminalAttachRetryableError, - withPromiseTimeout, -} from "./terminal-attach"; - -describe("terminal-attach", () => { - it("computes bounded exponential retry delays", () => { - expect(getTerminalAttachRetryDelayMs({ attempt: 0 })).toBe(250); - expect(getTerminalAttachRetryDelayMs({ attempt: 1 })).toBe(500); - expect(getTerminalAttachRetryDelayMs({ attempt: 2 })).toBe(1_000); - expect(getTerminalAttachRetryDelayMs({ attempt: 3 })).toBe(2_000); - expect(getTerminalAttachRetryDelayMs({ attempt: 8 })).toBe(2_000); - }); - - it("matches retryable attach errors", () => { - expect(isTerminalAttachRetryableError({ message: "Terminal not found while attaching" })).toBe( - true, - ); - expect(isTerminalAttachRetryableError({ message: "Network disconnected during attach" })).toBe( - true, - ); - expect(isTerminalAttachRetryableError({ message: "stream ended before snapshot" })).toBe( - true, - ); - expect(isTerminalAttachRetryableError({ message: "permission denied" })).toBe(false); - }); - - it("resolves before timeout when promise completes", async () => { - await expect( - withPromiseTimeout({ - promise: Promise.resolve("ok"), - timeoutMs: 50, - timeoutMessage: "timed out", - }), - ).resolves.toBe("ok"); - }); - - it("rejects when timeout wins", async () => { - await expect( - withPromiseTimeout({ - promise: new Promise(() => {}), - timeoutMs: 10, - timeoutMessage: "timed out", - }), - ).rejects.toThrow("timed out"); - }); -}); diff --git a/packages/app/src/utils/terminal-attach.ts b/packages/app/src/utils/terminal-attach.ts deleted file mode 100644 index 698aa0a30..000000000 --- a/packages/app/src/utils/terminal-attach.ts +++ /dev/null @@ -1,49 +0,0 @@ -const TERMINAL_ATTACH_RETRYABLE_ERROR_PATTERNS = [ - "terminal not found", - "timed out", - "timeout", - "connection", - "network", - "disconnected", - "stream ended", -] as const; - -export function getTerminalAttachRetryDelayMs(input: { attempt: number }): number { - const clampedAttempt = Math.max(0, input.attempt); - const exponentialDelay = 250 * 2 ** clampedAttempt; - return Math.min(2_000, exponentialDelay); -} - -export function isTerminalAttachRetryableError(input: { message: string }): boolean { - const normalized = input.message.toLowerCase(); - return TERMINAL_ATTACH_RETRYABLE_ERROR_PATTERNS.some((pattern) => normalized.includes(pattern)); -} - -export async function waitForDuration(input: { durationMs: number }): Promise { - await new Promise((resolve) => { - setTimeout(resolve, Math.max(0, input.durationMs)); - }); -} - -export async function withPromiseTimeout(input: { - promise: Promise; - timeoutMs: number; - timeoutMessage: string; -}): Promise { - let timeoutHandle: ReturnType | null = null; - const timeoutPromise = new Promise((_, reject) => { - timeoutHandle = setTimeout( - () => { - reject(new Error(input.timeoutMessage)); - }, - Math.max(0, input.timeoutMs), - ); - }); - try { - return await Promise.race([input.promise, timeoutPromise]); - } finally { - if (timeoutHandle) { - clearTimeout(timeoutHandle); - } - } -} diff --git a/packages/server/src/client/daemon-client.test.ts b/packages/server/src/client/daemon-client.test.ts index 14399fc26..fa97e7461 100644 --- a/packages/server/src/client/daemon-client.test.ts +++ b/packages/server/src/client/daemon-client.test.ts @@ -1257,7 +1257,6 @@ describe("DaemonClient", () => { type: "subscribe_terminal_response", payload: { terminalId: "term-1", - state: null, error: null, requestId: "sub-1", }, @@ -1308,7 +1307,6 @@ describe("DaemonClient", () => { type: "subscribe_terminal_response", payload: { terminalId: "term-1", - state: null, error: null, requestId: "sub-2", }, @@ -1356,7 +1354,6 @@ describe("DaemonClient", () => { type: "subscribe_terminal_response", payload: { terminalId: "term-1", - state: null, error: null, requestId: "sub-3", }, @@ -1418,7 +1415,6 @@ describe("DaemonClient", () => { type: "subscribe_terminal_response", payload: { terminalId: "term-1", - state: null, error: null, requestId: "sub-4", }, diff --git a/packages/server/src/server/daemon-e2e/terminal.e2e.test.ts b/packages/server/src/server/daemon-e2e/terminal.e2e.test.ts index 7961528de..9f259f09c 100644 --- a/packages/server/src/server/daemon-e2e/terminal.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/terminal.e2e.test.ts @@ -4,6 +4,11 @@ import { tmpdir } from "os"; import path from "path"; import WebSocket from "ws"; import { DaemonClient } from "../../client/daemon-client.js"; +import { + WSOutboundMessageSchema, + type TerminalState, + type WSOutboundMessage, +} from "../../shared/messages.js"; import { decodeTerminalStreamFrame, TerminalStreamOpcode, @@ -11,6 +16,8 @@ import { } from "../../shared/terminal-stream-protocol.js"; import { createDaemonTestContext, type DaemonTestContext } from "../test-utils/index.js"; +type RawSessionEnvelope = Extract; + function tmpCwd(): string { return mkdtempSync(path.join(tmpdir(), "daemon-terminal-e2e-")); } @@ -24,10 +31,7 @@ function createLogger() { }; } -function extractStateText(state: { - grid: Array>; - scrollback: Array>; -}): string { +function extractStateText(state: Pick): string { return [...state.scrollback, ...state.grid] .map((row) => row @@ -57,21 +61,9 @@ async function waitForCondition( async function waitForTerminalSnapshot( client: DaemonClient, terminalId: string, - predicate: (state: { - rows: number; - cols: number; - grid: Array>; - scrollback: Array>; - cursor: { row: number; col: number }; - }) => boolean, + predicate: (state: TerminalState) => boolean, timeout = 10000, -): Promise<{ - rows: number; - cols: number; - grid: Array>; - scrollback: Array>; - cursor: { row: number; col: number }; -}> { +): Promise { return new Promise((resolve, reject) => { const timeoutHandle = setTimeout(() => { unsubscribe(); @@ -156,9 +148,7 @@ async function connectRawWebSocket(port: number): Promise { const helloReady = waitForRawSessionMessage( ws, (message) => - message.type === "session" && - message.message?.type === "status" && - message.message.payload?.status === "server_info", + message.message.type === "status" && message.message.payload.status === "server_info", 10000, ); @@ -198,15 +188,9 @@ async function closeWebSocket(ws: WebSocket, timeout = 5000): Promise { async function waitForRawSessionMessage( ws: WebSocket, - predicate: (message: { - type?: string; - message?: { type?: string; payload?: Record }; - }) => boolean, + predicate: (message: RawSessionEnvelope) => boolean, timeout = 10000, -): Promise<{ - type?: string; - message?: { type?: string; payload?: Record }; -}> { +): Promise { return new Promise((resolve, reject) => { const timeoutHandle = setTimeout(() => { cleanup(); @@ -220,10 +204,11 @@ async function waitForRawSessionMessage( return; } try { - const parsed = JSON.parse(text) as { - type?: string; - message?: { type?: string; payload?: Record }; - }; + const parsedResult = WSOutboundMessageSchema.safeParse(JSON.parse(text)); + if (!parsedResult.success || parsedResult.data.type !== "session") { + return; + } + const parsed = parsedResult.data; if (!predicate(parsed)) { return; } @@ -294,9 +279,8 @@ async function subscribeRawTerminal(ws: WebSocket, terminalId: string, requestId const ready = waitForRawSessionMessage( ws, (message) => - message.type === "session" && - message.message?.type === "subscribe_terminal_response" && - message.message.payload?.requestId === requestId, + message.message.type === "subscribe_terminal_response" && + message.message.payload.requestId === requestId, 10000, ); @@ -311,7 +295,11 @@ async function subscribeRawTerminal(ws: WebSocket, terminalId: string, requestId }), ); - await ready; + const message = await ready; + if (message.message.type !== "subscribe_terminal_response") { + throw new Error("Expected subscribe_terminal_response"); + } + expect(message.message.payload).not.toHaveProperty("state"); } describe("daemon E2E terminal", () => { diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index aecdbba87..a5db8da24 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -279,8 +279,7 @@ type ActiveTerminalStream = { terminalId: string; unsubscribe: () => void; needsSnapshot: boolean; - primed: boolean; - snapshotTimer: ReturnType | null; + snapshotRetryTimer: ReturnType | null; }; export type SessionRuntimeMetrics = { @@ -1817,8 +1816,8 @@ export class Session { if (!resize) { return; } + activeStream.needsSnapshot = true; terminal.send({ type: "resize", rows: resize.rows, cols: resize.cols }); - this.queueTerminalSnapshot(activeStream, terminal); return; } @@ -7666,7 +7665,6 @@ export class Session { type: "subscribe_terminal_response", payload: { terminalId: msg.terminalId, - state: null, error: "Terminal manager not available", requestId: msg.requestId, }, @@ -7680,7 +7678,6 @@ export class Session { type: "subscribe_terminal_response", payload: { terminalId: msg.terminalId, - state: null, error: "Terminal not found", requestId: msg.requestId, }, @@ -7689,23 +7686,16 @@ export class Session { } this.ensureTerminalExitSubscription(session); - const activeStream = this.bindActiveTerminalStream(session); - - // Send initial state this.emit({ type: "subscribe_terminal_response", payload: { terminalId: msg.terminalId, - state: session.getState(), error: null, requestId: msg.requestId, }, }); - if (activeStream) { - this.sendTerminalSnapshot(activeStream, session); - activeStream.primed = true; - } + this.bindActiveTerminalStream(session); } private handleUnsubscribeTerminalRequest(msg: UnsubscribeTerminalRequest): void { @@ -7803,126 +7793,77 @@ export class Session { const activeStream: ActiveTerminalStream = { terminalId: terminal.id, unsubscribe: () => {}, - needsSnapshot: false, - primed: false, - snapshotTimer: null, + needsSnapshot: true, + snapshotRetryTimer: null, }; + + const trySendSnapshot = () => { + if (this.activeTerminalStream !== activeStream || !activeStream.needsSnapshot) { + return; + } + if (this.getCurrentBinaryBufferedAmount() > TERMINAL_STREAM_LOW_WATER_BYTES) { + if (!activeStream.snapshotRetryTimer) { + activeStream.snapshotRetryTimer = setTimeout(() => { + activeStream.snapshotRetryTimer = null; + trySendSnapshot(); + }, 33); + } + return; + } + if (activeStream.snapshotRetryTimer) { + clearTimeout(activeStream.snapshotRetryTimer); + activeStream.snapshotRetryTimer = null; + } + activeStream.needsSnapshot = false; + this.emitBinary( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Snapshot, + payload: encodeTerminalSnapshotPayload(terminal.getState()), + }), + ); + }; + activeStream.unsubscribe = terminal.subscribe((message) => { if (this.activeTerminalStream !== activeStream) { return; } - if (message.type === "output") { - this.handleActiveTerminalOutput(activeStream, terminal, message.data); + if (message.type === "snapshot") { + trySendSnapshot(); return; } - if (message.type === "snapshot") { - this.maybeSendQueuedTerminalSnapshot(activeStream, terminal); + if (activeStream.needsSnapshot || message.data.length === 0) { + return; + } + if (this.getCurrentBinaryBufferedAmount() >= TERMINAL_STREAM_HIGH_WATER_BYTES) { + activeStream.needsSnapshot = true; + trySendSnapshot(); + return; + } + this.emitBinary( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Output, + payload: new Uint8Array(Buffer.from(message.data, "utf8")), + }), + ); + if (this.getCurrentBinaryBufferedAmount() >= TERMINAL_STREAM_HIGH_WATER_BYTES) { + activeStream.needsSnapshot = true; + trySendSnapshot(); } }); this.activeTerminalStream = activeStream; return activeStream; } - private handleActiveTerminalOutput( - activeStream: ActiveTerminalStream, - terminal: TerminalSession, - data: string, - ): void { - if (!activeStream.primed || data.length === 0) { - return; - } - if (activeStream.needsSnapshot) { - this.maybeSendQueuedTerminalSnapshot(activeStream, terminal); - return; - } - if (this.getCurrentBinaryBufferedAmount() > TERMINAL_STREAM_HIGH_WATER_BYTES) { - activeStream.needsSnapshot = true; - this.scheduleTerminalSnapshotRetry(activeStream, terminal); - return; - } - this.emitBinary( - encodeTerminalStreamFrame({ - opcode: TerminalStreamOpcode.Output, - payload: new Uint8Array(Buffer.from(data, "utf8")), - }), - ); - if (this.getCurrentBinaryBufferedAmount() > TERMINAL_STREAM_HIGH_WATER_BYTES) { - activeStream.needsSnapshot = true; - this.scheduleTerminalSnapshotRetry(activeStream, terminal); - } - } - - private queueTerminalSnapshot( - activeStream: ActiveTerminalStream, - terminal: TerminalSession, - ): void { - if (this.getCurrentBinaryBufferedAmount() > TERMINAL_STREAM_HIGH_WATER_BYTES) { - activeStream.needsSnapshot = true; - this.scheduleTerminalSnapshotRetry(activeStream, terminal); - return; - } - this.sendTerminalSnapshot(activeStream, terminal); - } - - private maybeSendQueuedTerminalSnapshot( - activeStream: ActiveTerminalStream, - terminal: TerminalSession, - ): void { - if (!activeStream.needsSnapshot) { - return; - } - if (this.getCurrentBinaryBufferedAmount() >= TERMINAL_STREAM_LOW_WATER_BYTES) { - this.scheduleTerminalSnapshotRetry(activeStream, terminal); - return; - } - this.sendTerminalSnapshot(activeStream, terminal); - } - - private sendTerminalSnapshot( - activeStream: ActiveTerminalStream, - terminal: TerminalSession, - ): void { - this.clearTerminalSnapshotRetry(activeStream); - activeStream.needsSnapshot = false; - this.emitBinary( - encodeTerminalStreamFrame({ - opcode: TerminalStreamOpcode.Snapshot, - payload: encodeTerminalSnapshotPayload(terminal.getState()), - }), - ); - } - - private scheduleTerminalSnapshotRetry( - activeStream: ActiveTerminalStream, - terminal: TerminalSession, - ): void { - if (activeStream.snapshotTimer) { - return; - } - activeStream.snapshotTimer = setTimeout(() => { - activeStream.snapshotTimer = null; - if (this.activeTerminalStream !== activeStream || !activeStream.needsSnapshot) { - return; - } - this.maybeSendQueuedTerminalSnapshot(activeStream, terminal); - }, 33); - } - - private clearTerminalSnapshotRetry(activeStream: ActiveTerminalStream): void { - if (!activeStream.snapshotTimer) { - return; - } - clearTimeout(activeStream.snapshotTimer); - activeStream.snapshotTimer = null; - } - private detachActiveTerminalStream(options?: { emitExit: boolean }): boolean { const activeStream = this.activeTerminalStream; if (!activeStream) { return false; } this.activeTerminalStream = null; - this.clearTerminalSnapshotRetry(activeStream); + if (activeStream.snapshotRetryTimer) { + clearTimeout(activeStream.snapshotRetryTimer); + activeStream.snapshotRetryTimer = null; + } try { activeStream.unsubscribe(); } catch (error) { diff --git a/packages/server/src/shared/messages.ts b/packages/server/src/shared/messages.ts index 39a248f10..f0c0d3874 100644 --- a/packages/server/src/shared/messages.ts +++ b/packages/server/src/shared/messages.ts @@ -2104,15 +2104,15 @@ export const TerminalCellSchema = z.object({ bold: z.boolean().optional(), italic: z.boolean().optional(), underline: z.boolean().optional(), -}); +}).strict(); export const TerminalStateSchema = z.object({ rows: z.number(), cols: z.number(), grid: z.array(z.array(TerminalCellSchema)), scrollback: z.array(z.array(TerminalCellSchema)), - cursor: z.object({ row: z.number(), col: z.number() }), -}); + cursor: z.object({ row: z.number(), col: z.number() }).strict(), +}).strict(); export const ListTerminalsResponseSchema = z.object({ type: z.literal("list_terminals_response"), @@ -2144,7 +2144,6 @@ export const SubscribeTerminalResponseSchema = z.object({ type: z.literal("subscribe_terminal_response"), payload: z.object({ terminalId: z.string(), - state: TerminalStateSchema.nullable(), error: z.string().nullable(), requestId: z.string(), }), diff --git a/packages/server/src/shared/terminal-stream-protocol.test.ts b/packages/server/src/shared/terminal-stream-protocol.test.ts index 89db47595..a391eed72 100644 --- a/packages/server/src/shared/terminal-stream-protocol.test.ts +++ b/packages/server/src/shared/terminal-stream-protocol.test.ts @@ -56,4 +56,50 @@ describe("terminal stream protocol", () => { it("rejects unknown opcodes", () => { expect(decodeTerminalStreamFrame(new Uint8Array([0xff, 0x01]))).toBeNull(); }); + + it("rejects malformed JSON payloads", () => { + const malformed = new TextEncoder().encode("{"); + + expect(decodeTerminalResizePayload(malformed)).toBeNull(); + expect(decodeTerminalSnapshotPayload(malformed)).toBeNull(); + }); + + it("rejects invalid resize and snapshot shapes", () => { + expect( + decodeTerminalResizePayload(new TextEncoder().encode(JSON.stringify({ rows: "24", cols: 80 }))), + ).toBeNull(); + expect( + decodeTerminalSnapshotPayload( + new TextEncoder().encode( + JSON.stringify({ + rows: 1, + cols: 1, + grid: [[{ char: "A" }]], + scrollback: [], + }), + ), + ), + ).toBeNull(); + }); + + it("rejects unknown fields in resize and snapshot payloads", () => { + expect( + decodeTerminalResizePayload( + new TextEncoder().encode(JSON.stringify({ rows: 24, cols: 80, extra: true })), + ), + ).toBeNull(); + expect( + decodeTerminalSnapshotPayload( + new TextEncoder().encode( + JSON.stringify({ + rows: 1, + cols: 1, + grid: [[{ char: "A", extra: true }]], + scrollback: [], + cursor: { row: 0, col: 1 }, + }), + ), + ), + ).toBeNull(); + }); }); diff --git a/packages/server/src/shared/terminal-stream-protocol.ts b/packages/server/src/shared/terminal-stream-protocol.ts index 2c5c3ec51..f27a55621 100644 --- a/packages/server/src/shared/terminal-stream-protocol.ts +++ b/packages/server/src/shared/terminal-stream-protocol.ts @@ -4,7 +4,7 @@ import { TerminalStateSchema } from "./messages.js"; export const TerminalStreamResizeSchema = z.object({ rows: z.number().int().positive(), cols: z.number().int().positive(), -}); +}).strict(); export const TerminalStreamOpcode = { Output: 0x01, @@ -115,6 +115,10 @@ function encodeJsonPayload(value: unknown): Uint8Array { } function decodeJsonPayload(bytes: Uint8Array): unknown { - const text = new TextDecoder().decode(bytes); - return JSON.parse(text); + try { + const text = new TextDecoder().decode(bytes); + return JSON.parse(text); + } catch { + return null; + } } diff --git a/packages/server/src/terminal/terminal.test.ts b/packages/server/src/terminal/terminal.test.ts index 04926991a..ed3adf92a 100644 --- a/packages/server/src/terminal/terminal.test.ts +++ b/packages/server/src/terminal/terminal.test.ts @@ -438,6 +438,31 @@ describe("Terminal", () => { unsubscribe(); }); + it("emits output only after getState reflects the new data", async () => { + const session = trackSession( + await createTerminal({ + cwd: "/tmp", + shell: "/bin/sh", + env: { PS1: "$ " }, + }), + ); + + await waitForLines(session, ["$"]); + const outputSeenInState = new Promise((resolve) => { + const unsubscribe = session.subscribe((message) => { + if (message.type !== "output" || !message.data.includes("state-after-output")) { + return; + } + unsubscribe(); + const stateText = getLines(session.getState()).join("\n"); + resolve(stateText.includes("state-after-output")); + }); + }); + + session.send({ type: "input", data: "echo state-after-output\r" }); + expect(await outputSeenInState).toBe(true); + }); + it("unsubscribe stops receiving messages", async () => { const session = trackSession( await createTerminal({ diff --git a/packages/server/src/terminal/terminal.ts b/packages/server/src/terminal/terminal.ts index aa5cfaf53..5902aff5f 100644 --- a/packages/server/src/terminal/terminal.ts +++ b/packages/server/src/terminal/terminal.ts @@ -322,10 +322,13 @@ export async function createTerminal(options: CreateTerminalOptions): Promise { if (killed) return; - for (const listener of listeners) { - listener({ type: "output", data }); - } terminal.write(data, () => { + if (disposed || killed) { + return; + } + for (const listener of listeners) { + listener({ type: "output", data }); + } scheduleStateBroadcast(); }); }); @@ -359,6 +362,7 @@ export async function createTerminal(options: CreateTerminalOptions): Promise void): () => void { listeners.add(listener); - queueMicrotask(() => { - if (listeners.has(listener)) { + terminal.write("", () => { + if (!disposed && listeners.has(listener)) { listener({ type: "snapshot", state: getState() }); } });