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 ) }