From 69f851aa14428fe58ce008bf18a5c6c8e781c118 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 13 Feb 2026 09:09:39 +0000 Subject: [PATCH] Update files --- .tool-versions | 1 + .../app/src/utils/tauri-daemon-transport.ts | 23 +- packages/cli/src/utils/client.ts | 3 +- packages/relay/src/e2e.test.ts | 51 ++- .../server/src/client/daemon-client.test.ts | 190 ++++++++- packages/server/src/client/daemon-client.ts | 352 ++++++++++++++- .../daemon-e2e/relay-transport.e2e.test.ts | 55 ++- .../server/daemon-e2e/terminal.e2e.test.ts | 313 ++++++++++++++ packages/server/src/server/relay-transport.ts | 17 +- packages/server/src/server/session.ts | 401 ++++++++++++++++++ .../websocket-server.relay-reconnect.test.ts | 98 ++++- .../server/src/server/websocket-server.ts | 33 +- packages/server/src/shared/binary-mux.test.ts | 43 ++ packages/server/src/shared/binary-mux.ts | 120 ++++++ packages/server/src/shared/messages.ts | 56 +++ .../src/shared/terminal-key-input.test.ts | 25 ++ .../server/src/shared/terminal-key-input.ts | 146 +++++++ packages/server/src/terminal/terminal.test.ts | 85 ++++ packages/server/src/terminal/terminal.ts | 136 +++++- 19 files changed, 2120 insertions(+), 28 deletions(-) create mode 100644 packages/server/src/shared/binary-mux.test.ts create mode 100644 packages/server/src/shared/binary-mux.ts create mode 100644 packages/server/src/shared/terminal-key-input.test.ts create mode 100644 packages/server/src/shared/terminal-key-input.ts diff --git a/.tool-versions b/.tool-versions index ee114d2f1..2f1e58da7 100644 --- a/.tool-versions +++ b/.tool-versions @@ -1 +1,2 @@ rust 1.85.1 +nodejs 22.20.0 diff --git a/packages/app/src/utils/tauri-daemon-transport.ts b/packages/app/src/utils/tauri-daemon-transport.ts index 2e90bcb14..1f97193af 100644 --- a/packages/app/src/utils/tauri-daemon-transport.ts +++ b/packages/app/src/utils/tauri-daemon-transport.ts @@ -17,6 +17,18 @@ type TauriWebSocketModule = { connect(url: string, config?: unknown): Promise; }; +function toTauriOutgoingMessage( + data: string | Uint8Array | ArrayBuffer +): string | number[] { + if (typeof data === "string") { + return data; + } + if (data instanceof ArrayBuffer) { + return Array.from(new Uint8Array(data)); + } + return Array.from(data); +} + function isTauriEnvironment(): boolean { return typeof window !== "undefined" && (window as any).__TAURI__ !== undefined; } @@ -49,7 +61,7 @@ export function createTauriWebSocketTransportFactory(): DaemonTransportFactory | const errorHandlers = new Set<(event?: unknown) => void>(); const messageHandlers = new Set<(data: unknown) => void>(); - const pendingSends: string[] = []; + const pendingSends: Array = []; let closeRequested: { code?: number; reason?: string } | null = null; const emitOpen = () => { @@ -131,6 +143,10 @@ export function createTauriWebSocketTransportFactory(): DaemonTransportFactory | emitMessage({ data: msg.data }); return; } + if (msg.type === "Binary") { + emitMessage({ data: new Uint8Array(msg.data) }); + return; + } if (msg.type === "Close") { emitClose(msg.data ?? undefined); return; @@ -168,11 +184,12 @@ export function createTauriWebSocketTransportFactory(): DaemonTransportFactory | const transport: DaemonTransport = { send: (data) => { if (disposed) return; + const outgoing = toTauriOutgoingMessage(data); if (!ws) { - pendingSends.push(data); + pendingSends.push(outgoing); return; } - void ws.send(data).catch((error) => emitError(error)); + void ws.send(outgoing).catch((error) => emitError(error)); }, close: (code?: number, reason?: string) => { if (disposed) return; diff --git a/packages/cli/src/utils/client.ts b/packages/cli/src/utils/client.ts index d66ce8f98..51786a948 100644 --- a/packages/cli/src/utils/client.ts +++ b/packages/cli/src/utils/client.ts @@ -23,8 +23,9 @@ function createNodeWebSocketFactory() { return (url: string, options?: { headers?: Record }) => { return new WebSocket(url, { headers: options?.headers }) as unknown as { readyState: number - send: (data: string) => void + send: (data: string | Uint8Array | ArrayBuffer) => void close: (code?: number, reason?: string) => void + binaryType?: string on: (event: string, listener: (...args: unknown[]) => void) => void off: (event: string, listener: (...args: unknown[]) => void) => void } diff --git a/packages/relay/src/e2e.test.ts b/packages/relay/src/e2e.test.ts index 1c8b932a5..46cc761d2 100644 --- a/packages/relay/src/e2e.test.ts +++ b/packages/relay/src/e2e.test.ts @@ -12,6 +12,9 @@ import { decrypt, } from "./crypto.js"; +const nodeMajor = Number((process.versions.node ?? "0").split(".")[0] ?? "0"); +const shouldRunRelayE2e = process.env.FORCE_RELAY_E2E === "1" || nodeMajor < 25; + async function getAvailablePort(): Promise { return new Promise((resolve, reject) => { const server = net.createServer(); @@ -46,7 +49,36 @@ async function waitForServer(port: number, timeout = 15000): Promise { throw new Error(`Server did not start on port ${port} within ${timeout}ms`); } -describe("E2E Relay with E2EE", () => { +async function waitForRelayWebSocketReady(port: number, timeout = 60000): Promise { + const start = Date.now(); + while (Date.now() - start < timeout) { + const serverId = `probe-${Date.now().toString(36)}-${Math.random().toString(36).slice(2)}`; + const probeUrl = `ws://127.0.0.1:${port}/ws?serverId=${serverId}&role=server`; + const opened = await new Promise((resolve) => { + const ws = new WebSocket(probeUrl); + const timer = setTimeout(() => { + ws.terminate(); + resolve(false); + }, 5000); + ws.once("open", () => { + clearTimeout(timer); + ws.close(1000, "probe"); + resolve(true); + }); + ws.once("error", () => { + clearTimeout(timer); + resolve(false); + }); + }); + if (opened) { + return; + } + await new Promise((r) => setTimeout(r, 250)); + } + throw new Error(`Relay WebSocket endpoint not ready on port ${port} within ${timeout}ms`); +} + +(shouldRunRelayE2e ? describe : describe.skip)("E2E Relay with E2EE", () => { let relayPort: number; let relayProcess: ChildProcess | null = null; @@ -54,7 +86,17 @@ describe("E2E Relay with E2EE", () => { relayPort = await getAvailablePort(); relayProcess = spawn( "npx", - ["wrangler", "dev", "--local", "--ip", "127.0.0.1", "--port", String(relayPort)], + [ + "wrangler", + "dev", + "--local", + "--ip", + "127.0.0.1", + "--port", + String(relayPort), + "--live-reload=false", + "--show-interactive-dev-session=false", + ], { cwd: process.cwd(), env: { ...process.env }, @@ -79,6 +121,7 @@ describe("E2E Relay with E2EE", () => { }); await waitForServer(relayPort, 30000); + await waitForRelayWebSocketReady(relayPort, 60000); }); afterAll(async () => { @@ -88,7 +131,7 @@ describe("E2E Relay with E2EE", () => { } }); - it("full flow: daemon and client exchange encrypted messages through relay", { timeout: 20_000 }, async () => { + it("full flow: daemon and client exchange encrypted messages through relay", { timeout: 90_000 }, async () => { const serverId = "test-session-" + Date.now(); const clientId = "clt_test_" + Date.now() + "_" + Math.random().toString(36).slice(2); @@ -255,7 +298,7 @@ describe("E2E Relay with E2EE", () => { clientWs.close(); }); - it("relay only sees opaque bytes after handshake", async () => { + it("relay only sees opaque bytes after handshake", { timeout: 90_000 }, async () => { const serverId = "opaque-test-" + Date.now(); const clientId = "clt_opaque_" + Date.now() + "_" + Math.random().toString(36).slice(2); diff --git a/packages/server/src/client/daemon-client.test.ts b/packages/server/src/client/daemon-client.test.ts index ee9a054ee..f7cb5aa44 100644 --- a/packages/server/src/client/daemon-client.test.ts +++ b/packages/server/src/client/daemon-client.test.ts @@ -1,6 +1,13 @@ import { readFileSync } from "node:fs"; import { afterEach, describe, expect, expectTypeOf, test, vi } from "vitest"; import { DaemonClient, type DaemonTransport } from "./daemon-client"; +import { + BinaryMuxChannel, + TerminalBinaryMessageType, + asUint8Array, + decodeBinaryMuxFrame, + encodeBinaryMuxFrame, +} from "../shared/binary-mux.js"; expectTypeOf<"getGitDiff" extends keyof DaemonClient ? true : false>().toEqualTypeOf(); expectTypeOf<"getHighlightedDiff" extends keyof DaemonClient ? true : false>().toEqualTypeOf(); @@ -15,7 +22,7 @@ function createMockLogger() { } function createMockTransport() { - const sent: string[] = []; + const sent: Array = []; let onMessage: (data: unknown) => void = () => {}; let onOpen: () => void = () => {}; @@ -619,6 +626,187 @@ describe("DaemonClient", () => { }); }); + test("auto-acks terminal stream chunks after delivery", async () => { + const logger = createMockLogger(); + const mock = createMockTransport(); + + const client = new DaemonClient({ + url: "ws://test", + logger, + reconnect: { enabled: false }, + transportFactory: () => mock.transport, + }); + clients.push(client); + + const connectPromise = client.connect(); + mock.triggerOpen(); + await connectPromise; + + const seen: string[] = []; + const decoder = new TextDecoder(); + const unsubscribe = client.onTerminalStreamData(7, (chunk) => { + seen.push(decoder.decode(chunk.data)); + }); + + const payload = new TextEncoder().encode("hello"); + mock.triggerMessage( + encodeBinaryMuxFrame({ + channel: BinaryMuxChannel.Terminal, + messageType: TerminalBinaryMessageType.OutputUtf8, + streamId: 7, + offset: 10, + payload, + }) + ); + + expect(seen).toEqual(["hello"]); + expect(mock.sent).toHaveLength(1); + const ackBytes = asUint8Array(mock.sent[0]); + expect(ackBytes).not.toBeNull(); + const ackFrame = decodeBinaryMuxFrame(ackBytes!); + expect(ackFrame).not.toBeNull(); + expect(ackFrame!.channel).toBe(BinaryMuxChannel.Terminal); + expect(ackFrame!.messageType).toBe(TerminalBinaryMessageType.Ack); + expect(ackFrame!.streamId).toBe(7); + expect(ackFrame!.offset).toBe(15); + unsubscribe(); + }); + + test("acks buffered terminal chunks when handler is attached", async () => { + const logger = createMockLogger(); + const mock = createMockTransport(); + + const client = new DaemonClient({ + url: "ws://test", + logger, + reconnect: { enabled: false }, + transportFactory: () => mock.transport, + }); + clients.push(client); + + const connectPromise = client.connect(); + mock.triggerOpen(); + await connectPromise; + + mock.triggerMessage( + encodeBinaryMuxFrame({ + channel: BinaryMuxChannel.Terminal, + messageType: TerminalBinaryMessageType.OutputUtf8, + streamId: 19, + offset: 0, + payload: new TextEncoder().encode("buffered"), + }) + ); + + expect(mock.sent).toHaveLength(0); + + const seen: string[] = []; + client.onTerminalStreamData(19, (chunk) => { + seen.push(new TextDecoder().decode(chunk.data)); + }); + + expect(seen.join("")).toContain("buffered"); + expect(mock.sent).toHaveLength(1); + const ackFrame = decodeBinaryMuxFrame(asUint8Array(mock.sent[0])!); + expect(ackFrame?.messageType).toBe(TerminalBinaryMessageType.Ack); + expect(ackFrame?.streamId).toBe(19); + expect(ackFrame?.offset).toBe(8); + }); + + test("cleans terminal stream local state on terminal_stream_exit", async () => { + const logger = createMockLogger(); + const mock = createMockTransport(); + + const client = new DaemonClient({ + url: "ws://test", + logger, + reconnect: { enabled: false }, + transportFactory: () => mock.transport, + }); + clients.push(client); + + const connectPromise = client.connect(); + mock.triggerOpen(); + await connectPromise; + + const unsubscribe = client.onTerminalStreamData(23, () => { + // no-op + }); + const internal = client as unknown as { + terminalStreamHandlers: Map void>>; + bufferedTerminalStreamChunks: Map; + terminalStreamAckOffsets: Map; + }; + internal.bufferedTerminalStreamChunks.set(23, { chunks: [], bytes: 0 }); + internal.terminalStreamAckOffsets.set(23, 100); + + mock.triggerMessage( + wrapSessionMessage({ + type: "terminal_stream_exit", + payload: { + streamId: 23, + terminalId: "term-1", + }, + }) + ); + + expect(internal.terminalStreamHandlers.has(23)).toBe(false); + expect(internal.bufferedTerminalStreamChunks.has(23)).toBe(false); + expect(internal.terminalStreamAckOffsets.has(23)).toBe(false); + unsubscribe(); + }); + + test("cleans local stream state even when detach reports success=false", async () => { + const logger = createMockLogger(); + const mock = createMockTransport(); + + const client = new DaemonClient({ + url: "ws://test", + logger, + reconnect: { enabled: false }, + transportFactory: () => mock.transport, + }); + clients.push(client); + + const connectPromise = client.connect(); + mock.triggerOpen(); + await connectPromise; + + client.onTerminalStreamData(31, () => { + // no-op + }); + const internal = client as unknown as { + bufferedTerminalStreamChunks: Map; + terminalStreamAckOffsets: Map; + }; + internal.bufferedTerminalStreamChunks.set(31, { chunks: [], bytes: 0 }); + internal.terminalStreamAckOffsets.set(31, 55); + + const detachPromise = client.detachTerminalStream(31, "detach-31"); + const detachRequest = JSON.parse(String(mock.sent[0])) as { + type: "session"; + message: { type: string; streamId: number; requestId: string }; + }; + expect(detachRequest.message.type).toBe("detach_terminal_stream_request"); + expect(detachRequest.message.streamId).toBe(31); + + mock.triggerMessage( + wrapSessionMessage({ + type: "detach_terminal_stream_response", + payload: { + streamId: 31, + success: false, + requestId: "detach-31", + }, + }) + ); + + const payload = await detachPromise; + expect(payload.success).toBe(false); + expect(internal.bufferedTerminalStreamChunks.has(31)).toBe(false); + expect(internal.terminalStreamAckOffsets.has(31)).toBe(false); + }); + test("parses canonical agent_stream tool_call payloads without crashing", async () => { const logger = createMockLogger(); const mock = createMockTransport(); diff --git a/packages/server/src/client/daemon-client.ts b/packages/server/src/client/daemon-client.ts index 39355f94c..638131cda 100644 --- a/packages/server/src/client/daemon-client.ts +++ b/packages/server/src/client/daemon-client.ts @@ -39,6 +39,8 @@ import type { SubscribeTerminalResponse, TerminalOutput, KillTerminalResponse, + AttachTerminalStreamResponse, + DetachTerminalStreamResponse, TerminalInput, SessionInboundMessage, SessionOutboundMessage, @@ -57,6 +59,19 @@ import { type Transport as RelayTransport, } from "@getpaseo/relay/e2ee"; import { isRelayClientWebSocketUrl } from "../shared/daemon-endpoints.js"; +import { + asUint8Array, + decodeBinaryMuxFrame, + encodeBinaryMuxFrame, + BinaryMuxChannel, + TerminalBinaryFlags, + TerminalBinaryMessageType, + type BinaryMuxFrame, +} from "../shared/binary-mux.js"; +import { + encodeTerminalKeyInput, + type TerminalKeyInput, +} from "../shared/terminal-key-input.js"; export interface Logger { debug(obj: object, msg?: string): void; @@ -73,7 +88,7 @@ const consoleLogger: Logger = { }; export type DaemonTransport = { - send: (data: string) => void; + send: (data: string | Uint8Array | ArrayBuffer) => void; close: (code?: number, reason?: string) => void; onMessage: (handler: (data: unknown) => void) => () => void; onOpen: (handler: () => void) => () => void; @@ -93,8 +108,9 @@ export type WebSocketFactory = ( export type WebSocketLike = { readyState: number; - send: (data: string) => void; + send: (data: string | Uint8Array | ArrayBuffer) => void; close: (code?: number, reason?: string) => void; + binaryType?: string; on?: (event: string, listener: (...args: any[]) => void) => void; off?: (event: string, listener: (...args: any[]) => void) => void; removeListener?: (event: string, listener: (...args: any[]) => void) => void; @@ -217,6 +233,24 @@ type CreateTerminalPayload = CreateTerminalResponse["payload"]; type SubscribeTerminalPayload = SubscribeTerminalResponse["payload"]; type TerminalOutputPayload = TerminalOutput["payload"]; type KillTerminalPayload = KillTerminalResponse["payload"]; +type AttachTerminalStreamPayload = AttachTerminalStreamResponse["payload"]; +type DetachTerminalStreamPayload = DetachTerminalStreamResponse["payload"]; + +export type TerminalStreamChunk = { + streamId: number; + offset: number; + endOffset: number; + replay: boolean; + data: Uint8Array; +}; + +type BufferedTerminalStreamQueue = { + chunks: TerminalStreamChunk[]; + bytes: number; +}; + +const TERMINAL_STREAM_MAX_BUFFERED_CHUNKS = 2048; +const TERMINAL_STREAM_MAX_BUFFERED_BYTES = 2 * 1024 * 1024; type AgentRefreshedStatusPayload = z.infer< typeof AgentRefreshedStatusPayloadSchema @@ -321,6 +355,12 @@ export class DaemonClient { private logger: Logger; private pendingSendQueue: PendingSend[] = []; private relayClientId: string | null = null; + private terminalStreamHandlers: Map< + number, + Set<(chunk: TerminalStreamChunk) => void> + > = new Map(); + private bufferedTerminalStreamChunks: Map = new Map(); + private terminalStreamAckOffsets: Map = new Map(); constructor(private config: DaemonClientConfig) { this.logger = config.logger ?? consoleLogger; @@ -538,6 +578,9 @@ export class DaemonClient { } this.disposeTransport(1000, "Client closed"); this.clearWaiters(new Error("Daemon client closed")); + this.terminalStreamHandlers.clear(); + this.bufferedTerminalStreamChunks.clear(); + this.terminalStreamAckOffsets.clear(); this.updateConnectionState({ status: "disconnected", reason: "client_closed", @@ -656,6 +699,23 @@ export class DaemonClient { } } + private sendBinaryFrame(frame: BinaryMuxFrame): void { + if (!this.transport || this.connectionState.status !== "connected") { + if (this.config.suppressSendErrors) { + return; + } + throw new Error(`Transport not connected (status: ${this.connectionState.status})`); + } + try { + this.transport.send(encodeBinaryMuxFrame(frame)); + } catch (error) { + if (this.config.suppressSendErrors) { + return; + } + throw error instanceof Error ? error : new Error(String(error)); + } + } + /** * Send a session message for RPC methods that create waiters. * If the connection is still being established ("connecting"), the message @@ -2569,6 +2629,159 @@ export class DaemonClient { }); } + async attachTerminalStream( + terminalId: string, + options?: { + resumeOffset?: number; + rows?: number; + cols?: number; + }, + requestId?: string + ): Promise { + const resolvedRequestId = this.createRequestId(requestId); + const message = SessionInboundMessageSchema.parse({ + type: "attach_terminal_stream_request", + terminalId, + requestId: resolvedRequestId, + ...(options?.resumeOffset !== undefined ? { resumeOffset: options.resumeOffset } : {}), + ...(options?.rows !== undefined ? { rows: options.rows } : {}), + ...(options?.cols !== undefined ? { cols: options.cols } : {}), + }); + return this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 10000, + options: { skipQueue: true }, + select: (msg) => { + if (msg.type !== "attach_terminal_stream_response") { + return null; + } + if (msg.payload.requestId !== resolvedRequestId) { + return null; + } + return msg.payload; + }, + }); + } + + async detachTerminalStream( + streamId: number, + requestId?: string + ): Promise { + const resolvedRequestId = this.createRequestId(requestId); + const message = SessionInboundMessageSchema.parse({ + type: "detach_terminal_stream_request", + streamId, + requestId: resolvedRequestId, + }); + const payload = await this.sendRequest({ + requestId: resolvedRequestId, + message, + timeout: 10000, + options: { skipQueue: true }, + select: (msg) => { + if (msg.type !== "detach_terminal_stream_response") { + return null; + } + if (msg.payload.requestId !== resolvedRequestId) { + return null; + } + return msg.payload; + }, + }); + this.terminalStreamHandlers.delete(streamId); + this.bufferedTerminalStreamChunks.delete(streamId); + this.terminalStreamAckOffsets.delete(streamId); + return payload; + } + + onTerminalStreamData( + streamId: number, + handler: (chunk: TerminalStreamChunk) => void + ): () => void { + if (!this.terminalStreamHandlers.has(streamId)) { + this.terminalStreamHandlers.set(streamId, new Set()); + } + const handlers = this.terminalStreamHandlers.get(streamId)!; + handlers.add(handler); + + const buffered = this.bufferedTerminalStreamChunks.get(streamId); + if (buffered && buffered.chunks.length > 0) { + for (const chunk of buffered.chunks) { + try { + handler(chunk); + this.maybeAckTerminalStreamChunk(streamId, chunk.endOffset); + } catch { + // no-op + } + } + this.bufferedTerminalStreamChunks.delete(streamId); + } + + return () => { + handlers.delete(handler); + if (handlers.size === 0) { + this.terminalStreamHandlers.delete(streamId); + } + }; + } + + async waitForTerminalStreamData( + streamId: number, + predicate: (chunk: TerminalStreamChunk) => boolean, + timeout = 5000 + ): Promise { + return new Promise((resolve, reject) => { + const timeoutHandle = setTimeout(() => { + unsubscribe(); + reject(new Error(`Timeout waiting for terminal stream data (${timeout}ms)`)); + }, timeout); + + const unsubscribe = this.onTerminalStreamData(streamId, (chunk) => { + if (!predicate(chunk)) { + return; + } + clearTimeout(timeoutHandle); + unsubscribe(); + resolve(chunk); + }); + }); + } + + sendTerminalStreamInput(streamId: number, data: string | Uint8Array): void { + const payload = typeof data === "string" ? encodeUtf8String(data) : data; + this.sendBinaryFrame({ + channel: BinaryMuxChannel.Terminal, + messageType: TerminalBinaryMessageType.InputUtf8, + streamId, + offset: 0, + payload, + }); + } + + sendTerminalStreamKey(streamId: number, input: TerminalKeyInput): void { + const encoded = encodeTerminalKeyInput(input); + if (!encoded) { + return; + } + this.sendTerminalStreamInput(streamId, encoded); + } + + sendTerminalStreamAck(streamId: number, offset: number): void { + const normalizedOffset = Math.max(0, Math.floor(offset)); + const previousAck = this.terminalStreamAckOffsets.get(streamId) ?? -1; + if (normalizedOffset > previousAck) { + this.terminalStreamAckOffsets.set(streamId, normalizedOffset); + } + this.sendBinaryFrame({ + channel: BinaryMuxChannel.Terminal, + messageType: TerminalBinaryMessageType.Ack, + streamId, + offset: normalizedOffset, + payload: new Uint8Array(0), + }); + } + async waitForTerminalOutput( terminalId: string, timeout = 5000 @@ -2628,6 +2841,14 @@ export class DaemonClient { data && typeof data === "object" && "data" in data ? (data as { data: unknown }).data : data; + const rawBytes = asUint8Array(rawData); + if (rawBytes) { + const frame = decodeBinaryMuxFrame(rawBytes); + if (frame) { + this.handleBinaryFrame(frame); + return; + } + } const payload = decodeMessageData(rawData); if (!payload) { return; @@ -2654,6 +2875,75 @@ export class DaemonClient { this.handleSessionMessage(parsed.data.message); } + private handleBinaryFrame(frame: BinaryMuxFrame): void { + if ( + frame.channel === BinaryMuxChannel.Terminal && + frame.messageType === TerminalBinaryMessageType.OutputUtf8 + ) { + const handlers = this.terminalStreamHandlers.get(frame.streamId); + const chunk: TerminalStreamChunk = { + streamId: frame.streamId, + offset: frame.offset, + endOffset: frame.offset + (frame.payload?.byteLength ?? 0), + replay: Boolean((frame.flags ?? 0) & TerminalBinaryFlags.Replay), + data: frame.payload ?? new Uint8Array(0), + }; + if (!handlers || handlers.size === 0) { + const queue = this.bufferedTerminalStreamChunks.get(frame.streamId) ?? { + chunks: [], + bytes: 0, + }; + queue.chunks.push(chunk); + queue.bytes += chunk.data.byteLength; + while ( + queue.chunks.length > TERMINAL_STREAM_MAX_BUFFERED_CHUNKS || + queue.bytes > TERMINAL_STREAM_MAX_BUFFERED_BYTES + ) { + const removed = queue.chunks.shift(); + if (!removed) { + break; + } + queue.bytes -= removed.data.byteLength; + if (queue.bytes < 0) { + queue.bytes = 0; + } + } + this.bufferedTerminalStreamChunks.set(frame.streamId, queue); + return; + } + let delivered = false; + for (const handler of handlers) { + try { + handler(chunk); + delivered = true; + } catch { + // no-op + } + } + if (delivered) { + this.maybeAckTerminalStreamChunk(frame.streamId, chunk.endOffset); + } + return; + } + } + + private maybeAckTerminalStreamChunk(streamId: number, endOffset: number): void { + if (!Number.isFinite(endOffset) || endOffset < 0) { + return; + } + const normalizedEndOffset = Math.floor(endOffset); + const previousAck = this.terminalStreamAckOffsets.get(streamId) ?? -1; + if (normalizedEndOffset <= previousAck) { + return; + } + this.terminalStreamAckOffsets.set(streamId, normalizedEndOffset); + try { + this.sendTerminalStreamAck(streamId, normalizedEndOffset); + } catch { + // no-op + } + } + private updateConnectionState(next: ConnectionState): void { this.connectionState = next; for (const listener of this.connectionListeners) { @@ -2694,6 +2984,9 @@ 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.terminalStreamHandlers.clear(); + this.bufferedTerminalStreamChunks.clear(); + this.terminalStreamAckOffsets.clear(); this.updateConnectionState({ status: "disconnected", @@ -2709,6 +3002,12 @@ export class DaemonClient { } private handleSessionMessage(msg: SessionOutboundMessage): void { + if (msg.type === "terminal_stream_exit") { + this.terminalStreamHandlers.delete(msg.payload.streamId); + this.bufferedTerminalStreamChunks.delete(msg.payload.streamId); + this.terminalStreamAckOffsets.delete(msg.payload.streamId); + } + if (this.rawMessageListeners.size > 0) { for (const handler of this.rawMessageListeners) { try { @@ -2904,6 +3203,13 @@ function createWebSocketTransportFactory( ): DaemonTransportFactory { return ({ url, headers }) => { const ws = factory(url, { headers }); + if ("binaryType" in ws) { + try { + ws.binaryType = "arraybuffer"; + } catch { + // no-op + } + } return { send: (data) => { if (typeof ws.readyState === "number" && ws.readyState !== 1) { @@ -2997,16 +3303,23 @@ function createEncryptedTransport( base.send(data); return; } + if (ArrayBuffer.isView(data)) { + const view = new Uint8Array(data.buffer, data.byteOffset, data.byteLength); + const out = new Uint8Array(view.byteLength); + out.set(view); + base.send(out.buffer); + return; + } if (data instanceof ArrayBuffer) { if (typeof TextDecoder !== "undefined") { - base.send(new TextDecoder().decode(data)); + base.send(data); return; } if (typeof Buffer !== "undefined") { - base.send(Buffer.from(data).toString("utf8")); + base.send(data); return; } - base.send(String(data)); + base.send(data); return; } base.send(String(data)); @@ -3055,7 +3368,20 @@ function createEncryptedTransport( if (!channel) { throw new Error("Encrypted channel not ready"); } - void channel.send(data).catch((error) => { + const outbound = + typeof data === "string" + ? data + : data instanceof ArrayBuffer + ? data + : ArrayBuffer.isView(data) + ? (() => { + const view = new Uint8Array(data.buffer, data.byteOffset, data.byteLength); + const out = new Uint8Array(view.byteLength); + out.set(view); + return out.buffer; + })() + : String(data); + void channel.send(outbound).catch((error) => { emitError(error); }); }, @@ -3240,6 +3566,20 @@ function decodeMessageData(data: unknown): string | null { return null; } +function encodeUtf8String(value: string): Uint8Array { + if (typeof TextEncoder !== "undefined") { + return new TextEncoder().encode(value); + } + if (typeof Buffer !== "undefined") { + return new Uint8Array(Buffer.from(value, "utf8")); + } + const out = new Uint8Array(value.length); + for (let i = 0; i < value.length; i++) { + out[i] = value.charCodeAt(i) & 0xff; + } + return out; +} + function resolveAgentConfig(options: CreateAgentRequestOptions): AgentSessionConfig { const { config, diff --git a/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts b/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts index 258349559..1ec3ec1ae 100644 --- a/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts @@ -11,6 +11,9 @@ import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js"; import { createClientChannel, type Transport } from "@getpaseo/relay/e2ee"; import { buildRelayWebSocketUrl } from "../../shared/daemon-endpoints.js"; +const nodeMajor = Number((process.versions.node ?? "0").split(".")[0] ?? "0"); +const shouldRunRelayE2e = process.env.FORCE_RELAY_E2E === "1" || nodeMajor < 25; + function createCapturingLogger() { const lines: string[] = []; const stream = new Writable({ @@ -90,7 +93,40 @@ async function waitForServer(port: number, timeout = 15000): Promise { throw new Error(`Server did not start on port ${port} within ${timeout}ms`); } -describe("Relay transport (E2EE) - daemon E2E", () => { +async function waitForRelayWebSocketReady(port: number, timeout = 60000): Promise { + const start = Date.now(); + while (Date.now() - start < timeout) { + const serverId = `probe-${Date.now().toString(36)}-${Math.random().toString(36).slice(2)}`; + const url = buildRelayWebSocketUrl({ + endpoint: `127.0.0.1:${port}`, + serverId, + role: "server", + }); + const opened = await new Promise((resolve) => { + const ws = new WebSocket(url); + const timer = setTimeout(() => { + ws.terminate(); + resolve(false); + }, 5000); + ws.once("open", () => { + clearTimeout(timer); + ws.close(1000, "probe"); + resolve(true); + }); + ws.once("error", () => { + clearTimeout(timer); + resolve(false); + }); + }); + if (opened) { + return; + } + await new Promise((r) => setTimeout(r, 250)); + } + throw new Error(`Relay WebSocket endpoint not ready on port ${port} within ${timeout}ms`); +} + +(shouldRunRelayE2e ? describe : describe.skip)("Relay transport (E2EE) - daemon E2E", () => { let relayPort: number; let relayProcess: ChildProcess | null = null; @@ -99,7 +135,17 @@ describe("Relay transport (E2EE) - daemon E2E", () => { const relayDir = path.resolve(process.cwd(), "../relay"); relayProcess = spawn( "npx", - ["wrangler", "dev", "--local", "--ip", "127.0.0.1", "--port", String(relayPort)], + [ + "wrangler", + "dev", + "--local", + "--ip", + "127.0.0.1", + "--port", + String(relayPort), + "--live-reload=false", + "--show-interactive-dev-session=false", + ], { cwd: relayDir, env: { ...process.env }, @@ -124,6 +170,7 @@ describe("Relay transport (E2EE) - daemon E2E", () => { }); await waitForServer(relayPort, 30000); + await waitForRelayWebSocketReady(relayPort, 60000); }; const stopRelay = async () => { @@ -233,7 +280,7 @@ describe("Relay transport (E2EE) - daemon E2E", () => { await stopRelay(); } }, - 30000 + 90000 ); test( @@ -332,6 +379,6 @@ describe("Relay transport (E2EE) - daemon E2E", () => { await stopRelay(); } }, - 45000 + 90000 ); }); 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 10d307091..063b5426e 100644 --- a/packages/server/src/server/daemon-e2e/terminal.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/terminal.e2e.test.ts @@ -2,15 +2,46 @@ import { describe, test, expect, beforeEach, afterEach } from "vitest"; import { mkdtempSync, rmSync } from "fs"; import { tmpdir } from "os"; import path from "path"; +import WebSocket from "ws"; import { createDaemonTestContext, type DaemonTestContext, } from "../test-utils/index.js"; +import { + BinaryMuxChannel, + TerminalBinaryMessageType, + decodeBinaryMuxFrame, + encodeBinaryMuxFrame, +} from "../../shared/binary-mux.js"; + +const decoder = new TextDecoder(); function tmpCwd(): string { return mkdtempSync(path.join(tmpdir(), "daemon-terminal-e2e-")); } +async function waitForCondition( + predicate: () => boolean, + timeoutMs: number, + intervalMs = 25 +): Promise { + const start = Date.now(); + while (Date.now() - start < timeoutMs) { + if (predicate()) { + return; + } + await new Promise((resolve) => setTimeout(resolve, intervalMs)); + } + throw new Error(`Timed out after ${timeoutMs}ms waiting for condition`); +} + +function percentile(values: number[], p: number): number { + if (values.length === 0) return 0; + const sorted = [...values].sort((a, b) => a - b); + const index = Math.min(sorted.length - 1, Math.ceil((p / 100) * sorted.length) - 1); + return sorted[index]; +} + const shouldRun = !process.env.CI; (shouldRun ? describe : describe.skip)("daemon E2E terminal", () => { @@ -246,4 +277,286 @@ const shouldRun = !process.env.CI; }, 30000 ); + + test( + "streams terminal output over binary mux and supports modifier keys", + async () => { + const cwd = tmpCwd(); + const list = await ctx.client.listTerminals(cwd); + const terminalId = list.terminals[0].id; + + const attach = await ctx.client.attachTerminalStream(terminalId, { rows: 24, cols: 80 }); + expect(attach.error).toBeNull(); + expect(attach.streamId).toBeTypeOf("number"); + const streamId = attach.streamId!; + + let text = ""; + const unsubscribe = ctx.client.onTerminalStreamData(streamId, (chunk) => { + text += decoder.decode(chunk.data, { stream: true }); + }); + + ctx.client.sendTerminalStreamInput(streamId, "echo binary-stream\r"); + await waitForCondition(() => text.includes("binary-stream"), 10000); + + ctx.client.sendTerminalStreamInput(streamId, "cat -v\r"); + await waitForCondition(() => text.includes("cat -v"), 10000); + + // Ctrl+B is the default tmux prefix; cat -v should render it as ^B. + ctx.client.sendTerminalStreamKey(streamId, { key: "b", ctrl: true }); + ctx.client.sendTerminalStreamInput(streamId, "\r"); + await waitForCondition(() => text.includes("^B"), 10000); + + // Stop cat + ctx.client.sendTerminalStreamKey(streamId, { key: "c", ctrl: true }); + + const detach = await ctx.client.detachTerminalStream(streamId); + expect(detach.success).toBe(true); + unsubscribe(); + rmSync(cwd, { recursive: true, force: true }); + }, + 30000 + ); + + test( + "replays detached terminal output from resume offset (scrollback continuity)", + async () => { + const cwd = tmpCwd(); + const list = await ctx.client.listTerminals(cwd); + const terminalId = list.terminals[0].id; + + const attach = await ctx.client.attachTerminalStream(terminalId, { rows: 24, cols: 80 }); + expect(attach.error).toBeNull(); + const streamId = attach.streamId!; + + let output = ""; + let latestOffset = attach.currentOffset; + const unsubscribe = ctx.client.onTerminalStreamData(streamId, (chunk) => { + output += decoder.decode(chunk.data, { stream: true }); + latestOffset = chunk.endOffset; + }); + + ctx.client.sendTerminalStreamInput(streamId, "echo before-detach\r"); + await waitForCondition(() => output.includes("before-detach"), 10000); + + const firstDetach = await ctx.client.detachTerminalStream(streamId); + expect(firstDetach.success).toBe(true); + unsubscribe(); + + // Terminal keeps running while hidden, stream is detached. + ctx.client.sendTerminalInput(terminalId, { type: "input", data: "echo while-detached\r" }); + await new Promise((resolve) => setTimeout(resolve, 300)); + + const resumed = await ctx.client.attachTerminalStream(terminalId, { + resumeOffset: latestOffset, + }); + expect(resumed.error).toBeNull(); + expect(resumed.streamId).toBeTypeOf("number"); + expect(resumed.reset).toBe(false); + + let replayedText = ""; + const resumedUnsub = ctx.client.onTerminalStreamData(resumed.streamId!, (chunk) => { + if (chunk.replay) { + replayedText += decoder.decode(chunk.data, { stream: true }); + } + }); + + await waitForCondition(() => replayedText.includes("while-detached"), 10000); + + const secondDetach = await ctx.client.detachTerminalStream(resumed.streamId!); + expect(secondDetach.success).toBe(true); + resumedUnsub(); + rmSync(cwd, { recursive: true, force: true }); + }, + 30000 + ); + + test( + "applies stream backpressure window until client ack advances", + async () => { + const cwd = tmpCwd(); + const list = await ctx.client.listTerminals(cwd); + const terminalId = list.terminals[0].id; + + const ws = new WebSocket(`ws://127.0.0.1:${ctx.daemon.port}/ws`); + await new Promise((resolve, reject) => { + ws.once("open", () => resolve()); + ws.once("error", reject); + }); + + const attachRequestId = `attach-${Date.now()}`; + const detachRequestId = `detach-${Date.now()}`; + let streamId: number | null = null; + let latestEndOffset = 0; + let outputBytes = 0; + + const streamReady = new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + reject(new Error("Timed out waiting for attach_terminal_stream_response")); + }, 10000); + + ws.on("message", (raw) => { + if (Array.isArray(raw)) { + raw = Buffer.concat(raw.map((part) => (Buffer.isBuffer(part) ? part : Buffer.from(part)))); + } + if (typeof raw !== "string" && !Buffer.isBuffer(raw)) { + return; + } + const text = typeof raw === "string" ? raw : raw.toString("utf8"); + try { + const parsed = JSON.parse(text) as { + type?: string; + message?: { + type?: string; + payload?: { streamId?: number | null; requestId?: string }; + }; + }; + if ( + parsed.type === "session" && + parsed.message?.type === "attach_terminal_stream_response" && + parsed.message.payload?.requestId === attachRequestId + ) { + const nextStreamId = parsed.message.payload.streamId; + if (typeof nextStreamId === "number") { + streamId = nextStreamId; + clearTimeout(timeout); + resolve(); + } + } + } catch { + // Ignore non-JSON payloads (binary mux frames). + } + }); + }); + + ws.on("message", (raw) => { + if (typeof raw === "string") { + return; + } + if (Array.isArray(raw)) { + raw = Buffer.concat(raw.map((part) => (Buffer.isBuffer(part) ? part : Buffer.from(part)))); + } + const bytes = Buffer.isBuffer(raw) + ? new Uint8Array(raw.buffer, raw.byteOffset, raw.byteLength) + : new Uint8Array(raw as ArrayBuffer); + const frame = decodeBinaryMuxFrame(bytes); + if ( + !frame || + frame.channel !== BinaryMuxChannel.Terminal || + frame.messageType !== TerminalBinaryMessageType.OutputUtf8 + ) { + return; + } + const chunkBytes = frame.payload?.byteLength ?? 0; + outputBytes += chunkBytes; + latestEndOffset = frame.offset + chunkBytes; + }); + + ws.send( + JSON.stringify({ + type: "session", + message: { + type: "attach_terminal_stream_request", + terminalId, + requestId: attachRequestId, + }, + }) + ); + await streamReady; + expect(streamId).toBeTypeOf("number"); + + ws.send( + JSON.stringify({ + type: "session", + message: { + type: "terminal_input", + terminalId, + message: { + type: "input", + data: "head -c 1048576 /dev/zero | tr '\\0' 'A'\r", + }, + }, + }) + ); + + await waitForCondition(() => outputBytes > 0, 10000); + await new Promise((resolve) => setTimeout(resolve, 800)); + const beforeAckBytes = outputBytes; + expect(beforeAckBytes).toBeGreaterThan(0); + expect(beforeAckBytes).toBeLessThan(320 * 1024); + expect(latestEndOffset).toBeGreaterThan(0); + + ws.send( + encodeBinaryMuxFrame({ + channel: BinaryMuxChannel.Terminal, + messageType: TerminalBinaryMessageType.Ack, + streamId: streamId!, + offset: latestEndOffset, + payload: new Uint8Array(0), + }) + ); + + await waitForCondition(() => outputBytes > beforeAckBytes, 10000); + + ws.send( + JSON.stringify({ + type: "session", + message: { + type: "detach_terminal_stream_request", + streamId, + requestId: detachRequestId, + }, + }) + ); + + ws.close(); + rmSync(cwd, { recursive: true, force: true }); + }, + 30000 + ); + + test( + "measures local terminal round-trip latency via daemon client stream", + async () => { + const cwd = tmpCwd(); + const list = await ctx.client.listTerminals(cwd); + const terminalId = list.terminals[0].id; + + const attach = await ctx.client.attachTerminalStream(terminalId); + expect(attach.error).toBeNull(); + const streamId = attach.streamId!; + + let output = ""; + const unsubscribe = ctx.client.onTerminalStreamData(streamId, (chunk) => { + output += decoder.decode(chunk.data, { stream: true }); + }); + + const samplesMs: number[] = []; + const iterations = 8; + for (let i = 0; i < iterations; i++) { + const marker = `PASEO_LAT_${i}_${Date.now()}`; + const start = performance.now(); + ctx.client.sendTerminalStreamInput(streamId, `echo ${marker}\r`); + await waitForCondition(() => output.includes(marker), 10000); + samplesMs.push(performance.now() - start); + } + + const p50 = percentile(samplesMs, 50); + const p95 = percentile(samplesMs, 95); + + expect(samplesMs).toHaveLength(iterations); + // Localhost budget should be tight; keep margin for test noise. + expect(p95).toBeLessThan(350); + + // Emit measurements for diagnosis when this test regresses. + console.log( + `[terminal-latency] samples=${samplesMs.map((n) => n.toFixed(1)).join(",")} p50=${p50.toFixed(1)}ms p95=${p95.toFixed(1)}ms` + ); + + const detach = await ctx.client.detachTerminalStream(streamId); + expect(detach.success).toBe(true); + unsubscribe(); + rmSync(cwd, { recursive: true, force: true }); + }, + 30000 + ); }); diff --git a/packages/server/src/server/relay-transport.ts b/packages/server/src/server/relay-transport.ts index 494942294..533bc5eb0 100644 --- a/packages/server/src/server/relay-transport.ts +++ b/packages/server/src/server/relay-transport.ts @@ -25,7 +25,7 @@ export type RelayTransportController = { type RelaySocketLike = { readyState: number; - send: (data: string) => void; + send: (data: string | Uint8Array | ArrayBuffer) => void; close: (code?: number, reason?: string) => void; on: (event: "message" | "close" | "error", listener: (...args: any[]) => void) => void; once: (event: "close" | "error", listener: (...args: any[]) => void) => void; @@ -436,7 +436,20 @@ function createEncryptedSocket( return readyState; }, send: (data) => { - void channel.send(data).catch((error) => { + const outbound = + typeof data === "string" + ? data + : data instanceof ArrayBuffer + ? data + : ArrayBuffer.isView(data) + ? (() => { + const view = new Uint8Array(data.buffer, data.byteOffset, data.byteLength); + const out = new Uint8Array(view.byteLength); + out.set(view); + return out.buffer; + })() + : String(data); + void channel.send(outbound).catch((error) => { emitter.emit("error", error); }); }, diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index d8dfb0f8f..cbfb467b8 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -20,12 +20,20 @@ import { type UnsubscribeTerminalRequest, type TerminalInput, type KillTerminalRequest, + type AttachTerminalStreamRequest, + type DetachTerminalStreamRequest, type SubscribeCheckoutDiffRequest, type UnsubscribeCheckoutDiffRequest, type ProjectCheckoutLitePayload, type ProjectPlacementPayload, } from "./messages.js"; import type { TerminalManager } from "../terminal/terminal-manager.js"; +import { + BinaryMuxChannel, + TerminalBinaryFlags, + TerminalBinaryMessageType, + type BinaryMuxFrame, +} from "../shared/binary-mux.js"; import { TTSManager } from "./agent/tts-manager.js"; import { STTManager } from "./agent/stt-manager.js"; import type { SpeechToTextProvider, TextToSpeechProvider } from "./speech/speech-provider.js"; @@ -137,6 +145,9 @@ const PROJECT_PLACEMENT_CACHE_TTL_MS = 10_000; const MAX_AGENTS_PER_PROJECT = 5; const CHECKOUT_DIFF_WATCH_DEBOUNCE_MS = 150; const CHECKOUT_DIFF_FALLBACK_REFRESH_MS = 5_000; +const TERMINAL_STREAM_WINDOW_BYTES = 256 * 1024; +const TERMINAL_STREAM_MAX_PENDING_BYTES = 2 * 1024 * 1024; +const TERMINAL_STREAM_MAX_PENDING_CHUNKS = 2048; /** * Default model used for auto-generating commit messages and PR descriptions. @@ -255,6 +266,13 @@ type CheckoutErrorPayload = { message: string; }; +type TerminalStreamPendingChunk = { + data: string; + startOffset: number; + endOffset: number; + replay: boolean; +}; + const PCM_SAMPLE_RATE = 16000; const PCM_CHANNELS = 1; const PCM_BITS_PER_SAMPLE = 16; @@ -297,6 +315,7 @@ type VoiceTranscriptionResultPayload = { export type SessionOptions = { clientId: string; onMessage: (msg: SessionOutboundMessage) => void; + onBinaryMessage?: (frame: BinaryMuxFrame) => void; logger: pino.Logger; downloadTokenStore: DownloadTokenStore; pushTokenStore: PushTokenStore; @@ -409,6 +428,7 @@ export class Session { private readonly clientId: string; private readonly sessionId: string; private readonly onMessage: (msg: SessionOutboundMessage) => void; + private readonly onBinaryMessage: ((frame: BinaryMuxFrame) => void) | null; private readonly sessionLogger: pino.Logger; private readonly paseoHome: string; @@ -483,6 +503,19 @@ export class Session { private readonly MOBILE_BACKGROUND_STREAM_GRACE_MS = 60_000; private readonly terminalManager: TerminalManager | null; private terminalSubscriptions: Map void> = new Map(); + private readonly terminalStreams = new Map< + number, + { + terminalId: string; + unsubscribe: () => void; + lastOutputOffset: number; + lastAckOffset: number; + pendingChunks: TerminalStreamPendingChunk[]; + pendingBytes: number; + } + >(); + private readonly terminalStreamByTerminalId = new Map(); + private nextTerminalStreamId = 1; private readonly checkoutDiffSubscriptions = new Map(); private readonly checkoutDiffTargets = new Map(); private readonly voiceAgentMcpStdio: VoiceMcpStdioConfig | null; @@ -508,6 +541,7 @@ export class Session { const { clientId, onMessage, + onBinaryMessage, logger, downloadTokenStore, pushTokenStore, @@ -526,6 +560,7 @@ export class Session { this.clientId = clientId; this.sessionId = uuidv4(); this.onMessage = onMessage; + this.onBinaryMessage = onBinaryMessage ?? null; this.downloadTokenStore = downloadTokenStore; this.pushTokenStore = pushTokenStore; this.paseoHome = paseoHome; @@ -1367,6 +1402,14 @@ export class Session { case "kill_terminal_request": await this.handleKillTerminalRequest(msg); break; + + case "attach_terminal_stream_request": + await this.handleAttachTerminalStreamRequest(msg); + break; + + case "detach_terminal_stream_request": + this.handleDetachTerminalStreamRequest(msg); + break; } } catch (error: any) { const err = error instanceof Error ? error : new Error(String(error)); @@ -1404,6 +1447,72 @@ export class Session { } } + public handleBinaryFrame(frame: BinaryMuxFrame): void { + switch (frame.channel) { + case BinaryMuxChannel.Terminal: + this.handleTerminalBinaryFrame(frame); + break; + default: + this.sessionLogger.warn( + { channel: frame.channel, messageType: frame.messageType }, + "Unhandled binary mux channel" + ); + break; + } + } + + private handleTerminalBinaryFrame(frame: BinaryMuxFrame): void { + if (frame.messageType === TerminalBinaryMessageType.InputUtf8) { + const binding = this.terminalStreams.get(frame.streamId); + if (!binding) { + this.sessionLogger.warn({ streamId: frame.streamId }, "Terminal stream not found for input"); + return; + } + if (!this.terminalManager) { + return; + } + const session = this.terminalManager.getTerminal(binding.terminalId); + if (!session) { + this.detachTerminalStream(frame.streamId, { emitExit: true }); + return; + } + + const payload = frame.payload ?? new Uint8Array(0); + if (payload.byteLength === 0) { + return; + } + const text = Buffer.from(payload).toString("utf8"); + if (!text) { + return; + } + session.send({ type: "input", data: text }); + return; + } + + if (frame.messageType === TerminalBinaryMessageType.Ack) { + const binding = this.terminalStreams.get(frame.streamId); + if (binding) { + if (!Number.isFinite(frame.offset) || frame.offset < 0) { + return; + } + const nextAckOffset = Math.max( + binding.lastAckOffset, + Math.min(Math.floor(frame.offset), binding.lastOutputOffset) + ); + if (nextAckOffset > binding.lastAckOffset) { + binding.lastAckOffset = nextAckOffset; + this.flushPendingTerminalStreamChunks(frame.streamId, binding); + } + } + return; + } + + this.sessionLogger.warn( + { streamId: frame.streamId, messageType: frame.messageType }, + "Unhandled terminal binary frame" + ); + } + private async handleRestartServerRequest( requestId: string, reason?: string @@ -5642,6 +5751,17 @@ export class Session { this.onMessage(msg); } + private emitBinary(frame: BinaryMuxFrame): void { + if (!this.onBinaryMessage) { + return; + } + try { + this.onBinaryMessage(frame); + } catch (error) { + this.sessionLogger.error({ err: error }, "Failed to emit binary frame"); + } + } + /** * Clean up session resources */ @@ -5694,6 +5814,7 @@ export class Session { unsubscribe(); } this.terminalSubscriptions.clear(); + this.detachAllTerminalStreams({ emitExit: false }); for (const target of this.checkoutDiffTargets.values()) { this.closeCheckoutDiffWatchTarget(target); @@ -5883,6 +6004,11 @@ export class Session { this.terminalSubscriptions.delete(msg.terminalId); } + const streamId = this.terminalStreamByTerminalId.get(msg.terminalId); + if (typeof streamId === "number") { + this.detachTerminalStream(streamId, { emitExit: true }); + } + this.terminalManager.killTerminal(msg.terminalId); this.emit({ type: "kill_terminal_response", @@ -5893,4 +6019,279 @@ export class Session { }, }); } + + private async handleAttachTerminalStreamRequest( + msg: AttachTerminalStreamRequest + ): Promise { + if (!this.terminalManager || !this.onBinaryMessage) { + this.emit({ + type: "attach_terminal_stream_response", + payload: { + terminalId: msg.terminalId, + streamId: null, + replayedFrom: 0, + currentOffset: 0, + earliestAvailableOffset: 0, + reset: true, + error: "Terminal streaming not available", + requestId: msg.requestId, + }, + }); + return; + } + + const session = this.terminalManager.getTerminal(msg.terminalId); + if (!session) { + this.emit({ + type: "attach_terminal_stream_response", + payload: { + terminalId: msg.terminalId, + streamId: null, + replayedFrom: 0, + currentOffset: 0, + earliestAvailableOffset: 0, + reset: true, + error: "Terminal not found", + requestId: msg.requestId, + }, + }); + return; + } + + if (msg.rows || msg.cols) { + const state = session.getState(); + session.send({ + type: "resize", + rows: msg.rows ?? state.rows, + cols: msg.cols ?? state.cols, + }); + } + + const existingStreamId = this.terminalStreamByTerminalId.get(msg.terminalId); + if (typeof existingStreamId === "number") { + this.detachTerminalStream(existingStreamId, { emitExit: false }); + } + + const streamId = this.allocateTerminalStreamId(); + const initialOffset = Math.max(0, Math.floor(msg.resumeOffset ?? 0)); + const binding: { + terminalId: string; + unsubscribe: () => void; + lastOutputOffset: number; + lastAckOffset: number; + pendingChunks: TerminalStreamPendingChunk[]; + pendingBytes: number; + } = { + terminalId: msg.terminalId, + unsubscribe: () => {}, + lastOutputOffset: initialOffset, + lastAckOffset: initialOffset, + pendingChunks: [], + pendingBytes: 0, + }; + this.terminalStreams.set(streamId, binding); + this.terminalStreamByTerminalId.set(msg.terminalId, streamId); + + let rawSub; + try { + rawSub = session.subscribeRaw( + (chunk) => { + const currentBinding = this.terminalStreams.get(streamId); + if (!currentBinding) { + return; + } + this.enqueueOrEmitTerminalStreamChunk(streamId, currentBinding, { + data: chunk.data, + startOffset: chunk.startOffset, + endOffset: chunk.endOffset, + replay: chunk.replay, + }); + }, + { fromOffset: msg.resumeOffset ?? 0 } + ); + } catch (error) { + this.terminalStreams.delete(streamId); + this.terminalStreamByTerminalId.delete(msg.terminalId); + throw error; + } + + binding.unsubscribe = rawSub.unsubscribe; + binding.lastAckOffset = rawSub.replayedFrom; + if (binding.lastOutputOffset < rawSub.replayedFrom) { + binding.lastOutputOffset = rawSub.replayedFrom; + } + this.flushPendingTerminalStreamChunks(streamId, binding); + + this.emit({ + type: "attach_terminal_stream_response", + payload: { + terminalId: msg.terminalId, + streamId, + replayedFrom: rawSub.replayedFrom, + currentOffset: rawSub.currentOffset, + earliestAvailableOffset: rawSub.earliestAvailableOffset, + reset: rawSub.reset, + error: null, + requestId: msg.requestId, + }, + }); + } + + private getTerminalStreamChunkByteLength(chunk: TerminalStreamPendingChunk): number { + return Math.max(0, chunk.endOffset - chunk.startOffset); + } + + private canEmitTerminalStreamChunk( + binding: { + lastAckOffset: number; + }, + chunk: TerminalStreamPendingChunk + ): boolean { + return chunk.startOffset < binding.lastAckOffset + TERMINAL_STREAM_WINDOW_BYTES; + } + + private emitTerminalStreamChunk( + streamId: number, + binding: { + lastOutputOffset: number; + }, + chunk: TerminalStreamPendingChunk + ): void { + const payload = new Uint8Array(Buffer.from(chunk.data, "utf8")); + this.emitBinary({ + channel: BinaryMuxChannel.Terminal, + messageType: TerminalBinaryMessageType.OutputUtf8, + streamId, + offset: chunk.startOffset, + flags: chunk.replay ? TerminalBinaryFlags.Replay : 0, + payload, + }); + binding.lastOutputOffset = chunk.endOffset; + } + + private enqueueOrEmitTerminalStreamChunk( + streamId: number, + binding: { + lastAckOffset: number; + lastOutputOffset: number; + pendingChunks: TerminalStreamPendingChunk[]; + pendingBytes: number; + }, + chunk: TerminalStreamPendingChunk + ): void { + const chunkBytes = this.getTerminalStreamChunkByteLength(chunk); + + if (binding.pendingChunks.length > 0 || !this.canEmitTerminalStreamChunk(binding, chunk)) { + if ( + binding.pendingChunks.length >= TERMINAL_STREAM_MAX_PENDING_CHUNKS || + binding.pendingBytes + chunkBytes > TERMINAL_STREAM_MAX_PENDING_BYTES + ) { + this.sessionLogger.warn( + { + streamId, + pendingChunks: binding.pendingChunks.length, + pendingBytes: binding.pendingBytes, + chunkBytes, + }, + "Terminal stream pending buffer overflow; closing stream" + ); + this.detachTerminalStream(streamId, { emitExit: true }); + return; + } + binding.pendingChunks.push(chunk); + binding.pendingBytes += chunkBytes; + return; + } + + this.emitTerminalStreamChunk(streamId, binding, chunk); + } + + private flushPendingTerminalStreamChunks( + streamId: number, + binding: { + lastAckOffset: number; + lastOutputOffset: number; + pendingChunks: TerminalStreamPendingChunk[]; + pendingBytes: number; + } + ): void { + while (binding.pendingChunks.length > 0) { + const next = binding.pendingChunks[0]; + if (!next || !this.canEmitTerminalStreamChunk(binding, next)) { + break; + } + binding.pendingChunks.shift(); + binding.pendingBytes -= this.getTerminalStreamChunkByteLength(next); + if (binding.pendingBytes < 0) { + binding.pendingBytes = 0; + } + this.emitTerminalStreamChunk(streamId, binding, next); + } + } + + private handleDetachTerminalStreamRequest(msg: DetachTerminalStreamRequest): void { + const success = this.detachTerminalStream(msg.streamId, { emitExit: false }); + this.emit({ + type: "detach_terminal_stream_response", + payload: { + streamId: msg.streamId, + success, + requestId: msg.requestId, + }, + }); + } + + private detachAllTerminalStreams(options?: { emitExit: boolean }): void { + for (const streamId of Array.from(this.terminalStreams.keys())) { + this.detachTerminalStream(streamId, options); + } + } + + private detachTerminalStream( + streamId: number, + options?: { emitExit: boolean } + ): boolean { + const binding = this.terminalStreams.get(streamId); + if (!binding) { + return false; + } + + try { + binding.unsubscribe(); + } catch (error) { + this.sessionLogger.warn({ err: error, streamId }, "Failed to unsubscribe terminal stream"); + } + this.terminalStreams.delete(streamId); + if (this.terminalStreamByTerminalId.get(binding.terminalId) === streamId) { + this.terminalStreamByTerminalId.delete(binding.terminalId); + } + + if (options?.emitExit) { + this.emit({ + type: "terminal_stream_exit", + payload: { + streamId, + terminalId: binding.terminalId, + }, + }); + } + return true; + } + + private allocateTerminalStreamId(): number { + let attempts = 0; + while (attempts < 0xffffffff) { + const candidate = this.nextTerminalStreamId >>> 0; + this.nextTerminalStreamId = ((this.nextTerminalStreamId + 1) & 0xffffffff) >>> 0; + if (candidate === 0) { + attempts += 1; + continue; + } + if (!this.terminalStreams.has(candidate)) { + return candidate; + } + attempts += 1; + } + throw new Error("Unable to allocate terminal stream id"); + } } 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 4056c1ad5..37eeae34a 100644 --- a/packages/server/src/server/websocket-server.relay-reconnect.test.ts +++ b/packages/server/src/server/websocket-server.relay-reconnect.test.ts @@ -1,4 +1,11 @@ import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; +import { + BinaryMuxChannel, + TerminalBinaryMessageType, + asUint8Array, + decodeBinaryMuxFrame, + encodeBinaryMuxFrame, +} from "../shared/binary-mux.js"; const wsModuleMock = vi.hoisted(() => { class MockWebSocketServer { @@ -28,6 +35,7 @@ const sessionMock = vi.hoisted(() => { class MockSession { cleanup = vi.fn(async () => {}); handleMessage = vi.fn(async () => {}); + handleBinaryFrame = vi.fn((_frame: unknown) => {}); getClientActivity = vi.fn(() => null); readonly args: Record; @@ -71,7 +79,7 @@ import { class MockSocket { readyState = 1; - sent: string[] = []; + sent: unknown[] = []; private listeners = new Map void>>(); on(event: "message" | "close" | "error", listener: (...args: any[]) => void): void { @@ -88,7 +96,7 @@ class MockSocket { this.on(event, wrapped); } - send(data: string): void { + send(data: unknown): void { this.sent.push(data); } @@ -196,4 +204,90 @@ describe("relay external socket reconnect behavior", () => { await server.close(); }); + + test("routes inbound binary mux frames to session.handleBinaryFrame", async () => { + const server = createServer(); + const metadata: ExternalSocketMetadata = { + transport: "relay", + externalSessionKey: "relay:client-binary-inbound", + }; + + const socket = new MockSocket(); + await server.attachExternalSocket(socket, metadata); + expect(sessionMock.instances).toHaveLength(1); + const session = sessionMock.instances[0]!; + + socket.emit( + "message", + Buffer.from( + encodeBinaryMuxFrame({ + channel: BinaryMuxChannel.Terminal, + messageType: TerminalBinaryMessageType.InputUtf8, + streamId: 9, + offset: 0, + payload: new TextEncoder().encode("ls\r"), + }) + ) + ); + await Promise.resolve(); + + expect(session.handleBinaryFrame).toHaveBeenCalledTimes(1); + const frame = session.handleBinaryFrame.mock.calls[0]?.[0] as { + channel: number; + messageType: number; + streamId: number; + offset: number; + }; + expect(frame.channel).toBe(BinaryMuxChannel.Terminal); + expect(frame.messageType).toBe(TerminalBinaryMessageType.InputUtf8); + expect(frame.streamId).toBe(9); + expect(frame.offset).toBe(0); + + await server.close(); + }); + + test("sends outbound binary mux frames from session over websocket", async () => { + const server = createServer(); + const metadata: ExternalSocketMetadata = { + transport: "relay", + externalSessionKey: "relay:client-binary-outbound", + }; + + const socket = new MockSocket(); + await server.attachExternalSocket(socket, metadata); + expect(sessionMock.instances).toHaveLength(1); + const session = sessionMock.instances[0]!; + + const onBinaryMessage = session.args.onBinaryMessage as + | ((frame: { + channel: number; + messageType: number; + streamId: number; + offset: number; + payload?: Uint8Array; + }) => void) + | undefined; + expect(onBinaryMessage).toBeTypeOf("function"); + + onBinaryMessage?.({ + channel: BinaryMuxChannel.Terminal, + messageType: TerminalBinaryMessageType.OutputUtf8, + streamId: 11, + offset: 42, + payload: new TextEncoder().encode("ok"), + }); + + expect(socket.sent).toHaveLength(2); + const binaryPayload = asUint8Array(socket.sent[1]); + expect(binaryPayload).not.toBeNull(); + const frame = decodeBinaryMuxFrame(binaryPayload!); + expect(frame).not.toBeNull(); + expect(frame!.channel).toBe(BinaryMuxChannel.Terminal); + expect(frame!.messageType).toBe(TerminalBinaryMessageType.OutputUtf8); + expect(frame!.streamId).toBe(11); + expect(frame!.offset).toBe(42); + expect(new TextDecoder().decode(frame!.payload ?? new Uint8Array())).toBe("ok"); + + await server.close(); + }); }); diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index c4cbebe0d..fe934b0c5 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -13,6 +13,11 @@ import { type WSOutboundMessage, wrapSessionMessage, } from "./messages.js"; +import { + asUint8Array, + decodeBinaryMuxFrame, + encodeBinaryMuxFrame, +} from "../shared/binary-mux.js"; import type { AllowedHostsConfig } from "./allowed-hosts.js"; import { isHostAllowed } from "./allowed-hosts.js"; import { Session } from "./session.js"; @@ -54,7 +59,7 @@ function bufferFromWsData(data: Buffer | ArrayBuffer | Buffer[] | string): Buffe type WebSocketLike = { readyState: number; - send: (data: string) => void; + send: (data: string | Uint8Array | ArrayBuffer) => void; close: (code?: number, reason?: string) => void; on: (event: "message" | "close" | "error", listener: (...args: any[]) => void) => void; once: (event: "close" | "error", listener: (...args: any[]) => void) => void; @@ -253,6 +258,16 @@ export class VoiceAssistantWebSocketServer { } } + private sendBinaryToClient( + ws: WebSocketLike, + frame: Parameters[0] + ): void { + if (ws.readyState !== 1) { + return; + } + ws.send(encodeBinaryMuxFrame(frame)); + } + private async attachSocket( ws: WebSocketLike, _request?: unknown, @@ -301,6 +316,9 @@ export class VoiceAssistantWebSocketServer { onMessage: (msg) => { this.sendToClient(socketRef.current, wrapSessionMessage(msg)); }, + onBinaryMessage: (frame) => { + this.sendBinaryToClient(socketRef.current, frame); + }, logger: connectionLogger.child({ module: "session" }), downloadTokenStore: this.downloadTokenStore, pushTokenStore: this.pushTokenStore, @@ -481,6 +499,19 @@ export class VoiceAssistantWebSocketServer { ): Promise { try { const buffer = bufferFromWsData(data); + const asBytes = asUint8Array(buffer); + if (asBytes) { + const frame = decodeBinaryMuxFrame(asBytes); + if (frame) { + const connection = this.sessions.get(ws); + if (!connection) { + this.logger.error("No session found for client"); + return; + } + connection.session.handleBinaryFrame(frame); + return; + } + } const parsed = JSON.parse(buffer.toString()); const parsedMessage = WSInboundMessageSchema.safeParse(parsed); if (!parsedMessage.success) { diff --git a/packages/server/src/shared/binary-mux.test.ts b/packages/server/src/shared/binary-mux.test.ts new file mode 100644 index 000000000..2af8d41b7 --- /dev/null +++ b/packages/server/src/shared/binary-mux.test.ts @@ -0,0 +1,43 @@ +import { describe, expect, it } from "vitest"; +import { + BinaryMuxChannel, + TerminalBinaryFlags, + TerminalBinaryMessageType, + decodeBinaryMuxFrame, + encodeBinaryMuxFrame, +} from "./binary-mux.js"; + +describe("binary mux frame codec", () => { + it("encodes and decodes round trip", () => { + const payload = new TextEncoder().encode("hello"); + const encoded = encodeBinaryMuxFrame({ + channel: BinaryMuxChannel.Terminal, + messageType: TerminalBinaryMessageType.OutputUtf8, + streamId: 42, + offset: 1234, + flags: TerminalBinaryFlags.Replay, + payload, + }); + + const decoded = decodeBinaryMuxFrame(encoded); + expect(decoded).not.toBeNull(); + expect(decoded!.channel).toBe(BinaryMuxChannel.Terminal); + expect(decoded!.messageType).toBe(TerminalBinaryMessageType.OutputUtf8); + expect(decoded!.streamId).toBe(42); + expect(decoded!.offset).toBe(1234); + expect(decoded!.flags).toBe(TerminalBinaryFlags.Replay); + expect(Array.from(decoded!.payload ?? [])).toEqual(Array.from(payload)); + }); + + it("rejects malformed frame payload length", () => { + const encoded = encodeBinaryMuxFrame({ + channel: BinaryMuxChannel.Terminal, + messageType: TerminalBinaryMessageType.InputUtf8, + streamId: 1, + offset: 0, + payload: new Uint8Array([1, 2, 3]), + }); + const tampered = encoded.slice(0, encoded.byteLength - 1); + expect(decodeBinaryMuxFrame(tampered)).toBeNull(); + }); +}); diff --git a/packages/server/src/shared/binary-mux.ts b/packages/server/src/shared/binary-mux.ts new file mode 100644 index 000000000..410797c8b --- /dev/null +++ b/packages/server/src/shared/binary-mux.ts @@ -0,0 +1,120 @@ +const MUX_MAGIC_1 = 0x50; // 'P' +const MUX_MAGIC_2 = 0x58; // 'X' +const MUX_VERSION = 1; +const HEADER_SIZE = 24; +const UINT32_MAX = 0xffffffff; +const TWO_POW_32 = 0x1_0000_0000; + +export const BinaryMuxConstants = { + magic1: MUX_MAGIC_1, + magic2: MUX_MAGIC_2, + version: MUX_VERSION, + headerSize: HEADER_SIZE, +} as const; + +export const enum BinaryMuxChannel { + Terminal = 1, + FileTransfer = 2, +} + +export const enum TerminalBinaryMessageType { + InputUtf8 = 1, + OutputUtf8 = 2, + Ack = 3, +} + +export const enum TerminalBinaryFlags { + Replay = 1, +} + +export interface BinaryMuxFrame { + channel: number; + messageType: number; + streamId: number; + offset: number; + flags?: number; + payload?: Uint8Array; +} + +export function asUint8Array(data: unknown): Uint8Array | null { + if (data instanceof Uint8Array) { + return data; + } + if (typeof ArrayBuffer !== "undefined" && data instanceof ArrayBuffer) { + return new Uint8Array(data); + } + if (ArrayBuffer.isView(data)) { + return new Uint8Array(data.buffer, data.byteOffset, data.byteLength); + } + if (typeof Buffer !== "undefined" && Buffer.isBuffer(data)) { + return new Uint8Array(data.buffer, data.byteOffset, data.byteLength); + } + return null; +} + +export function isLikelyBinaryMuxFrame(data: Uint8Array): boolean { + if (data.byteLength < HEADER_SIZE) { + return false; + } + return data[0] === MUX_MAGIC_1 && data[1] === MUX_MAGIC_2 && data[2] === MUX_VERSION; +} + +export function encodeBinaryMuxFrame(frame: BinaryMuxFrame): Uint8Array { + const payload = frame.payload ?? new Uint8Array(0); + const out = new Uint8Array(HEADER_SIZE + payload.byteLength); + const view = new DataView(out.buffer, out.byteOffset, out.byteLength); + const flags = frame.flags ?? 0; + + if (!Number.isFinite(frame.streamId) || frame.streamId < 0 || frame.streamId > UINT32_MAX) { + throw new Error(`Invalid streamId: ${frame.streamId}`); + } + if (!Number.isFinite(frame.offset) || frame.offset < 0) { + throw new Error(`Invalid offset: ${frame.offset}`); + } + + const offsetHi = Math.floor(frame.offset / TWO_POW_32); + const offsetLo = frame.offset >>> 0; + if (offsetHi > UINT32_MAX) { + throw new Error(`Offset too large: ${frame.offset}`); + } + + view.setUint8(0, MUX_MAGIC_1); + view.setUint8(1, MUX_MAGIC_2); + view.setUint8(2, MUX_VERSION); + view.setUint8(3, frame.channel & 0xff); + view.setUint8(4, frame.messageType & 0xff); + view.setUint8(5, flags & 0xff); + view.setUint8(6, 0); + view.setUint8(7, 0); + view.setUint32(8, frame.streamId >>> 0); + view.setUint32(12, offsetHi >>> 0); + view.setUint32(16, offsetLo); + view.setUint32(20, payload.byteLength >>> 0); + + out.set(payload, HEADER_SIZE); + return out; +} + +export function decodeBinaryMuxFrame(data: Uint8Array): BinaryMuxFrame | null { + if (!isLikelyBinaryMuxFrame(data)) { + return null; + } + const view = new DataView(data.buffer, data.byteOffset, data.byteLength); + const payloadLength = view.getUint32(20); + if (payloadLength !== data.byteLength - HEADER_SIZE) { + return null; + } + + const offsetHi = view.getUint32(12); + const offsetLo = view.getUint32(16); + const offset = offsetHi * TWO_POW_32 + offsetLo; + + return { + channel: view.getUint8(3), + messageType: view.getUint8(4), + flags: view.getUint8(5), + streamId: view.getUint32(8), + offset, + payload: data.subarray(HEADER_SIZE), + }; +} diff --git a/packages/server/src/shared/messages.ts b/packages/server/src/shared/messages.ts index 748beaac3..cc4ed0d52 100644 --- a/packages/server/src/shared/messages.ts +++ b/packages/server/src/shared/messages.ts @@ -980,6 +980,21 @@ export const KillTerminalRequestSchema = z.object({ requestId: z.string(), }); +export const AttachTerminalStreamRequestSchema = z.object({ + type: z.literal("attach_terminal_stream_request"), + terminalId: z.string(), + resumeOffset: z.number().int().nonnegative().optional(), + rows: z.number().int().positive().optional(), + cols: z.number().int().positive().optional(), + requestId: z.string(), +}); + +export const DetachTerminalStreamRequestSchema = z.object({ + type: z.literal("detach_terminal_stream_request"), + streamId: z.number().int().nonnegative(), + requestId: z.string(), +}); + export const SessionInboundMessageSchema = z.discriminatedUnion("type", [ VoiceAudioChunkMessageSchema, AbortRequestMessageSchema, @@ -1041,6 +1056,8 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [ UnsubscribeTerminalRequestSchema, TerminalInputSchema, KillTerminalRequestSchema, + AttachTerminalStreamRequestSchema, + DetachTerminalStreamRequestSchema, ]); export type SessionInboundMessage = z.infer; @@ -1819,6 +1836,37 @@ export const KillTerminalResponseSchema = z.object({ }), }); +export const AttachTerminalStreamResponseSchema = z.object({ + type: z.literal("attach_terminal_stream_response"), + payload: z.object({ + terminalId: z.string(), + streamId: z.number().int().nonnegative().nullable(), + replayedFrom: z.number().int().nonnegative(), + currentOffset: z.number().int().nonnegative(), + earliestAvailableOffset: z.number().int().nonnegative(), + reset: z.boolean(), + error: z.string().nullable(), + requestId: z.string(), + }), +}); + +export const DetachTerminalStreamResponseSchema = z.object({ + type: z.literal("detach_terminal_stream_response"), + payload: z.object({ + streamId: z.number().int().nonnegative(), + success: z.boolean(), + requestId: z.string(), + }), +}); + +export const TerminalStreamExitSchema = z.object({ + type: z.literal("terminal_stream_exit"), + payload: z.object({ + streamId: z.number().int().nonnegative(), + terminalId: z.string(), + }), +}); + export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [ ActivityLogMessageSchema, AssistantChunkMessageSchema, @@ -1879,6 +1927,9 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [ SubscribeTerminalResponseSchema, TerminalOutputSchema, KillTerminalResponseSchema, + AttachTerminalStreamResponseSchema, + DetachTerminalStreamResponseSchema, + TerminalStreamExitSchema, ]); export type SessionOutboundMessage = z.infer< @@ -2028,6 +2079,11 @@ export type TerminalInput = z.infer; export type TerminalOutput = z.infer; export type KillTerminalRequest = z.infer; export type KillTerminalResponse = z.infer; +export type AttachTerminalStreamRequest = z.infer; +export type AttachTerminalStreamResponse = z.infer; +export type DetachTerminalStreamRequest = z.infer; +export type DetachTerminalStreamResponse = z.infer; +export type TerminalStreamExit = z.infer; // ============================================================================ // WebSocket Level Messages (wraps session messages) diff --git a/packages/server/src/shared/terminal-key-input.test.ts b/packages/server/src/shared/terminal-key-input.test.ts new file mode 100644 index 000000000..a0bc0b39c --- /dev/null +++ b/packages/server/src/shared/terminal-key-input.test.ts @@ -0,0 +1,25 @@ +import { describe, expect, it } from "vitest"; +import { encodeTerminalKeyInput } from "./terminal-key-input.js"; + +describe("encodeTerminalKeyInput", () => { + it("encodes ctrl+b for tmux prefix", () => { + expect(encodeTerminalKeyInput({ key: "b", ctrl: true })).toBe("\x02"); + }); + + it("encodes shifted arrow key modifiers", () => { + expect(encodeTerminalKeyInput({ key: "ArrowLeft", shift: true })).toBe("\x1b[1;2D"); + }); + + it("encodes alt-modified printable keys", () => { + expect(encodeTerminalKeyInput({ key: "x", alt: true })).toBe("\x1bx"); + }); + + it("encodes enter and backspace", () => { + expect(encodeTerminalKeyInput({ key: "Enter" })).toBe("\r"); + expect(encodeTerminalKeyInput({ key: "Backspace" })).toBe("\x7f"); + }); + + it("returns empty string for unsupported keys", () => { + expect(encodeTerminalKeyInput({ key: "UnidentifiedKey" })).toBe(""); + }); +}); diff --git a/packages/server/src/shared/terminal-key-input.ts b/packages/server/src/shared/terminal-key-input.ts new file mode 100644 index 000000000..b98f4bd06 --- /dev/null +++ b/packages/server/src/shared/terminal-key-input.ts @@ -0,0 +1,146 @@ +export interface TerminalKeyInput { + key: string; + ctrl?: boolean; + alt?: boolean; + shift?: boolean; + meta?: boolean; +} + +function modifierParam(input: TerminalKeyInput): number { + let value = 1; + if (input.shift) value += 1; + if (input.alt || input.meta) value += 2; + if (input.ctrl) value += 4; + return value; +} + +function applyAltLikePrefix(sequence: string, input: TerminalKeyInput): string { + return input.alt || input.meta ? `\x1b${sequence}` : sequence; +} + +function encodePrintableKey(input: TerminalKeyInput): string { + const raw = input.key; + const char = input.shift ? raw.toUpperCase() : raw; + + if (input.ctrl) { + const upper = char.toUpperCase(); + if (upper.length === 1 && upper >= "A" && upper <= "Z") { + return applyAltLikePrefix(String.fromCharCode(upper.charCodeAt(0) - 64), input); + } + switch (char) { + case " ": + case "@": + case "2": + return applyAltLikePrefix("\x00", input); + case "[": + case "3": + return applyAltLikePrefix("\x1b", input); + case "\\": + case "4": + return applyAltLikePrefix("\x1c", input); + case "]": + case "5": + return applyAltLikePrefix("\x1d", input); + case "^": + case "6": + return applyAltLikePrefix("\x1e", input); + case "_": + case "/": + case "7": + return applyAltLikePrefix("\x1f", input); + case "8": + case "?": + return applyAltLikePrefix("\x7f", input); + default: + break; + } + + if (char.length === 1) { + const code = char.charCodeAt(0) & 0x1f; + return applyAltLikePrefix(String.fromCharCode(code), input); + } + } + + return applyAltLikePrefix(char, input); +} + +function csiWithModifier(finalByte: string, input: TerminalKeyInput): string { + const mod = modifierParam(input); + return mod === 1 ? `\x1b[${finalByte}` : `\x1b[1;${mod}${finalByte}`; +} + +function csiTilde(base: number, input: TerminalKeyInput): string { + const mod = modifierParam(input); + return mod === 1 ? `\x1b[${base}~` : `\x1b[${base};${mod}~`; +} + +export function encodeTerminalKeyInput(input: TerminalKeyInput): string { + const key = input.key; + if (!key) { + return ""; + } + + if (key.length === 1) { + return encodePrintableKey(input); + } + + switch (key) { + case "Enter": + return applyAltLikePrefix("\r", input); + case "Tab": + if (input.shift && !input.ctrl && !input.alt && !input.meta) { + return "\x1b[Z"; + } + return applyAltLikePrefix("\t", input); + case "Backspace": + return applyAltLikePrefix("\x7f", input); + case "Escape": + return "\x1b"; + case "ArrowUp": + return csiWithModifier("A", input); + case "ArrowDown": + return csiWithModifier("B", input); + case "ArrowRight": + return csiWithModifier("C", input); + case "ArrowLeft": + return csiWithModifier("D", input); + case "Home": + return csiWithModifier("H", input); + case "End": + return csiWithModifier("F", input); + case "Insert": + return csiTilde(2, input); + case "Delete": + return csiTilde(3, input); + case "PageUp": + return csiTilde(5, input); + case "PageDown": + return csiTilde(6, input); + case "F1": + return modifierParam(input) === 1 ? "\x1bOP" : csiWithModifier("P", input); + case "F2": + return modifierParam(input) === 1 ? "\x1bOQ" : csiWithModifier("Q", input); + case "F3": + return modifierParam(input) === 1 ? "\x1bOR" : csiWithModifier("R", input); + case "F4": + return modifierParam(input) === 1 ? "\x1bOS" : csiWithModifier("S", input); + case "F5": + return csiTilde(15, input); + case "F6": + return csiTilde(17, input); + case "F7": + return csiTilde(18, input); + case "F8": + return csiTilde(19, input); + case "F9": + return csiTilde(20, input); + case "F10": + return csiTilde(21, input); + case "F11": + return csiTilde(23, input); + case "F12": + return csiTilde(24, input); + default: + return ""; + } +} diff --git a/packages/server/src/terminal/terminal.test.ts b/packages/server/src/terminal/terminal.test.ts index d47eef735..707017acb 100644 --- a/packages/server/src/terminal/terminal.test.ts +++ b/packages/server/src/terminal/terminal.test.ts @@ -429,6 +429,91 @@ describe("Terminal", () => { }); }); + describe("raw output stream", () => { + it("streams raw output chunks with byte offsets", async () => { + const session = trackSession( + await createTerminal({ + cwd: "/tmp", + shell: "/bin/sh", + env: { PS1: "$ " }, + }) + ); + + await waitForLines(session, ["$"]); + + const chunks: Array<{ data: string; startOffset: number; endOffset: number }> = []; + const sub = session.subscribeRaw((chunk) => { + chunks.push({ + data: chunk.data, + startOffset: chunk.startOffset, + endOffset: chunk.endOffset, + }); + }); + + session.send({ type: "input", data: "echo raw-stream\r" }); + await waitForLines(session, ["$ echo raw-stream", "raw-stream", "$"]); + + expect(chunks.length).toBeGreaterThan(0); + expect(chunks[chunks.length - 1].endOffset).toBe(session.getOutputOffset()); + for (let i = 1; i < chunks.length; i++) { + expect(chunks[i].startOffset).toBe(chunks[i - 1].endOffset); + } + + sub.unsubscribe(); + }); + + it("replays buffered output from offset", async () => { + const session = trackSession( + await createTerminal({ + cwd: "/tmp", + shell: "/bin/sh", + env: { PS1: "$ " }, + }) + ); + + await waitForLines(session, ["$"]); + + const seen: Array<{ data: string; endOffset: number; replay: boolean }> = []; + const firstSub = session.subscribeRaw((chunk) => { + seen.push({ + data: chunk.data, + endOffset: chunk.endOffset, + replay: chunk.replay, + }); + }); + + session.send({ type: "input", data: "echo before-detach\r" }); + await waitForLines(session, ["$ echo before-detach", "before-detach", "$"]); + + const resumeOffset = seen[seen.length - 1]?.endOffset ?? 0; + firstSub.unsubscribe(); + + session.send({ type: "input", data: "echo after-detach\r" }); + await waitForLines(session, [ + "$ echo before-detach", + "before-detach", + "$ echo after-detach", + "after-detach", + "$", + ]); + + const replayed: string[] = []; + const secondSub = session.subscribeRaw( + (chunk) => { + if (chunk.replay) { + replayed.push(chunk.data); + } + }, + { fromOffset: resumeOffset } + ); + + const replayText = replayed.join(""); + expect(replayText).toContain("after-detach"); + expect(secondSub.replayedFrom).toBeGreaterThanOrEqual(resumeOffset); + secondSub.unsubscribe(); + }); + }); + describe("getState", () => { it("returns current terminal state with grid", async () => { const session = trackSession( diff --git a/packages/server/src/terminal/terminal.ts b/packages/server/src/terminal/terminal.ts index 2394dbb22..8c2b5a6ad 100644 --- a/packages/server/src/terminal/terminal.ts +++ b/packages/server/src/terminal/terminal.ts @@ -34,6 +34,21 @@ export interface CellChange { cell: Cell; } +export interface TerminalRawChunk { + data: string; + startOffset: number; + endOffset: number; + replay: boolean; +} + +export interface TerminalRawSubscriptionResult { + unsubscribe: () => void; + replayedFrom: number; + currentOffset: number; + earliestAvailableOffset: number; + reset: boolean; +} + export type ClientMessage = | { type: "input"; data: string } | { type: "resize"; rows: number; cols: number } @@ -49,6 +64,11 @@ export interface TerminalSession { cwd: string; send(msg: ClientMessage): void; subscribe(listener: (msg: ServerMessage) => void): () => void; + subscribeRaw( + listener: (chunk: TerminalRawChunk) => void, + options?: { fromOffset?: number } + ): TerminalRawSubscriptionResult; + getOutputOffset(): number; getState(): TerminalState; kill(): void; } @@ -169,6 +189,15 @@ export async function createTerminal(options: CreateTerminalOptions): Promise void>(); + const rawListeners = new Set<(chunk: TerminalRawChunk) => void>(); + const rawOutputChunks: Array<{ + startOffset: number; + endOffset: number; + data: string; + }> = []; + const maxRawBufferBytes = 8 * 1024 * 1024; + let rawBufferBytes = 0; + let outputOffset = 0; let killed = false; // Create xterm.js headless terminal @@ -192,9 +221,66 @@ export async function createTerminal(options: CreateTerminalOptions): Promise maxRawBufferBytes && rawOutputChunks.length > 0) { + const removed = rawOutputChunks.shift(); + if (!removed) { + break; + } + rawBufferBytes -= removed.endOffset - removed.startOffset; + } + + const chunk: TerminalRawChunk = { + data, + startOffset, + endOffset, + replay: false, + }; + for (const listener of rawListeners) { + listener(chunk); + } + } + + function getEarliestAvailableOffset(): number { + const earliest = rawOutputChunks[0]; + return earliest ? earliest.startOffset : outputOffset; + } + + function alignOffsetToChunkBoundary(offset: number): number { + if (offset === outputOffset) { + return offset; + } + for (const chunk of rawOutputChunks) { + if (offset === chunk.startOffset || offset === chunk.endOffset) { + return offset; + } + if (offset > chunk.startOffset && offset < chunk.endOffset) { + return chunk.startOffset; + } + } + return offset; + } + // Pipe PTY output to terminal emulator ptyProcess.onData((data) => { if (killed) return; + appendRawOutputChunk(data); terminal.write(data, () => { // Notify listeners of changes const state = getState(); @@ -254,11 +340,51 @@ export async function createTerminal(options: CreateTerminalOptions): Promise void, + options?: { fromOffset?: number } + ): TerminalRawSubscriptionResult { + const requestedOffset = Math.max(0, Math.floor(options?.fromOffset ?? 0)); + const earliestAvailableOffset = getEarliestAvailableOffset(); + const clampedOffset = Math.max(requestedOffset, earliestAvailableOffset); + const replayedFrom = alignOffsetToChunkBoundary(clampedOffset); + const reset = replayedFrom !== requestedOffset; + + for (const chunk of rawOutputChunks) { + if (chunk.endOffset <= replayedFrom) { + continue; + } + listener({ + data: chunk.data, + startOffset: chunk.startOffset, + endOffset: chunk.endOffset, + replay: true, + }); + } + + rawListeners.add(listener); + + return { + unsubscribe: () => { + rawListeners.delete(listener); + }, + replayedFrom, + currentOffset: outputOffset, + earliestAvailableOffset, + reset, + }; + } + + function getOutputOffset(): number { + return outputOffset; + } + function kill(): void { if (killed) return; killed = true; ptyProcess.kill(); terminal.dispose(); + rawListeners.clear(); } // Small delay to let shell initialize @@ -268,9 +394,11 @@ export async function createTerminal(options: CreateTerminalOptions): Promise