fix(mobile): pipeline live terminal input instead of awaiting each RPC

Live keystrokes were stop-and-wait on terminal.send, so bursts still
paid RTT between the first write and the coalesced tail. Reuse the
existing subscribe Input frames so writes return after the socket send,
matching desktop multiplex input, and fall back to terminal.send when
the host has not advertised a binary stream id.

Fixes #12071

mobile/app/h/[hostId]/session/[worktreeId].tsx

@@ -125,7 +125,7 @@ import {
 import { useTerminalLiveInputFocus } from '../../../../src/terminal/use-terminal-live-input-focus'
 import { dismissTerminalKeyboard } from '../../../../src/terminal/terminal-keyboard-dismiss'
 import type { TerminalLiveInputSender } from '../../../../src/terminal/terminal-live-input-sender'
-import { isTerminalSendRpcAccepted } from '../../../../src/terminal/terminal-send-rpc-response'
+import { sendLiveTerminalInputBytes } from '../../../../src/terminal/send-live-terminal-input'
 import { sendMobileTerminalQueryReply } from '../../../../src/terminal/mobile-terminal-query-reply'
 import { TERMINAL_QUERY_REPLY_INPUT_RUNTIME_CAPABILITY } from '../../../../../src/shared/protocol-version'
 import { useTerminalLiveInputCommit } from '../../../../src/terminal/use-terminal-live-input-commit'
@@ -3024,30 +3024,17 @@ export default function SessionScreen() {
         showToast('Input too large (max 256 KiB)', 1500)
         return false
       }
-      const rpc = clientRef.current
-      // Why: callers suppress follow-up controls/toasts when this live send is stale.
-      if (
-        !rpc ||
-        connStateRef.current !== 'connected' ||
-        handle !== activeHandleRef.current ||
-        activeSessionTabTypeRef.current !== 'terminal'
-      ) {
-        return false
-      }
       // Why: live-mirror deltas queued behind a dying send drain into the connect
       // wait and replay stale bytes after reconnect (#6713's `YZZYecho …` corruption).
-      return rpc
-        .sendRequest(
-          'terminal.send',
-          buildTerminalSendParams({
-            terminal: handle,
-            text,
-            enter: false,
-            deviceToken: deviceTokenRef.current
-          }),
-          TERMINAL_INPUT_SEND_OPTIONS
-        )
-        .then(isTerminalSendRpcAccepted, () => false)
+      return sendLiveTerminalInputBytes({
+        rpc: clientRef.current,
+        handle,
+        bytes: text,
+        connected: connStateRef.current === 'connected',
+        activeHandle: activeHandleRef.current,
+        activeSessionTabType: activeSessionTabTypeRef.current,
+        deviceToken: deviceTokenRef.current
+      })
     },
     [showToast]
   )

mobile/src/terminal/send-live-terminal-input.test.ts

@@ -0,0 +1,148 @@
+import { describe, expect, it, vi } from 'vitest'
+import type { RpcResponse } from '../transport/types'
+import {
+  queueTerminalLiveMirrorSend,
+  createTerminalLivePendingFlushState
+} from './terminal-live-pending-flush-state'
+import { sendLiveTerminalInputBytes, type LiveTerminalInputRpc } from './send-live-terminal-input'
+
+function acceptedSend(): RpcResponse {
+  return {
+    id: 'rpc-1',
+    ok: true,
+    result: { send: { accepted: true } },
+    _meta: { runtimeId: 'runtime-1' }
+  }
+}
+
+function deferred<T>(): {
+  promise: Promise<T>
+  resolve: (value: T) => void
+} {
+  let resolve: (value: T) => void = () => {}
+  const promise = new Promise<T>((next) => {
+    resolve = next
+  })
+  return { promise, resolve }
+}
+
+describe('send live terminal input bytes', () => {
+  it('Given a held RPC response When later deltas queue Then later writes leave before the first response', async () => {
+    const streamWrites: string[] = []
+    const sendRequest = vi.fn(() => deferred<RpcResponse>().promise)
+    const rpc: LiveTerminalInputRpc = {
+      sendRequest,
+      sendTerminalInput: (_terminal, text) => {
+        streamWrites.push(text)
+        return 'sent'
+      }
+    }
+    const state = createTerminalLivePendingFlushState()
+    const sender = (handle: string, payload: string): Promise<boolean> =>
+      sendLiveTerminalInputBytes({
+        rpc,
+        handle,
+        bytes: payload,
+        connected: true,
+        activeHandle: handle,
+        activeSessionTabType: 'terminal',
+        deviceToken: 'phone-1'
+      })
+
+    const first = queueTerminalLiveMirrorSend(state, 'term-1', 'a', sender)
+    await Promise.resolve()
+    const second = queueTerminalLiveMirrorSend(state, 'term-1', 'b', sender)
+    const third = queueTerminalLiveMirrorSend(state, 'term-1', 'c', sender)
+    await Promise.resolve()
+
+    expect(streamWrites.join('')).toBe('abc')
+    expect(sendRequest).not.toHaveBeenCalled()
+    await expect(Promise.all([first, second, third])).resolves.toEqual([true, true, true])
+  })
+
+  it('Given a held RPC response When later deltas queue Then bytes stay in order and none drop', async () => {
+    const streamWrites: string[] = []
+    const rpc: LiveTerminalInputRpc = {
+      sendRequest: vi.fn(),
+      sendTerminalInput: (_terminal, text) => {
+        streamWrites.push(text)
+        return 'sent'
+      }
+    }
+    const state = createTerminalLivePendingFlushState()
+    const sender = (handle: string, payload: string): Promise<boolean> =>
+      sendLiveTerminalInputBytes({
+        rpc,
+        handle,
+        bytes: payload,
+        connected: true,
+        activeHandle: handle,
+        activeSessionTabType: 'terminal',
+        deviceToken: 'phone-1'
+      })
+
+    const first = queueTerminalLiveMirrorSend(state, 'term-1', 'h', sender)
+    await Promise.resolve()
+    const rest = 'ello world'
+      .split('')
+      .map((letter) => queueTerminalLiveMirrorSend(state, 'term-1', letter, sender))
+    await Promise.all([first, ...rest])
+
+    expect(streamWrites.join('')).toBe('hello world')
+  })
+
+  it('Given a failed stream write When input is live Then it does not fall back to terminal.send', async () => {
+    const sendRequest = vi.fn()
+    const rpc: LiveTerminalInputRpc = {
+      sendRequest,
+      sendTerminalInput: () => 'failed'
+    }
+
+    await expect(
+      sendLiveTerminalInputBytes({
+        rpc,
+        handle: 'term-1',
+        bytes: 'a',
+        connected: true,
+        activeHandle: 'term-1',
+        activeSessionTabType: 'terminal',
+        deviceToken: 'phone-1'
+      })
+    ).resolves.toBe(false)
+    expect(sendRequest).not.toHaveBeenCalled()
+  })
+
+  it('Given no stream input When RTT is held Then fallback still coalesces later bytes onto one follow-up RPC', async () => {
+    const firstRpc = deferred<RpcResponse>()
+    const sendRequest = vi.fn(() => firstRpc.promise)
+    const rpc: LiveTerminalInputRpc = { sendRequest }
+    const state = createTerminalLivePendingFlushState()
+    const sender = (handle: string, payload: string): Promise<boolean> =>
+      sendLiveTerminalInputBytes({
+        rpc,
+        handle,
+        bytes: payload,
+        connected: true,
+        activeHandle: handle,
+        activeSessionTabType: 'terminal',
+        deviceToken: 'phone-1'
+      })
+
+    const first = queueTerminalLiveMirrorSend(state, 'term-1', 'a', sender)
+    await Promise.resolve()
+    const second = queueTerminalLiveMirrorSend(state, 'term-1', 'b', sender)
+    const third = queueTerminalLiveMirrorSend(state, 'term-1', 'c', sender)
+    await Promise.resolve()
+
+    expect(sendRequest).toHaveBeenCalledTimes(1)
+    expect(sendRequest.mock.calls[0]?.[1]).toMatchObject({ text: 'a' })
+
+    const secondRpc = deferred<RpcResponse>()
+    sendRequest.mockImplementationOnce(() => secondRpc.promise)
+    firstRpc.resolve(acceptedSend())
+    await vi.waitFor(() => expect(sendRequest).toHaveBeenCalledTimes(2))
+    expect(sendRequest.mock.calls[1]?.[1]).toMatchObject({ text: 'bc' })
+    secondRpc.resolve(acceptedSend())
+    await expect(Promise.all([first, second, third])).resolves.toEqual([true, true, true])
+  })
+})

mobile/src/terminal/send-live-terminal-input.ts

@@ -0,0 +1,51 @@
+import type { RpcClient } from '../transport/rpc-client'
+import { isTerminalSendRpcAccepted } from './terminal-send-rpc-response'
+import { buildTerminalSendParams, TERMINAL_INPUT_SEND_OPTIONS } from './terminal-send-request'
+import { normalizeTerminalTextInput } from './terminal-text-input-normalization'
+
+export type LiveTerminalInputRpc = Pick<RpcClient, 'sendRequest'> &
+  Partial<Pick<RpcClient, 'sendTerminalInput'>>
+
+export async function sendLiveTerminalInputBytes(args: {
+  rpc: LiveTerminalInputRpc | null
+  handle: string
+  bytes: string
+  connected: boolean
+  activeHandle: string | null
+  activeSessionTabType: string | null
+  deviceToken: string | null
+}): Promise<boolean> {
+  const text = normalizeTerminalTextInput(args.bytes)
+  if (text.length === 0) {
+    return false
+  }
+  if (
+    !args.rpc ||
+    !args.connected ||
+    args.handle !== args.activeHandle ||
+    args.activeSessionTabType !== 'terminal'
+  ) {
+    return false
+  }
+  // Why: Input frames return after the socket write; waiting for terminal.send
+  // would reintroduce one RTT per mirror delta on every transport.
+  const streamResult = args.rpc.sendTerminalInput?.(args.handle, text)
+  if (streamResult === 'sent') {
+    return true
+  }
+  if (streamResult === 'failed') {
+    return false
+  }
+  return args.rpc
+    .sendRequest(
+      'terminal.send',
+      buildTerminalSendParams({
+        terminal: args.handle,
+        text,
+        enter: false,
+        deviceToken: args.deviceToken
+      }),
+      TERMINAL_INPUT_SEND_OPTIONS
+    )
+    .then(isTerminalSendRpcAccepted, () => false)
+}

mobile/src/terminal/terminal-live-accessory-raw-send.test.ts

@@ -17,6 +17,17 @@ const BASE_ARGS = {
 } as const
 
 describe('terminal live accessory raw send', () => {
+  it('writes accessory bytes on the subscribe stream when the host has one', async () => {
+    const sendTerminalInput = vi.fn(() => 'sent' as const)
+    const sendRequest = vi.fn()
+    const client = { sendRequest, sendTerminalInput }
+
+    await sendTerminalLiveAccessoryRawBytes({ ...BASE_ARGS, client })
+
+    expect(sendTerminalInput).toHaveBeenCalledWith('terminal-a', '')
+    expect(sendRequest).not.toHaveBeenCalled()
+  })
+
   it('sends raw bytes now-or-never with the device presence tag', async () => {
     const { client, sendRequest } = captureClient()
 

mobile/src/terminal/terminal-live-accessory-raw-send.ts

@@ -4,7 +4,7 @@ import type { RpcClient } from '../transport/rpc-client'
 import type { ConnectionState } from '../transport/types'
 
 type TerminalLiveAccessoryRawSendArgs = {
-  readonly client: Pick<RpcClient, 'sendRequest'> | null
+  readonly client: Pick<RpcClient, 'sendRequest' | 'sendTerminalInput'> | null
   readonly targetHandle: string
   readonly activeHandle: string | null
   readonly activeSessionTabType: string | null
@@ -25,6 +25,10 @@ export async function sendTerminalLiveAccessoryRawBytes(
   if (!args.client || !rawSendTarget || args.connState !== 'connected') {
     return
   }
+  const streamResult = args.client.sendTerminalInput?.(rawSendTarget, args.bytes)
+  if (streamResult === 'sent' || streamResult === 'failed') {
+    return
+  }
   await args.client
     .sendRequest(
       'terminal.send',

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

mobile/src/transport/mobile-relay-rpc-session.test.ts

@@ -3,7 +3,12 @@ import {
   BrowserScreencastOpcode,
   encodeBrowserScreencastFrame
 } from '../../../src/shared/browser-screencast-protocol'
-import { encodeTerminalStreamFrame, TerminalStreamOpcode } from './terminal-stream-protocol'
+import {
+  decodeTerminalStreamFrame,
+  decodeTerminalStreamText,
+  encodeTerminalStreamFrame,
+  TerminalStreamOpcode
+} from './terminal-stream-protocol'
 import { isRpcDeliveryUnknown } from './rpc-delivery-ambiguity'
 
 const fakes = vi.hoisted(() => ({
@@ -18,6 +23,7 @@ const fakes = vi.hoisted(() => ({
     onError(error: Error): void
   },
   sendText: vi.fn(() => true),
+  sendBinary: vi.fn(() => true),
   close: vi.fn()
 }))
 
@@ -27,6 +33,7 @@ vi.mock('./mobile-relay-e2ee-link', () => ({
       fakes.linkOptions = options
     }
     sendText = fakes.sendText
+    sendBinary = fakes.sendBinary
     close = fakes.close
   }
 }))
@@ -102,6 +109,7 @@ describe('mobile relay RPC session', () => {
     vi.clearAllMocks()
     fakes.linkOptions = null
     fakes.sendText.mockReturnValue(true)
+    fakes.sendBinary.mockReturnValue(true)
   })
   afterEach(() => vi.useRealTimers())
 
@@ -169,8 +177,18 @@ describe('mobile relay RPC session', () => {
       streamId: 42,
       chunk: 'hello'
     })
+    expect(session.sendTerminalInput('term-1', 'xy')).toBe('sent')
+    expect(fakes.sendBinary).toHaveBeenCalledOnce()
+    const binaryPayload = fakes.sendBinary.mock.calls[0]?.[0]
+    expect(binaryPayload).toBeInstanceOf(Uint8Array)
+    const input =
+      binaryPayload instanceof Uint8Array ? decodeTerminalStreamFrame(binaryPayload) : null
+    expect(input?.opcode).toBe(TerminalStreamOpcode.Input)
+    expect(input?.streamId).toBe(42)
+    expect(input ? decodeTerminalStreamText(input.payload) : '').toBe('xy')
 
     fakes.sendText.mockClear()
+    fakes.sendBinary.mockClear()
     const onBinaryFrame = vi.fn()
     session.subscribe('browser.screencast', {}, vi.fn(), { onBinaryFrame })
     await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledOnce())

mobile/src/transport/mobile-relay-rpc-session.ts

@@ -57,6 +57,8 @@ export function connectMobileRelayRpcSession(args: {
   const streams = new MobileRelayRpcStreams({
     nextId,
     sendFrame,
+    sendBinary: (bytes) => link.sendBinary(bytes),
+    isConnected: () => state === 'connected',
     waitForConnected: () => waitForConnected()
   })
 
@@ -105,6 +107,10 @@ export function connectMobileRelayRpcSession(args: {
       return streams.subscribe(method, params, listener, options)
     },
 
+    sendTerminalInput(terminal, text) {
+      return streams.sendTerminalInput(terminal, text)
+    },
+
     updateTerminalSubscriptionViewport(terminal, viewport) {
       streams.updateTerminalViewport(terminal, viewport)
     },

mobile/src/transport/mobile-relay-rpc-streams.test.ts

@@ -18,6 +18,8 @@ describe('MobileRelayRpcStreams failure parity', () => {
     const streams = new MobileRelayRpcStreams({
       nextId: () => 'stream-1',
       sendFrame,
+      sendBinary: () => true,
+      isConnected: () => true,
       waitForConnected: async () => {}
     })
     const cancel = streams.subscribe(
@@ -45,6 +47,8 @@ describe('MobileRelayRpcStreams failure parity', () => {
     const streams = new MobileRelayRpcStreams({
       nextId: () => 'stream-1',
       sendFrame,
+      sendBinary: () => true,
+      isConnected: () => true,
       waitForConnected: () => Promise.reject(waitError)
     })
     const cancel = streams.subscribe(
@@ -71,6 +75,8 @@ describe('MobileRelayRpcStreams failure parity', () => {
     const streams = new MobileRelayRpcStreams({
       nextId: () => 'stream-1',
       sendFrame,
+      sendBinary: () => true,
+      isConnected: () => true,
       waitForConnected: async () => {}
     })
     const cancel = streams.subscribe(
@@ -98,6 +104,8 @@ describe('MobileRelayRpcStreams failure parity', () => {
     const streams = new MobileRelayRpcStreams({
       nextId: () => 'stream-1',
       sendFrame,
+      sendBinary: () => true,
+      isConnected: () => true,
       waitForConnected: () => connection.promise
     })
     const cancel = streams.subscribe(
@@ -121,6 +129,8 @@ describe('MobileRelayRpcStreams failure parity', () => {
     const streams = new MobileRelayRpcStreams({
       nextId: () => 'stream-1',
       sendFrame,
+      sendBinary: () => true,
+      isConnected: () => true,
       waitForConnected: () => connection.promise
     })
     streams.subscribe('session.tabs.subscribe', { worktree: 'id:worktree-1' }, listener)
@@ -140,6 +150,8 @@ describe('MobileRelayRpcStreams failure parity', () => {
     const streams = new MobileRelayRpcStreams({
       nextId: () => 'stream-1',
       sendFrame: () => true,
+      sendBinary: () => true,
+      isConnected: () => true,
       waitForConnected: async () => {}
     })
     streams.subscribe('session.tabs.subscribe', { worktree: 'id:worktree-1' }, listener)

mobile/src/transport/mobile-relay-rpc-streams.ts

@@ -7,6 +7,7 @@ import {
   buildTerminalUnsubscribeParams,
   updateTerminalSubscriptionViewport
 } from './rpc-client-terminal-subscription'
+import { TerminalInputStreamRegistry } from './rpc-client-terminal-input-send'
 import type { RpcClient } from './rpc-client'
 import type { RpcResponse, RpcSuccess } from './types'
 
@@ -27,6 +28,8 @@ type StreamRecord = {
 type StreamManagerOptions = {
   nextId: () => string
   sendFrame: (request: { id: string; method: string; params?: unknown }) => boolean
+  sendBinary: (bytes: Uint8Array) => boolean
+  isConnected: () => boolean
   waitForConnected: () => Promise<void>
 }
 
@@ -34,6 +37,7 @@ export class MobileRelayRpcStreams {
   private readonly streams = new Map<string, StreamRecord>()
   private readonly terminalListeners = new Map<number, (result: unknown) => void>()
   private readonly terminalSnapshots = new Map<number, TerminalSnapshotState>()
+  private readonly terminalInputStreams = new TerminalInputStreamRegistry()
   private activeBrowserStream: StreamRecord | null = null
 
   constructor(private readonly options: StreamManagerOptions) {}
@@ -92,6 +96,7 @@ export class MobileRelayRpcStreams {
       if (typeof metadata.streamId === 'number') {
         stream.streamIds.add(metadata.streamId)
         this.terminalListeners.set(metadata.streamId, stream.listener)
+        this.terminalInputStreams.remember(stream.params, metadata.streamId)
       }
       if (stream.method === 'browser.screencast') {
         this.activeBrowserStream = stream
@@ -108,6 +113,15 @@ export class MobileRelayRpcStreams {
     return true
   }
 
+  sendTerminalInput(terminal: string, text: string) {
+    return this.terminalInputStreams.send(
+      terminal,
+      text,
+      this.options.isConnected(),
+      this.options.sendBinary
+    )
+  }
+
   handleBinary(bytes: Uint8Array): void {
     const browserFrame = decodeBrowserScreencastFrame(bytes)
     if (browserFrame && this.activeBrowserStream?.onBinaryFrame) {
@@ -127,6 +141,7 @@ export class MobileRelayRpcStreams {
     this.streams.clear()
     this.terminalListeners.clear()
     this.terminalSnapshots.clear()
+    this.terminalInputStreams.clear()
     this.activeBrowserStream = null
   }
 
@@ -160,6 +175,7 @@ export class MobileRelayRpcStreams {
     if (!stream) {
       return
     }
+    this.terminalInputStreams.forget(stream.params, stream.streamIds)
     for (const streamId of stream.streamIds) {
       this.terminalListeners.delete(streamId)
       this.terminalSnapshots.delete(streamId)

mobile/src/transport/rpc-client-encrypted-binary-send.test.ts

@@ -0,0 +1,69 @@
+import { describe, expect, it, vi } from 'vitest'
+import {
+  sendEncryptedBinaryPayload,
+  sendLanTerminalInputFrame
+} from './rpc-client-encrypted-binary-send'
+import { TerminalInputStreamRegistry } from './rpc-client-terminal-input-send'
+
+vi.mock('./e2ee', () => ({
+  encrypt: (plaintext: string) => plaintext,
+  encryptBytes: (bytes: Uint8Array) => bytes
+}))
+
+describe('sendEncryptedBinaryPayload', () => {
+  it('writes encrypted bytes on an open socket and reports a closed socket as failed', () => {
+    const sent: Uint8Array[] = []
+    const openSocket = {
+      OPEN: 1,
+      readyState: 1,
+      send: (payload: Uint8Array) => {
+        sent.push(payload)
+      }
+    } as unknown as WebSocket
+    expect(
+      sendEncryptedBinaryPayload({
+        socket: openSocket,
+        sharedKey: new Uint8Array(32),
+        plaintext: new Uint8Array([1, 2]),
+        onWriteError: () => {}
+      })
+    ).toBe(true)
+    expect(sent).toEqual([new Uint8Array([1, 2])])
+
+    const closedSocket = { OPEN: 1, readyState: 3, send: vi.fn() } as unknown as WebSocket
+    expect(
+      sendEncryptedBinaryPayload({
+        socket: closedSocket,
+        sharedKey: new Uint8Array(32),
+        plaintext: new Uint8Array([1]),
+        onWriteError: () => {}
+      })
+    ).toBe(false)
+  })
+
+  it('sends an Input frame for a remembered terminal stream', () => {
+    const registry = new TerminalInputStreamRegistry()
+    registry.remember({ terminal: 'term-1' }, 7)
+    const sent: Uint8Array[] = []
+    const openSocket = {
+      OPEN: 1,
+      readyState: 1,
+      send: (payload: Uint8Array) => {
+        sent.push(payload)
+      }
+    } as unknown as WebSocket
+    expect(
+      sendLanTerminalInputFrame({
+        registry,
+        terminal: 'term-1',
+        text: 'ab',
+        connected: true,
+        socket: openSocket,
+        sharedKey: new Uint8Array(32),
+        isCurrentSocket: (socket) => socket === openSocket,
+        onWriteError: () => {}
+      })
+    ).toBe('sent')
+    expect(sent).toHaveLength(1)
+  })
+})

mobile/src/transport/rpc-client-encrypted-binary-send.ts

@@ -0,0 +1,84 @@
+import { encrypt, encryptBytes } from './e2ee'
+import {
+  TerminalInputStreamRegistry,
+  type TerminalInputSendResult
+} from './rpc-client-terminal-input-send'
+import type { ConnectionState } from './types'
+
+export function sendEncryptedJson(args: {
+  socket: WebSocket | null
+  sharedKey: Uint8Array | null
+  request: unknown
+  state: ConnectionState
+  onWriteError: (socket: WebSocket) => void
+  onDesync: (socket: WebSocket) => void
+}): boolean {
+  const { socket, sharedKey, request, state, onWriteError, onDesync } = args
+  if (socket && socket.readyState === WebSocket.OPEN && sharedKey) {
+    try {
+      socket.send(encrypt(JSON.stringify(request), sharedKey))
+      return true
+    } catch {
+      onWriteError(socket)
+      return false
+    }
+  }
+  console.log('[net] sendEncrypted FAILED — channel not ready', {
+    hasWs: !!socket,
+    readyState: socket?.readyState,
+    hasKey: !!sharedKey,
+    state
+  })
+  // Why: RN can drop onclose, leaving state 'connected' over a dead socket; force reconnect or every send silently fails forever.
+  if (state === 'connected' && socket && socket.readyState !== WebSocket.OPEN) {
+    console.log('[net] sendEncrypted detected ws desync — forcing reconnect', {
+      readyState: socket.readyState
+    })
+    onDesync(socket)
+  }
+  return false
+}
+
+export function sendEncryptedBinaryPayload(args: {
+  socket: WebSocket | null
+  sharedKey: Uint8Array | null
+  plaintext: Uint8Array
+  onWriteError: (socket: WebSocket) => void
+}): boolean {
+  const { socket, sharedKey, plaintext, onWriteError } = args
+  if (!socket || socket.readyState !== WebSocket.OPEN || !sharedKey) {
+    return false
+  }
+  try {
+    socket.send(encryptBytes(plaintext, sharedKey))
+    return true
+  } catch {
+    onWriteError(socket)
+    return false
+  }
+}
+
+export function sendLanTerminalInputFrame(args: {
+  registry: TerminalInputStreamRegistry
+  terminal: string
+  text: string
+  connected: boolean
+  socket: WebSocket | null
+  sharedKey: Uint8Array | null
+  isCurrentSocket: (socket: WebSocket) => boolean
+  onWriteError: (socket: WebSocket) => void
+}): TerminalInputSendResult {
+  const sendingWs = args.socket
+  return args.registry.send(args.terminal, args.text, args.connected, (plaintext) =>
+    sendEncryptedBinaryPayload({
+      socket: sendingWs,
+      sharedKey: args.sharedKey,
+      plaintext,
+      onWriteError: (socket) => {
+        if (args.isCurrentSocket(socket)) {
+          args.onWriteError(socket)
+        }
+      }
+    })
+  )
+}

mobile/src/transport/rpc-client-terminal-input-send.test.ts

@@ -0,0 +1,82 @@
+import { describe, expect, it } from 'vitest'
+import {
+  encodeTerminalInputFrame,
+  forgetTerminalInputStream,
+  rememberTerminalInputStream,
+  sendRememberedTerminalInput,
+  terminalHandleFromSubscribeParams
+} from './rpc-client-terminal-input-send'
+import {
+  decodeTerminalStreamFrame,
+  decodeTerminalStreamText,
+  TerminalStreamOpcode
+} from './terminal-stream-protocol'
+
+function payloadText(bytes: Uint8Array): string {
+  const frame = decodeTerminalStreamFrame(bytes)
+  return frame ? decodeTerminalStreamText(frame.payload) : ''
+}
+
+describe('rpc-client terminal input send', () => {
+  it('encodes an Input frame for the remembered stream', () => {
+    const frame = decodeTerminalStreamFrame(encodeTerminalInputFrame(9, 'hi'))
+    expect(frame?.opcode).toBe(TerminalStreamOpcode.Input)
+    expect(frame?.streamId).toBe(9)
+    expect(frame).not.toBeNull()
+    if (!frame) {
+      return
+    }
+    expect(decodeTerminalStreamText(frame.payload)).toBe('hi')
+  })
+
+  it('sends in order and refuses a disconnected write', () => {
+    const streams = new Map<string, number>()
+    rememberTerminalInputStream(streams, 'term-1', 4)
+    const writes: string[] = []
+    expect(
+      sendRememberedTerminalInput({
+        streams,
+        terminal: 'term-1',
+        text: 'a',
+        connected: true,
+        sendBinary: (bytes) => {
+          writes.push(payloadText(bytes))
+          return true
+        }
+      })
+    ).toBe('sent')
+    expect(
+      sendRememberedTerminalInput({
+        streams,
+        terminal: 'term-1',
+        text: 'b',
+        connected: true,
+        sendBinary: (bytes) => {
+          writes.push(payloadText(bytes))
+          return true
+        }
+      })
+    ).toBe('sent')
+    expect(writes).toEqual(['a', 'b'])
+    expect(
+      sendRememberedTerminalInput({
+        streams,
+        terminal: 'term-1',
+        text: 'c',
+        connected: false,
+        sendBinary: () => true
+      })
+    ).toBe('failed')
+    expect(terminalHandleFromSubscribeParams({ terminal: 'term-1' })).toBe('term-1')
+    forgetTerminalInputStream(streams, 'term-1', 4)
+    expect(
+      sendRememberedTerminalInput({
+        streams,
+        terminal: 'term-1',
+        text: 'd',
+        connected: true,
+        sendBinary: () => true
+      })
+    ).toBe('no-stream')
+  })
+})

mobile/src/transport/rpc-client-terminal-input-send.ts

@@ -0,0 +1,96 @@
+import {
+  encodeTerminalStreamFrame,
+  encodeTerminalStreamText,
+  TerminalStreamOpcode
+} from './terminal-stream-protocol'
+
+export type TerminalInputSendResult = 'sent' | 'no-stream' | 'failed'
+
+export function terminalHandleFromSubscribeParams(params: unknown): string | null {
+  if (!params || typeof params !== 'object') {
+    return null
+  }
+  const terminal = (params as { terminal?: unknown }).terminal
+  return typeof terminal === 'string' ? terminal : null
+}
+
+export function rememberTerminalInputStream(
+  streams: Map<string, number>,
+  terminal: string,
+  streamId: number
+): void {
+  streams.set(terminal, streamId)
+}
+
+export function forgetTerminalInputStream(
+  streams: Map<string, number>,
+  terminal: string | null,
+  streamId: number
+): void {
+  if (terminal && streams.get(terminal) === streamId) {
+    streams.delete(terminal)
+  }
+}
+
+export function encodeTerminalInputFrame(streamId: number, text: string): Uint8Array {
+  return encodeTerminalStreamFrame({
+    opcode: TerminalStreamOpcode.Input,
+    streamId,
+    seq: 0,
+    payload: encodeTerminalStreamText(text)
+  })
+}
+
+export function sendRememberedTerminalInput(args: {
+  streams: ReadonlyMap<string, number>
+  terminal: string
+  text: string
+  connected: boolean
+  sendBinary: (bytes: Uint8Array) => boolean
+}): TerminalInputSendResult {
+  if (!args.connected) {
+    return 'failed'
+  }
+  const streamId = args.streams.get(args.terminal)
+  if (streamId === undefined) {
+    return 'no-stream'
+  }
+  return args.sendBinary(encodeTerminalInputFrame(streamId, args.text)) ? 'sent' : 'failed'
+}
+
+export class TerminalInputStreamRegistry {
+  private readonly streams = new Map<string, number>()
+
+  remember(params: unknown, streamId: number): void {
+    const terminal = terminalHandleFromSubscribeParams(params)
+    if (terminal) {
+      rememberTerminalInputStream(this.streams, terminal, streamId)
+    }
+  }
+
+  forget(params: unknown, streamIds: Iterable<number>): void {
+    const terminal = terminalHandleFromSubscribeParams(params)
+    for (const streamId of streamIds) {
+      forgetTerminalInputStream(this.streams, terminal, streamId)
+    }
+  }
+
+  clear(): void {
+    this.streams.clear()
+  }
+
+  send(
+    terminal: string,
+    text: string,
+    connected: boolean,
+    sendBinary: (bytes: Uint8Array) => boolean
+  ): TerminalInputSendResult {
+    return sendRememberedTerminalInput({
+      streams: this.streams,
+      terminal,
+      text,
+      connected,
+      sendBinary
+    })
+  }
+}

mobile/src/transport/rpc-client-terminal-input.test.ts

@@ -0,0 +1,137 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { connect } from './rpc-client'
+import {
+  decodeTerminalStreamFrame,
+  decodeTerminalStreamText,
+  TerminalStreamOpcode
+} from './terminal-stream-protocol'
+
+vi.mock('./e2ee', () => ({
+  generateKeyPair: () => ({
+    publicKey: new Uint8Array(32),
+    secretKey: new Uint8Array(32)
+  }),
+  deriveSharedKey: () => new Uint8Array(32),
+  publicKeyFromBase64: () => new Uint8Array(32),
+  publicKeyToBase64: () => 'client-public-key',
+  encrypt: (plaintext: string) => `encrypted:${plaintext}`,
+  decrypt: (raw: string) => raw.replace(/^encrypted:/, ''),
+  encryptBytes: (bytes: Uint8Array) => bytes,
+  decryptBytes: (bytes: Uint8Array) => bytes
+}))
+
+class MockWebSocket {
+  static OPEN = 1
+  readonly OPEN = MockWebSocket.OPEN
+  readyState = 0
+  onopen: (() => void) | null = null
+  onclose: (() => void) | null = null
+  onmessage: ((event: { data: unknown }) => void) | null = null
+  sent: Array<string | Uint8Array> = []
+  close = vi.fn(() => {
+    this.readyState = 3
+    this.onclose?.()
+  })
+
+  constructor(readonly endpoint: string) {
+    mockSockets.push(this)
+  }
+
+  send(payload: string | Uint8Array): void {
+    this.sent.push(payload)
+  }
+
+  open(): void {
+    this.readyState = MockWebSocket.OPEN
+    this.onopen?.()
+  }
+
+  receive(payload: unknown): void {
+    this.onmessage?.({ data: payload })
+  }
+}
+
+const mockSockets: MockWebSocket[] = []
+const originalWebSocket = globalThis.WebSocket
+
+function authenticate(socket: MockWebSocket): void {
+  socket.open()
+  socket.receive(JSON.stringify({ type: 'e2ee_ready' }))
+  socket.receive('encrypted:{"type":"e2ee_authenticated"}')
+}
+
+function sentRequest(socket: MockWebSocket, method: string): { id: string } {
+  for (const payload of socket.sent) {
+    if (typeof payload !== 'string') {
+      continue
+    }
+    const decoded = JSON.parse(payload.replace(/^encrypted:/, '')) as {
+      id: string
+      method: string
+    }
+    if (decoded.method === method) {
+      return { id: decoded.id }
+    }
+  }
+  throw new Error(`Request not sent: ${method}`)
+}
+
+describe('rpc-client terminal input frames', () => {
+  beforeEach(() => {
+    mockSockets.length = 0
+    globalThis.WebSocket = MockWebSocket as unknown as typeof WebSocket
+  })
+
+  afterEach(() => {
+    globalThis.WebSocket = originalWebSocket
+  })
+
+  it('writes live terminal input as a stream frame without waiting for an RPC response', () => {
+    const client = connect('ws://desktop.invalid', 'token', 'server-key')
+    const socket = mockSockets[0]!
+    authenticate(socket)
+    client.subscribe('terminal.subscribe', { terminal: 'term-1' }, () => {})
+    socket.receive(
+      `encrypted:${JSON.stringify({
+        id: sentRequest(socket, 'terminal.subscribe').id,
+        ok: true,
+        streaming: true,
+        result: { type: 'subscribed', streamId: 42 }
+      })}`
+    )
+
+    expect(client.sendTerminalInput?.('term-1', 'ab')).toBe('sent')
+    expect(client.sendTerminalInput?.('term-1', 'c')).toBe('sent')
+    const inputFrames = socket.sent
+      .filter((payload): payload is Uint8Array => payload instanceof Uint8Array)
+      .map((payload) => decodeTerminalStreamFrame(payload))
+      .filter(
+        (frame): frame is NonNullable<typeof frame> => frame?.opcode === TerminalStreamOpcode.Input
+      )
+    expect(inputFrames.map((frame) => decodeTerminalStreamText(frame.payload))).toEqual(['ab', 'c'])
+    expect(
+      socket.sent.some(
+        (payload) => typeof payload === 'string' && payload.includes('"method":"terminal.send"')
+      )
+    ).toBe(false)
+    client.close()
+  })
+
+  it('falls back to no-stream before subscribe and fails after disconnect', () => {
+    const client = connect('ws://desktop.invalid', 'token', 'server-key')
+    const socket = mockSockets[0]!
+    authenticate(socket)
+    expect(client.sendTerminalInput?.('term-1', 'a')).toBe('no-stream')
+    client.subscribe('terminal.subscribe', { terminal: 'term-1' }, () => {})
+    socket.receive(
+      `encrypted:${JSON.stringify({
+        id: sentRequest(socket, 'terminal.subscribe').id,
+        ok: true,
+        streaming: true,
+        result: { type: 'subscribed', streamId: 42 }
+      })}`
+    )
+    client.close()
+    expect(client.sendTerminalInput?.('term-1', 'a')).toBe('failed')
+  })
+})

mobile/src/transport/rpc-client.ts

@@ -11,7 +11,6 @@ import {
   deriveSharedKey,
   publicKeyFromBase64,
   publicKeyToBase64,
-  encrypt,
   decrypt,
   decryptBytes
 } from './e2ee'
@@ -28,6 +27,11 @@ import {
   buildTerminalUnsubscribeParams,
   updateTerminalSubscriptionViewport as updateCachedTerminalSubscriptionViewport
 } from './rpc-client-terminal-subscription'
+import { sendEncryptedJson, sendLanTerminalInputFrame } from './rpc-client-encrypted-binary-send'
+import {
+  TerminalInputStreamRegistry,
+  type TerminalInputSendResult
+} from './rpc-client-terminal-input-send'
 import { describeSocketEvent, redactSocketEndpoint } from './socket-event-debug'
 import {
   isStaleRpcSocketEvent,
@@ -102,6 +106,7 @@ export type RpcClient = {
     onData: StreamingListener,
     options?: SubscribeOptions
   ) => () => void
+  sendTerminalInput?: (terminal: string, text: string) => TerminalInputSendResult
   updateTerminalSubscriptionViewport: (
     terminal: string,
     viewport: { cols: number; rows: number }
@@ -202,6 +207,7 @@ export function connect(
   const streamListeners = new Map<string, StreamRequest>()
   const terminalStreamListeners = new Map<number, StreamingListener>()
   const terminalStreamIdsByRequest = new Map<string, Set<number>>()
+  const terminalInputStreams = new TerminalInputStreamRegistry()
   const terminalSnapshots = new Map<number, TerminalSnapshotState>()
   let activeBrowserScreencastRequestId: string | null = null
   let pendingBrowserScreencastRequestId: string | null = null
@@ -561,6 +567,7 @@ export function connect(
             }
             ids.add(result.streamId)
             terminalStreamListeners.set(result.streamId, stream.listener)
+            terminalInputStreams.remember(stream.params, result.streamId)
           }
           if (!stream.cancelled) {
             stream.listener(result)
@@ -839,6 +846,7 @@ export function connect(
     }
     const terminalStreamIds = terminalStreamIdsByRequest.get(id)
     if (terminalStreamIds) {
+      terminalInputStreams.forget(stream?.params, terminalStreamIds)
       for (const streamId of terminalStreamIds) {
         terminalStreamListeners.delete(streamId)
         terminalSnapshots.delete(streamId)
@@ -862,6 +870,7 @@ export function connect(
     if (!terminalStreamIds) {
       return
     }
+    terminalInputStreams.forget(streamListeners.get(id)?.params, terminalStreamIds)
     for (const streamId of terminalStreamIds) {
       terminalStreamListeners.delete(streamId)
       terminalSnapshots.delete(streamId)
@@ -936,33 +945,21 @@ export function connect(
   }
 
   function sendEncrypted(request: unknown): boolean {
-    if (ws && ws.readyState === WebSocket.OPEN && sharedKey) {
-      const sendingWs = ws
-      try {
-        sendingWs.send(encrypt(JSON.stringify(request), sharedKey))
-        return true
-      } catch {
-        if (ws === sendingWs) {
-          closeAndSynthesize(sendingWs)
+    return sendEncryptedJson({
+      socket: ws,
+      sharedKey,
+      request,
+      state,
+      onWriteError: (socket) => {
+        if (ws === socket) {
+          closeAndSynthesize(socket)
         }
-        return false
+      },
+      onDesync: (socket) => {
+        synthesizedCloses.remember(socket, authenticationGeneration)
+        handleSocketClosed(socket, { timedOut: false })
       }
-    }
-    console.log('[net] sendEncrypted FAILED — channel not ready', {
-      hasWs: !!ws,
-      readyState: ws?.readyState,
-      hasKey: !!sharedKey,
-      state
     })
-    // Why: RN can drop onclose, leaving state 'connected' over a dead socket; force reconnect or every send silently fails forever.
-    if (state === 'connected' && ws && ws.readyState !== WebSocket.OPEN) {
-      console.log('[net] sendEncrypted detected ws desync — forcing reconnect', {
-        readyState: ws.readyState
-      })
-      synthesizedCloses.remember(ws, authenticationGeneration)
-      handleSocketClosed(ws, { timedOut: false })
-    }
-    return false
   }
 
   function sendBrowserScreencastUnsubscribe(subscriptionId: string): void {
@@ -1066,6 +1063,18 @@ export function connect(
       })
     },
 
+    sendTerminalInput: (terminal, text) =>
+      sendLanTerminalInputFrame({
+        registry: terminalInputStreams,
+        terminal,
+        text,
+        connected: state === 'connected',
+        socket: ws,
+        sharedKey,
+        isCurrentSocket: (socket) => ws === socket,
+        onWriteError: closeAndSynthesize
+      }),
+
     subscribe(
       method: string,
       params: unknown,

mobile/src/transport/stable-logical-rpc-client.test.ts

@@ -18,6 +18,7 @@ class FakeSession implements RpcClient {
     vi.fn<RpcClient['updateTerminalSubscriptionViewport']>()
   readonly notifyForeground = vi.fn()
   readonly close = vi.fn()
+  sendTerminalInput = vi.fn(() => 'no-stream' as const)
   private state: ConnectionState
   private readonly stateListeners = new Set<(state: ConnectionState) => void>()
   private readonly streamListeners = new Set<(result: unknown) => void>()
@@ -67,6 +68,15 @@ function deferred<T>() {
 }
 
 describe('stable logical RPC client', () => {
+  it('forwards live terminal input to the active physical session', () => {
+    const session = new FakeSession('connected')
+    session.sendTerminalInput = vi.fn(() => 'sent')
+    const client = createStableLogicalRpcClient(session, 'lan')
+
+    expect(client.sendTerminalInput('term-1', 'ab')).toBe('sent')
+    expect(session.sendTerminalInput).toHaveBeenCalledWith('term-1', 'ab')
+  })
+
   it('advertises source-default support on worktree catalog requests', async () => {
     const session = new FakeSession('connected')
     session.sendRequest.mockResolvedValue(success([]))

mobile/src/transport/stable-logical-rpc-client.ts

@@ -113,6 +113,10 @@ export function createStableLogicalRpcClient(
       })
     },
 
+    sendTerminalInput(terminal, text) {
+      return activeSession.sendTerminalInput?.(terminal, text) ?? 'no-stream'
+    },
+
     subscribe(method, params, listener, options) {
       if (closed) {
         return () => {}

mobile/src/transport/terminal-stream-protocol.ts

@@ -9,6 +9,7 @@ export enum TerminalStreamOpcode {
   SnapshotEnd = 4,
   Resized = 5,
   Error = 6,
+  Input = 7,
   Metadata = 12
 }
 
@@ -64,6 +65,10 @@ export function decodeTerminalStreamJson<T>(payload: Uint8Array): T | null {
   }
 }
 
+export function encodeTerminalStreamText(value: string): Uint8Array {
+  return new TextEncoder().encode(value)
+}
+
 export function decodeTerminalStreamText(payload: Uint8Array): string {
   return new TextDecoder().decode(payload)
 }
@@ -76,6 +81,7 @@ function isTerminalStreamOpcode(value: number): value is TerminalStreamOpcode {
     value === TerminalStreamOpcode.SnapshotEnd ||
     value === TerminalStreamOpcode.Resized ||
     value === TerminalStreamOpcode.Error ||
+    value === TerminalStreamOpcode.Input ||
     value === TerminalStreamOpcode.Metadata
   )
 }