fix(mobile): stop stalled terminal input replaying into the PTY after reconnect

Device verification of the first commit surfaced the real replay vector for
the second defect: sendRequest parks in waitForConnected while disconnected,
so live-mirror deltas queued behind a dying send drain into the connect wait
and fire on the next socket — bytes typed during an outage executed tens of
seconds later (observed on device as the prompt reading `nOPQ` after
reconnect with no post-recovery typing).

Add SendRequestOptions.failWhenDisconnected — reject now instead of parking —
and opt in every keystroke-grade terminal send: live mirror, accessory keys,
buffered command send, and gesture arrows. Deliberate command sends
(initialPrompt on terminal create) keep the connect wait.

terminal.send param construction moves to terminal-send-request.ts and the
accessory raw-send tail to terminal-live-accessory-raw-send.ts.

Re-verified on simulator through a blackhole cut-proxy: text typed during the
stall no longer replays, and the first post-recovery command executes verbatim.
This commit is contained in:
Brennan Benson
2026-07-29 19:24:05 -07:00
parent 8200b46da8
commit 27b6b72178
7 changed files with 351 additions and 54 deletions
+54 -54
View File
@@ -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(
@@ -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 })
})
})
@@ -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<unknown> = Promise.resolve({ ok: true })) {
const sendRequest = vi.fn(() => result)
return { client: { sendRequest } as unknown as Pick<RpcClient, 'sendRequest'>, 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()
})
})
@@ -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<RpcClient, 'sendRequest'> | 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<void> {
// 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
)
}
@@ -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 } } : {})
}
}
@@ -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<unknown>): { 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)
}
})
})
+8
View File
@@ -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', {