fix(mobile): probe timed-out RPCs before reconnecting

mobile/src/transport/mobile-endpoint-hysteresis-policy.ts

@@ -0,0 +1,16 @@
+import { MobileEndpointHysteresis } from './mobile-endpoint-hysteresis'
+
+export type { MobileEndpointHysteresis } from './mobile-endpoint-hysteresis'
+
+const DIRECT_OBSERVATION_MS = 30_000
+const MINIMUM_DWELL_MS = 60_000
+const FAILURE_COOLDOWN_MS = 60_000
+
+export function createMobileEndpointHysteresis(now: number): MobileEndpointHysteresis {
+  return new MobileEndpointHysteresis(now, {
+    directSuccessesRequired: 3,
+    directObservationMs: DIRECT_OBSERVATION_MS,
+    failureCooldownMs: FAILURE_COOLDOWN_MS,
+    minimumDwellMs: MINIMUM_DWELL_MS
+  })
+}

mobile/src/transport/mobile-endpoint-lifecycle.ts

@@ -31,12 +31,17 @@ export function startMobileEndpointLifecycle(
   let stopped = false
   let foreground = true
   let owner: EndpointOwner
+  const publishHostUpdate = (host: HostProfile): void => {
+    if (!stopped) {
+      onHostUpdated(host)
+    }
+  }
 
   const startSupervisor = async (host: HostProfile): Promise<void> => {
     if (stopped) {
       return
     }
-    const supervisor = createSupervisor(logical, host, onLog, onHostUpdated)
+    const supervisor = createSupervisor(logical, host, onLog, publishHostUpdate)
     owner.stop()
     owner = supervisor
     supervisor.setForeground(foreground)
@@ -44,7 +49,7 @@ export function startMobileEndpointLifecycle(
   }
 
   if (initialHost.relay) {
-    owner = createSupervisor(logical, initialHost, onLog, onHostUpdated)
+    owner = createSupervisor(logical, initialHost, onLog, publishHostUpdate)
     void owner.start()
   } else {
     owner = new MobileRelayDirectUpgradeController(logical, initialHost, {
@@ -55,7 +60,7 @@ export function startMobileEndpointLifecycle(
           dependencies: { randomBytes: ExpoCrypto.getRandomBytes }
         }),
       onUpgraded: async ({ host }) => {
-        onHostUpdated(host)
+        publishHostUpdate(host)
         await startSupervisor(host)
       }
     })

mobile/src/transport/mobile-endpoint-supervisor-retirement.test.ts

@@ -0,0 +1,222 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import type { MobileRelayCredentialBundle } from './mobile-relay-credential-bundle'
+import { RelayOuterError } from './mobile-relay-e2ee-link'
+import type { MobileRelayRpcSession } from './mobile-relay-rpc-session'
+import {
+  MobileEndpointSupervisor,
+  type MobileEndpointSupervisorDependencies
+} from './mobile-endpoint-supervisor'
+import type { RpcClient } from './rpc-client'
+import type { MobileConnectionPath, StableLogicalRpcClient } from './stable-logical-rpc-client'
+import type { ConnectionState, HostProfile, RpcResponse } from './types'
+
+vi.mock('react-native', () => ({ Platform: { OS: 'ios' } }))
+vi.mock('expo-secure-store', () => ({ WHEN_UNLOCKED_THIS_DEVICE_ONLY: 'when-unlocked' }))
+vi.mock('expo-crypto', () => ({ getRandomBytes: (length: number) => new Uint8Array(length) }))
+
+class FakeSession implements RpcClient {
+  readonly sendRequest = vi.fn(
+    async (_method: string, _params?: unknown): Promise<RpcResponse> => ({
+      id: 'rpc-1',
+      ok: true,
+      result: {},
+      _meta: { runtimeId: 'runtime-1' }
+    })
+  )
+  readonly subscribe = vi.fn(() => () => {})
+  readonly updateTerminalSubscriptionViewport = vi.fn()
+  readonly notifyForeground = vi.fn()
+  readonly close = vi.fn()
+  private readonly listeners = new Set<(state: ConnectionState) => void>()
+
+  constructor(private state: ConnectionState) {}
+
+  getState = () => this.state
+  getReconnectAttempt = () => 0
+  getLastConnectedAt = () => null
+  onStateChange = (listener: (state: ConnectionState) => void) => {
+    this.listeners.add(listener)
+    return () => this.listeners.delete(listener)
+  }
+
+  publishState(state: ConnectionState): void {
+    this.state = state
+    for (const listener of this.listeners) {
+      listener(state)
+    }
+  }
+}
+
+class FakeRelaySession extends FakeSession implements MobileRelayRpcSession {
+  constructor(
+    state: ConnectionState,
+    private readonly failure: Error | null = null
+  ) {
+    super(state)
+  }
+  getLeaseExpiresAt = () => Date.now() + 120_000
+  getResumeConfirmation = () => null
+  getFailure = () => this.failure
+}
+
+class FakeLogicalClient extends FakeSession implements StableLogicalRpcClient {
+  private generation = 1
+
+  constructor(
+    state: ConnectionState,
+    private path: MobileConnectionPath
+  ) {
+    super(state)
+  }
+
+  migrateTo = vi.fn(async (session: RpcClient, path: MobileConnectionPath) => {
+    if (session.getState() !== 'connected') {
+      session.close()
+      throw new Error(`replacement session ${session.getState()}`)
+    }
+    this.path = path
+    this.generation += 1
+    this.publishState('connected')
+  })
+  suspendActiveSession = vi.fn(() => this.publishState('disconnected'))
+  getActivePath = () => this.path
+  getGeneration = () => this.generation
+}
+
+const relay = {
+  v: 1 as const,
+  directorUrl: 'https://relay.onorca.dev',
+  cellUrl: 'https://relay-c1.onorca.dev',
+  assignmentEpoch: 7,
+  relayHostId: 'AbCdEf0123_-xyZ9',
+  e2eeFraming: 2 as const
+}
+const host: HostProfile = {
+  id: 'host-1',
+  name: 'Blue Whale',
+  endpoint: 'ws://192.168.1.10:6768',
+  deviceToken: 'device-token',
+  publicKeyB64: 'A'.repeat(44),
+  lastConnected: 1,
+  relayHostId: relay.relayHostId,
+  relay
+}
+const bundle: MobileRelayCredentialBundle = {
+  v: 1,
+  hostId: host.id,
+  deviceToken: host.deviceToken,
+  current: {
+    token: 'A'.repeat(43),
+    hash: 'B'.repeat(43),
+    version: 2,
+    expiresAt: Number.MAX_SAFE_INTEGER
+  }
+}
+
+function dependencies(
+  overrides: Partial<MobileEndpointSupervisorDependencies> = {}
+): MobileEndpointSupervisorDependencies {
+  return {
+    openDirect: vi.fn(() => new FakeSession('connected')),
+    openRelay: vi.fn(() => new FakeRelaySession('connected')),
+    resolveRelay: vi.fn(async ({ relay }) => relay),
+    readBundle: vi.fn(async () => bundle),
+    writeBundle: vi.fn(async () => {}),
+    saveHost: vi.fn(async () => {}),
+    now: Date.now,
+    randomBytes: (length) => new Uint8Array(length).fill(1),
+    setTimer: setTimeout,
+    clearTimer: clearTimeout,
+    ...overrides
+  }
+}
+
+function mockCredentialRotation(logical: FakeLogicalClient): void {
+  let installed: Record<string, unknown> | null = null
+  logical.sendRequest.mockImplementation(async (method, params) => {
+    const request = params as { installReqId?: string; reqId?: string }
+    if (method === 'pairing.provisionRelay') {
+      installed = {
+        v: 1,
+        reqId: request.reqId,
+        authorizationMode: 'authenticated-direct',
+        currentVersion: 3,
+        resumeExpiresAt: Date.now() + 300_000,
+        graceExpiresAt: Date.now() + 60_000
+      }
+      return { id: 'rpc-2', ok: true, result: installed, _meta: { runtimeId: 'runtime-1' } }
+    }
+    return {
+      id: 'rpc-1',
+      ok: true,
+      result: {
+        v: 1,
+        relay,
+        installStatus: installed
+          ? { v: 1, reqId: request.installReqId, state: 'committed', result: installed }
+          : { v: 1, reqId: request.installReqId, state: 'not-found' }
+      },
+      _meta: { runtimeId: 'runtime-1' }
+    }
+  })
+}
+
+describe('mobile endpoint supervisor retirement', () => {
+  beforeEach(() => {
+    vi.useFakeTimers()
+    vi.setSystemTime(new Date('2026-07-13T12:00:00Z'))
+  })
+
+  afterEach(() => vi.useRealTimers())
+
+  it('does not publish a resolved relay profile after stopping', async () => {
+    const logical = new FakeLogicalClient('disconnected', 'lan')
+    let finishResolve: ((value: typeof relay) => void) | undefined
+    const resolvePending = new Promise<typeof relay>((resolve) => {
+      finishResolve = resolve
+    })
+    const onHostUpdated = vi.fn()
+    const deps = dependencies({
+      openRelay: vi.fn(() => new FakeRelaySession('disconnected', new RelayOuterError(4409))),
+      resolveRelay: vi.fn(() => resolvePending),
+      onHostUpdated
+    })
+    const supervisor = new MobileEndpointSupervisor(logical, host, deps)
+
+    const starting = supervisor.start()
+    await vi.waitFor(() => expect(deps.resolveRelay).toHaveBeenCalledOnce())
+    supervisor.stop()
+    finishResolve?.(relay)
+    await starting
+
+    expect(deps.saveHost).not.toHaveBeenCalled()
+    expect(onHostUpdated).not.toHaveBeenCalled()
+  })
+
+  it('does not publish a rotated relay profile after stopping', async () => {
+    const logical = new FakeLogicalClient('connected', 'lan')
+    let finishCredentialWrite: (() => void) | undefined
+    const credentialWritePending = new Promise<void>((resolve) => {
+      finishCredentialWrite = resolve
+    })
+    const writeBundle = vi
+      .fn<(value: MobileRelayCredentialBundle) => Promise<void>>()
+      .mockResolvedValue()
+      .mockResolvedValueOnce()
+      .mockReturnValueOnce(credentialWritePending)
+    const onHostUpdated = vi.fn()
+    mockCredentialRotation(logical)
+    const deps = dependencies({ writeBundle, onHostUpdated })
+    const supervisor = new MobileEndpointSupervisor(logical, host, deps)
+
+    await supervisor.start()
+    logical.publishState('connected')
+    await vi.waitFor(() => expect(writeBundle).toHaveBeenCalledTimes(2))
+    supervisor.stop()
+    finishCredentialWrite?.()
+    await vi.advanceTimersByTimeAsync(0)
+
+    expect(deps.saveHost).not.toHaveBeenCalled()
+    expect(onHostUpdated).not.toHaveBeenCalled()
+  })
+})

mobile/src/transport/mobile-endpoint-supervisor.ts

@@ -2,7 +2,10 @@ import { openAuthenticatedDirectEndpoint } from './mobile-direct-endpoint-probe'
 import type { MobileEndpointSupervisorDependencies } from './mobile-endpoint-supervisor-contract'
 import { RelayReconnectController } from './mobile-relay-reconnect-controller'
 import { RelayLeaseRotationTimer } from './mobile-relay-lease-rotation-timer'
-import { MobileEndpointHysteresis } from './mobile-endpoint-hysteresis'
+import {
+  createMobileEndpointHysteresis,
+  type MobileEndpointHysteresis
+} from './mobile-endpoint-hysteresis-policy'
 import {
   encodeBase64Url,
   isDirectorResolutionFailure,
@@ -21,9 +24,6 @@ import type { HostProfile } from './types'
 export type { MobileEndpointSupervisorDependencies } from './mobile-endpoint-supervisor-contract'
 
 const DIRECT_PROBE_INTERVAL_MS = 15_000
-const DIRECT_OBSERVATION_MS = 30_000
-const MINIMUM_DWELL_MS = 60_000
-const FAILURE_COOLDOWN_MS = 60_000
 
 export class MobileEndpointSupervisor {
   private bundle: MobileRelayCredentialBundle | null = null
@@ -43,12 +43,7 @@ export class MobileEndpointSupervisor {
     private host: HostProfile,
     private readonly dependencies: MobileEndpointSupervisorDependencies
   ) {
-    this.hysteresis = new MobileEndpointHysteresis(dependencies.now(), {
-      directSuccessesRequired: 3,
-      directObservationMs: DIRECT_OBSERVATION_MS,
-      failureCooldownMs: FAILURE_COOLDOWN_MS,
-      minimumDwellMs: MINIMUM_DWELL_MS
-    })
+    this.hysteresis = createMobileEndpointHysteresis(dependencies.now())
     this.relayReconnect = new RelayReconnectController(dependencies, this.recoverRelay.bind(this))
     this.leaseRotation = new RelayLeaseRotationTimer(dependencies, () => {
       this.relayRotationPending = true
@@ -179,6 +174,9 @@ export class MobileEndpointSupervisor {
         relay: this.host.relay,
         resumeToken: credential.token
       })
+      if (this.stopped) {
+        return { ok: false, error: new Error('relay state missing') }
+      }
       this.host = await persistRelayHost(this.host, resolved, this.dependencies.saveHost)
       this.dependencies.onHostUpdated?.(this.host)
       return await this.openAndMigrateRelay(credential)
@@ -300,6 +298,9 @@ export class MobileEndpointSupervisor {
         writeBundle: this.dependencies.writeBundle,
         randomBytes: this.dependencies.randomBytes
       })
+      if (this.stopped) {
+        return
+      }
       this.bundle = result.bundle
       // Why: a scheduled rotation can finish after the old credential enters the rejection gate.
       credentialRefreshed = true

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

@@ -218,7 +218,7 @@ describe('mobile relay RPC session', () => {
     await expect(pending.catch((error: unknown) => isRpcDeliveryUnknown(error))).resolves.toBe(true)
   })
 
-  it('marks a relay RPC timeout delivery-unknown', async () => {
+  it('probes before demoting a timed-out relay RPC', async () => {
     const { session } = await authenticateSession()
     vi.useFakeTimers()
     try {
@@ -234,9 +234,14 @@ describe('mobile relay RPC session', () => {
         message: 'relay RPC timed out: terminal.send',
         unknown: true
       })
-      expect(session.getFailure()).toMatchObject({
-        message: 'relay RPC timed out: terminal.send'
-      })
+      expect(session.getState()).toBe('connected')
+      expect(fakes.close).not.toHaveBeenCalled()
+      expect(
+        fakes.sendText.mock.calls.map(([payload]) => JSON.parse(payload as string)).at(-1)
+      ).toMatchObject({ method: 'status.get' })
+
+      await vi.advanceTimersByTimeAsync(8_000)
+      expect(session.getFailure()).toMatchObject({ message: 'relay RPC timed out: status.get' })
       expect(session.getState()).toBe('disconnected')
       expect(fakes.close).toHaveBeenCalledOnce()
     } finally {
@@ -244,6 +249,32 @@ describe('mobile relay RPC session', () => {
     }
   })
 
+  it('keeps a relay session when its post-timeout probe answers', async () => {
+    const { session } = await authenticateSession()
+    vi.useFakeTimers()
+    try {
+      const request = session.sendRequest('browser.screenshot', {}, { timeoutMs: 100 })
+      const outcome = request.catch((error: unknown) => error)
+
+      await vi.advanceTimersByTimeAsync(100)
+      await expect(outcome).resolves.toMatchObject({
+        message: 'relay RPC timed out: browser.screenshot'
+      })
+      const probe = fakes.sendText.mock.calls
+        .map(([payload]) => JSON.parse(payload as string) as { id: string; method: string })
+        .find(({ method }) => method === 'status.get')!
+      fakes.linkOptions!.onText(
+        JSON.stringify({ id: probe.id, ok: true, result: {}, _meta: { runtimeId: 'runtime-1' } })
+      )
+
+      await vi.advanceTimersByTimeAsync(8_000)
+      expect(session.getState()).toBe('connected')
+      expect(fakes.close).not.toHaveBeenCalled()
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
   it('keeps a Relay session when a fresh timeout probe proves it live', async () => {
     const { session } = await authenticateSession()
     vi.useFakeTimers()

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

@@ -152,18 +152,13 @@ export function connectMobileRelayRpcSession(args: {
       return Promise.reject(new Error('relay session not connected'))
     }
     const id = nextId()
-    const requestControlResponseSequence = controlResponseSequence
     return new Promise((resolve, reject) => {
       const timer = setTimeout(() => {
         pending.delete(id)
         // Why: the frame was written long ago — the desktop may have processed it.
         const error = markRpcDeliveryUnknown(new Error(`relay RPC timed out: ${method}`))
         reject(error)
-        if (controlResponseSequence === requestControlResponseSequence) {
-          fail(error)
-        } else {
-          probeControlPlane()
-        }
+        probeControlPlane()
       }, timeoutMs)
       pending.set(id, { resolve, reject, timer })
       if (!sendFrame({ id, method, params })) {

mobile/src/transport/rpc-client-request-deadline.test.ts

@@ -181,6 +181,48 @@ describe('mobile rpc-client request deadline', () => {
     }
   })
 
+  it('probes before demoting a request that exceeds its deadline', async () => {
+    const client = connect('ws://desktop.invalid', 'token', 'server-key')
+    const socket = mockSockets[0]!
+    socket.open()
+    const request = client.sendRequest('speech.dictation.finish', {}, { timeoutMs: 123 })
+    const outcome = track(request)
+
+    await vi.advanceTimersByTimeAsync(123)
+    expect(outcome.read()).toBe('Request timed out: speech.dictation.finish')
+    const probe = sentRequest(socket, 'status.get')
+    expect(socket.close).not.toHaveBeenCalled()
+    expect(client.getState()).toBe('connected')
+
+    socket.receive(`encrypted:${JSON.stringify({ id: probe.id, ok: true, result: {} })}`)
+    await vi.advanceTimersByTimeAsync(8_000)
+    expect(socket.close).not.toHaveBeenCalled()
+    expect(client.getState()).toBe('connected')
+
+    client.close()
+    await request.catch(() => undefined)
+  })
+
+  it('demotes when the post-timeout control probe also stalls', async () => {
+    const client = connect('ws://desktop.invalid', 'token', 'server-key')
+    const socket = mockSockets[0]!
+    socket.open()
+    const request = client.sendRequest('browser.screenshot', {}, { timeoutMs: 123 })
+    const outcome = track(request)
+
+    await vi.advanceTimersByTimeAsync(123)
+    expect(outcome.read()).toBe('Request timed out: browser.screenshot')
+    expect(sentRequest(socket, 'status.get')).toBeDefined()
+    expect(client.getState()).toBe('connected')
+
+    await vi.advanceTimersByTimeAsync(8_000)
+    expect(socket.close).toHaveBeenCalled()
+    expect(client.getState()).toBe('reconnecting')
+
+    client.close()
+    await request.catch(() => undefined)
+  })
+
   it('does not reconnect when another control response proves the socket live', async () => {
     const client = connect('ws://desktop.invalid', 'token', 'server-key')
     const socket = mockSockets[0]!

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

@@ -535,37 +535,6 @@ describe('mobile rpc-client connection timeout', () => {
     client.close()
   })
 
-  it('honors per-request timeout overrides', async () => {
-    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"}')
-    socket.emitCloseOnClose = false
-
-    const request = client.sendRequest(
-      'speech.dictation.finish',
-      { dictationId: 'd1' },
-      {
-        timeoutMs: 123
-      }
-    )
-    await Promise.resolve()
-
-    await vi.advanceTimersByTimeAsync(122)
-    await expect(
-      Promise.race([request.then(() => 'settled'), Promise.resolve('pending')])
-    ).resolves.toBe('pending')
-
-    await vi.advanceTimersByTimeAsync(1)
-    await expect(request).rejects.toThrow('Request timed out: speech.dictation.finish')
-    expect(socket.close).toHaveBeenCalled()
-    expect(client.getState()).toBe('reconnecting')
-
-    client.close()
-  })
-
   it('applies per-request timeout overrides while waiting for reconnect', async () => {
     const client = connect('ws://desktop.invalid', 'token', 'server-key')
     const socket = mockSockets[0]!

mobile/src/transport/rpc-client.ts

@@ -768,8 +768,8 @@ export function connect(
   }
 
   // Why: stream frames can flow while control RPC is wedged; only a control response satisfies this probe.
-  function runActivityProbe() {
-    if (state !== 'connected' || !ws || activityProbeInFlight) {
+  function runActivityProbe(expectedWs: WebSocket | null = ws) {
+    if (state !== 'connected' || !ws || ws !== expectedWs || activityProbeInFlight) {
       return
     }
     activityProbeInFlight = true
@@ -1037,7 +1037,7 @@ export function connect(
           // Why: the frame was written 30s ago — the host may have processed it.
           reject(markRpcDeliveryUnknown(new Error(`Request timed out: ${method}`)))
           if (controlResponseSequence === requestControlResponseSequence) {
-            forceSocketReconnect(requestWs)
+            runActivityProbe(requestWs)
           }
         }, timeoutMs)