diff --git a/mobile/src/session/MobileSessionCommandDock.tsx b/mobile/src/session/MobileSessionCommandDock.tsx
index 7c7f772e7d6..6bc2e60221e 100644
--- a/mobile/src/session/MobileSessionCommandDock.tsx
+++ b/mobile/src/session/MobileSessionCommandDock.tsx
@@ -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 && (
+
+ )}
{/* Accessory keys */}
{/* Why: fixed keyboard escape hatch; outside ScrollView + shortcut path so it can't scroll away or be hidden (#5106). */}
diff --git a/mobile/src/session/MobileTerminalInputRecovery.tsx b/mobile/src/session/MobileTerminalInputRecovery.tsx
new file mode 100644
index 00000000000..75a6bebd3d6
--- /dev/null
+++ b/mobile/src/session/MobileTerminalInputRecovery.tsx
@@ -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 (
+
+ Terminal input paused
+
+ {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.
+
+ {recoveryUnavailable && (
+
+ Recovery needs a new connection to a host that supports ordered input. Update the host if
+ needed.
+
+ )}
+
+ Reconnect input
+
+
+ )
+}
+
+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' }
+})
diff --git a/mobile/src/session/MobileTerminalLiveInputBar.tsx b/mobile/src/session/MobileTerminalLiveInputBar.tsx
index cda36e7dec1..c4921ff9464 100644
--- a/mobile/src/session/MobileTerminalLiveInputBar.tsx
+++ b/mobile/src/session/MobileTerminalLiveInputBar.tsx
@@ -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 (
{
const slice = sourceSlice(
sendActionsSource,
'async function handleAccessoryKey(',
- 'const sendLiveTerminalInput = useCallback('
+ 'const sendLiveTerminalInput = useCallback('
)
expect(slice).not.toContain('dismissKeyboardAfterAgentSend')
})
diff --git a/mobile/src/session/mobile-session-route-parity.test.ts b/mobile/src/session/mobile-session-route-parity.test.ts
index cf0c450ba4b..3d69db9bae9 100644
--- a/mobile/src/session/mobile-session-route-parity.test.ts
+++ b/mobile/src/session/mobile-session-route-parity.test.ts
@@ -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)
diff --git a/mobile/src/session/use-mobile-session-terminal-runtime.ts b/mobile/src/session/use-mobile-session-terminal-runtime.ts
index 47886acd8c6..ad72c7ddaad 100644
--- a/mobile/src/session/use-mobile-session-terminal-runtime.ts
+++ b/mobile/src/session/use-mobile-session-terminal-runtime.ts
@@ -58,6 +58,7 @@ export function useMobileSessionTerminalRuntime(scope: MobileSessionScreenStateM
const commandInputRef = useRef(null)
const liveInputFocusTimerRef = useRef | null>(null)
const sendLiveTerminalInputRef = useRef(async () => false)
+ const terminalInputSubscribedRef = useRef<(handle: string) => void>(() => {})
const sessionTabActionSheetKeyboardHideSubRef = useRef | null>(null)
@@ -172,6 +173,7 @@ export function useMobileSessionTerminalRuntime(scope: MobileSessionScreenStateM
liveInputFocusTimerRef,
sendLiveTerminalInputRef,
sessionTabActionSheetKeyboardHideSubRef,
+ terminalInputSubscribedRef,
sessionTabActionSheetRequestSeqRef,
dictationRouteContextRef,
terminalUnsubsRef,
diff --git a/mobile/src/session/use-mobile-session-terminal-send-actions.ts b/mobile/src/session/use-mobile-session-terminal-send-actions.ts
index 6909f71ca63..9ff034f1c22 100644
--- a/mobile/src/session/use-mobile-session-terminal-send-actions.ts
+++ b/mobile/src/session/use-mobile-session-terminal-send-actions.ts
@@ -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(
async (handle: string, bytes: string): Promise => {
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,
diff --git a/mobile/src/session/use-mobile-session-terminal-subscription.ts b/mobile/src/session/use-mobile-session-terminal-subscription.ts
index 333d472a30e..ec15d5bfe6e 100644
--- a/mobile/src/session/use-mobile-session-terminal-subscription.ts
+++ b/mobile/src/session/use-mobile-session-terminal-subscription.ts
@@ -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.
diff --git a/mobile/src/session/use-terminal-input-recovery.test.ts b/mobile/src/session/use-terminal-input-recovery.test.ts
new file mode 100644
index 00000000000..ee75288a983
--- /dev/null
+++ b/mobile/src/session/use-terminal-input-recovery.test.ts
@@ -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
+ function Probe() {
+ value = useTerminalInputRecovery(options)
+ return null
+ }
+ let renderer!: ReturnType
+ 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()
+ }
+ }
+ )
+})
diff --git a/mobile/src/session/use-terminal-input-recovery.ts b/mobile/src/session/use-terminal-input-recovery.ts
new file mode 100644
index 00000000000..321a85444e3
--- /dev/null
+++ b/mobile/src/session/use-terminal-input-recovery.ts
@@ -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
+ clientRef: RefObject
+ 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
+ }
+}
diff --git a/mobile/src/terminal/terminal-live-accessory-raw-send.test.ts b/mobile/src/terminal/terminal-live-accessory-raw-send.test.ts
index 4ecab7dd10c..7f992454b82 100644
--- a/mobile/src/terminal/terminal-live-accessory-raw-send.test.ts
+++ b/mobile/src/terminal/terminal-live-accessory-raw-send.test.ts
@@ -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()
diff --git a/mobile/src/terminal/terminal-live-accessory-raw-send.ts b/mobile/src/terminal/terminal-live-accessory-raw-send.ts
index 3824d750792..cf904bbde1d 100644
--- a/mobile/src/terminal/terminal-live-accessory-raw-send.ts
+++ b/mobile/src/terminal/terminal-live-accessory-raw-send.ts
@@ -5,7 +5,7 @@ import type { RpcClient } from '../transport/rpc-client'
import type { ConnectionState } from '../transport/types'
type TerminalLiveAccessoryRawSendArgs = {
- readonly client: Pick | null
+ readonly client: Pick | null
readonly targetHandle: string
readonly activeHandle: string | null
readonly activeSessionTabType: string | null
@@ -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)
}
diff --git a/mobile/src/terminal/terminal-live-input-sender.ts b/mobile/src/terminal/terminal-live-input-sender.ts
index c3538d1f723..77be3ca714c 100644
--- a/mobile/src/terminal/terminal-live-input-sender.ts
+++ b/mobile/src/terminal/terminal-live-input-sender.ts
@@ -1 +1,5 @@
-export type TerminalLiveInputSender = (handle: string, bytes: string) => Promise
+export type TerminalLiveInputSender = {
+ (handle: string, bytes: string): Promise
+ cancelPending?: (handle: string) => void
+ supportsPipeline?: (handle: string) => boolean
+}
diff --git a/mobile/src/terminal/terminal-live-pending-flush-state.test.ts b/mobile/src/terminal/terminal-live-pending-flush-state.test.ts
index a77d4315c29..b7834e2302f 100644
--- a/mobile/src/terminal/terminal-live-pending-flush-state.test.ts
+++ b/mobile/src/terminal/terminal-live-pending-flush-state.test.ts
@@ -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 () => {
diff --git a/mobile/src/terminal/terminal-live-pending-flush-state.ts b/mobile/src/terminal/terminal-live-pending-flush-state.ts
index fe0b4927f5a..8ee0e6af3ba 100644
--- a/mobile/src/terminal/terminal-live-pending-flush-state.ts
+++ b/mobile/src/terminal/terminal-live-pending-flush-state.ts
@@ -1,3 +1,5 @@
+import { TERMINAL_LIVE_INPUT_MAX_BYTES } from './terminal-live-input'
+
type TerminalLiveMirrorSender = (handle: string, payload: string) => Promise
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 | null
- activeRequests: TerminalLivePendingRequest[]
generation: number
pendingBatches: TerminalLivePendingBatch[]
+ activeBatches: Set
+ 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 {
- 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 {
- 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 {
+ 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((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((resolve) => {
+ state.finish = resolve
})
}
+ pumpMirrorSends(state)
return request
}
diff --git a/mobile/src/terminal/terminal-live-pipelined-flush.test.ts b/mobile/src/terminal/terminal-live-pipelined-flush.test.ts
new file mode 100644
index 00000000000..9a97d3366cf
--- /dev/null
+++ b/mobile/src/terminal/terminal-live-pipelined-flush.test.ts
@@ -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((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((resolve) => setTimeout(() => resolve(true), rtt))
+ }
+ const requests: Promise[] = []
+ 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)
+ }
+ )
+})
diff --git a/mobile/src/terminal/use-terminal-live-input-commit.test.ts b/mobile/src/terminal/use-terminal-live-input-commit.test.ts
index 616a7ef0fa7..6ce7af7676a 100644
--- a/mobile/src/terminal/use-terminal-live-input-commit.test.ts
+++ b/mobile/src/terminal/use-terminal-live-input-commit.test.ts
@@ -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' })
})
diff --git a/mobile/src/terminal/use-terminal-live-mirror-failure.test.ts b/mobile/src/terminal/use-terminal-live-mirror-failure.test.ts
new file mode 100644
index 00000000000..00b057d34bb
--- /dev/null
+++ b/mobile/src/terminal/use-terminal-live-mirror-failure.test.ts
@@ -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((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 = { current: sender }
+ let hook!: ReturnType>
+ 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((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)
+ })
+})
diff --git a/mobile/src/terminal/use-terminal-live-pending-input-flush.ts b/mobile/src/terminal/use-terminal-live-pending-input-flush.ts
index ff0c09c1c51..b9bf50b430f 100644
--- a/mobile/src/terminal/use-terminal-live-pending-input-flush.ts
+++ b/mobile/src/terminal/use-terminal-live-pending-input-flush.ts
@@ -58,6 +58,7 @@ export function useTerminalLivePendingInputFlush({
const liveInputComposingRef = useRef(undefined)
const sentLiveInputTextRef = useRef('')
const pendingLiveInputHandleRef = useRef(null)
+ const mirrorRevisionRef = useRef(0)
const runMirrorStepRef = useRef(async () => false)
const clearHeldCommitTimer = useCallback(() => {
@@ -67,20 +68,52 @@ export function useTerminalLivePendingInputFlush({
}
}, [])
- 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 => {
return waitForTerminalLivePendingFlush(pendingLiveInputFlushRef.current)
@@ -111,6 +144,8 @@ export function useTerminalLivePendingInputFlush({
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({
}
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({
}
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,
diff --git a/mobile/src/transport/direct-rpc-client.ts b/mobile/src/transport/direct-rpc-client.ts
index 16306f95cdd..9ecf9db2c06 100644
--- a/mobile/src/transport/direct-rpc-client.ts
+++ b/mobile/src/transport/direct-rpc-client.ts
@@ -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 | 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 {
diff --git a/mobile/src/transport/e2ee.ts b/mobile/src/transport/e2ee.ts
index 2732b3d615f..a6dbf6f346f 100644
--- a/mobile/src/transport/e2ee.ts
+++ b/mobile/src/transport/e2ee.ts
@@ -73,7 +73,7 @@ export function decrypt(encrypted: string, sharedKey: Uint8Array): string | null
return plaintext ? new TextDecoder().decode(plaintext) : null
}
-function encryptBytes(plaintext: Uint8Array, sharedKey: Uint8Array): Uint8Array {
+export function encryptBytes(plaintext: Uint8Array, sharedKey: Uint8Array): Uint8Array {
const nonce = u8(nacl.randomBytes(nacl.box.nonceLength))
const ciphertext = nacl.box.after(u8(plaintext), nonce, u8(sharedKey))
diff --git a/mobile/src/transport/logical-client-cutover.ts b/mobile/src/transport/logical-client-cutover.ts
new file mode 100644
index 00000000000..5b65652b831
--- /dev/null
+++ b/mobile/src/transport/logical-client-cutover.ts
@@ -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')
+ )
+}
diff --git a/mobile/src/transport/logical-terminal-stream-input.test.ts b/mobile/src/transport/logical-terminal-stream-input.test.ts
new file mode 100644
index 00000000000..8e01a2535a5
--- /dev/null
+++ b/mobile/src/transport/logical-terminal-stream-input.test.ts
@@ -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((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)
+ })
+})
diff --git a/mobile/src/transport/logical-terminal-stream-input.ts b/mobile/src/transport/logical-terminal-stream-input.ts
new file mode 100644
index 00000000000..ed9a38e6ad7
--- /dev/null
+++ b/mobile/src/transport/logical-terminal-stream-input.ts
@@ -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()
+ 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 | 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
+ }
+}
diff --git a/mobile/src/transport/mobile-relay-rpc-session.test.ts b/mobile/src/transport/mobile-relay-rpc-session.test.ts
index 5887dffc73d..7ebcd09d3ef 100644
--- a/mobile/src/transport/mobile-relay-rpc-session.test.ts
+++ b/mobile/src/transport/mobile-relay-rpc-session.test.ts
@@ -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()
diff --git a/mobile/src/transport/mobile-relay-rpc-session.ts b/mobile/src/transport/mobile-relay-rpc-session.ts
index 203a0329192..9ae0924dd50 100644
--- a/mobile/src/transport/mobile-relay-rpc-session.ts
+++ b/mobile/src/transport/mobile-relay-rpc-session.ts
@@ -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))
},
diff --git a/mobile/src/transport/mobile-relay-rpc-streams.ts b/mobile/src/transport/mobile-relay-rpc-streams.ts
index 2eacbd2168f..bd336bb835d 100644
--- a/mobile/src/transport/mobile-relay-rpc-streams.ts
+++ b/mobile/src/transport/mobile-relay-rpc-streams.ts
@@ -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
+ sendBinary?: (bytes: Uint8Array) => boolean
}
export class MobileRelayRpcStreams {
@@ -35,8 +37,26 @@ export class MobileRelayRpcStreams {
private readonly terminalListeners = new Map void>()
private readonly terminalSnapshots = new Map()
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 | 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
diff --git a/mobile/src/transport/rpc-client-request-tracker.ts b/mobile/src/transport/rpc-client-request-tracker.ts
index d96d1e97609..b679d2bc2d6 100644
--- a/mobile/src/transport/rpc-client-request-tracker.ts
+++ b/mobile/src/transport/rpc-client-request-tracker.ts
@@ -16,6 +16,7 @@ type RequestTrackerOptions = {
waitForConnected: (timeoutMs?: number) => Promise
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 {
return new Promise((resolve, reject) => {
+ this.options.validateRequest?.(method, params)
const id = this.options.nextId()
const timeout = setTimeout(() => {
this.pending.delete(id)
diff --git a/mobile/src/transport/rpc-client-socket-options.ts b/mobile/src/transport/rpc-client-socket-options.ts
new file mode 100644
index 00000000000..abb7ad48997
--- /dev/null
+++ b/mobile/src/transport/rpc-client-socket-options.ts
@@ -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
+}
diff --git a/mobile/src/transport/rpc-client-socket-send.ts b/mobile/src/transport/rpc-client-socket-send.ts
new file mode 100644
index 00000000000..6cedf810440
--- /dev/null
+++ b/mobile/src/transport/rpc-client-socket-send.ts
@@ -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
+ }
+}
diff --git a/mobile/src/transport/rpc-client-socket-session.ts b/mobile/src/transport/rpc-client-socket-session.ts
index bc8f9d13e2c..5f6cc1661e2 100644
--- a/mobile/src/transport/rpc-client-socket-session.ts
+++ b/mobile/src/transport/rpc-client-socket-session.ts
@@ -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)
diff --git a/mobile/src/transport/rpc-client-stream-options.ts b/mobile/src/transport/rpc-client-stream-options.ts
new file mode 100644
index 00000000000..9436715b097
--- /dev/null
+++ b/mobile/src/transport/rpc-client-stream-options.ts
@@ -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
+}
diff --git a/mobile/src/transport/rpc-client-stream-registry.ts b/mobile/src/transport/rpc-client-stream-registry.ts
index ac20acdecc0..a70a64f2d14 100644
--- a/mobile/src/transport/rpc-client-stream-registry.ts
+++ b/mobile/src/transport/rpc-client-stream-registry.ts
@@ -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()
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)
}
diff --git a/mobile/src/transport/rpc-client.ts b/mobile/src/transport/rpc-client.ts
index 10f586d2780..7a7c8a58481 100644
--- a/mobile/src/transport/rpc-client.ts
+++ b/mobile/src/transport/rpc-client.ts
@@ -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 | null
sendRequest: (
method: string,
params?: unknown,
diff --git a/mobile/src/transport/stable-logical-rpc-client.test.ts b/mobile/src/transport/stable-logical-rpc-client.test.ts
index faa236a88ca..7c23463b744 100644
--- a/mobile/src/transport/stable-logical-rpc-client.test.ts
+++ b/mobile/src/transport/stable-logical-rpc-client.test.ts
@@ -68,6 +68,43 @@ function deferred() {
}
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()
+ 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([]))
diff --git a/mobile/src/transport/stable-logical-rpc-client.ts b/mobile/src/transport/stable-logical-rpc-client.ts
index fb514128382..d0a440970e1 100644
--- a/mobile/src/transport/stable-logical-rpc-client.ts
+++ b/mobile/src/transport/stable-logical-rpc-client.ts
@@ -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()
- const pendingRequests = new Set()
+ const pendingRequests = new Set()
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
diff --git a/mobile/src/transport/terminal-input-request-fence.test.ts b/mobile/src/transport/terminal-input-request-fence.test.ts
new file mode 100644
index 00000000000..7a0b85f29d2
--- /dev/null
+++ b/mobile/src/transport/terminal-input-request-fence.test.ts
@@ -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)
+ })
+})
diff --git a/mobile/src/transport/terminal-input-request-fence.ts b/mobile/src/transport/terminal-input-request-fence.ts
new file mode 100644
index 00000000000..afdf2c3dc69
--- /dev/null
+++ b/mobile/src/transport/terminal-input-request-fence.ts
@@ -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.'
+ )
+ }
+}
diff --git a/mobile/src/transport/terminal-ordered-input-pty.integration.test.ts b/mobile/src/transport/terminal-ordered-input-pty.integration.test.ts
new file mode 100644
index 00000000000..69b1ff4d5a4
--- /dev/null
+++ b/mobile/src/transport/terminal-ordered-input-pty.integration.test.ts
@@ -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 void>()
+ const abort = new AbortController()
+ const errors: unknown[] = []
+ const sessions: E2EEChannel[] = []
+ const dispatches: Promise[] = []
+ const delayedSends = new Set>()
+ 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((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((resolve, reject) => {
+ phone!.once('open', resolve)
+ phone!.once('error', reject)
+ }),
+ 'encrypted client connection'
+ )
+ let authenticated!: () => void
+ const ready = new Promise((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((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[] = []
+ 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((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((resolve) => server.close(() => resolve())),
+ 'encrypted server close'
+ )
+ await rig.close()
+ }
+ },
+ 20_000
+)
diff --git a/mobile/src/transport/terminal-ordered-input-routing.test.ts b/mobile/src/transport/terminal-ordered-input-routing.test.ts
new file mode 100644
index 00000000000..b7e9ece27b5
--- /dev/null
+++ b/mobile/src/transport/terminal-ordered-input-routing.test.ts
@@ -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)
+ })
+})
diff --git a/mobile/src/transport/terminal-ordered-input.test.ts b/mobile/src/transport/terminal-ordered-input.test.ts
new file mode 100644
index 00000000000..cb30115a7cc
--- /dev/null
+++ b/mobile/src/transport/terminal-ordered-input.test.ts
@@ -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
+ byTerminal: Map
+ }
+ 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)
+ })
+})
diff --git a/mobile/src/transport/terminal-ordered-input.ts b/mobile/src/transport/terminal-ordered-input.ts
new file mode 100644
index 00000000000..f097ffcfa3c
--- /dev/null
+++ b/mobile/src/transport/terminal-ordered-input.ts
@@ -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
+}
+type InputStream = {
+ terminal: string
+ streamId: number
+ limits: Limits
+ sequence: number
+ failed: boolean
+ failure: TerminalStreamInputFailure | null
+ registration: number
+ failureRegistration: number
+ pendingBytes: number
+ pending: Map
+}
+
+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()
+ private readonly byTerminal = new Map()
+ 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 | 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((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
+ )
+ }
+}
diff --git a/mobile/src/transport/terminal-stream-input-failure.ts b/mobile/src/transport/terminal-stream-input-failure.ts
new file mode 100644
index 00000000000..c7772361985
--- /dev/null
+++ b/mobile/src/transport/terminal-stream-input-failure.ts
@@ -0,0 +1,4 @@
+export type TerminalStreamInputFailure = {
+ outcome: 'rejected' | 'unknown'
+ reason: string
+}
diff --git a/mobile/src/transport/terminal-stream-protocol.ts b/mobile/src/transport/terminal-stream-protocol.ts
index 384d66996f1..ef4ea94f053 100644
--- a/mobile/src/transport/terminal-stream-protocol.ts
+++ b/mobile/src/transport/terminal-stream-protocol.ts
@@ -9,6 +9,7 @@ export enum TerminalStreamOpcode {
SnapshotEnd = 4,
Resized = 5,
Error = 6,
+ Input = 7,
Metadata = 12
}
@@ -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
)
}