fix(mobile): fence reconnect liveness races

2026-10-22 08:03:37 +00:00 · 2026-08-01 03:07:14 -07:00
parent 693c43a4a6
commit 4e578fce25
17 changed files with 455 additions and 180 deletions
@@ -1,5 +1,6 @@
 import { useEffect, useState } from 'react'
-import { useRpcClientContext, type RpcClientContextValue } from './client-context'
+import { useRpcClientContext } from './client-context'
+import type { RpcClientContextValue } from './rpc-client-context-contract'

 export function useReconnectAttempt(hostId: string | undefined): number {
  return useHostMetric(hostId, (context, id) => context.getReconnectAttempt(id), 0)
@@ -365,6 +365,30 @@ describe('useHostClient', () => {
    harness.unmount()
  })

+  it('ignores relay metadata published by a superseded lifecycle', async () => {
+    const stale = makeFakeClient('connected')
+    const fresh = makeFakeClient('connecting')
+    fresh.sendRequest = vi.fn(async () => ({ id: 'health', ok: true, result: {} }))
+    connectMock.mockReturnValueOnce(stale).mockReturnValueOnce(fresh)
+    loadHostsMock.mockResolvedValue([HOST])
+
+    const harness = await renderHarness(HOST.id)
+    const updatedHost = { ...HOST, endpoint: 'ws://127.0.0.1:2' }
+    harness.primeHosts([updatedHost])
+    const publishHostUpdate = connectMock.mock.calls[0]?.[2] as
+      | ((host: HostProfile) => void)
+      | undefined
+    publishHostUpdate?.({
+      ...HOST,
+      endpoints: [{ id: 'direct-primary', kind: 'lan', url: HOST.endpoint }]
+    })
+
+    await act(async () => harness.forceReconnect(HOST.id))
+
+    expect(connectMock.mock.calls[1]?.[0]).toEqual(updatedHost)
+    harness.unmount()
+  })
+
  it('supersedes a pending old-profile open when Force Reconnect uses a saved endpoint', async () => {
    let resolveOldLookup: ((hosts: HostProfile[]) => void) | null = null
    const oldLookup = new Promise<HostProfile[]>((resolve) => {
@@ -423,17 +447,43 @@ describe('useHostClient', () => {
    harness.unmount()
  })

-  it('queues a changed endpoint behind an older reconnect health probe', async () => {
+  it('opens a changed endpoint without waiting for a superseded host lookup', async () => {
+    const stale = makeFakeClient('connected')
+    const fresh = makeFakeClient('connecting')
+    fresh.sendRequest = vi.fn(async () => ({ id: 'health', ok: true, result: {} }))
+    loadHostsMock.mockResolvedValueOnce([HOST]).mockReturnValueOnce(new Promise<never>(() => {}))
+    connectMock.mockReturnValueOnce(stale).mockReturnValueOnce(fresh)
+
+    const harness = await renderHarness(HOST.id)
+    void harness.forceReconnect(HOST.id)
+    await act(async () => {
+      await Promise.resolve()
+      await Promise.resolve()
+    })
+
+    const updatedHost = { ...HOST, endpoint: 'ws://127.0.0.1:2' }
+    harness.primeHosts([updatedHost])
+    await act(async () => harness.forceReconnect(HOST.id))
+
+    expect(connectMock).toHaveBeenCalledTimes(2)
+    expect(connectMock.mock.calls[1]?.[0]).toEqual(updatedHost)
+    harness.unmount()
+  })
+
+  it('supersedes an older reconnect health probe immediately', async () => {
    const stale = makeFakeClient('connected')
    const oldReplacement = makeFakeClient('connecting')
    const newReplacement = makeFakeClient('connecting')
-    let resolveOldHealthCheck: (() => void) | null = null
+    let rejectOldHealthCheck: ((error: Error) => void) | null = null
    oldReplacement.sendRequest = vi.fn(
      () =>
-        new Promise((resolve) => {
-          resolveOldHealthCheck = () => resolve({ id: 'old-health', ok: true, result: {} })
+        new Promise((_resolve, reject) => {
+          rejectOldHealthCheck = reject
        })
    )
+    oldReplacement.closeMock.mockImplementation(() => {
+      rejectOldHealthCheck?.(new Error('Client closed'))
+    })
    newReplacement.sendRequest = vi.fn(async () => ({ id: 'new-health', ok: true, result: {} }))
    connectMock
      .mockReturnValueOnce(stale)
@@ -452,14 +502,12 @@ describe('useHostClient', () => {
    harness.primeHosts([updatedHost])
    const second = harness.forceReconnect(HOST.id)
    expect(second).not.toBe(first)
-    expect(connectMock).toHaveBeenCalledTimes(2)
+    expect(connectMock).toHaveBeenCalledTimes(3)

-    resolveOldHealthCheck?.()
    await act(async () => {
      await Promise.all([first, second])
    })

-    expect(connectMock).toHaveBeenCalledTimes(3)
    expect(connectMock.mock.calls[2]?.[0]).toEqual(updatedHost)
    expect(oldReplacement.closeMock).toHaveBeenCalledOnce()
    expect(newReplacement.sendRequest).toHaveBeenCalledWith('status.get', undefined, {
@@ -471,9 +519,11 @@ describe('useHostClient', () => {
    harness.unmount()
  })

-  it('keeps a host disconnected when it closes with a changed-profile reconnect queued', async () => {
+  it('keeps a host disconnected when it closes during a changed-profile reconnect', async () => {
    const stale = makeFakeClient('connected')
    const oldReplacement = makeFakeClient('connecting')
+    const newReplacement = makeFakeClient('connecting')
+    newReplacement.sendRequest = vi.fn(async () => ({ id: 'new-health', ok: true, result: {} }))
    let rejectOldHealthCheck: ((error: Error) => void) | null = null
    oldReplacement.sendRequest = vi.fn(
      () =>
@@ -484,7 +534,10 @@ describe('useHostClient', () => {
    oldReplacement.closeMock.mockImplementation(() => {
      rejectOldHealthCheck?.(new Error('Client closed'))
    })
-    connectMock.mockReturnValueOnce(stale).mockReturnValueOnce(oldReplacement)
+    connectMock
+      .mockReturnValueOnce(stale)
+      .mockReturnValueOnce(oldReplacement)
+      .mockReturnValueOnce(newReplacement)
    loadHostsMock.mockResolvedValue([HOST])

    const harness = await renderHarness(HOST.id)
@@ -500,8 +553,9 @@ describe('useHostClient', () => {

    await act(async () => Promise.all([first, queued]))

-    expect(connectMock).toHaveBeenCalledTimes(2)
+    expect(connectMock).toHaveBeenCalledTimes(3)
    expect(oldReplacement.closeMock).toHaveBeenCalledOnce()
+    expect(newReplacement.closeMock).toHaveBeenCalledOnce()
    expect(harness.hook.client).toBeNull()
    expect(harness.hook.state).toBe('disconnected')

@@ -13,12 +13,13 @@ import {
 import type { RpcClient } from './rpc-client'
 import { connectionLogStore } from './connection-log-buffer'
 import { HostForceReconnectCoordinator, type HostReconnectEntry } from './host-force-reconnect'
-import { HostReconnectProfileCache } from './host-reconnect-profile-cache'
+import { HostReconnectProfileCache, type HostOpenProfile } from './host-reconnect-profile-cache'
 import { subscribeConnectionRevivalTriggers } from './connection-revival-triggers'
 import { HostClientOpenRegistry } from './host-client-open-registry'
 import { loadHosts } from './host-store'
 import { openHostLogicalClient } from './host-logical-client'
 import { clientActivePath } from './rpc-client-active-path'
+import type { RpcClientContextValue } from './rpc-client-context-contract'
 import type { MobileConnectionPath } from './stable-logical-rpc-client'
 import type { ConnectionLogSink, ConnectionState, HostProfile } from './types'

@@ -26,23 +27,6 @@ type StoreEntry = HostReconnectEntry & {
  state: ConnectionState
 }

-export type RpcClientContextValue = {
-  acquire: (hostId: string, host?: HostProfile) => RpcClient | null
-  release: (hostId: string) => void
-  forceReconnect: (hostId: string) => Promise<void>
-  closeHost: (hostId: string) => void
-  getState: (hostId: string) => ConnectionState
-  getReconnectAttempt: (hostId: string) => number
-  // Why: ms-epoch of the last 'connected' (null if never this session); UI escalates "Reconnecting…" into a re-pair prompt.
-  getLastConnectedAt: (hostId: string) => number | null
-  getActivePath: (hostId: string) => MobileConnectionPath
-  subscribeHostState: (hostId: string, listener: (state: ConnectionState) => void) => () => void
-  getAllClients: () => Array<{ hostId: string; client: RpcClient }>
-  subscribeAllHosts: (listener: () => void) => () => void
-  // Why: lets the home screen feed already-loaded HostProfiles so we don't pay loadHosts() latency twice.
-  primeHosts: (hosts: HostProfile[]) => void
-}
-
 const Ctx = createContext<RpcClientContextValue | null>(null)

 export function RpcClientProvider({ children }: { children: ReactNode }) {
@@ -86,85 +70,90 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
    notifyAllHosts()
  }, [])

-  const openEntry = useCallback(async (hostId: string): Promise<StoreEntry | null> => {
-    const profileVersion = primedHostsRef.current.version(hostId)
-    const existing = pendingOpensRef.current.getActivePromise(hostId, profileVersion)
-    if (existing) {
-      await existing
-      return storeRef.current.get(hostId) ?? null
-    }
-    let resolve: () => void = () => {}
-    const promise = new Promise<void>((res) => {
-      resolve = res
-    })
-    const pendingOpen = pendingOpensRef.current.register(hostId, profileVersion, promise)
-
-    try {
-      // Why: prefer the primed cache so we don't serialize a second Keychain pass on cold start.
-      let host = primedHostsRef.current.get(hostId)
-      if (!host) {
-        try {
-          const hosts = await loadHosts()
-          host = hosts.find((h) => h.id === hostId)
-        } catch {
-          // Why: cold-start Keychain failure (iOS mid-unlock / Android Keystore race); surface 'disconnected' so the user can Reconnect.
-          notifyHostState(hostId, 'disconnected')
-          notifyAllHosts()
-          return null
-        }
-        if (!host) {
-          // Why: silent return leaves screens on a permanent spinner (STA-1511); surface 'disconnected' so they show a retry affordance.
-          notifyHostState(hostId, 'disconnected')
-          notifyAllHosts()
-          return null
-        }
+  const openEntry = useCallback(
+    async (hostId: string, requestedProfile?: HostOpenProfile): Promise<StoreEntry | null> => {
+      const profileVersion = requestedProfile?.version ?? primedHostsRef.current.version(hostId)
+      const existing = pendingOpensRef.current.getActivePromise(hostId, profileVersion)
+      if (existing) {
+        await existing
+        return storeRef.current.get(hostId) ?? null
      }
-
-      if (pendingOpen.cancelled) {
-        return null
-      }
-
-      // Re-check after any await — another acquire() may have completed.
-      const after = storeRef.current.get(hostId)
-      if (after) {
-        return after
-      }
-
-      let client: RpcClient
-      const onLog: ConnectionLogSink = (entry) => connectionLogStore.append(hostId, entry)
-      try {
-        client = openHostLogicalClient(host, onLog, (updatedHost) =>
-          primedHostsRef.current.prime(updatedHost)
-        )
-      } catch {
-        // Why: openHostLogicalClient can throw synchronously (bad public key / invalid URL); notify so the UI leaves 'connecting'.
-        notifyHostState(hostId, 'disconnected')
-        notifyAllHosts()
-        return null
-      }
-      const unsubState = client.onStateChange((state) => {
-        const cur = storeRef.current.get(hostId)
-        if (!cur) {
-          return
-        }
-        cur.state = state
-        notifyHostState(hostId, state)
+      let resolve: () => void = () => {}
+      const promise = new Promise<void>((res) => {
+        resolve = res
      })
-      const entry: StoreEntry = {
-        client,
-        state: client.getState(),
-        refCount: 0,
-        unsubState
+      const pendingOpen = pendingOpensRef.current.register(hostId, profileVersion, promise)
+
+      try {
+        // Why: prefer the primed cache so we don't serialize a second Keychain pass on cold start.
+        let host = requestedProfile?.host ?? primedHostsRef.current.get(hostId)
+        if (!host) {
+          try {
+            const hosts = await loadHosts()
+            host = hosts.find((h) => h.id === hostId)
+          } catch {
+            // Why: cold-start Keychain failure (iOS mid-unlock / Android Keystore race); surface 'disconnected' so the user can Reconnect.
+            notifyHostState(hostId, 'disconnected')
+            notifyAllHosts()
+            return null
+          }
+          if (!host) {
+            // Why: silent return leaves screens on a permanent spinner (STA-1511); surface 'disconnected' so they show a retry affordance.
+            notifyHostState(hostId, 'disconnected')
+            notifyAllHosts()
+            return null
+          }
+        }
+
+        if (pendingOpen.cancelled) {
+          return null
+        }
+
+        // Re-check after any await — another acquire() may have completed.
+        const after = storeRef.current.get(hostId)
+        if (after) {
+          return after
+        }
+
+        let client: RpcClient
+        const onLog: ConnectionLogSink = (entry) => connectionLogStore.append(hostId, entry)
+        try {
+          client = openHostLogicalClient(
+            host,
+            onLog,
+            primedHostsRef.current.publisher(hostId, profileVersion)
+          )
+        } catch {
+          // Why: openHostLogicalClient can throw synchronously (bad public key / invalid URL); notify so the UI leaves 'connecting'.
+          notifyHostState(hostId, 'disconnected')
+          notifyAllHosts()
+          return null
+        }
+        const unsubState = client.onStateChange((state) => {
+          const cur = storeRef.current.get(hostId)
+          if (!cur) {
+            return
+          }
+          cur.state = state
+          notifyHostState(hostId, state)
+        })
+        const entry: StoreEntry = {
+          client,
+          state: client.getState(),
+          refCount: 0,
+          unsubState
+        }
+        storeRef.current.set(hostId, entry)
+        notifyHostState(hostId, entry.state)
+        notifyAllHosts()
+        return entry
+      } finally {
+        pendingOpensRef.current.deleteIfCurrent(hostId, pendingOpen)
+        resolve()
      }
-      storeRef.current.set(hostId, entry)
-      notifyHostState(hostId, entry.state)
-      notifyAllHosts()
-      return entry
-    } finally {
-      pendingOpensRef.current.deleteIfCurrent(hostId, pendingOpen)
-      resolve()
-    }
-  }, [])
+    },
+    []
+  )

  // Synchronous get-or-open: returns an existing client immediately, else kicks off an async open and returns null this tick.
  const acquire = useCallback(
@@ -205,10 +194,14 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
  }, [])

  const forceReconnect = useCallback(
-    (hostId: string): Promise<void> =>
-      forceReconnectCoordinatorRef.current.run({
+    (hostId: string): Promise<void> => {
+      const profile = {
+        host: primedHostsRef.current.get(hostId),
+        version: primedHostsRef.current.version(hostId)
+      }
+      return forceReconnectCoordinatorRef.current.run({
        hostId,
-        profileVersion: primedHostsRef.current.version(hostId),
+        profileVersion: profile.version,
        getEntry: () => storeRef.current.get(hostId),
        getListenerCount: () => stateListenersRef.current.get(hostId)?.size ?? 0,
        removeEntry: () => {
@@ -216,8 +209,10 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
          notifyHostState(hostId, 'disconnected')
          notifyAllHosts()
        },
-        openReplacement: () => openEntry(hostId)
-      }),
+        cancelPendingOpen: () => pendingOpensRef.current.cancel(hostId),
+        openReplacement: () => openEntry(hostId, profile)
+      })
+    },
    [openEntry]
  )

@@ -0,0 +1,28 @@
+import { HostProfileSchema, type HostProfile } from './types'
+import { mutateStoredHosts } from './host-store'
+import * as hostListLoads from './host-list-load-sharing'
+import { saveMobileRelayHostOverlay } from './mobile-relay-host-overlay-store'
+
+export class MobileRelayUpgradeHostRemovedError extends Error {}
+
+export async function saveExistingHostRelayRouting(host: HostProfile): Promise<void> {
+  const validated = HostProfileSchema.parse(host)
+  await mutateStoredHosts((hosts) => {
+    const existing = hosts.find(({ id }) => id === validated.id)
+    if (!existing || existing.publicKeyB64 !== validated.publicKeyB64) {
+      throw new MobileRelayUpgradeHostRemovedError('mobile relay upgrade host was removed')
+    }
+    return hosts
+  })
+  if (!validated.endpoints || !validated.relayHostId || !validated.relay) {
+    throw new Error('mobile relay upgrade routing metadata missing')
+  }
+  await saveMobileRelayHostOverlay({
+    v: 2,
+    hostId: validated.id,
+    endpoints: validated.endpoints,
+    relayHostId: validated.relayHostId,
+    relay: validated.relay
+  })
+  hostListLoads.dropSharedHostListLoad()
+}
@@ -13,6 +13,7 @@ type HostReconnectOperation = {
  getEntry: () => HostReconnectEntry | undefined
  getListenerCount: () => number
  removeEntry: () => void
+  cancelPendingOpen: () => void
  openReplacement: () => Promise<HostReconnectEntry | null>
 }

@@ -38,17 +39,17 @@ export class HostForceReconnectCoordinator {
  }

  run(operation: HostReconnectOperation): Promise<void> {
-    const generation = this.generation(operation.hostId)
    const pending = this.pendingByHost.get(operation.hostId)
+    let generation = this.generation(operation.hostId)
    if (pending?.profileVersion === operation.profileVersion && pending.generation === generation) {
      return pending.promise
    }
-    // Why: a changed endpoint must supersede the in-flight profile without racing its health probe.
-    const reconnect = pending
-      ? pending.promise
-          .catch(() => undefined)
-          .then(() => this.replaceAndVerify(operation, generation))
-      : this.replaceAndVerify(operation, generation)
+    if (pending) {
+      generation += 1
+      this.generations.set(operation.hostId, generation)
+      operation.cancelPendingOpen()
+    }
+    const reconnect = this.replaceAndVerify(operation, generation)
    this.pendingByHost.set(operation.hostId, {
      profileVersion: operation.profileVersion,
      generation,
@@ -5,11 +5,16 @@ type CachedHostProfile = {
  version: number
 }

+export type HostOpenProfile = {
+  host: HostProfile | undefined
+  version: number
+}
+
 export class HostReconnectProfileCache {
  private readonly profiles = new Map<string, CachedHostProfile>()
  private readonly latestVersions = new Map<string, number>()

-  prime(host: HostProfile): void {
+  prime(host: HostProfile): number {
    const current = this.profiles.get(host.id)
    const version =
      current && reconnectProfileMatches(current.host, host)
@@ -17,6 +22,27 @@ export class HostReconnectProfileCache {
        : (this.latestVersions.get(host.id) ?? 0) + 1
    this.latestVersions.set(host.id, version)
    this.profiles.set(host.id, { host, version })
+    return version
+  }
+
+  primeFromVersion(host: HostProfile, sourceVersion: number): number | null {
+    if (this.version(host.id) !== sourceVersion) {
+      return null
+    }
+    return this.prime(host)
+  }
+
+  publisher(hostId: string, initialVersion: number): (host: HostProfile) => void {
+    let sourceVersion = initialVersion
+    return (host) => {
+      if (host.id !== hostId) {
+        return
+      }
+      const nextVersion = this.primeFromVersion(host, sourceVersion)
+      if (nextVersion !== null) {
+        sourceVersion = nextVersion
+      }
+    }
  }

  get(hostId: string): HostProfile | undefined {
@@ -36,15 +36,17 @@ vi.mock('./host-credential-cleanup', () => ({

 import {
  loadHosts,
-  MobileRelayUpgradeHostRemovedError,
  removeHost,
  resolvePairingHostIdentity,
  resetHostStoreForTests,
  saveHost,
-  saveExistingHostRelayUpgrade,
  updateHostNameAndEndpoint,
  updateLastConnected
 } from './host-store'
+import {
+  MobileRelayUpgradeHostRemovedError,
+  saveExistingHostRelayRouting
+} from './existing-host-relay-routing'
 import { resetMobileRelayHostOverlayStoreForTests } from './mobile-relay-host-overlay-store'

 const HOSTS_STORAGE_KEY = 'orca:hosts'
@@ -247,13 +249,43 @@ describe('host-store list mutations', () => {
    storedHostsRaw = JSON.stringify([HOST_TWO])

    await expect(
-      saveExistingHostRelayUpgrade({ ...HOST_ONE, deviceToken: 'token-1' })
+      saveExistingHostRelayRouting({ ...HOST_ONE, deviceToken: 'token-1' })
    ).rejects.toBeInstanceOf(MobileRelayUpgradeHostRemovedError)

    expect(JSON.parse(storedHostsRaw)).toEqual([HOST_TWO])
    expect(secureStoreMock.setItemAsync).not.toHaveBeenCalled()
  })

+  it('publishes relay routing without overwriting a newer endpoint edit', async () => {
+    const relayHostId = 'AbCdEf0123_-xyZ9'
+    await updateHostNameAndEndpoint(HOST_ONE.id, { endpoint: 'ws://127.0.0.1:9' })
+
+    await saveExistingHostRelayRouting({
+      ...HOST_ONE,
+      deviceToken: 'token-1',
+      endpoints: [
+        { id: 'direct-primary', kind: 'lan', url: HOST_ONE.endpoint },
+        { id: 'relay-primary', kind: 'relay', url: 'wss://relay.onorca.dev/v1/connect/host' }
+      ],
+      relayHostId,
+      relay: {
+        v: 1,
+        directorUrl: 'https://relay.onorca.dev',
+        cellUrl: 'https://relay.onorca.dev',
+        assignmentEpoch: 1,
+        relayHostId,
+        e2eeFraming: 2
+      }
+    })
+
+    expect(JSON.parse(storedHostsRaw)).toContainEqual(
+      expect.objectContaining({ id: HOST_ONE.id, endpoint: 'ws://127.0.0.1:9' })
+    )
+    expect(JSON.parse(storedOverlayRaw ?? '[]')).toContainEqual(
+      expect.objectContaining({ hostId: HOST_ONE.id, relayHostId })
+    )
+  })
+
  it('awaits cleanup scheduling after metadata commit', async () => {
    let resolveSchedule: (() => void) | null = null
    scheduleCleanupMock.mockReturnValue(
@@ -186,14 +186,16 @@ async function readStoredHostsForMutation(): Promise<StoredHostProfile[]> {
  }
 }

-async function mutateStoredHosts(
+export async function mutateStoredHosts(
  update: (hosts: StoredHostProfile[]) => StoredHostProfile[]
 ): Promise<void> {
  const mutation = hostListMutation.then(async () => {
    const current = await readStoredHostsForMutation()
    const next = update(current)
-    await AsyncStorage.setItem(STORAGE_KEY, JSON.stringify(next))
-    hostListLoads.dropSharedHostListLoad()
+    if (next !== current) {
+      await AsyncStorage.setItem(STORAGE_KEY, JSON.stringify(next))
+      hostListLoads.dropSharedHostListLoad()
+    }
  })
  hostListMutation = mutation.catch(() => {})
  return mutation
@@ -209,17 +211,11 @@ function toStored(host: HostProfile): StoredHostProfile {
  }
 }

-export class MobileRelayUpgradeHostRemovedError extends Error {}
-
 export async function saveHost(host: HostProfile): Promise<void> {
-  await persistHost(host, false)
+  await persistHost(host)
 }

-export async function saveExistingHostRelayUpgrade(host: HostProfile): Promise<void> {
-  await persistHost(host, true)
-}
-
-async function persistHost(host: HostProfile, requireExisting: boolean): Promise<void> {
+async function persistHost(host: HostProfile): Promise<void> {
  const validated = HostProfileSchema.parse(host)
  const stored = toStored(validated)
  const duplicateHostIds = new Set<string>()
@@ -238,10 +234,6 @@ async function persistHost(host: HostProfile, requireExisting: boolean): Promise
        .filter(({ id }) => !duplicateHostIds.has(id))
        .map((candidate) => (candidate.id === stored.id ? stored : candidate))
    }
-    if (requireExisting) {
-      // Why: an in-flight relay upgrade must not resurrect a host the user removed.
-      throw new MobileRelayUpgradeHostRemovedError('mobile relay upgrade host was removed')
-    }
    return [...hosts.filter(({ id }) => !duplicateHostIds.has(id)), stored]
  })
  // Why: write metadata before the keychain token so a crash leaves recoverable orphaned metadata, not an orphaned token that persists forever.
@@ -8,7 +8,7 @@ import {
  readMobileRelayCredentialBundle,
  writeMobileRelayCredentialBundle
 } from './mobile-relay-credential-bundle'
-import { saveHost } from './host-store'
+import { saveExistingHostRelayRouting } from './existing-host-relay-routing'
 import { upgradeDirectMobileRelay } from './mobile-relay-direct-upgrade'
 import { MobileRelayDirectUpgradeController } from './mobile-relay-direct-upgrade-controller'
 import type { StableLogicalRpcClient } from './stable-logical-rpc-client'
@@ -99,7 +99,7 @@ function createSupervisor(
    resolveRelay: resolveMobileRelayEndpoint,
    readBundle: readMobileRelayCredentialBundle,
    writeBundle: writeMobileRelayCredentialBundle,
-    saveHost,
+    saveHost: saveExistingHostRelayRouting,
    onHostUpdated,
    now: Date.now,
    randomBytes: ExpoCrypto.getRandomBytes,
@@ -1,6 +1,6 @@
 import { describe, expect, it, vi } from 'vitest'
 import type { MobileRelayEndpoint } from '../../../src/shared/mobile-relay-credential-contract'
-import { MobileRelayUpgradeHostRemovedError } from './host-store'
+import { MobileRelayUpgradeHostRemovedError } from './existing-host-relay-routing'
 import {
  createMobileRelayDirectUpgradeJournal,
  type MobileRelayDirectUpgradeJournal
@@ -5,7 +5,10 @@ import {
  type DeviceCredentialInstalled,
  type PairingGetEndpointsResult
 } from '../../../src/shared/mobile-relay-credential-contract'
-import { MobileRelayUpgradeHostRemovedError, saveExistingHostRelayUpgrade } from './host-store'
+import {
+  MobileRelayUpgradeHostRemovedError,
+  saveExistingHostRelayRouting
+} from './existing-host-relay-routing'
 import { persistRelayHost } from './mobile-endpoint-supervisor-support'
 import {
  MobileRelayCredentialBundleSchema,
@@ -33,7 +36,7 @@ type Dependencies = {
  writeJournal: typeof writeMobileRelayDirectUpgradeJournal
  clearJournal: typeof deleteMobileRelayDirectUpgradeJournal
  writeBundle: typeof writeMobileRelayCredentialBundle
-  saveHost: typeof saveExistingHostRelayUpgrade
+  saveHost: typeof saveExistingHostRelayRouting
  deleteBundle: typeof deleteMobileRelayCredentialBundle
  randomBytes: (length: number) => Uint8Array
 }
@@ -51,7 +54,7 @@ export async function upgradeDirectMobileRelay(args: {
    writeJournal: writeMobileRelayDirectUpgradeJournal,
    clearJournal: deleteMobileRelayDirectUpgradeJournal,
    writeBundle: writeMobileRelayCredentialBundle,
-    saveHost: saveExistingHostRelayUpgrade,
+    saveHost: saveExistingHostRelayRouting,
    deleteBundle: deleteMobileRelayCredentialBundle,
    randomBytes: ExpoCrypto.getRandomBytes,
    ...args.dependencies
@@ -275,6 +275,66 @@ describe('mobile relay RPC session', () => {
    }
  })

+  it('counts a late timed-out reply as relay control-plane liveness', 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(0)
+      const timedOutRequest = fakes.sendText.mock.calls
+        .map(([payload]) => JSON.parse(payload as string) as { id: string; method: string })
+        .find(({ method }) => method === 'browser.screenshot')!
+
+      await vi.advanceTimersByTimeAsync(100)
+      await expect(outcome).resolves.toMatchObject({
+        message: 'relay RPC timed out: browser.screenshot'
+      })
+      fakes.linkOptions!.onText(
+        JSON.stringify({ id: timedOutRequest.id, ok: true, result: {}, _meta: {} })
+      )
+
+      await vi.advanceTimersByTimeAsync(8_000)
+      expect(session.getState()).toBe('connected')
+      expect(fakes.close).not.toHaveBeenCalled()
+    } finally {
+      session.close()
+      vi.useRealTimers()
+    }
+  })
+
+  it('periodically demotes a silent half-open Relay session', async () => {
+    vi.useFakeTimers()
+    try {
+      const { session } = await authenticateSession()
+
+      await vi.advanceTimersByTimeAsync(20_000)
+      expect(
+        fakes.sendText.mock.calls.map(([payload]) => JSON.parse(payload as string)).at(-1)
+      ).toMatchObject({ method: 'status.get' })
+      expect(session.getState()).toBe('connected')
+
+      await vi.advanceTimersByTimeAsync(8_000)
+      expect(session.getState()).toBe('disconnected')
+      expect(fakes.close).toHaveBeenCalledOnce()
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('stops periodic Relay probing when the session closes', async () => {
+    vi.useFakeTimers()
+    try {
+      const { session } = await authenticateSession()
+      session.close()
+
+      await vi.advanceTimersByTimeAsync(20_000)
+      expect(fakes.sendText).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()
@@ -13,6 +13,7 @@ import type { RpcClient } from './rpc-client'
 import type { ConnectionState, RpcResponse } from './types'

 const CONTROL_PROBE_TIMEOUT_MS = 8_000
+const CONTROL_PROBE_INTERVAL_MS = 20_000

 type PendingRequest = {
  resolve: (response: RpcResponse) => void
@@ -38,11 +39,13 @@ export function connectMobileRelayRpcSession(args: {
 }): MobileRelayRpcSession {
  const requestTimeoutMs = args.requestTimeoutMs ?? 30_000
  const pending = new Map<string, PendingRequest>()
+  const timedOutControlRequestIds = new Set<string>()
  const stateListeners = new Set<(state: ConnectionState) => void>()
  let state: ConnectionState = 'connecting'
  let requestCounter = 0
  let controlResponseSequence = 0
  let controlProbeInFlight = false
+  let controlProbeTimer: ReturnType<typeof setInterval> | null = null
  let lastConnectedAt: number | null = null
  let leaseExpiresAt: number | null = null
  let resumeConfirmation: DeviceResumeConfirmed | null = null
@@ -102,12 +105,16 @@ export function connectMobileRelayRpcSession(args: {
      stateListeners.add(listener)
      return () => stateListeners.delete(listener)
    },
-    notifyForeground: () => {},
+    notifyForeground() {
+      startControlProbeTimer()
+      probeControlPlane()
+    },
    close() {
      if (closed) {
        return
      }
      closed = true
+      stopControlProbeTimer()
      link.close()
      rejectPending(new Error('Client closed'))
      streams.clear()
@@ -137,6 +144,7 @@ export function connectMobileRelayRpcSession(args: {
      resumeConfirmation = result.resumeConfirmation
      lastConnectedAt = Date.now()
      publishState('connected')
+      startControlProbeTimer()
    } catch (error) {
      fail(asError(error))
    }
@@ -155,6 +163,7 @@ export function connectMobileRelayRpcSession(args: {
    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => {
        pending.delete(id)
+        timedOutControlRequestIds.add(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)
@@ -184,8 +193,10 @@ export function connectMobileRelayRpcSession(args: {
      return
    }
    const request = pending.get(value.id)
-    if (request) {
+    if (request || timedOutControlRequestIds.delete(value.id)) {
      controlResponseSequence += 1
+    }
+    if (request) {
      clearTimeout(request.timer)
      pending.delete(value.id)
      request.resolve(value)
@@ -206,25 +217,41 @@ export function connectMobileRelayRpcSession(args: {
    const probeControlResponseSequence = controlResponseSequence
    void sendRpc('status.get', undefined, CONTROL_PROBE_TIMEOUT_MS).then(
      () => {
-        controlProbeInFlight = false
+        finishControlProbe()
      },
      (error: unknown) => {
-        controlProbeInFlight = false
+        const controlResponded = controlResponseSequence > probeControlResponseSequence
+        finishControlProbe()
        if (closed || state !== 'connected') {
          return
        }
-        if (
-          !isRpcDeliveryUnknown(error) ||
-          controlResponseSequence === probeControlResponseSequence
-        ) {
+        if (!isRpcDeliveryUnknown(error) || !controlResponded) {
          fail(asError(error))
-        } else {
-          probeControlPlane()
        }
      }
    )
  }

+  function startControlProbeTimer(): void {
+    if (closed || state !== 'connected' || controlProbeTimer) {
+      return
+    }
+    controlProbeTimer = setInterval(probeControlPlane, CONTROL_PROBE_INTERVAL_MS)
+  }
+
+  function stopControlProbeTimer(): void {
+    if (controlProbeTimer) {
+      clearInterval(controlProbeTimer)
+      controlProbeTimer = null
+    }
+    finishControlProbe()
+  }
+
+  function finishControlProbe(): void {
+    controlProbeInFlight = false
+    timedOutControlRequestIds.clear()
+  }
+
  function waitForConnected(timeoutMs = requestTimeoutMs): Promise<void> {
    if (state === 'connected') {
      return Promise.resolve()
@@ -268,6 +295,7 @@ export function connectMobileRelayRpcSession(args: {
      return
    }
    closed = true
+    stopControlProbeTimer()
    failure = error
    link.close()
    rejectPending(error)
@@ -0,0 +1,18 @@
+import type { RpcClient } from './rpc-client'
+import type { MobileConnectionPath } from './stable-logical-rpc-client'
+import type { ConnectionState, HostProfile } from './types'
+
+export type RpcClientContextValue = {
+  acquire: (hostId: string, host?: HostProfile) => RpcClient | null
+  release: (hostId: string) => void
+  forceReconnect: (hostId: string) => Promise<void>
+  closeHost: (hostId: string) => void
+  getState: (hostId: string) => ConnectionState
+  getReconnectAttempt: (hostId: string) => number
+  getLastConnectedAt: (hostId: string) => number | null
+  getActivePath: (hostId: string) => MobileConnectionPath
+  subscribeHostState: (hostId: string, listener: (state: ConnectionState) => void) => () => void
+  getAllClients: () => Array<{ hostId: string; client: RpcClient }>
+  subscribeAllHosts: (listener: () => void) => () => void
+  primeHosts: (hosts: HostProfile[]) => void
+}
@@ -203,6 +203,28 @@ describe('mobile rpc-client request deadline', () => {
    await request.catch(() => undefined)
  })

+  it('counts a late timed-out reply as control-plane liveness', async () => {
+    const client = connect('ws://desktop.invalid', 'token', 'server-key')
+    const socket = mockSockets[0]!
+    socket.open()
+    const request = client.sendRequest('browser.screenshot', {}, { timeoutMs: 100 })
+    const outcome = track(request)
+    await vi.advanceTimersByTimeAsync(0)
+    const timedOutRequest = sentRequest(socket, 'browser.screenshot')
+
+    await vi.advanceTimersByTimeAsync(100)
+    expect(outcome.read()).toBe('Request timed out: browser.screenshot')
+    expect(sentRequest(socket, 'status.get')).toBeDefined()
+    socket.receive(`encrypted:${JSON.stringify({ id: timedOutRequest.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]!
@@ -36,6 +36,10 @@ import {
 import { markRpcDeliveryUnknown } from './rpc-delivery-ambiguity'
 import { openRpcRequestBudget, resolvePostConnectRequestTimeout } from './rpc-request-budget'
 import { isRpcResponse } from './rpc-response-shape'
+import {
+  isStreamingSubscriptionReadyResult,
+  isTerminalSubscribedResult
+} from './rpc-stream-response-shape'
 import { websocketPayloadToUint8 } from './websocket-payload-bytes'

 type PendingRequest = {
@@ -186,6 +190,7 @@ export function connect(
  const serverPublicKey = publicKeyFromBase64(serverPublicKeyB64)

  const pending = new Map<string, PendingRequest>()
+  const timedOutControlRequestIds = new Set<string>()
  const streamListeners = new Map<string, StreamRequest>()
  const terminalStreamListeners = new Map<number, StreamingListener>()
  const terminalStreamIdsByRequest = new Map<string, Set<number>>()
@@ -502,7 +507,7 @@ export function connect(
      if (!isRpcResponse(response)) {
        return
      }
-      if (pending.has(response.id)) {
+      if (pending.has(response.id) || timedOutControlRequestIds.delete(response.id)) {
        controlResponseSequence++
      }

@@ -658,6 +663,7 @@ export function connect(
      handshakeTimer = null
    }
    stopActivityProbe()
+    finishActivityProbe()
    if (intentionallyClosed) {
      console.log('[net] handleSocketClosed — intentional close')
      setState('disconnected')
@@ -779,9 +785,10 @@ export function connect(
    let timedOut = false
    const timeout = setTimeout(() => {
      timedOut = true
-      activityProbeInFlight = false
      pending.delete(id)
-      if (controlResponseSequence > probeControlResponseSequence) {
+      const controlResponded = controlResponseSequence > probeControlResponseSequence
+      finishActivityProbe()
+      if (controlResponded) {
        return
      }
      console.log('[net] activity-probe TIMEOUT — forcing reconnect', { state })
@@ -792,24 +799,29 @@ export function connect(
        if (timedOut) {
          return
        }
-        activityProbeInFlight = false
        clearTimeout(timeout)
+        finishActivityProbe()
      },
      reject: () => {
        if (timedOut) {
          return
        }
-        activityProbeInFlight = false
        clearTimeout(timeout)
+        finishActivityProbe()
      }
    })
    if (!sendEncrypted({ id, deviceToken, method: 'status.get' })) {
-      activityProbeInFlight = false
      clearTimeout(timeout)
      pending.delete(id)
+      finishActivityProbe()
    }
  }

+  function finishActivityProbe(): void {
+    activityProbeInFlight = false
+    timedOutControlRequestIds.clear()
+  }
+
  function forceSocketReconnect(socket: WebSocket | null): void {
    if (!socket || socket !== ws) {
      return
@@ -1037,6 +1049,9 @@ 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) {
+            if (requestWs === ws) {
+              timedOutControlRequestIds.add(id)
+            }
            runActivityProbe(requestWs)
          }
        }, timeoutMs)
@@ -1191,6 +1206,7 @@ export function connect(
        handshakeTimer = null
      }
      stopActivityProbe()
+      finishActivityProbe()
      if (ws) {
        ws.close()
        ws = null
@@ -1202,25 +1218,3 @@ export function connect(
    }
  }
 }
-
-function isTerminalSubscribedResult(
-  value: unknown
-): value is { type: 'subscribed'; streamId: number } {
-  return (
-    !!value &&
-    typeof value === 'object' &&
-    (value as { type?: unknown }).type === 'subscribed' &&
-    typeof (value as { streamId?: unknown }).streamId === 'number'
-  )
-}
-
-function isStreamingSubscriptionReadyResult(
-  value: unknown
-): value is { type: 'ready'; subscriptionId: string } {
-  return (
-    !!value &&
-    typeof value === 'object' &&
-    (value as { type?: unknown }).type === 'ready' &&
-    typeof (value as { subscriptionId?: unknown }).subscriptionId === 'string'
-  )
-}
@@ -0,0 +1,21 @@
+export function isTerminalSubscribedResult(
+  value: unknown
+): value is { type: 'subscribed'; streamId: number } {
+  return (
+    !!value &&
+    typeof value === 'object' &&
+    (value as { type?: unknown }).type === 'subscribed' &&
+    typeof (value as { streamId?: unknown }).streamId === 'number'
+  )
+}
+
+export function isStreamingSubscriptionReadyResult(
+  value: unknown
+): value is { type: 'ready'; subscriptionId: string } {
+  return (
+    !!value &&
+    typeof value === 'object' &&
+    (value as { type?: unknown }).type === 'ready' &&
+    typeof (value as { subscriptionId?: unknown }).subscriptionId === 'string'
+  )
+}