diff --git a/mobile/app/h/[hostId]/session/[worktreeId].tsx b/mobile/app/h/[hostId]/session/[worktreeId].tsx index 4191788f030..73c9d4a180c 100644 --- a/mobile/app/h/[hostId]/session/[worktreeId].tsx +++ b/mobile/app/h/[hostId]/session/[worktreeId].tsx @@ -114,7 +114,7 @@ import { loadTerminalAccessoryLayout } from '../../../../src/terminal/terminal-accessory-layout' import { createTerminalLiveAccessoryInput } from '../../../../src/terminal/terminal-live-accessory-input' -import { getTerminalLiveAccessoryRawSendTarget } from '../../../../src/terminal/terminal-live-accessory-raw-send-target' +import { sendTerminalLiveAccessoryRawBytes } from '../../../../src/terminal/terminal-live-accessory-raw-send' import { clearTerminalLiveInputFocusTimer, focusTerminalLiveInputTarget, @@ -128,6 +128,10 @@ import { sendMobileTerminalQueryReply } from '../../../../src/terminal/mobile-te import { TERMINAL_QUERY_REPLY_INPUT_RUNTIME_CAPABILITY } from '../../../../../src/shared/protocol-version' import { useTerminalLiveInputCommit } from '../../../../src/terminal/use-terminal-live-input-commit' import { resolveMobileTerminalInputGate } from '../../../../src/terminal/terminal-input-connection-gate' +import { + buildTerminalSendParams, + TERMINAL_INPUT_SEND_OPTIONS +} from '../../../../src/terminal/terminal-send-request' import { getTerminalCommandKeyboardType, getTerminalLiveInputKeyboardType @@ -3002,15 +3006,17 @@ export default function SessionScreen() { setInput('') try { - await client.sendRequest('terminal.send', { - terminal: activeHandle, - text, - enter: true, - // Why: presence-lock take-floor; marks this phone active so multi-mobile contention resolves to the last actor. - ...(deviceTokenRef.current - ? { client: { id: deviceTokenRef.current, type: 'mobile' as const } } - : {}) - }) + // Why: fail now and restore the text — a send parked across a reconnect would execute long after the tap. + await client.sendRequest( + 'terminal.send', + buildTerminalSendParams({ + terminal: activeHandle, + text, + enter: true, + deviceToken: deviceTokenRef.current + }), + TERMINAL_INPUT_SEND_OPTIONS + ) } catch { setInput(text) } finally { @@ -3027,29 +3033,15 @@ export default function SessionScreen() { if (accessoryCommit.kind !== 'allow-raw') { return } - const currentClient = clientRef.current - // Why: async IME flushing can outlive the original terminal selection. - const rawSendTarget = getTerminalLiveAccessoryRawSendTarget({ + await sendTerminalLiveAccessoryRawBytes({ + client: clientRef.current, targetHandle, activeHandle: activeHandleRef.current, - activeSessionTabType: activeSessionTabTypeRef.current + activeSessionTabType: activeSessionTabTypeRef.current, + connState: connStateRef.current, + bytes: input.bytes, + deviceToken: deviceTokenRef.current }) - if (!currentClient || !rawSendTarget || connStateRef.current !== 'connected') { - return - } - await currentClient - .sendRequest('terminal.send', { - terminal: rawSendTarget, - text: input.bytes, - enter: false, - ...(deviceTokenRef.current - ? { client: { id: deviceTokenRef.current, type: 'mobile' as const } } - : {}) - }) - .then( - () => undefined, - () => undefined - ) } const sendLiveTerminalInput = useCallback( @@ -3073,15 +3065,19 @@ export default function SessionScreen() { ) { 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', { - terminal: handle, - text, - enter: false, - ...(deviceTokenRef.current - ? { client: { id: deviceTokenRef.current, type: 'mobile' as const } } - : {}) - }) + .sendRequest( + 'terminal.send', + buildTerminalSendParams({ + terminal: handle, + text, + enter: false, + deviceToken: deviceTokenRef.current + }), + TERMINAL_INPUT_SEND_OPTIONS + ) .then(isTerminalSendRpcAccepted, () => false) }, [showToast] @@ -3343,14 +3339,17 @@ export default function SessionScreen() { terminalGestureInputInFlightRef.current.add(handle) try { - await rpc.sendRequest('terminal.send', { - terminal: handle, - text: queued.bytes, - enter: false, - ...(deviceTokenRef.current - ? { client: { id: deviceTokenRef.current, type: 'mobile' as const } } - : {}) - }) + // Why: gesture arrows parked across a reconnect would move a TUI long after the swipe. + await rpc.sendRequest( + 'terminal.send', + buildTerminalSendParams({ + terminal: handle, + text: queued.bytes, + enter: false, + deviceToken: deviceTokenRef.current + }), + TERMINAL_INPUT_SEND_OPTIONS + ) } catch { // Transient failure } finally { @@ -3831,14 +3830,15 @@ export default function SessionScreen() { subscribeToTerminal(createdHandle) if (options?.initialPrompt?.trim()) { void client - .sendRequest('terminal.send', { - terminal: createdHandle, - text: options.initialPrompt, - enter: options.enter !== false, - ...(deviceTokenRef.current - ? { client: { id: deviceTokenRef.current, type: 'mobile' as const } } - : {}) - }) + .sendRequest( + 'terminal.send', + buildTerminalSendParams({ + terminal: createdHandle, + text: options.initialPrompt, + enter: options.enter !== false, + deviceToken: deviceTokenRef.current + }) + ) .then((sendResponse) => { if (!sendResponse.ok) { throw new Error( diff --git a/mobile/src/terminal/terminal-input-connection-gate.test.ts b/mobile/src/terminal/terminal-input-connection-gate.test.ts index 15c72d92f9c..1cfa89d2d4e 100644 --- a/mobile/src/terminal/terminal-input-connection-gate.test.ts +++ b/mobile/src/terminal/terminal-input-connection-gate.test.ts @@ -1,6 +1,7 @@ import { readFileSync } from 'node:fs' import { describe, expect, it } from 'vitest' import { resolveMobileTerminalInputGate } from './terminal-input-connection-gate' +import { buildTerminalSendParams, TERMINAL_INPUT_SEND_OPTIONS } from './terminal-send-request' const sessionRouteSource = readFileSync( new URL('../../app/h/[hostId]/session/[worktreeId].tsx', import.meta.url), @@ -112,4 +113,22 @@ describe('session route offline-compose wiring', () => { const hookCall = routeSlice('useTerminalLiveInputCommit({', 'setLiveInputCapture') expect(hookCall).toContain("connected: connState === 'connected'") }) + + it('keeps every keystroke-grade terminal send now-or-never so nothing replays after reconnect', () => { + // Live mirror, buffered send, and gesture arrows must all opt out of the + // connect wait — a parked send replays stale bytes into the PTY. Accessory + // keys get the same option inside terminal-live-accessory-raw-send.ts. + const optOuts = sessionRouteSource.match(/TERMINAL_INPUT_SEND_OPTIONS/g)?.length ?? 0 + expect(optOuts).toBe(4) + expect(TERMINAL_INPUT_SEND_OPTIONS).toEqual({ failWhenDisconnected: true }) + }) + + it('tags terminal sends with the device presence lock only when a token exists', () => { + expect( + buildTerminalSendParams({ terminal: 't1', text: 'ls', enter: true, deviceToken: 'tok' }) + ).toEqual({ terminal: 't1', text: 'ls', enter: true, client: { id: 'tok', type: 'mobile' } }) + expect( + buildTerminalSendParams({ terminal: 't1', text: 'ls', enter: false, deviceToken: null }) + ).toEqual({ terminal: 't1', text: 'ls', enter: 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 new file mode 100644 index 00000000000..7a1110b407b --- /dev/null +++ b/mobile/src/terminal/terminal-live-accessory-raw-send.test.ts @@ -0,0 +1,56 @@ +import { describe, expect, it, vi } from 'vitest' +import { sendTerminalLiveAccessoryRawBytes } from './terminal-live-accessory-raw-send' +import type { RpcClient } from '../transport/rpc-client' + +function captureClient(result: Promise = Promise.resolve({ ok: true })) { + const sendRequest = vi.fn(() => result) + return { client: { sendRequest } as unknown as Pick, sendRequest } +} + +const BASE_ARGS = { + targetHandle: 'terminal-a', + activeHandle: 'terminal-a', + activeSessionTabType: 'terminal', + connState: 'connected', + bytes: '', + deviceToken: 'tok' +} as const + +describe('terminal live accessory raw send', () => { + it('sends raw bytes now-or-never with the device presence tag', async () => { + const { client, sendRequest } = captureClient() + + await sendTerminalLiveAccessoryRawBytes({ ...BASE_ARGS, client }) + + expect(sendRequest).toHaveBeenCalledWith( + 'terminal.send', + { terminal: 'terminal-a', text: '', enter: false, client: { id: 'tok', type: 'mobile' } }, + // Why: an accessory key parked in the connect wait would fire into the PTY long after the tap. + { failWhenDisconnected: true } + ) + }) + + it('drops the bytes instead of sending while disconnected', async () => { + const { client, sendRequest } = captureClient() + + await sendTerminalLiveAccessoryRawBytes({ ...BASE_ARGS, client, connState: 'reconnecting' }) + + expect(sendRequest).not.toHaveBeenCalled() + }) + + it('drops the bytes when the terminal selection went stale mid-flush', async () => { + const { client, sendRequest } = captureClient() + + await sendTerminalLiveAccessoryRawBytes({ ...BASE_ARGS, client, activeHandle: 'terminal-b' }) + + expect(sendRequest).not.toHaveBeenCalled() + }) + + it('swallows a rejected send so accessory taps never surface transport errors', async () => { + const { client } = captureClient(Promise.reject(new Error('Not connected: terminal.send'))) + + await expect( + sendTerminalLiveAccessoryRawBytes({ ...BASE_ARGS, client }) + ).resolves.toBeUndefined() + }) +}) diff --git a/mobile/src/terminal/terminal-live-accessory-raw-send.ts b/mobile/src/terminal/terminal-live-accessory-raw-send.ts new file mode 100644 index 00000000000..792491f6c63 --- /dev/null +++ b/mobile/src/terminal/terminal-live-accessory-raw-send.ts @@ -0,0 +1,43 @@ +import { getTerminalLiveAccessoryRawSendTarget } from './terminal-live-accessory-raw-send-target' +import { buildTerminalSendParams, TERMINAL_INPUT_SEND_OPTIONS } from './terminal-send-request' +import type { RpcClient } from '../transport/rpc-client' +import type { ConnectionState } from '../transport/types' + +type TerminalLiveAccessoryRawSendArgs = { + readonly client: Pick | null + readonly targetHandle: string + readonly activeHandle: string | null + readonly activeSessionTabType: string | null + readonly connState: ConnectionState + readonly bytes: string + readonly deviceToken: string | null +} + +export async function sendTerminalLiveAccessoryRawBytes( + args: TerminalLiveAccessoryRawSendArgs +): Promise { + // Why: async IME flushing can outlive the original terminal selection. + const rawSendTarget = getTerminalLiveAccessoryRawSendTarget({ + targetHandle: args.targetHandle, + activeHandle: args.activeHandle, + activeSessionTabType: args.activeSessionTabType + }) + if (!args.client || !rawSendTarget || args.connState !== 'connected') { + return + } + await args.client + .sendRequest( + 'terminal.send', + buildTerminalSendParams({ + terminal: rawSendTarget, + text: args.bytes, + enter: false, + deviceToken: args.deviceToken + }), + TERMINAL_INPUT_SEND_OPTIONS + ) + .then( + () => undefined, + () => undefined + ) +} diff --git a/mobile/src/terminal/terminal-send-request.ts b/mobile/src/terminal/terminal-send-request.ts new file mode 100644 index 00000000000..215c2837fe7 --- /dev/null +++ b/mobile/src/terminal/terminal-send-request.ts @@ -0,0 +1,28 @@ +import type { SendRequestOptions } from '../transport/rpc-client' + +type TerminalSendParams = { + readonly terminal: string + readonly text: string + readonly enter: boolean + readonly client?: { readonly id: string; readonly type: 'mobile' } +} + +// Why: keystroke-grade sends must never park in the connect wait — a send parked +// across a reconnect replays stale bytes into the PTY long after they were typed +// (#6713's `YZZYecho …` corruption of the first post-recovery command). +export const TERMINAL_INPUT_SEND_OPTIONS: SendRequestOptions = { failWhenDisconnected: true } + +export function buildTerminalSendParams(args: { + terminal: string + text: string + enter: boolean + // Why: presence-lock take-floor; marks this phone active so multi-mobile contention resolves to the last actor. + deviceToken: string | null +}): TerminalSendParams { + return { + terminal: args.terminal, + text: args.text, + enter: args.enter, + ...(args.deviceToken ? { client: { id: args.deviceToken, type: 'mobile' as const } } : {}) + } +} diff --git a/mobile/src/transport/rpc-client-connect-wait-replay.test.ts b/mobile/src/transport/rpc-client-connect-wait-replay.test.ts new file mode 100644 index 00000000000..24015ce829e --- /dev/null +++ b/mobile/src/transport/rpc-client-connect-wait-replay.test.ts @@ -0,0 +1,143 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { connect } from './rpc-client' + +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:/, ''), + decryptBytes: (bytes: Uint8Array) => bytes +})) + +class MockWebSocket { + static CONNECTING = 0 + static OPEN = 1 + static CLOSED = 3 + + readonly CONNECTING = MockWebSocket.CONNECTING + readonly OPEN = MockWebSocket.OPEN + readonly CLOSED = MockWebSocket.CLOSED + + readyState = MockWebSocket.CONNECTING + onopen: (() => void) | null = null + onclose: (() => void) | null = null + onmessage: ((event: { data: unknown }) => void) | null = null + sent: string[] = [] + close = vi.fn(() => { + if (this.readyState === MockWebSocket.CLOSED) { + return + } + this.readyState = MockWebSocket.CLOSED + this.onclose?.() + }) + + constructor(readonly endpoint: string) { + mockSockets.push(this) + } + + send(payload: string): void { + this.sent.push(payload) + } + + open(): void { + this.readyState = MockWebSocket.OPEN + this.onopen?.() + this.receive(JSON.stringify({ type: 'e2ee_ready' })) + this.receive('encrypted:{"type":"e2ee_authenticated"}') + } + + receive(payload: unknown): void { + this.onmessage?.({ data: payload }) + } +} + +const mockSockets: MockWebSocket[] = [] +const originalWebSocket = globalThis.WebSocket + +function track(request: Promise): { read: () => string } { + let outcome = 'pending' + request.then( + () => { + outcome = 'resolved' + }, + (error: Error) => { + outcome = error.message + } + ) + return { read: () => outcome } +} + +function terminalSendFrames(socket: MockWebSocket): string[] { + return socket.sent.filter((frame) => frame.includes('terminal.send')) +} + +describe('mobile rpc-client connect-wait replay', () => { + beforeEach(() => { + vi.useFakeTimers() + mockSockets.length = 0 + globalThis.WebSocket = MockWebSocket as unknown as typeof WebSocket + }) + + afterEach(() => { + vi.useRealTimers() + globalThis.WebSocket = originalWebSocket + }) + + it('Given a cut connection When a send opts into failWhenDisconnected Then it rejects now and nothing replays after reconnect', async () => { + const client = connect('ws://desktop.invalid', 'token', 'server-key') + const socket = mockSockets[0]! + socket.open() + socket.close() + + const request = client.sendRequest( + 'terminal.send', + { terminal: 't1', text: 'YZZY' }, + { failWhenDisconnected: true } + ) + const outcome = track(request) + + try { + await vi.advanceTimersByTimeAsync(0) + expect(outcome.read()).toBe('Not connected: terminal.send') + + // Reconnect; the rejected keystroke must not ride the new socket. + await vi.advanceTimersByTimeAsync(500) + mockSockets[1]!.open() + await vi.advanceTimersByTimeAsync(0) + expect(terminalSendFrames(mockSockets[1]!)).toEqual([]) + } finally { + client.close() + await request.catch(() => undefined) + } + }) + + it('Given a cut connection When a caller does not opt in Then the send parks and is delivered on the next socket', async () => { + const client = connect('ws://desktop.invalid', 'token', 'server-key') + const socket = mockSockets[0]! + socket.open() + socket.close() + + // Pins the default behavior that motivates the opt-out: parked requests + // replay after reconnect, which is what corrupted post-recovery input. + const request = client.sendRequest('terminal.send', { terminal: 't1', text: 'YZZY' }) + const outcome = track(request) + + try { + await vi.advanceTimersByTimeAsync(0) + expect(outcome.read()).toBe('pending') + + await vi.advanceTimersByTimeAsync(500) + mockSockets[1]!.open() + await vi.advanceTimersByTimeAsync(0) + expect(terminalSendFrames(mockSockets[1]!)).toHaveLength(1) + } finally { + client.close() + await request.catch(() => undefined) + } + }) +}) diff --git a/mobile/src/transport/rpc-client.ts b/mobile/src/transport/rpc-client.ts index 62ead1851ab..421d9c64d80 100644 --- a/mobile/src/transport/rpc-client.ts +++ b/mobile/src/transport/rpc-client.ts @@ -54,6 +54,11 @@ export type SendRequestOptions = { * against the post-connect clock, and squeezing them to the floor after a slow * reconnect would fail sends that used to land. */ budgetSpansConnect?: boolean + /** Reject immediately when not connected instead of parking in the connect + * wait. Keystroke-grade terminal input needs it: a send parked across a + * reconnect replays stale bytes into the PTY long after they were typed + * (observed as `YZZYecho …` corrupting the first post-recovery command). */ + failWhenDisconnected?: boolean } type SubscribeOptions = { @@ -999,6 +1004,9 @@ export function connect( const budget = openRpcRequestBudget(options) const waitStart = budget.startedAt const wasConnected = state === 'connected' + if (options?.failWhenDisconnected && !wasConnected) { + throw new Error(`Not connected: ${method}`) + } await waitForConnected(options?.timeoutMs) if (!wasConnected) { console.log('[net] sendRequest waited for connect', {