test: tighten transport, recovery, and diagnostic checks (#33559)

* test: tighten transport, recovery, and diagnostic checks * test: use at for last mock socket access --------- Co-authored-by: OrcaWin <293788423+OrcaWin@users.noreply.github.com>
2026-10-30 16:47:34 +00:00 · 2026-10-25 08:11:25 -07:00
parent 33c76d89f1
commit cc0e35ebcc
19 changed files with 319 additions and 210 deletions
@@ -37,6 +37,7 @@ export function createFixturePageRunner(baseUrl, browserPromise) {
      workers.set(scale, createWorker(scale))
    }
    const worker = await workers.get(scale)
+    worker.pageErrors = []
    if (worker.revision !== sourceRevision) {
      const revision = sourceRevision
      await worker.page.goto(`${baseUrl}/fixtures.html?worker=1`)
@@ -47,7 +48,6 @@ export function createFixturePageRunner(baseUrl, browserPromise) {
      )
      worker.revision = revision
    }
-    worker.pageErrors = []
    return worker
  }

@@ -90,7 +90,10 @@ async function screenshotFixtureTool(args) {
      content.push({ type: 'image', data, mimeType: 'image/png' })
    }
  }
-  return { content, isError: payload.results.some((result) => result.error) }
+  return {
+    content,
+    isError: payload.results.some((result) => result.error || result.pageErrors?.length)
+  }
 }

 async function handleRequest(message) {
@@ -52,6 +52,12 @@ async function main() {
    scale: Number(values.scale),
    outDir: values.out ? path.resolve(values.out) : undefined
  })
+  if (
+    payload.error ||
+    payload.results?.some((result) => result.error || result.pageErrors?.length)
+  ) {
+    process.exitCode = 1
+  }
  if (values.json) {
    console.log(JSON.stringify({ ...payload, totalMs: Math.round(performance.now() - startedAt) }))
    return
@@ -99,6 +99,11 @@ function runOracle(label, targetRoot, operation, userDataPath, options = {}) {
    }
  })
  const output = `${result.stdout ?? ''}${result.stderr ?? ''}`
+  if (result.error || result.signal || result.status === null) {
+    throw new Error(
+      `${label} could not complete (${result.error?.message ?? result.signal ?? 'no exit status'})\n${output}`
+    )
+  }
  if (options.expectFailure) {
    if (result.status === 0 || !output.includes(options.expectFailure)) {
      throw new Error(
@@ -21,7 +21,7 @@ const ORCAD_ENTRY = join(ORCAD_DIR, ORCAD_SERVER_ENTRY_FILENAME)
 const READY_TIMEOUT_MS = 120_000
 const OUTPUT_TIMEOUT_MS = 30_000
 const SHUTDOWN_TIMEOUT_MS = 15_000
-// Why a random high port: a fixed one collides with a developer's own `orca serve`.
+// Offset the port when a developer's own `orca serve` is already using it.
 const PORT = 6800 + Math.floor(Number(process.env.ORCA_SMOKE_PORT_OFFSET ?? '0'))

 function log(message) {
@@ -66,6 +66,11 @@ function orca(pairingCode, args) {
  if (result.error) {
    throw new Error(`orca ${args[0]} failed to spawn: ${result.error.message}`)
  }
+  if (result.status !== 0) {
+    throw new Error(
+      `orca ${args.join(' ')} exited ${result.signal ?? result.status}:\n${result.stderr || result.stdout}`
+    )
+  }
  const line = (result.stdout ?? '').trim()
  if (!line.startsWith('{')) {
    throw new Error(`orca ${args.join(' ')} produced no JSON:\n${result.stdout}\n${result.stderr}`)
@@ -311,18 +316,18 @@ async function main() {
    log(`created ${terminal.handle}`)

    // Why invoke node rather than `echo`: the shell differs per platform, node does not.
-    const nonce = `ORCA_SMOKE_${randomBytes(8).toString('hex')}`
+    const nonce = randomBytes(8).toString('hex')
    orca(pairingCode, [
      'terminal',
      'send',
      '--terminal',
      terminal.handle,
      '--text',
-      `"${process.execPath}" -e "console.log('${nonce}')"`,
+      `"${process.execPath}" -e "console.log('ORCA_SMOKE_' + '${nonce}')"`,
      '--enter'
    ])

-    if (!(await waitForNonce(pairingCode, terminal.handle, nonce))) {
+    if (!(await waitForNonce(pairingCode, terminal.handle, `ORCA_SMOKE_${nonce}`))) {
      throw new Error(
        `terminal produced no output containing ${nonce} within ${OUTPUT_TIMEOUT_MS}ms — ` +
          `the server started and answered RPC, but its PTY path is dead`
@@ -280,6 +280,9 @@ async function main() {
    log('SKIP: Windows ConPTY is required')
    return
  }
+  if (!Number.isSafeInteger(iterations) || iterations <= 0) {
+    throw new Error('ORCA_WINDOWS_DAEMON_CLOSE_ITERATIONS must be a positive safe integer')
+  }
  if (!existsSync(entryPath)) {
    throw new Error(`Missing ${entryPath}; run pnpm build:electron-vite first`)
  }
@@ -55,6 +55,11 @@ async function authenticateSession(onLog?: ConnectionLogSink) {
  fakes.linkOptions!.onAuthenticated()
  await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledOnce())
  const confirmation = sentRequests()[0]!
+  expect(confirmation).toMatchObject({
+    method: 'pairing.getEndpoints',
+    params: { resumeConfirmReqId: 'confirm-1' },
+    deviceToken: 'device-token'
+  })
  fakes.linkOptions!.onText(
    JSON.stringify({
      id: confirmation.id,
@@ -76,6 +81,11 @@ async function authenticateSession(onLog?: ConnectionLogSink) {
  )
  await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledTimes(2))
  const capabilities = sentRequests()[1]!
+  expect(capabilities).toMatchObject({
+    method: 'runtime.clientCapabilities.update',
+    params: { clientCapabilities: expect.arrayContaining(['agent-session.structured.v1']) },
+    deviceToken: 'device-token'
+  })
  fakes.linkOptions!.onText(
    JSON.stringify({
      id: capabilities.id,
@@ -89,10 +99,21 @@ async function authenticateSession(onLog?: ConnectionLogSink) {
  return session
 }

-function sentRequests(): Array<{ id: string; method: string }> {
-  return fakes.sendText.mock.calls.map(([value]): { id: string; method: string } =>
-    JSON.parse(value)
-  )
+function sentRequests() {
+  return fakes.sendText.mock.calls.map(([value]) => {
+    const request: unknown = JSON.parse(value)
+    if (
+      !request ||
+      typeof request !== 'object' ||
+      !('id' in request) ||
+      typeof request.id !== 'string' ||
+      !('method' in request) ||
+      typeof request.method !== 'string'
+    ) {
+      throw new Error('Expected a serialized RPC request')
+    }
+    return request
+  })
 }

 describe('mobile relay RPC session liveness', () => {
@@ -203,11 +203,15 @@ describe('mobile relay RPC session', () => {
  })

  it('connects when the relay never answers capability negotiation', async () => {
+    vi.useFakeTimers()
    const { session } = await confirmResume()
+    expect(session.getState()).toBe('handshaking')

-    // An unanswered capability advisory must not trap a slow link before connected.
-    await vi.waitFor(() => expect(session.getState()).toBe('connected'), { timeout: 5_000 })
+    await vi.advanceTimersByTimeAsync(1_000)
+
+    expect(session.getState()).toBe('connected')
    expect(session.getFailure()).toBeNull()
+    session.close()
  })

  // Dial stages distinguish a missing upgrade from slow cell assignment while both report connecting.
@@ -333,7 +337,10 @@ describe('mobile relay RPC session', () => {
      .sendRequest('terminal.send', { terminal: 'term', text: 'hi' })
      .catch((cause: unknown) => cause)

-    expect((error as Error).message).toBe('relay outbound overflow')
+    if (!(error instanceof Error)) {
+      throw new Error('Expected the relay failure')
+    }
+    expect(error.message).toBe('relay outbound overflow')
    expect(isRpcDeliveryUnknown(error)).toBe(false)
  })

@@ -352,17 +359,32 @@ describe('mobile relay RPC session', () => {
    vi.useFakeTimers()
    try {
      const pending = session.sendRequest('terminal.send', { terminal: 'term', text: 'hi' })
-      const outcome = pending.catch((error: unknown) => ({
-        message: (error as Error).message,
-        unknown: isRpcDeliveryUnknown(error)
-      }))
+      let settled = false
+      const outcome = pending.then(
+        (response) => {
+          settled = true
+          return response
+        },
+        (error: unknown) => {
+          settled = true
+          return error
+        }
+      )
      // Let sendRequest pass its connected-check microtask and register the timer.
      await vi.advanceTimersByTimeAsync(0)
-      await vi.advanceTimersByTimeAsync(1_000)
-      await expect(outcome).resolves.toEqual({
-        message: 'relay RPC timed out: terminal.send',
-        unknown: true
-      })
+      expect(fakes.sendText).toHaveBeenCalledExactlyOnceWith(
+        expect.stringContaining('"method":"terminal.send"')
+      )
+      await vi.advanceTimersByTimeAsync(999)
+      expect(settled).toBe(false)
+      await vi.advanceTimersByTimeAsync(1)
+      const error = await outcome
+      if (!(error instanceof Error)) {
+        throw new Error('Expected the relay timeout')
+      }
+      expect(error.message).toBe('relay RPC timed out: terminal.send')
+      expect(isRpcDeliveryUnknown(error)).toBe(true)
+      session.close()
    } finally {
      vi.useRealTimers()
    }
@@ -1,9 +1,7 @@
 import { describe, expect, it } from 'vitest'
 import { MOBILE_WEB_BUNDLE_CAPABILITY } from '../../../src/shared/mobile-web-bundle/mobile-web-bundle-capability'
-import { MOBILE_WEB_BUNDLE_SCHEMA_VERSION } from '../../../src/shared/mobile-web-bundle/manifest-contract'
 import {
  evaluateMobileWebBundleCompat,
-  SUPPORTED_MOBILE_WEB_BUNDLE_SCHEMA_VERSIONS,
  type MobileWebBundleCompatManifest,
  type MobileWebBundleCompatVerdict,
  type MobileWebBundleHostStatus
@@ -175,10 +173,6 @@ describe('evaluateMobileWebBundleCompat', () => {
      })
    ).toMatchObject({ reason: 'bundle-incompatible', side: 'mobile' })
  })
-
-  it('supports the schema the desktop writes today, so a current bundle opens', () => {
-    expect(SUPPORTED_MOBILE_WEB_BUNDLE_SCHEMA_VERSIONS).toContain(MOBILE_WEB_BUNDLE_SCHEMA_VERSION)
-  })
 })

 // Pin the page-version floor independently of manifest-schema support.
@@ -1,5 +1,5 @@
 import { sha256 } from '@noble/hashes/sha256'
-import { describe, expect, it, vi } from 'vitest'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import { computeMobileWebBundleId } from '../../../src/shared/mobile-web-bundle/manifest-contract'
 import { fetchMobileWebBundle } from './mobile-web-bundle-fetch'
 import { MobileWebBundleFetchError } from './mobile-web-bundle-fetch-refusal'
@@ -119,7 +119,7 @@ function waveHost(files: Record<string, string>, options: WaveHostOptions = {})
    close: vi.fn()
  }

-  const settleMicrotasks = () => new Promise((resolve) => setTimeout(resolve, 0))
+  const settleMicrotasks = () => vi.advanceTimersByTimeAsync(0)

  // Drain after settlement too so unstopped sibling reads remain visible.
  const runWaves = async (
@@ -168,6 +168,9 @@ function syntheticBundle(): Record<string, string> {
  }
 }

+beforeEach(() => vi.useFakeTimers())
+afterEach(() => vi.useRealTimers())
+
 describe('fetchMobileWebBundle chunk pipeline', () => {
  it('keeps four chunk reads in flight across the whole bundle, largest asset first', async () => {
    const files = syntheticBundle()
@@ -174,15 +174,6 @@ function rangeHost(
    calls,
    wireBase64Bytes: () => wireBase64Bytes,
    peakInFlight: () => peakInFlight,
-    // Drain to quiescence so missing sibling stops become extra requests instead of hidden pending work.
-    drain: async (): Promise<void> => {
-      let quiet = 0
-      for (let tick = 0; tick < 2000 && quiet < 10; tick += 1) {
-        const sent = calls.length
-        await new Promise((resolve) => setTimeout(resolve, 0))
-        quiet = inFlight === 0 && calls.length === sent ? quiet + 1 : 0
-      }
-    },
    inflations: (): Inflation[] =>
      inflationLog
        .filter((entry) => sentGzipBodies.has(encodeBase64(entry.body)))
@@ -285,26 +276,32 @@ describe('fetchMobileWebBundle from a host whose manifest names a range grid', (

  // Bound a gzip bomb at one byte over its slot; enough planned reads expose continued dispatch after refusal.
  it('refuses a gzip bomb at one byte over the window and sends little after it', async () => {
-    const bomb = encodeBase64(gzipSync(new Uint8Array(4 * 1024 * 1024), { level: 9 }))
-    const many = Object.fromEntries(
-      Array.from({ length: 10 }, (_, index) => [
-        `assets/part-${String(index)}.js`,
-        scriptBytes(RANGE_GRID * 2, index)
-      ])
-    )
-    const host = rangeHost(many, {
-      tamper: (call, body) =>
-        call.params.path === 'assets/part-0.js' && call.params.offset === 0 ? bomb : body
-    })
+    vi.useFakeTimers()
+    try {
+      const bomb = encodeBase64(gzipSync(new Uint8Array(4 * 1024 * 1024), { level: 9 }))
+      const many = Object.fromEntries(
+        Array.from({ length: 10 }, (_, index) => [
+          `assets/part-${String(index)}.js`,
+          scriptBytes(RANGE_GRID * 2, index)
+        ])
+      )
+      const host = rangeHost(many, {
+        tamper: (call, body) =>
+          call.params.path === 'assets/part-0.js' && call.params.offset === 0 ? bomb : body
+      })

-    expect(await refusalOf(fetchMobileWebBundle({ client: host.client }))).toBe('chunk-oversize')
-    await host.drain()
-    const window = RANGE_GRID + 1
-    expect(host.inflations().filter((inflation) => inflation.resultLength === window)).toEqual([
-      { outLength: window, resultLength: window }
-    ])
-    // Sibling stop permits only in-flight work and immediate follow-ups, never the full planned grid.
-    expect(readsOf(host.calls, 'mobileWeb.bundle.range').length).toBeLessThanOrEqual(8)
+      const refusal = refusalOf(fetchMobileWebBundle({ client: host.client }))
+      await vi.runAllTimersAsync()
+      expect(await refusal).toBe('chunk-oversize')
+      const window = RANGE_GRID + 1
+      expect(host.inflations().filter((inflation) => inflation.resultLength === window)).toEqual([
+        { outLength: window, resultLength: window }
+      ])
+      // Sibling stop permits only in-flight work and immediate follow-ups, never the full planned grid.
+      expect(readsOf(host.calls, 'mobileWeb.bundle.range').length).toBeLessThanOrEqual(8)
+    } finally {
+      vi.useRealTimers()
+    }
  })

  it('refuses a range that inflates short of its window as short', async () => {
@@ -181,6 +181,9 @@ describe('fetchMobileWebBundle over a slow link', () => {

    expect(isMobileWebBundleTransportFailure(outcome.error)).toBe(true)
    expect(String(outcome.error)).toContain('stalled')
+    expect(outcome.settledAt - startedAt).toBe(
+      2 * ONE_WAY_LATENCY_MS + MOBILE_WEB_BUNDLE_WINDOW_STALL_MS
+    )
    // Fail against fetch-wide inactivity, not the per-read backstop.
    expect(outcome.settledAt - startedAt).toBeLessThan(MOBILE_WEB_BUNDLE_WINDOW_STALL_MS + 1_000)
    expect(host.delivered()).toBe(1)
@@ -139,13 +139,6 @@ function chunkCallCount(calls: readonly HostCall[]): number {
  return calls.filter((call) => call.method === 'mobileWeb.bundle.chunk').length
 }

-// Drain enough ticks to expose a worker pool missing its stop check.
-async function drainPendingHostWork(): Promise<void> {
-  for (let tick = 0; tick < 300; tick += 1) {
-    await new Promise((resolve) => setTimeout(resolve, 0))
-  }
-}
-
 describe('fetchMobileWebBundle', () => {
  it('pages every asset to eof and returns the verified bytes', async () => {
    const host = bundleHost({ 'index.html': '<h1>orca</h1>', 'assets/app.js': 'x=1' })
@@ -438,35 +431,41 @@ describe('fetchMobileWebBundle', () => {
  })

  it('stops the other reads mid-asset once one chunk is refused', async () => {
-    const host = bundleHost(
-      {
-        'a.js': 'x',
-        'b.js': 'b'.repeat(40),
-        'c.js': 'c'.repeat(40),
-        'd.js': 'd'.repeat(40)
-      },
-      {
-        chunkBytes: 1,
-        intercept: (call) =>
-          call.method === 'mobileWeb.bundle.chunk' &&
-          paramField(call.params, 'path') === 'b.js' &&
-          paramField(call.params, 'offset') === 1
-            ? new Error('mobile_web_bundle_asset_unknown')
-            : undefined
-      }
-    )
+    vi.useFakeTimers()
+    try {
+      const host = bundleHost(
+        {
+          'a.js': 'x',
+          'b.js': 'b'.repeat(40),
+          'c.js': 'c'.repeat(40),
+          'd.js': 'd'.repeat(40)
+        },
+        {
+          chunkBytes: 1,
+          intercept: (call) =>
+            call.method === 'mobileWeb.bundle.chunk' &&
+            paramField(call.params, 'path') === 'b.js' &&
+            paramField(call.params, 'offset') === 1
+              ? new Error('mobile_web_bundle_asset_unknown')
+              : undefined
+        }
+      )

-    const error = await fetchMobileWebBundle({ client: host.client }).catch(
-      (thrown: unknown) => thrown
-    )
-    const atRejection = chunkCallCount(host.calls)
-    await drainPendingHostWork()
+      const outcome = fetchMobileWebBundle({ client: host.client }).then(
+        () => ({ error: undefined, atRejection: chunkCallCount(host.calls) }),
+        (error: unknown) => ({ error, atRejection: chunkCallCount(host.calls) })
+      )
+      await vi.runAllTimersAsync()
+      const { error, atRejection } = await outcome

-    // The refusal is what the caller sees; the internal stop never surfaces.
-    expect(readMobileWebBundleErrorCode(error)).toBe('mobile_web_bundle_asset_unknown')
-    // Stop checks allow at most another concurrency window, never all remaining chunks.
-    expect(chunkCallCount(host.calls)).toBeLessThanOrEqual(atRejection + 4)
-    expect(chunkCallCount(host.calls)).toBeLessThan(10)
+      // The refusal is what the caller sees; the internal stop never surfaces.
+      expect(readMobileWebBundleErrorCode(error)).toBe('mobile_web_bundle_asset_unknown')
+      // Stop checks allow at most another concurrency window, never all remaining chunks.
+      expect(chunkCallCount(host.calls)).toBeLessThanOrEqual(atRejection + 4)
+      expect(chunkCallCount(host.calls)).toBeLessThan(10)
+    } finally {
+      vi.useRealTimers()
+    }
  })

  it('asks for nothing at all when the caller arrives already aborted', async () => {
@@ -1,11 +1,7 @@
 import { describe, expect, it } from 'vitest'
-import {
-  MOBILE_WEB_BUNDLE_CHUNK_BYTES,
-  MOBILE_WEB_BUNDLE_ERROR_CODES
-} from '../../../src/shared/mobile-web-bundle/bundle-rpc-contract'
+import { MOBILE_WEB_BUNDLE_CHUNK_BYTES } from '../../../src/shared/mobile-web-bundle/bundle-rpc-contract'
 import {
  computeMobileWebBundleId,
-  MobileWebBundleAssetSchema,
  MobileWebBundleRouteSchema,
  MOBILE_WEB_BUNDLE_MAX_ASSETS,
  MOBILE_WEB_BUNDLE_MAX_ASSET_BYTES,
@@ -25,6 +21,9 @@ import {
  MobileWebBundleManifestReplySchema
 } from './mobile-web-bundle-reply-schemas'
 import type { RpcReadResult } from './rpc-operation-contract'
+import { runRpcOperation } from './rpc-operation'
+import { RpcIncompatibleReplyError } from './rpc-incompatible-reply-error'
+import type { RpcResponse } from './types'

 const ASSET_SHA = 'b'.repeat(64)
 const MAX_DATA_BASE64_LENGTH = Math.ceil(MOBILE_WEB_BUNDLE_CHUNK_BYTES / 3) * 4 + 8
@@ -40,9 +39,8 @@ function asset(overrides: Record<string, unknown> = {}) {
 }

 // Derive ids after fixture asset edits unless the test explicitly selects its own buildId.
-function contentIdOf(assets: unknown): string {
-  const parsed = MobileWebBundleAssetSchema.array().safeParse(assets)
-  return parsed.success ? computeMobileWebBundleId(parsed.data) : 'a'.repeat(64)
+function contentIdOf(assets: ReturnType<typeof asset>[]): string {
+  return computeMobileWebBundleId(assets)
 }

 const BUILD_ID = contentIdOf([asset()])
@@ -331,9 +329,15 @@ describe('mobile web bundle chunk reply reader', () => {
 })

 describe('mobile web bundle error codes', () => {
-  it('maps every code the host declares, from both positions one can occupy', () => {
-    expect(MOBILE_WEB_BUNDLE_ERROR_CODES).toHaveLength(6)
-    for (const code of MOBILE_WEB_BUNDLE_ERROR_CODES) {
+  it('reads released wire codes bare, wrapped, and followed by diagnostic prose', () => {
+    for (const code of [
+      'mobile_web_bundle_unavailable',
+      'mobile_web_bundle_build_changed',
+      'mobile_web_bundle_asset_unknown',
+      'mobile_web_bundle_asset_changed',
+      'mobile_web_bundle_offset_invalid',
+      'mobile_web_bundle_read_limited'
+    ]) {
      expect(readMobileWebBundleErrorCode(new Error(`invalid_argument: ${code}`))).toBe(code)
      expect(readMobileWebBundleErrorCode(new Error(code))).toBe(code)
      expect(readMobileWebBundleErrorCode(new Error(`invalid_argument: ${code}: no bundle`))).toBe(
@@ -378,11 +382,29 @@ describe('mobile web bundle error codes', () => {
 })

 describe('mobile web bundle operation descriptors', () => {
-  it('accepts no reply but a result', () => {
-    // Whole-result policy must refuse immediately rather than salvage a corrupt partial bundle.
-    for (const operation of [mobileWebBundleManifestRead, mobileWebBundleChunkRead]) {
-      expect(operation.acceptance).toBe('require-result-or-throw')
-      expect(operation.barrier).toBe('on-settle')
+  it('rejects host refusals and unreadable successful results for both reads', async () => {
+    let reply: RpcResponse = {
+      id: 'read-1',
+      ok: false,
+      error: { code: 'invalid_argument', message: 'mobile_web_bundle_unavailable' },
+      _meta: { runtimeId: 'host-1' }
+    }
+    const client = { sendRequest: async () => reply }
+    const reads = [
+      () => runRpcOperation(client, mobileWebBundleManifestRead, null),
+      () =>
+        runRpcOperation(client, mobileWebBundleChunkRead, {
+          buildId: BUILD_ID,
+          path: 'index.html',
+          offset: 0
+        })
+    ]
+    for (const read of reads) {
+      await expect(read()).rejects.toThrow('invalid_argument: mobile_web_bundle_unavailable')
+    }
+    reply = { id: 'read-1', ok: true, result: null, _meta: { runtimeId: 'host-1' } }
+    for (const read of reads) {
+      await expect(read()).rejects.toBeInstanceOf(RpcIncompatibleReplyError)
    }
  })
 })
@@ -399,7 +421,7 @@ describe('which side a bundle read failed on', () => {
  })

  it.each([
-    ['a host refusal', `invalid_argument: ${MOBILE_WEB_BUNDLE_ERROR_CODES[0]}`],
+    ['a host refusal', 'invalid_argument: mobile_web_bundle_unavailable'],
    ['bytes that do not hash', 'bundle asset index.html hashed aa, not bb'],
    ['a build that changed mid-fetch', 'bundle build changed mid-fetch: asked aa, served bb'],
    ['an unread reply', 'The host sent a reply this app could not read (mobileWeb.bundle.manifest)']
@@ -86,8 +86,21 @@ function sentRequest(socket: MockWebSocket, method: string): SentRpcRequest {

 function sentRequests(socket: MockWebSocket, method: string): SentRpcRequest[] {
  return socket.sent
-    .map((payload) => JSON.parse(payload.replace(/^encrypted:/, '')) as SentRpcRequest)
-    .filter((request) => request.method === method)
+    .map((payload) => {
+      const request: unknown = JSON.parse(payload.replace(/^encrypted:/, ''))
+      if (
+        !request ||
+        typeof request !== 'object' ||
+        !('id' in request) ||
+        typeof request.id !== 'string' ||
+        !('method' in request) ||
+        typeof request.method !== 'string'
+      ) {
+        return null
+      }
+      return request
+    })
+    .filter((request): request is SentRpcRequest => request !== null && request.method === method)
 }

 function encodeBrowserFrame(): Uint8Array {
@@ -116,6 +129,12 @@ function encodeTerminalOutput(streamId: number, chunk: string): Uint8Array {
  })
 }

+function openAndAuthenticate(socket: MockWebSocket) {
+  socket.open()
+  socket.receive(JSON.stringify({ type: 'e2ee_ready' }))
+  socket.receive('encrypted:{"type":"e2ee_authenticated"}')
+}
+
 describe('mobile rpc-client connection timeout', () => {
  beforeEach(() => {
    vi.useFakeTimers()
@@ -174,9 +193,7 @@ describe('mobile rpc-client connection timeout', () => {
    const client = connect('ws://desktop.invalid', 'token', 'server-key')
    const socket = mockSockets[0]!

-    socket.open()
-    socket.receive(JSON.stringify({ type: 'e2ee_ready' }))
-    socket.receive('encrypted:{"type":"e2ee_authenticated"}')
+    openAndAuthenticate(socket)

    expect(client.getState()).toBe('connected')

@@ -192,9 +209,7 @@ describe('mobile rpc-client connection timeout', () => {
    const client = connect('ws://desktop.invalid', 'token', 'server-key')
    const socket = mockSockets[0]!

-    socket.open()
-    socket.receive(JSON.stringify({ type: 'e2ee_ready' }))
-    socket.receive('encrypted:{"type":"e2ee_authenticated"}')
+    openAndAuthenticate(socket)

    const unsubscribe = client.subscribe(
      'session.tabs.subscribe',
@@ -228,9 +243,7 @@ describe('mobile rpc-client connection timeout', () => {
    })
    const socket = mockSockets[0]!

-    socket.open()
-    socket.receive(JSON.stringify({ type: 'e2ee_ready' }))
-    socket.receive('encrypted:{"type":"e2ee_authenticated"}')
+    openAndAuthenticate(socket)

    expect(sentRequests(socket, 'notifications.subscribe')).toHaveLength(1)

@@ -242,9 +255,7 @@ describe('mobile rpc-client connection timeout', () => {
    const socket = mockSockets[0]!
    const frames: unknown[] = []

-    socket.open()
-    socket.receive(JSON.stringify({ type: 'e2ee_ready' }))
-    socket.receive('encrypted:{"type":"e2ee_authenticated"}')
+    openAndAuthenticate(socket)

    const unsubscribe = client.subscribe(
      'browser.screencast',
@@ -285,9 +296,7 @@ describe('mobile rpc-client connection timeout', () => {
    const client = connect('ws://desktop.invalid', 'token', 'server-key')
    const socket = mockSockets[0]!

-    socket.open()
-    socket.receive(JSON.stringify({ type: 'e2ee_ready' }))
-    socket.receive('encrypted:{"type":"e2ee_authenticated"}')
+    openAndAuthenticate(socket)

    const unsubscribe = client.subscribe(
      'browser.screencast',
@@ -324,9 +333,7 @@ describe('mobile rpc-client connection timeout', () => {
    const events: unknown[] = []
    const frames: unknown[] = []

-    socket.open()
-    socket.receive(JSON.stringify({ type: 'e2ee_ready' }))
-    socket.receive('encrypted:{"type":"e2ee_authenticated"}')
+    openAndAuthenticate(socket)

    client.subscribe(
      'browser.screencast',
@@ -372,9 +379,7 @@ describe('mobile rpc-client connection timeout', () => {
    )
    unsubscribe()

-    socket.open()
-    socket.receive(JSON.stringify({ type: 'e2ee_ready' }))
-    socket.receive('encrypted:{"type":"e2ee_authenticated"}')
+    openAndAuthenticate(socket)

    expect(sentRequests(socket, 'browser.screencast')).toHaveLength(0)

@@ -385,9 +390,7 @@ describe('mobile rpc-client connection timeout', () => {
    const client = connect('ws://desktop.invalid', 'token', 'server-key')
    const socket = mockSockets[0]!

-    socket.open()
-    socket.receive(JSON.stringify({ type: 'e2ee_ready' }))
-    socket.receive('encrypted:{"type":"e2ee_authenticated"}')
+    openAndAuthenticate(socket)

    client.subscribe('browser.screencast', { worktree: 'id:wt-1', page: 'page-1' }, () => {}, {
      onBinaryFrame: () => {}
@@ -425,9 +428,7 @@ describe('mobile rpc-client connection timeout', () => {
    const firstFrames: unknown[] = []
    const secondFrames: unknown[] = []

-    socket.open()
-    socket.receive(JSON.stringify({ type: 'e2ee_ready' }))
-    socket.receive('encrypted:{"type":"e2ee_authenticated"}')
+    openAndAuthenticate(socket)

    client.subscribe('browser.screencast', { worktree: 'id:wt-1', page: 'page-1' }, () => {}, {
      onBinaryFrame: (frame) => firstFrames.push(frame)
@@ -476,9 +477,7 @@ describe('mobile rpc-client connection timeout', () => {
    const socket = mockSockets[0]!
    const events: unknown[] = []

-    socket.open()
-    socket.receive(JSON.stringify({ type: 'e2ee_ready' }))
-    socket.receive('encrypted:{"type":"e2ee_authenticated"}')
+    openAndAuthenticate(socket)

    client.subscribe('terminal.subscribe', { terminal: 'term-1' }, (event) => events.push(event))
    const request = sentRequest(socket, 'terminal.subscribe')
@@ -504,9 +503,7 @@ describe('mobile rpc-client connection timeout', () => {
    const client = connect('ws://desktop.invalid', 'token', 'server-key')
    const firstSocket = mockSockets[0]!

-    firstSocket.open()
-    firstSocket.receive(JSON.stringify({ type: 'e2ee_ready' }))
-    firstSocket.receive('encrypted:{"type":"e2ee_authenticated"}')
+    openAndAuthenticate(firstSocket)

    client.subscribe(
      'terminal.subscribe',
@@ -525,9 +522,7 @@ describe('mobile rpc-client connection timeout', () => {
    firstSocket.close()
    vi.advanceTimersByTime(500)
    const secondSocket = mockSockets[1]!
-    secondSocket.open()
-    secondSocket.receive(JSON.stringify({ type: 'e2ee_ready' }))
-    secondSocket.receive('encrypted:{"type":"e2ee_authenticated"}')
+    openAndAuthenticate(secondSocket)

    expect(sentRequest(secondSocket, 'terminal.subscribe').params).toMatchObject({
      viewport: { cols: 60, rows: 24 }
@@ -540,9 +535,7 @@ describe('mobile rpc-client connection timeout', () => {
    const client = connect('ws://desktop.invalid', 'token', 'server-key')
    const socket = mockSockets[0]!

-    socket.open()
-    socket.receive(JSON.stringify({ type: 'e2ee_ready' }))
-    socket.receive('encrypted:{"type":"e2ee_authenticated"}')
+    openAndAuthenticate(socket)

    const request = client.sendRequest(
      'speech.dictation.finish',
@@ -551,15 +544,29 @@ describe('mobile rpc-client connection timeout', () => {
        timeoutMs: 123
      }
    )
+    let settled = false
+    const outcome = request.then(
+      (response) => {
+        settled = true
+        return response
+      },
+      (error: unknown) => {
+        settled = true
+        return error
+      }
+    )
    await Promise.resolve()
+    expect(sentRequest(socket, 'speech.dictation.finish').params).toEqual({ dictationId: 'd1' })

    await vi.advanceTimersByTimeAsync(122)
-    await expect(
-      Promise.race([request.then(() => 'settled'), Promise.resolve('pending')])
-    ).resolves.toBe('pending')
+    expect(settled).toBe(false)

    await vi.advanceTimersByTimeAsync(1)
-    await expect(request).rejects.toThrow('Request timed out: speech.dictation.finish')
+    const error = await outcome
+    if (!(error instanceof Error)) {
+      throw new Error('Expected the request timeout')
+    }
+    expect(error.message).toBe('Request timed out: speech.dictation.finish')

    client.close()
  })
@@ -568,9 +575,7 @@ describe('mobile rpc-client connection timeout', () => {
    const client = connect('ws://desktop.invalid', 'token', 'server-key')
    const socket = mockSockets[0]!

-    socket.open()
-    socket.receive(JSON.stringify({ type: 'e2ee_ready' }))
-    socket.receive('encrypted:{"type":"e2ee_authenticated"}')
+    openAndAuthenticate(socket)
    socket.close()

    const request = client.sendRequest(
@@ -606,11 +611,6 @@ describe('mobile rpc-client connection timeout', () => {

  // Foreground recovery must heal Android sessions left half-open after suspension (#5049).
  describe('foreground recovery', () => {
-    function openAndAuthenticate(socket: MockWebSocket) {
-      socket.open()
-      socket.receive(JSON.stringify({ type: 'e2ee_ready' }))
-      socket.receive('encrypted:{"type":"e2ee_authenticated"}')
-    }
    function connectAuthenticated() {
      const client = connect('ws://desktop.invalid', 'token', 'server-key')
      const socket = mockSockets[0]!
@@ -749,7 +749,7 @@ describe('mobile rpc-client connection timeout', () => {
      const probe = sentRequest(socket, 'status.get')
      socket.receive(`encrypted:${JSON.stringify({ id: probe.id, ok: true, result: {} })}`)

-      await vi.advanceTimersByTimeAsync(10_000)
+      await vi.advanceTimersByTimeAsync(24_000)
      expect(socket.close).not.toHaveBeenCalled()
      expect(client.getState()).toBe('connected')

@@ -811,7 +811,7 @@ describe('mobile rpc-client connection timeout', () => {
        error: { code: 'download_failed' }
      })

-      await vi.advanceTimersByTimeAsync(8_000)
+      await vi.advanceTimersByTimeAsync(24_000)
      expect(socket.close).not.toHaveBeenCalled()
      expect(client.getState()).toBe('connected')

@@ -839,7 +839,7 @@ describe('mobile rpc-client connection timeout', () => {
      socket.receive(encodeTerminalOutput(42, 'still alive'))
      await Promise.resolve()
      await Promise.resolve()
-      await vi.advanceTimersByTimeAsync(8_000)
+      await vi.advanceTimersByTimeAsync(24_000)

      expect(events).toContainEqual({ type: 'data', streamId: 42, chunk: 'still alive' })
      expect(socket.close).not.toHaveBeenCalled()
@@ -849,18 +849,23 @@ describe('mobile rpc-client connection timeout', () => {
    })

    it('counts authenticated unknown payloads before semantic decoding', async () => {
-      const { client, socket } = connectAuthenticated()
+      for (const payload of [
+        'encrypted:{"unexpected":true}',
+        'encrypted:{"id":"rpc-incomplete","ok":true}',
+        new Uint8Array([0xff, 0x00, 0x01])
+      ]) {
+        const client = connect('ws://desktop.invalid', 'token', 'server-key')
+        const socket = mockSockets.at(-1)!
+        openAndAuthenticate(socket)
+        client.notifyForeground()
+        socket.receive(payload)

-      client.notifyForeground()
-      socket.receive('encrypted:{"unexpected":true}')
-      socket.receive('encrypted:{"id":"rpc-incomplete","ok":true}')
-      socket.receive(new Uint8Array([0xff, 0x00, 0x01]))
+        await vi.advanceTimersByTimeAsync(24_000)
+        expect(socket.close).not.toHaveBeenCalled()
+        expect(client.getState()).toBe('connected')

-      await vi.advanceTimersByTimeAsync(8_000)
-      expect(socket.close).not.toHaveBeenCalled()
-      expect(client.getState()).toBe('connected')
-
-      client.close()
+        client.close()
+      }
    })

    it('does not count undecryptable payloads as probe activity', async () => {
@@ -888,11 +893,6 @@ describe('mobile rpc-client connection timeout', () => {

  // Transient auth rejection must not force re-pairing while the desktop still has valid credentials (#5200).
  describe('auth rejection retry (issue #5200)', () => {
-    function authenticate(socket: MockWebSocket) {
-      socket.open()
-      socket.receive(JSON.stringify({ type: 'e2ee_ready' }))
-      socket.receive('encrypted:{"type":"e2ee_authenticated"}')
-    }
    function unauthorizedResponsePayload(id: string): string {
      return `encrypted:${JSON.stringify({ id, ok: false, error: { code: 'unauthorized', message: 'Unauthorized' } })}`
    }
@@ -907,7 +907,7 @@ describe('mobile rpc-client connection timeout', () => {
      expect(client.getState()).toBe('reconnecting')

      await vi.advanceTimersByTimeAsync(500)
-      authenticate(mockSockets[mockSockets.length - 1]!)
+      openAndAuthenticate(mockSockets.at(-1)!)
      expect(client.getState()).toBe('connected')

      client.close()
@@ -946,7 +946,7 @@ describe('mobile rpc-client connection timeout', () => {
        socket.receive('encrypted:{"type":"e2ee_error","error":{"code":"unauthorized"}}')
      }
      await vi.advanceTimersByTimeAsync(1_000)
-      authenticate(mockSockets[mockSockets.length - 1]!)
+      openAndAuthenticate(mockSockets.at(-1)!)
      expect(client.getState()).toBe('connected')

      const live = mockSockets[mockSockets.length - 1]!
@@ -966,9 +966,7 @@ describe('mobile rpc-client connection timeout', () => {
    const client = connect('ws://desktop.invalid', 'token', 'server-key')
    const socket = mockSockets[0]!

-    socket.open()
-    socket.receive(JSON.stringify({ type: 'e2ee_ready' }))
-    socket.receive('encrypted:{"type":"e2ee_authenticated"}')
+    openAndAuthenticate(socket)
    socket.close()

    const waitingRequestError = client.sendRequest('status.get').then(
@@ -1,4 +1,4 @@
-import { describe, expect, it } from 'vitest'
+import { describe, expect, it, vi } from 'vitest'
 import type { RpcResponse } from './types'
 import { FakeSession } from './mobile-endpoint-supervisor-test-fakes'
 import {
@@ -165,25 +165,31 @@ describe('transport rejection stays on the promise channel', () => {
  })

  it('fails a Promise.all group immediately instead of waiting for a stalled peer', async () => {
-    const error = new Error('socket closed')
-    const failing = new FakeSession('connected')
-    failing.sendRequest.mockRejectedValue(error)
-    const stalled = new FakeSession('connected')
-    stalled.sendRequest.mockReturnValue(new Promise<RpcResponse>(() => {}))
+    vi.useFakeTimers()
+    try {
+      const error = new Error('socket closed')
+      const failing = new FakeSession('connected')
+      failing.sendRequest.mockRejectedValue(error)
+      const stalled = new FakeSession('connected')
+      stalled.sendRequest.mockReturnValue(new Promise<RpcResponse>(() => {}))

-    const group = Promise.all([
-      runRpcOperation(failing, workspaceListOrThrow, {}),
-      runRpcOperation(stalled, workspaceListOrThrow, {})
-    ])
-    const raced = await Promise.race([
-      group.then(
-        () => 'resolved' as const,
-        (caught: unknown) => caught
-      ),
-      new Promise((resolve) => setTimeout(() => resolve('still waiting'), 50))
-    ])
+      const group = Promise.all([
+        runRpcOperation(failing, workspaceListOrThrow, {}),
+        runRpcOperation(stalled, workspaceListOrThrow, {})
+      ])
+      const raced = Promise.race([
+        group.then(
+          () => 'resolved' as const,
+          (caught: unknown) => caught
+        ),
+        new Promise((resolve) => setTimeout(() => resolve('still waiting'), 50))
+      ])

-    expect(raced).toBe(error)
+      await vi.advanceTimersByTimeAsync(50)
+      expect(await raced).toBe(error)
+    } finally {
+      vi.useRealTimers()
+    }
  })

  it('only captures a rejection when a caller names the all-settled helper', async () => {
@@ -294,10 +300,12 @@ describe('an incompatible reply', () => {
      {}
    ).catch((thrown: unknown) => thrown)

-    expect(caught).toBeInstanceOf(RpcIncompatibleReplyError)
-    expect((caught as RpcIncompatibleReplyError).method).toBe('worktree.ps')
-    expect((caught as RpcIncompatibleReplyError).operationName).toBe('test.workspaceListOrThrow')
-    expect((caught as RpcIncompatibleReplyError).issues.length).toBeGreaterThan(0)
+    if (!(caught instanceof RpcIncompatibleReplyError)) {
+      throw new Error('expected an incompatible-reply error')
+    }
+    expect(caught.method).toBe('worktree.ps')
+    expect(caught.operationName).toBe('test.workspaceListOrThrow')
+    expect(caught.issues.length).toBeGreaterThan(0)
  })

  // Unreadable replies prove no mutation failure and must not match errors that authorize retry.
@@ -308,6 +316,7 @@ describe('an incompatible reply', () => {
      {}
    ).catch((thrown: unknown) => thrown)

+    expect(caught).toBeInstanceOf(RpcIncompatibleReplyError)
    expect(isRpcDeliveryUnknown(caught)).toBe(false)
    expect(isLogicalClientCutoverError(caught)).toBe(false)
  })
@@ -111,6 +111,9 @@ describe('stable logical RPC client', () => {
      undefined
    )
    expect(oldSession.close).toHaveBeenCalledOnce()
+    expect(nextSession.subscribe.mock.invocationCallOrder[0]).toBeLessThan(
+      oldSession.close.mock.invocationCallOrder[0]!
+    )
    expect(client.getActivePath()).toBe('relay')
    expect(client.getGeneration()).toBe(2)
    oldSession.emitStream('stale')
@@ -16,11 +16,11 @@ describe('withSelectedTabs', () => {

  it('keeps a selection the move did not touch', () => {
    const before = [
-      { id: 'g1', activeTabId: 'a', tabOrder: ['a', 'b'] },
+      { id: 'g1', activeTabId: 'a', tabOrder: ['d', 'a', 'b'] },
      { id: 'g2', activeTabId: 'c', tabOrder: ['c'] }
    ]
    const after = [
-      { id: 'g1', tabOrder: ['a'] },
+      { id: 'g1', tabOrder: ['d', 'a'] },
      { id: 'g2', tabOrder: ['b', 'c'] }
    ]
    expect(withSelectedTabs(before, after, 'b').map((group) => group.activeTabId)).toEqual([
@@ -17,6 +17,7 @@ const { OrcaRuntimeService } = await import('./orca-runtime-test-mocks.spec')
 await import('./orca-runtime-test-lifecycle.spec')
 const {
  HEADLESS_LEAF_ID,
+  makeDeferred,
  TEST_WORKTREE_ID,
  makeRuntimeStoreWithWorkspaceSession,
  makeWorkspaceSessionWithHeadlessTerminal
@@ -80,7 +81,22 @@ describe("a paired desktop's Reopen of a held pane (layout-owner build)", () =>
    const { runtime, spawn } = await heldPaneRuntime()
    expect(publishedStatus(runtime)).toBe('held')

-    await expect(runtime.reopenHeldTerminalPanes([PANE])).resolves.toEqual({ reopened: true })
+    const stopStarted = makeDeferred()
+    const stopFinished = makeDeferred()
+    daemonHold.reopen.mockImplementationOnce(async () => {
+      stopStarted.resolve()
+      await stopFinished.promise
+      return { reopened: true }
+    })
+    const reopening = runtime.reopenHeldTerminalPanes([PANE])
+    await stopStarted.promise
+    try {
+      expect(spawn).toHaveBeenCalledOnce()
+      expect(publishedStatus(runtime)).toBe('held')
+    } finally {
+      stopFinished.resolve()
+    }
+    await expect(reopening).resolves.toEqual({ reopened: true })

    expect(daemonHold.reopen).toHaveBeenCalledOnce()
    expect(spawn).toHaveBeenCalledTimes(2)