From 2b70ec6f3a81149027a2a44779f2ac4cc73d9d4a Mon Sep 17 00:00:00 2001 From: Yifan <74950140+geekifan@users.noreply.github.com> Date: Thu, 2 Apr 2026 00:31:21 +0800 Subject: [PATCH] feat(terminal): switch terminal transport to pure websocket with fallback (#762) * feat(terminal): add resumable websocket transport Unify terminal input and stream traffic on `/api/terminal/ws` with a v2 control-frame protocol and advertised transport capabilities. Buffer recent PTY output on the server so rebinding clients can replay missed chunks after reconnects or startup races, while keeping SSE as a fallback stream path. Update the web terminal client and store to negotiate the new transport, track tab lifecycle, and avoid reopening exited sessions when restoring tabs. * fix(terminal): retry rehydrated websocket reconnects * fix(terminal): keep reconnect retries silent * fix(terminal): wire ws stream output and replay in runtime * fix(web): enable ws proxy for /api in dev --------- Co-authored-by: Bohdan Triapitsyn --- .../ui/src/components/views/TerminalView.tsx | 67 +- packages/ui/src/lib/api/types.ts | 21 +- packages/ui/src/lib/terminalApi.ts | 733 ++++++++++++------ packages/ui/src/stores/useTerminalStore.ts | 50 +- .../web/server/TERMINAL_INPUT_WS_PROTOCOL.md | 44 -- packages/web/server/TERMINAL_WS_PROTOCOL.md | 48 ++ .../web/server/lib/terminal/DOCUMENTATION.md | 121 +-- packages/web/server/lib/terminal/index.js | 35 +- .../lib/terminal/output-replay-buffer.js | 66 ++ .../lib/terminal/output-replay-buffer.test.js | 66 ++ packages/web/server/lib/terminal/runtime.js | 127 ++- ...ws-protocol.js => terminal-ws-protocol.js} | 24 +- ...l.test.js => terminal-ws-protocol.test.js} | 71 +- packages/web/vite.config.ts | 1 + 14 files changed, 1012 insertions(+), 462 deletions(-) delete mode 100644 packages/web/server/TERMINAL_INPUT_WS_PROTOCOL.md create mode 100644 packages/web/server/TERMINAL_WS_PROTOCOL.md create mode 100644 packages/web/server/lib/terminal/output-replay-buffer.js create mode 100644 packages/web/server/lib/terminal/output-replay-buffer.test.js rename packages/web/server/lib/terminal/{input-ws-protocol.js => terminal-ws-protocol.js} (59%) rename packages/web/server/lib/terminal/{input-ws-protocol.test.js => terminal-ws-protocol.test.js} (58%) diff --git a/packages/ui/src/components/views/TerminalView.tsx b/packages/ui/src/components/views/TerminalView.tsx index b64727f5..fb630353 100644 --- a/packages/ui/src/components/views/TerminalView.tsx +++ b/packages/ui/src/components/views/TerminalView.tsx @@ -54,7 +54,7 @@ const STREAM_OPTIONS = { const REHYDRATED_STREAM_OPTIONS = { retry: { - maxRetries: 0, + ...STREAM_OPTIONS.retry, initialDelayMs: 200, maxDelayMs: 500, }, @@ -106,6 +106,7 @@ export const TerminalView: React.FC = () => { const setActiveTab = terminalStore.setActiveTab; const closeTab = terminalStore.closeTab; const setTabSessionId = terminalStore.setTabSessionId; + const setTabLifecycle = terminalStore.setTabLifecycle; const setConnecting = terminalStore.setConnecting; const appendToBuffer = terminalStore.appendToBuffer; @@ -132,11 +133,13 @@ export const TerminalView: React.FC = () => { }, [directoryTerminalState, activeTabId]); const terminalSessionId = activeTab?.terminalSessionId ?? null; + const terminalLifecycle = activeTab?.lifecycle ?? 'idle'; const bufferChunks = activeTab?.bufferChunks ?? []; const isConnecting = activeTab?.isConnecting ?? false; const [connectionError, setConnectionError] = React.useState(null); const [isFatalError, setIsFatalError] = React.useState(false); + const [isReconnectPending, setIsReconnectPending] = React.useState(false); const [activeModifier, setActiveModifier] = React.useState(null); const [isRestarting, setIsRestarting] = React.useState(false); const [viewportLayoutVersion, setViewportLayoutVersion] = React.useState(0); @@ -234,6 +237,7 @@ export const TerminalView: React.FC = () => { streamCleanupRef.current?.(); streamCleanupRef.current = null; activeTerminalIdRef.current = null; + setIsReconnectPending(false); }, []); React.useEffect( @@ -273,6 +277,7 @@ export const TerminalView: React.FC = () => { setConnecting(directory, tabId, false); setConnectionError(null); setIsFatalError(false); + setIsReconnectPending(false); focusTerminalWhenWindowActive(); // After a reload, buffer is empty and a reused PTY can look "stuck" @@ -286,10 +291,10 @@ export const TerminalView: React.FC = () => { break; } case 'reconnecting': { - const attempt = event.attempt ?? 0; - const maxAttempts = event.maxAttempts ?? 3; - setConnectionError(`Reconnecting (${attempt}/${maxAttempts})...`); + void event; + setConnectionError(null); setIsFatalError(false); + setIsReconnectPending(true); break; } case 'data': { @@ -313,10 +318,12 @@ export const TerminalView: React.FC = () => { exitCode !== null ? ` with code ${exitCode}` : '' }${signal !== null ? ` (signal ${signal})` : ''}]\r\n` ); + setTabLifecycle(directory, tabId, 'exited'); setTabSessionId(directory, tabId, null); setConnecting(directory, tabId, false); setConnectionError(isActionTab ? null : 'Terminal session ended'); setIsFatalError(false); + setIsReconnectPending(false); disconnectStream(); break; } @@ -327,18 +334,19 @@ export const TerminalView: React.FC = () => { return; } - const errorMsg = fatal - ? `Connection failed: ${error.message}` - : error.message || 'Terminal stream connection error'; - - setConnectionError(errorMsg); - setIsFatalError(!!fatal); - - if (fatal) { - setConnecting(directory, tabId, false); - setTabSessionId(directory, tabId, null); - disconnectStream(); + if (!fatal) { + setConnectionError(null); + setIsFatalError(false); + return; } + + setIsReconnectPending(false); + setConnectionError(`Connection failed: ${error.message}`); + setIsFatalError(true); + setConnecting(directory, tabId, false); + setTabLifecycle(directory, tabId, 'exited'); + setTabSessionId(directory, tabId, null); + disconnectStream(); }, }, streamOptions @@ -349,7 +357,7 @@ export const TerminalView: React.FC = () => { activeTerminalIdRef.current = null; }; }, - [appendToBuffer, disconnectStream, focusTerminalWhenWindowActive, setConnecting, setTabSessionId, terminal] + [appendToBuffer, disconnectStream, focusTerminalWhenWindowActive, setConnecting, setTabLifecycle, setTabSessionId, terminal] ); React.useEffect(() => { @@ -389,6 +397,7 @@ export const TerminalView: React.FC = () => { const tab = state.tabs.find((t) => t.id === tabId) ?? state.tabs[0]; let terminalId = tab?.terminalSessionId ?? null; + const terminalLifecycle = tab?.lifecycle ?? 'idle'; const isActionTab = Boolean(tab?.label?.startsWith('Action:')); const hasBufferedOutput = (tab?.bufferLength ?? 0) > 0 || (tab?.bufferChunks?.length ?? 0) > 0; @@ -402,6 +411,11 @@ export const TerminalView: React.FC = () => { Boolean(terminalId) && rehydratedTerminalIdsRef.current.has(terminalId as string); if (!terminalId) { + if (terminalLifecycle === 'exited') { + setConnecting(directory, tabId, false); + return; + } + if (isActionTab && hasBufferedOutput) { setConnecting(directory, tabId, false); return; @@ -409,6 +423,7 @@ export const TerminalView: React.FC = () => { setConnectionError(null); setIsFatalError(false); + setIsReconnectPending(false); setConnecting(directory, tabId, true); try { const size = lastViewportSizeRef.current; @@ -440,6 +455,7 @@ export const TerminalView: React.FC = () => { : 'Failed to start terminal session' ); setIsFatalError(true); + setIsReconnectPending(false); setConnecting(directory, tabId, false); } return; @@ -476,12 +492,14 @@ export const TerminalView: React.FC = () => { hasActiveContext, effectiveDirectory, terminalSessionId, + terminalLifecycle, activeTabId, hasOpenedTerminalViewport, enableTabs, terminalHydrated, ensureDirectory, setConnecting, + setTabLifecycle, setTabSessionId, startStream, disconnectStream, @@ -520,6 +538,7 @@ export const TerminalView: React.FC = () => { setIsRestarting(true); setConnectionError(null); setIsFatalError(false); + setIsReconnectPending(false); disconnectStream(); @@ -528,6 +547,7 @@ export const TerminalView: React.FC = () => { } catch (error) { setConnectionError(error instanceof Error ? error.message : 'Failed to restart terminal'); setIsFatalError(true); + setIsReconnectPending(false); } finally { setIsRestarting(false); } @@ -544,6 +564,7 @@ export const TerminalView: React.FC = () => { setActiveTab(effectiveDirectory, tabId); setConnectionError(null); setIsFatalError(false); + setIsReconnectPending(false); disconnectStream(); }, [createTab, disconnectStream, effectiveDirectory, setActiveTab]); @@ -553,6 +574,7 @@ export const TerminalView: React.FC = () => { setActiveTab(effectiveDirectory, tabId); setConnectionError(null); setIsFatalError(false); + setIsReconnectPending(false); disconnectStream(); }, [disconnectStream, effectiveDirectory, setActiveTab] @@ -568,6 +590,7 @@ export const TerminalView: React.FC = () => { setConnectionError(null); setIsFatalError(false); + setIsReconnectPending(false); void closeTab(effectiveDirectory, tabId); }, [activeTabId, closeTab, disconnectStream, effectiveDirectory] @@ -576,7 +599,7 @@ export const TerminalView: React.FC = () => { const handleViewportInput = React.useCallback( (data: string) => { - if (!data) { + if (!data || isReconnectPending) { return; } @@ -602,7 +625,9 @@ export const TerminalView: React.FC = () => { if (!terminalId) return; void terminal.sendInput(terminalId, payload).catch((error) => { - setConnectionError(error instanceof Error ? error.message : 'Failed to send input'); + if (!isReconnectPending) { + setConnectionError(error instanceof Error ? error.message : 'Failed to send input'); + } }); if (modifierConsumed) { @@ -610,7 +635,7 @@ export const TerminalView: React.FC = () => { terminalControllerRef.current?.focus(); } }, - [activeModifier, setActiveModifier, terminal] + [activeModifier, isReconnectPending, setActiveModifier, terminal] ); const handleViewportResize = React.useCallback( @@ -855,7 +880,7 @@ export const TerminalView: React.FC = () => { ); } - const quickKeysDisabled = !terminalSessionId || isConnecting || isRestarting; + const quickKeysDisabled = !terminalSessionId || isConnecting || isRestarting || isReconnectPending; const shouldRenderViewport = isMobile ? isTerminalVisible : hasOpenedTerminalViewport; const quickKeysControls = ( <> @@ -1075,7 +1100,7 @@ export const TerminalView: React.FC = () => { /> ) : null} - {connectionError && ( + {!isReconnectPending && connectionError && (
{connectionError} {isFatalError && isMobile && ( diff --git a/packages/ui/src/lib/api/types.ts b/packages/ui/src/lib/api/types.ts index 24a6f473..9f9c253c 100644 --- a/packages/ui/src/lib/api/types.ts +++ b/packages/ui/src/lib/api/types.ts @@ -29,20 +29,23 @@ export interface RetryPolicy { maxDelayMs: number; } +export interface TerminalTransportCapability { + preferred?: 'ws' | 'http' | 'sse'; + transports?: Array<'ws' | 'http' | 'sse'>; + ws?: { + path: string; + v?: number; + enc?: string; + }; +} + export interface TerminalSession { sessionId: string; cols: number; rows: number; capabilities?: { - input?: { - preferred?: 'ws' | 'http'; - transports?: Array<'ws' | 'http'>; - ws?: { - path: string; - v?: number; - enc?: string; - }; - }; + input?: TerminalTransportCapability; + stream?: TerminalTransportCapability; }; } diff --git a/packages/ui/src/lib/terminalApi.ts b/packages/ui/src/lib/terminalApi.ts index c11ae71c..491a8e74 100644 --- a/packages/ui/src/lib/terminalApi.ts +++ b/packages/ui/src/lib/terminalApi.ts @@ -1,21 +1,22 @@ +export interface TerminalWebSocketDescriptor { + path: string; + v?: number; + enc?: string; +} +export interface TerminalTransportCapability { + preferred?: 'ws' | 'http' | 'sse'; + transports?: Array<'ws' | 'http' | 'sse'>; + ws?: TerminalWebSocketDescriptor; +} export interface TerminalSession { sessionId: string; cols: number; rows: number; capabilities?: { - input?: TerminalInputCapability; - }; -} - -export interface TerminalInputCapability { - preferred?: 'ws' | 'http'; - transports?: Array<'ws' | 'http'>; - ws?: { - path: string; - v?: number; - enc?: string; + input?: TerminalTransportCapability; + stream?: TerminalTransportCapability; }; } @@ -26,6 +27,8 @@ export interface TerminalStreamEvent { signal?: number | null; attempt?: number; maxAttempts?: number; + runtime?: 'node' | 'bun'; + ptyBackend?: string; } export interface CreateTerminalOptions { @@ -41,24 +44,43 @@ export interface ConnectStreamOptions { connectionTimeout?: number; } -type TerminalInputControlMessage = { +type TerminalControlMessage = { t: string; s?: string; c?: string; f?: boolean; v?: number; + exitCode?: number; + signal?: number | null; + runtime?: 'node' | 'bun'; + ptyBackend?: string; +}; + +type StreamSubscription = { + token: symbol; + sessionId: string; + onEvent: (event: TerminalStreamEvent) => void; + onError?: (error: Error, fatal?: boolean) => void; + maxRetries: number; + initialRetryDelay: number; + maxRetryDelay: number; + connectionTimeout: number; + retryCount: number; + connected: boolean; + connectionTimeoutId: ReturnType | null; }; const CONTROL_TAG_JSON = 0x01; const WS_READY_STATE_OPEN = 1; -const DEFAULT_TERMINAL_INPUT_WS_PATH = '/api/terminal/input-ws'; +const WS_READY_STATE_CONNECTING = 0; +const DEFAULT_TERMINAL_WS_PATH = '/api/terminal/ws'; const WS_SEND_WAIT_MS = 1200; const WS_RECONNECT_INITIAL_DELAY_MS = 1000; const WS_RECONNECT_MAX_DELAY_MS = 30000; const WS_RECONNECT_JITTER_MS = 250; const WS_KEEPALIVE_INTERVAL_MS = 20000; const WS_CONNECT_TIMEOUT_MS = 5000; -const GLOBAL_TERMINAL_INPUT_STATE_KEY = '__openchamberTerminalInputWsState'; +const GLOBAL_TERMINAL_TRANSPORT_STATE_KEY = '__openchamberTerminalTransportState'; const textEncoder = new TextEncoder(); const textDecoder = new TextDecoder(); @@ -83,7 +105,7 @@ const normalizeWebSocketPath = (pathValue: string): string => { return `${protocol}//${window.location.host}${normalizedPath}`; }; -const encodeControlFrame = (payload: TerminalInputControlMessage): Uint8Array => { +const encodeControlFrame = (payload: TerminalControlMessage): Uint8Array => { const jsonBytes = textEncoder.encode(JSON.stringify(payload)); const bytes = new Uint8Array(jsonBytes.length + 1); bytes[0] = CONTROL_TAG_JSON; @@ -91,33 +113,58 @@ const encodeControlFrame = (payload: TerminalInputControlMessage): Uint8Array => return bytes; }; -const isWsInputSupported = (capability: TerminalInputCapability | null): boolean => { +const isWsTransportSupported = (capability: TerminalTransportCapability | null | undefined): boolean => { if (!capability) return false; const transports = capability.transports ?? []; const supportsTransport = transports.includes('ws') || capability.preferred === 'ws'; return supportsTransport && typeof capability.ws?.path === 'string' && capability.ws.path.length > 0; }; -class TerminalInputWsManager { +const getPreferredTerminalWsPath = (state: TerminalTransportGlobalState): string => ( + state.streamCapability?.ws?.path + ?? state.inputCapability?.ws?.path + ?? DEFAULT_TERMINAL_WS_PATH +); + +const createTransportError = (code: string | undefined): Error => { + switch (code) { + case 'SESSION_NOT_FOUND': + return new Error('Terminal session not found'); + case 'NOT_BOUND': + return new Error('Terminal session is not bound'); + case 'WRITE_FAIL': + return new Error('Failed to write to terminal'); + case 'RATE_LIMIT': + return new Error('Terminal websocket is rate limited'); + case 'BAD_FRAME': + return new Error('Terminal websocket protocol violation'); + default: + return new Error('Terminal websocket error'); + } +}; + +class TerminalTransportManager { private socket: WebSocket | null = null; private socketUrl = ''; private boundSessionId: string | null = null; + private requestedSessionId: string | null = null; private openPromise: Promise | null = null; private reconnectTimeout: ReturnType | null = null; - private reconnectAttempt = 0; private keepaliveInterval: ReturnType | null = null; private closed = false; + private subscriptions = new Map(); + private activeSubscriptionToken: symbol | null = null; configure(socketUrl: string): void { - if (!socketUrl) return; + if (!socketUrl) { + return; + } if (this.socketUrl === socketUrl) { this.closed = false; - if (this.isConnectedOrConnecting()) { - return; + if (!this.isConnectedOrConnecting()) { + this.ensureConnected(); } - - this.ensureConnected(); return; } @@ -127,6 +174,50 @@ class TerminalInputWsManager { this.ensureConnected(); } + subscribe( + sessionId: string, + onEvent: (event: TerminalStreamEvent) => void, + onError?: (error: Error, fatal?: boolean) => void, + options?: ConnectStreamOptions + ): () => void { + const token = Symbol(sessionId); + const subscription: StreamSubscription = { + token, + sessionId, + onEvent, + onError, + maxRetries: options?.maxRetries ?? 3, + initialRetryDelay: options?.initialRetryDelay ?? 1000, + maxRetryDelay: options?.maxRetryDelay ?? 8000, + connectionTimeout: options?.connectionTimeout ?? 10000, + retryCount: 0, + connected: false, + connectionTimeoutId: null, + }; + + this.subscriptions.set(token, subscription); + this.activeSubscriptionToken = token; + this.boundSessionId = null; + this.requestedSessionId = sessionId; + this.ensureConnected(); + this.startConnectionTimeout(subscription); + this.bindActiveSession(); + + return () => { + this.clearConnectionTimeout(subscription); + this.subscriptions.delete(token); + if (this.activeSubscriptionToken === token) { + this.activeSubscriptionToken = null; + } + if (this.boundSessionId === sessionId) { + this.boundSessionId = null; + } + if (this.requestedSessionId === sessionId) { + this.requestedSessionId = null; + } + }; + } + async sendInput(sessionId: string, data: string): Promise { if (!sessionId || !data || this.closed || !this.socketUrl) { return false; @@ -139,37 +230,43 @@ class TerminalInputWsManager { try { if (this.boundSessionId !== sessionId) { - socket.send(encodeControlFrame({ t: 'b', s: sessionId, v: 1 })); - this.boundSessionId = sessionId; + this.requestedSessionId = sessionId; + socket.send(encodeControlFrame({ t: 'b', s: sessionId, v: 2 })); } socket.send(data); return true; } catch { - this.handleSocketFailure(); + this.handleSocketFailure(new Error('Terminal websocket send failed')); return false; } } unbindSession(sessionId: string): void { - if (!sessionId) return; + if (!sessionId) { + return; + } if (this.boundSessionId === sessionId) { this.boundSessionId = null; } + if (this.requestedSessionId === sessionId) { + this.requestedSessionId = null; + } } close(): void { this.closed = true; this.clearReconnectTimeout(); + for (const subscription of this.subscriptions.values()) { + this.clearConnectionTimeout(subscription); + } this.resetConnection(); this.socketUrl = ''; + this.subscriptions.clear(); + this.activeSubscriptionToken = null; } prime(): void { - if (this.closed || !this.socketUrl) { - return; - } - - if (this.isConnectedOrConnecting()) { + if (this.closed || !this.socketUrl || this.isConnectedOrConnecting()) { return; } @@ -185,73 +282,39 @@ class TerminalInputWsManager { return false; } - if (this.socket && (this.socket.readyState === WebSocket.OPEN || this.socket.readyState === WebSocket.CONNECTING)) { + if (this.socket && (this.socket.readyState === WS_READY_STATE_OPEN || this.socket.readyState === WS_READY_STATE_CONNECTING)) { return true; } return this.openPromise !== null; } - private sendControl(payload: TerminalInputControlMessage): boolean { - if (!this.socket || this.socket.readyState !== WS_READY_STATE_OPEN) { - return false; + private getActiveSubscription(): StreamSubscription | null { + if (!this.activeSubscriptionToken) { + return null; } - try { - this.socket.send(encodeControlFrame(payload)); - return true; - } catch { - this.handleSocketFailure(); - return false; - } + return this.subscriptions.get(this.activeSubscriptionToken) ?? null; } - private startKeepalive(): void { - this.stopKeepalive(); - this.keepaliveInterval = setInterval(() => { - if (this.closed) { + private startConnectionTimeout(subscription: StreamSubscription): void { + this.clearConnectionTimeout(subscription); + subscription.connectionTimeoutId = setTimeout(() => { + if (this.getActiveSubscription()?.token !== subscription.token || subscription.connected) { return; } - this.sendControl({ t: 'p', v: 1 }); - }, WS_KEEPALIVE_INTERVAL_MS); + this.handleSocketFailure(new Error('Connection timeout')); + }, subscription.connectionTimeout); } - private stopKeepalive(): void { - if (!this.keepaliveInterval) { + private clearConnectionTimeout(subscription: StreamSubscription): void { + if (!subscription.connectionTimeoutId) { return; } - clearInterval(this.keepaliveInterval); - this.keepaliveInterval = null; - } - - private scheduleReconnect(): void { - if (this.closed || !this.socketUrl || this.reconnectTimeout) { - return; - } - - const baseDelay = Math.min( - WS_RECONNECT_INITIAL_DELAY_MS * Math.pow(2, this.reconnectAttempt), - WS_RECONNECT_MAX_DELAY_MS - ); - const jitter = Math.floor(Math.random() * WS_RECONNECT_JITTER_MS); - const delay = baseDelay + jitter; - - this.reconnectTimeout = setTimeout(() => { - this.reconnectTimeout = null; - this.reconnectAttempt += 1; - this.ensureConnected(); - }, delay); - } - - private clearReconnectTimeout(): void { - if (!this.reconnectTimeout) { - return; - } - - clearTimeout(this.reconnectTimeout); - this.reconnectTimeout = null; + clearTimeout(subscription.connectionTimeoutId); + subscription.connectionTimeoutId = null; } private async getOpenSocket(waitMs: number): Promise { @@ -288,7 +351,7 @@ class TerminalInputWsManager { return; } - if (this.socket && (this.socket.readyState === WebSocket.OPEN || this.socket.readyState === WebSocket.CONNECTING)) { + if (this.socket && (this.socket.readyState === WS_READY_STATE_OPEN || this.socket.readyState === WS_READY_STATE_CONNECTING)) { return; } @@ -321,7 +384,6 @@ class TerminalInputWsManager { socket.onopen = () => { this.socket = socket; - this.reconnectAttempt = 0; this.startKeepalive(); settle(socket); }; @@ -330,13 +392,19 @@ class TerminalInputWsManager { void this.handleSocketMessage(event.data); }; + socket.onerror = () => { + if (!this.closed && !this.getActiveSubscription()) { + this.scheduleReconnect(new Error('Terminal websocket error')); + } + }; + socket.onclose = () => { if (this.socket === socket) { this.socket = null; this.boundSessionId = null; this.stopKeepalive(); if (!this.closed) { - this.scheduleReconnect(); + this.scheduleReconnect(new Error('Terminal stream connection error')); } } settle(null); @@ -345,7 +413,7 @@ class TerminalInputWsManager { this.socket = socket; connectTimeout = setTimeout(() => { - if (socket.readyState === WebSocket.CONNECTING) { + if (socket.readyState === WS_READY_STATE_CONNECTING) { socket.close(); settle(null); } @@ -353,38 +421,209 @@ class TerminalInputWsManager { } catch { settle(null); if (!this.closed) { - this.scheduleReconnect(); + this.scheduleReconnect(new Error('Terminal websocket open failed')); } } }); } - private async handleSocketMessage(messageData: unknown): Promise { - const bytes = await this.asUint8Array(messageData); - if (!bytes || bytes.length < 2) { + private bindActiveSession(): void { + const activeSubscription = this.getActiveSubscription(); + if (!activeSubscription || !this.socket || this.socket.readyState !== WS_READY_STATE_OPEN) { return; } - if (bytes[0] !== CONTROL_TAG_JSON) { - return; - } + this.requestedSessionId = activeSubscription.sessionId; try { - const payload = JSON.parse(textDecoder.decode(bytes.subarray(1))) as TerminalInputControlMessage; - if (payload.t === 'po') { + this.socket.send(encodeControlFrame({ t: 'b', s: activeSubscription.sessionId, v: 2 })); + } catch { + this.handleSocketFailure(new Error('Terminal websocket bind failed')); + } + } + + private scheduleReconnect(error: Error): void { + if (this.closed || !this.socketUrl || this.reconnectTimeout) { + return; + } + + const activeSubscription = this.getActiveSubscription(); + const attempt = (activeSubscription?.retryCount ?? 0) + 1; + const initialDelay = activeSubscription?.initialRetryDelay ?? WS_RECONNECT_INITIAL_DELAY_MS; + const maxDelay = activeSubscription?.maxRetryDelay ?? WS_RECONNECT_MAX_DELAY_MS; + const maxRetries = activeSubscription?.maxRetries ?? Number.POSITIVE_INFINITY; + + if (activeSubscription) { + if (attempt > maxRetries) { + this.clearConnectionTimeout(activeSubscription); + activeSubscription.onError?.(error, true); return; } - if (payload.t === 'e') { + activeSubscription.retryCount = attempt; + activeSubscription.connected = false; + activeSubscription.onEvent({ + type: 'reconnecting', + attempt, + maxAttempts: maxRetries, + }); + this.startConnectionTimeout(activeSubscription); + } + + const baseDelay = Math.min(initialDelay * Math.pow(2, Math.max(attempt - 1, 0)), maxDelay); + const jitter = Math.floor(Math.random() * WS_RECONNECT_JITTER_MS); + const delay = baseDelay + jitter; + + this.reconnectTimeout = setTimeout(() => { + this.reconnectTimeout = null; + this.ensureConnected(); + this.bindActiveSession(); + }, delay); + } + + private clearReconnectTimeout(): void { + if (!this.reconnectTimeout) { + return; + } + + clearTimeout(this.reconnectTimeout); + this.reconnectTimeout = null; + } + + private sendControl(payload: TerminalControlMessage): boolean { + if (!this.socket || this.socket.readyState !== WS_READY_STATE_OPEN) { + return false; + } + + try { + this.socket.send(encodeControlFrame(payload)); + return true; + } catch { + this.handleSocketFailure(new Error('Terminal websocket control send failed')); + return false; + } + } + + private startKeepalive(): void { + this.stopKeepalive(); + this.keepaliveInterval = setInterval(() => { + if (this.closed) { + return; + } + + this.sendControl({ t: 'p', v: 2 }); + }, WS_KEEPALIVE_INTERVAL_MS); + } + + private stopKeepalive(): void { + if (!this.keepaliveInterval) { + return; + } + + clearInterval(this.keepaliveInterval); + this.keepaliveInterval = null; + } + + private async handleSocketMessage(messageData: unknown): Promise { + const bytes = await this.asUint8Array(messageData); + if (bytes && bytes.length > 0 && bytes[0] === CONTROL_TAG_JSON) { + this.handleControlMessage(bytes); + return; + } + + const text = await this.asText(messageData); + if (!text) { + return; + } + + const activeSubscription = this.getActiveSubscription(); + if (!activeSubscription) { + return; + } + + activeSubscription.onEvent({ type: 'data', data: text }); + } + + private handleControlMessage(bytes: Uint8Array): void { + if (bytes.length < 2) { + return; + } + + let payload: TerminalControlMessage; + try { + payload = JSON.parse(textDecoder.decode(bytes.subarray(1))) as TerminalControlMessage; + } catch { + this.handleSocketFailure(new Error('Terminal websocket control parse failed')); + return; + } + + const activeSubscription = this.getActiveSubscription(); + + switch (payload.t) { + case 'ok': + this.bindActiveSession(); + return; + case 'po': + return; + case 'bok': { + this.boundSessionId = payload.s ?? this.requestedSessionId; + if (!activeSubscription) { + return; + } + activeSubscription.retryCount = 0; + activeSubscription.connected = true; + this.clearConnectionTimeout(activeSubscription); + activeSubscription.onEvent({ + type: 'connected', + runtime: payload.runtime, + ptyBackend: payload.ptyBackend, + }); + return; + } + case 'x': { + if (!activeSubscription) { + this.boundSessionId = null; + return; + } + + if (payload.s && payload.s !== activeSubscription.sessionId) { + return; + } + + activeSubscription.connected = false; + this.clearConnectionTimeout(activeSubscription); + this.boundSessionId = null; + this.requestedSessionId = null; + activeSubscription.onEvent({ + type: 'exit', + exitCode: payload.exitCode, + signal: payload.signal ?? null, + }); + return; + } + case 'e': { + const error = createTransportError(payload.c); + const isFatal = payload.f === true || payload.c === 'SESSION_NOT_FOUND'; + if (payload.c === 'NOT_BOUND' || payload.c === 'SESSION_NOT_FOUND') { this.boundSessionId = null; } - if (payload.f === true) { - this.handleSocketFailure(); + + if (activeSubscription) { + activeSubscription.connected = false; + if (isFatal) { + this.clearConnectionTimeout(activeSubscription); + } + activeSubscription.onError?.(error, isFatal); } + + if (payload.f === true) { + this.handleSocketFailure(error); + } + return; } - } catch { - this.handleSocketFailure(); + default: + return; } } @@ -405,10 +644,24 @@ class TerminalInputWsManager { return null; } - private handleSocketFailure(): void { + private async asText(messageData: unknown): Promise { + if (typeof messageData === 'string') { + return messageData; + } + + const bytes = await this.asUint8Array(messageData); + if (!bytes) { + return ''; + } + + return textDecoder.decode(bytes); + } + + private handleSocketFailure(error: Error): void { this.boundSessionId = null; + this.requestedSessionId = null; this.resetConnection(); - this.scheduleReconnect(); + this.scheduleReconnect(error); } private resetConnection(): void { @@ -421,7 +674,7 @@ class TerminalInputWsManager { socket.onmessage = null; socket.onerror = null; socket.onclose = null; - if (socket.readyState === WebSocket.OPEN || socket.readyState === WebSocket.CONNECTING) { + if (socket.readyState === WS_READY_STATE_OPEN || socket.readyState === WS_READY_STATE_CONNECTING) { socket.close(); } } @@ -429,51 +682,54 @@ class TerminalInputWsManager { } } -type TerminalInputWsGlobalState = { - capability: TerminalInputCapability | null; - manager: TerminalInputWsManager | null; +type TerminalTransportGlobalState = { + inputCapability: TerminalTransportCapability | null; + streamCapability: TerminalTransportCapability | null; + manager: TerminalTransportManager | null; }; -const getTerminalInputWsGlobalState = (): TerminalInputWsGlobalState => { +const getTerminalTransportGlobalState = (): TerminalTransportGlobalState => { const globalScope = globalThis as typeof globalThis & { - [GLOBAL_TERMINAL_INPUT_STATE_KEY]?: TerminalInputWsGlobalState; + [GLOBAL_TERMINAL_TRANSPORT_STATE_KEY]?: TerminalTransportGlobalState; }; - if (!globalScope[GLOBAL_TERMINAL_INPUT_STATE_KEY]) { - globalScope[GLOBAL_TERMINAL_INPUT_STATE_KEY] = { - capability: null, + if (!globalScope[GLOBAL_TERMINAL_TRANSPORT_STATE_KEY]) { + globalScope[GLOBAL_TERMINAL_TRANSPORT_STATE_KEY] = { + inputCapability: null, + streamCapability: null, manager: null, }; } - return globalScope[GLOBAL_TERMINAL_INPUT_STATE_KEY]; + return globalScope[GLOBAL_TERMINAL_TRANSPORT_STATE_KEY]; }; -const applyTerminalInputCapability = (capability: TerminalInputCapability | undefined): void => { - const globalState = getTerminalInputWsGlobalState(); - globalState.capability = capability ?? null; +const ensureTerminalTransportManager = (): TerminalTransportManager => { + const globalState = getTerminalTransportGlobalState(); + if (!globalState.manager) { + globalState.manager = new TerminalTransportManager(); + } + return globalState.manager; +}; - if (!isWsInputSupported(globalState.capability)) { +const applyTerminalTransportCapabilities = (capabilities: TerminalSession['capabilities'] | undefined): void => { + const globalState = getTerminalTransportGlobalState(); + globalState.inputCapability = capabilities?.input ?? null; + globalState.streamCapability = capabilities?.stream ?? null; + + if (!isWsTransportSupported(globalState.inputCapability) && !isWsTransportSupported(globalState.streamCapability)) { globalState.manager?.close(); globalState.manager = null; return; } - const wsPath = globalState.capability?.ws?.path; - if (!wsPath) { - return; - } - - const socketUrl = normalizeWebSocketPath(wsPath); + const socketUrl = normalizeWebSocketPath(getPreferredTerminalWsPath(globalState)); if (!socketUrl) { return; } - if (!globalState.manager) { - globalState.manager = new TerminalInputWsManager(); - } - - globalState.manager.configure(socketUrl); + const manager = ensureTerminalTransportManager(); + manager.configure(socketUrl); }; const sendTerminalInputHttp = async (sessionId: string, data: string): Promise => { @@ -484,21 +740,19 @@ const sendTerminalInputHttp = async (sessionId: string, data: string): Promise ({ error: 'Failed to send input' })); + const error = await response.json().catch(() => ({ error: 'Failed to send terminal input' })); throw new Error(error.error || 'Failed to send terminal input'); } }; -export async function createTerminalSession( - options: CreateTerminalOptions -): Promise { +export async function createTerminalSession(options: CreateTerminalOptions): Promise { const response = await fetch('/api/terminal/create', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ cwd: options.cwd, - cols: options.cols || 80, - rows: options.rows || 24, + cols: options.cols ?? 80, + rows: options.rows ?? 24, }), }); @@ -508,22 +762,20 @@ export async function createTerminalSession( } const session = await response.json() as TerminalSession; - applyTerminalInputCapability(session.capabilities?.input); + applyTerminalTransportCapabilities(session.capabilities); return session; } -export function connectTerminalStream( +const connectTerminalStreamViaSse = ( sessionId: string, onEvent: (event: TerminalStreamEvent) => void, onError?: (error: Error, fatal?: boolean) => void, options: ConnectStreamOptions = {} -): () => void { - const { - maxRetries = 3, - initialRetryDelay = 1000, - maxRetryDelay = 8000, - connectionTimeout = 10000, - } = options; +): (() => void) => { + const maxRetries = options.maxRetries ?? 3; + const initialRetryDelay = options.initialRetryDelay ?? 1000; + const maxRetryDelay = options.maxRetryDelay ?? 8000; + const connectionTimeout = options.connectionTimeout ?? 10000; let eventSource: EventSource | null = null; let retryCount = 0; @@ -545,6 +797,10 @@ export function connectTerminalStream( }; const cleanup = () => { + if (isClosed) { + return; + } + isClosed = true; clearTimeouts(); if (eventSource) { @@ -553,81 +809,15 @@ export function connectTerminalStream( } }; - const connect = () => { - if (isClosed || terminalExited) { - return; - } - - if (eventSource && eventSource.readyState !== EventSource.CLOSED) { - console.warn('Attempted to create duplicate EventSource, skipping'); - return; - } - - hasDispatchedOpen = false; - eventSource = new EventSource(`/api/terminal/${sessionId}/stream`); - - connectionTimeoutId = setTimeout(() => { - if (!hasDispatchedOpen && eventSource?.readyState !== EventSource.OPEN) { - console.error('Terminal connection timeout'); - eventSource?.close(); - handleError(new Error('Connection timeout'), false); - } - }, connectionTimeout); - - eventSource.onopen = () => { - if (hasDispatchedOpen) { - return; - } - hasDispatchedOpen = true; - retryCount = 0; - clearTimeouts(); - - onEvent({ type: 'connected' }); - }; - - eventSource.onmessage = (event) => { - try { - const data = JSON.parse(event.data) as TerminalStreamEvent; - - if (data.type === 'exit') { - getTerminalInputWsGlobalState().manager?.unbindSession(sessionId); - terminalExited = true; - cleanup(); - } - - onEvent(data); - } catch (error) { - console.error('Failed to parse terminal event:', error); - onError?.(error as Error, false); - } - }; - - eventSource.onerror = (error) => { - console.error('Terminal stream error:', error, 'readyState:', eventSource?.readyState); - clearTimeouts(); - - const isFatalError = terminalExited || eventSource?.readyState === EventSource.CLOSED; - - eventSource?.close(); - eventSource = null; - - if (!terminalExited) { - handleError(new Error('Terminal stream connection error'), isFatalError); - } - }; - }; - const handleError = (error: Error, isFatal: boolean) => { if (isClosed || terminalExited) { return; } if (retryCount < maxRetries && !isFatal) { - retryCount++; + retryCount += 1; const delay = Math.min(initialRetryDelay * Math.pow(2, retryCount - 1), maxRetryDelay); - console.log(`Reconnecting to terminal stream (attempt ${retryCount}/${maxRetries}) in ${delay}ms`); - onEvent({ type: 'reconnecting', attempt: retryCount, @@ -639,24 +829,100 @@ export function connectTerminalStream( connect(); } }, delay); - } else { - - console.error(`Terminal connection failed after ${retryCount} attempts`); - onError?.(error, true); - cleanup(); + return; } + + onError?.(error, true); + cleanup(); + }; + + const connect = () => { + if (isClosed || terminalExited) { + return; + } + + if (eventSource && eventSource.readyState !== EventSource.CLOSED) { + return; + } + + eventSource = new EventSource(`/api/terminal/${sessionId}/stream`); + + connectionTimeoutId = setTimeout(() => { + if (!hasDispatchedOpen && eventSource?.readyState !== EventSource.OPEN) { + eventSource?.close(); + handleError(new Error('Connection timeout'), false); + } + }, connectionTimeout); + + eventSource.onopen = () => { + if (hasDispatchedOpen) { + return; + } + + hasDispatchedOpen = true; + retryCount = 0; + clearTimeouts(); + onEvent({ type: 'connected' }); + }; + + eventSource.onmessage = (event) => { + try { + const data = JSON.parse(event.data) as TerminalStreamEvent; + + if (data.type === 'exit') { + getTerminalTransportGlobalState().manager?.unbindSession(sessionId); + terminalExited = true; + cleanup(); + } + + onEvent(data); + } catch (error) { + onError?.(error as Error, false); + } + }; + + eventSource.onerror = () => { + clearTimeouts(); + const isFatalError = terminalExited || eventSource?.readyState === EventSource.CLOSED; + eventSource?.close(); + eventSource = null; + + if (!terminalExited) { + handleError(new Error('Terminal stream connection error'), isFatalError); + } + }; }; connect(); - return cleanup; +}; + +export function connectTerminalStream( + sessionId: string, + onEvent: (event: TerminalStreamEvent) => void, + onError?: (error: Error, fatal?: boolean) => void, + options: ConnectStreamOptions = {} +): () => void { + const globalState = getTerminalTransportGlobalState(); + if (!isWsTransportSupported(globalState.streamCapability)) { + return connectTerminalStreamViaSse(sessionId, onEvent, onError, options); + } + + const manager = ensureTerminalTransportManager(); + const socketUrl = normalizeWebSocketPath(getPreferredTerminalWsPath(globalState)); + if (!socketUrl) { + return connectTerminalStreamViaSse(sessionId, onEvent, onError, options); + } + + manager.configure(socketUrl); + return manager.subscribe(sessionId, onEvent, onError, options); } export async function sendTerminalInput( sessionId: string, data: string ): Promise { - const globalState = getTerminalInputWsGlobalState(); + const globalState = getTerminalTransportGlobalState(); if (globalState.manager && await globalState.manager.sendInput(sessionId, data)) { return; } @@ -682,7 +948,7 @@ export async function resizeTerminal( } export async function closeTerminal(sessionId: string): Promise { - getTerminalInputWsGlobalState().manager?.unbindSession(sessionId); + getTerminalTransportGlobalState().manager?.unbindSession(sessionId); const response = await fetch(`/api/terminal/${sessionId}`, { method: 'DELETE', @@ -698,7 +964,7 @@ export async function restartTerminalSession( currentSessionId: string, options: { cwd: string; cols?: number; rows?: number } ): Promise { - getTerminalInputWsGlobalState().manager?.unbindSession(currentSessionId); + getTerminalTransportGlobalState().manager?.unbindSession(currentSessionId); const response = await fetch(`/api/terminal/${currentSessionId}/restart`, { method: 'POST', @@ -716,7 +982,7 @@ export async function restartTerminalSession( } const session = await response.json() as TerminalSession; - applyTerminalInputCapability(session.capabilities?.input); + applyTerminalTransportCapabilities(session.capabilities); return session; } @@ -736,39 +1002,42 @@ export async function forceKillTerminal(options: { } if (options.sessionId) { - getTerminalInputWsGlobalState().manager?.unbindSession(options.sessionId); + getTerminalTransportGlobalState().manager?.unbindSession(options.sessionId); } } export function disposeTerminalInputTransport(): void { - const globalState = getTerminalInputWsGlobalState(); + const globalState = getTerminalTransportGlobalState(); globalState.manager?.close(); globalState.manager = null; - globalState.capability = null; + globalState.inputCapability = null; + globalState.streamCapability = null; } export function primeTerminalInputTransport(): void { - const globalState = getTerminalInputWsGlobalState(); - if (globalState.capability && !isWsInputSupported(globalState.capability)) { + const globalState = getTerminalTransportGlobalState(); + if ( + globalState.inputCapability && + globalState.streamCapability && + !isWsTransportSupported(globalState.inputCapability) && + !isWsTransportSupported(globalState.streamCapability) + ) { return; } - const wsPath = globalState.capability?.ws?.path ?? DEFAULT_TERMINAL_INPUT_WS_PATH; - const socketUrl = normalizeWebSocketPath(wsPath); + const preferredPath = getPreferredTerminalWsPath(globalState) || DEFAULT_TERMINAL_WS_PATH; + const socketUrl = normalizeWebSocketPath(preferredPath); if (!socketUrl) { return; } - if (!globalState.manager) { - globalState.manager = new TerminalInputWsManager(); - } - - if (globalState.manager.isConnectedOrConnecting(socketUrl)) { + const manager = ensureTerminalTransportManager(); + if (manager.isConnectedOrConnecting(socketUrl)) { return; } - globalState.manager.configure(socketUrl); - globalState.manager.prime(); + manager.configure(socketUrl); + manager.prime(); } const hotModule = (import.meta as ImportMeta & { diff --git a/packages/ui/src/stores/useTerminalStore.ts b/packages/ui/src/stores/useTerminalStore.ts index 58681463..3f42cc3e 100644 --- a/packages/ui/src/stores/useTerminalStore.ts +++ b/packages/ui/src/stores/useTerminalStore.ts @@ -9,9 +9,12 @@ export interface TerminalChunk { data: string; } +export type TerminalTabLifecycle = 'idle' | 'running' | 'exited'; + export type TerminalTab = { id: string; terminalSessionId: string | null; + lifecycle: TerminalTabLifecycle; label: string; bufferChunks: TerminalChunk[]; bufferLength: number; @@ -40,6 +43,7 @@ interface TerminalStore { closeTab: (directory: string, tabId: string) => Promise; setTabSessionId: (directory: string, tabId: string, sessionId: string | null) => void; + setTabLifecycle: (directory: string, tabId: string, lifecycle: TerminalTabLifecycle) => void; setConnecting: (directory: string, tabId: string, isConnecting: boolean) => void; appendToBuffer: (directory: string, tabId: string, chunk: string) => void; clearBuffer: (directory: string, tabId: string) => void; @@ -52,7 +56,7 @@ const TERMINAL_BUFFER_LIMIT = 1_000_000; const TERMINAL_STORE_NAME = 'terminal-store'; let hydrationListenerAttached = false; -type PersistedTerminalTab = Pick; +type PersistedTerminalTab = Pick; type PersistedDirectoryTerminalState = { tabs: PersistedTerminalTab[]; @@ -85,6 +89,7 @@ function normalizeDirectory(dir: string): string { const createEmptyTab = (id: string, label: string): TerminalTab => ({ id, terminalSessionId: null, + lifecycle: 'idle', label, bufferChunks: [], bufferLength: 0, @@ -298,9 +303,14 @@ export const useTerminalStore = create()( const tab = existing.tabs[idx]; const shouldResetBuffer = sessionId !== null && tab.terminalSessionId !== sessionId; + const nextLifecycle = sessionId + ? 'running' + : (tab.terminalSessionId ? 'exited' : tab.lifecycle); + const nextTab: TerminalTab = { ...tab, terminalSessionId: sessionId, + lifecycle: nextLifecycle, isConnecting: false, ...(shouldResetBuffer ? { bufferChunks: [], bufferLength: 0 } : {}), }; @@ -312,6 +322,27 @@ export const useTerminalStore = create()( }); }, + setTabLifecycle: (directory: string, tabId: string, lifecycle: TerminalTabLifecycle) => { + const key = normalizeDirectory(directory); + set((state) => { + const newSessions = new Map(state.sessions); + const existing = newSessions.get(key); + if (!existing) { + return state; + } + + const idx = findTabIndex(existing, tabId); + if (idx < 0) { + return state; + } + + const nextTabs = [...existing.tabs]; + nextTabs[idx] = { ...nextTabs[idx], lifecycle, isConnecting: false }; + newSessions.set(key, { ...existing, tabs: nextTabs }); + return { sessions: newSessions }; + }); + }, + setConnecting: (directory: string, tabId: string, isConnecting: boolean) => { const key = normalizeDirectory(directory); set((state) => { @@ -428,6 +459,7 @@ export const useTerminalStore = create()( id: tab.id, label: tab.label, terminalSessionId: tab.terminalSessionId, + lifecycle: tab.lifecycle, createdAt: tab.createdAt, })), }, @@ -474,13 +506,21 @@ export const useTerminalStore = create()( maxTabNum = Math.max(maxTabNum, num); } + const terminalSessionId = + typeof rawTab.terminalSessionId === 'string' || rawTab.terminalSessionId === null + ? (rawTab.terminalSessionId as string | null) + : null; + const lifecycleRaw = rawTab.lifecycle; + const lifecycle = + lifecycleRaw === 'idle' || lifecycleRaw === 'running' || lifecycleRaw === 'exited' + ? lifecycleRaw + : (terminalSessionId ? 'running' : 'idle'); + tabs.push({ id, label: typeof rawTab.label === 'string' ? rawTab.label : 'Terminal', - terminalSessionId: - typeof rawTab.terminalSessionId === 'string' || rawTab.terminalSessionId === null - ? (rawTab.terminalSessionId as string | null) - : null, + terminalSessionId, + lifecycle, createdAt: typeof rawTab.createdAt === 'number' ? rawTab.createdAt : Date.now(), bufferChunks: [], bufferLength: 0, diff --git a/packages/web/server/TERMINAL_INPUT_WS_PROTOCOL.md b/packages/web/server/TERMINAL_INPUT_WS_PROTOCOL.md deleted file mode 100644 index a518ca1a..00000000 --- a/packages/web/server/TERMINAL_INPUT_WS_PROTOCOL.md +++ /dev/null @@ -1,44 +0,0 @@ -# Terminal Input WS Protocol - -## Goal -Reduce terminal input latency by replacing per-keystroke HTTP requests with a persistent WebSocket input channel, while keeping SSE output and HTTP endpoints as compatibility fallback. - -## Scope -- Input path: WebSocket (`/api/terminal/input-ws`) -- Output path: SSE (`/api/terminal/:sessionId/stream`) -- HTTP input fallback remains: `POST /api/terminal/:sessionId/input` - -## Framing -- Text frame: terminal keystroke payload (hot path) - - Examples: `"\r"`, `"\u001b[A"`, `"\u0003"` -- Binary frame: control envelope - - Byte 0: tag (`0x01` = JSON control) - - Bytes 1..N: UTF-8 JSON payload - -## Control Messages -- Bind active socket to terminal session: - - client -> server: `{"t":"b","s":"","v":1}` -- Keepalive ping: - - client -> server: `{"t":"p","v":1}` - - server -> client: `{"t":"po","v":1}` -- Server control responses: - - ready: `{"t":"ok","v":1}` - - bind ok: `{"t":"bok","v":1}` - - error: `{"t":"e","c":"","f":true|false}` - -## Multiplexing Model -- Single shared socket per client runtime. -- Socket has one mutable `boundSessionId`. -- Client sends bind control when active terminal changes. -- Keystroke frames apply to currently bound session. -- Client keeps socket open and sends periodic keepalive pings so the channel stays ready for next input. -- Client primes/opens this socket when the Terminal tab is opened (not per keystroke). - -## Security -- UI auth session required when UI password is enabled. -- Origin validation enforced for cookie-authenticated browser upgrades. -- Invalid/malformed frames are rate-limited and may close socket. - -## Fallback Behavior -- On WS unavailable/error/close, client falls back to HTTP input immediately. -- Existing terminal behavior remains functional during mixed-version rollout. diff --git a/packages/web/server/TERMINAL_WS_PROTOCOL.md b/packages/web/server/TERMINAL_WS_PROTOCOL.md new file mode 100644 index 00000000..39674ae2 --- /dev/null +++ b/packages/web/server/TERMINAL_WS_PROTOCOL.md @@ -0,0 +1,48 @@ +# Terminal WebSocket Transport Protocol + +## Goal +Use a single persistent WebSocket for terminal input and output, while keeping the legacy SSE output route and HTTP input route as compatibility fallbacks. + +## Scope +- Primary full-duplex path: WebSocket (`/api/terminal/ws`) +- Legacy output fallback: SSE (`/api/terminal/:sessionId/stream`) +- HTTP input fallback remains: `POST /api/terminal/:sessionId/input` + +## Framing +- Text frame: + - client -> server: terminal keystroke payload + - server -> client: raw PTY output chunk +- Binary frame: control envelope + - Byte 0: tag (`0x01` = JSON control) + - Bytes 1..N: UTF-8 JSON payload + +## Control Messages +- Bind active socket to terminal session: + - client -> server: `{"t":"b","s":"","v":2}` +- Keepalive ping: + - client -> server: `{"t":"p","v":2}` + - server -> client: `{"t":"po","v":2}` +- Server control responses: + - ready: `{"t":"ok","v":2}` + - bind ok: `{"t":"bok","s":"","runtime":"node|bun","ptyBackend":"...","v":2}` + - exit: `{"t":"x","s":"","exitCode":0,"signal":null}` + - error: `{"t":"e","c":"","f":true|false}` + +## Multiplexing Model +- Single shared socket per client runtime. +- Socket has one mutable bound session. +- Client sends a bind control when the active terminal changes. +- Text frames always apply to the currently bound session. +- PTY output is pushed back over the same socket as text frames. +- Client keeps the socket primed so both stream subscription and input reuse the same transport. + +## Security +- UI auth session required when UI password is enabled. +- Origin validation enforced for cookie-authenticated browser upgrades. +- Invalid or malformed frames are rate-limited and may close the socket. + +## Fallback Behavior +- New clients prefer `capabilities.stream.ws` and reuse the same socket for input. +- If stream WebSocket capability is unavailable, clients fall back to SSE output. +- If terminal input cannot be sent over WebSocket, clients fall back to HTTP input. +- The removed `/api/terminal/input-ws` path should fail with `404 Not Found`. diff --git a/packages/web/server/lib/terminal/DOCUMENTATION.md b/packages/web/server/lib/terminal/DOCUMENTATION.md index 2c477f7d..6cfa5474 100644 --- a/packages/web/server/lib/terminal/DOCUMENTATION.md +++ b/packages/web/server/lib/terminal/DOCUMENTATION.md @@ -1,115 +1,76 @@ # Terminal Module Documentation ## Purpose -This module provides WebSocket protocol utilities for terminal input handling in the web server runtime, including message normalization, control frame parsing, rate limiting, and pathname resolution for terminal WebSocket connections. +This module provides WebSocket transport utilities for terminal input and output in the web server runtime, including message normalization, control frame parsing, rate limiting, pathname resolution, and short-lived output replay buffering for terminal WebSocket connections. ## Entrypoints and structure - `packages/web/server/lib/terminal/`: Terminal module directory. - - `index.js`: Stable module entrypoint that re-exports protocol helpers/constants. + - `index.js`: Stable module entrypoint that re-exports protocol helpers and replay-buffer helpers. - `runtime.js`: Runtime module that owns terminal session state, WS server setup, and `/api/terminal/*` route registration. - - `input-ws-protocol.js`: Single-file module containing all terminal input WebSocket protocol utilities. -- `packages/web/server/lib/terminal/input-ws-protocol.test.js`: Test file for protocol utilities. + - `terminal-ws-protocol.js`: Single-file module containing terminal WebSocket protocol utilities. + - `output-replay-buffer.js`: Helper module for buffering recent terminal output so late subscribers can receive startup prompt data. +- `packages/web/server/lib/terminal/terminal-ws-protocol.test.js`: Test file for protocol utilities. +- `packages/web/server/lib/terminal/output-replay-buffer.test.js`: Test file for replay buffer helpers. Public API entry point: imported by `packages/web/server/index.js` from `./lib/terminal/index.js`. ## Public exports ### Constants -- `TERMINAL_INPUT_WS_PATH`: WebSocket endpoint path (`/api/terminal/input-ws`). -- `TERMINAL_INPUT_WS_CONTROL_TAG_JSON`: Control frame tag byte (0x01) indicating JSON payload. -- `TERMINAL_INPUT_WS_MAX_PAYLOAD_BYTES`: Maximum payload size (64KB). +- `TERMINAL_WS_PATH`: Primary WebSocket endpoint path (`/api/terminal/ws`). +- `TERMINAL_WS_CONTROL_TAG_JSON`: Control frame tag byte (`0x01`) indicating JSON payload. +- `TERMINAL_WS_MAX_PAYLOAD_BYTES`: Maximum inbound WebSocket payload size (64KB). +- `TERMINAL_OUTPUT_REPLAY_MAX_BYTES`: Maximum buffered terminal output retained for replay (64KB). ### Request Parsing - `parseRequestPathname(requestUrl)`: Extracts pathname from request URL string. Returns empty string for invalid inputs. +- `isTerminalWsPathname(pathname)`: Returns whether a pathname matches a supported terminal WebSocket route. ### Message Normalization -- `normalizeTerminalInputWsMessageToBuffer(rawData)`: Normalizes various data types (Buffer, Uint8Array, ArrayBuffer, string, chunk arrays) to a single Buffer. -- `normalizeTerminalInputWsMessageToText(rawData)`: Normalizes data to UTF-8 text string. Passes through strings directly, converts binary data to text. +- `normalizeTerminalWsMessageToBuffer(rawData)`: Normalizes various data types (Buffer, Uint8Array, ArrayBuffer, string, chunk arrays) to a single Buffer. +- `normalizeTerminalWsMessageToText(rawData)`: Normalizes data to UTF-8 text string. ### Control Frame Handling -- `readTerminalInputWsControlFrame(rawData)`: Parses WebSocket message as control frame. Returns parsed JSON object or null if invalid/malformed. Validates control tag prefix and JSON structure. -- `createTerminalInputWsControlFrame(payload)`: Creates a control frame with JSON payload. Prepends control tag byte. +- `readTerminalWsControlFrame(rawData)`: Parses WebSocket message as control frame. Returns parsed JSON object or null if invalid or malformed. +- `createTerminalWsControlFrame(payload)`: Creates a control frame with JSON payload and prepends the control tag byte. + +### Replay Buffer Helpers +- `createTerminalOutputReplayBuffer()`: Creates mutable state for recent terminal output replay. +- `appendTerminalOutputReplayChunk(bufferState, data, maxBytes?)`: Appends a chunk, trimming older buffered data to stay within the configured byte budget. +- `listTerminalOutputReplayChunksSince(bufferState, lastSeenId)`: Returns buffered chunks newer than the provided replay cursor. +- `getLatestTerminalOutputReplayChunkId(bufferState)`: Returns the latest chunk id in the replay buffer, or `0` when empty. ### Rate Limiting - `pruneRebindTimestamps(timestamps, now, windowMs)`: Filters timestamps to keep only those within the active time window. -- `isRebindRateLimited(timestamps, maxPerWindow)`: Checks if rebind operations have exceeded rate limit threshold. - -## Response contracts - -### Control Frame -Control frames use binary encoding: -- First byte: `TERMINAL_INPUT_WS_CONTROL_TAG_JSON` (0x01) -- Remaining bytes: UTF-8 encoded JSON object -- Parsed result: Object or null on parse failure - -### Normalized Buffer -Input types are normalized to Buffer: -- `Buffer`: Returned as-is -- `Uint8Array`/`ArrayBuffer`: Converted to Buffer -- `String`: Converted to UTF-8 Buffer -- `Array`: Concatenated to single Buffer - -### Rate Limiting -Rate limiting uses timestamp arrays: -- `pruneRebindTimestamps`: Returns filtered array of active timestamps -- `isRebindRateLimited`: Returns boolean indicating if limit is reached +- `isRebindRateLimited(timestamps, maxPerWindow)`: Checks if rebind operations have exceeded the configured threshold. ## Usage in web server - -The terminal protocol utilities are used by `packages/web/server/index.js` for: -- WebSocket endpoint path definition (`TERMINAL_INPUT_WS_PATH`) -- Message normalization for input handling -- Control frame parsing for session binding +The terminal helpers are used by `packages/web/server/index.js` for: +- WebSocket endpoint path definition and matching +- Message normalization for terminal input payloads +- Control frame parsing for session binding, keepalive, and exit signaling - Rate limiting for session rebind operations - Request pathname parsing for WebSocket routing +- Replaying startup output such as shell prompts when the client binds after the PTY already emitted data -The web server uses these utilities in combination with `bun-pty` or `node-pty` for PTY session management. +The web server combines these utilities with `bun-pty` or `node-pty` to drive full-duplex PTY sessions. ## Notes for contributors - -### Adding New Control Frame Types -1. Define new control tag constants (e.g., `TERMINAL_INPUT_WS_CONTROL_TAG_CUSTOM = 0x02`) -2. Update `readTerminalInputWsControlFrame` to handle new tag type -3. Update `createTerminalInputWsControlFrame` or create new frame creation function -4. Add corresponding tests in `terminal-input-ws-protocol.test.js` - -### Message Normalization -- Always normalize incoming WebSocket messages before processing -- Use `normalizeTerminalInputWsMessageToBuffer` for binary data -- Use `normalizeTerminalInputWsMessageToText` for text data (terminal escape sequences) -- Normalize chunked messages from WebSocket fragmentation handling - -### Rate Limiting -- Rate limiting is time-window based: tracks timestamps within a rolling window -- Use `pruneRebindTimestamps` to clean up stale timestamps before rate limit checks -- Configure `maxPerWindow` based on operational requirements (prevent abuse) - -### Error Handling -- `readTerminalInputWsControlFrame` returns null for invalid/malformed frames -- `parseRequestPathname` returns empty string for invalid URLs -- Callers should handle null/empty returns gracefully - -### Testing -- Run `bun run type-check`, `bun run lint`, and `bun run build` before finalizing changes -- Test edge cases: empty payloads, malformed JSON, chunked messages, rate limit boundaries -- Verify control frame roundtrip: create → read → validate payload equality -- Test pathname parsing with relative URLs, absolute URLs, and invalid inputs +- Keep control frames backward-compatible when possible; use explicit `v` values for protocol changes. +- Always normalize incoming WebSocket messages before processing them. +- Keep replay buffering small and memory-only; it exists to cover startup races, not to implement persistent scrollback. +- Add tests for new control frame types, websocket path changes, malformed payload handling, and replay trimming semantics. +- Keep HTTP input and SSE output fallbacks functional unless the rollout explicitly removes them. ## Verification notes - ### Manual verification -1. Start web server and create terminal session via `/api/terminal/create` -2. Connect to `/api/terminal/input-ws` WebSocket -3. Send control frames with valid/invalid payloads to verify parsing -4. Test message normalization with various data types -5. Verify rate limiting by issuing rapid rebind requests +1. Start the web server and create a terminal session via `/api/terminal/create`. +2. Wait briefly before binding the client to ensure the shell emits its prompt first. +3. Connect to `/api/terminal/ws` WebSocket and bind to the session. +4. Verify the startup prompt and early shell output are replayed before interactive input begins. +5. Verify `/api/terminal/input-ws` is rejected with `404 Not Found` and `/api/terminal/:sessionId/stream` still works as a fallback path. ### Automated verification -- Run test file: `bun test packages/web/server/lib/terminal/input-ws-protocol.test.js` -- Protocol tests should pass covering: - - WebSocket path constant - - Control frame encoding/decoding - - Payload validation - - Message normalization (all data types) - - Pathname parsing - - Rate limiting logic +- Run `bun test packages/web/server/lib/terminal/terminal-ws-protocol.test.js` +- Run `bun test packages/web/server/lib/terminal/output-replay-buffer.test.js` +- Run `bun run type-check`, `bun run lint`, and `bun run build` before finalizing changes. diff --git a/packages/web/server/lib/terminal/index.js b/packages/web/server/lib/terminal/index.js index 9051c6ae..d23d9eb2 100644 --- a/packages/web/server/lib/terminal/index.js +++ b/packages/web/server/lib/terminal/index.js @@ -1,12 +1,31 @@ export { - TERMINAL_INPUT_WS_PATH, - TERMINAL_INPUT_WS_CONTROL_TAG_JSON, - TERMINAL_INPUT_WS_MAX_PAYLOAD_BYTES, + TERMINAL_WS_PATH, + TERMINAL_WS_CONTROL_TAG_JSON, + TERMINAL_WS_MAX_PAYLOAD_BYTES, + isTerminalWsPathname, parseRequestPathname, - normalizeTerminalInputWsMessageToBuffer, - normalizeTerminalInputWsMessageToText, - readTerminalInputWsControlFrame, - createTerminalInputWsControlFrame, + normalizeTerminalWsMessageToBuffer, + normalizeTerminalWsMessageToText, + readTerminalWsControlFrame, + createTerminalWsControlFrame, pruneRebindTimestamps, isRebindRateLimited, -} from './input-ws-protocol.js'; +} from './terminal-ws-protocol.js'; + +export { + TERMINAL_WS_PATH as TERMINAL_INPUT_WS_PATH, + TERMINAL_WS_CONTROL_TAG_JSON as TERMINAL_INPUT_WS_CONTROL_TAG_JSON, + TERMINAL_WS_MAX_PAYLOAD_BYTES as TERMINAL_INPUT_WS_MAX_PAYLOAD_BYTES, + normalizeTerminalWsMessageToBuffer as normalizeTerminalInputWsMessageToBuffer, + normalizeTerminalWsMessageToText as normalizeTerminalInputWsMessageToText, + readTerminalWsControlFrame as readTerminalInputWsControlFrame, + createTerminalWsControlFrame as createTerminalInputWsControlFrame, +} from './terminal-ws-protocol.js'; + +export { + TERMINAL_OUTPUT_REPLAY_MAX_BYTES, + createTerminalOutputReplayBuffer, + appendTerminalOutputReplayChunk, + listTerminalOutputReplayChunksSince, + getLatestTerminalOutputReplayChunkId, +} from './output-replay-buffer.js'; diff --git a/packages/web/server/lib/terminal/output-replay-buffer.js b/packages/web/server/lib/terminal/output-replay-buffer.js new file mode 100644 index 00000000..d7c64bf0 --- /dev/null +++ b/packages/web/server/lib/terminal/output-replay-buffer.js @@ -0,0 +1,66 @@ +export const TERMINAL_OUTPUT_REPLAY_MAX_BYTES = 64 * 1024; + +const trimTerminalOutputChunkToMaxBytes = (data, maxBytes) => { + if (typeof data !== 'string' || data.length === 0) { + return ''; + } + + const bytes = Buffer.byteLength(data, 'utf8'); + if (bytes <= maxBytes) { + return data; + } + + const trimmedBuffer = Buffer.from(data, 'utf8').subarray(-maxBytes); + return trimmedBuffer.toString('utf8'); +}; + +export const createTerminalOutputReplayBuffer = () => ({ + chunks: [], + totalBytes: 0, + nextId: 1, +}); + +export const appendTerminalOutputReplayChunk = (bufferState, data, maxBytes = TERMINAL_OUTPUT_REPLAY_MAX_BYTES) => { + if (!bufferState || typeof bufferState !== 'object') { + return null; + } + + const normalizedData = trimTerminalOutputChunkToMaxBytes(data, maxBytes); + if (!normalizedData) { + return null; + } + + const bytes = Buffer.byteLength(normalizedData, 'utf8'); + const chunk = { + id: bufferState.nextId, + data: normalizedData, + bytes, + }; + + bufferState.nextId += 1; + bufferState.chunks.push(chunk); + bufferState.totalBytes += bytes; + + while (bufferState.totalBytes > maxBytes && bufferState.chunks.length > 1) { + const removedChunk = bufferState.chunks.shift(); + bufferState.totalBytes -= removedChunk?.bytes ?? 0; + } + + return chunk; +}; + +export const listTerminalOutputReplayChunksSince = (bufferState, lastSeenId = 0) => { + if (!bufferState || typeof bufferState !== 'object' || !Array.isArray(bufferState.chunks)) { + return []; + } + + return bufferState.chunks.filter((chunk) => chunk.id > lastSeenId); +}; + +export const getLatestTerminalOutputReplayChunkId = (bufferState) => { + if (!bufferState || typeof bufferState !== 'object' || !Array.isArray(bufferState.chunks) || bufferState.chunks.length === 0) { + return 0; + } + + return bufferState.chunks[bufferState.chunks.length - 1]?.id ?? 0; +}; diff --git a/packages/web/server/lib/terminal/output-replay-buffer.test.js b/packages/web/server/lib/terminal/output-replay-buffer.test.js new file mode 100644 index 00000000..98dd2456 --- /dev/null +++ b/packages/web/server/lib/terminal/output-replay-buffer.test.js @@ -0,0 +1,66 @@ +import { describe, expect, it } from 'bun:test'; + +import { + TERMINAL_OUTPUT_REPLAY_MAX_BYTES, + appendTerminalOutputReplayChunk, + createTerminalOutputReplayBuffer, + getLatestTerminalOutputReplayChunkId, + listTerminalOutputReplayChunksSince, +} from './output-replay-buffer.js'; + +describe('terminal output replay buffer', () => { + it('starts empty', () => { + const bufferState = createTerminalOutputReplayBuffer(); + expect(bufferState).toEqual({ chunks: [], totalBytes: 0, nextId: 1 }); + expect(getLatestTerminalOutputReplayChunkId(bufferState)).toBe(0); + }); + + it('appends chunks with incrementing ids', () => { + const bufferState = createTerminalOutputReplayBuffer(); + const first = appendTerminalOutputReplayChunk(bufferState, 'prompt> '); + const second = appendTerminalOutputReplayChunk(bufferState, 'ls\r\n'); + + expect(first).toEqual({ id: 1, data: 'prompt> ', bytes: 8 }); + expect(second).toEqual({ id: 2, data: 'ls\r\n', bytes: 4 }); + expect(getLatestTerminalOutputReplayChunkId(bufferState)).toBe(2); + }); + + it('lists chunks after a replay cursor', () => { + const bufferState = createTerminalOutputReplayBuffer(); + appendTerminalOutputReplayChunk(bufferState, 'prompt> '); + appendTerminalOutputReplayChunk(bufferState, 'ls\r\n'); + appendTerminalOutputReplayChunk(bufferState, 'file.txt\r\n'); + + expect(listTerminalOutputReplayChunksSince(bufferState, 1).map((chunk) => chunk.data)).toEqual([ + 'ls\r\n', + 'file.txt\r\n', + ]); + }); + + it('trims old chunks beyond max bytes', () => { + const bufferState = createTerminalOutputReplayBuffer(); + appendTerminalOutputReplayChunk(bufferState, '1234', 8); + appendTerminalOutputReplayChunk(bufferState, '5678', 8); + appendTerminalOutputReplayChunk(bufferState, '90', 8); + + expect(bufferState.chunks.map((chunk) => chunk.data)).toEqual(['5678', '90']); + expect(bufferState.totalBytes).toBe(6); + }); + + it('trims oversized single chunks to the configured max bytes', () => { + const bufferState = createTerminalOutputReplayBuffer(); + const chunk = appendTerminalOutputReplayChunk(bufferState, 'abcdefghij', 4); + + expect(chunk?.data).toBe('ghij'); + expect(chunk?.bytes).toBe(4); + expect(bufferState.totalBytes).toBe(4); + }); + + it('uses the default max bytes when not provided', () => { + const bufferState = createTerminalOutputReplayBuffer(); + const chunk = appendTerminalOutputReplayChunk(bufferState, 'ok'); + + expect(chunk?.bytes).toBe(2); + expect(TERMINAL_OUTPUT_REPLAY_MAX_BYTES).toBe(64 * 1024); + }); +}); diff --git a/packages/web/server/lib/terminal/runtime.js b/packages/web/server/lib/terminal/runtime.js index 9453761a..2d104b69 100644 --- a/packages/web/server/lib/terminal/runtime.js +++ b/packages/web/server/lib/terminal/runtime.js @@ -2,8 +2,12 @@ import { WebSocketServer } from 'ws'; import { TERMINAL_INPUT_WS_MAX_PAYLOAD_BYTES, TERMINAL_INPUT_WS_PATH, + TERMINAL_OUTPUT_REPLAY_MAX_BYTES, + appendTerminalOutputReplayChunk, + createTerminalOutputReplayBuffer, createTerminalInputWsControlFrame, isRebindRateLimited, + listTerminalOutputReplayChunksSince, normalizeTerminalInputWsMessageToText, parseRequestPathname, pruneRebindTimestamps, @@ -150,8 +154,10 @@ export function createTerminalRuntime({ }; const terminalSessions = new Map(); + const terminalWsConnections = new Set(); const MAX_TERMINAL_SESSIONS = 20; const TERMINAL_IDLE_TIMEOUT = 30 * 60 * 1000; + const terminalRuntimeName = typeof globalThis.Bun === 'undefined' ? 'node' : 'bun'; const sanitizeTerminalEnv = (env) => { const next = { ...env }; delete next.BASH_XTRACEFD; @@ -159,13 +165,22 @@ export function createTerminalRuntime({ delete next.ENV; return next; }; - const terminalInputCapabilities = { + const terminalTransportCapabilities = { input: { preferred: 'ws', transports: ['http', 'ws'], ws: { path: TERMINAL_INPUT_WS_PATH, - v: 1, + v: 2, + enc: 'text+json-bin-control', + }, + }, + stream: { + preferred: 'ws', + transports: ['sse', 'ws'], + ws: { + path: TERMINAL_INPUT_WS_PATH, + v: 2, enc: 'text+json-bin-control', }, }, @@ -189,13 +204,17 @@ export function createTerminalRuntime({ terminalInputWsServer.on('connection', (socket) => { const connectionState = { + socket, boundSessionId: null, invalidFrames: 0, rebindTimestamps: [], + replayCursorBySession: new Map(), lastActivityAt: Date.now(), }; - sendTerminalInputWsControl(socket, { t: 'ok', v: 1 }); + terminalWsConnections.add(connectionState); + + sendTerminalInputWsControl(socket, { t: 'ok', v: 2 }); const heartbeatInterval = setInterval(() => { if (socket.readyState !== 1) { @@ -231,7 +250,7 @@ export function createTerminalRuntime({ } if (controlMessage.t === 'p') { - sendTerminalInputWsControl(socket, { t: 'po', v: 1 }); + sendTerminalInputWsControl(socket, { t: 'po', v: 2 }); return; } @@ -268,9 +287,32 @@ export function createTerminalRuntime({ return; } + const replaySinceRaw = + typeof controlMessage.r === 'number' && Number.isFinite(controlMessage.r) + ? Math.max(0, Math.trunc(controlMessage.r)) + : 0; + const rememberedReplayCursor = connectionState.replayCursorBySession.get(nextSessionId) ?? 0; + const replaySince = Math.max(replaySinceRaw, rememberedReplayCursor); + connectionState.rebindTimestamps.push(now); connectionState.boundSessionId = nextSessionId; - sendTerminalInputWsControl(socket, { t: 'bok', v: 1 }); + sendTerminalInputWsControl(socket, { + t: 'bok', + v: 2, + s: nextSessionId, + runtime: terminalRuntimeName, + ptyBackend: targetSession.ptyBackend || 'unknown', + }); + + const replayChunks = listTerminalOutputReplayChunksSince(targetSession.outputReplayBuffer, replaySince); + for (const replayChunk of replayChunks) { + try { + socket.send(replayChunk.data); + connectionState.replayCursorBySession.set(nextSessionId, replayChunk.id); + } catch { + break; + } + } return; } @@ -301,6 +343,8 @@ export function createTerminalRuntime({ socket.on('close', () => { clearInterval(heartbeatInterval); + connectionState.boundSessionId = null; + terminalWsConnections.delete(connectionState); }); socket.on('error', (error) => { @@ -347,6 +391,56 @@ export function createTerminalRuntime({ void handleUpgrade(); }); + const wireTerminalSession = (sessionId, session) => { + session.ptyProcess.onData((data) => { + session.lastActivity = Date.now(); + const replayChunk = appendTerminalOutputReplayChunk( + session.outputReplayBuffer, + data, + TERMINAL_OUTPUT_REPLAY_MAX_BYTES + ); + + for (const wsConnection of terminalWsConnections) { + if (wsConnection.boundSessionId !== sessionId) { + continue; + } + + if (!wsConnection.socket || wsConnection.socket.readyState !== 1) { + continue; + } + + try { + wsConnection.socket.send(data); + if (replayChunk) { + wsConnection.replayCursorBySession.set(sessionId, replayChunk.id); + } + } catch { + } + } + }); + + session.ptyProcess.onExit(({ exitCode, signal }) => { + console.log(`Terminal session ${sessionId} exited with code ${exitCode}, signal ${signal}`); + for (const wsConnection of terminalWsConnections) { + if (wsConnection.boundSessionId !== sessionId) { + continue; + } + + wsConnection.boundSessionId = null; + wsConnection.replayCursorBySession.delete(sessionId); + sendTerminalInputWsControl(wsConnection.socket, { + t: 'x', + v: 2, + s: sessionId, + exitCode, + signal, + }); + } + + terminalSessions.delete(sessionId); + }); + }; + const idleSweepInterval = setInterval(() => { const now = Date.now(); for (const [sessionId, session] of terminalSessions.entries()) { @@ -399,17 +493,14 @@ export function createTerminalRuntime({ cwd, lastActivity: Date.now(), clients: new Set(), + outputReplayBuffer: createTerminalOutputReplayBuffer(), }; terminalSessions.set(sessionId, session); - - ptyProcess.onExit(({ exitCode, signal }) => { - console.log(`Terminal session ${sessionId} exited with code ${exitCode}, signal ${signal}`); - terminalSessions.delete(sessionId); - }); + wireTerminalSession(sessionId, session); console.log(`Created terminal session: ${sessionId} in ${cwd} using shell ${shell}`); - res.json({ sessionId, cols: cols || 80, rows: rows || 24, capabilities: terminalInputCapabilities }); + res.json({ sessionId, cols: cols || 80, rows: rows || 24, capabilities: terminalTransportCapabilities }); } catch (error) { console.error('Failed to create terminal session:', error); res.status(500).json({ error: error.message || 'Failed to create terminal session' }); @@ -433,9 +524,8 @@ export function createTerminalRuntime({ session.clients.add(clientId); session.lastActivity = Date.now(); - const runtime = typeof globalThis.Bun === 'undefined' ? 'node' : 'bun'; const ptyBackend = session.ptyBackend || 'unknown'; - res.write(`data: ${JSON.stringify({ type: 'connected', runtime, ptyBackend })}\n\n`); + res.write(`data: ${JSON.stringify({ type: 'connected', runtime: terminalRuntimeName, ptyBackend })}\n\n`); const heartbeatInterval = setInterval(() => { try { @@ -501,7 +591,7 @@ export function createTerminalRuntime({ req.on('close', cleanup); req.on('error', cleanup); - console.log(`Terminal connected: session=${sessionId} client=${clientId} runtime=${runtime} pty=${ptyBackend}`); + console.log(`Terminal connected: session=${sessionId} client=${clientId} runtime=${terminalRuntimeName} pty=${ptyBackend}`); }); app.post('/api/terminal/:sessionId/input', express.text({ type: '*/*' }), (req, res) => { @@ -613,17 +703,14 @@ export function createTerminalRuntime({ cwd, lastActivity: Date.now(), clients: new Set(), + outputReplayBuffer: createTerminalOutputReplayBuffer(), }; terminalSessions.set(newSessionId, session); - - ptyProcess.onExit(({ exitCode, signal }) => { - console.log(`Terminal session ${newSessionId} exited with code ${exitCode}, signal ${signal}`); - terminalSessions.delete(newSessionId); - }); + wireTerminalSession(newSessionId, session); console.log(`Restarted terminal session: ${sessionId} -> ${newSessionId} in ${cwd} using shell ${shell}`); - res.json({ sessionId: newSessionId, cols: cols || 80, rows: rows || 24, capabilities: terminalInputCapabilities }); + res.json({ sessionId: newSessionId, cols: cols || 80, rows: rows || 24, capabilities: terminalTransportCapabilities }); } catch (error) { console.error('Failed to restart terminal session:', error); res.status(500).json({ error: error.message || 'Failed to restart terminal session' }); diff --git a/packages/web/server/lib/terminal/input-ws-protocol.js b/packages/web/server/lib/terminal/terminal-ws-protocol.js similarity index 59% rename from packages/web/server/lib/terminal/input-ws-protocol.js rename to packages/web/server/lib/terminal/terminal-ws-protocol.js index cdf69699..263565b6 100644 --- a/packages/web/server/lib/terminal/input-ws-protocol.js +++ b/packages/web/server/lib/terminal/terminal-ws-protocol.js @@ -1,6 +1,6 @@ -export const TERMINAL_INPUT_WS_PATH = '/api/terminal/input-ws'; -export const TERMINAL_INPUT_WS_CONTROL_TAG_JSON = 0x01; -export const TERMINAL_INPUT_WS_MAX_PAYLOAD_BYTES = 64 * 1024; +export const TERMINAL_WS_PATH = '/api/terminal/ws'; +export const TERMINAL_WS_CONTROL_TAG_JSON = 0x01; +export const TERMINAL_WS_MAX_PAYLOAD_BYTES = 64 * 1024; export const parseRequestPathname = (requestUrl) => { if (typeof requestUrl !== 'string' || requestUrl.length === 0) { @@ -14,7 +14,9 @@ export const parseRequestPathname = (requestUrl) => { } }; -export const normalizeTerminalInputWsMessageToBuffer = (rawData) => { +export const isTerminalWsPathname = (pathname) => pathname === TERMINAL_WS_PATH; + +export const normalizeTerminalWsMessageToBuffer = (rawData) => { if (Buffer.isBuffer(rawData)) { return rawData; } @@ -26,21 +28,21 @@ export const normalizeTerminalInputWsMessageToBuffer = (rawData) => { return Buffer.from(rawData); }; -export const normalizeTerminalInputWsMessageToText = (rawData) => { +export const normalizeTerminalWsMessageToText = (rawData) => { if (typeof rawData === 'string') { return rawData; } - return normalizeTerminalInputWsMessageToBuffer(rawData).toString('utf8'); + return normalizeTerminalWsMessageToBuffer(rawData).toString('utf8'); }; -export const readTerminalInputWsControlFrame = (rawData) => { +export const readTerminalWsControlFrame = (rawData) => { if (!rawData) { return null; } - const buffer = normalizeTerminalInputWsMessageToBuffer(rawData); - if (buffer.length < 2 || buffer[0] !== TERMINAL_INPUT_WS_CONTROL_TAG_JSON) { + const buffer = normalizeTerminalWsMessageToBuffer(rawData); + if (buffer.length < 2 || buffer[0] !== TERMINAL_WS_CONTROL_TAG_JSON) { return null; } @@ -55,9 +57,9 @@ export const readTerminalInputWsControlFrame = (rawData) => { } }; -export const createTerminalInputWsControlFrame = (payload) => { +export const createTerminalWsControlFrame = (payload) => { const jsonBytes = Buffer.from(JSON.stringify(payload), 'utf8'); - return Buffer.concat([Buffer.from([TERMINAL_INPUT_WS_CONTROL_TAG_JSON]), jsonBytes]); + return Buffer.concat([Buffer.from([TERMINAL_WS_CONTROL_TAG_JSON]), jsonBytes]); }; export const pruneRebindTimestamps = (timestamps, now, windowMs) => diff --git a/packages/web/server/lib/terminal/input-ws-protocol.test.js b/packages/web/server/lib/terminal/terminal-ws-protocol.test.js similarity index 58% rename from packages/web/server/lib/terminal/input-ws-protocol.test.js rename to packages/web/server/lib/terminal/terminal-ws-protocol.test.js index b9844ce4..2fe9f969 100644 --- a/packages/web/server/lib/terminal/input-ws-protocol.test.js +++ b/packages/web/server/lib/terminal/terminal-ws-protocol.test.js @@ -1,86 +1,93 @@ import { describe, expect, it } from 'bun:test'; import { - TERMINAL_INPUT_WS_CONTROL_TAG_JSON, - TERMINAL_INPUT_WS_PATH, - createTerminalInputWsControlFrame, + TERMINAL_WS_PATH, + TERMINAL_WS_CONTROL_TAG_JSON, + createTerminalWsControlFrame, + isTerminalWsPathname, isRebindRateLimited, - normalizeTerminalInputWsMessageToBuffer, - normalizeTerminalInputWsMessageToText, + normalizeTerminalWsMessageToBuffer, + normalizeTerminalWsMessageToText, parseRequestPathname, pruneRebindTimestamps, - readTerminalInputWsControlFrame, -} from './input-ws-protocol.js'; + readTerminalWsControlFrame, +} from './terminal-ws-protocol.js'; -describe('terminal input websocket protocol', () => { - it('uses fixed websocket path', () => { - expect(TERMINAL_INPUT_WS_PATH).toBe('/api/terminal/input-ws'); +describe('terminal websocket protocol', () => { + it('uses fixed websocket paths', () => { + expect(TERMINAL_WS_PATH).toBe('/api/terminal/ws'); + }); + + it('matches supported websocket pathnames', () => { + expect(isTerminalWsPathname('/api/terminal/ws')).toBe(true); + expect(isTerminalWsPathname('/api/terminal/input-ws')).toBe(false); + expect(isTerminalWsPathname('/api/terminal/other')).toBe(false); }); it('encodes control frames with control tag prefix', () => { - const frame = createTerminalInputWsControlFrame({ t: 'ok', v: 1 }); - expect(frame[0]).toBe(TERMINAL_INPUT_WS_CONTROL_TAG_JSON); + const frame = createTerminalWsControlFrame({ t: 'ok', v: 1 }); + expect(frame[0]).toBe(TERMINAL_WS_CONTROL_TAG_JSON); }); it('roundtrips control frame payload', () => { const payload = { t: 'b', s: 'abc123', v: 1 }; - const frame = createTerminalInputWsControlFrame(payload); - expect(readTerminalInputWsControlFrame(frame)).toEqual(payload); + const frame = createTerminalWsControlFrame(payload); + expect(readTerminalWsControlFrame(frame)).toEqual(payload); }); it('rejects control frame without protocol tag', () => { const frame = Buffer.from(JSON.stringify({ t: 'b', s: 'abc123' }), 'utf8'); - expect(readTerminalInputWsControlFrame(frame)).toBeNull(); + expect(readTerminalWsControlFrame(frame)).toBeNull(); }); it('rejects malformed control json', () => { const frame = Buffer.concat([ - Buffer.from([TERMINAL_INPUT_WS_CONTROL_TAG_JSON]), + Buffer.from([TERMINAL_WS_CONTROL_TAG_JSON]), Buffer.from('{not json', 'utf8'), ]); - expect(readTerminalInputWsControlFrame(frame)).toBeNull(); + expect(readTerminalWsControlFrame(frame)).toBeNull(); }); it('rejects empty control payloads', () => { - expect(readTerminalInputWsControlFrame(null)).toBeNull(); - expect(readTerminalInputWsControlFrame(undefined)).toBeNull(); - expect(readTerminalInputWsControlFrame(Buffer.alloc(0))).toBeNull(); + expect(readTerminalWsControlFrame(null)).toBeNull(); + expect(readTerminalWsControlFrame(undefined)).toBeNull(); + expect(readTerminalWsControlFrame(Buffer.alloc(0))).toBeNull(); }); it('rejects control json that is not object', () => { const frame = Buffer.concat([ - Buffer.from([TERMINAL_INPUT_WS_CONTROL_TAG_JSON]), + Buffer.from([TERMINAL_WS_CONTROL_TAG_JSON]), Buffer.from('"str"', 'utf8'), ]); - expect(readTerminalInputWsControlFrame(frame)).toBeNull(); + expect(readTerminalWsControlFrame(frame)).toBeNull(); }); it('parses control frame from chunk arrays', () => { - const frame = createTerminalInputWsControlFrame({ t: 'bok', v: 1 }); + const frame = createTerminalWsControlFrame({ t: 'bok', v: 1 }); const chunks = [frame.subarray(0, 2), frame.subarray(2)]; - expect(readTerminalInputWsControlFrame(chunks)).toEqual({ t: 'bok', v: 1 }); + expect(readTerminalWsControlFrame(chunks)).toEqual({ t: 'bok', v: 1 }); }); it('normalizes buffer passthrough', () => { const raw = Buffer.from('abc', 'utf8'); - const normalized = normalizeTerminalInputWsMessageToBuffer(raw); + const normalized = normalizeTerminalWsMessageToBuffer(raw); expect(normalized).toBe(raw); expect(normalized.toString('utf8')).toBe('abc'); }); it('normalizes uint8 arrays', () => { - const normalized = normalizeTerminalInputWsMessageToBuffer(new Uint8Array([97, 98, 99])); + const normalized = normalizeTerminalWsMessageToBuffer(new Uint8Array([97, 98, 99])); expect(normalized.toString('utf8')).toBe('abc'); }); it('normalizes array buffer payloads', () => { const source = new Uint8Array([97, 98, 99]).buffer; - const normalized = normalizeTerminalInputWsMessageToBuffer(source); + const normalized = normalizeTerminalWsMessageToBuffer(source); expect(normalized.toString('utf8')).toBe('abc'); }); it('normalizes chunk array payloads', () => { - const normalized = normalizeTerminalInputWsMessageToBuffer([ + const normalized = normalizeTerminalWsMessageToBuffer([ Buffer.from('ab', 'utf8'), Buffer.from('c', 'utf8'), ]); @@ -88,19 +95,19 @@ describe('terminal input websocket protocol', () => { }); it('normalizes text payload from string', () => { - expect(normalizeTerminalInputWsMessageToText('\u001b[A')).toBe('\u001b[A'); + expect(normalizeTerminalWsMessageToText('\u001b[A')).toBe('\u001b[A'); }); it('normalizes text payload from binary data', () => { - expect(normalizeTerminalInputWsMessageToText(Buffer.from('\r', 'utf8'))).toBe('\r'); + expect(normalizeTerminalWsMessageToText(Buffer.from('\r', 'utf8'))).toBe('\r'); }); it('parses relative request pathname', () => { - expect(parseRequestPathname('/api/terminal/input-ws?x=1')).toBe('/api/terminal/input-ws'); + expect(parseRequestPathname('/api/terminal/ws?x=1')).toBe('/api/terminal/ws'); }); it('parses absolute request pathname', () => { - expect(parseRequestPathname('http://localhost:3000/api/terminal/input-ws')).toBe('/api/terminal/input-ws'); + expect(parseRequestPathname('http://localhost:3000/api/terminal/ws')).toBe('/api/terminal/ws'); }); it('returns empty pathname for non-string request url', () => { diff --git a/packages/web/vite.config.ts b/packages/web/vite.config.ts index 7fec9b8e..6e1babf0 100644 --- a/packages/web/vite.config.ts +++ b/packages/web/vite.config.ts @@ -92,6 +92,7 @@ export default defineConfig({ '/api': { target: `http://127.0.0.1:${process.env.OPENCHAMBER_PORT || 3001}`, changeOrigin: true, + ws: true, }, }, },