fix(e2ee): stop dropping terminal output on a backed-up legacy connection (#26528)

* fix(browser): retry a screencast frame a viewer's socket refused

An idle page sends no next frame, so a refused last frame left the viewer on a stale picture. The
desktop's relay leg to a server uses legacy E2EE, which drops on a full socket where a direct v2 phone
queues; retrying the newest pending frame restores parity. Retries never count toward ghost eviction.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(e2ee): queue legacy binary replies with text instead of dropping them

Legacy E2EE (every direct phone, the desktop's server link) sent binary straight to the socket and
silently dropped it once the outbound budget was full, so terminals lost output and binary could
overtake queued text. Binary now shares the ordered text queue and closes on overflow like v2; the
lossy screencast opts into dropWhenBacklogged and its retry delivers the newest frame after a drain.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(e2ee): keep a backlogged screencast viewer instead of evicting it as a ghost

A lossy frame skipped on a busy socket returned false like a dead socket, so a long backlog evicted a
live viewer. Skips now return 'backlogged', which retries without counting toward eviction; legacy
judges idleness on the sealed size before encrypting, so skipped frames cost no encryption.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* test(browser): expect the ready gate to forward screencast send options

The pre-ready gate now passes the per-frame send options through to the
socket sender, so the replayed startup frame arrives with an options slot.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* test(browser): share the streamed-guest double across screencast tests

Main's renderer throttle lease now calls setBackgroundThrottling on the guest,
so the bare { isDestroyed } guest in two branch tests threw. One shared double
keeps all three screencast suites on the shape the lease needs.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

src/main/runtime/browser-screencast-subscriber-frame-delivery.ts

@@ -0,0 +1,57 @@
+import type { ActiveBrowserScreencastSubscriber } from './runtime-browser-commands-browser-command-target-params'
+import { recordScreencastSubscriberSend } from './browser-screencast-ghost-subscriber-eviction'
+import { sendRemoteBrowserScreencastFrame } from './remote-browser-screencast-frame-admission'
+
+// Why 50: the frame pacer's own backpressure retry interval.
+export const SCREENCAST_PENDING_FRAME_RETRY_MS = 50
+
+/**
+ * Sends one frame to one viewer. A frame its socket did not take is kept as the viewer's newest
+ * pending frame and retried until taken: an idle page produces no next frame, so without the retry
+ * a viewer behind a briefly full socket keeps a stale picture until the page changes.
+ */
+export function deliverScreencastSubscriberFrame(
+  subscriber: ActiveBrowserScreencastSubscriber,
+  bytes: Uint8Array<ArrayBufferLike>
+): void {
+  const outcome = sendRemoteBrowserScreencastFrame(subscriber.sendBinary, bytes)
+  subscriber.pendingFrame = outcome === 'handled' ? null : bytes
+  // Why: a backlogged socket is alive and busy, so only a refusal is evidence of a ghost viewer.
+  if (outcome !== 'backlogged') {
+    subscriber.delivery = recordScreencastSubscriberSend(subscriber.delivery, outcome === 'handled')
+  }
+  if (outcome !== 'handled') {
+    schedulePendingFrameRetry(subscriber)
+  }
+}
+
+/** Paths that remove a viewer that may have received frames call this, so no retry outlives it. */
+export function cancelScreencastSubscriberFrameRetry(
+  subscriber: ActiveBrowserScreencastSubscriber
+): void {
+  if (subscriber.pendingFrameRetry) {
+    clearTimeout(subscriber.pendingFrameRetry)
+    subscriber.pendingFrameRetry = null
+  }
+  subscriber.pendingFrame = null
+}
+
+function schedulePendingFrameRetry(subscriber: ActiveBrowserScreencastSubscriber): void {
+  if (subscriber.pendingFrameRetry) {
+    return
+  }
+  subscriber.pendingFrameRetry = setTimeout(() => {
+    subscriber.pendingFrameRetry = null
+    const bytes = subscriber.pendingFrame
+    if (!bytes) {
+      return
+    }
+    // Why a refused retry is not recorded: ghost eviction must advance only on produced frames.
+    if (sendRemoteBrowserScreencastFrame(subscriber.sendBinary, bytes) === 'handled') {
+      subscriber.pendingFrame = null
+      subscriber.delivery = recordScreencastSubscriberSend(subscriber.delivery, true)
+    } else {
+      schedulePendingFrameRetry(subscriber)
+    }
+  }, SCREENCAST_PENDING_FRAME_RETRY_MS)
+}

src/main/runtime/orca-runtime-browser-screencast-e2ee-backlog.test.ts

@@ -0,0 +1,126 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import type { WebSocket } from 'ws'
+import { BROWSER_SCREENCAST_GHOST_SUBSCRIBER_REFUSAL_LIMIT } from './browser-screencast-ghost-subscriber-eviction'
+import { SCREENCAST_PENDING_FRAME_RETRY_MS } from './browser-screencast-subscriber-frame-delivery'
+import {
+  createSinglePageBrowserCommandsHost,
+  createStreamedGuest
+} from './single-page-browser-commands-host-test-double'
+import { E2EEChannel } from './rpc/e2ee-channel'
+import { decrypt, decryptBytes, deriveSharedKey, encrypt, generateKeyPair } from './rpc/e2ee-crypto'
+import type { RpcBinarySender } from './rpc/rpc-binary-sender'
+
+const { webContentsFromId, startBrowserScreencast } = vi.hoisted(() => ({
+  webContentsFromId: vi.fn(),
+  startBrowserScreencast: vi.fn()
+}))
+
+vi.mock('electron', () => ({
+  ipcMain: { on: vi.fn(), removeListener: vi.fn() },
+  webContents: { fromId: webContentsFromId }
+}))
+vi.mock('../browser/browser-screencast-stream', () => ({ startBrowserScreencast }))
+vi.mock('../telemetry/client', () => ({ track: vi.fn() }))
+
+/** A real legacy E2EE channel, as every directly connected phone uses. */
+function connectLegacyChannel() {
+  const serverKeys = generateKeyPair()
+  const clientKeys = generateKeyPair()
+  const sent: (string | Buffer)[] = []
+  const ws = {
+    OPEN: 1 as const,
+    readyState: 1,
+    bufferedAmount: 0,
+    send: vi.fn((data: string | Buffer) => sent.push(data)),
+    close: vi.fn()
+  }
+  const onError = vi.fn()
+  const channel = new E2EEChannel(ws as unknown as WebSocket, {
+    serverSecretKey: serverKeys.secretKey,
+    resolveAuthenticatedDevice: (token) =>
+      token === 'valid-token'
+        ? { deviceId: 'device-1', deviceToken: token, scope: 'mobile' }
+        : null,
+    onReady: vi.fn(),
+    onError
+  })
+  const sharedKey = deriveSharedKey(clientKeys.secretKey, serverKeys.publicKey)
+  channel.handleRawMessage(
+    JSON.stringify({
+      type: 'e2ee_hello',
+      publicKeyB64: Buffer.from(clientKeys.publicKey).toString('base64')
+    })
+  )
+  channel.handleRawMessage(
+    encrypt(JSON.stringify({ type: 'e2ee_auth', deviceToken: 'valid-token' }), sharedKey)
+  )
+  let reply!: (response: string) => void
+  let sendBinary!: RpcBinarySender
+  channel.onMessage((_plaintext, textReply, binaryReply) => {
+    reply = textReply
+    sendBinary = binaryReply
+  })
+  channel.handleRawMessage(encrypt('{"id":"s","method":"browser.screencast"}', sharedKey))
+  const decode = (frame: string | Buffer): string =>
+    typeof frame === 'string'
+      ? decrypt(frame, sharedKey)!
+      : `binary:${Buffer.from(decryptBytes(new Uint8Array(frame), sharedKey)!).join(',')}`
+  return { ws, sent, onError, reply, sendBinary, decode, channel }
+}
+
+async function subscribe(sendBinary: RpcBinarySender) {
+  const { RuntimeBrowserCommands } = await import('./orca-runtime-browser')
+  let resolveDone!: () => void
+  const done = new Promise<void>((resolve) => {
+    resolveDone = resolve
+  })
+  const stop = vi.fn(() => resolveDone())
+  startBrowserScreencast.mockResolvedValue({ stop, done, updateViewport: vi.fn(async () => {}) })
+  const commands = new RuntimeBrowserCommands(createSinglePageBrowserCommandsHost())
+  const subscription = await commands.browserScreencast(
+    { worktree: 'id:wt-1', page: 'page-1', format: 'jpeg' },
+    { sendBinary }
+  )
+  const onFrame: (bytes: Uint8Array<ArrayBufferLike>) => unknown =
+    startBrowserScreencast.mock.calls[0][1].onFrame
+  return { subscription, onFrame, stop }
+}
+
+describe('screencast over a backlogged legacy E2EE socket', () => {
+  beforeEach(() => {
+    vi.useFakeTimers()
+    webContentsFromId.mockReset()
+    webContentsFromId.mockReturnValue(createStreamedGuest())
+    startBrowserScreencast.mockReset()
+  })
+  afterEach(() => {
+    vi.useRealTimers()
+  })
+
+  // Why: a busy socket is alive; evicting its viewer as a ghost would blank a working phone.
+  it('keeps the viewer through a long backlog and then delivers only the newest frame', async () => {
+    const transport = connectLegacyChannel()
+    const { subscription, onFrame, stop } = await subscribe(transport.sendBinary)
+    onFrame(new Uint8Array([0]))
+    const baseline = transport.sent.length
+
+    transport.ws.bufferedAmount = 9 * 1024 * 1024
+    transport.reply('{"parked":true}')
+    const frameCount = BROWSER_SCREENCAST_GHOST_SUBSCRIBER_REFUSAL_LIMIT + 10
+    for (let frame = 1; frame <= frameCount; frame++) {
+      onFrame(new Uint8Array([frame]))
+    }
+    expect(stop).not.toHaveBeenCalled()
+    expect(transport.sent.length).toBe(baseline)
+
+    transport.ws.bufferedAmount = 0
+    await vi.advanceTimersByTimeAsync(SCREENCAST_PENDING_FRAME_RETRY_MS * 2)
+    expect(transport.sent.slice(baseline).map(transport.decode)).toEqual([
+      '{"parked":true}',
+      `binary:${frameCount}`
+    ])
+    expect(transport.onError).not.toHaveBeenCalled()
+    subscription.session.stop()
+    transport.channel.destroy()
+  })
+})

src/main/runtime/orca-runtime-browser-screencast-fanout.test.ts

@@ -2,7 +2,10 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'
 import { REMOTE_RUNTIME_MAX_OUTBOUND_BINARY_FRAME_BYTES } from '../../shared/remote-runtime-memory-limits'
 import { BROWSER_SCREENCAST_GHOST_SUBSCRIBER_REFUSAL_LIMIT } from './browser-screencast-ghost-subscriber-eviction'
 import type { RuntimeBrowserCommands } from './orca-runtime-browser'
-import { createSinglePageBrowserCommandsHost } from './single-page-browser-commands-host-test-double'
+import {
+  createSinglePageBrowserCommandsHost,
+  createStreamedGuest
+} from './single-page-browser-commands-host-test-double'
 
 const { webContentsFromId, startBrowserScreencast } = vi.hoisted(() => ({
   webContentsFromId: vi.fn(),
@@ -23,15 +26,6 @@ function deferred() {
   return { promise, resolve }
 }
 
-// A guest the page stream keeps painting while it streams.
-function createStreamedGuest() {
-  return {
-    isDestroyed: () => false,
-    setBackgroundThrottling: vi.fn(),
-    capturePage: vi.fn(async () => null)
-  }
-}
-
 describe('RuntimeBrowserCommands screencast fanout', () => {
   beforeEach(() => {
     webContentsFromId.mockReset()
@@ -72,8 +66,8 @@ describe('RuntimeBrowserCommands screencast fanout', () => {
     const frame = new Uint8Array([1, 2, 3])
     expect(startBrowserScreencast).toHaveBeenCalledOnce()
     expect(startBrowserScreencast.mock.calls[0][1].onFrame(frame)).toBe(true)
-    expect(firstSend).toHaveBeenCalledWith(frame)
-    expect(secondSend).toHaveBeenCalledWith(frame)
+    expect(firstSend).toHaveBeenCalledWith(frame, { dropWhenBacklogged: true })
+    expect(secondSend).toHaveBeenCalledWith(frame, { dropWhenBacklogged: true })
     expect(updateViewport).toHaveBeenLastCalledWith(
       expect.objectContaining({ viewportWidth: 800, viewportHeight: 600 })
     )
@@ -167,12 +161,12 @@ describe('RuntimeBrowserCommands screencast fanout', () => {
       { sendBinary: joinerSend }
     )
 
-    expect(creatorSend).toHaveBeenCalledWith(snapshot)
-    expect(joinerSend).toHaveBeenCalledWith(snapshot)
+    expect(creatorSend).toHaveBeenCalledWith(snapshot, { dropWhenBacklogged: true })
+    expect(joinerSend).toHaveBeenCalledWith(snapshot, { dropWhenBacklogged: true })
     joinerSend.mockClear()
     gateOpen = true
     joiner.flushPendingFrame()
-    expect(joinerSend).toHaveBeenCalledExactlyOnceWith(snapshot)
+    expect(joinerSend).toHaveBeenCalledExactlyOnceWith(snapshot, { dropWhenBacklogged: true })
 
     // The accepted replay is not retained, so a second flush cannot duplicate it.
     joinerSend.mockClear()

src/main/runtime/orca-runtime-browser-screencast-pending-frame-retry.test.ts

@@ -0,0 +1,136 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { BROWSER_SCREENCAST_GHOST_SUBSCRIBER_REFUSAL_LIMIT } from './browser-screencast-ghost-subscriber-eviction'
+import { SCREENCAST_PENDING_FRAME_RETRY_MS } from './browser-screencast-subscriber-frame-delivery'
+import {
+  createSinglePageBrowserCommandsHost,
+  createStreamedGuest
+} from './single-page-browser-commands-host-test-double'
+import type { RpcBinarySender } from './rpc/rpc-binary-sender'
+
+const { webContentsFromId, startBrowserScreencast } = vi.hoisted(() => ({
+  webContentsFromId: vi.fn(),
+  startBrowserScreencast: vi.fn()
+}))
+
+vi.mock('electron', () => ({
+  ipcMain: { on: vi.fn(), removeListener: vi.fn() },
+  webContents: { fromId: webContentsFromId }
+}))
+vi.mock('../browser/browser-screencast-stream', () => ({ startBrowserScreencast }))
+
+async function subscribe(sendBinary: RpcBinarySender) {
+  const { RuntimeBrowserCommands } = await import('./orca-runtime-browser')
+  let resolveDone!: () => void
+  const done = new Promise<void>((resolve) => {
+    resolveDone = resolve
+  })
+  const stop = vi.fn(() => resolveDone())
+  startBrowserScreencast.mockResolvedValue({
+    stop,
+    done,
+    updateViewport: vi.fn(async () => {})
+  })
+  const commands = new RuntimeBrowserCommands(createSinglePageBrowserCommandsHost())
+  const subscription = await commands.browserScreencast(
+    { worktree: 'id:wt-1', page: 'page-1', format: 'jpeg' },
+    { sendBinary }
+  )
+  const onFrame: (bytes: Uint8Array<ArrayBufferLike>) => unknown =
+    startBrowserScreencast.mock.calls[0][1].onFrame
+  return { subscription, onFrame, stop }
+}
+
+describe('screencast frames a viewer refused', () => {
+  beforeEach(() => {
+    vi.useFakeTimers()
+    webContentsFromId.mockReset()
+    webContentsFromId.mockReturnValue(createStreamedGuest())
+    startBrowserScreencast.mockReset()
+  })
+  afterEach(() => {
+    vi.useRealTimers()
+  })
+
+  // Why: an idle page sends no next frame, so a refused last frame must reach the viewer once its
+  // socket drains, or a relayed phone (whose server leg drops on a full socket) keeps a stale page.
+  it('delivers the last refused frame once the socket drains, with no new frame produced', async () => {
+    let socketFull = true
+    const received: Uint8Array<ArrayBufferLike>[] = []
+    const { subscription, onFrame } = await subscribe((bytes) => {
+      if (socketFull) {
+        return false
+      }
+      received.push(bytes)
+      return true
+    })
+    const first = new Uint8Array([1])
+    const last = new Uint8Array([2])
+    onFrame(first)
+    onFrame(last)
+    await vi.advanceTimersByTimeAsync(SCREENCAST_PENDING_FRAME_RETRY_MS * 4)
+    expect(received).toEqual([])
+
+    socketFull = false
+    await vi.advanceTimersByTimeAsync(SCREENCAST_PENDING_FRAME_RETRY_MS)
+    expect(received).toEqual([last])
+    await vi.advanceTimersByTimeAsync(SCREENCAST_PENDING_FRAME_RETRY_MS * 4)
+    expect(received).toEqual([last])
+    subscription.session.stop()
+  })
+
+  it('retries the newest refused frame when a newer one is refused mid-retry', async () => {
+    let socketFull = true
+    const received: Uint8Array<ArrayBufferLike>[] = []
+    const { subscription, onFrame } = await subscribe((bytes) => {
+      if (socketFull) {
+        return false
+      }
+      received.push(bytes)
+      return true
+    })
+    const stale = new Uint8Array([1])
+    const newest = new Uint8Array([2])
+    onFrame(stale)
+    await vi.advanceTimersByTimeAsync(SCREENCAST_PENDING_FRAME_RETRY_MS * 2)
+    onFrame(newest)
+    socketFull = false
+    await vi.advanceTimersByTimeAsync(SCREENCAST_PENDING_FRAME_RETRY_MS * 4)
+    expect(received).toEqual([newest])
+    subscription.session.stop()
+  })
+
+  // Why: a lossy frame must not queue behind a backlog; the retry above delivers the newest instead.
+  it('asks the transport to drop frames while the socket is backlogged', async () => {
+    const sendBinary = vi.fn(() => true)
+    const { subscription, onFrame } = await subscribe(sendBinary)
+    onFrame(new Uint8Array([1]))
+    expect(sendBinary).toHaveBeenCalledWith(new Uint8Array([1]), { dropWhenBacklogged: true })
+    subscription.session.stop()
+  })
+
+  it('stops retrying once the viewer leaves', async () => {
+    const sendBinary = vi.fn(() => false)
+    const { subscription, onFrame } = await subscribe(sendBinary)
+    onFrame(new Uint8Array([1]))
+    subscription.session.stop()
+    await subscription.session.done
+    sendBinary.mockClear()
+    await vi.advanceTimersByTimeAsync(SCREENCAST_PENDING_FRAME_RETRY_MS * 4)
+    expect(sendBinary).not.toHaveBeenCalled()
+  })
+
+  it('does not count refused retries toward ghost eviction', async () => {
+    let refusing = false
+    const { subscription, onFrame, stop } = await subscribe(() => !refusing)
+    onFrame(new Uint8Array([0]))
+    refusing = true
+    onFrame(new Uint8Array([1]))
+    await vi.advanceTimersByTimeAsync(
+      SCREENCAST_PENDING_FRAME_RETRY_MS * BROWSER_SCREENCAST_GHOST_SUBSCRIBER_REFUSAL_LIMIT * 2
+    )
+    // Eviction is judged when a frame is produced.
+    onFrame(new Uint8Array([2]))
+    expect(stop).not.toHaveBeenCalled()
+    subscription.session.stop()
+  })
+})

src/main/runtime/orca-runtime-browser.test.ts

@@ -669,8 +669,8 @@ describe('RuntimeBrowserCommands browser screencast', () => {
     expect(first.subscriptionId).not.toBe(second.subscriptionId)
     const frame = new Uint8Array([1, 2, 3])
     expect(startBrowserScreencastMock.mock.calls[0][1].onFrame(frame)).toBe(true)
-    expect(firstSend).toHaveBeenCalledWith(frame)
-    expect(secondSend).toHaveBeenCalledWith(frame)
+    expect(firstSend).toHaveBeenCalledWith(frame, { dropWhenBacklogged: true })
+    expect(secondSend).toHaveBeenCalledWith(frame, { dropWhenBacklogged: true })
     expect(updateViewport).toHaveBeenLastCalledWith(
       expect.objectContaining({ viewportWidth: 800, viewportHeight: 600 })
     )

src/main/runtime/orca-runtime-tests/mobile-creation-and-orchestration.spec.ts

@@ -672,7 +672,7 @@ describe('OrcaRuntimeService', () => {
     )
     // The gate still refuses the frame on arrival; it reaches the client only via the
     // post-ready replay, so a static page does not leave the subscriber frameless.
-    expect(sendBinary).toHaveBeenCalledExactlyOnceWith(startupFrame)
+    expect(sendBinary).toHaveBeenCalledExactlyOnceWith(startupFrame, undefined)
 
     runtime.cleanupSubscription('browser-screencast:page-1:first')
     await task

src/main/runtime/remote-browser-screencast-frame-admission.test.ts

@@ -7,6 +7,7 @@ import { REMOTE_RUNTIME_MAX_OUTBOUND_BINARY_FRAME_BYTES } from '../../shared/rem
 import { startBrowserScreencast } from '../browser/browser-screencast-stream'
 import { E2EEChannel } from './rpc/e2ee-channel'
 import { sendRemoteBrowserScreencastFrame } from './remote-browser-screencast-frame-admission'
+import type { RpcBinarySender } from './rpc/rpc-binary-sender'
 
 function createMockWebContents() {
   let attached = false
@@ -58,7 +59,7 @@ function createRuntimeBinarySender() {
   channel.handleRawMessage(
     encrypt(JSON.stringify({ type: 'e2ee_auth', deviceToken: 'valid-token' }), sharedKey)
   )
-  let sendBinary: ((bytes: Uint8Array<ArrayBufferLike>) => boolean | void) | undefined
+  let sendBinary: RpcBinarySender | undefined
   channel.onMessage((_plaintext, _sendText, sendBinaryReply) => {
     sendBinary = sendBinaryReply
   })
@@ -74,7 +75,7 @@ describe('sendRemoteBrowserScreencastFrame', () => {
     const sendBinary = vi.fn(() => true)
     const oversized = new Uint8Array(REMOTE_RUNTIME_MAX_OUTBOUND_BINARY_FRAME_BYTES + 1)
 
-    expect(sendRemoteBrowserScreencastFrame(sendBinary, oversized)).toBe(true)
+    expect(sendRemoteBrowserScreencastFrame(sendBinary, oversized)).toBe('handled')
     expect(sendBinary).not.toHaveBeenCalled()
   })
 
@@ -86,14 +87,20 @@ describe('sendRemoteBrowserScreencastFrame', () => {
         vi.fn(() => false),
         withinLimit
       )
-    ).toBe(false)
+    ).toBe('refused')
     expect(
       sendRemoteBrowserScreencastFrame(
         vi.fn(() => true),
         withinLimit
       )
-    ).toBe(true)
-    expect(sendRemoteBrowserScreencastFrame(vi.fn(), withinLimit)).toBe(true)
+    ).toBe('handled')
+    expect(sendRemoteBrowserScreencastFrame(vi.fn(), withinLimit)).toBe('handled')
+    expect(
+      sendRemoteBrowserScreencastFrame(
+        vi.fn(() => 'backlogged' as const),
+        withinLimit
+      )
+    ).toBe('backlogged')
   })
 
   it('drops an oversized encoded frame without closing or stalling the paired stream', async () => {
@@ -107,7 +114,8 @@ describe('sendRemoteBrowserScreencastFrame', () => {
       maxHeight: 2160,
       everyNthFrame: 1,
       minFrameIntervalMs: 0,
-      onFrame: (bytes) => sendRemoteBrowserScreencastFrame(transport.sendBinary, bytes)
+      onFrame: (bytes) =>
+        sendRemoteBrowserScreencastFrame(transport.sendBinary, bytes) !== 'refused'
     })
 
     try {

src/main/runtime/remote-browser-screencast-frame-admission.ts

@@ -1,14 +1,19 @@
 import { isRemoteRuntimeBinaryFrameWithinLimit } from '../../shared/remote-runtime-memory-limits'
+import type { RpcBinarySender } from './rpc/rpc-binary-sender'
+
+/** 'backlogged': a live socket skipped the frame; 'refused': the transport would not take it. */
+export type ScreencastFrameSendOutcome = 'handled' | 'backlogged' | 'refused'
 
 // Why: the E2EE channel closes the socket (1013) on an over-limit binary frame, and the
-// screencast producer reads `false` as backpressure and retries the identical frame. Reporting
+// screencast producer reads a refusal as backpressure and retries the identical frame. Reporting
 // an over-limit frame as handled drops it so the stream advances instead of retrying forever.
 export function sendRemoteBrowserScreencastFrame(
-  sendBinary: (bytes: Uint8Array<ArrayBufferLike>) => boolean | void,
+  sendBinary: RpcBinarySender,
   bytes: Uint8Array<ArrayBufferLike>
-): boolean {
+): ScreencastFrameSendOutcome {
   if (!isRemoteRuntimeBinaryFrameWithinLimit(bytes)) {
-    return true
+    return 'handled'
   }
-  return sendBinary(bytes) !== false
+  const sent = sendBinary(bytes, { dropWhenBacklogged: true })
+  return sent === 'backlogged' ? 'backlogged' : sent === false ? 'refused' : 'handled'
 }

src/main/runtime/rpc/core.ts

@@ -16,6 +16,7 @@ import type {
 import type { OrchestrationCompatibilityEvidence } from '../../../shared/orchestration-compatibility-evidence'
 import type { OrchestrationSessionCaller } from '../orchestration/orchestration-caller-identity'
 import type { RpcCallerIdentity } from './rpc-caller-identity'
+import type { RpcBinarySender } from './rpc-binary-sender'
 import type { RpcMethodPermission } from './rpc-method-permission'
 
 export type PairingRpcContext = {
@@ -123,7 +124,7 @@ export type RpcContext = {
     sync(phones: readonly DelegatedPhone[]): DelegatedMobileDeviceSyncResult
   }
   // Why: mobile terminal traffic bypasses JSON streaming; undefined on Unix/socket and non-E2EE WebSocket paths.
-  sendBinary?: (bytes: Uint8Array<ArrayBufferLike>) => boolean | void
+  sendBinary?: RpcBinarySender
   // Why: binary terminal frames arrive outside JSON-RPC once a stream is established; handlers register only the stream IDs they created.
   registerBinaryStreamHandler?: (
     streamId: number,

src/main/runtime/rpc/e2ee-channel-text-backpressure.test.ts

@@ -1,7 +1,8 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import type { WebSocket } from 'ws'
+import type { RpcBinarySendResult } from './rpc-binary-sender'
 import { E2EEChannel, type E2EEChannelOptions } from './e2ee-channel'
-import { deriveSharedKey, decrypt, encrypt, generateKeyPair } from './e2ee-crypto'
+import { deriveSharedKey, decrypt, decryptBytes, encrypt, generateKeyPair } from './e2ee-crypto'
 import { createMobileE2EEOutboundMemoryBudget } from './mobile-e2ee-outbound-memory-budget'
 import { REMOTE_RUNTIME_MAX_OUTBOUND_JSON_BYTES } from '../../../shared/remote-runtime-memory-limits'
 
@@ -20,12 +21,12 @@ function publicKeyToBase64(key: Uint8Array): string {
 }
 
 function createMockWs() {
-  const sent: string[] = []
+  const sent: (string | Buffer)[] = []
   return {
     OPEN: 1 as const,
     readyState: 1,
     bufferedAmount: 0,
-    send: vi.fn((data: string) => {
+    send: vi.fn((data: string | Buffer) => {
       sent.push(data)
     }),
     close: vi.fn(),
@@ -93,7 +94,9 @@ describe('E2EE text reply backpressure', () => {
     ctx.ws.bufferedAmount = 0
     vi.runOnlyPendingTimers()
 
-    const replies = ctx.ws.sent.slice(baseline).map((frame) => decrypt(frame, ctx.sharedKey))
+    const replies = ctx.ws.sent
+      .slice(baseline)
+      .map((frame) => decrypt(String(frame), ctx.sharedKey))
     expect(replies).toEqual(['{"seq":1}', '{"seq":2}', '{"seq":3}'])
     expect(ctx.onError).not.toHaveBeenCalled()
   })
@@ -103,7 +106,7 @@ describe('E2EE text reply backpressure', () => {
     const baseline = ctx.ws.sent.length
     emitReply(ctx, '{"ok":true}')
     expect(ctx.ws.sent.length).toBe(baseline + 1)
-    expect(decrypt(ctx.ws.sent[baseline]!, ctx.sharedKey)).toBe('{"ok":true}')
+    expect(decrypt(String(ctx.ws.sent[baseline]), ctx.sharedKey)).toBe('{"ok":true}')
   })
 
   it('still closes an oversized reply when telemetry throws', () => {
@@ -143,3 +146,102 @@ describe('E2EE text reply backpressure', () => {
     expect(outboundMemoryBudget.evidence().queuedBytes).toBe(0)
   })
 })
+
+describe('E2EE legacy binary reply backpressure', () => {
+  beforeEach(() => {
+    vi.useFakeTimers()
+    trackMock.mockReset()
+  })
+  afterEach(() => {
+    vi.useRealTimers()
+  })
+
+  function decodeSent(ctx: ReturnType<typeof setup>, frame: string | Buffer): string {
+    return typeof frame === 'string'
+      ? decrypt(frame, ctx.sharedKey)!
+      : `binary:${Buffer.from(decryptBytes(new Uint8Array(frame), ctx.sharedKey)!).toString()}`
+  }
+
+  // Why: legacy terminal.subscribe has no seq-gap check, so a dropped or reordered binary frame corrupts the phone's screen.
+  it('holds binary replies in one order with text while the outbound budget is full', () => {
+    const outboundMemoryBudget = createMobileE2EEOutboundMemoryBudget({ maxBufferedBytes: 1_000 })
+    const ctx = setup({ outboundMemoryBudget })
+    const baseline = ctx.ws.sent.length
+    ctx.ws.bufferedAmount = 1_001
+    const accepted: RpcBinarySendResult[] = []
+
+    ctx.channel.onMessage((_plaintext, encryptedReply, encryptedBinaryReply) => {
+      encryptedReply('{"seq":1}')
+      accepted.push(encryptedBinaryReply(Buffer.from('2')))
+      encryptedReply('{"seq":3}')
+      accepted.push(encryptedBinaryReply(Buffer.from('4')))
+    })
+    ctx.channel.handleRawMessage(encrypt('{"id":"x","method":"terminal.subscribe"}', ctx.sharedKey))
+    expect(ctx.ws.sent.length).toBe(baseline)
+
+    ctx.ws.bufferedAmount = 0
+    vi.runOnlyPendingTimers()
+
+    expect(accepted).toEqual([true, true])
+    expect(ctx.ws.sent.slice(baseline).map((frame) => decodeSent(ctx, frame))).toEqual([
+      '{"seq":1}',
+      'binary:2',
+      '{"seq":3}',
+      'binary:4'
+    ])
+    expect(ctx.onError).not.toHaveBeenCalled()
+  })
+
+  it('closes for budget instead of dropping when queued binary overflows', () => {
+    const outboundMemoryBudget = createMobileE2EEOutboundMemoryBudget({
+      maxBufferedBytes: 1_000,
+      maxQueuedBytes: 150
+    })
+    const ctx = setup({ outboundMemoryBudget })
+    ctx.ws.bufferedAmount = 1_001
+
+    ctx.channel.onMessage((_plaintext, _encryptedReply, encryptedBinaryReply) => {
+      encryptedBinaryReply(new Uint8Array(100))
+      encryptedBinaryReply(new Uint8Array(100))
+    })
+    ctx.channel.handleRawMessage(encrypt('{"id":"x","method":"terminal.subscribe"}', ctx.sharedKey))
+
+    expect(ctx.onError).toHaveBeenCalledWith(1013, 'Outbound reply buffer overflow')
+    expect(trackMock).toHaveBeenCalledWith('remote_outbound_budget_close', { emitter: 'queue' })
+  })
+
+  it('refuses binary once the socket has left OPEN', () => {
+    const ctx = setup()
+    ctx.ws.readyState = 3
+    let accepted: RpcBinarySendResult = undefined
+
+    ctx.channel.onMessage((_plaintext, _encryptedReply, encryptedBinaryReply) => {
+      accepted = encryptedBinaryReply(new Uint8Array([1]))
+    })
+    ctx.channel.handleRawMessage(encrypt('{"id":"x","method":"terminal.subscribe"}', ctx.sharedKey))
+
+    expect(accepted).toBe(false)
+  })
+
+  it('drops a backlogged lossy binary reply instead of queueing it, then sends once drained', () => {
+    const ctx = setup()
+    const baseline = ctx.ws.sent.length
+    ctx.ws.bufferedAmount = 9 * 1024 * 1024
+    const accepted: RpcBinarySendResult[] = []
+    ctx.channel.onMessage((_plaintext, _encryptedReply, encryptedBinaryReply) => {
+      accepted.push(encryptedBinaryReply(Buffer.from('frame'), { dropWhenBacklogged: true }))
+    })
+    const request = encrypt('{"id":"x","method":"browser.screencast"}', ctx.sharedKey)
+    ctx.channel.handleRawMessage(request)
+    ctx.ws.bufferedAmount = 0
+    vi.runOnlyPendingTimers()
+    expect(ctx.ws.sent.length).toBe(baseline)
+
+    ctx.channel.handleRawMessage(request)
+    expect(accepted).toEqual(['backlogged', true])
+    expect(ctx.ws.sent.slice(baseline).map((frame) => decodeSent(ctx, frame))).toEqual([
+      'binary:frame'
+    ])
+    expect(ctx.onError).not.toHaveBeenCalled()
+  })
+})

src/main/runtime/rpc/e2ee-channel-v2.test.ts

@@ -12,6 +12,7 @@ import {
   sealMobileE2EEV2Frame
 } from '../../../shared/mobile-e2ee-v2-framing'
 import { deriveSharedKey } from './e2ee-crypto'
+import type { RpcBinarySendResult } from './rpc-binary-sender'
 import { E2EEChannel } from './e2ee-channel'
 import { deriveMobileE2EEV2KeySchedule } from './mobile-e2ee-v2-key-schedule'
 
@@ -276,4 +277,25 @@ describe('E2EEChannel v2', () => {
     )
     expect(ctx.ws.sent[3]!.options).toEqual({ binary: true })
   })
+
+  it('drops a backlogged lossy binary reply without spending a counter, then sends once drained', () => {
+    const ctx = setup()
+    const { schedule } = startV2(ctx)
+    authenticate(ctx, schedule)
+    ctx.ws.bufferedAmount = 9 * 1024 * 1024
+    const accepted: RpcBinarySendResult[] = []
+    ctx.channel.onMessage((_request, _textReply, binaryReply) => {
+      accepted.push(binaryReply(new Uint8Array([1]), { dropWhenBacklogged: true }))
+    })
+    ctx.channel.handleRawMessage(clientText('{"method":"browser.screencast"}', schedule, 1n))
+    ctx.ws.bufferedAmount = 0
+    vi.runOnlyPendingTimers()
+    expect(ctx.ws.sent).toHaveLength(2)
+
+    ctx.channel.handleRawMessage(clientText('{"method":"browser.screencast"}', schedule, 2n))
+    expect(accepted).toEqual(['backlogged', true])
+    expect(openServerFrame(ctx.ws.sent[2]!.data, 'binary', schedule, 1n)).toEqual(
+      new Uint8Array([1])
+    )
+  })
 })

src/main/runtime/rpc/e2ee-channel.test.ts

@@ -1,5 +1,6 @@
 import { describe, expect, it, vi, beforeEach, afterEach } from 'vitest'
 import type { WebSocket } from 'ws'
+import type { RpcBinarySendResult } from './rpc-binary-sender'
 import { E2EEChannel, type E2EEChannelOptions } from './e2ee-channel'
 import { generateKeyPair, deriveSharedKey, encrypt, decrypt, encryptBytes } from './e2ee-crypto'
 import {
@@ -308,7 +309,7 @@ describe('E2EEChannel', () => {
       const ctx = setup()
       const sharedKey = doHandshake(ctx)
       const sentBefore = ctx.ws.sent.length
-      let accepted: boolean | void = undefined
+      let accepted: RpcBinarySendResult = undefined
 
       ctx.channel.onMessage((_plaintext, _encryptedReply, encryptedBinaryReply) => {
         accepted = encryptedBinaryReply(

src/main/runtime/rpc/e2ee-channel.ts

@@ -1,6 +1,6 @@
 // Why: this channel keeps E2EE framing out of RPC handlers, which consume plaintext across transports.
 import type { WebSocket } from 'ws'
-import { deriveSharedKey, encrypt, decrypt, encryptBytes, decryptBytes } from './e2ee-crypto'
+import { deriveSharedKey, encrypt, decrypt, decryptBytes } from './e2ee-crypto'
 import {
   DesktopMobileE2EEV2Session,
   type DesktopMobileE2EEV2Context
@@ -17,6 +17,11 @@ import { parseRemoteRuntimeJsonText } from '../../../shared/remote-runtime-reque
 import type { MobileE2EEOutboundMemoryBudget } from './mobile-e2ee-outbound-memory-budget'
 import { MobileE2EEDesktopOutboundOwner } from './mobile-e2ee-desktop-outbound-owner'
 import { parseRuntimeClientCapabilities } from './runtime-client-capabilities'
+import type {
+  RpcBinarySendOptions,
+  RpcBinarySendResult,
+  RpcBinarySender
+} from './rpc-binary-sender'
 import type { RuntimeCapability } from '../../../shared/protocol-version'
 import type { EventProps } from '../../../shared/telemetry-events'
 import { track } from '../../telemetry/client'
@@ -61,7 +66,7 @@ export class E2EEChannel {
     | ((
         plaintext: string,
         encryptedReply: (response: string) => void,
-        encryptedBinaryReply: (response: Uint8Array<ArrayBufferLike>) => boolean | void
+        encryptedBinaryReply: RpcBinarySender
       ) => void)
     | null = null
   private binaryMessageHandler: ((plaintext: Uint8Array<ArrayBufferLike>) => void) | null = null
@@ -78,7 +83,12 @@ export class E2EEChannel {
     this.onError = options.onError
     this.transportContext = options.transportContext ?? { transport: 'direct' }
     this.requireV2 = options.requireV2 ?? false
-    this.outbound = new MobileE2EEDesktopOutboundOwner(ws, options.outboundMemoryBudget)
+    this.outbound = new MobileE2EEDesktopOutboundOwner(
+      ws,
+      () => Boolean(this.sharedKey),
+      () => this.closeForOutboundBudget('queue'),
+      options.outboundMemoryBudget
+    )
 
     this.handshakeTimer = setTimeout(() => {
       this.onError(4002, 'E2EE handshake timeout')
@@ -89,7 +99,7 @@ export class E2EEChannel {
     handler: (
       plaintext: string,
       encryptedReply: (response: string) => void,
-      encryptedBinaryReply: (response: Uint8Array<ArrayBufferLike>) => boolean | void
+      encryptedBinaryReply: RpcBinarySender
     ) => void
   ): void {
     this.messageHandler = handler
@@ -154,13 +164,9 @@ export class E2EEChannel {
         this.closeForOutboundBudget('size')
         return
       }
-      this.outbound.enqueueLegacyText(
-        encrypt(response, this.sharedKey),
-        () => Boolean(this.sharedKey),
-        () => this.closeForOutboundBudget('queue')
-      )
+      this.outbound.enqueueLegacyText(encrypt(response, this.sharedKey))
     }
-    const encryptedBinaryReply = (response: Uint8Array<ArrayBufferLike>): boolean => {
+    const encryptedBinaryReply: RpcBinarySender = (response, options) => {
       if (!this.sharedKey || this.ws.readyState !== this.ws.OPEN) {
         return false
       }
@@ -168,11 +174,7 @@ export class E2EEChannel {
         this.closeForOutboundBudget('size')
         return false
       }
-      if (!this.outbound.canSend(response.byteLength + 40)) {
-        return false
-      }
-      this.ws.send(Buffer.from(encryptBytes(response, this.sharedKey)), { binary: true })
-      return true
+      return this.outbound.enqueueLegacyBinary(response, this.sharedKey, options)
     }
     this.messageHandler?.(plaintext, encryptedReply, encryptedBinaryReply)
   }
@@ -291,13 +293,13 @@ export class E2EEChannel {
         this.messageHandler?.(
           plaintext,
           (response) => this.enqueueV2({ kind: 'text', plaintext: response }),
-          (response) => this.enqueueV2({ kind: 'binary', plaintext: response })
+          (response, options) => this.enqueueV2({ kind: 'binary', plaintext: response }, options)
         ),
       onProtocolError: () => this.onError(4001, 'Invalid binary message before authentication')
     })
   }
 
-  private enqueueV2(item: V2OutboundItem): boolean {
+  private enqueueV2(item: V2OutboundItem, options?: RpcBinarySendOptions): RpcBinarySendResult {
     if (!this.v2Session || this.ws.readyState !== this.ws.OPEN) {
       return false
     }
@@ -305,7 +307,7 @@ export class E2EEChannel {
       this.closeForOutboundBudget('size')
       return false
     }
-    return this.outbound.enqueueV2(item, this.v2Session, () => this.closeForOutboundBudget('queue'))
+    return this.outbound.enqueueV2(item, this.v2Session, options)
   }
 
   // Why: this close kills the whole remote session. `size` means a producer emitted something
@@ -324,7 +326,7 @@ export class E2EEChannel {
       this.enqueueV2({ kind: 'text', plaintext: JSON.stringify(message) })
     } else if (this.ws.readyState === this.ws.OPEN && this.sharedKey) {
       const frame = encrypt(JSON.stringify(message), this.sharedKey)
-      this.outbound.sendLegacyFrame(frame, () => this.closeForOutboundBudget('queue'))
+      this.outbound.sendLegacyFrame(frame)
     }
   }
 

src/main/runtime/rpc/methods/terminal/terminal-multiplex-frame-delivery.ts

@@ -13,6 +13,7 @@ import {
 import type { TerminalOutputFrameChunk } from '../../terminal-output-frame-chunks'
 import type { TerminalStreamInputOutcome } from './terminal-input-delivery'
 import type { TerminalMultiplexStream } from './terminal-stream-types'
+import type { RpcBinarySendResult } from '../../rpc-binary-sender'
 
 export function installMultiplexFrameDelivery(state: TerminalMultiplexConnection): void {
   const { sendBinary, emit } = state
@@ -30,7 +31,7 @@ export function installMultiplexFrameDelivery(state: TerminalMultiplexConnection
     // Why: a seq-less Output chunk must carry sentinel 0, not the control-frame state.cursor, or it poisons the client's frame-drop tracker.
     const resolvedSeq =
       typeof seq === 'number' ? seq : opcode === TerminalStreamOpcode.Output ? 0 : state.cursor++
-    let sent: boolean | void
+    let sent: RpcBinarySendResult
     try {
       sent = sendBinary(encodeTerminalStreamFrame({ opcode, streamId, seq: resolvedSeq, payload }))
     } catch {

src/main/runtime/rpc/mobile-e2ee-desktop-outbound-owner.ts

@@ -1,8 +1,10 @@
 import type { WebSocket } from 'ws'
 import type { WsOutboundBackpressureQueue } from '../../../shared/ws-outbound-backpressure-queue'
-import { createLegacyMobileE2EETextReplyQueue } from './mobile-e2ee-outbound-admission'
+import { createLegacyMobileE2EEReplyQueue } from './mobile-e2ee-outbound-admission'
+import { encryptBytes } from './e2ee-crypto'
 import {
   createDesktopMobileE2EEV2OutboundQueue,
+  desktopMobileE2EEV2OutboundItemBytes,
   type DesktopMobileE2EEV2OutboundItem
 } from './mobile-e2ee-v2-desktop-outbound'
 import type { DesktopMobileE2EEV2Session } from './mobile-e2ee-v2-desktop-session'
@@ -11,56 +13,61 @@ import {
   type MobileE2EEOutboundMemoryBudget,
   type MobileE2EEOutboundSocketMemory
 } from './mobile-e2ee-outbound-memory-budget'
+import type { RpcBinarySendOptions } from './rpc-binary-sender'
 
 export class MobileE2EEDesktopOutboundOwner {
   private readonly memoryBudget: MobileE2EEOutboundMemoryBudget
   private readonly socketMemory: MobileE2EEOutboundSocketMemory | null
-  private legacyQueue: WsOutboundBackpressureQueue<string> | null = null
+  // Why: one queue for text and binary, so a binary frame can neither overtake parked text nor drop silently.
+  private legacyQueue: WsOutboundBackpressureQueue<string | Buffer> | null = null
   private v2Queue: WsOutboundBackpressureQueue<DesktopMobileE2EEV2OutboundItem> | null = null
 
   constructor(
     private readonly ws: WebSocket,
+    private readonly isLegacyKeyed: () => boolean,
+    private readonly onOverflow: () => void,
     memoryBudget: MobileE2EEOutboundMemoryBudget = createMobileE2EEOutboundMemoryBudget()
   ) {
     this.memoryBudget = memoryBudget
     this.socketMemory = memoryBudget.registerBufferedAmount(() => ws.bufferedAmount)
   }
 
-  canSend(bytes: number): boolean {
-    return this.socketMemory?.canSend(bytes) === true
-  }
-
-  sendLegacyFrame(frame: string, onOverflow: () => void): boolean {
-    if (!this.canSend(frame.length) || this.ws.readyState !== this.ws.OPEN) {
-      onOverflow()
+  sendLegacyFrame(frame: string): boolean {
+    if (this.socketMemory?.canSend(frame.length) !== true || this.ws.readyState !== this.ws.OPEN) {
+      this.onOverflow()
       return false
     }
     this.ws.send(frame)
     return true
   }
 
-  enqueueLegacyText(frame: string, isKeyed: () => boolean, onOverflow: () => void): boolean {
-    if (!this.socketMemory) {
-      onOverflow()
+  enqueueLegacyText(frame: string): boolean {
+    return this.legacy()?.enqueue(frame) ?? false
+  }
+
+  enqueueLegacyBinary(
+    bytes: Uint8Array<ArrayBufferLike>,
+    sharedKey: Uint8Array,
+    options?: RpcBinarySendOptions
+  ): boolean | 'backlogged' {
+    const queue = this.legacy()
+    if (!queue) {
       return false
     }
-    this.legacyQueue ??= createLegacyMobileE2EETextReplyQueue({
-      ws: this.ws,
-      isKeyed,
-      memoryBudget: this.memoryBudget,
-      socketMemory: this.socketMemory,
-      onOverflow
-    })
-    return this.legacyQueue.enqueue(frame)
+    // Why: idleness is judged on the sealed size (+40) before sealing, so a skipped frame costs no encryption.
+    if (options?.dropWhenBacklogged && !queue.isIdle(bytes.byteLength + 40)) {
+      return 'backlogged'
+    }
+    return queue.enqueue(Buffer.from(encryptBytes(bytes, sharedKey)))
   }
 
   enqueueV2(
     item: DesktopMobileE2EEV2OutboundItem,
     session: DesktopMobileE2EEV2Session,
-    onOverflow: () => void
-  ): boolean {
+    options?: RpcBinarySendOptions
+  ): boolean | 'backlogged' {
     if (!this.socketMemory) {
-      onOverflow()
+      this.onOverflow()
       return false
     }
     this.v2Queue ??= createDesktopMobileE2EEV2OutboundQueue({
@@ -68,11 +75,32 @@ export class MobileE2EEDesktopOutboundOwner {
       session,
       memoryBudget: this.memoryBudget,
       socketMemory: this.socketMemory,
-      onOverflow
+      onOverflow: this.onOverflow
     })
+    if (
+      options?.dropWhenBacklogged &&
+      !this.v2Queue.isIdle(desktopMobileE2EEV2OutboundItemBytes(item))
+    ) {
+      return 'backlogged'
+    }
     return this.v2Queue.enqueue(item)
   }
 
+  private legacy(): WsOutboundBackpressureQueue<string | Buffer> | null {
+    if (!this.socketMemory) {
+      this.onOverflow()
+      return null
+    }
+    this.legacyQueue ??= createLegacyMobileE2EEReplyQueue({
+      ws: this.ws,
+      isKeyed: this.isLegacyKeyed,
+      memoryBudget: this.memoryBudget,
+      socketMemory: this.socketMemory,
+      onOverflow: this.onOverflow
+    })
+    return this.legacyQueue
+  }
+
   dispose(): void {
     this.legacyQueue?.dispose()
     this.legacyQueue = null

src/main/runtime/rpc/mobile-e2ee-outbound-admission.ts

@@ -41,17 +41,18 @@ export function isMobileE2EEOutboundItemWithinLimit(
     : isMobileE2EEBinaryPayloadWithinLimit(item.plaintext)
 }
 
-export function createLegacyMobileE2EETextReplyQueue(args: {
+export function createLegacyMobileE2EEReplyQueue(args: {
   ws: WebSocket
   isKeyed: () => boolean
   onOverflow: () => void
   memoryBudget: MobileE2EEOutboundMemoryBudget
   socketMemory: MobileE2EEOutboundSocketMemory
-}): WsOutboundBackpressureQueue<string> {
-  return createWsOutboundBackpressureQueue<string>({
-    send: (frame) => args.ws.send(frame),
-    // Encrypted replies are base64 ASCII strings, so length === byte count.
-    byteLengthOf: (frame) => frame.length,
+}): WsOutboundBackpressureQueue<string | Buffer> {
+  return createWsOutboundBackpressureQueue<string | Buffer>({
+    send: (frame) =>
+      typeof frame === 'string' ? args.ws.send(frame) : args.ws.send(frame, { binary: true }),
+    // Encrypted text replies are base64 ASCII strings, so length === byte count.
+    byteLengthOf: (frame) => (typeof frame === 'string' ? frame.length : frame.byteLength),
     getBufferedAmount: () => args.ws.bufferedAmount,
     isWritable: () => args.isKeyed() && args.ws.readyState === args.ws.OPEN,
     canSend: (bytes) => args.socketMemory.canSend(bytes),

src/main/runtime/rpc/mobile-e2ee-v2-desktop-outbound.ts

@@ -17,6 +17,16 @@ export type DesktopMobileE2EEV2OutboundItem =
   | { kind: 'text'; plaintext: string }
   | { kind: 'binary'; plaintext: Uint8Array<ArrayBufferLike> }
 
+export function desktopMobileE2EEV2OutboundItemBytes(
+  item: DesktopMobileE2EEV2OutboundItem
+): number {
+  const bytes =
+    item.kind === 'text'
+      ? mobileE2EETextPayloadAdmissionBytes(item.plaintext)
+      : mobileE2EEBinaryPayloadAdmissionBytes(item.plaintext)
+  return Number.isFinite(bytes) ? bytes + 82 : bytes
+}
+
 export function createDesktopMobileE2EEV2OutboundQueue(args: {
   ws: WebSocket
   session: DesktopMobileE2EEV2Session
@@ -34,13 +44,7 @@ export function createDesktopMobileE2EEV2OutboundQueue(args: {
         args.ws.send(Buffer.from(args.session.sealBinary(item.plaintext)), { binary: true })
       }
     },
-    byteLengthOf: (item) => {
-      const bytes =
-        item.kind === 'text'
-          ? mobileE2EETextPayloadAdmissionBytes(item.plaintext)
-          : mobileE2EEBinaryPayloadAdmissionBytes(item.plaintext)
-      return Number.isFinite(bytes) ? bytes + 82 : bytes
-    },
+    byteLengthOf: desktopMobileE2EEV2OutboundItemBytes,
     getBufferedAmount: () => args.ws.bufferedAmount,
     isWritable: () => args.ws.readyState === args.ws.OPEN,
     canSend: (bytes) => args.socketMemory.canSend(bytes),

src/main/runtime/rpc/mobile-socket-wiring.ts

@@ -5,6 +5,7 @@ import type { E2EEKeypair } from '../e2ee-keypair'
 import { E2EEChannel, type E2EEAuthenticatedDevice } from './e2ee-channel'
 import { createMobileE2EEOutboundMemoryBudget } from './mobile-e2ee-outbound-memory-budget'
 import type { RuntimeCapability } from '../../../shared/protocol-version'
+import type { RpcBinarySender } from './rpc-binary-sender'
 
 type MobileSocketPayload = string | Uint8Array<ArrayBufferLike>
 
@@ -48,7 +49,7 @@ type MobileSocketWiringOptions = {
     socket: AuthenticatedMobileSocket,
     plaintext: string,
     reply: (response: string) => void,
-    sendBinary: (response: Uint8Array<ArrayBufferLike>) => boolean | void
+    sendBinary: RpcBinarySender
   ) => void
   onBinary: (socket: AuthenticatedMobileSocket, bytes: Uint8Array<ArrayBufferLike>) => void
   onClose: (socket: AuthenticatedMobileSocket | null, hasOtherConnections: boolean) => void

src/main/runtime/rpc/rpc-binary-sender.ts

@@ -0,0 +1,12 @@
+export type RpcBinarySendOptions = {
+  // Why: a lossy stream resends its newest frame later, so queueing stale frames behind a backlog only delays it.
+  dropWhenBacklogged?: boolean
+}
+
+/** False: not sent or queued. 'backlogged': skipped under dropWhenBacklogged; the socket is alive. */
+export type RpcBinarySendResult = boolean | void | 'backlogged'
+
+export type RpcBinarySender = (
+  bytes: Uint8Array<ArrayBufferLike>,
+  options?: RpcBinarySendOptions
+) => RpcBinarySendResult

src/main/runtime/runtime-browser-commands-browser-click.ts

@@ -20,6 +20,7 @@ import type {
   BrowserTypeResult
 } from '../../shared/runtime-types'
 import type { BrowserScreencastSession } from '../browser/browser-screencast-stream-types'
+import { cancelScreencastSubscriberFrameRetry } from './browser-screencast-subscriber-frame-delivery'
 
 export class RuntimeBrowserCommandsWithBrowserClick extends RuntimeBrowserCommandsWithActiveScreencastsByPageId {
   async browserClick(
@@ -152,6 +153,7 @@ export class RuntimeBrowserCommandsWithBrowserClick extends RuntimeBrowserComman
       return
     }
     active.subscribers.delete(subscriptionId)
+    cancelScreencastSubscriberFrameRetry(subscriber)
     subscriber.resolveDone()
     if (active.viewportOwnerSubscriptionId === subscriptionId) {
       const fallback = Array.from(active.subscribers.entries()).findLast(([, candidate]) =>

src/main/runtime/runtime-browser-commands-browser-command-target-params.ts

@@ -18,6 +18,7 @@ import type { RuntimeBrowserPageRegistry } from './runtime-browser-page-registry
 import type { BrowserWindow } from 'electron'
 import type { BrowserBackend } from '../browser/browser-backend'
 import type { BrowserSessionTabSelectionOptions } from './browser-tab-create-publication'
+import type { RpcBinarySender } from './rpc/rpc-binary-sender'
 
 export type BrowserCommandTargetParams = {
   worktree?: string
@@ -58,13 +59,14 @@ export type BrowserScreencastStartResult = {
 }
 
 export type ActiveBrowserScreencastSubscriber = {
-  sendBinary: (bytes: Uint8Array<ArrayBufferLike>) => boolean | void
+  sendBinary: RpcBinarySender
   emit?: (event: BrowserScreencastResult) => void
   done: Promise<void>
   resolveDone: () => void
   viewport: BrowserScreencastViewport
   budget: BrowserScreencastFrameBudget
   pendingFrame: Uint8Array<ArrayBufferLike> | null
+  pendingFrameRetry: ReturnType<typeof setTimeout> | null
   // Why: identifies the viewer across reconnects, which the RPC connectionId cannot — a new
   // socket never reuses the old id, so a reconnecting device would stack a second subscription.
   pairedDeviceId?: string

src/main/runtime/runtime-browser-commands-browser-screencast.ts

@@ -18,19 +18,22 @@ import { BrowserError } from '../browser/browser-error'
 import { randomUUID } from 'node:crypto'
 import { startBrowserScreencast } from '../browser/browser-screencast-stream'
 import { rendererPublicationThrottle } from '../window/renderer-publication-throttle'
-import { sendRemoteBrowserScreencastFrame } from './remote-browser-screencast-frame-admission'
+import {
+  cancelScreencastSubscriberFrameRetry,
+  deliverScreencastSubscriberFrame
+} from './browser-screencast-subscriber-frame-delivery'
 import {
   INITIAL_SCREENCAST_SUBSCRIBER_DELIVERY,
-  recordScreencastSubscriberSend,
   screencastSubscriberIsGhost
 } from './browser-screencast-ghost-subscriber-eviction'
 import type { BrowserScreencastSession } from '../browser/browser-screencast-stream-types'
+import type { RpcBinarySender } from './rpc/rpc-binary-sender'
 
 export class RuntimeBrowserCommandsWithBrowserScreencast extends RuntimeBrowserCommandsWithBrowserClick {
   async browserScreencast(
     params: BrowserScreencastParams,
     stream: {
-      sendBinary: (bytes: Uint8Array<ArrayBufferLike>) => boolean | void
+      sendBinary: RpcBinarySender
       emit?: (event: BrowserScreencastResult) => void
       pairedDeviceId?: string
     }
@@ -82,11 +85,8 @@ export class RuntimeBrowserCommandsWithBrowserScreencast extends RuntimeBrowserC
         onFrame: (bytes) => {
           const ghosts: string[] = []
           for (const [subscriptionId, subscriber] of record.subscribers) {
-            // A slow viewer drops this frame without stalling every other viewer, but the
-            // newest refusal is retained so a gate that opens later can still be filled.
-            const delivered = sendRemoteBrowserScreencastFrame(subscriber.sendBinary, bytes)
-            subscriber.pendingFrame = delivered ? null : bytes
-            subscriber.delivery = recordScreencastSubscriberSend(subscriber.delivery, delivered)
+            // A slow viewer drops this frame without stalling every other viewer.
+            deliverScreencastSubscriberFrame(subscriber, bytes)
             if (screencastSubscriberIsGhost(subscriber.delivery)) {
               ghosts.push(subscriptionId)
             }
@@ -131,6 +131,7 @@ export class RuntimeBrowserCommandsWithBrowserScreencast extends RuntimeBrowserC
             this.activeScreencastsByPageId.delete(browserPageId)
           }
           for (const subscriber of record.subscribers.values()) {
+            cancelScreencastSubscriberFrameRetry(subscriber)
             subscriber.resolveDone()
           }
           record.subscribers.clear()
@@ -145,6 +146,7 @@ export class RuntimeBrowserCommandsWithBrowserScreencast extends RuntimeBrowserC
       viewport,
       budget,
       pendingFrame: null,
+      pendingFrameRetry: null,
       pairedDeviceId: stream.pairedDeviceId,
       delivery: INITIAL_SCREENCAST_SUBSCRIBER_DELIVERY
     })
@@ -180,17 +182,16 @@ export class RuntimeBrowserCommandsWithBrowserScreencast extends RuntimeBrowserC
     }
     return {
       subscriptionId,
+      // Why: the frame the pre-ready gate refused goes out with ready, not a retry tick later.
       flushPendingFrame: () => {
         const subscriber = active.subscribers.get(subscriptionId)
         const bytes = subscriber?.pendingFrame
         if (!subscriber || !bytes) {
           return
         }
-        const delivered = sendRemoteBrowserScreencastFrame(subscriber.sendBinary, bytes)
-        subscriber.pendingFrame = delivered ? null : bytes
         // The replay is this subscriber's first chance to reach its socket, so it is also where
         // an eviction-eligible delivery history starts.
-        subscriber.delivery = recordScreencastSubscriberSend(subscriber.delivery, delivered)
+        deliverScreencastSubscriberFrame(subscriber, bytes)
       },
       session: {
         done: subscriberDone,

src/main/runtime/runtime-browser-screencast-controller.ts

@@ -6,6 +6,7 @@ import {
   type BrowserScreencastSubscriber
 } from './browser-screencast-driver-scope'
 import type { RuntimeBrowserCommands } from './orca-runtime-browser'
+import type { RpcBinarySender } from './rpc/rpc-binary-sender'
 
 type RuntimeBrowserScreencastControllerDeps = {
   getCommands: () => RuntimeBrowserCommands
@@ -61,7 +62,7 @@ export class RuntimeBrowserScreencastController {
       connectionId?: string
       pairedDeviceId?: string
       clientKind?: 'mobile' | 'runtime'
-      sendBinary?: (bytes: Uint8Array<ArrayBufferLike>) => boolean | void
+      sendBinary?: RpcBinarySender
       signal?: AbortSignal
       emit: (result: BrowserScreencastResult) => void
     }
@@ -114,11 +115,11 @@ export class RuntimeBrowserScreencastController {
       end(emitEnd)
     }
     const abortScreencast = (): void => cancel()
-    const sendBinaryAfterReady = (bytes: Uint8Array<ArrayBufferLike>): boolean | void => {
+    const sendBinaryAfterReady: RpcBinarySender = (bytes, sendOptions) => {
       if (!readyEmitted) {
         return false
       }
-      return options.sendBinary?.(bytes)
+      return options.sendBinary?.(bytes, sendOptions)
     }
 
     // Why: rotation can happen before ready, so replacements are connection-scoped immediately.

src/main/runtime/runtime-edge-command-controller.ts

@@ -4,6 +4,7 @@ import { RuntimeEmulatorCommands } from './orca-runtime-emulator'
 import { RuntimeBrowserScreencastController } from './runtime-browser-screencast-controller'
 import { createRuntimeBrowserCommands } from './runtime-browser-commands-factory'
 import { RuntimeJiraCommands } from './runtime-jira-commands'
+import type { RpcBinarySender } from './rpc/rpc-binary-sender'
 
 type PublicMethods<T> = Pick<T, keyof T>
 type BrowserSurface = Omit<PublicMethods<RuntimeBrowserCommands>, 'browserScreencast'> & {
@@ -13,7 +14,7 @@ type BrowserSurface = Omit<PublicMethods<RuntimeBrowserCommands>, 'browserScreen
       connectionId?: string
       pairedDeviceId?: string
       clientKind?: 'mobile' | 'runtime'
-      sendBinary?: (bytes: Uint8Array<ArrayBufferLike>) => boolean | void
+      sendBinary?: RpcBinarySender
       signal?: AbortSignal
       emit: (result: BrowserScreencastResult) => void
     }

src/main/runtime/runtime-rpc/runtime-rpc-websocket-dispatch.ts

@@ -10,6 +10,7 @@ import type { WebSocketTransport } from '../rpc/ws-transport'
 import type { DeviceScope } from '../device-registry'
 import { RuntimeRpcMobileDevices } from './runtime-rpc-mobile-devices'
 import { classifyRuntimeLongPoll } from './runtime-rpc-long-poll'
+import type { RpcBinarySender } from '../rpc/rpc-binary-sender'
 import type { RpcCallerScope } from '../rpc/rpc-caller-scope'
 import { limitRuntimeRpcReplySize } from './runtime-rpc-reply-size-limit'
 
@@ -32,7 +33,7 @@ export class RuntimeRpcWebSocketDispatch extends RuntimeRpcMobileDevices {
   protected async handleWebSocketMessage(
     rawMessage: string,
     reply: (response: string) => void,
-    sendBinary: (response: Uint8Array<ArrayBufferLike>) => boolean | void,
+    sendBinary: RpcBinarySender,
     wsTransport?: WebSocketTransport,
     ws?: WebSocket,
     authenticatedDeviceToken?: string | null,

src/main/runtime/single-page-browser-commands-host-test-double.ts

@@ -9,6 +9,15 @@ export type AuthoritativeWindowDouble = {
   webContents: RendererPublicationThrottleTarget
 }
 
+// A guest the page stream keeps painting while it streams.
+export function createStreamedGuest(): RendererPublicationThrottleTarget {
+  return {
+    isDestroyed: () => false,
+    setBackgroundThrottling: vi.fn(),
+    capturePage: vi.fn(async () => null)
+  }
+}
+
 /** A browser command host whose worktree `wt-1` has one registered page, `page-1`. */
 export function createSinglePageBrowserCommandsHost(
   window: AuthoritativeWindowDouble | null = null

src/shared/ws-outbound-backpressure-queue.ts

@@ -62,6 +62,8 @@ export type WsOutboundBackpressureQueue<TFrame> = {
   enqueue: (frame: TFrame) => boolean
   /** Queue-or-send a frame and allow its owner to cancel it before wire delivery. */
   enqueueCancelable: (frame: TFrame) => WsOutboundEnqueueResult
+  /** True when nothing is parked and the wire has room, so enqueue would send this size directly. */
+  isIdle: (frameBytes: number) => boolean
   /** Bytes currently held (not yet handed to the wire). */
   queuedBytes: () => number
   evidence: () => { queuedBytes: number; queuedFrames: number; storageSlots: number }
@@ -257,6 +259,12 @@ export function createWsOutboundBackpressureQueue<TFrame>(
     }
   }
 
+  const isIdle = (bytes: number): boolean =>
+    queuedFrames === 0 &&
+    options.isWritable() &&
+    bufferedAmount() <= softCapBytes &&
+    (options.canSend?.(bytes, false) ?? true)
+
   const enqueueCancelable = (frame: TFrame): WsOutboundEnqueueResult => {
     if (disposed || overflowed) {
       return { accepted: false, queued: false, cancel: () => false }
@@ -267,12 +275,7 @@ export function createWsOutboundBackpressureQueue<TFrame>(
       return { accepted: false, queued: false, cancel: () => false }
     }
     // Fast path: nothing parked and the wire is under the cap — send directly.
-    if (
-      queuedFrames === 0 &&
-      options.isWritable() &&
-      bufferedAmount() <= softCapBytes &&
-      (options.canSend?.(bytes, false) ?? true)
-    ) {
+    if (isIdle(bytes)) {
       return {
         accepted: sendFrame(frame),
         queued: false,
@@ -312,6 +315,7 @@ export function createWsOutboundBackpressureQueue<TFrame>(
       return enqueueCancelable(frame).accepted
     },
     enqueueCancelable,
+    isIdle,
     queuedBytes: () => queued,
     evidence: () => ({
       queuedBytes: queued,