From eaca3437352b3dd0fdd4e4b101255ae1578dfbdb Mon Sep 17 00:00:00 2001 From: Jinwoo-H Date: Mon, 11 May 2026 16:36:20 -0400 Subject: [PATCH] Improve mobile terminal streaming performance Co-authored-by: Orca --- .../app/h/[hostId]/session/[worktreeId].tsx | 84 +++--- mobile/src/terminal/TerminalWebView.tsx | 38 ++- mobile/src/transport/e2ee.ts | 26 +- mobile/src/transport/protocol-version.ts | 4 +- mobile/src/transport/rpc-client.ts | 164 +++++++++- .../src/transport/terminal-stream-protocol.ts | 79 +++++ src/main/runtime/rpc/core.ts | 4 + src/main/runtime/rpc/dispatcher.ts | 11 +- src/main/runtime/rpc/e2ee-channel.ts | 48 ++- src/main/runtime/rpc/e2ee-crypto.test.ts | 22 +- src/main/runtime/rpc/e2ee-crypto.ts | 28 +- src/main/runtime/rpc/methods/terminal.ts | 283 +++++++++++++++--- src/main/runtime/rpc/ws-transport.ts | 25 +- src/main/runtime/runtime-rpc.ts | 13 +- src/shared/protocol-compat.test.ts | 17 ++ src/shared/protocol-version.ts | 4 +- src/shared/terminal-stream-protocol.test.ts | 63 ++++ src/shared/terminal-stream-protocol.ts | 87 ++++++ 18 files changed, 869 insertions(+), 131 deletions(-) create mode 100644 mobile/src/transport/terminal-stream-protocol.ts create mode 100644 src/shared/terminal-stream-protocol.test.ts create mode 100644 src/shared/terminal-stream-protocol.ts diff --git a/mobile/app/h/[hostId]/session/[worktreeId].tsx b/mobile/app/h/[hostId]/session/[worktreeId].tsx index 3ddf7c523c9..a22e9a0da77 100644 --- a/mobile/app/h/[hostId]/session/[worktreeId].tsx +++ b/mobile/app/h/[hostId]/session/[worktreeId].tsx @@ -97,6 +97,7 @@ const STATUS_LABELS: Record = { function TerminalPaneView({ handle, active, + keyboardLift, onRef, onWebReady, onSelectionMode, @@ -107,6 +108,7 @@ function TerminalPaneView({ }: { handle: string active: boolean + keyboardLift: number onRef: (handle: string, ref: TerminalWebViewHandle | null) => void onWebReady: (handle: string) => void onSelectionMode: (handle: string, active: boolean) => void @@ -125,7 +127,11 @@ function TerminalPaneView({ return ( 0 && { transform: [{ translateY: -keyboardLift }] }, + !active && styles.terminalPaneHidden + ]} > (null) // Why: in Expo SDK 55 edge-to-edge mode the OS does NOT resize the window when // the IME opens — the keyboard draws on top of the app. We track the keyboard - // height ourselves and apply it as paddingBottom on the input/accessory area - // so the input lifts above the IME and the terminal flex container shrinks. + // height ourselves and translate the input/accessory area above the IME without + // changing the terminal frame height, so keyboard open/close does not resize + // the desktop PTY. const [keyboardHeight, setKeyboardHeight] = useState(0) // Why: server-authoritative display mode per terminal. The runtime is the // single source of truth — this state is populated from subscribe responses. @@ -286,7 +293,8 @@ export default function SessionScreen() { { terminal: handle, client: { id: deviceTokenRef.current!, type: 'mobile' as const }, - viewport: viewportRef.current ?? undefined + viewport: viewportRef.current ?? undefined, + capabilities: { terminalBinaryStream: 1 } }, (result) => { if (subscribeSeqRef.current.get(handle) !== seq) return @@ -319,6 +327,9 @@ export default function SessionScreen() { } else if (eventSeq != null && data.type === 'scrollback') { layoutSeqRef.current.set(handle, eventSeq) } + if (data.type === 'subscribed') { + return + } if (data.type === 'scrollback') { if (initializedHandlesRef.current.has(handle)) { return @@ -439,11 +450,12 @@ export default function SessionScreen() { // dims with fresh scrollback. No resubscribe needed. const cols = (data.cols as number) || 80 const rows = (data.rows as number) || 24 - const serialized = - typeof data.serialized === 'string' && data.serialized.length > 0 - ? data.serialized - : '' - getTerminalRef(handle)?.init(cols, rows, serialized) + const serialized = typeof data.serialized === 'string' ? data.serialized : null + if (serialized != null) { + getTerminalRef(handle)?.init(cols, rows, serialized) + } else { + getTerminalRef(handle)?.resize(cols, rows) + } if (data.displayMode) { setTerminalModes((prev) => new Map(prev).set(handle, data.displayMode as MobileDisplayMode) @@ -612,11 +624,7 @@ export default function SessionScreen() { void loadCustomKeys().then(setCustomKeys) }, []) - // Why: drive the bottom padding from the keyboard height (edge-to-edge mode - // doesn't resize the window) and refit xterm once the layout settles so the - // terminal grid matches the new visible area. iOS exposes 'will' events that - // animate in sync with the IME; Android only fires 'did' events reliably. - // Also drives re-measurement when other layout-affecting state changes + // Why: re-measure when non-keyboard layout-affecting state changes // (e.g. tab strip toggling visibility when the terminal count crosses // 0↔1 — without this, a freshly-created 2nd tab subscribes with a // stale viewport that doesn't account for the now-visible tab strip, @@ -665,11 +673,9 @@ export default function SessionScreen() { useEffect(() => { const onShow = (e: KeyboardEvent) => { setKeyboardHeight(e.endCoordinates?.height ?? 0) - scheduleViewportRefit() } const onHide = () => { setKeyboardHeight(0) - scheduleViewportRefit() } const showEvent = Platform.OS === 'ios' ? 'keyboardWillShow' : 'keyboardDidShow' const hideEvent = Platform.OS === 'ios' ? 'keyboardWillHide' : 'keyboardDidHide' @@ -680,7 +686,7 @@ export default function SessionScreen() { showSub.remove() hideSub.remove() } - }, [scheduleViewportRefit]) + }, []) // Why: the tab strip is hidden when only one terminal exists and shown // once a second is created. Crossing the 1↔2 boundary changes the @@ -726,13 +732,12 @@ export default function SessionScreen() { useEffect(() => { if (connState !== 'connected') return - // Why: on reconnect the RPC client auto-resends terminal.subscribe, - // creating new server-side handlers. Clear local subscription state - // so subscribeToTerminal's guards don't block fresh subscriptions, - // and clear xterm buffers so the new scrollback snapshot replaces - // stale content (including data that arrived while disconnected). - clearTerminalCache() - setTerminalsLoaded(false) + // Why: the RPC client auto-resends terminal.subscribe on reconnect. + // Keep the current xterm visible while the binary snapshot hydrates, + // instead of clearing to a blank "Loading terminals" surface. + if (initializedHandlesRef.current.size === 0) { + setTerminalsLoaded(false) + } let disposed = false const timers: ReturnType[] = [] function addTimer(fn: () => void, ms: number) { @@ -1200,17 +1205,15 @@ export default function SessionScreen() { : `${terminals.length} terminals` : STATUS_LABELS[connState] - // Why: on Android (Samsung 3-button) the keyboard event reports only the IME - // height; the system nav bar sits below the keyboard and adds its own height - // on top, so we must add insets.bottom too or the input stays clipped behind - // the nav bar. On iOS the keyboard coordinates already include the home - // indicator region, so adding insets.bottom would double-count. - const bottomPadding = + // Why: keep safe-area padding in layout at all times, then visually translate + // the controls over the terminal when the keyboard appears. iOS keyboard + // height includes the home-indicator inset; Android IME height does not. + const keyboardLift = keyboardHeight > 0 ? Platform.OS === 'ios' - ? keyboardHeight - : keyboardHeight + insets.bottom - : insets.bottom + ? Math.max(0, keyboardHeight - insets.bottom) + : keyboardHeight + : 0 return ( @@ -1313,6 +1316,7 @@ export default function SessionScreen() { key={terminal.handle} handle={terminal.handle} active={terminal.handle === activeHandle} + keyboardLift={keyboardLift} onRef={setTerminalWebViewRef} onWebReady={handleTerminalWebReady} onSelectionMode={handleSelectionMode} @@ -1333,9 +1337,14 @@ export default function SessionScreen() { )} - {/* Why: bottomPadding lifts the entire input region above the keyboard - (when shown) or above the system nav bar / home indicator (when hidden). */} - + {/* Why: translate instead of resizing so keyboard open/close does not + trigger a server-side PTY viewport change. */} + {/* Accessory keys */} void init: (cols: number, rows: number, initialData?: string) => void + resize: (cols: number, rows: number) => void clear: () => void measureFitDimensions: (containerHeight?: number) => Promise<{ cols: number; rows: number } | null> resetZoom: () => void @@ -40,6 +41,7 @@ type Props = { type TerminalMessage = | { type: 'write'; id?: number; data: string } | { type: 'init'; id?: number; cols: number; rows: number; initialData?: string } + | { type: 'resize'; id?: number; cols: number; rows: number } | { type: 'clear'; id?: number } | { type: 'measure'; id?: number; containerHeight?: number } | { type: 'reset-zoom'; id?: number } @@ -432,7 +434,20 @@ const XTERM_HTML = ` firstDataPending = true; var replayData = normalizeInitialData(initialData); activeAltScreenSnapshot = isAltScreenActive(replayData); - if (term) term.dispose(); + var oldTerm = term; + var oldSurface = surface; + var nextSurface = null; + if (oldTerm) { + nextSurface = document.createElement('div'); + nextSurface.id = 'terminal-surface'; + nextSurface.style.visibility = 'hidden'; + nextSurface.style.position = 'absolute'; + nextSurface.style.left = '0'; + nextSurface.style.top = '0'; + document.getElementById('terminal-container').appendChild(nextSurface); + surface = nextSurface; + oldSurface.removeAttribute('id'); + } term = new Terminal({ cols: cols || 80, @@ -485,6 +500,14 @@ const XTERM_HTML = ` ready = true; afterWritesDrained(function() { if (gen !== terminalGeneration) return; + if (nextSurface && oldSurface) { + nextSurface.style.visibility = 'visible'; + nextSurface.style.position = ''; + nextSurface.style.left = ''; + nextSurface.style.top = ''; + oldSurface.remove(); + if (oldTerm) oldTerm.dispose(); + } applyFitScale('init-replay'); notify({ type: 'ready', cols: cols, rows: rows }); }); @@ -508,6 +531,14 @@ const XTERM_HTML = ` } } + function resize(cols, rows) { + if (!term) return; + initRows = rows || initRows; + term.resize(cols || term.cols, rows || term.rows); + applyFitScale('resize-msg'); + notify({ type: 'ready', cols: cols, rows: rows }); + } + function notify(msg) { if (window.ReactNativeWebView) { window.ReactNativeWebView.postMessage(JSON.stringify(msg)); @@ -576,6 +607,8 @@ const XTERM_HTML = ` } if (msg.type === 'init') { init(msg.cols, msg.rows, msg.initialData); + } else if (msg.type === 'resize') { + resize(msg.cols, msg.rows); } else if (msg.type === 'write') { write(msg.data); } else if (msg.type === 'clear') { @@ -1393,6 +1426,9 @@ export const TerminalWebView = forwardRef(function }) postMessage({ type: 'init', cols, rows, initialData }) }, + resize(cols: number, rows: number) { + postMessage({ type: 'resize', cols, rows }) + }, clear() { postMessage({ type: 'clear' }) }, diff --git a/mobile/src/transport/e2ee.ts b/mobile/src/transport/e2ee.ts index 87d599ec077..a6dbf6f346f 100644 --- a/mobile/src/transport/e2ee.ts +++ b/mobile/src/transport/e2ee.ts @@ -1,6 +1,7 @@ // Why: E2EE primitives for the mobile side. Uses tweetnacl for Curve25519 ECDH -// key exchange and XSalsa20-Poly1305 authenticated encryption. Wire format is -// base64([24-byte nonce][ciphertext]) over WebSocket text frames. +// key exchange and XSalsa20-Poly1305 authenticated encryption. JSON RPC uses +// base64([24-byte nonce][ciphertext]) over WebSocket text frames; terminal +// stream frames use the raw byte bundle. import nacl from 'tweetnacl' import * as ExpoCrypto from 'expo-crypto' @@ -62,19 +63,28 @@ export function publicKeyToBase64(key: Uint8Array): string { } export function encrypt(plaintext: string, sharedKey: Uint8Array): string { - const nonce = u8(nacl.randomBytes(nacl.box.nonceLength)) const messageBytes = u8(new TextEncoder().encode(plaintext)) - const ciphertext = nacl.box.after(messageBytes, nonce, u8(sharedKey)) + return uint8ToBase64(encryptBytes(messageBytes, sharedKey)) +} + +export function decrypt(encrypted: string, sharedKey: Uint8Array): string | null { + const bundle = base64ToUint8(encrypted) + const plaintext = decryptBytes(bundle, sharedKey) + return plaintext ? new TextDecoder().decode(plaintext) : null +} + +export function encryptBytes(plaintext: Uint8Array, sharedKey: Uint8Array): Uint8Array { + const nonce = u8(nacl.randomBytes(nacl.box.nonceLength)) + const ciphertext = nacl.box.after(u8(plaintext), nonce, u8(sharedKey)) const bundle = new Uint8Array(nonce.length + ciphertext.length) bundle.set(nonce) bundle.set(ciphertext, nonce.length) - return uint8ToBase64(bundle) + return bundle } -export function decrypt(encrypted: string, sharedKey: Uint8Array): string | null { - const bundle = base64ToUint8(encrypted) +export function decryptBytes(bundle: Uint8Array, sharedKey: Uint8Array): Uint8Array | null { if (bundle.length < nacl.box.nonceLength + nacl.box.overheadLength) { return null } @@ -87,5 +97,5 @@ export function decrypt(encrypted: string, sharedKey: Uint8Array): string | null return null } - return new TextDecoder().decode(plaintext) + return u8(plaintext) } diff --git a/mobile/src/transport/protocol-version.ts b/mobile/src/transport/protocol-version.ts index 9445c1f2881..3ba4ede3878 100644 --- a/mobile/src/transport/protocol-version.ts +++ b/mobile/src/transport/protocol-version.ts @@ -16,5 +16,5 @@ // desktop feature added at a specific desktop protocol version. This // triggers a hard-block screen for users paired to older desktops. -export const MOBILE_PROTOCOL_VERSION = 1 -export const MIN_COMPATIBLE_DESKTOP_VERSION = 0 +export const MOBILE_PROTOCOL_VERSION = 2 +export const MIN_COMPATIBLE_DESKTOP_VERSION = 2 diff --git a/mobile/src/transport/rpc-client.ts b/mobile/src/transport/rpc-client.ts index d3e080441c3..790723239a7 100644 --- a/mobile/src/transport/rpc-client.ts +++ b/mobile/src/transport/rpc-client.ts @@ -11,8 +11,15 @@ import { publicKeyFromBase64, publicKeyToBase64, encrypt, - decrypt + decrypt, + decryptBytes } from './e2ee' +import { + TerminalStreamOpcode, + decodeTerminalStreamFrame, + decodeTerminalStreamJson, + decodeTerminalStreamText +} from './terminal-stream-protocol' type PendingRequest = { resolve: (response: RpcResponse) => void @@ -27,6 +34,12 @@ type StreamRequest = { listener: StreamingListener } +type TerminalSnapshotState = { + streamId: number + meta: Record + chunks: string[] +} + export type RpcClient = { sendRequest: (method: string, params?: unknown) => Promise subscribe: (method: string, params: unknown, onData: StreamingListener) => () => void @@ -144,6 +157,9 @@ export function connect( const pending = new Map() const streamListeners = new Map() + const terminalStreamListeners = new Map() + const terminalStreamIdsByRequest = new Map>() + const terminalSnapshots = new Map() const stateListeners = new Set<(state: ConnectionState) => void>() const connectWaiters: Array<{ resolve: () => void; reject: (e: Error) => void }> = [] @@ -293,14 +309,21 @@ export function connect( } ws.onmessage = (event) => { + void handleSocketMessage(event.data) + } + + async function handleSocketMessage(rawData: unknown) { // Why: track last-inbound for the openConnection diagnostic. Server // pongs and stream events both bump this — anything from the wire. lastInboundAt = Date.now() - const raw = typeof event.data === 'string' ? event.data : String(event.data) + const raw = typeof rawData === 'string' ? rawData : null // Why: during handshaking, e2ee_ready is plaintext because it precedes // encrypted auth; e2ee_authenticated/e2ee_error are encrypted. if (state === 'handshaking') { + if (raw === null) { + return + } try { const msg = JSON.parse(raw) if (msg.type === 'e2ee_ready') { @@ -362,6 +385,19 @@ export function connect( return } + if (raw === null) { + const bytes = await websocketPayloadToUint8(rawData) + if (!bytes) { + return + } + const plaintextBytes = decryptBytes(bytes, sharedKey) + if (!plaintextBytes) { + return + } + handleTerminalBinaryFrame(plaintextBytes) + return + } + const plaintext = decrypt(raw, sharedKey) if (plaintext === null) { return @@ -390,7 +426,17 @@ export function connect( if (isStreaming) { const stream = streamListeners.get(response.id) if (stream && response.ok) { - stream.listener((response as RpcSuccess).result) + const result = (response as RpcSuccess).result + if (isTerminalSubscribedResult(result)) { + let ids = terminalStreamIdsByRequest.get(response.id) + if (!ids) { + ids = new Set() + terminalStreamIdsByRequest.set(response.id, ids) + } + ids.add(result.streamId) + terminalStreamListeners.set(result.streamId, stream.listener) + } + stream.listener(result) } return } @@ -654,6 +700,75 @@ export function connect( } } + function handleTerminalBinaryFrame(bytes: Uint8Array) { + const frame = decodeTerminalStreamFrame(bytes) + if (!frame) { + return + } + const listener = terminalStreamListeners.get(frame.streamId) + if (!listener) { + return + } + if (frame.opcode === TerminalStreamOpcode.Output) { + listener({ + type: 'data', + streamId: frame.streamId, + chunk: decodeTerminalStreamText(frame.payload) + }) + return + } + if (frame.opcode === TerminalStreamOpcode.SnapshotStart) { + const meta = decodeTerminalStreamJson>(frame.payload) + if (!meta) { + return + } + terminalSnapshots.set(frame.streamId, { streamId: frame.streamId, meta, chunks: [] }) + return + } + if (frame.opcode === TerminalStreamOpcode.SnapshotChunk) { + const snapshot = terminalSnapshots.get(frame.streamId) + if (!snapshot) { + return + } + snapshot.chunks.push(decodeTerminalStreamText(frame.payload)) + return + } + if (frame.opcode === TerminalStreamOpcode.SnapshotEnd) { + const snapshot = terminalSnapshots.get(frame.streamId) + if (!snapshot) { + return + } + terminalSnapshots.delete(frame.streamId) + const kind = snapshot.meta.kind === 'resized' ? 'resized' : 'scrollback' + listener({ + ...snapshot.meta, + type: kind, + streamId: frame.streamId, + serialized: snapshot.chunks.join('') + }) + return + } + if (frame.opcode === TerminalStreamOpcode.Resized) { + const meta = decodeTerminalStreamJson>(frame.payload) + if (!meta) { + return + } + listener({ + ...meta, + type: 'resized', + streamId: frame.streamId + }) + return + } + if (frame.opcode === TerminalStreamOpcode.Error) { + listener({ + type: 'error', + streamId: frame.streamId, + message: decodeTerminalStreamText(frame.payload) + }) + } + } + function sendEncrypted(request: unknown): boolean { if (ws && ws.readyState === WebSocket.OPEN && sharedKey) { ws.send(encrypt(JSON.stringify(request), sharedKey)) @@ -740,6 +855,14 @@ export function connect( return () => { const stream = streamListeners.get(id) streamListeners.delete(id) + const terminalStreamIds = terminalStreamIdsByRequest.get(id) + if (terminalStreamIds) { + for (const streamId of terminalStreamIds) { + terminalStreamListeners.delete(streamId) + terminalSnapshots.delete(streamId) + } + terminalStreamIdsByRequest.delete(id) + } if ( stream?.method === 'terminal.subscribe' && stream.params && @@ -813,3 +936,38 @@ export function connect( } } } + +function isTerminalSubscribedResult( + value: unknown +): value is { type: 'subscribed'; streamId: number } { + return ( + !!value && + typeof value === 'object' && + (value as { type?: unknown }).type === 'subscribed' && + typeof (value as { streamId?: unknown }).streamId === 'number' + ) +} + +async function websocketPayloadToUint8(value: unknown): Promise { + if (value instanceof Uint8Array) { + return value + } + if (value instanceof ArrayBuffer) { + return new Uint8Array(value) + } + if (value && typeof value === 'object' && 'arrayBuffer' in value) { + const blob = value as { arrayBuffer: () => Promise } + return new Uint8Array(await blob.arrayBuffer()) + } + if (typeof FileReader !== 'undefined' && value instanceof Blob) { + return new Promise((resolve) => { + const reader = new FileReader() + reader.onload = () => { + resolve(reader.result instanceof ArrayBuffer ? new Uint8Array(reader.result) : null) + } + reader.onerror = () => resolve(null) + reader.readAsArrayBuffer(value) + }) + } + return null +} diff --git a/mobile/src/transport/terminal-stream-protocol.ts b/mobile/src/transport/terminal-stream-protocol.ts new file mode 100644 index 00000000000..0d5a04340c5 --- /dev/null +++ b/mobile/src/transport/terminal-stream-protocol.ts @@ -0,0 +1,79 @@ +const TERMINAL_STREAM_KIND = 0x74 +const TERMINAL_STREAM_VERSION = 1 +const HEADER_BYTES = 16 + +export enum TerminalStreamOpcode { + Output = 1, + SnapshotStart = 2, + SnapshotChunk = 3, + SnapshotEnd = 4, + Resized = 5, + Error = 6 +} + +export type TerminalStreamFrame = { + opcode: TerminalStreamOpcode + streamId: number + seq: number + payload: Uint8Array +} + +export function encodeTerminalStreamFrame(frame: TerminalStreamFrame): Uint8Array { + const out = new Uint8Array(HEADER_BYTES + frame.payload.length) + const view = new DataView(out.buffer, out.byteOffset, out.byteLength) + view.setUint8(0, TERMINAL_STREAM_KIND) + view.setUint8(1, TERMINAL_STREAM_VERSION) + view.setUint8(2, frame.opcode) + view.setUint8(3, 0) + view.setUint32(4, frame.streamId, true) + const seq = Math.max(0, Math.floor(frame.seq)) + view.setUint32(8, Math.floor(seq / 0x100000000), true) + view.setUint32(12, seq >>> 0, true) + out.set(frame.payload, HEADER_BYTES) + return out +} + +export function decodeTerminalStreamFrame(bytes: Uint8Array): TerminalStreamFrame | null { + if (bytes.length < HEADER_BYTES) { + return null + } + const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength) + if (view.getUint8(0) !== TERMINAL_STREAM_KIND || view.getUint8(1) !== TERMINAL_STREAM_VERSION) { + return null + } + const opcode = view.getUint8(2) + if (!isTerminalStreamOpcode(opcode)) { + return null + } + const high = view.getUint32(8, true) + const low = view.getUint32(12, true) + return { + opcode, + streamId: view.getUint32(4, true), + seq: high * 0x100000000 + low, + payload: bytes.slice(HEADER_BYTES) + } +} + +export function decodeTerminalStreamJson(payload: Uint8Array): T | null { + try { + return JSON.parse(new TextDecoder().decode(payload)) as T + } catch { + return null + } +} + +export function decodeTerminalStreamText(payload: Uint8Array): string { + return new TextDecoder().decode(payload) +} + +function isTerminalStreamOpcode(value: number): value is TerminalStreamOpcode { + return ( + value === TerminalStreamOpcode.Output || + value === TerminalStreamOpcode.SnapshotStart || + value === TerminalStreamOpcode.SnapshotChunk || + value === TerminalStreamOpcode.SnapshotEnd || + value === TerminalStreamOpcode.Resized || + value === TerminalStreamOpcode.Error + ) +} diff --git a/src/main/runtime/rpc/core.ts b/src/main/runtime/rpc/core.ts index 0340561938e..eceb98bf405 100644 --- a/src/main/runtime/rpc/core.ts +++ b/src/main/runtime/rpc/core.ts @@ -52,6 +52,10 @@ export type RpcContext = { // server reap all subscriptions for a closing socket, even when other // sockets for the same deviceToken stay alive (multi-screen mobile). connectionId?: string + // Why: mobile terminal traffic is byte-oriented and bypasses JSON streaming + // responses after the binary terminal cutover. Undefined on Unix/socket + // transports and non-E2EE WebSocket paths. + sendBinary?: (bytes: Uint8Array) => void } export type RpcHandler = (params: TParams, ctx: RpcContext) => Promise | unknown diff --git a/src/main/runtime/rpc/dispatcher.ts b/src/main/runtime/rpc/dispatcher.ts index bdd2f181b30..ba346a51b45 100644 --- a/src/main/runtime/rpc/dispatcher.ts +++ b/src/main/runtime/rpc/dispatcher.ts @@ -77,7 +77,7 @@ export class RpcDispatcher { async dispatchStreaming( request: RpcRequest, reply: (response: string) => void, - options?: { connectionId?: string } + options?: { connectionId?: string; sendBinary?: (bytes: Uint8Array) => void } ): Promise { const meta = this.meta() const method = this.registry.get(request.method) @@ -100,7 +100,8 @@ export class RpcDispatcher { try { const result = await method.handler(parsedParams.value, { runtime: this.runtime, - connectionId: options?.connectionId + connectionId: options?.connectionId, + sendBinary: options?.sendBinary }) reply(JSON.stringify(successResponse(request.id, meta, result))) } catch (error) { @@ -118,7 +119,11 @@ export class RpcDispatcher { try { await method.handler( parsedParams.value, - { runtime: this.runtime, connectionId: options?.connectionId }, + { + runtime: this.runtime, + connectionId: options?.connectionId, + sendBinary: options?.sendBinary + }, emit ) } catch (error) { diff --git a/src/main/runtime/rpc/e2ee-channel.ts b/src/main/runtime/rpc/e2ee-channel.ts index f3100584d56..a105b882a0a 100644 --- a/src/main/runtime/rpc/e2ee-channel.ts +++ b/src/main/runtime/rpc/e2ee-channel.ts @@ -2,7 +2,7 @@ // It owns the handshake state machine and transparent encrypt/decrypt so the RPC // handler only sees plaintext JSON, identical to the Unix socket path. import type { WebSocket } from 'ws' -import { deriveSharedKey, encrypt, decrypt } from './e2ee-crypto' +import { deriveSharedKey, encrypt, decrypt, encryptBytes, decryptBytes } from './e2ee-crypto' type ChannelState = 'awaiting_hello' | 'awaiting_auth' | 'ready' @@ -40,7 +40,11 @@ export class E2EEChannel { // can forward decrypted messages. Kept as a callback rather than constructor // param because the handler needs the encrypt function for replies. private messageHandler: - | ((plaintext: string, encryptedReply: (response: string) => void) => void) + | (( + plaintext: string, + encryptedReply: (response: string) => void, + encryptedBinaryReply: (response: Uint8Array) => void + ) => void) | null = null deviceToken: string | null = null @@ -58,13 +62,21 @@ export class E2EEChannel { } onMessage( - handler: (plaintext: string, encryptedReply: (response: string) => void) => void + handler: ( + plaintext: string, + encryptedReply: (response: string) => void, + encryptedBinaryReply: (response: Uint8Array) => void + ) => void ): void { this.messageHandler = handler } - handleRawMessage(raw: string): void { + handleRawMessage(raw: string | Uint8Array): void { if (this.state === 'awaiting_hello') { + if (typeof raw !== 'string') { + this.onError(4001, 'Invalid handshake message') + return + } this.handleHello(raw) return } @@ -73,12 +85,17 @@ export class E2EEChannel { return } + if (typeof raw !== 'string') { + const plaintextBytes = decryptBytes(raw, this.sharedKey) + if (plaintextBytes === null) { + this.trackDecryptFailure() + } + return + } + const plaintext = decrypt(raw, this.sharedKey) if (plaintext === null) { - this.consecutiveFailures++ - if (this.consecutiveFailures >= MAX_CONSECUTIVE_DECRYPT_FAILURES) { - this.onError(4003, 'Too many decryption failures') - } + this.trackDecryptFailure() return } @@ -100,7 +117,20 @@ export class E2EEChannel { } this.ws.send(encrypt(response, this.sharedKey)) } - this.messageHandler?.(plaintext, encryptedReply) + const encryptedBinaryReply = (response: Uint8Array) => { + if (!this.sharedKey || this.ws.readyState !== this.ws.OPEN) { + return + } + this.ws.send(Buffer.from(encryptBytes(response, this.sharedKey)), { binary: true }) + } + this.messageHandler?.(plaintext, encryptedReply, encryptedBinaryReply) + } + + private trackDecryptFailure(): void { + this.consecutiveFailures++ + if (this.consecutiveFailures >= MAX_CONSECUTIVE_DECRYPT_FAILURES) { + this.onError(4003, 'Too many decryption failures') + } } private handleHello(raw: string): void { diff --git a/src/main/runtime/rpc/e2ee-crypto.test.ts b/src/main/runtime/rpc/e2ee-crypto.test.ts index 09e546db63c..aaa5db016f4 100644 --- a/src/main/runtime/rpc/e2ee-crypto.test.ts +++ b/src/main/runtime/rpc/e2ee-crypto.test.ts @@ -1,5 +1,12 @@ import { describe, expect, it } from 'vitest' -import { generateKeyPair, deriveSharedKey, encrypt, decrypt } from './e2ee-crypto' +import { + generateKeyPair, + deriveSharedKey, + encrypt, + decrypt, + encryptBytes, + decryptBytes +} from './e2ee-crypto' describe('e2ee-crypto', () => { it('encrypt/decrypt round-trips with shared key', () => { @@ -76,4 +83,17 @@ describe('e2ee-crypto', () => { expect(decrypt(tampered, shared)).toBeNull() }) + + it('encrypt/decrypt round-trips raw bytes for binary terminal frames', () => { + const server = generateKeyPair() + const client = generateKeyPair() + const serverShared = deriveSharedKey(server.secretKey, client.publicKey) + const clientShared = deriveSharedKey(client.secretKey, server.publicKey) + const payload = new Uint8Array([0, 1, 2, 127, 128, 255]) + + const encrypted = encryptBytes(payload, clientShared) + const decrypted = decryptBytes(encrypted, serverShared) + + expect(decrypted).toEqual(payload) + }) }) diff --git a/src/main/runtime/rpc/e2ee-crypto.ts b/src/main/runtime/rpc/e2ee-crypto.ts index 39adec27160..40818c424b5 100644 --- a/src/main/runtime/rpc/e2ee-crypto.ts +++ b/src/main/runtime/rpc/e2ee-crypto.ts @@ -1,6 +1,6 @@ // Why: shared E2EE primitives for the desktop side. Wraps tweetnacl to provide -// encrypt/decrypt with the NaCl box format: [24-byte nonce][ciphertext], encoded -// as base64 for transmission over WebSocket text frames. +// encrypt/decrypt with the NaCl box format: [24-byte nonce][ciphertext]. JSON +// RPC uses base64 text frames; terminal streams use the raw byte bundle. import nacl from 'tweetnacl' export function generateKeyPair(): nacl.BoxKeyPair { @@ -12,19 +12,31 @@ export function deriveSharedKey(ourSecretKey: Uint8Array, peerPublicKey: Uint8Ar } export function encrypt(plaintext: string, sharedKey: Uint8Array): string { - const nonce = nacl.randomBytes(nacl.box.nonceLength) const messageBytes = new TextEncoder().encode(plaintext) - const ciphertext = nacl.box.after(messageBytes, nonce, sharedKey) + return Buffer.from(encryptBytes(messageBytes, sharedKey)).toString('base64') +} + +export function decrypt(encrypted: string, sharedKey: Uint8Array): string | null { + const bundle = Uint8Array.from(Buffer.from(encrypted, 'base64')) + const plaintext = decryptBytes(bundle, sharedKey) + return plaintext ? new TextDecoder().decode(plaintext) : null +} + +export function encryptBytes( + plaintext: Uint8Array, + sharedKey: Uint8Array +): Uint8Array { + const nonce = nacl.randomBytes(nacl.box.nonceLength) + const ciphertext = nacl.box.after(plaintext, nonce, sharedKey) const bundle = new Uint8Array(nonce.length + ciphertext.length) bundle.set(nonce) bundle.set(ciphertext, nonce.length) - return Buffer.from(bundle).toString('base64') + return bundle } -export function decrypt(encrypted: string, sharedKey: Uint8Array): string | null { - const bundle = Uint8Array.from(Buffer.from(encrypted, 'base64')) +export function decryptBytes(bundle: Uint8Array, sharedKey: Uint8Array): Uint8Array | null { if (bundle.length < nacl.box.nonceLength + nacl.box.overheadLength) { return null } @@ -37,5 +49,5 @@ export function decrypt(encrypted: string, sharedKey: Uint8Array): string | null return null } - return new TextDecoder().decode(plaintext) + return plaintext } diff --git a/src/main/runtime/rpc/methods/terminal.ts b/src/main/runtime/rpc/methods/terminal.ts index fbf4136aa7a..71cf2f7d90d 100644 --- a/src/main/runtime/rpc/methods/terminal.ts +++ b/src/main/runtime/rpc/methods/terminal.ts @@ -2,6 +2,13 @@ import { z } from 'zod' import { defineMethod, defineStreamingMethod, type RpcAnyMethod } from '../core' import { OptionalFiniteNumber, OptionalString, requiredString } from '../schemas' +import type { OrcaRuntimeService } from '../../orca-runtime' +import { + TerminalStreamOpcode, + encodeTerminalStreamFrame, + encodeTerminalStreamJson, + encodeTerminalStreamText +} from '../../../../shared/terminal-stream-protocol' // Why: when a mobile client subscribes the server resizes the PTY to phone // dims and serializes the buffer. Sending only the visible screen meant @@ -10,6 +17,87 @@ import { OptionalFiniteNumber, OptionalString, requiredString } from '../schemas // agent runs (Claude Code chats, command output) reachable. The mobile // WebView's xterm has a 5000-row buffer so this fits comfortably. const MOBILE_SUBSCRIBE_SCROLLBACK_ROWS = 1000 +const MOBILE_SNAPSHOT_BYTE_BUDGET = 512 * 1024 +const TERMINAL_STREAM_CHUNK_BYTES = 48 * 1024 +let nextTerminalStreamId = 1 + +type SnapshotFrameOptions = { + kind: 'scrollback' | 'resized' + cols: number + rows: number + data: string + displayMode?: string + reason?: string + seq?: number + truncated?: boolean + truncatedByByteBudget?: boolean +} + +type SerializedSnapshot = { + data: string + cols: number + rows: number + scrollbackRows: number + truncatedByByteBudget: boolean +} | null + +function sendSnapshotFrames( + sendFrame: (opcode: TerminalStreamOpcode, payload?: Uint8Array) => void, + options: SnapshotFrameOptions +): { bytes: number; chunks: number } { + sendFrame( + TerminalStreamOpcode.SnapshotStart, + encodeTerminalStreamJson({ + kind: options.kind, + cols: options.cols, + rows: options.rows, + displayMode: options.displayMode, + reason: options.reason, + seq: options.seq, + truncated: options.truncated === true, + truncatedByByteBudget: options.truncatedByByteBudget === true + }) + ) + const bytes = encodeTerminalStreamText(options.data) + let chunks = 0 + for (let offset = 0; offset < bytes.length; offset += TERMINAL_STREAM_CHUNK_BYTES) { + chunks++ + sendFrame( + TerminalStreamOpcode.SnapshotChunk, + bytes.slice(offset, offset + TERMINAL_STREAM_CHUNK_BYTES) + ) + } + sendFrame(TerminalStreamOpcode.SnapshotEnd) + return { bytes: bytes.byteLength, chunks } +} + +async function serializeBudgetedMobileSnapshot( + runtime: OrcaRuntimeService, + ptyId: string, + isMobile: boolean +): Promise { + if (!isMobile) { + const serialized = await runtime.serializeTerminalBuffer(ptyId, { scrollbackRows: 0 }) + return serialized ? { ...serialized, scrollbackRows: 0, truncatedByByteBudget: false } : null + } + const candidates = [MOBILE_SUBSCRIBE_SCROLLBACK_ROWS, 500, 250, 100, 25, 0] + for (const rows of candidates) { + const serialized = await runtime.serializeTerminalBuffer(ptyId, { scrollbackRows: rows }) + if (!serialized) { + return null + } + const bytes = new TextEncoder().encode(serialized.data).byteLength + if (bytes <= MOBILE_SNAPSHOT_BYTE_BUDGET || rows === 0) { + return { + ...serialized, + scrollbackRows: rows, + truncatedByByteBudget: + rows < MOBILE_SUBSCRIBE_SCROLLBACK_ROWS || bytes > MOBILE_SNAPSHOT_BYTE_BUDGET + } + } + } + return null +} const TerminalHandle = z.object({ terminal: requiredString('Missing terminal handle') @@ -127,6 +215,11 @@ const TerminalSubscribe = TerminalHandle.extend({ cols: z.number().int().min(20).max(240), rows: z.number().int().min(8).max(120) }) + .optional(), + capabilities: z + .object({ + terminalBinaryStream: z.literal(1).optional() + }) .optional() }) @@ -384,9 +477,10 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ defineStreamingMethod({ name: 'terminal.subscribe', params: TerminalSubscribe, - handler: async (params, { runtime }, emit) => { + handler: async (params, { runtime, connectionId, sendBinary }, emit) => { let leaf = runtime.resolveLeafForHandle(params.terminal) const isMobile = params.client?.type === 'mobile' + const useBinaryStream = isMobile && params.capabilities?.terminalBinaryStream === 1 // Why: the left pane's PTY spawns asynchronously after the tab is created. // Mobile clients that subscribe before the PTY is ready would get a bare @@ -403,30 +497,99 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ if (!leaf?.ptyId) { const read = await runtime.readTerminal(params.terminal) - emit({ - type: 'scrollback', - lines: read.tail, - truncated: read.truncated, - serialized: undefined, - cols: undefined, - rows: undefined - }) + emit({ type: 'subscribed', streamId: null, lines: read.tail, truncated: read.truncated }) emit({ type: 'end' }) return } + if (isMobile && (!useBinaryStream || !sendBinary)) { + throw new Error('binary_terminal_stream_required') + } + const ptyId = leaf.ptyId const clientId = params.client?.id + if (!isMobile) { + const read = await runtime.readTerminal(params.terminal) + const serialized = await serializeBudgetedMobileSnapshot(runtime, ptyId, false) + const size = runtime.getTerminalSize(ptyId) + const displayMode = runtime.getMobileDisplayMode(ptyId) + const seq = runtime.getLayout(ptyId)?.seq + emit({ + type: 'scrollback', + lines: read.tail, + truncated: read.truncated, + serialized: serialized?.data, + cols: serialized?.cols ?? size?.cols, + rows: serialized?.rows ?? size?.rows, + displayMode, + seq + }) + await new Promise((resolve) => { + const unsubscribeData = runtime.subscribeToTerminalData(ptyId, (data) => { + emit({ type: 'data', chunk: data }) + }) + const unsubscribeFit = runtime.subscribeToFitOverrideChanges(ptyId, (event) => { + emit({ + type: 'fit-override-changed', + mode: event.mode, + cols: event.cols, + rows: event.rows + }) + }) + runtime.registerSubscriptionCleanup( + params.terminal, + () => { + unsubscribeData() + unsubscribeFit() + emit({ type: 'end' }) + resolve() + }, + connectionId + ) + }) + return + } + + const streamId = nextTerminalStreamId++ + let cursor = 0 + let closed = false + let buffering = true + const pendingOutput: string[] = [] + const sendFrame = ( + opcode: TerminalStreamOpcode, + payload: Uint8Array = new Uint8Array() + ): void => { + if (closed || !sendBinary) { + return + } + sendBinary(encodeTerminalStreamFrame({ opcode, streamId, seq: cursor++, payload })) + } // Server-side auto-fit: resize PTY to phone dims before serializing scrollback if (isMobile && clientId) { await runtime.handleMobileSubscribe(ptyId, clientId, params.viewport) } - const read = await runtime.readTerminal(params.terminal) - const serialized = await runtime.serializeTerminalBuffer(ptyId, { - scrollbackRows: isMobile ? MOBILE_SUBSCRIBE_SCROLLBACK_ROWS : 0 + const unsubscribeData = runtime.subscribeToTerminalData(ptyId, (data) => { + if (closed) { + return + } + if (buffering) { + pendingOutput.push(data) + return + } + sendBinary!( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Output, + streamId, + seq: cursor++, + payload: encodeTerminalStreamText(data) + }) + ) }) + + const read = await runtime.readTerminal(params.terminal) + const serialized = await serializeBudgetedMobileSnapshot(runtime, ptyId, true) const size = runtime.getTerminalSize(ptyId) const displayMode = runtime.getMobileDisplayMode(ptyId) // Why: emit the current layout seq with the initial scrollback so @@ -435,40 +598,61 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ // See docs/mobile-terminal-layout-state-machine.md. const seq = runtime.getLayout(ptyId)?.seq emit({ - type: 'scrollback', + type: 'subscribed', + streamId, lines: read.tail, truncated: read.truncated, - serialized: serialized?.data, cols: serialized?.cols ?? size?.cols, rows: serialized?.rows ?? size?.rows, displayMode, seq }) + const snapshotStats = sendSnapshotFrames(sendFrame, { + kind: 'scrollback', + cols: serialized?.cols ?? size?.cols ?? 80, + rows: serialized?.rows ?? size?.rows ?? 24, + displayMode, + seq, + truncated: read.truncated, + truncatedByByteBudget: serialized?.truncatedByByteBudget, + data: serialized?.data ?? '' + }) + console.log('[mobile-terminal-stream] snapshot', { + terminal: params.terminal, + streamId, + kind: 'scrollback', + bytes: snapshotStats.bytes, + chunks: snapshotStats.chunks, + scrollbackRows: serialized?.scrollbackRows, + truncatedByByteBudget: serialized?.truncatedByByteBudget === true + }) + buffering = false + for (const item of pendingOutput.splice(0)) { + sendBinary!( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Output, + streamId, + seq: cursor++, + payload: encodeTerminalStreamText(item) + }) + ) + } await new Promise((resolve) => { - const unsubscribeData = runtime.subscribeToTerminalData(ptyId, (data) => { - emit({ type: 'data', chunk: data }) - }) - - // Inline resize events replace the old fit-override-changed event for - // mobile clients. They include fresh serialized scrollback so the client - // can reinitialize xterm without resubscribing. - const unsubscribeResize = runtime.subscribeToTerminalResize(ptyId, async (event) => { - // Why: mobile subscriptions need the same scrollback on inline - // resize as on initial subscribe — without it, toggling phone/desktop - // mode or the keyboard-driven refit would silently wipe history. - const fresh = await runtime.serializeTerminalBuffer(ptyId, { - scrollbackRows: isMobile ? MOBILE_SUBSCRIBE_SCROLLBACK_ROWS : 0 - }) - emit({ - type: 'resized', - cols: event.cols, - rows: event.rows, - serialized: fresh?.data, - displayMode: event.displayMode, - reason: event.reason, - seq: event.seq - }) + const unsubscribeResize = runtime.subscribeToTerminalResize(ptyId, (event) => { + // Why: true PTY geometry changes should be followed by the TUI's + // redraw output, not a full scrollback replay. The client resizes + // xterm geometry and consumes subsequent live output on this stream. + sendFrame( + TerminalStreamOpcode.Resized, + encodeTerminalStreamJson({ + cols: event.cols, + rows: event.rows, + displayMode: event.displayMode, + reason: event.reason, + seq: event.seq + }) + ) }) // Legacy fit-override-changed for non-mobile (desktop) subscribers @@ -488,16 +672,21 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ // evict each other via registerSubscriptionCleanup's // duplicate-key cleanup. See docs/mobile-presence-lock.md. const subscriptionId = clientId ? `${params.terminal}:${clientId}` : params.terminal - runtime.registerSubscriptionCleanup(subscriptionId, () => { - unsubscribeData() - unsubscribeResize() - unsubscribeFit() - if (isMobile && clientId) { - runtime.handleMobileUnsubscribe(ptyId, clientId) - } - emit({ type: 'end' }) - resolve() - }) + runtime.registerSubscriptionCleanup( + subscriptionId, + () => { + closed = true + unsubscribeData() + unsubscribeResize() + unsubscribeFit() + if (isMobile && clientId) { + runtime.handleMobileUnsubscribe(ptyId, clientId) + } + emit({ type: 'end' }) + resolve() + }, + connectionId + ) }) } }), diff --git a/src/main/runtime/rpc/ws-transport.ts b/src/main/runtime/rpc/ws-transport.ts index 5a97642a30c..b6f34c71113 100644 --- a/src/main/runtime/rpc/ws-transport.ts +++ b/src/main/runtime/rpc/ws-transport.ts @@ -10,6 +10,14 @@ import type { RpcTransport } from './transport' const MAX_WS_MESSAGE_BYTES = 1024 * 1024 const MAX_WS_CONNECTIONS = 32 +type WebSocketMessagePayload = string | Uint8Array +type WebSocketMessageHandler = { + bivarianceHack( + msg: WebSocketMessagePayload, + reply: (response: string) => void, + ws: WebSocket + ): void +}['bivarianceHack'] // Why: mobile clients (iOS/Android) regularly background-suspend their // sockets without the OS sending a TCP FIN/RST, leaving the server with @@ -45,9 +53,7 @@ export class WebSocketTransport implements RpcTransport { // sweep. A socket missing from the set when the next sweep fires is // assumed dead and terminated. private wsAlive = new WeakSet() - private messageHandler: - | ((msg: string, reply: (response: string) => void, ws: WebSocket) => void) - | null = null + private messageHandler: WebSocketMessageHandler | null = null private connectionCloseHandler: | ((clientId: string, ws: WebSocket, hasOtherConnections: boolean) => void) | null = null @@ -63,9 +69,7 @@ export class WebSocketTransport implements RpcTransport { this.heartbeatIntervalMs = heartbeatIntervalMs ?? HEARTBEAT_INTERVAL_MS } - onMessage( - handler: (msg: string, reply: (response: string) => void, ws: WebSocket) => void - ): void { + onMessage(handler: WebSocketMessageHandler): void { this.messageHandler = handler } @@ -240,13 +244,18 @@ export class WebSocketTransport implements RpcTransport { this.wsAlive.add(ws) }) - ws.on('message', (data) => { + ws.on('message', (data, isBinary) => { // Why: any inbound traffic counts as proof of life, not just pongs. // RN's WebSocket runtime auto-pongs server pings transparently, but // app-level frames also count toward liveness so an actively-talking // client doesn't get terminated mid-request. this.wsAlive.add(ws) - const msg = typeof data === 'string' ? data : data.toString('utf-8') + const msg = + typeof data === 'string' + ? data + : isBinary + ? new Uint8Array(data as Buffer) + : data.toString() this.messageHandler?.( msg, (response) => { diff --git a/src/main/runtime/runtime-rpc.ts b/src/main/runtime/runtime-rpc.ts index c1f946e6990..95c1bae1064 100644 --- a/src/main/runtime/runtime-rpc.ts +++ b/src/main/runtime/runtime-rpc.ts @@ -250,8 +250,14 @@ export class OrcaRuntimeRpcServer { ws.close(code, reason) } }) - channel.onMessage((plaintext, encryptedReply) => { - void this.handleWebSocketMessage(plaintext, encryptedReply, wsTransport, ws) + channel.onMessage((plaintext, encryptedReply, encryptedBinaryReply) => { + void this.handleWebSocketMessage( + plaintext, + encryptedReply, + encryptedBinaryReply, + wsTransport, + ws + ) }) this.e2eeChannels.set(ws, channel) } @@ -411,6 +417,7 @@ export class OrcaRuntimeRpcServer { private async handleWebSocketMessage( rawMessage: string, reply: (response: string) => void, + sendBinary: (response: Uint8Array) => void, wsTransport?: WebSocketTransport, ws?: WebSocket ): Promise { @@ -451,7 +458,7 @@ export class OrcaRuntimeRpcServer { } const connectionId = ws ? this.wsConnectionIds.get(ws) : undefined - await this.dispatcher.dispatchStreaming(request, reply, { connectionId }) + await this.dispatcher.dispatchStreaming(request, reply, { connectionId, sendBinary }) } private buildError(id: string, code: string, message: string): RpcResponse { diff --git a/src/shared/protocol-compat.test.ts b/src/shared/protocol-compat.test.ts index 1798e97fc87..22d5d4f5fa2 100644 --- a/src/shared/protocol-compat.test.ts +++ b/src/shared/protocol-compat.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from 'vitest' import { evaluateCompat } from './protocol-compat' +import { DESKTOP_PROTOCOL_VERSION, MIN_COMPATIBLE_MOBILE_VERSION } from './protocol-version' const MOBILE_V = 1 @@ -104,4 +105,20 @@ describe('evaluateCompat', () => { ).toEqual({ kind: 'ok' }) } }) + + it('hard-blocks protocol-1 mobile for the binary terminal stream cutover', () => { + const verdict = evaluateCompat({ + mobileProtocolVersion: 1, + minCompatibleDesktopVersion: DESKTOP_PROTOCOL_VERSION, + desktopProtocolVersion: DESKTOP_PROTOCOL_VERSION, + desktopMinCompatibleMobileVersion: MIN_COMPATIBLE_MOBILE_VERSION + }) + + expect(verdict).toEqual({ + kind: 'blocked', + reason: 'mobile-too-old', + desktopVersion: DESKTOP_PROTOCOL_VERSION, + requiredMobileVersion: MIN_COMPATIBLE_MOBILE_VERSION + }) + }) }) diff --git a/src/shared/protocol-version.ts b/src/shared/protocol-version.ts index e8232991356..63aebbdc312 100644 --- a/src/shared/protocol-version.ts +++ b/src/shared/protocol-version.ts @@ -19,5 +19,5 @@ // "kill switch": desktop can refuse old mobile builds without needing // a desktop release of mobile. -export const DESKTOP_PROTOCOL_VERSION = 1 -export const MIN_COMPATIBLE_MOBILE_VERSION = 0 +export const DESKTOP_PROTOCOL_VERSION = 2 +export const MIN_COMPATIBLE_MOBILE_VERSION = 2 diff --git a/src/shared/terminal-stream-protocol.test.ts b/src/shared/terminal-stream-protocol.test.ts new file mode 100644 index 00000000000..5dec28c811e --- /dev/null +++ b/src/shared/terminal-stream-protocol.test.ts @@ -0,0 +1,63 @@ +import { describe, expect, it } from 'vitest' +import { + TerminalStreamOpcode, + decodeTerminalStreamFrame, + decodeTerminalStreamJson, + decodeTerminalStreamText, + encodeTerminalStreamFrame, + encodeTerminalStreamJson, + encodeTerminalStreamText +} from './terminal-stream-protocol' + +describe('terminal-stream-protocol', () => { + it('round-trips fixed-width binary frame headers and payloads', () => { + const payload = encodeTerminalStreamText('hello terminal') + const encoded = encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Output, + streamId: 42, + seq: 9, + payload + }) + + const decoded = decodeTerminalStreamFrame(encoded) + + expect(decoded?.opcode).toBe(TerminalStreamOpcode.Output) + expect(decoded?.streamId).toBe(42) + expect(decoded?.seq).toBe(9) + expect(decoded ? decodeTerminalStreamText(decoded.payload) : '').toBe('hello terminal') + }) + + it('round-trips snapshot metadata JSON payloads', () => { + const encoded = encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.SnapshotStart, + streamId: 7, + seq: 1, + payload: encodeTerminalStreamJson({ kind: 'scrollback', cols: 49, rows: 28 }) + }) + + const decoded = decodeTerminalStreamFrame(encoded) + + expect(decoded && decodeTerminalStreamJson(decoded.payload)).toEqual({ + kind: 'scrollback', + cols: 49, + rows: 28 + }) + }) + + it('rejects unknown frame versions and opcodes', () => { + const encoded = encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Output, + streamId: 1, + seq: 1, + payload: new Uint8Array() + }) + + const badVersion = encoded.slice() + badVersion[1] = 99 + expect(decodeTerminalStreamFrame(badVersion)).toBeNull() + + const badOpcode = encoded.slice() + badOpcode[2] = 99 + expect(decodeTerminalStreamFrame(badOpcode)).toBeNull() + }) +}) diff --git a/src/shared/terminal-stream-protocol.ts b/src/shared/terminal-stream-protocol.ts new file mode 100644 index 00000000000..daa709f415b --- /dev/null +++ b/src/shared/terminal-stream-protocol.ts @@ -0,0 +1,87 @@ +const TERMINAL_STREAM_KIND = 0x74 +const TERMINAL_STREAM_VERSION = 1 +const HEADER_BYTES = 16 + +export enum TerminalStreamOpcode { + Output = 1, + SnapshotStart = 2, + SnapshotChunk = 3, + SnapshotEnd = 4, + Resized = 5, + Error = 6 +} + +export type TerminalStreamFrame = { + opcode: TerminalStreamOpcode + streamId: number + seq: number + payload: Uint8Array +} + +export function encodeTerminalStreamFrame(frame: TerminalStreamFrame): Uint8Array { + const out = new Uint8Array(HEADER_BYTES + frame.payload.length) + const view = new DataView(out.buffer, out.byteOffset, out.byteLength) + view.setUint8(0, TERMINAL_STREAM_KIND) + view.setUint8(1, TERMINAL_STREAM_VERSION) + view.setUint8(2, frame.opcode) + view.setUint8(3, 0) + view.setUint32(4, frame.streamId, true) + const seq = Math.max(0, Math.floor(frame.seq)) + view.setUint32(8, Math.floor(seq / 0x100000000), true) + view.setUint32(12, seq >>> 0, true) + out.set(frame.payload, HEADER_BYTES) + return out +} + +export function decodeTerminalStreamFrame(bytes: Uint8Array): TerminalStreamFrame | null { + if (bytes.length < HEADER_BYTES) { + return null + } + const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength) + if (view.getUint8(0) !== TERMINAL_STREAM_KIND || view.getUint8(1) !== TERMINAL_STREAM_VERSION) { + return null + } + const opcode = view.getUint8(2) + if (!isTerminalStreamOpcode(opcode)) { + return null + } + const high = view.getUint32(8, true) + const low = view.getUint32(12, true) + return { + opcode, + streamId: view.getUint32(4, true), + seq: high * 0x100000000 + low, + payload: bytes.slice(HEADER_BYTES) + } +} + +export function encodeTerminalStreamJson(value: unknown): Uint8Array { + return new TextEncoder().encode(JSON.stringify(value)) +} + +export function decodeTerminalStreamJson(payload: Uint8Array): T | null { + try { + return JSON.parse(new TextDecoder().decode(payload)) as T + } catch { + return null + } +} + +export function encodeTerminalStreamText(value: string): Uint8Array { + return new TextEncoder().encode(value) +} + +export function decodeTerminalStreamText(payload: Uint8Array): string { + return new TextDecoder().decode(payload) +} + +function isTerminalStreamOpcode(value: number): value is TerminalStreamOpcode { + return ( + value === TerminalStreamOpcode.Output || + value === TerminalStreamOpcode.SnapshotStart || + value === TerminalStreamOpcode.SnapshotChunk || + value === TerminalStreamOpcode.SnapshotEnd || + value === TerminalStreamOpcode.Resized || + value === TerminalStreamOpcode.Error + ) +}