From ea41dbc8e78ceba3a03907f5b83077f4112f1050 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 23 Mar 2026 12:36:35 +0700 Subject: [PATCH 1/2] Add multiplexed terminal stream slots --- .../server/src/client/daemon-client.test.ts | 103 +++++++++- packages/server/src/client/daemon-client.ts | 55 +++-- .../server/daemon-e2e/terminal.e2e.test.ts | 122 +++++++---- packages/server/src/server/session.ts | 191 ++++++++++++------ .../websocket-server.relay-reconnect.test.ts | 6 +- packages/server/src/shared/messages.ts | 61 +++--- .../shared/terminal-stream-protocol.test.ts | 17 +- .../src/shared/terminal-stream-protocol.ts | 29 ++- 8 files changed, 430 insertions(+), 154 deletions(-) diff --git a/packages/server/src/client/daemon-client.test.ts b/packages/server/src/client/daemon-client.test.ts index fa97e7461..eed79a7e0 100644 --- a/packages/server/src/client/daemon-client.test.ts +++ b/packages/server/src/client/daemon-client.test.ts @@ -1257,6 +1257,7 @@ describe("DaemonClient", () => { type: "subscribe_terminal_response", payload: { terminalId: "term-1", + slot: 11, error: null, requestId: "sub-1", }, @@ -1267,6 +1268,7 @@ describe("DaemonClient", () => { mock.triggerMessage( encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Output, + slot: 11, payload: new TextEncoder().encode("hello"), }), ); @@ -1276,7 +1278,7 @@ describe("DaemonClient", () => { unsubscribe(); }); - test("emits snapshot events for the active terminal stream", async () => { + test("emits snapshot events for the subscribed terminal stream", async () => { const logger = createMockLogger(); const mock = createMockTransport(); @@ -1307,6 +1309,7 @@ describe("DaemonClient", () => { type: "subscribe_terminal_response", payload: { terminalId: "term-1", + slot: 12, error: null, requestId: "sub-2", }, @@ -1324,6 +1327,7 @@ describe("DaemonClient", () => { mock.triggerMessage( encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Snapshot, + slot: 12, payload: encodeTerminalSnapshotPayload(state), }), ); @@ -1331,7 +1335,7 @@ describe("DaemonClient", () => { expect(snapshots).toEqual([state]); }); - test("sends input and resize frames for the active terminal", async () => { + test("sends input and resize frames for the subscribed terminal slot", async () => { const logger = createMockLogger(); const mock = createMockTransport(); @@ -1354,6 +1358,7 @@ describe("DaemonClient", () => { type: "subscribe_terminal_response", payload: { terminalId: "term-1", + slot: 13, error: null, requestId: "sub-3", }, @@ -1376,14 +1381,105 @@ describe("DaemonClient", () => { const resizeFrame = decodeTerminalStreamFrame(asUint8Array(mock.sent[1])!); expect(inputFrame?.opcode).toBe(TerminalStreamOpcode.Input); + expect(inputFrame?.slot).toBe(13); expect(new TextDecoder().decode(inputFrame?.payload ?? new Uint8Array())).toBe("echo hello\r"); expect(resizeFrame?.opcode).toBe(TerminalStreamOpcode.Resize); + expect(resizeFrame?.slot).toBe(13); expect(decodeTerminalResizePayload(resizeFrame?.payload ?? new Uint8Array())).toEqual({ rows: 24, cols: 80, }); }); + test("routes concurrent terminal stream frames by slot", async () => { + const logger = createMockLogger(); + const mock = createMockTransport(); + + const client = new DaemonClient({ + url: "ws://test", + clientId: "clsk_unit_test", + logger, + reconnect: { enabled: false }, + transportFactory: () => mock.transport, + }); + clients.push(client); + + const connectPromise = client.connect(); + mock.triggerOpen(); + await connectPromise; + + const seen: string[] = []; + client.onTerminalStreamEvent((event) => { + if (event.type !== "output") { + return; + } + seen.push(`${event.terminalId}:${new TextDecoder().decode(event.data)}`); + }); + + const subscribeFirstPromise = client.subscribeTerminal("term-1", "sub-multi-1"); + mock.triggerMessage( + wrapSessionMessage({ + type: "subscribe_terminal_response", + payload: { + terminalId: "term-1", + slot: 21, + error: null, + requestId: "sub-multi-1", + }, + }), + ); + await subscribeFirstPromise; + + const subscribeSecondPromise = client.subscribeTerminal("term-2", "sub-multi-2"); + mock.triggerMessage( + wrapSessionMessage({ + type: "subscribe_terminal_response", + payload: { + terminalId: "term-2", + slot: 22, + error: null, + requestId: "sub-multi-2", + }, + }), + ); + await subscribeSecondPromise; + mock.sent.length = 0; + + mock.triggerMessage( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Output, + slot: 22, + payload: new TextEncoder().encode("beta"), + }), + ); + mock.triggerMessage( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Output, + slot: 21, + payload: new TextEncoder().encode("alpha"), + }), + ); + + client.sendTerminalInput("term-2", { + type: "input", + data: "echo beta\r", + }); + client.sendTerminalInput("term-1", { + type: "resize", + rows: 10, + cols: 20, + }); + + const inputFrame = decodeTerminalStreamFrame(asUint8Array(mock.sent[0])!); + const resizeFrame = decodeTerminalStreamFrame(asUint8Array(mock.sent[1])!); + + expect(seen).toEqual(["term-2:beta", "term-1:alpha"]); + expect(inputFrame?.opcode).toBe(TerminalStreamOpcode.Input); + expect(inputFrame?.slot).toBe(22); + expect(resizeFrame?.opcode).toBe(TerminalStreamOpcode.Resize); + expect(resizeFrame?.slot).toBe(21); + }); + test("ignores terminal stream frames after terminal_stream_exit", async () => { const logger = createMockLogger(); const mock = createMockTransport(); @@ -1415,6 +1511,7 @@ describe("DaemonClient", () => { type: "subscribe_terminal_response", payload: { terminalId: "term-1", + slot: 14, error: null, requestId: "sub-4", }, @@ -1425,6 +1522,7 @@ describe("DaemonClient", () => { mock.triggerMessage( encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Output, + slot: 14, payload: new TextEncoder().encode("before-exit"), }), ); @@ -1442,6 +1540,7 @@ describe("DaemonClient", () => { mock.triggerMessage( encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Output, + slot: 14, payload: new TextEncoder().encode("after-exit"), }), ); diff --git a/packages/server/src/client/daemon-client.ts b/packages/server/src/client/daemon-client.ts index 520e7c538..df9ed4945 100644 --- a/packages/server/src/client/daemon-client.ts +++ b/packages/server/src/client/daemon-client.ts @@ -419,7 +419,8 @@ export class DaemonClient { { cwd: string; compare: { mode: "uncommitted" | "base"; baseRef?: string } } >(); private terminalDirectorySubscriptions = new Set(); - private activeTerminalId: string | null = null; + private terminalSlots = new Map(); + private slotTerminals = new Map(); private readonly terminalStreamListeners = new Set<(event: TerminalStreamEvent) => void>(); private logger: Logger; private pendingSendQueue: PendingSend[] = []; @@ -659,7 +660,7 @@ export class DaemonClient { this.disposeTransport(1000, "Client closed"); this.clearWaiters(new Error("Daemon client closed")); this.rejectPendingSendQueue(new Error("Daemon client closed")); - this.activeTerminalId = null; + this.clearTerminalSlots(); this.lastServerInfoMessage = null; this.updateConnectionState( { status: "disposed" }, @@ -2613,15 +2614,13 @@ export class DaemonClient { options: { skipQueue: true }, }); if (!payload.error) { - this.activeTerminalId = terminalId; + this.setTerminalSlot(terminalId, payload.slot); } return payload; } unsubscribeTerminal(terminalId: string): void { - if (this.activeTerminalId === terminalId) { - this.activeTerminalId = null; - } + this.removeTerminalSlot(terminalId); this.sendSessionMessage({ type: "unsubscribe_terminal_request", terminalId, @@ -2629,11 +2628,13 @@ export class DaemonClient { } sendTerminalInput(terminalId: string, message: TerminalInput["message"]): void { - if (this.activeTerminalId === terminalId) { + const slot = this.terminalSlots.get(terminalId); + if (typeof slot === "number") { if (message.type === "input") { this.sendBinaryFrame( encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Input, + slot, payload: encodeUtf8String(message.data), }), ); @@ -2643,6 +2644,7 @@ export class DaemonClient { this.sendBinaryFrame( encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Resize, + slot, payload: encodeTerminalResizePayload({ rows: message.rows, cols: message.cols, @@ -2711,6 +2713,37 @@ export class DaemonClient { return requestId ?? crypto.randomUUID(); } + private setTerminalSlot(terminalId: string, slot: number): void { + const existingTerminalId = this.slotTerminals.get(slot); + if (existingTerminalId && existingTerminalId !== terminalId) { + this.terminalSlots.delete(existingTerminalId); + } + + const existingSlot = this.terminalSlots.get(terminalId); + if (typeof existingSlot === "number" && existingSlot !== slot) { + this.slotTerminals.delete(existingSlot); + } + + this.terminalSlots.set(terminalId, slot); + this.slotTerminals.set(slot, terminalId); + } + + private removeTerminalSlot(terminalId: string): void { + const slot = this.terminalSlots.get(terminalId); + if (typeof slot !== "number") { + return; + } + this.terminalSlots.delete(terminalId); + if (this.slotTerminals.get(slot) === terminalId) { + this.slotTerminals.delete(slot); + } + } + + private clearTerminalSlots(): void { + this.terminalSlots.clear(); + this.slotTerminals.clear(); + } + getLastServerInfoMessage(): ServerInfoStatusPayload | null { return this.lastServerInfoMessage; } @@ -2840,7 +2873,7 @@ export class DaemonClient { } private handleBinaryFrame(frame: TerminalStreamFrame): void { - const terminalId = this.activeTerminalId; + const terminalId = this.slotTerminals.get(frame.slot); if (!terminalId) { return; } @@ -2934,7 +2967,7 @@ export class DaemonClient { // and responses from the previous connection will never arrive. this.clearWaiters(new Error(reason ?? "Connection lost")); this.rejectPendingSendQueue(new Error(reason ?? "Connection lost")); - this.activeTerminalId = null; + this.clearTerminalSlots(); this.lastServerInfoMessage = null; if (wasDisposed) { @@ -2989,9 +3022,7 @@ export class DaemonClient { } if (msg.type === "terminal_stream_exit") { - if (this.activeTerminalId === msg.payload.terminalId) { - this.activeTerminalId = null; - } + this.removeTerminalSlot(msg.payload.terminalId); } if (this.rawMessageListeners.size > 0) { 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 d79a3dd34..4b044fabd 100644 --- a/packages/server/src/server/daemon-e2e/terminal.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/terminal.e2e.test.ts @@ -12,6 +12,7 @@ import { import { decodeTerminalSnapshotPayload, decodeTerminalStreamFrame, + encodeTerminalStreamFrame, TerminalStreamOpcode, type TerminalStreamFrame, } from "../../shared/terminal-stream-protocol.js"; @@ -375,7 +376,11 @@ function getFrameText(frame: TerminalStreamFrame): string { return ""; } -async function subscribeRawTerminal(ws: WebSocket, terminalId: string, requestId: string): Promise { +async function subscribeRawTerminal( + ws: WebSocket, + terminalId: string, + requestId: string, +): Promise { const ready = waitForRawSessionMessage( ws, (message) => @@ -400,6 +405,7 @@ async function subscribeRawTerminal(ws: WebSocket, terminalId: string, requestId throw new Error("Expected subscribe_terminal_response"); } expect(message.message.payload).not.toHaveProperty("state"); + return message.message.payload.slot; } describe("daemon E2E terminal", () => { @@ -435,10 +441,8 @@ describe("daemon E2E terminal", () => { }); await new Promise((resolve) => setTimeout(resolve, 300)); - const snapshotPromise = waitForTerminalSnapshot( - ctx.client, - terminalId, - (state) => extractStateText(state).includes("hello"), + const snapshotPromise = waitForTerminalSnapshot(ctx.client, terminalId, (state) => + extractStateText(state).includes("hello"), ); await ctx.client.subscribeTerminal(terminalId); const snapshot = await snapshotPromise; @@ -453,10 +457,8 @@ describe("daemon E2E terminal", () => { const created = await ctx.client.createTerminal(cwd); const terminalId = created.terminal!.id; - const outputPromise = waitForTerminalOutput( - ctx.client, - terminalId, - (text) => text.includes("binary-stream"), + const outputPromise = waitForTerminalOutput(ctx.client, terminalId, (text) => + text.includes("binary-stream"), ); await ctx.client.subscribeTerminal(terminalId); ctx.client.sendTerminalInput(terminalId, { @@ -469,6 +471,42 @@ describe("daemon E2E terminal", () => { rmSync(cwd, { recursive: true, force: true }); }, 30000); + test("one client can stream two terminals concurrently", async () => { + const cwd = tmpCwd(); + const firstCreated = await ctx.client.createTerminal(cwd, "first"); + const secondCreated = await ctx.client.createTerminal(cwd, "second"); + const firstTerminalId = firstCreated.terminal!.id; + const secondTerminalId = secondCreated.terminal!.id; + + const firstSubscribe = await ctx.client.subscribeTerminal(firstTerminalId); + const secondSubscribe = await ctx.client.subscribeTerminal(secondTerminalId); + + expect(firstSubscribe.error).toBeNull(); + expect(secondSubscribe.error).toBeNull(); + expect(firstSubscribe.slot).not.toBe(secondSubscribe.slot); + + const firstOutput = waitForTerminalOutput(ctx.client, firstTerminalId, (text) => + text.includes("from-first"), + ); + const secondOutput = waitForTerminalOutput(ctx.client, secondTerminalId, (text) => + text.includes("from-second"), + ); + + ctx.client.sendTerminalInput(firstTerminalId, { + type: "input", + data: "echo from-first\r", + }); + ctx.client.sendTerminalInput(secondTerminalId, { + type: "input", + data: "echo from-second\r", + }); + + expect(await firstOutput).toContain("from-first"); + expect(await secondOutput).toContain("from-second"); + + rmSync(cwd, { recursive: true, force: true }); + }, 30000); + test("disconnect and reconnect both receive the current snapshot", async () => { const cwd = tmpCwd(); const created = await ctx.client.createTerminal(cwd); @@ -483,10 +521,8 @@ describe("daemon E2E terminal", () => { }); await new Promise((resolve) => setTimeout(resolve, 300)); - const snapshotPromise = waitForTerminalSnapshot( - ctx.client, - terminalId, - (state) => extractStateText(state).includes("while-detached"), + const snapshotPromise = waitForTerminalSnapshot(ctx.client, terminalId, (state) => + extractStateText(state).includes("while-detached"), ); await ctx.client.subscribeTerminal(terminalId); @@ -502,9 +538,14 @@ describe("daemon E2E terminal", () => { const ws = await connectRawWebSocket(ctx.daemon.port); await subscribeRawTerminal(ws, terminalId, "sub-raw"); - await waitForRawBinaryFrame(ws, (frame) => frame.opcode === TerminalStreamOpcode.Snapshot, 10000); + await waitForRawBinaryFrame( + ws, + (frame) => frame.opcode === TerminalStreamOpcode.Snapshot, + 10000, + ); - const rawSocket = (ws as WebSocket & { _socket?: { pause: () => void; resume: () => void } })._socket; + const rawSocket = (ws as WebSocket & { _socket?: { pause: () => void; resume: () => void } }) + ._socket; expect(rawSocket).toBeDefined(); rawSocket!.pause(); @@ -537,15 +578,11 @@ describe("daemon E2E terminal", () => { ); try { - const firstOutput = waitForTerminalOutput( - ctx.client, - terminalId, - (text) => text.includes("fanout"), + const firstOutput = waitForTerminalOutput(ctx.client, terminalId, (text) => + text.includes("fanout"), ); - const secondOutput = waitForTerminalOutput( - secondClient, - terminalId, - (text) => text.includes("fanout"), + const secondOutput = waitForTerminalOutput(secondClient, terminalId, (text) => + text.includes("fanout"), ); await ctx.client.subscribeTerminal(terminalId); @@ -689,15 +726,11 @@ describe("daemon E2E terminal", () => { await ctx.client.subscribeTerminal(terminalId); await secondClient.subscribeTerminal(terminalId); - const firstOwnOutput = waitForTerminalOutput( - ctx.client, - terminalId, - (text) => text.includes("from-a"), + const firstOwnOutput = waitForTerminalOutput(ctx.client, terminalId, (text) => + text.includes("from-a"), ); - const secondOwnOutput = waitForTerminalOutput( - secondClient, - terminalId, - (text) => text.includes("from-b"), + const secondOwnOutput = waitForTerminalOutput(secondClient, terminalId, (text) => + text.includes("from-b"), ); ctx.client.sendTerminalInput(terminalId, { @@ -764,7 +797,11 @@ describe("daemon E2E terminal", () => { try { await subscribeRawTerminal(ws, terminalId, "sub-exit"); - await waitForRawBinaryFrame(ws, (frame) => frame.opcode === TerminalStreamOpcode.Snapshot, 10000); + await waitForRawBinaryFrame( + ws, + (frame) => frame.opcode === TerminalStreamOpcode.Snapshot, + 10000, + ); const exitMessagePromise = waitForRawSessionMessage( ws, @@ -794,10 +831,19 @@ describe("daemon E2E terminal", () => { const ws = await connectRawWebSocket(ctx.daemon.port); try { - await subscribeRawTerminal(ws, terminalId, "sub-empty-input"); - await waitForRawBinaryFrame(ws, (frame) => frame.opcode === TerminalStreamOpcode.Snapshot, 10000); + const slot = await subscribeRawTerminal(ws, terminalId, "sub-empty-input"); + await waitForRawBinaryFrame( + ws, + (frame) => frame.opcode === TerminalStreamOpcode.Snapshot, + 10000, + ); - ws.send(new Uint8Array([TerminalStreamOpcode.Input])); + ws.send( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Input, + slot, + }), + ); await new Promise((resolve) => setTimeout(resolve, 200)); expect(ws.readyState).toBe(WebSocket.OPEN); @@ -825,7 +871,11 @@ describe("daemon E2E terminal", () => { try { await subscribeRawTerminal(ws, terminalId, "sub-large-output"); - await waitForRawBinaryFrame(ws, (frame) => frame.opcode === TerminalStreamOpcode.Snapshot, 10000); + await waitForRawBinaryFrame( + ws, + (frame) => frame.opcode === TerminalStreamOpcode.Snapshot, + 10000, + ); ctx.client.sendTerminalInput(terminalId, { type: "input", diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 77549b69d..3d2979c1c 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -192,6 +192,7 @@ const WORKSPACE_GIT_WATCH_DEBOUNCE_MS = 500; const WORKSPACE_GIT_WATCH_REMOVED_FINGERPRINT = "__removed__"; const TERMINAL_STREAM_HIGH_WATER_BYTES = 256 * 1024; const TERMINAL_STREAM_LOW_WATER_BYTES = 16 * 1024; +const MAX_TERMINAL_STREAM_SLOTS = 256; function deriveInitialAgentTitle(prompt: string): string | null { const firstContentLine = prompt @@ -277,6 +278,7 @@ type CheckoutErrorPayload = { type ActiveTerminalStream = { terminalId: string; + slot: number; unsubscribe: () => void; needsSnapshot: boolean; snapshotRetryTimer: ReturnType | null; @@ -605,7 +607,9 @@ export class Session { private readonly subscribedTerminalDirectories = new Set(); private unsubscribeTerminalsChanged: (() => void) | null = null; private terminalExitSubscriptions: Map void> = new Map(); - private activeTerminalStream: ActiveTerminalStream | null = null; + private readonly activeTerminalStreams = new Map(); + private readonly terminalIdToSlot = new Map(); + private nextTerminalSlot = 0; private readonly checkoutDiffSubscriptions = new Map(); private readonly checkoutDiffTargets = new Map(); private readonly workspaceGitWatchTargets = new Map(); @@ -749,7 +753,7 @@ export class Session { checkoutDiffWatcherCount, checkoutDiffFallbackRefreshTargetCount, terminalDirectorySubscriptionCount: this.subscribedTerminalDirectories.size, - terminalSubscriptionCount: this.activeTerminalStream ? 1 : 0, + terminalSubscriptionCount: this.activeTerminalStreams.size, }; } @@ -1750,7 +1754,6 @@ export class Session { case "kill_terminal_request": await this.handleKillTerminalRequest(msg); break; - } } catch (error: any) { const err = error instanceof Error ? error : new Error(String(error)); @@ -1786,13 +1789,13 @@ export class Session { } public handleBinaryFrame(frame: TerminalStreamFrame): void { - const activeStream = this.activeTerminalStream; + const activeStream = this.activeTerminalStreams.get(frame.slot); if (!activeStream || !this.terminalManager) { return; } const terminal = this.terminalManager.getTerminal(activeStream.terminalId); if (!terminal) { - this.detachActiveTerminalStream({ emitExit: true }); + this.detachTerminalStream(activeStream.terminalId, { emitExit: true }); return; } @@ -1990,10 +1993,7 @@ export class Session { try { await this.agentManager.closeAgent(agentId); } catch (error) { - this.sessionLogger.warn( - { err: error, agentId }, - "Failed to close agent during archive", - ); + this.sessionLogger.warn({ err: error, agentId }, "Failed to close agent during archive"); } } @@ -7474,7 +7474,7 @@ export class Session { unsubscribeExit(); } this.terminalExitSubscriptions.clear(); - this.detachActiveTerminalStream({ emitExit: false }); + this.disposeTerminalSubscriptions(); for (const target of this.checkoutDiffTargets.values()) { this.closeCheckoutDiffWatchTarget(target); @@ -7510,9 +7510,7 @@ export class Session { this.terminalExitSubscriptions.delete(terminalId); } - if (this.activeTerminalStream?.terminalId === terminalId) { - this.detachActiveTerminalStream({ emitExit: true }); - } + this.detachTerminalStream(terminalId, { emitExit: true }); } private emitTerminalsChangedSnapshot(input: { @@ -7663,6 +7661,7 @@ export class Session { type: "subscribe_terminal_response", payload: { terminalId: msg.terminalId, + slot: 0, error: "Terminal manager not available", requestId: msg.requestId, }, @@ -7676,6 +7675,7 @@ export class Session { type: "subscribe_terminal_response", payload: { terminalId: msg.terminalId, + slot: 0, error: "Terminal not found", requestId: msg.requestId, }, @@ -7684,22 +7684,33 @@ export class Session { } this.ensureTerminalExitSubscription(session); + const slot = this.bindActiveTerminalStream(session); + if (slot === null) { + this.emit({ + type: "subscribe_terminal_response", + payload: { + terminalId: msg.terminalId, + slot: 0, + error: "No terminal stream slots available", + requestId: msg.requestId, + }, + }); + return; + } + this.emit({ type: "subscribe_terminal_response", payload: { terminalId: msg.terminalId, + slot, error: null, requestId: msg.requestId, }, }); - - this.bindActiveTerminalStream(session); } private handleUnsubscribeTerminalRequest(msg: UnsubscribeTerminalRequest): void { - if (this.activeTerminalStream?.terminalId === msg.terminalId) { - this.detachActiveTerminalStream({ emitExit: false }); - } + this.detachTerminalStream(msg.terminalId, { emitExit: false }); } private handleTerminalInput(msg: TerminalInput): void { @@ -7718,10 +7729,7 @@ export class Session { } private killTrackedTerminal(terminalId: string, options?: { emitExit: boolean }): void { - if (this.activeTerminalStream?.terminalId === terminalId) { - this.detachActiveTerminalStream({ emitExit: options?.emitExit ?? true }); - } - + this.detachTerminalStream(terminalId, { emitExit: options?.emitExit ?? true }); this.terminalManager?.killTerminal(terminalId); } @@ -7782,82 +7790,137 @@ export class Session { }); } - private bindActiveTerminalStream(terminal: TerminalSession): ActiveTerminalStream | null { - this.detachActiveTerminalStream({ emitExit: false }); + private bindActiveTerminalStream(terminal: TerminalSession): number | null { if (!this.onBinaryMessage) { return null; } + const existingSlot = this.terminalIdToSlot.get(terminal.id); + if (typeof existingSlot === "number") { + const existingStream = this.activeTerminalStreams.get(existingSlot); + if (existingStream) { + existingStream.needsSnapshot = true; + this.trySendTerminalSnapshot(existingStream); + return existingSlot; + } + this.terminalIdToSlot.delete(terminal.id); + } + + const slot = this.allocateTerminalSlot(); + if (slot === null) { + return null; + } + const activeStream: ActiveTerminalStream = { terminalId: terminal.id, + slot, unsubscribe: () => {}, 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()), - }), - ); - }; + this.activeTerminalStreams.set(slot, activeStream); + this.terminalIdToSlot.set(terminal.id, slot); activeStream.unsubscribe = terminal.subscribe((message) => { - if (this.activeTerminalStream !== activeStream) { + if (this.activeTerminalStreams.get(slot) !== activeStream) { return; } if (message.type === "snapshot") { - trySendSnapshot(); + this.trySendTerminalSnapshot(activeStream); return; } if (activeStream.needsSnapshot || message.data.length === 0) { return; } if (this.getCurrentBinaryBufferedAmount() >= TERMINAL_STREAM_HIGH_WATER_BYTES) { - activeStream.needsSnapshot = true; - trySendSnapshot(); + this.markAllActiveTerminalStreamsForSnapshot(); return; } this.emitBinary( encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Output, + slot, payload: new Uint8Array(Buffer.from(message.data, "utf8")), }), ); if (this.getCurrentBinaryBufferedAmount() >= TERMINAL_STREAM_HIGH_WATER_BYTES) { - activeStream.needsSnapshot = true; - trySendSnapshot(); + this.markAllActiveTerminalStreamsForSnapshot(); } }); - this.activeTerminalStream = activeStream; - return activeStream; + return slot; } - private detachActiveTerminalStream(options?: { emitExit: boolean }): boolean { - const activeStream = this.activeTerminalStream; - if (!activeStream) { + private trySendTerminalSnapshot(activeStream: ActiveTerminalStream): void { + if ( + this.activeTerminalStreams.get(activeStream.slot) !== activeStream || + !activeStream.needsSnapshot + ) { + return; + } + + if (this.getCurrentBinaryBufferedAmount() > TERMINAL_STREAM_LOW_WATER_BYTES) { + if (!activeStream.snapshotRetryTimer) { + activeStream.snapshotRetryTimer = setTimeout(() => { + activeStream.snapshotRetryTimer = null; + this.trySendTerminalSnapshot(activeStream); + }, 33); + } + return; + } + + if (activeStream.snapshotRetryTimer) { + clearTimeout(activeStream.snapshotRetryTimer); + activeStream.snapshotRetryTimer = null; + } + + const terminal = this.terminalManager?.getTerminal(activeStream.terminalId); + if (!terminal) { + this.detachTerminalStream(activeStream.terminalId, { emitExit: true }); + return; + } + + activeStream.needsSnapshot = false; + this.emitBinary( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Snapshot, + slot: activeStream.slot, + payload: encodeTerminalSnapshotPayload(terminal.getState()), + }), + ); + } + + private markAllActiveTerminalStreamsForSnapshot(): void { + for (const activeStream of this.activeTerminalStreams.values()) { + activeStream.needsSnapshot = true; + this.trySendTerminalSnapshot(activeStream); + } + } + + private allocateTerminalSlot(): number | null { + for (let attempt = 0; attempt < MAX_TERMINAL_STREAM_SLOTS; attempt += 1) { + const slot = (this.nextTerminalSlot + attempt) % MAX_TERMINAL_STREAM_SLOTS; + if (this.activeTerminalStreams.has(slot)) { + continue; + } + this.nextTerminalSlot = (slot + 1) % MAX_TERMINAL_STREAM_SLOTS; + return slot; + } + return null; + } + + private detachTerminalStream(terminalId: string, options?: { emitExit: boolean }): boolean { + const slot = this.terminalIdToSlot.get(terminalId); + if (typeof slot !== "number") { return false; } - this.activeTerminalStream = null; + const activeStream = this.activeTerminalStreams.get(slot); + if (!activeStream) { + this.terminalIdToSlot.delete(terminalId); + return false; + } + this.activeTerminalStreams.delete(slot); + this.terminalIdToSlot.delete(terminalId); if (activeStream.snapshotRetryTimer) { clearTimeout(activeStream.snapshotRetryTimer); activeStream.snapshotRetryTimer = null; @@ -7878,6 +7941,12 @@ export class Session { return true; } + private disposeTerminalSubscriptions(): void { + for (const terminalId of [...this.terminalIdToSlot.keys()]) { + this.detachTerminalStream(terminalId, { emitExit: false }); + } + } + private getCurrentBinaryBufferedAmount(): number { const bufferedAmount = this.getBinaryBufferedAmount?.() ?? 0; if (!Number.isFinite(bufferedAmount) || bufferedAmount < 0) { diff --git a/packages/server/src/server/websocket-server.relay-reconnect.test.ts b/packages/server/src/server/websocket-server.relay-reconnect.test.ts index feb312f0b..0e251a9d6 100644 --- a/packages/server/src/server/websocket-server.relay-reconnect.test.ts +++ b/packages/server/src/server/websocket-server.relay-reconnect.test.ts @@ -639,6 +639,7 @@ describe("relay external socket reconnect behavior", () => { Buffer.from( encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Input, + slot: 9, payload: new TextEncoder().encode("ls\r"), }), ), @@ -648,9 +649,11 @@ describe("relay external socket reconnect behavior", () => { expect(session.handleBinaryFrame).toHaveBeenCalledTimes(1); const frame = session.handleBinaryFrame.mock.calls[0]?.[0] as { opcode: number; + slot: number; payload: Uint8Array; }; expect(frame.opcode).toBe(TerminalStreamOpcode.Input); + expect(frame.slot).toBe(9); expect(new TextDecoder().decode(frame.payload)).toBe("ls\r"); await server.close(); @@ -673,7 +676,7 @@ describe("relay external socket reconnect behavior", () => { | undefined; expect(onBinaryMessage).toBeTypeOf("function"); - onBinaryMessage?.(new Uint8Array([TerminalStreamOpcode.Output, 0x6f, 0x6b])); + onBinaryMessage?.(new Uint8Array([TerminalStreamOpcode.Output, 12, 0x6f, 0x6b])); expect(socket.sent).toHaveLength(2); const binaryPayload = asUint8Array(socket.sent[1]); @@ -681,6 +684,7 @@ describe("relay external socket reconnect behavior", () => { const frame = decodeTerminalStreamFrame(binaryPayload!); expect(frame).not.toBeNull(); expect(frame!.opcode).toBe(TerminalStreamOpcode.Output); + expect(frame!.slot).toBe(12); expect(new TextDecoder().decode(frame!.payload ?? new Uint8Array())).toBe("ok"); await server.close(); diff --git a/packages/server/src/shared/messages.ts b/packages/server/src/shared/messages.ts index e86a4cfd2..f96199bae 100644 --- a/packages/server/src/shared/messages.ts +++ b/packages/server/src/shared/messages.ts @@ -2095,37 +2095,43 @@ const TerminalInfoSchema = z.object({ cwd: z.string(), }); -export const TerminalCellSchema = z.object({ - char: z.string(), - fg: z.number().optional(), - bg: z.number().optional(), - fgMode: z.number().optional(), - bgMode: z.number().optional(), - bold: z.boolean().optional(), - italic: z.boolean().optional(), - underline: z.boolean().optional(), - dim: z.boolean().optional(), - inverse: z.boolean().optional(), - strikethrough: z.boolean().optional(), -}).strict(); +export const TerminalCellSchema = z + .object({ + char: z.string(), + fg: z.number().optional(), + bg: z.number().optional(), + fgMode: z.number().optional(), + bgMode: z.number().optional(), + bold: z.boolean().optional(), + italic: z.boolean().optional(), + underline: z.boolean().optional(), + dim: z.boolean().optional(), + inverse: z.boolean().optional(), + strikethrough: z.boolean().optional(), + }) + .strict(); export const TerminalCursorStyleSchema = z.enum(["block", "underline", "bar"]); -export const TerminalCursorSchema = z.object({ - row: z.number(), - col: z.number(), - hidden: z.boolean().optional(), - style: TerminalCursorStyleSchema.optional(), - blink: z.boolean().optional(), -}).strict(); +export const TerminalCursorSchema = z + .object({ + row: z.number(), + col: z.number(), + hidden: z.boolean().optional(), + style: TerminalCursorStyleSchema.optional(), + blink: 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: TerminalCursorSchema, -}).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: TerminalCursorSchema, + }) + .strict(); export const ListTerminalsResponseSchema = z.object({ type: z.literal("list_terminals_response"), @@ -2157,6 +2163,7 @@ export const SubscribeTerminalResponseSchema = z.object({ type: z.literal("subscribe_terminal_response"), payload: z.object({ terminalId: z.string(), + slot: z.number().int().min(0).max(255), 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 a391eed72..ac395fd28 100644 --- a/packages/server/src/shared/terminal-stream-protocol.test.ts +++ b/packages/server/src/shared/terminal-stream-protocol.test.ts @@ -11,19 +11,22 @@ import { } from "./terminal-stream-protocol.js"; describe("terminal stream protocol", () => { - it("encodes output frames as a 1-byte opcode prefix plus payload", () => { + it("encodes output frames as opcode plus slot plus payload", () => { const payload = new TextEncoder().encode("hello"); const encoded = encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Output, + slot: 7, payload, }); expect(encoded[0]).toBe(TerminalStreamOpcode.Output); - expect(Array.from(encoded.subarray(1))).toEqual(Array.from(payload)); + expect(encoded[1]).toBe(7); + expect(Array.from(encoded.subarray(2))).toEqual(Array.from(payload)); const decoded = decodeTerminalStreamFrame(encoded); expect(decoded).toEqual({ opcode: TerminalStreamOpcode.Output, + slot: 7, payload, }); }); @@ -54,7 +57,11 @@ describe("terminal stream protocol", () => { }); it("rejects unknown opcodes", () => { - expect(decodeTerminalStreamFrame(new Uint8Array([0xff, 0x01]))).toBeNull(); + expect(decodeTerminalStreamFrame(new Uint8Array([0xff, 0x01, 0x02]))).toBeNull(); + }); + + it("rejects frames without a slot byte", () => { + expect(decodeTerminalStreamFrame(new Uint8Array([TerminalStreamOpcode.Output]))).toBeNull(); }); it("rejects malformed JSON payloads", () => { @@ -66,7 +73,9 @@ describe("terminal stream protocol", () => { it("rejects invalid resize and snapshot shapes", () => { expect( - decodeTerminalResizePayload(new TextEncoder().encode(JSON.stringify({ rows: "24", cols: 80 }))), + decodeTerminalResizePayload( + new TextEncoder().encode(JSON.stringify({ rows: "24", cols: 80 })), + ), ).toBeNull(); expect( decodeTerminalSnapshotPayload( diff --git a/packages/server/src/shared/terminal-stream-protocol.ts b/packages/server/src/shared/terminal-stream-protocol.ts index f27a55621..33461f367 100644 --- a/packages/server/src/shared/terminal-stream-protocol.ts +++ b/packages/server/src/shared/terminal-stream-protocol.ts @@ -1,10 +1,12 @@ import { z } from "zod"; import { TerminalStateSchema } from "./messages.js"; -export const TerminalStreamResizeSchema = z.object({ - rows: z.number().int().positive(), - cols: z.number().int().positive(), -}).strict(); +export const TerminalStreamResizeSchema = z + .object({ + rows: z.number().int().positive(), + cols: z.number().int().positive(), + }) + .strict(); export const TerminalStreamOpcode = { Output: 0x01, @@ -13,11 +15,11 @@ export const TerminalStreamOpcode = { Snapshot: 0x04, } as const; -export type TerminalStreamOpcode = - (typeof TerminalStreamOpcode)[keyof typeof TerminalStreamOpcode]; +export type TerminalStreamOpcode = (typeof TerminalStreamOpcode)[keyof typeof TerminalStreamOpcode]; export type TerminalStreamFrame = { opcode: TerminalStreamOpcode; + slot: number; payload: Uint8Array; }; @@ -61,17 +63,19 @@ function isTerminalStreamOpcode(value: number): value is TerminalStreamOpcode { export function encodeTerminalStreamFrame(input: { opcode: TerminalStreamOpcode; + slot: number; payload?: Uint8Array | ArrayBuffer | string; }): Uint8Array { const payload = asUint8Array(input.payload ?? new Uint8Array(0)) ?? new Uint8Array(0); - const bytes = new Uint8Array(1 + payload.byteLength); + const bytes = new Uint8Array(2 + payload.byteLength); bytes[0] = input.opcode; - bytes.set(payload, 1); + bytes[1] = input.slot & 0xff; + bytes.set(payload, 2); return bytes; } export function decodeTerminalStreamFrame(bytes: Uint8Array): TerminalStreamFrame | null { - if (bytes.byteLength < 1) { + if (bytes.byteLength < 2) { return null; } const opcode = bytes[0]; @@ -80,11 +84,14 @@ export function decodeTerminalStreamFrame(bytes: Uint8Array): TerminalStreamFram } return { opcode, - payload: bytes.subarray(1), + slot: bytes[1] ?? 0, + payload: bytes.subarray(2), }; } -export function encodeTerminalSnapshotPayload(state: z.infer): Uint8Array { +export function encodeTerminalSnapshotPayload( + state: z.infer, +): Uint8Array { return encodeJsonPayload(state); } From df99442123e1cfe37f12eb9f569a5f6e859f6c28 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 23 Mar 2026 12:44:18 +0700 Subject: [PATCH 2/2] Fix terminal slot review issues --- packages/server/src/client/daemon-client.ts | 2 +- .../server/daemon-e2e/terminal.e2e.test.ts | 5 +++++ packages/server/src/server/session.ts | 10 +++++++--- packages/server/src/shared/messages.ts | 19 +++++++++++++------ .../src/shared/terminal-stream-protocol.ts | 2 +- 5 files changed, 27 insertions(+), 11 deletions(-) diff --git a/packages/server/src/client/daemon-client.ts b/packages/server/src/client/daemon-client.ts index df9ed4945..f927f3fa6 100644 --- a/packages/server/src/client/daemon-client.ts +++ b/packages/server/src/client/daemon-client.ts @@ -2613,7 +2613,7 @@ export class DaemonClient { timeout: 10000, options: { skipQueue: true }, }); - if (!payload.error) { + if (payload.error === null) { this.setTerminalSlot(terminalId, payload.slot); } return payload; 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 4b044fabd..73f3b5410 100644 --- a/packages/server/src/server/daemon-e2e/terminal.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/terminal.e2e.test.ts @@ -404,6 +404,11 @@ async function subscribeRawTerminal( if (message.message.type !== "subscribe_terminal_response") { throw new Error("Expected subscribe_terminal_response"); } + if (message.message.payload.error !== null) { + throw new Error( + `Expected successful subscribe_terminal_response: ${message.message.payload.error}`, + ); + } expect(message.message.payload).not.toHaveProperty("state"); return message.message.payload.slot; } diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 3d2979c1c..a68bcf37d 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -7661,7 +7661,6 @@ export class Session { type: "subscribe_terminal_response", payload: { terminalId: msg.terminalId, - slot: 0, error: "Terminal manager not available", requestId: msg.requestId, }, @@ -7675,7 +7674,6 @@ export class Session { type: "subscribe_terminal_response", payload: { terminalId: msg.terminalId, - slot: 0, error: "Terminal not found", requestId: msg.requestId, }, @@ -7686,11 +7684,17 @@ export class Session { const slot = this.bindActiveTerminalStream(session); if (slot === null) { + this.sessionLogger.warn( + { + terminalId: msg.terminalId, + activeTerminalStreamCount: this.activeTerminalStreams.size, + }, + "Terminal stream slot exhaustion", + ); this.emit({ type: "subscribe_terminal_response", payload: { terminalId: msg.terminalId, - slot: 0, error: "No terminal stream slots available", requestId: msg.requestId, }, diff --git a/packages/server/src/shared/messages.ts b/packages/server/src/shared/messages.ts index f96199bae..ee51e0b2a 100644 --- a/packages/server/src/shared/messages.ts +++ b/packages/server/src/shared/messages.ts @@ -2161,12 +2161,19 @@ export const CreateTerminalResponseSchema = z.object({ export const SubscribeTerminalResponseSchema = z.object({ type: z.literal("subscribe_terminal_response"), - payload: z.object({ - terminalId: z.string(), - slot: z.number().int().min(0).max(255), - error: z.string().nullable(), - requestId: z.string(), - }), + payload: z.union([ + z.object({ + terminalId: z.string(), + slot: z.number().int().min(0).max(255), + error: z.null(), + requestId: z.string(), + }), + z.object({ + terminalId: z.string(), + error: z.string(), + requestId: z.string(), + }), + ]), }); export const KillTerminalResponseSchema = z.object({ diff --git a/packages/server/src/shared/terminal-stream-protocol.ts b/packages/server/src/shared/terminal-stream-protocol.ts index 33461f367..6970b9620 100644 --- a/packages/server/src/shared/terminal-stream-protocol.ts +++ b/packages/server/src/shared/terminal-stream-protocol.ts @@ -84,7 +84,7 @@ export function decodeTerminalStreamFrame(bytes: Uint8Array): TerminalStreamFram } return { opcode, - slot: bytes[1] ?? 0, + slot: bytes[1], payload: bytes.subarray(2), }; }