From 3af90c0ffed301bcebf7010902bb6bef47be2366 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 20 Aug 2026 19:11:15 -0700 Subject: [PATCH] fix(mobile): pipeline live terminal input instead of awaiting each RPC Live keystrokes were stop-and-wait on terminal.send, so bursts still paid RTT between the first write and the coalesced tail. Reuse the existing subscribe Input frames so writes return after the socket send, matching desktop multiplex input, and fall back to terminal.send when the host has not advertised a binary stream id. Fixes #12071 --- .../app/h/[hostId]/session/[worktreeId].tsx | 33 ++-- .../terminal/send-live-terminal-input.test.ts | 148 ++++++++++++++++++ .../src/terminal/send-live-terminal-input.ts | 51 ++++++ .../terminal-live-accessory-raw-send.test.ts | 11 ++ .../terminal-live-accessory-raw-send.ts | 6 +- mobile/src/transport/e2ee.ts | 2 +- .../mobile-relay-rpc-session.test.ts | 20 ++- .../src/transport/mobile-relay-rpc-session.ts | 6 + .../mobile-relay-rpc-streams.test.ts | 12 ++ .../src/transport/mobile-relay-rpc-streams.ts | 16 ++ .../rpc-client-encrypted-binary-send.test.ts | 69 ++++++++ .../rpc-client-encrypted-binary-send.ts | 84 ++++++++++ .../rpc-client-terminal-input-send.test.ts | 82 ++++++++++ .../rpc-client-terminal-input-send.ts | 96 ++++++++++++ .../rpc-client-terminal-input.test.ts | 137 ++++++++++++++++ mobile/src/transport/rpc-client.ts | 59 ++++--- .../stable-logical-rpc-client.test.ts | 10 ++ .../transport/stable-logical-rpc-client.ts | 4 + .../src/transport/terminal-stream-protocol.ts | 6 + 19 files changed, 801 insertions(+), 51 deletions(-) create mode 100644 mobile/src/terminal/send-live-terminal-input.test.ts create mode 100644 mobile/src/terminal/send-live-terminal-input.ts create mode 100644 mobile/src/transport/rpc-client-encrypted-binary-send.test.ts create mode 100644 mobile/src/transport/rpc-client-encrypted-binary-send.ts create mode 100644 mobile/src/transport/rpc-client-terminal-input-send.test.ts create mode 100644 mobile/src/transport/rpc-client-terminal-input-send.ts create mode 100644 mobile/src/transport/rpc-client-terminal-input.test.ts diff --git a/mobile/app/h/[hostId]/session/[worktreeId].tsx b/mobile/app/h/[hostId]/session/[worktreeId].tsx index 14a94513507..55df4a35a64 100644 --- a/mobile/app/h/[hostId]/session/[worktreeId].tsx +++ b/mobile/app/h/[hostId]/session/[worktreeId].tsx @@ -125,7 +125,7 @@ import { import { useTerminalLiveInputFocus } from '../../../../src/terminal/use-terminal-live-input-focus' import { dismissTerminalKeyboard } from '../../../../src/terminal/terminal-keyboard-dismiss' import type { TerminalLiveInputSender } from '../../../../src/terminal/terminal-live-input-sender' -import { isTerminalSendRpcAccepted } from '../../../../src/terminal/terminal-send-rpc-response' +import { sendLiveTerminalInputBytes } from '../../../../src/terminal/send-live-terminal-input' import { sendMobileTerminalQueryReply } from '../../../../src/terminal/mobile-terminal-query-reply' import { TERMINAL_QUERY_REPLY_INPUT_RUNTIME_CAPABILITY } from '../../../../../src/shared/protocol-version' import { useTerminalLiveInputCommit } from '../../../../src/terminal/use-terminal-live-input-commit' @@ -3024,30 +3024,17 @@ export default function SessionScreen() { showToast('Input too large (max 256 KiB)', 1500) return false } - const rpc = clientRef.current - // Why: callers suppress follow-up controls/toasts when this live send is stale. - if ( - !rpc || - connStateRef.current !== 'connected' || - handle !== activeHandleRef.current || - activeSessionTabTypeRef.current !== 'terminal' - ) { - return false - } // Why: live-mirror deltas queued behind a dying send drain into the connect // wait and replay stale bytes after reconnect (#6713's `YZZYecho …` corruption). - return rpc - .sendRequest( - 'terminal.send', - buildTerminalSendParams({ - terminal: handle, - text, - enter: false, - deviceToken: deviceTokenRef.current - }), - TERMINAL_INPUT_SEND_OPTIONS - ) - .then(isTerminalSendRpcAccepted, () => false) + return sendLiveTerminalInputBytes({ + rpc: clientRef.current, + handle, + bytes: text, + connected: connStateRef.current === 'connected', + activeHandle: activeHandleRef.current, + activeSessionTabType: activeSessionTabTypeRef.current, + deviceToken: deviceTokenRef.current + }) }, [showToast] ) diff --git a/mobile/src/terminal/send-live-terminal-input.test.ts b/mobile/src/terminal/send-live-terminal-input.test.ts new file mode 100644 index 00000000000..ebf97eb58b7 --- /dev/null +++ b/mobile/src/terminal/send-live-terminal-input.test.ts @@ -0,0 +1,148 @@ +import { describe, expect, it, vi } from 'vitest' +import type { RpcResponse } from '../transport/types' +import { + queueTerminalLiveMirrorSend, + createTerminalLivePendingFlushState +} from './terminal-live-pending-flush-state' +import { sendLiveTerminalInputBytes, type LiveTerminalInputRpc } from './send-live-terminal-input' + +function acceptedSend(): RpcResponse { + return { + id: 'rpc-1', + ok: true, + result: { send: { accepted: true } }, + _meta: { runtimeId: 'runtime-1' } + } +} + +function deferred(): { + promise: Promise + resolve: (value: T) => void +} { + let resolve: (value: T) => void = () => {} + const promise = new Promise((next) => { + resolve = next + }) + return { promise, resolve } +} + +describe('send live terminal input bytes', () => { + it('Given a held RPC response When later deltas queue Then later writes leave before the first response', async () => { + const streamWrites: string[] = [] + const sendRequest = vi.fn(() => deferred().promise) + const rpc: LiveTerminalInputRpc = { + sendRequest, + sendTerminalInput: (_terminal, text) => { + streamWrites.push(text) + return 'sent' + } + } + const state = createTerminalLivePendingFlushState() + const sender = (handle: string, payload: string): Promise => + sendLiveTerminalInputBytes({ + rpc, + handle, + bytes: payload, + connected: true, + activeHandle: handle, + activeSessionTabType: 'terminal', + deviceToken: 'phone-1' + }) + + const first = queueTerminalLiveMirrorSend(state, 'term-1', 'a', sender) + await Promise.resolve() + const second = queueTerminalLiveMirrorSend(state, 'term-1', 'b', sender) + const third = queueTerminalLiveMirrorSend(state, 'term-1', 'c', sender) + await Promise.resolve() + + expect(streamWrites.join('')).toBe('abc') + expect(sendRequest).not.toHaveBeenCalled() + await expect(Promise.all([first, second, third])).resolves.toEqual([true, true, true]) + }) + + it('Given a held RPC response When later deltas queue Then bytes stay in order and none drop', async () => { + const streamWrites: string[] = [] + const rpc: LiveTerminalInputRpc = { + sendRequest: vi.fn(), + sendTerminalInput: (_terminal, text) => { + streamWrites.push(text) + return 'sent' + } + } + const state = createTerminalLivePendingFlushState() + const sender = (handle: string, payload: string): Promise => + sendLiveTerminalInputBytes({ + rpc, + handle, + bytes: payload, + connected: true, + activeHandle: handle, + activeSessionTabType: 'terminal', + deviceToken: 'phone-1' + }) + + const first = queueTerminalLiveMirrorSend(state, 'term-1', 'h', sender) + await Promise.resolve() + const rest = 'ello world' + .split('') + .map((letter) => queueTerminalLiveMirrorSend(state, 'term-1', letter, sender)) + await Promise.all([first, ...rest]) + + expect(streamWrites.join('')).toBe('hello world') + }) + + it('Given a failed stream write When input is live Then it does not fall back to terminal.send', async () => { + const sendRequest = vi.fn() + const rpc: LiveTerminalInputRpc = { + sendRequest, + sendTerminalInput: () => 'failed' + } + + await expect( + sendLiveTerminalInputBytes({ + rpc, + handle: 'term-1', + bytes: 'a', + connected: true, + activeHandle: 'term-1', + activeSessionTabType: 'terminal', + deviceToken: 'phone-1' + }) + ).resolves.toBe(false) + expect(sendRequest).not.toHaveBeenCalled() + }) + + it('Given no stream input When RTT is held Then fallback still coalesces later bytes onto one follow-up RPC', async () => { + const firstRpc = deferred() + const sendRequest = vi.fn(() => firstRpc.promise) + const rpc: LiveTerminalInputRpc = { sendRequest } + const state = createTerminalLivePendingFlushState() + const sender = (handle: string, payload: string): Promise => + sendLiveTerminalInputBytes({ + rpc, + handle, + bytes: payload, + connected: true, + activeHandle: handle, + activeSessionTabType: 'terminal', + deviceToken: 'phone-1' + }) + + const first = queueTerminalLiveMirrorSend(state, 'term-1', 'a', sender) + await Promise.resolve() + const second = queueTerminalLiveMirrorSend(state, 'term-1', 'b', sender) + const third = queueTerminalLiveMirrorSend(state, 'term-1', 'c', sender) + await Promise.resolve() + + expect(sendRequest).toHaveBeenCalledTimes(1) + expect(sendRequest.mock.calls[0]?.[1]).toMatchObject({ text: 'a' }) + + const secondRpc = deferred() + sendRequest.mockImplementationOnce(() => secondRpc.promise) + firstRpc.resolve(acceptedSend()) + await vi.waitFor(() => expect(sendRequest).toHaveBeenCalledTimes(2)) + expect(sendRequest.mock.calls[1]?.[1]).toMatchObject({ text: 'bc' }) + secondRpc.resolve(acceptedSend()) + await expect(Promise.all([first, second, third])).resolves.toEqual([true, true, true]) + }) +}) diff --git a/mobile/src/terminal/send-live-terminal-input.ts b/mobile/src/terminal/send-live-terminal-input.ts new file mode 100644 index 00000000000..971fde4a275 --- /dev/null +++ b/mobile/src/terminal/send-live-terminal-input.ts @@ -0,0 +1,51 @@ +import type { RpcClient } from '../transport/rpc-client' +import { isTerminalSendRpcAccepted } from './terminal-send-rpc-response' +import { buildTerminalSendParams, TERMINAL_INPUT_SEND_OPTIONS } from './terminal-send-request' +import { normalizeTerminalTextInput } from './terminal-text-input-normalization' + +export type LiveTerminalInputRpc = Pick & + Partial> + +export async function sendLiveTerminalInputBytes(args: { + rpc: LiveTerminalInputRpc | null + handle: string + bytes: string + connected: boolean + activeHandle: string | null + activeSessionTabType: string | null + deviceToken: string | null +}): Promise { + const text = normalizeTerminalTextInput(args.bytes) + if (text.length === 0) { + return false + } + if ( + !args.rpc || + !args.connected || + args.handle !== args.activeHandle || + args.activeSessionTabType !== 'terminal' + ) { + return false + } + // Why: Input frames return after the socket write; waiting for terminal.send + // would reintroduce one RTT per mirror delta on every transport. + const streamResult = args.rpc.sendTerminalInput?.(args.handle, text) + if (streamResult === 'sent') { + return true + } + if (streamResult === 'failed') { + return false + } + return args.rpc + .sendRequest( + 'terminal.send', + buildTerminalSendParams({ + terminal: args.handle, + text, + enter: false, + deviceToken: args.deviceToken + }), + TERMINAL_INPUT_SEND_OPTIONS + ) + .then(isTerminalSendRpcAccepted, () => false) +} diff --git a/mobile/src/terminal/terminal-live-accessory-raw-send.test.ts b/mobile/src/terminal/terminal-live-accessory-raw-send.test.ts index 7a1110b407b..4bee4fb630b 100644 --- a/mobile/src/terminal/terminal-live-accessory-raw-send.test.ts +++ b/mobile/src/terminal/terminal-live-accessory-raw-send.test.ts @@ -17,6 +17,17 @@ const BASE_ARGS = { } as const describe('terminal live accessory raw send', () => { + it('writes accessory bytes on the subscribe stream when the host has one', async () => { + const sendTerminalInput = vi.fn(() => 'sent' as const) + const sendRequest = vi.fn() + const client = { sendRequest, sendTerminalInput } + + await sendTerminalLiveAccessoryRawBytes({ ...BASE_ARGS, client }) + + expect(sendTerminalInput).toHaveBeenCalledWith('terminal-a', '') + expect(sendRequest).not.toHaveBeenCalled() + }) + it('sends raw bytes now-or-never with the device presence tag', async () => { const { client, sendRequest } = captureClient() diff --git a/mobile/src/terminal/terminal-live-accessory-raw-send.ts b/mobile/src/terminal/terminal-live-accessory-raw-send.ts index 792491f6c63..48de268c3ab 100644 --- a/mobile/src/terminal/terminal-live-accessory-raw-send.ts +++ b/mobile/src/terminal/terminal-live-accessory-raw-send.ts @@ -4,7 +4,7 @@ import type { RpcClient } from '../transport/rpc-client' import type { ConnectionState } from '../transport/types' type TerminalLiveAccessoryRawSendArgs = { - readonly client: Pick | null + readonly client: Pick | null readonly targetHandle: string readonly activeHandle: string | null readonly activeSessionTabType: string | null @@ -25,6 +25,10 @@ export async function sendTerminalLiveAccessoryRawBytes( if (!args.client || !rawSendTarget || args.connState !== 'connected') { return } + const streamResult = args.client.sendTerminalInput?.(rawSendTarget, args.bytes) + if (streamResult === 'sent' || streamResult === 'failed') { + return + } await args.client .sendRequest( 'terminal.send', diff --git a/mobile/src/transport/e2ee.ts b/mobile/src/transport/e2ee.ts index 2732b3d615f..a6dbf6f346f 100644 --- a/mobile/src/transport/e2ee.ts +++ b/mobile/src/transport/e2ee.ts @@ -73,7 +73,7 @@ export function decrypt(encrypted: string, sharedKey: Uint8Array): string | null return plaintext ? new TextDecoder().decode(plaintext) : null } -function encryptBytes(plaintext: Uint8Array, sharedKey: Uint8Array): Uint8Array { +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)) diff --git a/mobile/src/transport/mobile-relay-rpc-session.test.ts b/mobile/src/transport/mobile-relay-rpc-session.test.ts index d5b547885cf..122841cec05 100644 --- a/mobile/src/transport/mobile-relay-rpc-session.test.ts +++ b/mobile/src/transport/mobile-relay-rpc-session.test.ts @@ -3,7 +3,12 @@ import { BrowserScreencastOpcode, encodeBrowserScreencastFrame } from '../../../src/shared/browser-screencast-protocol' -import { encodeTerminalStreamFrame, TerminalStreamOpcode } from './terminal-stream-protocol' +import { + decodeTerminalStreamFrame, + decodeTerminalStreamText, + encodeTerminalStreamFrame, + TerminalStreamOpcode +} from './terminal-stream-protocol' import { isRpcDeliveryUnknown } from './rpc-delivery-ambiguity' const fakes = vi.hoisted(() => ({ @@ -18,6 +23,7 @@ const fakes = vi.hoisted(() => ({ onError(error: Error): void }, sendText: vi.fn(() => true), + sendBinary: vi.fn(() => true), close: vi.fn() })) @@ -27,6 +33,7 @@ vi.mock('./mobile-relay-e2ee-link', () => ({ fakes.linkOptions = options } sendText = fakes.sendText + sendBinary = fakes.sendBinary close = fakes.close } })) @@ -102,6 +109,7 @@ describe('mobile relay RPC session', () => { vi.clearAllMocks() fakes.linkOptions = null fakes.sendText.mockReturnValue(true) + fakes.sendBinary.mockReturnValue(true) }) afterEach(() => vi.useRealTimers()) @@ -169,8 +177,18 @@ describe('mobile relay RPC session', () => { streamId: 42, chunk: 'hello' }) + expect(session.sendTerminalInput('term-1', 'xy')).toBe('sent') + expect(fakes.sendBinary).toHaveBeenCalledOnce() + const binaryPayload = fakes.sendBinary.mock.calls[0]?.[0] + expect(binaryPayload).toBeInstanceOf(Uint8Array) + const input = + binaryPayload instanceof Uint8Array ? decodeTerminalStreamFrame(binaryPayload) : null + expect(input?.opcode).toBe(TerminalStreamOpcode.Input) + expect(input?.streamId).toBe(42) + expect(input ? decodeTerminalStreamText(input.payload) : '').toBe('xy') fakes.sendText.mockClear() + fakes.sendBinary.mockClear() const onBinaryFrame = vi.fn() session.subscribe('browser.screencast', {}, vi.fn(), { onBinaryFrame }) await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledOnce()) diff --git a/mobile/src/transport/mobile-relay-rpc-session.ts b/mobile/src/transport/mobile-relay-rpc-session.ts index 84b1e7cdbeb..83b5fd34003 100644 --- a/mobile/src/transport/mobile-relay-rpc-session.ts +++ b/mobile/src/transport/mobile-relay-rpc-session.ts @@ -57,6 +57,8 @@ export function connectMobileRelayRpcSession(args: { const streams = new MobileRelayRpcStreams({ nextId, sendFrame, + sendBinary: (bytes) => link.sendBinary(bytes), + isConnected: () => state === 'connected', waitForConnected: () => waitForConnected() }) @@ -105,6 +107,10 @@ export function connectMobileRelayRpcSession(args: { return streams.subscribe(method, params, listener, options) }, + sendTerminalInput(terminal, text) { + return streams.sendTerminalInput(terminal, text) + }, + updateTerminalSubscriptionViewport(terminal, viewport) { streams.updateTerminalViewport(terminal, viewport) }, diff --git a/mobile/src/transport/mobile-relay-rpc-streams.test.ts b/mobile/src/transport/mobile-relay-rpc-streams.test.ts index b0e59c6bf5b..6eb62e2d25c 100644 --- a/mobile/src/transport/mobile-relay-rpc-streams.test.ts +++ b/mobile/src/transport/mobile-relay-rpc-streams.test.ts @@ -18,6 +18,8 @@ describe('MobileRelayRpcStreams failure parity', () => { const streams = new MobileRelayRpcStreams({ nextId: () => 'stream-1', sendFrame, + sendBinary: () => true, + isConnected: () => true, waitForConnected: async () => {} }) const cancel = streams.subscribe( @@ -45,6 +47,8 @@ describe('MobileRelayRpcStreams failure parity', () => { const streams = new MobileRelayRpcStreams({ nextId: () => 'stream-1', sendFrame, + sendBinary: () => true, + isConnected: () => true, waitForConnected: () => Promise.reject(waitError) }) const cancel = streams.subscribe( @@ -71,6 +75,8 @@ describe('MobileRelayRpcStreams failure parity', () => { const streams = new MobileRelayRpcStreams({ nextId: () => 'stream-1', sendFrame, + sendBinary: () => true, + isConnected: () => true, waitForConnected: async () => {} }) const cancel = streams.subscribe( @@ -98,6 +104,8 @@ describe('MobileRelayRpcStreams failure parity', () => { const streams = new MobileRelayRpcStreams({ nextId: () => 'stream-1', sendFrame, + sendBinary: () => true, + isConnected: () => true, waitForConnected: () => connection.promise }) const cancel = streams.subscribe( @@ -121,6 +129,8 @@ describe('MobileRelayRpcStreams failure parity', () => { const streams = new MobileRelayRpcStreams({ nextId: () => 'stream-1', sendFrame, + sendBinary: () => true, + isConnected: () => true, waitForConnected: () => connection.promise }) streams.subscribe('session.tabs.subscribe', { worktree: 'id:worktree-1' }, listener) @@ -140,6 +150,8 @@ describe('MobileRelayRpcStreams failure parity', () => { const streams = new MobileRelayRpcStreams({ nextId: () => 'stream-1', sendFrame: () => true, + sendBinary: () => true, + isConnected: () => true, waitForConnected: async () => {} }) streams.subscribe('session.tabs.subscribe', { worktree: 'id:worktree-1' }, listener) diff --git a/mobile/src/transport/mobile-relay-rpc-streams.ts b/mobile/src/transport/mobile-relay-rpc-streams.ts index 2eacbd2168f..f0884735d70 100644 --- a/mobile/src/transport/mobile-relay-rpc-streams.ts +++ b/mobile/src/transport/mobile-relay-rpc-streams.ts @@ -7,6 +7,7 @@ import { buildTerminalUnsubscribeParams, updateTerminalSubscriptionViewport } from './rpc-client-terminal-subscription' +import { TerminalInputStreamRegistry } from './rpc-client-terminal-input-send' import type { RpcClient } from './rpc-client' import type { RpcResponse, RpcSuccess } from './types' @@ -27,6 +28,8 @@ type StreamRecord = { type StreamManagerOptions = { nextId: () => string sendFrame: (request: { id: string; method: string; params?: unknown }) => boolean + sendBinary: (bytes: Uint8Array) => boolean + isConnected: () => boolean waitForConnected: () => Promise } @@ -34,6 +37,7 @@ export class MobileRelayRpcStreams { private readonly streams = new Map() private readonly terminalListeners = new Map void>() private readonly terminalSnapshots = new Map() + private readonly terminalInputStreams = new TerminalInputStreamRegistry() private activeBrowserStream: StreamRecord | null = null constructor(private readonly options: StreamManagerOptions) {} @@ -92,6 +96,7 @@ export class MobileRelayRpcStreams { if (typeof metadata.streamId === 'number') { stream.streamIds.add(metadata.streamId) this.terminalListeners.set(metadata.streamId, stream.listener) + this.terminalInputStreams.remember(stream.params, metadata.streamId) } if (stream.method === 'browser.screencast') { this.activeBrowserStream = stream @@ -108,6 +113,15 @@ export class MobileRelayRpcStreams { return true } + sendTerminalInput(terminal: string, text: string) { + return this.terminalInputStreams.send( + terminal, + text, + this.options.isConnected(), + this.options.sendBinary + ) + } + handleBinary(bytes: Uint8Array): void { const browserFrame = decodeBrowserScreencastFrame(bytes) if (browserFrame && this.activeBrowserStream?.onBinaryFrame) { @@ -127,6 +141,7 @@ export class MobileRelayRpcStreams { this.streams.clear() this.terminalListeners.clear() this.terminalSnapshots.clear() + this.terminalInputStreams.clear() this.activeBrowserStream = null } @@ -160,6 +175,7 @@ export class MobileRelayRpcStreams { if (!stream) { return } + this.terminalInputStreams.forget(stream.params, stream.streamIds) for (const streamId of stream.streamIds) { this.terminalListeners.delete(streamId) this.terminalSnapshots.delete(streamId) diff --git a/mobile/src/transport/rpc-client-encrypted-binary-send.test.ts b/mobile/src/transport/rpc-client-encrypted-binary-send.test.ts new file mode 100644 index 00000000000..db02d932cf5 --- /dev/null +++ b/mobile/src/transport/rpc-client-encrypted-binary-send.test.ts @@ -0,0 +1,69 @@ +import { describe, expect, it, vi } from 'vitest' +import { + sendEncryptedBinaryPayload, + sendLanTerminalInputFrame +} from './rpc-client-encrypted-binary-send' +import { TerminalInputStreamRegistry } from './rpc-client-terminal-input-send' + +vi.mock('./e2ee', () => ({ + encrypt: (plaintext: string) => plaintext, + encryptBytes: (bytes: Uint8Array) => bytes +})) + +describe('sendEncryptedBinaryPayload', () => { + it('writes encrypted bytes on an open socket and reports a closed socket as failed', () => { + const sent: Uint8Array[] = [] + const openSocket = { + OPEN: 1, + readyState: 1, + send: (payload: Uint8Array) => { + sent.push(payload) + } + } as unknown as WebSocket + expect( + sendEncryptedBinaryPayload({ + socket: openSocket, + sharedKey: new Uint8Array(32), + plaintext: new Uint8Array([1, 2]), + onWriteError: () => {} + }) + ).toBe(true) + expect(sent).toEqual([new Uint8Array([1, 2])]) + + const closedSocket = { OPEN: 1, readyState: 3, send: vi.fn() } as unknown as WebSocket + expect( + sendEncryptedBinaryPayload({ + socket: closedSocket, + sharedKey: new Uint8Array(32), + plaintext: new Uint8Array([1]), + onWriteError: () => {} + }) + ).toBe(false) + }) + + it('sends an Input frame for a remembered terminal stream', () => { + const registry = new TerminalInputStreamRegistry() + registry.remember({ terminal: 'term-1' }, 7) + const sent: Uint8Array[] = [] + const openSocket = { + OPEN: 1, + readyState: 1, + send: (payload: Uint8Array) => { + sent.push(payload) + } + } as unknown as WebSocket + expect( + sendLanTerminalInputFrame({ + registry, + terminal: 'term-1', + text: 'ab', + connected: true, + socket: openSocket, + sharedKey: new Uint8Array(32), + isCurrentSocket: (socket) => socket === openSocket, + onWriteError: () => {} + }) + ).toBe('sent') + expect(sent).toHaveLength(1) + }) +}) diff --git a/mobile/src/transport/rpc-client-encrypted-binary-send.ts b/mobile/src/transport/rpc-client-encrypted-binary-send.ts new file mode 100644 index 00000000000..07a0f298a8a --- /dev/null +++ b/mobile/src/transport/rpc-client-encrypted-binary-send.ts @@ -0,0 +1,84 @@ +import { encrypt, encryptBytes } from './e2ee' +import { + TerminalInputStreamRegistry, + type TerminalInputSendResult +} from './rpc-client-terminal-input-send' +import type { ConnectionState } from './types' + +export function sendEncryptedJson(args: { + socket: WebSocket | null + sharedKey: Uint8Array | null + request: unknown + state: ConnectionState + onWriteError: (socket: WebSocket) => void + onDesync: (socket: WebSocket) => void +}): boolean { + const { socket, sharedKey, request, state, onWriteError, onDesync } = args + if (socket && socket.readyState === WebSocket.OPEN && sharedKey) { + try { + socket.send(encrypt(JSON.stringify(request), sharedKey)) + return true + } catch { + onWriteError(socket) + return false + } + } + console.log('[net] sendEncrypted FAILED — channel not ready', { + hasWs: !!socket, + readyState: socket?.readyState, + hasKey: !!sharedKey, + state + }) + // Why: RN can drop onclose, leaving state 'connected' over a dead socket; force reconnect or every send silently fails forever. + if (state === 'connected' && socket && socket.readyState !== WebSocket.OPEN) { + console.log('[net] sendEncrypted detected ws desync — forcing reconnect', { + readyState: socket.readyState + }) + onDesync(socket) + } + return false +} + +export function sendEncryptedBinaryPayload(args: { + socket: WebSocket | null + sharedKey: Uint8Array | null + plaintext: Uint8Array + onWriteError: (socket: WebSocket) => void +}): boolean { + const { socket, sharedKey, plaintext, onWriteError } = args + if (!socket || socket.readyState !== WebSocket.OPEN || !sharedKey) { + return false + } + try { + socket.send(encryptBytes(plaintext, sharedKey)) + return true + } catch { + onWriteError(socket) + return false + } +} + +export function sendLanTerminalInputFrame(args: { + registry: TerminalInputStreamRegistry + terminal: string + text: string + connected: boolean + socket: WebSocket | null + sharedKey: Uint8Array | null + isCurrentSocket: (socket: WebSocket) => boolean + onWriteError: (socket: WebSocket) => void +}): TerminalInputSendResult { + const sendingWs = args.socket + return args.registry.send(args.terminal, args.text, args.connected, (plaintext) => + sendEncryptedBinaryPayload({ + socket: sendingWs, + sharedKey: args.sharedKey, + plaintext, + onWriteError: (socket) => { + if (args.isCurrentSocket(socket)) { + args.onWriteError(socket) + } + } + }) + ) +} diff --git a/mobile/src/transport/rpc-client-terminal-input-send.test.ts b/mobile/src/transport/rpc-client-terminal-input-send.test.ts new file mode 100644 index 00000000000..24d71b32c4f --- /dev/null +++ b/mobile/src/transport/rpc-client-terminal-input-send.test.ts @@ -0,0 +1,82 @@ +import { describe, expect, it } from 'vitest' +import { + encodeTerminalInputFrame, + forgetTerminalInputStream, + rememberTerminalInputStream, + sendRememberedTerminalInput, + terminalHandleFromSubscribeParams +} from './rpc-client-terminal-input-send' +import { + decodeTerminalStreamFrame, + decodeTerminalStreamText, + TerminalStreamOpcode +} from './terminal-stream-protocol' + +function payloadText(bytes: Uint8Array): string { + const frame = decodeTerminalStreamFrame(bytes) + return frame ? decodeTerminalStreamText(frame.payload) : '' +} + +describe('rpc-client terminal input send', () => { + it('encodes an Input frame for the remembered stream', () => { + const frame = decodeTerminalStreamFrame(encodeTerminalInputFrame(9, 'hi')) + expect(frame?.opcode).toBe(TerminalStreamOpcode.Input) + expect(frame?.streamId).toBe(9) + expect(frame).not.toBeNull() + if (!frame) { + return + } + expect(decodeTerminalStreamText(frame.payload)).toBe('hi') + }) + + it('sends in order and refuses a disconnected write', () => { + const streams = new Map() + rememberTerminalInputStream(streams, 'term-1', 4) + const writes: string[] = [] + expect( + sendRememberedTerminalInput({ + streams, + terminal: 'term-1', + text: 'a', + connected: true, + sendBinary: (bytes) => { + writes.push(payloadText(bytes)) + return true + } + }) + ).toBe('sent') + expect( + sendRememberedTerminalInput({ + streams, + terminal: 'term-1', + text: 'b', + connected: true, + sendBinary: (bytes) => { + writes.push(payloadText(bytes)) + return true + } + }) + ).toBe('sent') + expect(writes).toEqual(['a', 'b']) + expect( + sendRememberedTerminalInput({ + streams, + terminal: 'term-1', + text: 'c', + connected: false, + sendBinary: () => true + }) + ).toBe('failed') + expect(terminalHandleFromSubscribeParams({ terminal: 'term-1' })).toBe('term-1') + forgetTerminalInputStream(streams, 'term-1', 4) + expect( + sendRememberedTerminalInput({ + streams, + terminal: 'term-1', + text: 'd', + connected: true, + sendBinary: () => true + }) + ).toBe('no-stream') + }) +}) diff --git a/mobile/src/transport/rpc-client-terminal-input-send.ts b/mobile/src/transport/rpc-client-terminal-input-send.ts new file mode 100644 index 00000000000..cc6567d96c6 --- /dev/null +++ b/mobile/src/transport/rpc-client-terminal-input-send.ts @@ -0,0 +1,96 @@ +import { + encodeTerminalStreamFrame, + encodeTerminalStreamText, + TerminalStreamOpcode +} from './terminal-stream-protocol' + +export type TerminalInputSendResult = 'sent' | 'no-stream' | 'failed' + +export function terminalHandleFromSubscribeParams(params: unknown): string | null { + if (!params || typeof params !== 'object') { + return null + } + const terminal = (params as { terminal?: unknown }).terminal + return typeof terminal === 'string' ? terminal : null +} + +export function rememberTerminalInputStream( + streams: Map, + terminal: string, + streamId: number +): void { + streams.set(terminal, streamId) +} + +export function forgetTerminalInputStream( + streams: Map, + terminal: string | null, + streamId: number +): void { + if (terminal && streams.get(terminal) === streamId) { + streams.delete(terminal) + } +} + +export function encodeTerminalInputFrame(streamId: number, text: string): Uint8Array { + return encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Input, + streamId, + seq: 0, + payload: encodeTerminalStreamText(text) + }) +} + +export function sendRememberedTerminalInput(args: { + streams: ReadonlyMap + terminal: string + text: string + connected: boolean + sendBinary: (bytes: Uint8Array) => boolean +}): TerminalInputSendResult { + if (!args.connected) { + return 'failed' + } + const streamId = args.streams.get(args.terminal) + if (streamId === undefined) { + return 'no-stream' + } + return args.sendBinary(encodeTerminalInputFrame(streamId, args.text)) ? 'sent' : 'failed' +} + +export class TerminalInputStreamRegistry { + private readonly streams = new Map() + + remember(params: unknown, streamId: number): void { + const terminal = terminalHandleFromSubscribeParams(params) + if (terminal) { + rememberTerminalInputStream(this.streams, terminal, streamId) + } + } + + forget(params: unknown, streamIds: Iterable): void { + const terminal = terminalHandleFromSubscribeParams(params) + for (const streamId of streamIds) { + forgetTerminalInputStream(this.streams, terminal, streamId) + } + } + + clear(): void { + this.streams.clear() + } + + send( + terminal: string, + text: string, + connected: boolean, + sendBinary: (bytes: Uint8Array) => boolean + ): TerminalInputSendResult { + return sendRememberedTerminalInput({ + streams: this.streams, + terminal, + text, + connected, + sendBinary + }) + } +} diff --git a/mobile/src/transport/rpc-client-terminal-input.test.ts b/mobile/src/transport/rpc-client-terminal-input.test.ts new file mode 100644 index 00000000000..acc88ceda1c --- /dev/null +++ b/mobile/src/transport/rpc-client-terminal-input.test.ts @@ -0,0 +1,137 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { connect } from './rpc-client' +import { + decodeTerminalStreamFrame, + decodeTerminalStreamText, + TerminalStreamOpcode +} from './terminal-stream-protocol' + +vi.mock('./e2ee', () => ({ + generateKeyPair: () => ({ + publicKey: new Uint8Array(32), + secretKey: new Uint8Array(32) + }), + deriveSharedKey: () => new Uint8Array(32), + publicKeyFromBase64: () => new Uint8Array(32), + publicKeyToBase64: () => 'client-public-key', + encrypt: (plaintext: string) => `encrypted:${plaintext}`, + decrypt: (raw: string) => raw.replace(/^encrypted:/, ''), + encryptBytes: (bytes: Uint8Array) => bytes, + decryptBytes: (bytes: Uint8Array) => bytes +})) + +class MockWebSocket { + static OPEN = 1 + readonly OPEN = MockWebSocket.OPEN + readyState = 0 + onopen: (() => void) | null = null + onclose: (() => void) | null = null + onmessage: ((event: { data: unknown }) => void) | null = null + sent: Array = [] + close = vi.fn(() => { + this.readyState = 3 + this.onclose?.() + }) + + constructor(readonly endpoint: string) { + mockSockets.push(this) + } + + send(payload: string | Uint8Array): void { + this.sent.push(payload) + } + + open(): void { + this.readyState = MockWebSocket.OPEN + this.onopen?.() + } + + receive(payload: unknown): void { + this.onmessage?.({ data: payload }) + } +} + +const mockSockets: MockWebSocket[] = [] +const originalWebSocket = globalThis.WebSocket + +function authenticate(socket: MockWebSocket): void { + socket.open() + socket.receive(JSON.stringify({ type: 'e2ee_ready' })) + socket.receive('encrypted:{"type":"e2ee_authenticated"}') +} + +function sentRequest(socket: MockWebSocket, method: string): { id: string } { + for (const payload of socket.sent) { + if (typeof payload !== 'string') { + continue + } + const decoded = JSON.parse(payload.replace(/^encrypted:/, '')) as { + id: string + method: string + } + if (decoded.method === method) { + return { id: decoded.id } + } + } + throw new Error(`Request not sent: ${method}`) +} + +describe('rpc-client terminal input frames', () => { + beforeEach(() => { + mockSockets.length = 0 + globalThis.WebSocket = MockWebSocket as unknown as typeof WebSocket + }) + + afterEach(() => { + globalThis.WebSocket = originalWebSocket + }) + + it('writes live terminal input as a stream frame without waiting for an RPC response', () => { + const client = connect('ws://desktop.invalid', 'token', 'server-key') + const socket = mockSockets[0]! + authenticate(socket) + client.subscribe('terminal.subscribe', { terminal: 'term-1' }, () => {}) + socket.receive( + `encrypted:${JSON.stringify({ + id: sentRequest(socket, 'terminal.subscribe').id, + ok: true, + streaming: true, + result: { type: 'subscribed', streamId: 42 } + })}` + ) + + expect(client.sendTerminalInput?.('term-1', 'ab')).toBe('sent') + expect(client.sendTerminalInput?.('term-1', 'c')).toBe('sent') + const inputFrames = socket.sent + .filter((payload): payload is Uint8Array => payload instanceof Uint8Array) + .map((payload) => decodeTerminalStreamFrame(payload)) + .filter( + (frame): frame is NonNullable => frame?.opcode === TerminalStreamOpcode.Input + ) + expect(inputFrames.map((frame) => decodeTerminalStreamText(frame.payload))).toEqual(['ab', 'c']) + expect( + socket.sent.some( + (payload) => typeof payload === 'string' && payload.includes('"method":"terminal.send"') + ) + ).toBe(false) + client.close() + }) + + it('falls back to no-stream before subscribe and fails after disconnect', () => { + const client = connect('ws://desktop.invalid', 'token', 'server-key') + const socket = mockSockets[0]! + authenticate(socket) + expect(client.sendTerminalInput?.('term-1', 'a')).toBe('no-stream') + client.subscribe('terminal.subscribe', { terminal: 'term-1' }, () => {}) + socket.receive( + `encrypted:${JSON.stringify({ + id: sentRequest(socket, 'terminal.subscribe').id, + ok: true, + streaming: true, + result: { type: 'subscribed', streamId: 42 } + })}` + ) + client.close() + expect(client.sendTerminalInput?.('term-1', 'a')).toBe('failed') + }) +}) diff --git a/mobile/src/transport/rpc-client.ts b/mobile/src/transport/rpc-client.ts index b3c44504ae6..7343148c9a1 100644 --- a/mobile/src/transport/rpc-client.ts +++ b/mobile/src/transport/rpc-client.ts @@ -11,7 +11,6 @@ import { deriveSharedKey, publicKeyFromBase64, publicKeyToBase64, - encrypt, decrypt, decryptBytes } from './e2ee' @@ -28,6 +27,11 @@ import { buildTerminalUnsubscribeParams, updateTerminalSubscriptionViewport as updateCachedTerminalSubscriptionViewport } from './rpc-client-terminal-subscription' +import { sendEncryptedJson, sendLanTerminalInputFrame } from './rpc-client-encrypted-binary-send' +import { + TerminalInputStreamRegistry, + type TerminalInputSendResult +} from './rpc-client-terminal-input-send' import { describeSocketEvent, redactSocketEndpoint } from './socket-event-debug' import { isStaleRpcSocketEvent, @@ -102,6 +106,7 @@ export type RpcClient = { onData: StreamingListener, options?: SubscribeOptions ) => () => void + sendTerminalInput?: (terminal: string, text: string) => TerminalInputSendResult updateTerminalSubscriptionViewport: ( terminal: string, viewport: { cols: number; rows: number } @@ -202,6 +207,7 @@ export function connect( const streamListeners = new Map() const terminalStreamListeners = new Map() const terminalStreamIdsByRequest = new Map>() + const terminalInputStreams = new TerminalInputStreamRegistry() const terminalSnapshots = new Map() let activeBrowserScreencastRequestId: string | null = null let pendingBrowserScreencastRequestId: string | null = null @@ -561,6 +567,7 @@ export function connect( } ids.add(result.streamId) terminalStreamListeners.set(result.streamId, stream.listener) + terminalInputStreams.remember(stream.params, result.streamId) } if (!stream.cancelled) { stream.listener(result) @@ -839,6 +846,7 @@ export function connect( } const terminalStreamIds = terminalStreamIdsByRequest.get(id) if (terminalStreamIds) { + terminalInputStreams.forget(stream?.params, terminalStreamIds) for (const streamId of terminalStreamIds) { terminalStreamListeners.delete(streamId) terminalSnapshots.delete(streamId) @@ -862,6 +870,7 @@ export function connect( if (!terminalStreamIds) { return } + terminalInputStreams.forget(streamListeners.get(id)?.params, terminalStreamIds) for (const streamId of terminalStreamIds) { terminalStreamListeners.delete(streamId) terminalSnapshots.delete(streamId) @@ -936,33 +945,21 @@ export function connect( } function sendEncrypted(request: unknown): boolean { - if (ws && ws.readyState === WebSocket.OPEN && sharedKey) { - const sendingWs = ws - try { - sendingWs.send(encrypt(JSON.stringify(request), sharedKey)) - return true - } catch { - if (ws === sendingWs) { - closeAndSynthesize(sendingWs) + return sendEncryptedJson({ + socket: ws, + sharedKey, + request, + state, + onWriteError: (socket) => { + if (ws === socket) { + closeAndSynthesize(socket) } - return false + }, + onDesync: (socket) => { + synthesizedCloses.remember(socket, authenticationGeneration) + handleSocketClosed(socket, { timedOut: false }) } - } - console.log('[net] sendEncrypted FAILED — channel not ready', { - hasWs: !!ws, - readyState: ws?.readyState, - hasKey: !!sharedKey, - state }) - // Why: RN can drop onclose, leaving state 'connected' over a dead socket; force reconnect or every send silently fails forever. - if (state === 'connected' && ws && ws.readyState !== WebSocket.OPEN) { - console.log('[net] sendEncrypted detected ws desync — forcing reconnect', { - readyState: ws.readyState - }) - synthesizedCloses.remember(ws, authenticationGeneration) - handleSocketClosed(ws, { timedOut: false }) - } - return false } function sendBrowserScreencastUnsubscribe(subscriptionId: string): void { @@ -1066,6 +1063,18 @@ export function connect( }) }, + sendTerminalInput: (terminal, text) => + sendLanTerminalInputFrame({ + registry: terminalInputStreams, + terminal, + text, + connected: state === 'connected', + socket: ws, + sharedKey, + isCurrentSocket: (socket) => ws === socket, + onWriteError: closeAndSynthesize + }), + subscribe( method: string, params: unknown, diff --git a/mobile/src/transport/stable-logical-rpc-client.test.ts b/mobile/src/transport/stable-logical-rpc-client.test.ts index 0ba7a1f2ea7..da663f46776 100644 --- a/mobile/src/transport/stable-logical-rpc-client.test.ts +++ b/mobile/src/transport/stable-logical-rpc-client.test.ts @@ -18,6 +18,7 @@ class FakeSession implements RpcClient { vi.fn() readonly notifyForeground = vi.fn() readonly close = vi.fn() + sendTerminalInput = vi.fn(() => 'no-stream' as const) private state: ConnectionState private readonly stateListeners = new Set<(state: ConnectionState) => void>() private readonly streamListeners = new Set<(result: unknown) => void>() @@ -67,6 +68,15 @@ function deferred() { } describe('stable logical RPC client', () => { + it('forwards live terminal input to the active physical session', () => { + const session = new FakeSession('connected') + session.sendTerminalInput = vi.fn(() => 'sent') + const client = createStableLogicalRpcClient(session, 'lan') + + expect(client.sendTerminalInput('term-1', 'ab')).toBe('sent') + expect(session.sendTerminalInput).toHaveBeenCalledWith('term-1', 'ab') + }) + it('advertises source-default support on worktree catalog requests', async () => { const session = new FakeSession('connected') session.sendRequest.mockResolvedValue(success([])) diff --git a/mobile/src/transport/stable-logical-rpc-client.ts b/mobile/src/transport/stable-logical-rpc-client.ts index d1f1701aed9..20f425b67f4 100644 --- a/mobile/src/transport/stable-logical-rpc-client.ts +++ b/mobile/src/transport/stable-logical-rpc-client.ts @@ -113,6 +113,10 @@ export function createStableLogicalRpcClient( }) }, + sendTerminalInput(terminal, text) { + return activeSession.sendTerminalInput?.(terminal, text) ?? 'no-stream' + }, + subscribe(method, params, listener, options) { if (closed) { return () => {} diff --git a/mobile/src/transport/terminal-stream-protocol.ts b/mobile/src/transport/terminal-stream-protocol.ts index 384d66996f1..409884f0c93 100644 --- a/mobile/src/transport/terminal-stream-protocol.ts +++ b/mobile/src/transport/terminal-stream-protocol.ts @@ -9,6 +9,7 @@ export enum TerminalStreamOpcode { SnapshotEnd = 4, Resized = 5, Error = 6, + Input = 7, Metadata = 12 } @@ -64,6 +65,10 @@ export function decodeTerminalStreamJson(payload: Uint8Array): T | null { } } +export function encodeTerminalStreamText(value: string): Uint8Array { + return new TextEncoder().encode(value) +} + export function decodeTerminalStreamText(payload: Uint8Array): string { return new TextDecoder().decode(payload) } @@ -76,6 +81,7 @@ function isTerminalStreamOpcode(value: number): value is TerminalStreamOpcode { value === TerminalStreamOpcode.SnapshotEnd || value === TerminalStreamOpcode.Resized || value === TerminalStreamOpcode.Error || + value === TerminalStreamOpcode.Input || value === TerminalStreamOpcode.Metadata ) }