starred/orca
1
0
Fork 0
mirror of https://github.com/stablyai/orca.git synced 2026-09-29 08:03:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: pipeline receipted mobile input with explicit no-replay recovery

Browse Source
This commit is contained in:
Neil
2026-09-05 16:14:16 -07:00
parent a0ef01f11e
commit c8a98363a3
44 changed files with 2433 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mobile/src/session/MobileSessionCommandDock.tsx
+10 -1
View File
@@ -10,6 +10,7 @@ import {
} from 'lucide-react-native'
import { triggerMediumImpact } from '../platform/haptics'
import { MobileTerminalLiveInputBar } from './MobileTerminalLiveInputBar'
import { MobileTerminalInputRecovery } from './MobileTerminalInputRecovery'
import { useHardwareKeyboardTextInputFocus } from '../hardware-keyboard/use-hardware-keyboard-text-input-focus'
import { createTerminalLiveAccessoryInput } from '../terminal/terminal-live-accessory-input'
import { getTerminalCommandKeyboardType } from '../terminal/terminal-keyboard-type'
@@ -36,7 +37,7 @@ export function MobileSessionCommandDock({ controller }: { controller: MobileSes
liveInputRef,
commandInputRef,
activeSessionTab,
canSend,
canSend: connectionCanSend,
canCompose,
liveInputEnabled,
showNativeChat,
@@ -60,6 +61,7 @@ export function MobileSessionCommandDock({ controller }: { controller: MobileSes
activeBrowserTab,
keyboardLift
} = controller
const canSend = connectionCanSend && !controller.terminalInputFailure
const hardwareInputFocus = useHardwareKeyboardTextInputFocus({
enabled:
!activeMarkdownTab &&
@@ -84,6 +86,13 @@ export function MobileSessionCommandDock({ controller }: { controller: MobileSes
}
]}
>
{controller.terminalInputFailure && (
<MobileTerminalInputRecovery
failure={controller.terminalInputFailure}
onRecover={controller.recoverTerminalInput}
recoveryUnavailable={controller.terminalInputRecoveryUnavailable}
/>
)}
{/* Accessory keys */}
<View style={styles.accessoryBar}>
{/* Why: fixed keyboard escape hatch; outside ScrollView + shortcut path so it can't scroll away or be hidden (#5106). */}
mobile/src/session/MobileTerminalInputRecovery.tsx
+41
View File
@@ -0,0 +1,41 @@
import { Pressable, StyleSheet, Text, View } from 'react-native'
import { colors, typography } from '../theme/mobile-theme'
import type { TerminalStreamInputFailure } from '../transport/terminal-stream-input-failure'
export function MobileTerminalInputRecovery({
failure,
onRecover,
recoveryUnavailable
}: {
failure: TerminalStreamInputFailure
onRecover: () => void
recoveryUnavailable: boolean
}) {
return (
<View style={styles.container} accessibilityLiveRegion="polite">
<Text style={styles.title}>Terminal input paused</Text>
<Text style={styles.detail}>
{failure.outcome === 'unknown'
? 'Some input may have reached the terminal. Check its contents before continuing.'
: 'The host did not accept the input. Check the terminal before continuing.'}{' '}
Input will not be replayed.
</Text>
{recoveryUnavailable && (
<Text style={styles.detail}>
Recovery needs a new connection to a host that supports ordered input. Update the host if
needed.
</Text>
)}
<Pressable accessibilityRole="button" onPress={onRecover} style={styles.button}>
<Text style={styles.title}>Reconnect input</Text>
</Pressable>
</View>
)
}
const styles = StyleSheet.create({
container: { padding: 12, gap: 8, backgroundColor: colors.terminalBg },
title: { color: colors.textPrimary, fontSize: typography.metaSize, fontWeight: '600' },
detail: { color: colors.textSecondary, fontSize: typography.metaSize },
button: { alignSelf: 'flex-start', minHeight: 36, justifyContent: 'center' }
})
mobile/src/session/MobileTerminalLiveInputBar.tsx
+2 -1
View File
@@ -26,7 +26,7 @@ export function MobileTerminalLiveInputBar({
)
const {
handleLiveInputHardwareKey,
canSend,
canSend: connectionCanSend,
focusLiveInput,
dictation,
isAttaching,
@@ -46,6 +46,7 @@ export function MobileTerminalLiveInputBar({
getLiveInteractionGeneration,
dismissKeyboardAfterAgentSend
} = controller
const canSend = connectionCanSend && !controller.terminalInputFailure
return (
<HardwareKeyboardCaptureView
style={[styles.inputBar, styles.liveInputBar]}
mobile/src/session/agent-send-keyboard-dismissal-wiring.test.ts
+1 -1
View File
@@ -189,7 +189,7 @@ describe('terminal send keyboard dismissal wiring', () => {
const slice = sourceSlice(
sendActionsSource,
'async function handleAccessoryKey(',
'const sendLiveTerminalInput = useCallback('
'const sendLiveTerminalInput = useCallback<TerminalLiveInputSender>('
)
expect(slice).not.toContain('dismissKeyboardAfterAgentSend')
})
mobile/src/session/mobile-session-route-parity.test.ts
+8 -8
View File
@@ -63,15 +63,15 @@ const HOST_COMPONENT_NAMES = new Set([
'View'
])
const HEAD_MAIN_HOOK_SHA256 = 'ae0325005289a8df7a479aa50b4d65dc3f70234bc0bce7ef4dfd923031944178'
const HEAD_HOOK_BINDING_SHA256 = 'a6c0e22ca841301a97e3cf8f3a9f7ba2a90f848bc7b430845d3df2ae79fea3bf'
const HEAD_MAIN_HOOK_SHA256 = '8d30f0d8f02761c52690ad162d61a47befa35c2bbb61202e802a4cee4ab38a42'
const HEAD_HOOK_BINDING_SHA256 = 'de164c375eb0317d8e3ac7b76ff636ea8bc12949297feed5c235c31e21349525'
const HEAD_CALLBACK_IDENTITY_SHA256 =
'2a9e4825df007f6ef53b81aa5004991d6318eee7507b44d625c07e630be432eb'
const HEAD_CALLBACK_BODY_SHA256 = '22103ba85a86e3a3fcb80a7509c7a455d79863010cde3af02db6565b55e3ebe9'
'4324a33cfe4081380515ca1afddf50b1858fd34f72833cba426829edead92d57'
const HEAD_CALLBACK_BODY_SHA256 = '15a2e663ddd70ea79677ab8486cf3753b38c349ab83150dd35dae4ed2d32bf0f'
const HEAD_EFFECT_SHA256 = 'd9ebfaabc1e79773cdada7ab370b20459ed972f1f8edce1652199f4d0391cd13'
const HEAD_CONTENT_HOOK_SHA256 = '9c3b612fef3f370d66873aefdbe1d701f20cb64ded31fef5cc45fde6f8189581'
const HEAD_NESTED_FUNCTION_SHA256 =
'6a13919ede2a8033436fb03e0ff7c426fbed97f470875a7b21b00aaada17fb73'
'0720d9ee9a848386db780b3ec198635fabd3ca33d376a8838dba0b9d3d748926'
const HEAD_NATIVE_REGISTRATION_SHA256 =
'cab85e4e4a3f43289ba93ddea9ccce57aea83e0bf14fd1620a965aad0c1cb49e'
const HEAD_NATIVE_REMOVAL_SHA256 =
@@ -82,7 +82,7 @@ const HEAD_TIMER_CLEANUP_SHA256 = 'c73f1d1c2cc89642f3d727d6f3b6b81860a9d6f342345
const HEAD_RUNTIME_STRING_SHA256 =
'951ee0022e65cc1b15dfbaf11c78841840059e590c6df6368c56d70d30cf4015'
const HEAD_HOST_JSX_SHA256 = '6cc35b6fe4de4abf3948b4ef24924b1992301bb7c592e1e3de8c1323804ab163'
const HEAD_LEAF_JSX_SHA256 = 'c94d687d5cb1aa02467863904b1d79c678ab7081521a6586f7b22a6c1f669ab6'
const HEAD_LEAF_JSX_SHA256 = 'c3815cc0fcc5a7150d0c6e171684583e3287f2af1e83f83b1600c7e9810a7982'
const HEAD_STYLE_REFERENCE_SHA256 =
'7cbf94b352fe9cf5a6cf30e72a4fa9d8f1bb768684563ab14a44cb3a96cab4cb'
const HEAD_IDENTITY_FIELD_SHA256 =
@@ -473,7 +473,7 @@ describe('mobile session route extraction parity', () => {
const contentBindings = CONTENT_COMPONENT_NAMES.flatMap(
(name) => readHookFacts(name, definitions).bindings
)
expect(main.hooks).toHaveLength(267)
expect(main.hooks).toHaveLength(269)
expect(hash(main.hooks)).toBe(HEAD_MAIN_HOOK_SHA256)
expect(hash(main.bindings)).toBe(HEAD_HOOK_BINDING_SHA256)
expect(main.callbacks).toHaveLength(77)
@@ -523,7 +523,7 @@ describe('mobile session route extraction parity', () => {
const jsx = readJsxFacts(readDefinitions())
expect(jsx.host).toHaveLength(123)
expect(hash(jsx.host)).toBe(HEAD_HOST_JSX_SHA256)
expect(jsx.leaf).toHaveLength(62)
expect(jsx.leaf).toHaveLength(63)
expect(hash(jsx.leaf)).toBe(HEAD_LEAF_JSX_SHA256)
expect(jsx.styleReferences).toHaveLength(172)
expect(hash(jsx.styleReferences)).toBe(HEAD_STYLE_REFERENCE_SHA256)
mobile/src/session/use-mobile-session-terminal-runtime.ts
+2
View File
@@ -58,6 +58,7 @@ export function useMobileSessionTerminalRuntime(scope: MobileSessionScreenStateM
const commandInputRef = useRef<TextInput>(null)
const liveInputFocusTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null)
const sendLiveTerminalInputRef = useRef<TerminalLiveInputSender>(async () => false)
const terminalInputSubscribedRef = useRef<(handle: string) => void>(() => {})
const sessionTabActionSheetKeyboardHideSubRef = useRef<ReturnType<
typeof Keyboard.addListener
> | null>(null)
@@ -172,6 +173,7 @@ export function useMobileSessionTerminalRuntime(scope: MobileSessionScreenStateM
liveInputFocusTimerRef,
sendLiveTerminalInputRef,
sessionTabActionSheetKeyboardHideSubRef,
terminalInputSubscribedRef,
sessionTabActionSheetRequestSeqRef,
dictationRouteContextRef,
terminalUnsubsRef,
mobile/src/session/use-mobile-session-terminal-send-actions.ts
+44 -15
View File
@@ -15,6 +15,8 @@ import {
} from '../terminal/terminal-send-request'
import { normalizeTerminalTextInput } from '../terminal/terminal-text-input-normalization'
import { useAgentSendKeyboardDismissal } from './use-agent-send-keyboard-dismissal'
import { useTerminalInputRecovery } from './use-terminal-input-recovery'
import type { TerminalLiveInputSender } from '../terminal/terminal-live-input-sender'
import type { MobileSessionTab } from './mobile-session-route-types'
import { useMobileSessionTabActionSheetOpener } from './use-mobile-session-tab-action-targets'
import type { MobileSessionTerminalWebviewModel } from './use-mobile-session-terminal-webview'
@@ -50,6 +52,7 @@ export function useMobileSessionTerminalSendActions(scope: MobileSessionTerminal
showToast
} = scope
const TERMINAL_KEYBOARD_DISMISS_ACTION_SHEET_FALLBACK_MS = 450
const inputRecovery = useTerminalInputRecovery(scope)
const dismissSoftwareKeyboard = useCallback(() => {
dismissTerminalKeyboard({
@@ -121,7 +124,11 @@ export function useMobileSessionTerminalSendActions(scope: MobileSessionTerminal
if (accessoryCommit.kind !== 'allow-raw') {
return
}
await sendTerminalLiveAccessoryRawBytes({
const rpc = clientRef.current
const reportFailure = rpc
? inputRecovery.captureTerminalInputFailureReporter(targetHandle, rpc)
: null
const accepted = await sendTerminalLiveAccessoryRawBytes({
client: clientRef.current,
targetHandle,
activeHandle: activeHandleRef.current,
@@ -130,9 +137,12 @@ export function useMobileSessionTerminalSendActions(scope: MobileSessionTerminal
bytes: input.bytes,
deviceToken: deviceTokenRef.current
})
if (!accepted && rpc) {
reportFailure?.()
}
}
const sendLiveTerminalInput = useCallback(
const sendLiveTerminalInput = useCallback<TerminalLiveInputSender>(
async (handle: string, bytes: string): Promise<boolean> => {
const text = normalizeTerminalTextInput(bytes)
if (text.length === 0) {
@@ -155,21 +165,39 @@ export function useMobileSessionTerminalSendActions(scope: MobileSessionTerminal
}
// 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)
const reportFailure = inputRecovery.captureTerminalInputFailureReporter(handle, rpc)
const streamSend = rpc.sendTerminalStreamInput?.(handle, text) ?? null
const accepted = await (
streamSend ??
rpc
.sendRequest(
'terminal.send',
buildTerminalSendParams({
terminal: handle,
text,
enter: false,
deviceToken: deviceTokenRef.current
}),
TERMINAL_INPUT_SEND_OPTIONS
)
.then(isTerminalSendRpcAccepted, () => false)
).catch(() => false)
if (!accepted) {
reportFailure()
}
return accepted
},
[showToast]
[showToast, inputRecovery.captureTerminalInputFailureReporter]
)
sendLiveTerminalInput.cancelPending = (handle: string) => {
const rpc = clientRef.current
rpc?.cancelTerminalStreamInput?.(handle)
if (rpc) {
inputRecovery.reportTerminalInputFailure(handle, rpc)
}
}
sendLiveTerminalInput.supportsPipeline = (handle: string) =>
clientRef.current?.supportsTerminalStreamInput?.(handle) ?? false
sendLiveTerminalInputRef.current = sendLiveTerminalInput
const clearSessionTabActionSheetKeyboardListener = useCallback(() => {
@@ -228,6 +256,7 @@ export function useMobileSessionTerminalSendActions(scope: MobileSessionTerminal
)
return {
...inputRecovery,
handleSend,
handleAccessoryKey,
sendLiveTerminalInput,
mobile/src/session/use-mobile-session-terminal-subscription.ts
+2
View File
@@ -34,6 +34,7 @@ export function useMobileSessionTerminalSubscription(
scheduleDelayedAction,
showToast,
markNativeChatInputLeaseReady,
terminalInputSubscribedRef,
showNativeChatRef,
getTerminalRef,
unsubscribeTerminal,
@@ -113,6 +114,7 @@ export function useMobileSessionTerminalSubscription(
}
if (data.type === 'subscribed') {
markNativeChatInputLeaseReady(handle)
terminalInputSubscribedRef.current(handle)
return
}
// Why: keep the subscription as the input-floor lease but don't mutate covered xterm state; return-to-terminal resubscribes.
mobile/src/session/use-terminal-input-recovery.test.ts
+173
View File
@@ -0,0 +1,173 @@
import { createElement } from 'react'
import { act, create } from 'react-test-renderer'
import { describe, expect, it, vi } from 'vitest'
import type { RpcClient } from '../transport/rpc-client'
import type { TerminalStreamInputFailure } from '../transport/terminal-stream-input-failure'
import { useTerminalInputRecovery } from './use-terminal-input-recovery'
function harness() {
let detail: TerminalStreamInputFailure | null = { outcome: 'unknown', reason: 'timeout' }
const recover = vi.fn(() => {
detail = null
return true
})
const client = {
getTerminalStreamInputFailure: () => detail,
cancelTerminalStreamInput: vi.fn(),
recoverTerminalStreamInput: recover
} as unknown as RpcClient
const options = {
client,
getSendCompletionGeneration: vi.fn(() => 1),
getLiveInteractionGeneration: vi.fn(() => 1),
activeHandle: 'a',
activeHandleRef: { current: 'a' },
clientRef: { current: client },
clearPendingLiveInputCommit: vi.fn(),
unsubscribeTerminal: vi.fn(),
subscribeToTerminal: vi.fn(),
terminalInputSubscribedRef: { current: (_handle: string) => {} }
}
let value!: ReturnType<typeof useTerminalInputRecovery>
function Probe() {
value = useTerminalInputRecovery(options)
return null
}
let renderer!: ReturnType<typeof create>
act(() => {
renderer = create(createElement(Probe))
})
return {
options,
client,
recover,
get: () => value,
render: () => act(() => renderer.update(createElement(Probe))),
close: () => act(() => renderer.unmount())
}
}
describe('terminal input explicit recovery', () => {
it('does not turn a late pre-recovery failure into a new legacy warning', () => {
const h = harness()
try {
const lateFailure = h.get().captureTerminalInputFailureReporter('a', h.client)
act(() => h.get().recoverTerminalInput())
act(() => h.options.terminalInputSubscribedRef.current('a'))
act(() => lateFailure())
expect(h.get().terminalInputFailure).toBeNull()
} finally {
h.close()
}
})
it('requires a user action and fresh subscription, and never replays text', () => {
const h = harness()
try {
act(() => h.options.terminalInputSubscribedRef.current('a'))
expect(h.recover).not.toHaveBeenCalled()
act(() => h.get().recoverTerminalInput())
expect(h.options.clearPendingLiveInputCommit).toHaveBeenCalledTimes(1)
expect(h.client.cancelTerminalStreamInput).toHaveBeenCalledWith('a')
expect(h.options.unsubscribeTerminal).toHaveBeenCalledWith('a')
expect(h.options.subscribeToTerminal).toHaveBeenCalledWith('a')
expect(h.get().terminalInputFailure?.outcome).toBe('unknown')
act(() => h.options.terminalInputSubscribedRef.current('a'))
expect(h.recover).toHaveBeenCalledWith('a')
expect(h.get().terminalInputFailure).toBeNull()
} finally {
h.close()
}
})
it('does not recover a new route from a late subscription event', () => {
const h = harness()
try {
act(() => h.get().recoverTerminalInput())
h.options.activeHandleRef.current = 'b'
act(() => h.options.terminalInputSubscribedRef.current('a'))
expect(h.recover).not.toHaveBeenCalled()
} finally {
h.close()
}
})
it('keeps the warning if the replacement subscription cannot recover', () => {
const h = harness()
try {
h.recover.mockReturnValue(false)
act(() => h.get().recoverTerminalInput())
act(() => h.options.terminalInputSubscribedRef.current('a'))
expect(h.get().terminalInputFailure?.outcome).toBe('unknown')
expect(h.get().terminalInputRecoveryUnavailable).toBe(true)
} finally {
h.close()
}
})
it('shows an old-host JSON failure and resumes only after explicit resubscription', () => {
const h = harness()
try {
h.client.getTerminalStreamInputFailure = () => null
h.recover.mockReturnValue(false)
act(() => h.get().reportTerminalInputFailure('a', h.client, true))
expect(h.get().terminalInputFailure?.reason).toBe('legacy_send_failed')
act(() => h.options.terminalInputSubscribedRef.current('a'))
expect(h.get().terminalInputFailure).not.toBeNull()
act(() => h.get().recoverTerminalInput())
act(() => h.options.terminalInputSubscribedRef.current('a'))
expect(h.get().terminalInputFailure).toBeNull()
} finally {
h.close()
}
})
it('a legacy warning cannot clear a later negotiated uncertain-prefix fence', () => {
const h = harness()
try {
h.client.getTerminalStreamInputFailure = () => null
h.recover.mockReturnValue(false)
act(() => h.get().reportTerminalInputFailure('a', h.client, true))
h.client.getTerminalStreamInputFailure = () => ({
outcome: 'unknown',
reason: 'receipt_timeout'
})
act(() => h.get().recoverTerminalInput())
act(() => h.options.terminalInputSubscribedRef.current('a'))
expect(h.get().terminalInputFailure?.reason).toBe('receipt_timeout')
expect(h.get().terminalInputRecoveryUnavailable).toBe(true)
} finally {
h.close()
}
})
it('invalidates recovery across A to B to A even when the late event arrives back on A', () => {
const h = harness()
try {
act(() => h.get().recoverTerminalInput())
h.options.activeHandle = h.options.activeHandleRef.current = 'b'
h.render()
h.options.activeHandle = h.options.activeHandleRef.current = 'a'
h.render()
act(() => h.options.terminalInputSubscribedRef.current('a'))
expect(h.recover).not.toHaveBeenCalled()
} finally {
h.close()
}
})
it.each(['getSendCompletionGeneration', 'getLiveInteractionGeneration'] as const)(
'does not clear newer input after %s changes',
(field) => {
const h = harness()
try {
act(() => h.get().recoverTerminalInput())
h.options[field].mockReturnValue(2)
act(() => h.options.terminalInputSubscribedRef.current('a'))
expect(h.recover).not.toHaveBeenCalled()
expect(h.options.clearPendingLiveInputCommit).toHaveBeenCalledTimes(1)
} finally {
h.close()
}
}
)
})
mobile/src/session/use-terminal-input-recovery.ts
+157
View File
@@ -0,0 +1,157 @@
import { useCallback, useLayoutEffect, useRef, useReducer, useState, type RefObject } from 'react'
import type { RpcClient } from '../transport/rpc-client'
import type { TerminalStreamInputFailure } from '../transport/terminal-stream-input-failure'
type RecoveryOptions = {
activeHandle: string | null
client: RpcClient | null
getSendCompletionGeneration: () => number
getLiveInteractionGeneration: () => number
activeHandleRef: RefObject<string | null>
clientRef: RefObject<RpcClient | null>
clearPendingLiveInputCommit: () => void
unsubscribeTerminal: (handle: string) => void
subscribeToTerminal: (handle: string) => void
terminalInputSubscribedRef: RefObject<(handle: string) => void>
}
export function useTerminalInputRecovery(options: RecoveryOptions) {
const [, refreshFailure] = useReducer((revision: number) => revision + 1, 0)
const recoveryEpochRef = useRef(0)
const [legacyFailure, setLegacyFailure] = useState<{ handle: string; client: RpcClient } | null>(
null
)
const [unavailable, setUnavailable] = useState<{ handle: string; client: RpcClient } | null>(null)
const recoveryRef = useRef<{
handle: string
client: RpcClient
generation: number
interaction: number
} | null>(null)
useLayoutEffect(() => {
return () => {
recoveryRef.current = null
}
}, [options.activeHandle, options.client])
const reportTerminalInputFailure = useCallback(
(handle: string, client: RpcClient, legacySendFailed = false) => {
if (options.clientRef.current !== client || options.activeHandleRef.current !== handle) {
return
}
const detail = client.getTerminalStreamInputFailure?.(handle)
if (detail) {
refreshFailure()
} else if (legacySendFailed) {
setLegacyFailure({ handle, client })
}
},
[options.activeHandleRef, options.clientRef]
)
const recoverTerminalInput = useCallback(() => {
const handle = options.activeHandleRef.current
const client = options.clientRef.current
if (!handle || !client) {
return
}
recoveryEpochRef.current += 1
setUnavailable(null)
// Recovery never replays the optimistic field or assumes what reached the shell.
options.clearPendingLiveInputCommit()
client.cancelTerminalStreamInput?.(handle)
options.unsubscribeTerminal(handle)
recoveryRef.current = {
handle,
client,
generation: options.getSendCompletionGeneration(),
interaction: options.getLiveInteractionGeneration()
}
options.subscribeToTerminal(handle)
}, [
options.activeHandleRef,
options.clientRef,
options.clearPendingLiveInputCommit,
options.unsubscribeTerminal,
options.subscribeToTerminal,
options.getSendCompletionGeneration,
options.getLiveInteractionGeneration
])
const captureTerminalInputFailureReporter = useCallback(
(handle: string, client: RpcClient) => {
const epoch = recoveryEpochRef.current
const surface = options.getSendCompletionGeneration()
return () => {
if (
epoch === recoveryEpochRef.current &&
surface === options.getSendCompletionGeneration()
) {
reportTerminalInputFailure(handle, client, true)
}
}
},
[options.getSendCompletionGeneration, reportTerminalInputFailure]
)
const handleSubscribed = useCallback(
(handle: string) => {
const recovery = recoveryRef.current
if (
!recovery ||
recovery.handle !== handle ||
options.activeHandleRef.current !== handle ||
options.clientRef.current !== recovery.client ||
options.getSendCompletionGeneration() !== recovery.generation ||
options.getLiveInteractionGeneration() !== recovery.interaction
) {
recoveryRef.current = null
return
}
recoveryRef.current = null
const legacyOnly =
legacyFailure?.handle === handle &&
legacyFailure.client === recovery.client &&
!recovery.client.getTerminalStreamInputFailure?.(handle)
if (recovery.client.recoverTerminalStreamInput?.(handle) || legacyOnly) {
options.clearPendingLiveInputCommit()
setLegacyFailure(null)
refreshFailure()
} else {
setUnavailable(recovery)
reportTerminalInputFailure(handle, recovery.client)
}
},
[
options.activeHandleRef,
options.clientRef,
options.getSendCompletionGeneration,
options.getLiveInteractionGeneration,
options.clearPendingLiveInputCommit,
reportTerminalInputFailure,
legacyFailure
]
)
useLayoutEffect(() => {
options.terminalInputSubscribedRef.current = handleSubscribed
return () => {
options.terminalInputSubscribedRef.current = () => {}
}
}, [handleSubscribed, options.terminalInputSubscribedRef])
const legacyDetail: TerminalStreamInputFailure | null =
legacyFailure?.handle === options.activeHandle && legacyFailure?.client === options.client
? { outcome: 'unknown', reason: 'legacy_send_failed' }
: null
return {
terminalInputRecoveryUnavailable:
unavailable?.handle === options.activeHandle && unavailable?.client === options.client,
terminalInputFailure: options.activeHandle
? (options.clientRef.current?.getTerminalStreamInputFailure?.(options.activeHandle) ??
legacyDetail)
: null,
reportTerminalInputFailure,
captureTerminalInputFailureReporter,
recoverTerminalInput
}
}
mobile/src/terminal/terminal-live-accessory-raw-send.test.ts
+24
View File
@@ -25,6 +25,30 @@ const BASE_ARGS = {
} as const
describe('terminal live accessory raw send', () => {
it.each([true, false])('uses negotiated receipts without RPC fallback: %s', async (accepted) => {
const { client, sendRequest } = captureClient()
const sendTerminalStreamInput = vi.fn(() => Promise.resolve(accepted))
await expect(
sendTerminalLiveAccessoryRawBytes({
...BASE_ARGS,
client: { ...client, sendTerminalStreamInput }
})
).resolves.toBe(accepted)
expect(sendTerminalStreamInput).toHaveBeenCalledWith('terminal-a', '\u001b')
expect(sendRequest).not.toHaveBeenCalled()
})
it('falls back only before stream admission when negotiation is absent', async () => {
const { client, sendRequest } = captureClient()
await expect(
sendTerminalLiveAccessoryRawBytes({
...BASE_ARGS,
client: { ...client, sendTerminalStreamInput: () => null }
})
).resolves.toBe(true)
expect(sendRequest).toHaveBeenCalledTimes(1)
})
it('sends raw bytes now-or-never with the device presence tag', async () => {
const { client, sendRequest } = captureClient()
mobile/src/terminal/terminal-live-accessory-raw-send.ts
+17 -13
View File
@@ -5,7 +5,7 @@ import type { RpcClient } from '../transport/rpc-client'
import type { ConnectionState } from '../transport/types'
type TerminalLiveAccessoryRawSendArgs = {
readonly client: Pick<RpcClient, 'sendRequest'> | null
readonly client: Pick<RpcClient, 'sendRequest' | 'sendTerminalStreamInput'> | null
readonly targetHandle: string
readonly activeHandle: string | null
readonly activeSessionTabType: string | null
@@ -26,16 +26,20 @@ export async function sendTerminalLiveAccessoryRawBytes(
if (!args.client || !rawSendTarget || args.connState !== 'connected') {
return false
}
return args.client
.sendRequest(
'terminal.send',
buildTerminalSendParams({
terminal: rawSendTarget,
text: args.bytes,
enter: false,
deviceToken: args.deviceToken
}),
TERMINAL_INPUT_SEND_OPTIONS
)
.then(isTerminalSendRpcAccepted, () => false)
const streamSend = args.client.sendTerminalStreamInput?.(rawSendTarget, args.bytes) ?? null
return (
streamSend ??
args.client
.sendRequest(
'terminal.send',
buildTerminalSendParams({
terminal: rawSendTarget,
text: args.bytes,
enter: false,
deviceToken: args.deviceToken
}),
TERMINAL_INPUT_SEND_OPTIONS
)
.then(isTerminalSendRpcAccepted, () => false)
).catch(() => false)
}
mobile/src/terminal/terminal-live-input-sender.ts
+5 -1
View File
@@ -1 +1,5 @@
export type TerminalLiveInputSender = (handle: string, bytes: string) => Promise<boolean>
export type TerminalLiveInputSender = {
(handle: string, bytes: string): Promise<boolean>
cancelPending?: (handle: string) => void
supportsPipeline?: (handle: string) => boolean
}
mobile/src/terminal/terminal-live-pending-flush-state.test.ts
+6 -5
View File
@@ -98,7 +98,7 @@ describe('terminal live mirror send queue', () => {
expect(payloads).toEqual(['a', 'bc'])
})
it('Given a failed previous send When a mirror send queues Then it still runs in order', async () => {
it('Given a failed prefix When a mirror send queues Then dependent input is not sent', async () => {
// Given
const state = createTerminalLivePendingFlushState()
const order: string[] = []
@@ -115,11 +115,12 @@ describe('terminal live mirror send queue', () => {
// Then
await expect(first).resolves.toBe(false)
await expect(second).resolves.toBe(true)
expect(order).toEqual(['first', 'second'])
await expect(second).resolves.toBe(false)
expect(order).toEqual(['first'])
await expect(waitForTerminalLivePendingFlush(state)).resolves.toBe(false)
})
it('Given a throwing send When a mirror send queues Then the promise resolves false and the chain continues', async () => {
it('Given a throwing send When a mirror send queues Then both resolve false', async () => {
// Given
const state = createTerminalLivePendingFlushState()
const first = queueTerminalLiveMirrorSend(state, 'terminal-1', 'first', async () => {
@@ -131,7 +132,7 @@ describe('terminal live mirror send queue', () => {
// Then
await expect(first).resolves.toBe(false)
await expect(second).resolves.toBe(true)
await expect(second).resolves.toBe(false)
})
it('Given a settled mirror send When it was the newest Then the state resets to null', async () => {
mobile/src/terminal/terminal-live-pending-flush-state.ts
+121 -47
View File
@@ -1,3 +1,5 @@
import { TERMINAL_LIVE_INPUT_MAX_BYTES } from './terminal-live-input'
type TerminalLiveMirrorSender = (handle: string, payload: string) => Promise<boolean>
type TerminalLivePendingRequest = {
@@ -7,102 +9,174 @@ type TerminalLivePendingRequest = {
type TerminalLivePendingBatch = {
readonly handle: string
payload: string
bytes: number
readonly requests: TerminalLivePendingRequest[]
readonly sender: TerminalLiveMirrorSender
readonly pipeline: boolean
}
const MAX_PENDING_REQUESTS = 64
const MAX_PENDING_BYTES = 1024 * 1024
const encoder = new TextEncoder()
export type TerminalLivePendingFlushState = {
current: Promise<boolean> | null
activeRequests: TerminalLivePendingRequest[]
generation: number
pendingBatches: TerminalLivePendingBatch[]
activeBatches: Set<TerminalLivePendingBatch>
retainedBytes: number
requestCount: number
failed: boolean
finish: ((sent: boolean) => void) | null
}
export function createTerminalLivePendingFlushState(): TerminalLivePendingFlushState {
return {
current: null,
activeRequests: [],
generation: 0,
pendingBatches: []
pendingBatches: [],
activeBatches: new Set(),
retainedBytes: 0,
requestCount: 0,
failed: false,
finish: null
}
}
export function waitForTerminalLivePendingFlush(
state: TerminalLivePendingFlushState
): Promise<boolean> {
return state.current ?? Promise.resolve(true)
return state.failed ? Promise.resolve(false) : (state.current ?? Promise.resolve(true))
}
function releaseBatch(
state: TerminalLivePendingFlushState,
batch: TerminalLivePendingBatch,
sent: boolean
): void {
state.retainedBytes -= batch.bytes
state.requestCount -= batch.requests.length
batch.payload = ''
batch.requests.splice(0).forEach(({ resolve }) => resolve(sent))
}
function rejectPending(state: TerminalLivePendingFlushState): void {
for (const batch of state.pendingBatches.splice(0)) {
releaseBatch(state, batch, false)
}
}
export function cancelTerminalLivePendingFlush(state: TerminalLivePendingFlushState): void {
state.generation += 1
const requests = [
...state.activeRequests,
...state.pendingBatches.flatMap((batch) => batch.requests)
]
state.activeRequests = []
state.pendingBatches = []
state.current = null
requests.forEach(({ resolve }) => resolve(false))
}
async function drainTerminalLiveMirrorSends(
state: TerminalLivePendingFlushState,
generation: number
): Promise<boolean> {
let allSent = true
while (state.generation === generation) {
const batch = state.pendingBatches.shift()
if (!batch) {
state.current = null
return allSent
}
state.activeRequests = batch.requests
const sent = await batch.sender(batch.handle, batch.payload).catch(() => false)
if (state.generation !== generation) {
return false
}
state.activeRequests = []
batch.requests.forEach(({ resolve }) => resolve(sent))
allSent &&= sent
rejectPending(state)
for (const batch of state.activeBatches) {
releaseBatch(state, batch, false)
}
return false
state.activeBatches.clear()
state.finish?.(false)
state.finish = null
state.current = null
state.failed = false
}
// Mirror deltas are ordered PTY bytes; batching pending bytes avoids one RTT per keystroke.
function pumpMirrorSends(state: TerminalLivePendingFlushState): void {
if (state.failed) {
rejectPending(state)
}
while (state.pendingBatches.length > 0) {
const batch = state.pendingBatches[0]
if (
[...state.activeBatches].some(
(active) =>
!active.pipeline ||
!batch.pipeline ||
active.sender !== batch.sender ||
active.handle !== batch.handle
)
) {
return
}
state.pendingBatches.shift()
state.activeBatches.add(batch)
const generation = state.generation
const settle = (sent: boolean): void => {
if (state.generation !== generation) {
return
}
state.activeBatches.delete(batch)
releaseBatch(state, batch, sent)
state.failed ||= !sent
pumpMirrorSends(state)
}
try {
void batch.sender(batch.handle, batch.payload).then(settle, () => settle(false))
} catch {
settle(false)
}
batch.payload = ''
}
if (state.activeBatches.size === 0) {
state.finish?.(!state.failed)
state.finish = null
state.current = null
}
}
// Pipelining changes dispatch timing, never the meaning of an accepted completion.
export function queueTerminalLiveMirrorSend(
state: TerminalLivePendingFlushState,
handle: string,
payload: string,
sender: TerminalLiveMirrorSender
sender: TerminalLiveMirrorSender,
options: { pipeline?: boolean } = {}
): Promise<boolean> {
if (state.failed) {
return Promise.resolve(false)
}
const bytes =
payload.length > TERMINAL_LIVE_INPUT_MAX_BYTES ? Infinity : encoder.encode(payload).byteLength
if (
bytes > TERMINAL_LIVE_INPUT_MAX_BYTES ||
state.retainedBytes + bytes > MAX_PENDING_BYTES ||
state.requestCount >= MAX_PENDING_REQUESTS
) {
state.failed = true
rejectPending(state)
return Promise.resolve(false)
}
let resolveRequest: (sent: boolean) => void = () => {}
const request = new Promise<boolean>((resolve) => {
resolveRequest = resolve
})
state.retainedBytes += bytes
state.requestCount += 1
const pipeline = options.pipeline === true
const pendingTail = state.pendingBatches.at(-1)
if (pendingTail?.handle === handle && pendingTail.sender === sender) {
if (
pendingTail?.handle === handle &&
pendingTail.sender === sender &&
pendingTail.pipeline === pipeline &&
pendingTail.bytes + bytes <= TERMINAL_LIVE_INPUT_MAX_BYTES
) {
pendingTail.payload += payload
pendingTail.bytes += bytes
pendingTail.requests.push({ resolve: resolveRequest })
} else {
state.pendingBatches.push({
handle,
payload,
bytes,
requests: [{ resolve: resolveRequest }],
sender
sender,
pipeline
})
}
if (!state.current) {
const generation = state.generation
const drain = drainTerminalLiveMirrorSends(state, generation).catch(() => false)
state.current = drain
void drain.then(() => {
if (state.current === drain) {
state.current = null
}
state.current = new Promise<boolean>((resolve) => {
state.finish = resolve
})
}
pumpMirrorSends(state)
return request
}
mobile/src/terminal/terminal-live-pipelined-flush.test.ts
+173
View File
@@ -0,0 +1,173 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import {
cancelTerminalLivePendingFlush,
createTerminalLivePendingFlushState,
queueTerminalLiveMirrorSend,
waitForTerminalLivePendingFlush
} from './terminal-live-pending-flush-state'
import { sendTerminalLiveControlAfterPendingFlush } from './terminal-live-control-send-order'
function deferred() {
let resolve = (_sent: boolean): void => {}
const promise = new Promise<boolean>((done) => {
resolve = done
})
return { promise, resolve }
}
afterEach(() => vi.useRealTimers())
describe('negotiated live-input pipelining', () => {
it('dispatches later bytes before receipts, but holds Enter until every receipt', async () => {
const state = createTerminalLivePendingFlushState()
const receipts = [deferred(), deferred()]
const sent: string[] = []
const sender = (_handle: string, text: string) => {
sent.push(text)
return receipts[sent.length - 1].promise
}
const a = queueTerminalLiveMirrorSend(state, 'pty', 'a', sender, { pipeline: true })
const b = queueTerminalLiveMirrorSend(state, 'pty', 'b', sender, { pipeline: true })
expect(sent).toEqual(['a', 'b'])
const enter = sendTerminalLiveControlAfterPendingFlush(
() => waitForTerminalLivePendingFlush(state),
async () => {
sent.push('\r')
return true
}
)
receipts[1].resolve(true)
await b
expect(sent).toEqual(['a', 'b'])
receipts[0].resolve(true)
await a
await expect(enter).resolves.toBe(true)
expect(sent).toEqual(['a', 'b', '\r'])
expect(state.retainedBytes).toBe(0)
expect(state.requestCount).toBe(0)
})
it('latches an out-of-order failure after earlier receipts and refuses Enter', async () => {
const state = createTerminalLivePendingFlushState()
const receipts = [deferred(), deferred()]
let index = 0
const sender = () => receipts[index++].promise
const a = queueTerminalLiveMirrorSend(state, 'pty', 'a', sender, { pipeline: true })
const b = queueTerminalLiveMirrorSend(state, 'pty', 'b', sender, { pipeline: true })
receipts[1].resolve(false)
await expect(b).resolves.toBe(false)
receipts[0].resolve(true)
await a
expect(state.current).toBeNull()
const control = vi.fn(async () => true)
await expect(
sendTerminalLiveControlAfterPendingFlush(
() => waitForTerminalLivePendingFlush(state),
control
)
).resolves.toBe(false)
expect(control).not.toHaveBeenCalled()
await expect(
queueTerminalLiveMirrorSend(state, 'pty', 'suffix', sender, { pipeline: true })
).resolves.toBe(false)
expect(index).toBe(2)
})
it('does not pipeline past a legacy RPC still awaiting its reply', async () => {
const state = createTerminalLivePendingFlushState()
const first = deferred()
const sent: string[] = []
const sender = async (_handle: string, text: string) => {
sent.push(text)
return text === 'old' ? first.promise : true
}
const legacy = queueTerminalLiveMirrorSend(state, 'pty', 'old', sender)
const modern = queueTerminalLiveMirrorSend(state, 'pty', 'new', sender, { pipeline: true })
expect(sent).toEqual(['old'])
first.resolve(true)
await Promise.all([legacy, modern])
expect(sent).toEqual(['old', 'new'])
})
it('bounds outstanding request bookkeeping and resets only on explicit cancellation', async () => {
const state = createTerminalLivePendingFlushState()
const receipt = deferred()
const sender = vi.fn(() => receipt.promise)
const requests = Array.from({ length: 64 }, () =>
queueTerminalLiveMirrorSend(state, 'pty', 'x', sender, { pipeline: true })
)
await expect(
queueTerminalLiveMirrorSend(state, 'pty', 'overflow', sender, { pipeline: true })
).resolves.toBe(false)
expect(sender).toHaveBeenCalledTimes(64)
expect(state.requestCount).toBe(64)
cancelTerminalLivePendingFlush(state)
await expect(Promise.all(requests)).resolves.toEqual(Array(64).fill(false))
expect(state.retainedBytes).toBe(0)
expect(state.requestCount).toBe(0)
await expect(
queueTerminalLiveMirrorSend(state, 'replacement', 'fresh', async () => true, {
pipeline: true
})
).resolves.toBe(true)
receipt.resolve(false)
await Promise.resolve()
await expect(waitForTerminalLivePendingFlush(state)).resolves.toBe(true)
})
it('bounds UTF-8 frame and total bytes, including acknowledged-later batches', async () => {
const state = createTerminalLivePendingFlushState()
const receipt = deferred()
const sender = vi.fn(() => receipt.promise)
const payload = 'é'.repeat(128 * 1024)
const requests = Array.from({ length: 4 }, () =>
queueTerminalLiveMirrorSend(state, 'pty', payload, sender, { pipeline: true })
)
expect(state.retainedBytes).toBe(1024 * 1024)
await expect(
queueTerminalLiveMirrorSend(state, 'pty', 'x', sender, { pipeline: true })
).resolves.toBe(false)
expect(sender).toHaveBeenCalledTimes(4)
cancelTerminalLivePendingFlush(state)
await Promise.all(requests)
await expect(
queueTerminalLiveMirrorSend(state, 'pty', payload + 'x', sender, { pipeline: true })
).resolves.toBe(false)
expect(sender).toHaveBeenCalledTimes(4)
receipt.resolve(false)
})
it.each([300, 400, 500])(
'keeps 50ms typing cadence under %ims receipts without weakening completion',
async (rtt) => {
vi.useFakeTimers()
const state = createTerminalLivePendingFlushState()
const arrivals: { text: string; at: number }[] = []
const started = Date.now()
const sender = (_handle: string, text: string) => {
arrivals.push({ text, at: Date.now() - started })
return new Promise<boolean>((resolve) => setTimeout(() => resolve(true), rtt))
}
const requests: Promise<boolean>[] = []
for (const text of ['a', 'b', 'c', 'd']) {
requests.push(queueTerminalLiveMirrorSend(state, 'pty', text, sender, { pipeline: true }))
await vi.advanceTimersByTimeAsync(50)
}
expect(arrivals).toEqual([
{ text: 'a', at: 0 },
{ text: 'b', at: 50 },
{ text: 'c', at: 100 },
{ text: 'd', at: 150 }
])
let completed = false
void waitForTerminalLivePendingFlush(state).then(() => {
completed = true
})
await vi.advanceTimersByTimeAsync(rtt - 51)
expect(completed).toBe(false)
await vi.advanceTimersByTimeAsync(1)
await expect(Promise.all(requests)).resolves.toEqual([true, true, true, true])
expect(completed).toBe(true)
}
)
})
mobile/src/terminal/use-terminal-live-input-commit.test.ts
+2 -2
View File
@@ -260,7 +260,7 @@ describe('terminal live input commit hook', () => {
await vi.waitFor(() => expect(sent).toEqual(['한']))
})
it('Given a failed mirrored Backspace When accessory input commits Then reports failure', async () => {
it('Given a failed mirrored prefix When accessory Backspace commits Then suppresses the dependent erase', async () => {
const { handlers, sent } = createTerminalLiveInputCommitHarness({ sendResult: false })
changeLiveInput(handlers, 'a', false)
await vi.waitFor(() => expect(sent).toEqual(['a']))
@@ -270,7 +270,7 @@ describe('terminal live input commit hook', () => {
localEdit: 'backspace'
})
expect(sent).toEqual(['a', '\x7f'])
expect(sent).toEqual(['a'])
expect(result).toEqual({ kind: 'suppress-raw' })
})
mobile/src/terminal/use-terminal-live-mirror-failure.test.ts
+213
View File
@@ -0,0 +1,213 @@
import { createElement, type RefObject } from 'react'
import { act, create, type ReactTestRenderer } from 'react-test-renderer'
import type { TextInput } from 'react-native'
import { afterEach, describe, expect, it, vi } from 'vitest'
import { useTerminalLivePendingInputFlush } from './use-terminal-live-pending-input-flush'
import type { TerminalLiveInputSender } from './terminal-live-input-sender'
function deferred() {
let resolve!: (sent: boolean) => void
const promise = new Promise<boolean>((done) => {
resolve = done
})
return { promise, resolve }
}
const unmounts: (() => void)[] = []
afterEach(() => {
unmounts.splice(0).forEach((unmount) => unmount())
vi.useRealTimers()
})
function harness(sender: TerminalLiveInputSender) {
const activeHandleRef = { current: 'terminal-a' as string | null }
const activeSessionTabTypeRef = { current: 'terminal' as string | null }
let capturedText = ''
let nativeText = ''
const setLiveInputCapture = vi.fn((text: string) => {
capturedText = text
})
const setNativeProps = vi.fn(({ text }: { text: string }) => {
nativeText = text
})
const liveInputRef = { current: { setNativeProps } as unknown as TextInput }
const sendLiveTerminalInputRef: RefObject<TerminalLiveInputSender> = { current: sender }
let hook!: ReturnType<typeof useTerminalLivePendingInputFlush<string>>
let renderer!: ReactTestRenderer
function Harness() {
hook = useTerminalLivePendingInputFlush({
activeHandleRef,
activeSessionTabTypeRef,
liveInputRef,
liveInputTerminalHandlesRef: { current: new Set(['terminal-a', 'terminal-b']) },
sendLiveTerminalInputRef,
setLiveInputCapture
})
return null
}
act(() => {
renderer = create(createElement(Harness))
})
const unmount = () => {
act(() => renderer.unmount())
}
unmounts.push(unmount)
return {
hook,
activeHandleRef,
activeSessionTabTypeRef,
setNativeProps,
setLiveInputCapture,
text: () => ({ capturedText, nativeText }),
type: (text: string, composing?: boolean) => {
capturedText = text
nativeText = text
return hook.applyLiveInputMirror(activeHandleRef.current!, text, composing)
},
unmount
}
}
describe('failed terminal live mirrors', () => {
it.each([300, 400, 500])(
'pipelines negotiated typing while retaining the %ims receipt barrier',
async (delay) => {
vi.useFakeTimers()
const writes: { text: string; at: number }[] = []
const start = Date.now()
const sender = Object.assign(
(_handle: string, text: string) => {
writes.push({ text, at: Date.now() - start })
return new Promise<boolean>((resolve) => setTimeout(() => resolve(true), delay))
},
{ supportsPipeline: () => true }
)
const h = harness(sender)
const sends = []
for (const text of ['a', 'ab', 'abc']) {
sends.push(h.type(text))
await vi.advanceTimersByTimeAsync(50)
}
expect(writes).toEqual([
{ text: 'a', at: 0 },
{ text: 'b', at: 50 },
{ text: 'c', at: 100 }
])
let finished = false
void h.hook.waitForPendingLiveInputFlush().then(() => {
finished = true
})
await vi.advanceTimersByTimeAsync(delay - 51)
expect(finished).toBe(false)
await vi.advanceTimersByTimeAsync(1)
await Promise.all(sends)
expect(finished).toBe(true)
}
)
it.each(['refused', 'throwing'] as const)(
'clears native and JS text on a %s send and retains the failure latch',
async (kind) => {
const sender = vi.fn(async () => {
if (kind === 'throwing') {
throw new Error('connection lost')
}
return false
})
const h = harness(sender)
await expect(h.type('optimistic')).resolves.toBe(false)
expect(h.text()).toEqual({ capturedText: '', nativeText: '' })
expect(h.hook.sentLiveInputTextRef.current).toBe('')
expect(h.hook.heldLiveInputTextRef.current).toBe('')
expect(h.hook.pendingLiveInputHandleRef.current).toBe('terminal-a')
await expect(h.hook.flushPendingLiveInputText('terminal-a')).resolves.toBe(false)
await expect(h.hook.waitForPendingLiveInputFlush()).resolves.toBe(false)
await expect(h.type('dependent suffix')).resolves.toBe(false)
expect(sender).toHaveBeenCalledTimes(1)
}
)
it('clears a newer dependent composition when its mirrored prefix fails', async () => {
vi.useFakeTimers()
const receipt = deferred()
const sender = vi.fn(() => receipt.promise)
const h = harness(sender)
const first = h.type('x')
const preedit = h.type('x한', true)
receipt.resolve(false)
await Promise.all([first, preedit])
expect(h.text()).toEqual({ capturedText: '', nativeText: '' })
expect(h.hook.liveInputComposingRef.current).toBeUndefined()
await vi.advanceTimersByTimeAsync(1000)
expect(sender).toHaveBeenCalledTimes(1)
})
it('does not clear another active terminal before its lifecycle reset runs', async () => {
const receipt = deferred()
const h = harness(() => receipt.promise)
const first = h.type('old')
h.activeHandleRef.current = 'terminal-b'
receipt.resolve(false)
await first
expect(h.setLiveInputCapture).not.toHaveBeenCalled()
expect(h.setNativeProps).not.toHaveBeenCalled()
})
it('a canceled failure and flush cannot clear a newer same-terminal interaction', async () => {
const old = deferred()
const fresh = deferred()
const h = harness((_handle, bytes) => (bytes === 'old' ? old.promise : fresh.promise))
const first = h.type('old')
const flush = h.hook.flushPendingLiveInputText('terminal-a')
h.hook.clearPendingLiveInputCommit()
const next = h.type('new')
await expect(first).resolves.toBe(false)
await expect(flush).resolves.toBe(false)
expect(h.text()).toEqual({ capturedText: 'new', nativeText: 'new' })
old.resolve(false)
fresh.resolve(true)
await next
expect(h.text()).toEqual({ capturedText: 'new', nativeText: 'new' })
})
it('an older successful flush cannot clear text typed while it waited', async () => {
const firstReceipt = deferred()
const h = harness(async (_handle, text) => (text === 'a' ? firstReceipt.promise : true))
const first = h.type('a')
const flush = h.hook.flushPendingLiveInputText('terminal-a')
const newer = h.type('ab')
firstReceipt.resolve(true)
await Promise.all([first, flush, newer])
expect(h.text()).toEqual({ capturedText: 'ab', nativeText: 'ab' })
})
it('unmount cancellation cannot update the native or JS field after settlement', async () => {
const receipt = deferred()
const h = harness(() => receipt.promise)
const pending = h.type('pending')
h.unmount()
unmounts.pop()
receipt.resolve(false)
await pending
expect(h.setLiveInputCapture).not.toHaveBeenCalled()
expect(h.setNativeProps).not.toHaveBeenCalled()
})
it('cancels transport input only for an unsettled mirror, before releasing its queue', async () => {
const receipt = deferred()
const cancelPending = vi.fn()
const sender = Object.assign(() => receipt.promise, { cancelPending })
const h = harness(sender)
h.hook.clearPendingLiveInputCommit()
expect(cancelPending).not.toHaveBeenCalled()
const first = h.type('pending')
h.hook.clearPendingLiveInputCommit()
expect(cancelPending).toHaveBeenCalledExactlyOnceWith('terminal-a')
await expect(first).resolves.toBe(false)
receipt.resolve(true)
const second = h.type('accepted')
await second
await h.hook.flushPendingLiveInputText('terminal-a')
expect(cancelPending).toHaveBeenCalledTimes(1)
})
})
mobile/src/terminal/use-terminal-live-pending-input-flush.ts
+93 -30
View File
@@ -58,6 +58,7 @@ export function useTerminalLivePendingInputFlush<TTabType extends string>({
const liveInputComposingRef = useRef<boolean | undefined>(undefined)
const sentLiveInputTextRef = useRef('')
const pendingLiveInputHandleRef = useRef<string | null>(null)
const mirrorRevisionRef = useRef(0)
const runMirrorStepRef = useRef<RunTerminalLiveMirrorStep>(async () => false)
const clearHeldCommitTimer = useCallback(() => {
@@ -67,20 +68,52 @@ export function useTerminalLivePendingInputFlush<TTabType extends string>({
}
}, [])
const resetMirrorState = useCallback(() => {
const clearMirrorState = useCallback(() => {
mirrorRevisionRef.current += 1
clearHeldCommitTimer()
cancelTerminalLivePendingFlush(pendingLiveInputFlushRef.current)
heldLiveInputTextRef.current = ''
liveInputComposingRef.current = undefined
sentLiveInputTextRef.current = ''
pendingLiveInputHandleRef.current = null
}, [clearHeldCommitTimer])
const clearPendingLiveInputCommit = useCallback(() => {
resetMirrorState()
const cancelPendingMirror = useCallback(() => {
const state = pendingLiveInputFlushRef.current
if (state.current && state.requestCount > 0) {
const handles = new Set(
[...state.activeBatches, ...state.pendingBatches].map((batch) => batch.handle)
)
for (const handle of handles) {
sendLiveTerminalInputRef.current.cancelPending?.(handle)
}
}
cancelTerminalLivePendingFlush(state)
}, [sendLiveTerminalInputRef])
const resetMirrorState = useCallback(() => {
cancelPendingMirror()
clearMirrorState()
}, [cancelPendingMirror, clearMirrorState])
const clearMirrorCapture = useCallback(() => {
clearMirrorState()
setLiveInputCapture('')
liveInputRef.current?.setNativeProps({ text: '' })
}, [liveInputRef, resetMirrorState, setLiveInputCapture])
}, [clearMirrorState, liveInputRef, setLiveInputCapture])
const clearPendingLiveInputCommit = useCallback(() => {
cancelPendingMirror()
clearMirrorCapture()
}, [cancelPendingMirror, clearMirrorCapture])
const clearRejectedMirror = useCallback(
(handle: string) => {
clearMirrorCapture()
// Keep the failed lane owned so terminal switches can cancel it.
pendingLiveInputHandleRef.current = handle
},
[clearMirrorCapture]
)
const waitForPendingLiveInputFlush = useCallback(async (): Promise<boolean> => {
return waitForTerminalLivePendingFlush(pendingLiveInputFlushRef.current)
@@ -111,6 +144,8 @@ export function useTerminalLivePendingInputFlush<TTabType extends string>({
commitHeld,
composing
})
const revision = ++mirrorRevisionRef.current
const generation = pendingLiveInputFlushRef.current.generation
sentLiveInputTextRef.current = step.nextSentText
heldLiveInputTextRef.current = step.heldText
liveInputComposingRef.current = composing
@@ -130,23 +165,37 @@ export function useTerminalLivePendingInputFlush<TTabType extends string>({
}
const payload = buildTerminalLiveMirrorPayload(step)
if (payload.length === 0) {
return waitForPendingLiveInputFlush()
const sent = await (payload.length === 0
? waitForPendingLiveInputFlush()
: queueTerminalLiveMirrorSend(
pendingLiveInputFlushRef.current,
handle,
payload,
sendQueuedMirrorPayload,
{ pipeline: sendLiveTerminalInputRef.current.supportsPipeline?.(handle) === true }
))
if (
!sent &&
generation === pendingLiveInputFlushRef.current.generation &&
revision === mirrorRevisionRef.current &&
activeHandleRef.current === handle &&
(activeSessionTabTypeRef.current == null ||
activeSessionTabTypeRef.current === 'terminal') &&
liveInputTerminalHandlesRef.current.has(handle)
) {
clearRejectedMirror(handle)
}
return queueTerminalLiveMirrorSend(
pendingLiveInputFlushRef.current,
handle,
payload,
sendQueuedMirrorPayload
)
return sent
},
[
activeHandleRef,
activeSessionTabTypeRef,
clearHeldCommitTimer,
clearRejectedMirror,
liveInputTerminalHandlesRef,
resetMirrorState,
sendQueuedMirrorPayload,
sendLiveTerminalInputRef,
waitForPendingLiveInputFlush
]
)
@@ -174,32 +223,46 @@ export function useTerminalLivePendingInputFlush<TTabType extends string>({
}
const heldText = heldLiveInputTextRef.current
const result =
const flush =
heldText.length > 0
? await runMirrorStep(handle, sentLiveInputTextRef.current + heldText, true)
: await waitForPendingLiveInputFlush()
? runMirrorStep(handle, sentLiveInputTextRef.current + heldText, true)
: waitForPendingLiveInputFlush()
const generation = pendingLiveInputFlushRef.current.generation
const revision = mirrorRevisionRef.current
const result = await flush
// Why: an explicit flush ends the field's editing session; the echoed PTY
// text stays, so local mirror state must restart from empty.
clearPendingLiveInputCommit()
if (
generation === pendingLiveInputFlushRef.current.generation &&
revision === mirrorRevisionRef.current &&
activeHandleRef.current === handle &&
(activeSessionTabTypeRef.current == null ||
activeSessionTabTypeRef.current === 'terminal') &&
liveInputTerminalHandlesRef.current.has(handle)
) {
if (result) {
clearPendingLiveInputCommit()
} else {
clearRejectedMirror(handle)
}
}
return result
},
[clearPendingLiveInputCommit, runMirrorStep, waitForPendingLiveInputFlush]
[
activeHandleRef,
activeSessionTabTypeRef,
liveInputTerminalHandlesRef,
clearPendingLiveInputCommit,
clearRejectedMirror,
runMirrorStep,
waitForPendingLiveInputFlush
]
)
useEffect(() => {
return () => {
if (heldCommitTimerRef.current) {
clearTimeout(heldCommitTimerRef.current)
heldCommitTimerRef.current = null
}
heldLiveInputTextRef.current = ''
liveInputComposingRef.current = undefined
sentLiveInputTextRef.current = ''
pendingLiveInputHandleRef.current = null
cancelTerminalLivePendingFlush(pendingLiveInputFlushRef.current)
}
}, [])
return resetMirrorState
}, [resetMirrorState])
return {
applyLiveInputMirror,
mobile/src/transport/direct-rpc-client.ts
+24 -24
View File
@@ -1,5 +1,6 @@
import type { ConnectOptions, RpcClient, SendRequestOptions } from './rpc-client'
import { DirectConnectionLog } from './direct-connection-log'
import { assertTerminalInputRequestAllowed } from './terminal-input-request-fence'
import { RpcClientAuthenticationRetry } from './rpc-client-authentication-retry'
import { RpcClientConnectionState } from './rpc-client-connection-state'
import {
@@ -19,6 +20,7 @@ import { RpcSessionLivenessWatchdog } from './rpc-session-liveness-watchdog'
import { isStaleForegroundDial } from './rpc-stale-dial'
import type { ConnectionState, ForegroundNudgeReason, RpcResponse } from './types'
import { negotiateMobileRuntimeCapabilities } from './mobile-runtime-capability-negotiation'
import { sendSessionEncrypted } from './rpc-client-socket-send'
const LIVENESS_REQUEST_ID_PREFIX = 'mobile-liveness-'
@@ -61,9 +63,12 @@ export class DirectRpcClient implements RpcClient {
nextId: () => this.nextId(),
deviceToken,
getState: () => this.connectionState.get(),
sendEncrypted: (request) => this.sendEncrypted(request)
sendEncrypted: (request) => this.sendEncrypted(request),
sendBinary: (bytes) => this.socketSession?.sendBinary(bytes) ?? false
})
this.requests = new RpcClientRequestTracker({
validateRequest: (method, params) =>
assertTerminalInputRequestAllowed(method, params, this.getTerminalStreamInputFailure),
nextId: () => this.nextId(),
deviceToken,
getState: () => this.connectionState.get(),
@@ -137,6 +142,17 @@ export class DirectRpcClient implements RpcClient {
return this.requests.sendRequest(method, params, options)
}
sendTerminalStreamInput = (terminal: string, text: string): Promise<boolean> | null =>
this.streams.sendTerminalStreamInput(terminal, text)
supportsTerminalStreamInput = (terminal: string): boolean =>
this.streams.supportsTerminalStreamInput(terminal)
getTerminalStreamInputFailure = (terminal: string) =>
this.streams.getTerminalStreamInputFailure(terminal)
recoverTerminalStreamInput = (terminal: string) =>
this.streams.recoverTerminalStreamInput(terminal)
cancelTerminalStreamInput = (terminal: string) => this.streams.cancelTerminalStreamInput(terminal)
subscribe(
method: string,
params: unknown,
@@ -153,23 +169,14 @@ export class DirectRpcClient implements RpcClient {
this.streams.updateTerminalViewport(terminal, viewport)
}
getState(): ConnectionState {
return this.connectionState.get()
}
getReconnectAttempt(): number {
return this.reconnect.getAttempt()
}
getLastConnectedAt(): number | null {
return this.connectionState.getLastConnectedAt()
}
getState = (): ConnectionState => this.connectionState.get()
getReconnectAttempt = (): number => this.reconnect.getAttempt()
getLastConnectedAt = (): number | null => this.connectionState.getLastConnectedAt()
getLastInboundAt = (): number | null => this.liveness.getLastInboundAt() || null
onStateChange(listener: (state: ConnectionState) => void): () => void {
return this.connectionState.addListener(listener)
}
onStateChange = (listener: (state: ConnectionState) => void): (() => void) =>
this.connectionState.addListener(listener)
notifyForeground(_reason?: ForegroundNudgeReason): void {
if (this.intentionallyClosed) {
@@ -204,6 +211,7 @@ export class DirectRpcClient implements RpcClient {
close(): void {
this.intentionallyClosed = true
this.streams.markForReplay()
this.reconnect.cancel()
const session = this.socketSession
session?.clearTimers()
@@ -285,15 +293,7 @@ export class DirectRpcClient implements RpcClient {
}
private sendEncrypted(request: unknown): boolean {
if (this.socketSession) {
return this.socketSession.sendEncrypted(request)
}
console.log('[net] sendEncrypted FAILED — channel not ready', {
hasWs: false,
hasKey: false,
state: this.getState()
})
return false
return sendSessionEncrypted(this.socketSession, request, this.getState())
}
private sendLivenessProbe(identity: object): boolean {
mobile/src/transport/e2ee.ts
+1 -1
View File
@@ -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))
mobile/src/transport/logical-client-cutover.ts
+13
View File
@@ -0,0 +1,13 @@
export class LogicalClientCutoverError extends Error {
constructor() {
super('RPC interrupted by connection migration')
}
}
export type PendingLogicalRequest = { reject: (error: Error) => void }
export function isLogicalClientCutoverError(error: unknown): boolean {
return (
error instanceof LogicalClientCutoverError ||
(error instanceof Error && error.message === 'RPC interrupted by connection migration')
)
}
mobile/src/transport/logical-terminal-stream-input.test.ts
+34
View File
@@ -0,0 +1,34 @@
import { describe, expect, it, vi } from 'vitest'
import { LogicalTerminalStreamInput } from './logical-terminal-stream-input'
import type { RpcClient } from './rpc-client'
describe('logical terminal input cutover', () => {
it('keeps a pending old-stream prefix from falling back on a replacement old host', async () => {
let resolve = (_accepted: boolean) => {}
const original = new Promise<boolean>((r) => {
resolve = r
})
let generation = 1
let session = {
supportsTerminalStreamInput: () => true,
sendTerminalStreamInput: vi.fn(() => original)
} as unknown as RpcClient
const input = new LogicalTerminalStreamInput(() => ({ generation, session, available: true }))
const prefix = input.send('t', 'prefix')!
generation = 2
session = {} as RpcClient
expect(input.supports('t')).toBe(true)
expect(await input.send('t', '\r')).toBe(false)
resolve(true)
expect(await prefix).toBe(false)
expect(await input.send('t', '\r')).toBe(false)
session = {
supportsTerminalStreamInput: () => true,
recoverTerminalStreamInput: () => true,
sendTerminalStreamInput: vi.fn(() => Promise.resolve(true))
} as unknown as RpcClient
expect(await input.send('t', 'blocked')).toBe(false)
expect(input.recover('t')).toBe(true)
expect(await input.send('t', 'fresh')).toBe(true)
})
})
mobile/src/transport/logical-terminal-stream-input.ts
+85
View File
@@ -0,0 +1,85 @@
import type { RpcClient } from './rpc-client'
import type { TerminalStreamInputFailure } from './terminal-stream-input-failure'
type InputAttempt = {
generation: number
pending: number
failure: TerminalStreamInputFailure | null
}
type InputContext = { generation: number; session: RpcClient; available: boolean }
const interrupted: TerminalStreamInputFailure = {
outcome: 'unknown',
reason: 'connection_interrupted'
}
export class LogicalTerminalStreamInput {
private readonly attempts = new Map<string, InputAttempt>()
constructor(private readonly context: () => InputContext) {}
supports = (terminal: string): boolean => {
const { session } = this.context()
return this.attempts.has(terminal) || (session.supportsTerminalStreamInput?.(terminal) ?? false)
}
failure = (terminal: string): TerminalStreamInputFailure | null => {
const current = this.context()
const attempt = this.attempts.get(terminal)
if (attempt && (!current.available || attempt.generation !== current.generation)) {
attempt.failure ??= interrupted
}
return attempt?.failure ?? current.session.getTerminalStreamInputFailure?.(terminal) ?? null
}
recover = (terminal: string): boolean => {
const current = this.context()
if (!current.available || !current.session.recoverTerminalStreamInput?.(terminal)) {
return false
}
this.attempts.delete(terminal)
return true
}
cancel = (terminal: string): void => {
const attempt = this.attempts.get(terminal)
if (attempt?.pending) {
attempt.failure ??= { outcome: 'unknown', reason: 'cancelled' }
}
this.context().session.cancelTerminalStreamInput?.(terminal)
}
send = (terminal: string, text: string): Promise<boolean> | null => {
const context = this.context()
if (!context.available || this.failure(terminal)) {
return Promise.resolve(false)
}
const previous = this.attempts.get(terminal)
const result = context.session.sendTerminalStreamInput?.(terminal, text) ?? null
if (!result) {
return previous ? Promise.resolve(false) : null
}
const attempt = previous ?? { generation: context.generation, pending: 0, failure: null }
attempt.pending += 1
this.attempts.set(terminal, attempt)
return result.then(
(accepted) => this.settle(terminal, attempt, accepted),
() => this.settle(terminal, attempt, false)
)
}
private settle(terminal: string, attempt: InputAttempt, accepted: boolean): boolean {
const current = this.context()
const valid =
accepted && current.available && current.generation === attempt.generation && !attempt.failure
attempt.pending -= 1
if (!valid) {
attempt.failure ??=
current.generation === attempt.generation
? (current.session.getTerminalStreamInputFailure?.(terminal) ?? interrupted)
: interrupted
}
if (this.attempts.get(terminal) === attempt && attempt.pending === 0 && !attempt.failure) {
this.attempts.delete(terminal)
}
return valid
}
}
mobile/src/transport/mobile-relay-rpc-session.test.ts
+32
View File
@@ -5,6 +5,7 @@ import {
} from '../../../src/shared/browser-screencast-protocol'
import { encodeTerminalStreamFrame, TerminalStreamOpcode } from './terminal-stream-protocol'
import { isRpcDeliveryUnknown } from './rpc-delivery-ambiguity'
import { TERMINAL_ORDERED_INPUT_CAPABILITY } from '../../../src/shared/terminal-ordered-input'
const fakes = vi.hoisted(() => ({
linkOptions: null as null | {
@@ -19,6 +20,7 @@ const fakes = vi.hoisted(() => ({
onError(error: Error): void
},
sendText: vi.fn(() => true),
sendBinary: vi.fn(() => true),
close: vi.fn()
}))
@@ -28,6 +30,7 @@ vi.mock('./mobile-relay-e2ee-link', () => ({
fakes.linkOptions = options
}
sendText = fakes.sendText
sendBinary = fakes.sendBinary
close = fakes.close
}
}))
@@ -133,9 +136,38 @@ describe('mobile relay RPC session', () => {
vi.clearAllMocks()
fakes.linkOptions = null
fakes.sendText.mockReturnValue(true)
fakes.sendBinary.mockReturnValue(true)
})
afterEach(() => vi.useRealTimers())
it('rechecks failed input after the connected await before sending JSON controls', async () => {
const { session } = await authenticateSession()
const dispose = session.subscribe('terminal.subscribe', { terminal: 't' }, vi.fn())
await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledOnce())
const request = JSON.parse(fakes.sendText.mock.calls[0]![0] as string)
fakes.linkOptions!.onText(
JSON.stringify({
id: request.id,
ok: true,
streaming: true,
result: {
type: 'subscribed',
streamId: 7,
capabilities: { orderedInput: TERMINAL_ORDERED_INPUT_CAPABILITY }
},
_meta: { runtimeId: 'runtime-1' }
})
)
fakes.sendText.mockClear()
const enter = session.sendRequest('terminal.send', { terminal: 't', text: '\r' })
fakes.sendBinary.mockReturnValue(false)
expect(await session.sendTerminalStreamInput?.('t', 'prefix')).toBe(false)
await expect(enter).rejects.toThrow('Terminal input stopped')
expect(fakes.sendText).not.toHaveBeenCalled()
dispose()
session.close()
})
it('requires exact resume observations and confirms by request ID before becoming connected', async () => {
const { session, confirmationRequest, capabilityRequest } = await authenticateSession()
mobile/src/transport/mobile-relay-rpc-session.ts
+9
View File
@@ -1,3 +1,4 @@
import { assertTerminalInputRequestAllowed } from './terminal-input-request-fence'
import {
PairingGetEndpointsResultSchema,
type DeviceResumeConfirmed,
@@ -61,6 +62,7 @@ export function connectMobileRelayRpcSession(args: {
const streams = new MobileRelayRpcStreams({
nextId: () => pending.nextId(),
sendFrame,
sendBinary: (bytes) => !closed && state === 'connected' && link.sendBinary(bytes),
waitForConnected: () => waitForConnected()
})
@@ -99,9 +101,16 @@ export function connectMobileRelayRpcSession(args: {
})
const client: MobileRelayRpcSession = {
supportsTerminalStreamInput: (terminal) => streams.supportsTerminalStreamInput(terminal),
getTerminalStreamInputFailure: (terminal) => streams.getTerminalStreamInputFailure(terminal),
recoverTerminalStreamInput: (terminal) => streams.recoverTerminalStreamInput(terminal),
cancelTerminalStreamInput: (terminal) => streams.cancelTerminalStreamInput(terminal),
sendTerminalStreamInput: (terminal, text) => streams.sendTerminalStreamInput(terminal, text),
async sendRequest(method, params, options) {
assertTerminalInputRequestAllowed(method, params, streams.getTerminalStreamInputFailure)
const budget = openRpcRequestBudget(options)
await waitForConnected(budget.timeoutMs)
assertTerminalInputRequestAllowed(method, params, streams.getTerminalStreamInputFailure)
return sendRpc(method, params, resolvePostConnectRequestTimeout(budget, requestTimeoutMs))
},
mobile/src/transport/mobile-relay-rpc-streams.ts
+30 -3
View File
@@ -9,6 +9,7 @@ import {
} from './rpc-client-terminal-subscription'
import type { RpcClient } from './rpc-client'
import type { RpcResponse, RpcSuccess } from './types'
import { TerminalOrderedInput, advertiseTerminalOrderedInput } from './terminal-ordered-input'
type StreamRecord = {
method: string
@@ -28,6 +29,7 @@ type StreamManagerOptions = {
nextId: () => string
sendFrame: (request: { id: string; method: string; params?: unknown }) => boolean
waitForConnected: () => Promise<void>
sendBinary?: (bytes: Uint8Array) => boolean
}
export class MobileRelayRpcStreams {
@@ -35,8 +37,26 @@ export class MobileRelayRpcStreams {
private readonly terminalListeners = new Map<number, (result: unknown) => void>()
private readonly terminalSnapshots = new Map<number, TerminalSnapshotState>()
private activeBrowserStream: StreamRecord | null = null
private readonly orderedInput: TerminalOrderedInput
constructor(private readonly options: StreamManagerOptions) {}
constructor(private readonly options: StreamManagerOptions) {
this.orderedInput = new TerminalOrderedInput((bytes) => options.sendBinary?.(bytes) ?? false)
}
sendTerminalStreamInput(terminal: string, text: string): Promise<boolean> | null {
return this.orderedInput.send(terminal, text)
}
getTerminalStreamInputFailure = (terminal: string) => this.orderedInput.failure(terminal)
recoverTerminalStreamInput = (terminal: string) => this.orderedInput.recover(terminal)
cancelTerminalStreamInput(terminal: string): void {
for (const id of this.orderedInput.cancel(terminal)) {
this.cancel(id)
}
}
supportsTerminalStreamInput(terminal: string): boolean {
return this.orderedInput.supports(terminal)
}
subscribe(
method: string,
@@ -47,7 +67,7 @@ export class MobileRelayRpcStreams {
const id = this.options.nextId()
const stream: StreamRecord = {
method,
params,
params: advertiseTerminalOrderedInput(method, params),
listener,
onBinaryFrame: subscribeOptions?.onBinaryFrame,
streamIds: new Set(),
@@ -90,8 +110,13 @@ export class MobileRelayRpcStreams {
stream.subscriptionId = metadata.subscriptionId
}
if (typeof metadata.streamId === 'number') {
const streamId = metadata.streamId
this.orderedInput.register(response.id, stream.params, result)
stream.streamIds.add(metadata.streamId)
this.terminalListeners.set(metadata.streamId, stream.listener)
this.terminalListeners.set(metadata.streamId, (event) => {
this.orderedInput.handle(streamId, event)
stream.listener(event)
})
}
if (stream.method === 'browser.screencast') {
this.activeBrowserStream = stream
@@ -121,6 +146,7 @@ export class MobileRelayRpcStreams {
}
clear(): void {
this.orderedInput.clear()
for (const stream of this.streams.values()) {
stream.cancelled = true
}
@@ -156,6 +182,7 @@ export class MobileRelayRpcStreams {
}
private remove(id: string): void {
this.orderedInput.reset(id)
const stream = this.streams.get(id)
if (!stream) {
return
mobile/src/transport/rpc-client-request-tracker.ts
+2
View File
@@ -16,6 +16,7 @@ type RequestTrackerOptions = {
waitForConnected: (timeoutMs?: number) => Promise<void>
sendEncrypted: (request: unknown) => boolean
deviceToken: string
validateRequest?: (method: string, params: unknown) => void
}
export class RpcClientRequestTracker {
@@ -63,6 +64,7 @@ export class RpcClientRequestTracker {
timeoutMs: number
): Promise<RpcResponse> {
return new Promise((resolve, reject) => {
this.options.validateRequest?.(method, params)
const id = this.options.nextId()
const timeout = setTimeout(() => {
this.pending.delete(id)
mobile/src/transport/rpc-client-socket-options.ts
+21
View File
@@ -0,0 +1,21 @@
import type { RpcClientSocketSession } from './rpc-client-socket-session'
import type { ConnectionLogEmitter, ConnectionState, RpcResponse } from './types'
export type SocketSessionOptions = {
endpoint: string
deviceToken: string
serverPublicKey: Uint8Array
getCurrentSocket: () => WebSocket | null
getState: () => ConnectionState
getReconnectAttempt: () => number
isIntentionallyClosed: () => boolean
emitLog: ConnectionLogEmitter
onHandshakeStarted: () => void
onAuthenticated: (session: RpcClientSocketSession) => void
onAuthRejected: (reason: string) => void
onRpcResponse: (response: RpcResponse) => void
onBinary: (bytes: Uint8Array) => void
onAnyInbound: (receivedAt: number) => void
onAuthenticatedInbound: (session: RpcClientSocketSession) => void
onClosed: (session: RpcClientSocketSession, closeCode?: number) => void
onForcedClose: (session: RpcClientSocketSession) => void
}
mobile/src/transport/rpc-client-socket-send.ts
+38
View File
@@ -0,0 +1,38 @@
import { encryptBytes } from './e2ee'
import type { RpcClientSocketSession } from './rpc-client-socket-session'
import type { ConnectionState } from './types'
export function sendSessionEncrypted(
session: RpcClientSocketSession | null,
request: unknown,
state: ConnectionState
): boolean {
if (session) {
return session.sendEncrypted(request)
}
console.log('[net] sendEncrypted FAILED — channel not ready', {
hasWs: false,
hasKey: false,
state
})
return false
}
export function sendSocketEncryptedBinary(
socket: WebSocket,
key: Uint8Array | null,
ready: boolean,
bytes: Uint8Array,
onFailure: () => void
): boolean {
if (!ready || !key || socket.readyState !== WebSocket.OPEN) {
return false
}
try {
socket.send(encryptBytes(bytes, key))
return true
} catch {
onFailure()
return false
}
}
mobile/src/transport/rpc-client-socket-session.ts
+12 -21
View File
@@ -9,33 +9,14 @@ import {
import { isRpcResponse } from './rpc-response-shape'
import { isStaleRpcSocketEvent, logRpcSocketClose } from './rpc-socket-close-evidence'
import { describeSocketEvent, redactSocketEndpoint } from './socket-event-debug'
import type { ConnectionLogEmitter, ConnectionState, RpcResponse } from './types'
import type { SocketSessionOptions } from './rpc-client-socket-options'
import { websocketPayloadToUint8 } from './websocket-payload-bytes'
import { sendSocketEncryptedBinary } from './rpc-client-socket-send'
const CONNECT_TIMEOUT_MS = 12_000
const HANDSHAKE_TIMEOUT_MS = 5_000
const WEBSOCKET_CONNECTING_STATE = 0
type SocketSessionOptions = {
endpoint: string
deviceToken: string
serverPublicKey: Uint8Array
getCurrentSocket: () => WebSocket | null
getState: () => ConnectionState
getReconnectAttempt: () => number
isIntentionallyClosed: () => boolean
emitLog: ConnectionLogEmitter
onHandshakeStarted: () => void
onAuthenticated: (session: RpcClientSocketSession) => void
onAuthRejected: (reason: string) => void
onRpcResponse: (response: RpcResponse) => void
onBinary: (bytes: Uint8Array) => void
onAnyInbound: (receivedAt: number) => void
onAuthenticatedInbound: (session: RpcClientSocketSession) => void
onClosed: (session: RpcClientSocketSession, closeCode?: number) => void
onForcedClose: (session: RpcClientSocketSession) => void
}
export class RpcClientSocketSession {
readonly socket: WebSocket
readonly constructedAt = Date.now()
@@ -86,6 +67,16 @@ export class RpcClientSocketSession {
this.socket.close()
}
sendBinary(bytes: Uint8Array): boolean {
return sendSocketEncryptedBinary(
this.socket,
this.sharedKey,
this.authenticated && this.options.getCurrentSocket() === this.socket,
bytes,
() => this.options.onForcedClose(this)
)
}
clearTimers(): void {
if (this.connectTimer) {
clearTimeout(this.connectTimer)
mobile/src/transport/rpc-client-stream-options.ts
+9
View File
@@ -0,0 +1,9 @@
import type { ConnectionState } from './types'
export type StreamRegistryOptions = {
nextId: () => string
deviceToken: string
getState: () => ConnectionState
sendEncrypted: (request: unknown) => boolean
sendBinary?: (bytes: Uint8Array) => boolean
}
mobile/src/transport/rpc-client-stream-registry.ts
+23 -14
View File
@@ -13,7 +13,9 @@ import {
isTerminalSubscribedResult
} from './rpc-subscription-result-shapes'
import { RpcClientTerminalStreamRouter } from './rpc-client-terminal-stream-router'
import type { ConnectionState, RpcResponse, RpcSuccess } from './types'
import type { RpcResponse, RpcSuccess } from './types'
import type { StreamRegistryOptions } from './rpc-client-stream-options'
import { TerminalOrderedInput, advertiseTerminalOrderedInput } from './terminal-ordered-input'
export type RpcStreamingListener = (result: unknown) => void
@@ -31,20 +33,24 @@ type StreamRequest = {
sent?: boolean
}
type StreamRegistryOptions = {
nextId: () => string
deviceToken: string
getState: () => ConnectionState
sendEncrypted: (request: unknown) => boolean
}
export class RpcClientStreamRegistry {
private readonly streams = new Map<string, StreamRequest>()
private readonly terminalRouter = new RpcClientTerminalStreamRouter()
private activeBrowserRequestId: string | null = null
private pendingBrowserRequestId: string | null = null
private readonly orderedInput = new TerminalOrderedInput(
(bytes) => this.options.sendBinary?.(bytes) ?? false
)
constructor(private readonly options: StreamRegistryOptions) {}
readonly sendTerminalStreamInput = this.orderedInput.send.bind(this.orderedInput)
readonly supportsTerminalStreamInput = this.orderedInput.supports.bind(this.orderedInput)
readonly getTerminalStreamInputFailure = this.orderedInput.failure.bind(this.orderedInput)
readonly recoverTerminalStreamInput = this.orderedInput.recover.bind(this.orderedInput)
cancelTerminalStreamInput(terminal: string): void {
for (const id of this.orderedInput.cancel(terminal)) {
this.dispose(id)
}
}
subscribe(
method: string,
@@ -55,7 +61,7 @@ export class RpcClientStreamRegistry {
const id = this.options.nextId()
const stream: StreamRequest = {
method,
params,
params: advertiseTerminalOrderedInput(method, params),
listener,
onBinaryFrame: subscribeOptions?.onBinaryFrame
}
@@ -156,9 +162,7 @@ export class RpcClientStreamRegistry {
updateTerminalSubscriptionViewport(this.streams.values(), terminal, viewport)
}
size(): number {
return this.streams.size
}
readonly size = (): number => this.streams.size
private handleStreamingResponse(response: RpcSuccess): void {
const stream = this.streams.get(response.id)
@@ -187,7 +191,11 @@ export class RpcClientStreamRegistry {
}
}
if (isTerminalSubscribedResult(result)) {
this.terminalRouter.register(response.id, result.streamId, stream.listener)
this.orderedInput.register(response.id, stream.params, result)
this.terminalRouter.register(response.id, result.streamId, (event) => {
this.orderedInput.handle(result.streamId, event)
stream.listener(event)
})
}
if (!stream.cancelled) {
stream.listener(result)
@@ -293,6 +301,7 @@ export class RpcClientStreamRegistry {
}
private resetTerminalRouting(id: string): void {
this.orderedInput.reset(id)
this.terminalRouter.reset(id)
}
mobile/src/transport/rpc-client.ts
+8
View File
@@ -22,6 +22,14 @@ type SubscribeOptions = {
type StreamingListener = (result: unknown) => void
export type RpcClient = {
supportsTerminalStreamInput?: (terminal: string) => boolean
getTerminalStreamInputFailure?: (
terminal: string
) => import('./terminal-stream-input-failure').TerminalStreamInputFailure | null
recoverTerminalStreamInput?: (terminal: string) => boolean
cancelTerminalStreamInput?: (terminal: string) => void
/** null means no negotiated stream; a returned promise must never be retried over RPC. */
sendTerminalStreamInput?: (terminal: string, text: string) => Promise<boolean> | null
sendRequest: (
method: string,
params?: unknown,
mobile/src/transport/stable-logical-rpc-client.test.ts
+37
View File
@@ -68,6 +68,43 @@ function deferred<T>() {
}
describe('stable logical RPC client', () => {
it('fences JSON terminal.send after an ordered prefix crosses a logical cutover', async () => {
const old = new FakeSession('connected')
const pending = deferred<boolean>()
Object.assign(old, { sendTerminalStreamInput: () => pending.promise })
const client = createStableLogicalRpcClient(old, 'lan')
const prefix = client.sendTerminalStreamInput?.('t', 'prefix')
const fresh = new FakeSession('connected')
Object.assign(fresh, { recoverTerminalStreamInput: () => true })
fresh.sendRequest.mockResolvedValue(success({}))
await client.migrateTo(fresh, 'relay')
await expect(
client.sendRequest('terminal.send', { terminal: 't', text: '\r' })
).rejects.toThrow('Terminal input stopped')
expect(fresh.sendRequest).not.toHaveBeenCalled()
expect(client.getTerminalStreamInputFailure?.('t')?.outcome).toBe('unknown')
expect(client.recoverTerminalStreamInput?.('t')).toBe(true)
pending.resolve(true)
expect(await prefix).toBe(false)
expect(client.getTerminalStreamInputFailure?.('t')).toBe(null)
await client.sendRequest('terminal.send', { terminal: 't', text: 'new' })
expect(fresh.sendRequest).toHaveBeenCalledOnce()
client.close()
})
it('exposes negotiated stream support without probing or sending input', async () => {
const session = new FakeSession('connected')
const supports = vi.fn(() => true)
const send = vi.fn(() => Promise.resolve(false))
Object.assign(session, { supportsTerminalStreamInput: supports, sendTerminalStreamInput: send })
const client = createStableLogicalRpcClient(session, 'lan')
expect(client.supportsTerminalStreamInput?.('t')).toBe(true)
expect(send).not.toHaveBeenCalled()
expect(await client.sendTerminalStreamInput?.('t', 'x')).toBe(false)
expect(client.supportsTerminalStreamInput?.('t')).toBe(true)
client.close()
expect(client.supportsTerminalStreamInput?.('t')).toBe(true)
expect(await client.sendTerminalStreamInput?.('t', '\r')).toBe(false)
})
it('advertises source-default support on worktree catalog requests', async () => {
const session = new FakeSession('connected')
session.sendRequest.mockResolvedValue(success([]))
mobile/src/transport/stable-logical-rpc-client.ts
+20 -22
View File
@@ -1,4 +1,5 @@
import type { ConnectionState, RpcResponse } from './types'
import { assertTerminalInputRequestAllowed } from './terminal-input-request-fence'
import type { RpcClient } from './rpc-client'
import {
forwardMigrationDialState,
@@ -7,23 +8,12 @@ import {
import { waitForAuthenticated } from './replacement-session-authentication'
import { projectMobileRpcRequestParams } from './mobile-rpc-request-projection'
import { LogicalClientConnectionPath } from './logical-client-connection-path'
import { LogicalClientCutoverError, type PendingLogicalRequest } from './logical-client-cutover'
import { LogicalTerminalStreamInput } from './logical-terminal-stream-input'
export { LogicalClientCutoverError, isLogicalClientCutoverError } from './logical-client-cutover'
export type MobileConnectionPath = 'lan' | 'tailscale' | 'relay'
export class LogicalClientCutoverError extends Error {
constructor() {
super('RPC interrupted by connection migration')
}
}
// Why: instanceof can miss across bundle copies, so also match by message.
export function isLogicalClientCutoverError(error: unknown): boolean {
return (
error instanceof LogicalClientCutoverError ||
(error instanceof Error && error.message === 'RPC interrupted by connection migration')
)
}
type SubscriptionRecord = {
method: string
params: unknown
@@ -33,10 +23,6 @@ type SubscriptionRecord = {
cancelled: boolean
}
type PendingRequest = {
reject: (error: Error) => void
}
export type StableLogicalRpcClient = RpcClient & {
migrateTo(
session: RpcClient,
@@ -75,20 +61,32 @@ export function createStableLogicalRpcClient(
let nextSubscriptionId = 0
let activeStateUnsubscribe: (() => void) | null = null
const subscriptions = new Map<number, SubscriptionRecord>()
const pendingRequests = new Set<PendingRequest>()
const pendingRequests = new Set<PendingLogicalRequest>()
const stateListeners = new Set<(state: ConnectionState) => void>()
let state = initialSession.getState()
const connectionPath = new LogicalClientConnectionPath(() => state === 'connected')
bindActiveState(initialSession, generation)
const terminalInput = new LogicalTerminalStreamInput(() => ({
generation,
session: activeSession,
available: !closed && !suspended && state === 'connected'
}))
const logical: StableLogicalRpcClient = {
sendRequest(method, params, options) {
supportsTerminalStreamInput: terminalInput.supports,
sendTerminalStreamInput: terminalInput.send,
getTerminalStreamInputFailure: terminalInput.failure,
recoverTerminalStreamInput: terminalInput.recover,
cancelTerminalStreamInput: terminalInput.cancel,
async sendRequest(method, params, options) {
assertTerminalInputRequestAllowed(method, params, terminalInput.failure)
if (closed) {
return Promise.reject(new Error('Client closed'))
throw new Error('Client closed')
}
if (suspended) {
return Promise.reject(new Error('Client suspended'))
throw new Error('Client suspended')
}
const requestGeneration = generation
const session = activeSession
mobile/src/transport/terminal-input-request-fence.test.ts
+41
View File
@@ -0,0 +1,41 @@
import { describe, expect, it, vi } from 'vitest'
import { assertTerminalInputRequestAllowed } from './terminal-input-request-fence'
import { RpcClientRequestTracker } from './rpc-client-request-tracker'
describe('failed ordered prefix RPC fence', () => {
it('fences only terminal.send for the failed terminal', () => {
const failure = (terminal: string) =>
terminal === 't' ? { outcome: 'unknown' as const, reason: 'lost' } : null
expect(() =>
assertTerminalInputRequestAllowed('terminal.send', { terminal: 't' }, failure)
).toThrow('Terminal input stopped')
expect(() =>
assertTerminalInputRequestAllowed('terminal.send', { terminal: 'other' }, failure)
).not.toThrow()
expect(() =>
assertTerminalInputRequestAllowed('terminal.subscribe', { terminal: 't' }, failure)
).not.toThrow()
})
it('rechecks direct requests after a connection wait before writing JSON', async () => {
let failed = false
const send = vi.fn(() => true)
const tracker = new RpcClientRequestTracker({
nextId: () => '1',
deviceToken: 'd',
getState: () => 'connected',
waitForConnected: async () => {
failed = true
},
sendEncrypted: send,
validateRequest: (method, params) =>
assertTerminalInputRequestAllowed(method, params, () =>
failed ? { outcome: 'unknown', reason: 'lost' } : null
)
})
await expect(
tracker.sendRequest('terminal.send', { terminal: 't', text: '\r' })
).rejects.toThrow('Terminal input stopped')
expect(send).not.toHaveBeenCalled()
expect(tracker.size()).toBe(0)
})
})
mobile/src/transport/terminal-input-request-fence.ts
+17
View File
@@ -0,0 +1,17 @@
import type { TerminalStreamInputFailure } from './terminal-stream-input-failure'
export function assertTerminalInputRequestAllowed(
method: string,
params: unknown,
failure: (terminal: string) => TerminalStreamInputFailure | null
): void {
if (method !== 'terminal.send' || !params || typeof params !== 'object') {
return
}
const terminal = (params as { terminal?: unknown }).terminal
if (typeof terminal === 'string' && failure(terminal)) {
throw new Error(
'Terminal input stopped; explicitly recover a fresh subscription before sending more input.'
)
}
}
mobile/src/transport/terminal-ordered-input-pty.integration.test.ts
+254
View File
@@ -0,0 +1,254 @@
import { randomBytes } from 'node:crypto'
import { expect, it, vi } from 'vitest'
import nacl from 'tweetnacl'
import WebSocket, { WebSocketServer } from 'ws'
import '../../../config/scripts/vitest-host-ports-setup'
import { E2EEChannel } from '../../../src/main/runtime/rpc/e2ee-channel'
import { RpcDispatcher } from '../../../src/main/runtime/rpc/dispatcher'
import { TERMINAL_METHODS } from '../../../src/main/runtime/rpc/methods/terminal'
import {
createOrderedInputPtyTestRig,
inputProofDeadline
} from '../../../src/main/runtime/rpc/terminal-ordered-input-pty-test-rig'
import {
decodeTerminalStreamFrame,
type TerminalStreamFrame
} from '../../../src/shared/terminal-stream-protocol'
import { MobileE2EEV2ClientSession } from './mobile-e2ee-v2-client-session'
import { MobileE2EEV2PhysicalChannel } from './mobile-e2ee-v2-physical-channel'
import { MobileRelayRpcStreams } from './mobile-relay-rpc-streams'
import {
queueTerminalLiveMirrorSend,
createTerminalLivePendingFlushState,
waitForTerminalLivePendingFlush
} from '../terminal/terminal-live-pending-flush-state'
vi.mock('expo-crypto', () => ({
getRandomBytes: (length: number) => new Uint8Array(randomBytes(length))
}))
it.each(
[300, 400, 500].flatMap((receiptDelayMs) => [
{ receiptDelayMs, pipeline: true },
{ receiptDelayMs, pipeline: false }
])
)(
'encrypted real-PTY cadence: receipt delay $receiptDelayMs ms, pipeline $pipeline',
async ({ receiptDelayMs, pipeline }) => {
const payloads = ['BOM:\ufeff', 'abcdefghijklmnop', 'é'.repeat(9000), '\t\x1b[A\x7f\u0000']
const suffix = '\r'
const expected = Buffer.from(payloads.join('') + suffix)
const rig = await createOrderedInputPtyTestRig(expected)
const server = new WebSocketServer({ host: '127.0.0.1', port: 0, perMessageDeflate: false })
const handlers = new Map<number, (frame: TerminalStreamFrame) => void>()
const abort = new AbortController()
const errors: unknown[] = []
const sessions: E2EEChannel[] = []
const dispatches: Promise<void>[] = []
const delayedSends = new Set<ReturnType<typeof setTimeout>>()
const keys = nacl.box.keyPair()
const dispatcher = new RpcDispatcher({ runtime: rig.runtime, methods: TERMINAL_METHODS })
let phone: WebSocket | undefined
let mobile: MobileE2EEV2PhysicalChannel | undefined
let disposeSubscription: (() => void) | undefined
let clearStreams: (() => void) | undefined
let delayReplies = false
let receipts = 0
let bytesBeforeFirstReceipt: Buffer | undefined
const pending = createTerminalLivePendingFlushState()
try {
await inputProofDeadline(
new Promise<void>((resolve, reject) => {
server.once('listening', resolve)
server.once('error', reject)
}),
'encrypted server listener'
)
server.on('connection', (socket) => {
const channel = new E2EEChannel(socket, {
serverSecretKey: keys.secretKey,
resolveAuthenticatedDevice: (token) =>
token === 'test-device'
? { deviceId: 'phone', deviceToken: token, scope: 'mobile' }
: null,
transportContext: { transport: 'relay', relayHostId: 'AbCdEf0123_-xyZ9' },
requireV2: true,
onReady: () => {},
onError: (code, reason) => errors.push({ code, reason })
})
sessions.push(channel)
channel.onMessage((plaintext, reply, sendBinary) => {
const request = JSON.parse(plaintext)
const dispatched = dispatcher.dispatchStreaming(request, reply, {
connectionId: 'encrypted-input-proof',
signal: abort.signal,
clientKind: 'mobile',
sendBinary: (bytes) => {
if (!delayReplies) {
return sendBinary(bytes)
}
const schedule = () => {
const timer = setTimeout(() => {
delayedSends.delete(timer)
sendBinary(bytes)
}, receiptDelayMs)
delayedSends.add(timer)
}
schedule()
return true
},
registerBinaryStreamHandler: (id, handler) => {
handlers.set(id, handler)
return () => {
handlers.delete(id)
}
}
})
dispatches.push(dispatched)
void dispatched.catch((error) => errors.push(error))
})
channel.onBinaryMessage((bytes) => {
const frame = decodeTerminalStreamFrame(bytes)
if (frame) {
handlers.get(frame.streamId)?.(frame)
}
})
socket.on('message', (raw, binary) =>
channel.handleRawMessage(binary ? new Uint8Array(raw as Buffer) : raw.toString())
)
socket.on('error', (error) => errors.push(error))
})
const address = server.address()
if (!address || typeof address === 'string') {
throw new Error('Missing encrypted listener port')
}
phone = new WebSocket(`ws://127.0.0.1:${address.port}`, { perMessageDeflate: false })
await inputProofDeadline(
new Promise<void>((resolve, reject) => {
phone!.once('open', resolve)
phone!.once('error', reject)
}),
'encrypted client connection'
)
let authenticated!: () => void
const ready = new Promise<void>((resolve) => {
authenticated = resolve
})
let id = 0
const streams = new MobileRelayRpcStreams({
nextId: () => String(++id),
waitForConnected: () => ready,
sendFrame: (request) => mobile!.sendText(JSON.stringify(request)),
sendBinary: (bytes) => mobile!.sendBinary(bytes)
})
clearStreams = () => streams.clear()
mobile = new MobileE2EEV2PhysicalChannel({
session: MobileE2EEV2ClientSession.create({
desktopPublicKeyB64: Buffer.from(keys.publicKey).toString('base64'),
transport: 'relay',
relayHostId: 'AbCdEf0123_-xyZ9'
}),
socket: phone,
deviceToken: 'test-device',
decodeBinary: async (raw) => (raw instanceof Uint8Array ? raw : null),
onAuthenticated: authenticated,
onText: (text) => {
streams.handleResponse(JSON.parse(text))
},
onBinary: (bytes) => streams.handleBinary(bytes),
onError: (error) => errors.push(error)
})
phone.on('message', (raw, binary) => {
void mobile!.handleMessage(binary ? new Uint8Array(raw as Buffer) : raw.toString())
})
mobile.start()
await inputProofDeadline(ready, 'real E2EE authentication')
let subscribed!: () => void
const subscriptionReady = new Promise<void>((resolve) => {
subscribed = resolve
})
disposeSubscription = streams.subscribe(
'terminal.subscribe',
{
terminal: 'terminal-1',
client: { id: 'phone', type: 'mobile' },
capabilities: { terminalBinaryStream: 1 }
},
(result) => {
if ((result as { type?: string }).type === 'subscribed') {
subscribed()
}
}
)
await inputProofDeadline(subscriptionReady, 'ordered-input negotiation')
expect(streams.supportsTerminalStreamInput('terminal-1')).toBe(true)
delayReplies = true
const started = performance.now()
const sender = (terminal: string, text: string) =>
streams.sendTerminalStreamInput(terminal, text)!
const sends: Promise<boolean>[] = []
for (const text of payloads) {
sends.push(
queueTerminalLiveMirrorSend(pending, 'terminal-1', text, sender, { pipeline }).then(
(accepted) => {
bytesBeforeFirstReceipt ??= rig.bytes()
receipts++
return accepted
}
)
)
// Real-clock typing cadence, not a lifecycle readiness delay.
await new Promise<void>((resolve) => setTimeout(resolve, 50))
}
expect(receipts).toBe(0)
expect(
await inputProofDeadline(waitForTerminalLivePendingFlush(pending), 'prefix receipts')
).toBe(true)
expect(await Promise.all(sends)).toEqual(payloads.map(() => true))
expect(bytesBeforeFirstReceipt).toEqual(
Buffer.from(pipeline ? payloads.join('') : payloads[0]!)
)
expect(await inputProofDeadline(sender('terminal-1', suffix), 'control receipt')).toBe(true)
await inputProofDeadline(rig.inputDelivered, 'real PTY bytes')
expect(rig.bytes()).toEqual(expected)
expect(errors).toEqual([])
console.log(
'[ordered-input-pty-proof]',
JSON.stringify({
payloads: payloads.length,
bytes: expected.length,
receiptDelayMs,
pipeline,
bytesBeforeFirstReceipt: bytesBeforeFirstReceipt!.length,
firstPtyInputMs: Math.round(rig.firstInputAt()! - started),
allReceiptsMs: Math.round(performance.now() - started)
})
)
disposeSubscription()
disposeSubscription = undefined
streams.clear()
} finally {
disposeSubscription?.()
clearStreams?.()
abort.abort()
for (const timer of delayedSends) {
clearTimeout(timer)
}
mobile?.dispose()
phone?.terminate()
for (const channel of sessions) {
channel.destroy()
}
for (const socket of server.clients) {
socket.terminate()
}
await inputProofDeadline(Promise.allSettled(dispatches), 'host subscription cleanup')
await inputProofDeadline(
new Promise<void>((resolve) => server.close(() => resolve())),
'encrypted server close'
)
await rig.close()
}
},
20_000
)
mobile/src/transport/terminal-ordered-input-routing.test.ts
+120
View File
@@ -0,0 +1,120 @@
import { describe, expect, it, vi } from 'vitest'
import { RpcClientStreamRegistry } from './rpc-client-stream-registry'
import { MobileRelayRpcStreams } from './mobile-relay-rpc-streams'
import { encodeTerminalStreamFrame, TerminalStreamOpcode } from './terminal-stream-protocol'
import { TERMINAL_ORDERED_INPUT_CAPABILITY } from '../../../src/shared/terminal-ordered-input'
import type { RpcResponse } from './types'
describe.each(['direct', 'relay'] as const)('%s ordered input routing', (path) => {
function setup() {
const sent: { id: string; params?: unknown }[] = []
const send = (request: unknown) => {
sent.push(request as { id: string; params?: unknown })
return true
}
const binary = vi.fn((_bytes: Uint8Array) => true)
let next = 0
const options = { nextId: () => `${++next}`, sendBinary: binary }
const registry =
path === 'direct'
? new RpcClientStreamRegistry({
...options,
deviceToken: 'device',
getState: () => 'connected',
sendEncrypted: send
})
: new MobileRelayRpcStreams({
...options,
sendFrame: send,
waitForConnected: async () => {}
})
return { registry, sent, binary }
}
it('advertises, routes receipts, and cancels pending input when unsubscribed', async () => {
const { registry, sent, binary } = setup()
const dispose = registry.subscribe('terminal.subscribe', { terminal: 't' }, vi.fn())
await vi.waitFor(() => expect(sent).toHaveLength(1))
expect(sent[0]?.params).toMatchObject({ capabilities: { orderedInput: 1 } })
const response: RpcResponse = {
id: sent[0]!.id,
ok: true,
streaming: true,
result: {
type: 'subscribed',
streamId: 7,
capabilities: { orderedInput: TERMINAL_ORDERED_INPUT_CAPABILITY }
},
_meta: { runtimeId: 'r' }
}
registry.handleResponse(response)
expect(registry.supportsTerminalStreamInput('t')).toBe(true)
const first = registry.sendTerminalStreamInput('t', 'a')!
expect(binary).toHaveBeenCalledOnce()
registry.handleBinary(
encodeTerminalStreamFrame({
opcode: TerminalStreamOpcode.Metadata,
streamId: 7,
seq: 999,
payload: new TextEncoder().encode(
JSON.stringify({ inputReceipt: { sequence: 1, outcome: 'accepted' } })
)
})
)
expect(await first).toBe(true)
const second = registry.sendTerminalStreamInput('t', 'b')!
registry.cancelTerminalStreamInput('t')
expect(sent).toContainEqual(expect.objectContaining({ method: 'terminal.unsubscribe' }))
dispose()
expect(await second).toBe(false)
expect(registry.getTerminalStreamInputFailure('t')).toEqual({
outcome: 'unknown',
reason: 'cancelled'
})
expect(registry.supportsTerminalStreamInput('t')).toBe(true)
expect(await registry.sendTerminalStreamInput('t', '\r')).toBe(false)
})
it('preserves old-host fallback without sending the new opcode', async () => {
const { registry, sent, binary } = setup()
const dispose = registry.subscribe('terminal.subscribe', { terminal: 't' }, vi.fn())
await vi.waitFor(() => expect(sent).toHaveLength(1))
registry.handleResponse({
id: sent[0]!.id,
ok: true,
streaming: true,
result: { type: 'subscribed', streamId: 7 },
_meta: { runtimeId: 'r' }
})
expect(registry.sendTerminalStreamInput('t', 'a')).toBe(null)
expect(registry.supportsTerminalStreamInput('t')).toBe(false)
expect(binary).not.toHaveBeenCalled()
dispose()
})
it('settles admitted input on physical connection loss', async () => {
const { registry, sent } = setup()
registry.subscribe('terminal.subscribe', { terminal: 't' }, vi.fn())
await vi.waitFor(() => expect(sent).toHaveLength(1))
registry.handleResponse({
id: sent[0]!.id,
ok: true,
streaming: true,
result: {
type: 'subscribed',
streamId: 7,
capabilities: { orderedInput: TERMINAL_ORDERED_INPUT_CAPABILITY }
},
_meta: { runtimeId: 'r' }
})
const input = registry.sendTerminalStreamInput('t', 'a')!
if (registry instanceof RpcClientStreamRegistry) {
registry.markForReplay()
} else {
registry.clear()
}
expect(await input).toBe(false)
expect(registry.supportsTerminalStreamInput('t')).toBe(true)
expect(await registry.sendTerminalStreamInput('t', '\r')).toBe(false)
})
})
mobile/src/transport/terminal-ordered-input.test.ts
+196
View File
@@ -0,0 +1,196 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import { TerminalOrderedInput, advertiseTerminalOrderedInput } from './terminal-ordered-input'
import { decodeTerminalStreamFrame, TerminalStreamOpcode } from './terminal-stream-protocol'
import { TERMINAL_ORDERED_INPUT_CAPABILITY as limits } from '../../../src/shared/terminal-ordered-input'
function setup(overrides = {}) {
const binary = vi.fn((_bytes: Uint8Array) => true)
const input = new TerminalOrderedInput(binary)
input.register(
'request',
advertiseTerminalOrderedInput('terminal.subscribe', { terminal: 't' }),
{ type: 'subscribed', streamId: 8, capabilities: { orderedInput: { ...limits, ...overrides } } }
)
const receipt = (sequence: number, outcome: string) =>
input.handle(8, { type: 'metadata', inputReceipt: { sequence, outcome } })
return { input, binary, receipt }
}
afterEach(() => vi.useRealTimers())
describe('negotiated terminal input receipts', () => {
it.each([
[1, 'rejected'],
[2, 'unknown']
] as const)('classifies refusal of pending sequence %s as %s', async (sequence, outcome) => {
const { input, receipt } = setup()
const first = input.send('t', 'a')!
const second = input.send('t', 'b')!
receipt(sequence, 'rejected')
expect(await first).toBe(false)
expect(await second).toBe(false)
expect(input.failure('t')?.outcome).toBe(outcome)
expect(input.recover('t')).toBe(false)
})
it('does not recover using a subscription registered before the failed admission', async () => {
const { input, receipt } = setup()
const params = advertiseTerminalOrderedInput('terminal.subscribe', { terminal: 't' })
input.register('candidate', params, {
type: 'subscribed',
streamId: 9,
capabilities: { orderedInput: limits }
})
const pending = input.send('t', 'x')!
input.handle(9, { type: 'metadata', inputReceipt: { sequence: 1, outcome: 'unknown' } })
expect(await pending).toBe(false)
expect(input.recover('t')).toBe(false)
receipt(1, 'accepted')
expect(input.failure('t')?.outcome).toBe('unknown')
})
it('releases clean terminal records across 2048 subscribe/unsubscribe cycles', async () => {
const input = new TerminalOrderedInput(() => true)
for (let i = 1; i <= 2048; i += 1) {
const terminal = `t-${i}`
const request = `request-${i}`
input.register(request, advertiseTerminalOrderedInput('terminal.subscribe', { terminal }), {
type: 'subscribed',
streamId: i,
capabilities: { orderedInput: limits }
})
if (i % 2 === 0) {
const accepted = input.send(terminal, 'a')!
input.handle(i, { type: 'metadata', inputReceipt: { sequence: 1, outcome: 'accepted' } })
expect(await accepted).toBe(true)
}
if (i % 3 === 0) {
input.clear()
} else {
input.reset(request)
}
expect(input.supports(terminal)).toBe(false)
}
const retained = input as unknown as {
streams: Map<string, unknown>
byTerminal: Map<string, unknown>
}
expect(retained.streams.size).toBe(0)
expect(retained.byTerminal.size).toBe(0)
})
it('waits for accepted receipts and sequences input independently of output metadata', async () => {
const { input, binary, receipt } = setup()
let settled = false
const first = input.send('t', '한')!
void first.then(() => {
settled = true
})
const second = input.send('t', '\r')!
await Promise.resolve()
expect(settled).toBe(false)
expect(binary.mock.calls.map(([bytes]) => decodeTerminalStreamFrame(bytes))).toMatchObject([
{ opcode: TerminalStreamOpcode.Input, streamId: 8, seq: 1 },
{ opcode: TerminalStreamOpcode.Input, streamId: 8, seq: 2 }
])
input.handle(8, { type: 'metadata', seq: 200 })
expect(settled).toBe(false)
receipt(1, 'accepted')
receipt(2, 'accepted')
expect(await first).toBe(true)
expect(await second).toBe(true)
})
it.each(['rejected', 'unknown'])('latches %s without replay or later sends', async (outcome) => {
const { input, binary, receipt } = setup()
const first = input.send('t', 'text')!
const second = input.send('t', '\r')!
receipt(1, outcome)
expect(await first).toBe(false)
expect(await second).toBe(false)
expect(await input.send('t', 'later')).toBe(false)
expect(input.supports('t')).toBe(true)
expect(binary).toHaveBeenCalledTimes(2)
})
it.each([{ maxPendingFrames: 1 }, { maxPendingBytes: 1 }])(
'bounds admitted outstanding input %j',
async (limit) => {
const { input, binary } = setup(limit)
const first = input.send('t', 'a')!
expect(await input.send('t', 'b')).toBe(false)
expect(await first).toBe(false)
expect(binary).toHaveBeenCalledTimes(1)
}
)
it('checks UTF-8 frame bytes and latches local refusal', async () => {
const { input, binary } = setup({ maxFrameBytes: 2 })
expect(await input.send('t', '한')).toBe(false)
expect(await input.send('t', 'a')).toBe(false)
expect(binary).not.toHaveBeenCalled()
})
it('settles and releases receipts on unsubscribe and ignores old receipts', async () => {
const { input, receipt } = setup()
const first = input.send('t', 'a')!
input.reset('request')
expect(await first).toBe(false)
receipt(1, 'accepted')
expect(await input.send('t', 'b')).toBe(false)
expect(input.supports('t')).toBe(true)
})
it('times out missing receipts without permitting further input', async () => {
vi.useFakeTimers()
const { input, binary } = setup()
const first = input.send('t', 'a')!
await vi.advanceTimersByTimeAsync(30_000)
expect(await first).toBe(false)
expect(await input.send('t', 'b')).toBe(false)
expect(binary).toHaveBeenCalledTimes(1)
expect(vi.getTimerCount()).toBe(0)
})
it('returns fallback only when no compatible host echo exists', () => {
const input = new TerminalOrderedInput(vi.fn())
const params = advertiseTerminalOrderedInput('terminal.subscribe', { terminal: 't' })
input.register('old', params, { type: 'subscribed', streamId: 1 })
expect(input.send('t', 'a')).toBe(null)
expect(input.supports('t')).toBe(false)
input.register('future', params, {
type: 'subscribed',
streamId: 2,
capabilities: { orderedInput: { ...limits, version: 2 } }
})
expect(input.send('t', 'a')).toBe(null)
})
it('keeps uncertainty latched across an old-host reconnect and recovers only after fresh negotiation', async () => {
const { input, receipt } = setup()
const first = input.send('t', 'prefix')!
input.clear()
expect(await first).toBe(false)
const params = advertiseTerminalOrderedInput('terminal.subscribe', { terminal: 't' })
input.register('old', params, { type: 'subscribed', streamId: 9 })
expect(input.recover('t')).toBe(false)
expect(input.supports('t')).toBe(true)
expect(await input.send('t', '\r')).toBe(false)
input.register('fresh', params, {
type: 'subscribed',
streamId: 10,
capabilities: { orderedInput: limits }
})
expect(await input.send('t', 'blocked')).toBe(false)
expect(input.recover('t')).toBe(true)
const fresh = input.send('t', 'fresh')!
receipt(1, 'accepted')
input.handle(10, { type: 'metadata', inputReceipt: { sequence: 1, outcome: 'accepted' } })
expect(await fresh).toBe(true)
})
it('latches synchronous socket failure and disposes receipt timers', async () => {
vi.useFakeTimers()
const { input, binary } = setup()
binary.mockReturnValue(false)
expect(await input.send('t', 'a')).toBe(false)
expect(await input.send('t', 'b')).toBe(false)
expect(vi.getTimerCount()).toBe(0)
})
})
mobile/src/transport/terminal-ordered-input.ts
+307
View File
@@ -0,0 +1,307 @@
import { encodeTerminalStreamFrame, TerminalStreamOpcode } from './terminal-stream-protocol'
import { TERMINAL_ORDERED_INPUT_CAPABILITY } from '../../../src/shared/terminal-ordered-input'
import type { TerminalStreamInputFailure } from './terminal-stream-input-failure'
type Limits = {
version: 1
maxFrameBytes: number
maxPendingBytes: number
maxPendingFrames: number
}
type Pending = {
bytes: number
resolve: (accepted: boolean) => void
timer: ReturnType<typeof setTimeout>
}
type InputStream = {
terminal: string
streamId: number
limits: Limits
sequence: number
failed: boolean
failure: TerminalStreamInputFailure | null
registration: number
failureRegistration: number
pendingBytes: number
pending: Map<number, Pending>
}
export function advertiseTerminalOrderedInput(method: string, params: unknown): unknown {
if (method !== 'terminal.subscribe' || !params || typeof params !== 'object') {
return params
}
const existing = (params as { capabilities?: unknown }).capabilities
return {
...params,
capabilities: { ...(existing && typeof existing === 'object' ? existing : {}), orderedInput: 1 }
}
}
export class TerminalOrderedInput {
private readonly streams = new Map<string, InputStream>()
private readonly byTerminal = new Map<string, InputStream>()
private registration = 0
constructor(private readonly sendBinary: (bytes: Uint8Array) => boolean) {}
supports(terminal: string): boolean {
return this.byTerminal.has(terminal)
}
failure(terminal: string): TerminalStreamInputFailure | null {
return this.byTerminal.get(terminal)?.failure ?? null
}
recover(terminal: string): boolean {
const failed = this.byTerminal.get(terminal)
if (!failed?.failed) {
return !!failed && failed.sequence === 0
}
const fresh = [...this.streams.values()].findLast(
(stream) =>
stream.terminal === terminal &&
!stream.failed &&
stream.sequence === 0 &&
stream.registration > failed.failureRegistration
)
if (!fresh) {
return false
}
this.byTerminal.set(terminal, fresh)
return true
}
cancel(terminal: string): string[] {
const requests: string[] = []
for (const [request, stream] of this.streams) {
if (stream.terminal !== terminal) {
continue
}
if (stream.pending.size > 0) {
this.fail(stream, { outcome: 'unknown', reason: 'cancelled' })
}
requests.push(request)
}
return requests
}
register(requestId: string, params: unknown, result: unknown): void {
if (!params || typeof params !== 'object' || !result || typeof result !== 'object') {
return
}
const { terminal, capabilities: offered } = params as {
terminal?: unknown
capabilities?: { orderedInput?: unknown }
}
if (offered?.orderedInput !== 1) {
return
}
const { type, streamId, capabilities } = result as {
type?: unknown
streamId?: unknown
capabilities?: { orderedInput?: unknown }
}
if (
type !== 'subscribed' ||
typeof terminal !== 'string' ||
!Number.isInteger(streamId) ||
Number(streamId) <= 0 ||
Number(streamId) > 0xffffffff
) {
return
}
const limits = parseLimits(capabilities?.orderedInput)
if (!limits || this.streams.has(requestId)) {
return
}
const previous = this.byTerminal.get(terminal)
if (previous && previous.pending.size > 0) {
this.fail(previous, { outcome: 'unknown', reason: 'subscription_replaced' })
}
const stream: InputStream = {
terminal,
streamId: Number(streamId),
limits,
sequence: 0,
failed: false,
failure: null,
registration: ++this.registration,
failureRegistration: 0,
pendingBytes: 0,
pending: new Map()
}
this.streams.set(requestId, stream)
if (!this.byTerminal.get(terminal)?.failed) {
this.byTerminal.set(terminal, stream)
}
}
send(terminal: string, text: string): Promise<boolean> | null {
const stream = this.byTerminal.get(terminal)
if (!stream) {
return null
}
if (stream.failed) {
return Promise.resolve(false)
}
// Check code units before allocating a potentially oversized encoded copy.
if (text.length === 0 || text.length > stream.limits.maxFrameBytes) {
this.fail(stream, {
outcome: stream.pending.size > 0 ? 'unknown' : 'rejected',
reason: 'too_large'
})
return Promise.resolve(false)
}
const payload = new TextEncoder().encode(text)
if (
payload.length > stream.limits.maxFrameBytes ||
stream.pendingBytes + payload.length > stream.limits.maxPendingBytes ||
stream.pending.size >= stream.limits.maxPendingFrames ||
stream.sequence >= Number.MAX_SAFE_INTEGER
) {
this.fail(stream, {
outcome: stream.pending.size > 0 ? 'unknown' : 'rejected',
reason: 'queue_full'
})
return Promise.resolve(false)
}
const sequence = ++stream.sequence
return new Promise<boolean>((resolve) => {
const timer = setTimeout(
() => this.fail(stream, { outcome: 'unknown', reason: 'receipt_timeout' }),
30_000
)
stream.pending.set(sequence, { bytes: payload.length, resolve, timer })
stream.pendingBytes += payload.length
try {
if (
!this.sendBinary(
encodeTerminalStreamFrame({
opcode: TerminalStreamOpcode.Input,
streamId: stream.streamId,
seq: sequence,
payload
})
)
) {
this.fail(stream, { outcome: 'unknown', reason: 'connection_interrupted' })
}
} catch {
this.fail(stream, { outcome: 'unknown', reason: 'connection_interrupted' })
}
})
}
handle(streamId: number, result: unknown): void {
if (!result || typeof result !== 'object') {
return
}
const metadata = result as { type?: unknown; inputReceipt?: unknown }
const streams = [...this.streams.values()].filter((stream) => stream.streamId === streamId)
for (const stream of streams) {
if (metadata.type === 'error') {
this.fail(stream)
continue
}
if (metadata.type !== 'metadata' || metadata.inputReceipt === undefined) {
continue
}
const receipt = metadata.inputReceipt as {
sequence?: unknown
outcome?: unknown
reason?: unknown
} | null
if (!receipt || !Number.isSafeInteger(receipt.sequence) || Number(receipt.sequence) <= 0) {
this.fail(stream)
continue
}
const sequence = Number(receipt.sequence)
if (sequence > stream.sequence) {
this.fail(stream)
continue
}
const pending = stream.pending.get(sequence)
if (!pending) {
continue
}
if (receipt.outcome !== 'accepted') {
this.fail(stream, {
outcome:
receipt.outcome === 'rejected' && stream.pending.keys().next().value === sequence
? 'rejected'
: 'unknown',
reason: typeof receipt.reason === 'string' ? receipt.reason.slice(0, 128) : 'write_failed'
})
continue
}
clearTimeout(pending.timer)
stream.pending.delete(sequence)
stream.pendingBytes -= pending.bytes
pending.resolve(true)
}
}
reset(requestId: string): void {
const stream = this.streams.get(requestId)
if (stream) {
const retainFailure = stream.pending.size > 0 || stream.failed
this.fail(stream)
if (!retainFailure && this.byTerminal.get(stream.terminal) === stream) {
this.byTerminal.delete(stream.terminal)
}
}
this.streams.delete(requestId)
}
clear(): void {
for (const requestId of this.streams.keys()) {
this.reset(requestId)
}
}
private fail(
stream: InputStream,
failure: TerminalStreamInputFailure = {
outcome: stream.pending.size > 0 ? 'unknown' : 'rejected',
reason: 'stream_closed'
}
): void {
if (!stream.failed) {
stream.failure = failure
stream.failureRegistration = this.registration
}
stream.failed = true
for (const pending of stream.pending.values()) {
clearTimeout(pending.timer)
pending.resolve(false)
}
stream.pending.clear()
stream.pendingBytes = 0
}
}
function parseLimits(value: unknown): Limits | null {
if (!value || typeof value !== 'object') {
return null
}
const limits = value as Limits
if (limits.version !== 1) {
return null
}
for (const key of ['maxFrameBytes', 'maxPendingBytes', 'maxPendingFrames'] as const) {
if (!Number.isSafeInteger(limits[key]) || limits[key] <= 0) {
return null
}
}
return {
version: 1,
maxFrameBytes: Math.min(limits.maxFrameBytes, TERMINAL_ORDERED_INPUT_CAPABILITY.maxFrameBytes),
maxPendingBytes: Math.min(
limits.maxPendingBytes,
TERMINAL_ORDERED_INPUT_CAPABILITY.maxPendingBytes
),
maxPendingFrames: Math.min(
limits.maxPendingFrames,
TERMINAL_ORDERED_INPUT_CAPABILITY.maxPendingFrames
)
}
}
mobile/src/transport/terminal-stream-input-failure.ts
+4
View File
@@ -0,0 +1,4 @@
export type TerminalStreamInputFailure = {
outcome: 'rejected' | 'unknown'
reason: string
}
mobile/src/transport/terminal-stream-protocol.ts
+2
View File
@@ -9,6 +9,7 @@ export enum TerminalStreamOpcode {
SnapshotEnd = 4,
Resized = 5,
Error = 6,
Input = 7,
Metadata = 12
}
@@ -76,6 +77,7 @@ function isTerminalStreamOpcode(value: number): value is TerminalStreamOpcode {
value === TerminalStreamOpcode.SnapshotEnd ||
value === TerminalStreamOpcode.Resized ||
value === TerminalStreamOpcode.Error ||
value === TerminalStreamOpcode.Input ||
value === TerminalStreamOpcode.Metadata
)
}