fix(mobile): close reconnect lifecycle review gaps

mobile/app/h/[hostId]/edit.tsx

@@ -17,10 +17,7 @@ import { colors, radii, spacing, typography } from '../../../src/theme/mobile-th
 import { loadHosts, updateHostNameAndEndpoint } from '../../../src/transport/host-store'
 import { getHostListLoadRevision } from '../../../src/transport/host-list-load-sharing'
 import { displayHostEndpoint } from '../../../src/transport/host-endpoint'
-import {
-  hostProfileAfterEdit,
-  resolveHostEndpointEdit
-} from '../../../src/transport/host-endpoint-edit'
+import { resolveHostEndpointEdit } from '../../../src/transport/host-endpoint-edit'
 import { useForceReconnect, usePrimeHosts } from '../../../src/transport/client-context'
 import { showForceReconnectError } from '../../../src/transport/force-reconnect-feedback'
 import type { HostProfile } from '../../../src/transport/types'
@@ -109,7 +106,7 @@ export default function EditHostScreen() {
       ...(willRename ? { name: nextName } : {}),
       ...(nextEndpoint !== undefined ? { endpoint: nextEndpoint } : {})
     }
-    let reconnectProfile: HostProfile | null = hostProfileAfterEdit(host, updates)
+    let reconnectWithCurrentProfile = false
 
     savingRef.current = true
     setSaving(true)
@@ -130,11 +127,20 @@ export default function EditHostScreen() {
       // Why: the write already committed above; a re-prime failure here
       // must not be reported as a save failure — the next loadHosts() call
       // elsewhere in the app picks up the fresh state regardless.
-      const hostLoadRevision = getHostListLoadRevision()
-      const hosts = await loadHosts()
-      if (hostLoadRevision === getHostListLoadRevision()) {
+      for (let attempt = 0; attempt < 3; attempt += 1) {
+        const hostLoadRevision = getHostListLoadRevision()
+        const hosts = await loadHosts()
+        if (hostLoadRevision !== getHostListLoadRevision()) {
+          continue
+        }
+        if (!hosts.some(({ id }) => id === host.id)) {
+          break
+        }
         primeHosts(hosts, hostLoadRevision)
-        reconnectProfile = hosts.find(({ id }) => id === host.id) ?? null
+        if (hostLoadRevision === getHostListLoadRevision()) {
+          reconnectWithCurrentProfile = true
+          break
+        }
       }
     } catch {
       // best-effort re-prime; persisted data is unaffected
@@ -144,11 +150,11 @@ export default function EditHostScreen() {
     setSaving(false)
     router.back()
 
-    if (nextEndpoint !== undefined && reconnectProfile !== null) {
+    if (nextEndpoint !== undefined && reconnectWithCurrentProfile) {
       // Why: reconnect is a follow-on side effect of a save that already
       // committed — its failure or a hang must not be reported as a save
       // failure or block navigating back.
-      void forceReconnectHost(host.id, reconnectProfile).catch(showForceReconnectError)
+      void forceReconnectHost(host.id).catch(showForceReconnectError)
     }
   }
 

mobile/src/host-edit-save-flow.test.ts

@@ -186,7 +186,8 @@ describe('edit host handleSave', () => {
     expect(dependencies.updateHostNameAndEndpoint).toHaveBeenCalledWith('host-1', {
       endpoint: 'ws://192.168.1.20:6768'
     })
-    expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1', savedHost)
+    expect(dependencies.primeHosts).toHaveBeenCalledWith([savedHost], 0)
+    expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1')
     expect(dependencies.back).toHaveBeenCalledTimes(1)
 
     act(() => renderer.unmount())
@@ -209,7 +210,8 @@ describe('edit host handleSave', () => {
       name: 'Home Desk',
       endpoint: 'ws://192.168.1.20:6768'
     })
-    expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1', savedHost)
+    expect(dependencies.primeHosts).toHaveBeenCalledWith([savedHost], 0)
+    expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1')
     expect(dependencies.back).toHaveBeenCalledTimes(1)
 
     act(() => renderer.unmount())
@@ -265,7 +267,7 @@ describe('edit host handleSave', () => {
       await Promise.resolve()
     })
 
-    expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1', savedHost)
+    expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1')
     expect(dependencies.back).toHaveBeenCalledTimes(1)
     expect(dependencies.alert).toHaveBeenCalledWith('Unable to reconnect', 'connect failed')
 
@@ -286,24 +288,29 @@ describe('edit host handleSave', () => {
     setFieldValue(renderer, 'Address', '192.168.1.20:6768')
     await pressSave(renderer)
 
-    expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1', relayHost)
+    expect(dependencies.primeHosts).toHaveBeenCalledWith([relayHost], 0)
+    expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1')
 
     act(() => renderer.unmount())
   })
 
-  it('preserves the edited profile when an unrelated write invalidates the loaded snapshot', async () => {
+  it('reloads the current edited profile when an unrelated write invalidates a snapshot', async () => {
     const editedHost = { ...HOST_FIXTURE, endpoint: 'ws://192.168.1.20:6768' }
     const invalidatedHost = { ...HOST_FIXTURE, name: 'Stale snapshot' }
     dependencies.loadHosts
       .mockResolvedValueOnce([HOST_FIXTURE])
       .mockResolvedValueOnce([invalidatedHost])
-    dependencies.getHostListLoadRevision.mockReturnValueOnce(1).mockReturnValueOnce(2)
+      .mockResolvedValueOnce([editedHost])
+    dependencies.getHostListLoadRevision
+      .mockReturnValueOnce(1)
+      .mockReturnValueOnce(2)
+      .mockReturnValue(2)
     const renderer = await renderEditHostRoute()
     setFieldValue(renderer, 'Address', '192.168.1.20:6768')
     await pressSave(renderer)
 
-    expect(dependencies.primeHosts).not.toHaveBeenCalled()
-    expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1', editedHost)
+    expect(dependencies.primeHosts).toHaveBeenCalledWith([editedHost], 2)
+    expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1')
 
     act(() => renderer.unmount())
   })

mobile/src/transport/existing-host-relay-routing-lifecycle.test.ts

@@ -12,6 +12,7 @@ vi.mock('./mobile-relay-host-overlay-store', () => overlayStoreMock)
 vi.mock('./host-list-load-sharing', () => ({ dropSharedHostListLoad: vi.fn() }))
 
 import {
+  MobileRelayUpgradeLifecycleRetiredError,
   MobileRelayUpgradeHostSupersededError,
   saveExistingHostRelayRouting,
   writeExistingHostRelayCredentialBundle
@@ -69,7 +70,7 @@ describe('existing host relay endpoint lifecycle publication', () => {
     )
     releaseOldToken?.(HOST.deviceToken)
 
-    await expect(oldWrite).rejects.toBeInstanceOf(MobileRelayUpgradeHostSupersededError)
+    await expect(oldWrite).rejects.toBeInstanceOf(MobileRelayUpgradeLifecycleRetiredError)
     expect(writes).toEqual(['replacement'])
   })
 
@@ -97,7 +98,7 @@ describe('existing host relay endpoint lifecycle publication', () => {
     )
     releaseOldToken?.(HOST.deviceToken)
 
-    await expect(stalePublication).rejects.toBeInstanceOf(MobileRelayUpgradeHostSupersededError)
+    await expect(stalePublication).rejects.toBeInstanceOf(MobileRelayUpgradeLifecycleRetiredError)
     expect(overlayStoreMock.saveMobileRelayHostOverlay).toHaveBeenCalledOnce()
     expect(overlayStoreMock.saveMobileRelayHostOverlay).toHaveBeenCalledWith(
       expect.objectContaining({ relayHostId: 'AbCdEf0123_-new9' })
@@ -125,7 +126,7 @@ describe('existing host relay endpoint lifecycle publication', () => {
     beginHostEndpointPublicationLifecycle(HOST.id)
     releaseMetadata?.(HOST)
 
-    await expect(oldWrite).rejects.toBeInstanceOf(MobileRelayUpgradeHostSupersededError)
+    await expect(oldWrite).rejects.toBeInstanceOf(MobileRelayUpgradeLifecycleRetiredError)
     expect(writeBundle).not.toHaveBeenCalled()
   })
 
@@ -151,7 +152,7 @@ describe('existing host relay endpoint lifecycle publication', () => {
     beginHostEndpointPublicationLifecycle(HOST.id)
     releaseCredentialWrite?.()
 
-    await expect(oldWrite).rejects.toBeInstanceOf(MobileRelayUpgradeHostSupersededError)
+    await expect(oldWrite).rejects.toBeInstanceOf(MobileRelayUpgradeLifecycleRetiredError)
     expect(overlayStoreMock.saveMobileRelayHostOverlay).not.toHaveBeenCalled()
   })
 
@@ -191,6 +192,40 @@ describe('existing host relay endpoint lifecycle publication', () => {
     await expect(staleWrite).rejects.toBeInstanceOf(MobileRelayUpgradeHostSupersededError)
     expect(writeBundle).not.toHaveBeenCalled()
   })
+
+  it('binds an endpoint lifecycle to the profile revision before same-identity recovery', async () => {
+    let releaseRecovery: (() => void) | null = null
+    let markRecoveryStarted: (() => void) | null = null
+    const recoveryStarted = new Promise<void>((resolve) => {
+      markRecoveryStarted = resolve
+    })
+    const recoveryPending = new Promise<void>((resolve) => {
+      releaseRecovery = resolve
+    })
+    const oldLifecycle = beginHostEndpointPublicationLifecycle(HOST.id)
+    const recovery = publishHostProfileTransaction(
+      HOST,
+      async () => {
+        markRecoveryStarted?.()
+        await recoveryPending
+      },
+      async () => {}
+    )
+    await recoveryStarted
+
+    const writeBundle = vi.fn(async () => {})
+    const staleWrite = writeExistingHostRelayCredentialBundle(
+      HOST,
+      credentialBundle(1),
+      writeBundle,
+      oldLifecycle
+    )
+    releaseRecovery?.()
+
+    await recovery
+    await expect(staleWrite).rejects.toBeInstanceOf(MobileRelayUpgradeLifecycleRetiredError)
+    expect(writeBundle).not.toHaveBeenCalled()
+  })
 })
 
 function credentialBundle(version: number): MobileRelayCredentialBundle {

mobile/src/transport/existing-host-relay-routing.ts

@@ -2,9 +2,10 @@ import { HostProfileSchema, type HostProfile } from './types'
 import { loadStoredHostIdentity } from './host-store'
 import { readHostDeviceToken } from './host-device-token-store'
 import {
-  getHostEndpointPublicationGeneration,
+  getHostEndpointPublicationLifecycle,
   getHostProfilePublicationRevision,
-  serializeHostProfilePublication
+  serializeHostProfilePublication,
+  type HostEndpointPublicationLifecycle
 } from './host-profile-publication'
 import * as hostListLoads from './host-list-load-sharing'
 import { saveMobileRelayHostOverlay } from './mobile-relay-host-overlay-store'
@@ -12,28 +13,28 @@ import type { MobileRelayCredentialBundle } from './mobile-relay-credential-bund
 
 export class MobileRelayUpgradeHostRemovedError extends Error {}
 export class MobileRelayUpgradeHostSupersededError extends Error {}
+export class MobileRelayUpgradeLifecycleRetiredError extends Error {}
 
 export async function saveExistingHostRelayRouting(
   host: HostProfile,
   beforePublish?: () => Promise<void>,
-  endpointGeneration = getHostEndpointPublicationGeneration(host.id)
+  endpointLifecycle = getHostEndpointPublicationLifecycle(host.id)
 ): Promise<void> {
   const validated = HostProfileSchema.parse(host)
-  const revision = getHostProfilePublicationRevision(validated.id)
   await requireCurrentHostCredential(validated)
   await serializeHostProfilePublication(validated.id, async () => {
-    requireCurrentPublication(validated.id, revision, endpointGeneration)
+    requireCurrentEndpointGeneration(validated.id, endpointLifecycle)
     await requireCurrentHostDeviceToken(validated)
-    requireCurrentPublication(validated.id, revision, endpointGeneration)
+    requireCurrentEndpointGeneration(validated.id, endpointLifecycle)
     const existing = await requireCurrentHostMetadata(validated)
-    requireCurrentPublication(validated.id, revision, endpointGeneration)
+    requireCurrentPublication(validated.id, endpointLifecycle)
     const { endpoints, relayHostId, relay } = validated
     if (!endpoints || !relayHostId || !relay) {
       throw new Error('mobile relay upgrade routing metadata missing')
     }
     // Why: serialize the credential before its routing overlay so neither side can cross a re-pair.
     await beforePublish?.()
-    requireCurrentPublication(validated.id, revision, endpointGeneration)
+    requireCurrentPublication(validated.id, endpointLifecycle)
     await saveMobileRelayHostOverlay({
       v: 2,
       hostId: validated.id,
@@ -53,20 +54,19 @@ export async function writeExistingHostRelayCredentialBundle(
   host: HostProfile,
   bundle: MobileRelayCredentialBundle,
   writeBundle: (bundle: MobileRelayCredentialBundle) => Promise<void>,
-  endpointGeneration = getHostEndpointPublicationGeneration(host.id)
+  endpointLifecycle = getHostEndpointPublicationLifecycle(host.id)
 ): Promise<void> {
   const validated = HostProfileSchema.parse(host)
-  const revision = getHostProfilePublicationRevision(validated.id)
   await requireCurrentHostCredential(validated)
   await serializeHostProfilePublication(validated.id, async () => {
-    requireCurrentPublication(validated.id, revision, endpointGeneration)
+    requireCurrentEndpointGeneration(validated.id, endpointLifecycle)
     await requireCurrentHostDeviceToken(validated)
-    requireCurrentPublication(validated.id, revision, endpointGeneration)
+    requireCurrentEndpointGeneration(validated.id, endpointLifecycle)
     await requireCurrentHostMetadata(validated)
     if (bundle.hostId !== validated.id || bundle.deviceToken !== validated.deviceToken) {
       throw new MobileRelayUpgradeHostSupersededError('mobile relay credential identity mismatch')
     }
-    requireCurrentPublication(validated.id, revision, endpointGeneration)
+    requireCurrentPublication(validated.id, endpointLifecycle)
     await writeBundle(bundle)
   })
 }
@@ -106,13 +106,21 @@ function requireMatchingHostMetadata(
 
 function requireCurrentPublication(
   hostId: string,
-  revision: number,
-  endpointGeneration: number
+  endpointLifecycle: HostEndpointPublicationLifecycle
 ): void {
   if (
-    getHostProfilePublicationRevision(hostId) !== revision ||
-    getHostEndpointPublicationGeneration(hostId) !== endpointGeneration
+    getHostProfilePublicationRevision(hostId) !== endpointLifecycle.profileRevision ||
+    getHostEndpointPublicationLifecycle(hostId).generation !== endpointLifecycle.generation
   ) {
-    throw new MobileRelayUpgradeHostSupersededError('mobile relay upgrade host was re-paired')
+    throw new MobileRelayUpgradeLifecycleRetiredError('mobile relay endpoint lifecycle was retired')
+  }
+}
+
+function requireCurrentEndpointGeneration(
+  hostId: string,
+  endpointLifecycle: HostEndpointPublicationLifecycle
+): void {
+  if (getHostEndpointPublicationLifecycle(hostId).generation !== endpointLifecycle.generation) {
+    throw new MobileRelayUpgradeLifecycleRetiredError('mobile relay endpoint lifecycle was retired')
   }
 }

mobile/src/transport/force-reconnect-rpc-health.test.ts

@@ -2,6 +2,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest'
 import type { RpcClient, SendRequestOptions } from './rpc-client'
 import type { ConnectionState, RpcResponse } from './types'
 import { verifyForceReconnectRpcHealth } from './force-reconnect-rpc-health'
+import { markRpcDeliveryUnknown } from './rpc-delivery-ambiguity'
 import { LogicalClientCutoverError } from './stable-logical-rpc-client'
 
 describe('Force Reconnect RPC health', () => {
@@ -79,4 +80,34 @@ describe('Force Reconnect RPC health', () => {
     )
     expect(sendRequest).toHaveBeenCalledOnce()
   })
+
+  it('waits for a Relay replacement after the active session drops', async () => {
+    let state: ConnectionState = 'connected'
+    const listeners = new Set<(next: ConnectionState) => void>()
+    const sendRequest = vi
+      .fn<() => Promise<RpcResponse>>()
+      .mockImplementationOnce(async () => {
+        state = 'disconnected'
+        throw markRpcDeliveryUnknown(new Error('relay RPC interrupted'))
+      })
+      .mockResolvedValueOnce({ id: 'rpc-2', ok: true, result: {} })
+    const client = {
+      sendRequest,
+      getState: () => state,
+      onStateChange: (listener: (next: ConnectionState) => void) => {
+        listeners.add(listener)
+        return () => listeners.delete(listener)
+      }
+    } as unknown as RpcClient
+
+    const verification = verifyForceReconnectRpcHealth(client)
+    await vi.waitFor(() => expect(listeners.size).toBe(1))
+    state = 'connected'
+    for (const listener of listeners) {
+      listener(state)
+    }
+
+    await expect(verification).resolves.toBeUndefined()
+    expect(sendRequest).toHaveBeenCalledTimes(2)
+  })
 })

mobile/src/transport/force-reconnect-rpc-health.ts

@@ -22,15 +22,61 @@ export async function verifyForceReconnectRpcHealth(
       })
       return
     } catch (error) {
-      if (
-        client.getState() !== 'reconnecting' &&
-        !isLogicalClientCutoverError(error) &&
-        !isRpcDeliveryUnknown(error) &&
-        !(error instanceof Error && error.message === 'Connection interrupted')
-      ) {
+      const state = client.getState()
+      if (state === 'auth-failed' || !isRecoverableHealthError(error, state)) {
         throw error
       }
       lastError = error
+      if (state === 'disconnected') {
+        await waitForReconnectState(client, deadline, error)
+      }
     }
   }
 }
+
+function isRecoverableHealthError(
+  error: unknown,
+  state: ReturnType<RpcClient['getState']>
+): boolean {
+  return (
+    state === 'reconnecting' ||
+    isLogicalClientCutoverError(error) ||
+    isRpcDeliveryUnknown(error) ||
+    (error instanceof Error &&
+      ['Connection interrupted', 'relay session not connected'].includes(error.message))
+  )
+}
+
+function waitForReconnectState(
+  client: RpcClient,
+  deadline: number,
+  lastError: unknown
+): Promise<void> {
+  if (client.getState() !== 'disconnected') {
+    return Promise.resolve()
+  }
+  return new Promise((resolve, reject) => {
+    const timeoutMs = deadline - Date.now()
+    if (timeoutMs <= 0) {
+      reject(lastError)
+      return
+    }
+    const timer = setTimeout(() => finish(() => reject(lastError)), timeoutMs)
+    const unsubscribe = client.onStateChange((state) => {
+      if (state === 'disconnected') {
+        return
+      }
+      finish(state === 'auth-failed' ? () => reject(lastError) : resolve)
+    })
+    const currentState = client.getState()
+    if (currentState !== 'disconnected') {
+      finish(currentState === 'auth-failed' ? () => reject(lastError) : resolve)
+    }
+
+    function finish(settle: () => void): void {
+      clearTimeout(timer)
+      unsubscribe()
+      settle()
+    }
+  })
+}

mobile/src/transport/host-profile-publication.ts

@@ -4,18 +4,30 @@ const pendingByHost = new Map<string, Promise<void>>()
 const revisionByHost = new Map<string, number>()
 const endpointGenerationByHost = new Map<string, number>()
 
+export type HostEndpointPublicationLifecycle = Readonly<{
+  generation: number
+  profileRevision: number
+}>
+
 export function getHostProfilePublicationRevision(hostId: string): number {
   return revisionByHost.get(hostId) ?? 0
 }
 
-export function beginHostEndpointPublicationLifecycle(hostId: string): number {
+export function beginHostEndpointPublicationLifecycle(
+  hostId: string
+): HostEndpointPublicationLifecycle {
   const generation = (endpointGenerationByHost.get(hostId) ?? 0) + 1
   endpointGenerationByHost.set(hostId, generation)
-  return generation
+  return { generation, profileRevision: getHostProfilePublicationRevision(hostId) }
 }
 
-export function getHostEndpointPublicationGeneration(hostId: string): number {
-  return endpointGenerationByHost.get(hostId) ?? 0
+export function getHostEndpointPublicationLifecycle(
+  hostId: string
+): HostEndpointPublicationLifecycle {
+  return {
+    generation: endpointGenerationByHost.get(hostId) ?? 0,
+    profileRevision: getHostProfilePublicationRevision(hostId)
+  }
 }
 
 export function serializeHostProfilePublication<T>(

mobile/src/transport/host-store.test.ts

@@ -432,7 +432,6 @@ describe('host-store list mutations', () => {
           releaseOldToken = resolve
         })
     )
-    const relayHostId = 'AbCdEf0123_-xyZ9'
     const stalePublication = saveExistingHostRelayRouting({
       ...HOST_ONE,
       deviceToken: 'token-1',
@@ -440,13 +439,13 @@ describe('host-store list mutations', () => {
         { id: 'direct-primary', kind: 'lan', url: HOST_ONE.endpoint },
         { id: 'relay-primary', kind: 'relay', url: 'wss://relay.onorca.dev/v1/connect/host' }
       ],
-      relayHostId,
+      relayHostId: 'AbCdEf0123_-xyZ9',
       relay: {
         v: 1,
         directorUrl: 'https://relay.onorca.dev',
         cellUrl: 'https://relay.onorca.dev',
         assignmentEpoch: 1,
-        relayHostId,
+        relayHostId: 'AbCdEf0123_-xyZ9',
         e2eeFraming: 2
       }
     })
@@ -476,13 +475,14 @@ describe('host-store list mutations', () => {
 
   it('lets same-host re-pair supersede a publication stalled on identity lookup', async () => {
     let releaseOldToken: ((token: string | null) => void) | null = null
-    secureStoreMock.getItemAsync.mockImplementationOnce(
-      () =>
-        new Promise<string | null>((resolve) => {
-          releaseOldToken = resolve
-        })
-    )
-    const relayHostId = 'AbCdEf0123_-xyZ9'
+    secureStoreMock.getItemAsync
+      .mockImplementationOnce(
+        () =>
+          new Promise<string | null>((resolve) => {
+            releaseOldToken = resolve
+          })
+      )
+      .mockResolvedValue('replacement-token')
     const stalePublication = saveExistingHostRelayRouting({
       ...HOST_ONE,
       deviceToken: 'token-1',
@@ -490,13 +490,13 @@ describe('host-store list mutations', () => {
         { id: 'direct-primary', kind: 'lan', url: HOST_ONE.endpoint },
         { id: 'relay-primary', kind: 'relay', url: 'wss://relay.onorca.dev/v1/connect/host' }
       ],
-      relayHostId,
+      relayHostId: 'AbCdEf0123_-xyZ9',
       relay: {
         v: 1,
         directorUrl: 'https://relay.onorca.dev',
         cellUrl: 'https://relay.onorca.dev',
         assignmentEpoch: 1,
-        relayHostId,
+        relayHostId: 'AbCdEf0123_-xyZ9',
         e2eeFraming: 2
       }
     })

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

@@ -12,7 +12,10 @@ import {
   saveExistingHostRelayRouting,
   writeExistingHostRelayCredentialBundle
 } from './existing-host-relay-routing'
-import { beginHostEndpointPublicationLifecycle } from './host-profile-publication'
+import {
+  beginHostEndpointPublicationLifecycle,
+  type HostEndpointPublicationLifecycle
+} from './host-profile-publication'
 import { upgradeDirectMobileRelay } from './mobile-relay-direct-upgrade'
 import { MobileRelayDirectUpgradeController } from './mobile-relay-direct-upgrade-controller'
 import type { StableLogicalRpcClient } from './stable-logical-rpc-client'
@@ -35,7 +38,7 @@ export function startMobileEndpointLifecycle(
   let stopped = false
   let foreground = true
   let owner: EndpointOwner
-  const endpointGeneration = beginHostEndpointPublicationLifecycle(initialHost.id)
+  const endpointLifecycle = beginHostEndpointPublicationLifecycle(initialHost.id)
   const publishHostUpdate = (host: HostProfile): void => {
     if (!stopped) {
       onHostUpdated(host)
@@ -46,7 +49,7 @@ export function startMobileEndpointLifecycle(
     if (stopped) {
       return
     }
-    const supervisor = createSupervisor(logical, host, onLog, publishHostUpdate, endpointGeneration)
+    const supervisor = createSupervisor(logical, host, onLog, publishHostUpdate, endpointLifecycle)
     owner.stop()
     owner = supervisor
     supervisor.setForeground(foreground)
@@ -54,7 +57,7 @@ export function startMobileEndpointLifecycle(
   }
 
   if (initialHost.relay) {
-    owner = createSupervisor(logical, initialHost, onLog, publishHostUpdate, endpointGeneration)
+    owner = createSupervisor(logical, initialHost, onLog, publishHostUpdate, endpointLifecycle)
     void owner.start()
   } else {
     owner = new MobileRelayDirectUpgradeController(logical, initialHost, {
@@ -65,7 +68,7 @@ export function startMobileEndpointLifecycle(
           dependencies: {
             randomBytes: ExpoCrypto.getRandomBytes,
             saveHost: (updatedHost, beforePublish) =>
-              saveExistingHostRelayRouting(updatedHost, beforePublish, endpointGeneration)
+              saveExistingHostRelayRouting(updatedHost, beforePublish, endpointLifecycle)
           }
         }),
       onUpgraded: async ({ host }) => {
@@ -93,7 +96,7 @@ function createSupervisor(
   host: HostProfile,
   onLog: ConnectionLogSink,
   onHostUpdated: (host: HostProfile) => void,
-  endpointGeneration: number
+  endpointLifecycle: HostEndpointPublicationLifecycle
 ): MobileEndpointSupervisor {
   return new MobileEndpointSupervisor(logical, host, {
     openDirect: (endpoint) => connect(endpoint, host.deviceToken, host.publicKeyB64, { onLog }),
@@ -113,10 +116,10 @@ function createSupervisor(
         host,
         bundle,
         writeMobileRelayCredentialBundle,
-        endpointGeneration
+        endpointLifecycle
       ),
     saveHost: (updatedHost, beforePublish) =>
-      saveExistingHostRelayRouting(updatedHost, beforePublish, endpointGeneration),
+      saveExistingHostRelayRouting(updatedHost, beforePublish, endpointLifecycle),
     onHostUpdated,
     now: Date.now,
     randomBytes: ExpoCrypto.getRandomBytes,

mobile/src/transport/mobile-relay-direct-upgrade.test.ts

@@ -1,6 +1,7 @@
 import { describe, expect, it, vi } from 'vitest'
 import type { MobileRelayEndpoint } from '../../../src/shared/mobile-relay-credential-contract'
 import {
+  MobileRelayUpgradeLifecycleRetiredError,
   MobileRelayUpgradeHostRemovedError,
   MobileRelayUpgradeHostSupersededError
 } from './existing-host-relay-routing'
@@ -181,6 +182,30 @@ describe('existing direct pairing relay upgrade', () => {
     await expect(deps.readJournal()).resolves.toEqual(replacement)
   })
 
+  it('retains a committed journal when an endpoint lifecycle is retired', async () => {
+    const journal = createMobileRelayDirectUpgradeJournal(host.id, (length) =>
+      new Uint8Array(length).fill(6)
+    )
+    const committed = installed(journal)
+    const deps = dependencies(journal)
+    deps.saveHost.mockRejectedValue(
+      new MobileRelayUpgradeLifecycleRetiredError('mobile relay endpoint lifecycle was retired')
+    )
+    const client = clientWith([
+      success({
+        v: 1,
+        relay,
+        installStatus: { v: 1, reqId: journal.reqId, state: 'committed', result: committed }
+      })
+    ])
+
+    await expect(
+      upgradeDirectMobileRelay({ client, host, dependencies: deps })
+    ).rejects.toBeInstanceOf(MobileRelayUpgradeLifecycleRetiredError)
+    expect(deps.clearJournal).not.toHaveBeenCalled()
+    await expect(deps.readJournal()).resolves.toEqual(journal)
+  })
+
   it('retains recovery state when host identity credentials are temporarily unavailable', async () => {
     const journal = createMobileRelayDirectUpgradeJournal(host.id, (length) =>
       new Uint8Array(length).fill(8)

mobile/src/transport/mobile-relay-pending-requests.ts

@@ -0,0 +1,24 @@
+import { markRpcDeliveryUnknown } from './rpc-delivery-ambiguity'
+import type { RpcResponse } from './types'
+
+export type MobileRelayPendingRequest = {
+  resolve: (response: RpcResponse) => void
+  reject: (error: Error) => void
+  timer: ReturnType<typeof setTimeout>
+}
+
+export function rejectMobileRelayPendingRequests(
+  pending: Map<string, MobileRelayPendingRequest>,
+  error: Error
+): void {
+  if (pending.size === 0) {
+    return
+  }
+  // Written frames may have executed before the Relay session failed.
+  markRpcDeliveryUnknown(error)
+  for (const request of pending.values()) {
+    clearTimeout(request.timer)
+    request.reject(error)
+  }
+  pending.clear()
+}

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

@@ -334,6 +334,63 @@ describe('mobile relay RPC session', () => {
     }
   })
 
+  it.each([
+    ['lease-only terminal', { type: 'subscribed', streamId: null }],
+    ['native-chat snapshot', { type: 'snapshot' }],
+    ['session-tabs snapshot', { type: 'snapshot' }]
+  ])('counts a Relay %s reply as control-plane liveness', async (_name, result) => {
+    const { session } = await authenticateSession()
+    vi.useFakeTimers()
+    try {
+      session.subscribe('nativeChat.subscribe', {}, vi.fn())
+      await vi.advanceTimersByTimeAsync(0)
+      const subscribe = fakes.sendText.mock.calls
+        .map(([payload]) => JSON.parse(payload as string) as { id: string; method: string })
+        .find(({ method }) => method === 'nativeChat.subscribe')!
+
+      session.notifyForeground()
+      fakes.linkOptions!.onText(
+        JSON.stringify({ id: subscribe.id, ok: true, streaming: true, result, _meta: {} })
+      )
+      await vi.advanceTimersByTimeAsync(8_000)
+
+      expect(session.getState()).toBe('connected')
+      expect(fakes.close).not.toHaveBeenCalled()
+    } finally {
+      session.close()
+      vi.useRealTimers()
+    }
+  })
+
+  it('does not count repeated Relay subscription frames as fresh control responses', async () => {
+    const { session } = await authenticateSession()
+    vi.useFakeTimers()
+    try {
+      session.subscribe('nativeChat.subscribe', {}, vi.fn())
+      await vi.advanceTimersByTimeAsync(0)
+      const subscribe = fakes.sendText.mock.calls
+        .map(([payload]) => JSON.parse(payload as string) as { id: string; method: string })
+        .find(({ method }) => method === 'nativeChat.subscribe')!
+      const response = JSON.stringify({
+        id: subscribe.id,
+        ok: true,
+        streaming: true,
+        result: { type: 'snapshot' },
+        _meta: {}
+      })
+      fakes.linkOptions!.onText(response)
+
+      session.notifyForeground()
+      fakes.linkOptions!.onText(response)
+      await vi.advanceTimersByTimeAsync(8_000)
+
+      expect(session.getState()).toBe('disconnected')
+      expect(fakes.close).toHaveBeenCalledOnce()
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
   it('does not let Relay terminal payload traffic mask a stalled control channel', async () => {
     const { session } = await authenticateSession()
     vi.useFakeTimers()
@@ -621,4 +678,41 @@ describe('mobile relay RPC session', () => {
       vi.useRealTimers()
     }
   })
+
+  it('retains late Relay reply evidence after an earlier probe completes', 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'
+      })
+      const firstProbe = 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: firstProbe.id, ok: true, result: {}, _meta: {} })
+      )
+      await vi.advanceTimersByTimeAsync(0)
+
+      session.notifyForeground()
+      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()
+    }
+  })
 })

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

@@ -8,21 +8,20 @@ import { MobileRelayRpcStreams } from './mobile-relay-rpc-streams'
 import { waitForMobileRelayRpcConnected } from './mobile-relay-rpc-connect-wait'
 import { MobileE2EEAuthenticationError } from './mobile-e2ee-v2-physical-channel'
 import { isRpcDeliveryUnknown, markRpcDeliveryUnknown } from './rpc-delivery-ambiguity'
+import {
+  rejectMobileRelayPendingRequests,
+  type MobileRelayPendingRequest
+} from './mobile-relay-pending-requests'
 import { RpcControlProbeFollowUp } from './rpc-control-probe-follow-up'
 import { openRpcRequestBudget, resolvePostConnectRequestTimeout } from './rpc-request-budget'
 import { isRpcResponse } from './rpc-response-shape'
 import type { RpcClient } from './rpc-client'
 import type { ConnectionState, RpcResponse } from './types'
+import { TimedOutControlRequestIndex } from './timed-out-control-request-index'
 
 const CONTROL_PROBE_TIMEOUT_MS = 8_000
 const CONTROL_PROBE_INTERVAL_MS = 20_000
 
-type PendingRequest = {
-  resolve: (response: RpcResponse) => void
-  reject: (error: Error) => void
-  timer: ReturnType<typeof setTimeout>
-}
-
 export type MobileRelayRpcSession = RpcClient & {
   getLeaseExpiresAt(): number | null
   getResumeConfirmation(): DeviceResumeConfirmed | null
@@ -40,16 +39,15 @@ export function connectMobileRelayRpcSession(args: {
   createSocket?: (url: string) => WebSocket
 }): MobileRelayRpcSession {
   const requestTimeoutMs = args.requestTimeoutMs ?? 30_000
-  const pending = new Map<string, PendingRequest>()
-  const timedOutControlRequestIds = new Set<string>()
+  const pending = new Map<string, MobileRelayPendingRequest>()
+  const timedOutControlRequestIds = new TimedOutControlRequestIndex()
   const stateListeners = new Set<(state: ConnectionState) => void>()
   let state: ConnectionState = 'connecting'
   let requestCounter = 0
   let controlResponseSequence = 0
   const controlProbeFollowUp = new RpcControlProbeFollowUp<boolean>(
     () => (!closed && state === 'connected' ? true : null),
-    probeControlPlane,
-    (hasQueuedFollowUp) => !hasQueuedFollowUp && timedOutControlRequestIds.clear()
+    probeControlPlane
   )
   let controlProbeTimer: ReturnType<typeof setInterval> | null = null
   let lastConnectedAt: number | null = null
@@ -121,8 +119,9 @@ export function connectMobileRelayRpcSession(args: {
       }
       closed = true
       stopControlProbeTimer()
+      timedOutControlRequestIds.clear()
       link.close()
-      rejectPending(new Error('Client closed'))
+      rejectMobileRelayPendingRequests(pending, new Error('Client closed'))
       streams.clear()
       publishState('disconnected')
     },
@@ -170,7 +169,7 @@ export function connectMobileRelayRpcSession(args: {
     return new Promise((resolve, reject) => {
       const timer = setTimeout(() => {
         pending.delete(id)
-        timedOutControlRequestIds.add(id)
+        timedOutControlRequestIds.remember(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)
@@ -213,7 +212,11 @@ export function connectMobileRelayRpcSession(args: {
       fail(error, new Error(error.message))
       return
     }
-    if (request || timedOutControlRequestIds.delete(value.id) || streams.isControlResponse(value)) {
+    if (
+      request ||
+      timedOutControlRequestIds.consume(value.id) ||
+      streams.isControlResponse(value)
+    ) {
       controlResponseSequence += 1
     }
     if (request) {
@@ -297,27 +300,13 @@ export function connectMobileRelayRpcSession(args: {
     }
     closed = true
     stopControlProbeTimer()
+    timedOutControlRequestIds.clear()
     failure = error
     link.close()
-    rejectPending(pendingError)
+    rejectMobileRelayPendingRequests(pending, pendingError)
     publishState(error instanceof MobileE2EEAuthenticationError ? 'auth-failed' : 'disconnected')
   }
 
-  function rejectPending(error: Error): void {
-    if (pending.size === 0) {
-      return
-    }
-    // Why: pending entries only exist after their frame reached the authenticated
-    // link (sendFrame failures delete them synchronously), so the desktop may
-    // have processed them — mark the ambiguity for callers.
-    markRpcDeliveryUnknown(error)
-    for (const request of pending.values()) {
-      clearTimeout(request.timer)
-      request.reject(error)
-    }
-    pending.clear()
-  }
-
   function nextId(): string {
     return `relay-rpc-${++requestCounter}-${Date.now()}`
   }

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

@@ -9,7 +9,6 @@ import {
 } from './rpc-client-terminal-subscription'
 import type { RpcClient } from './rpc-client'
 import type { RpcResponse, RpcSuccess } from './types'
-import { isStreamControlResponse } from './rpc-stream-response-shape'
 
 type StreamRecord = {
   method: string
@@ -23,6 +22,8 @@ type StreamRecord = {
   streamIds: Set<number>
   subscriptionId?: string
   cancelled: boolean
+  sent: boolean
+  controlResponseReceived: boolean
 }
 
 type StreamManagerOptions = {
@@ -52,14 +53,19 @@ export class MobileRelayRpcStreams {
       listener,
       onBinaryFrame: subscribeOptions?.onBinaryFrame,
       streamIds: new Set(),
-      cancelled: false
+      cancelled: false,
+      sent: false,
+      controlResponseReceived: false
     }
     this.streams.set(id, stream)
     void this.options
       .waitForConnected()
       .then(() => {
-        if (!stream.cancelled && !this.options.sendFrame({ id, method, params: stream.params })) {
-          this.fail(id, stream, 'Connection interrupted')
+        if (!stream.cancelled) {
+          stream.sent = this.options.sendFrame({ id, method, params: stream.params })
+          if (!stream.sent) {
+            this.fail(id, stream, 'Connection interrupted')
+          }
         }
       })
       .catch((error: unknown) => {
@@ -108,7 +114,12 @@ export class MobileRelayRpcStreams {
   }
 
   isControlResponse(response: RpcResponse): boolean {
-    return this.streams.has(response.id) && isStreamControlResponse(response)
+    const stream = this.streams.get(response.id)
+    if (!stream?.sent || stream.controlResponseReceived) {
+      return false
+    }
+    stream.controlResponseReceived = true
+    return true
   }
 
   handleBinary(bytes: Uint8Array): void {

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

@@ -231,6 +231,31 @@ describe('mobile rpc-client request deadline', () => {
     await request.catch(() => undefined)
   })
 
+  it('retains late reply evidence after an earlier probe completes', 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')
+    const firstProbe = sentRequest(socket, 'status.get')
+    socket.receive(`encrypted:${JSON.stringify({ id: firstProbe.id, ok: true, result: {} })}`)
+    await vi.advanceTimersByTimeAsync(0)
+
+    client.notifyForeground()
+    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]!

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

@@ -798,10 +798,13 @@ describe('mobile rpc-client connection timeout', () => {
     })
 
     it.each([
-      ['subscribed', { ok: true, streaming: true, result: { type: 'subscribed', streamId: 42 } }],
+      [
+        'lease-only subscribed',
+        { ok: true, streaming: true, result: { type: 'subscribed', streamId: null } }
+      ],
       ['ready', { ok: true, streaming: true, result: { type: 'ready', subscriptionId: 'sub-1' } }],
-      ['error', { ok: false, error: { code: 'stream_failed', message: 'stream failed' } }],
-      ['end', { ok: true, result: { type: 'end' } }]
+      ['native-chat snapshot', { ok: true, streaming: true, result: { type: 'snapshot' } }],
+      ['session-tabs snapshot', { ok: true, streaming: true, result: { type: 'snapshot' } }]
     ])('counts a subscription %s reply as control liveness', async (_name, reply) => {
       const { client, socket } = connectAuthenticated()
       client.subscribe('terminal.subscribe', { terminal: 'term-1' }, () => {})
@@ -816,6 +819,27 @@ describe('mobile rpc-client connection timeout', () => {
       client.close()
     })
 
+    it('does not count repeated subscription frames as fresh control responses', async () => {
+      const { client, socket } = connectAuthenticated()
+      client.subscribe('nativeChat.subscribe', {}, () => {})
+      const subscribe = sentRequest(socket, 'nativeChat.subscribe')
+      const snapshot = {
+        id: subscribe.id,
+        ok: true,
+        streaming: true,
+        result: { type: 'ready', subscriptionId: 'sub-1' }
+      }
+      socket.receive(`encrypted:${JSON.stringify(snapshot)}`)
+
+      client.notifyForeground()
+      socket.receive(`encrypted:${JSON.stringify(snapshot)}`)
+      await vi.advanceTimersByTimeAsync(8_000)
+
+      expect(socket.close).toHaveBeenCalled()
+      expect(client.getState()).toBe('reconnecting')
+      client.close()
+    })
+
     it('does not let terminal stream traffic mask a stalled control channel', async () => {
       const { client, socket } = connectAuthenticated()
       const events: unknown[] = []

mobile/src/transport/rpc-client.ts

@@ -38,11 +38,12 @@ import { RpcControlProbeFollowUp } from './rpc-control-probe-follow-up'
 import { openRpcRequestBudget, resolvePostConnectRequestTimeout } from './rpc-request-budget'
 import { isRpcResponse } from './rpc-response-shape'
 import {
-  isStreamControlResponse,
+  consumeFirstStreamControlResponse,
   isStreamingSubscriptionReadyResult,
   isTerminalSubscribedResult
 } from './rpc-stream-response-shape'
 import { websocketPayloadToUint8 } from './websocket-payload-bytes'
+import { TimedOutControlRequestIndex } from './timed-out-control-request-index'
 
 type PendingRequest = {
   resolve: (response: RpcResponse) => void
@@ -86,7 +87,7 @@ type StreamRequest = {
   onBinaryFrame?: (frame: BrowserScreencastFrame) => void
   subscriptionId?: string
   cancelled?: boolean
-  sent?: boolean
+  sent?: 'awaiting' | 'received'
 }
 
 export type RpcClient = {
@@ -177,8 +178,7 @@ export function connect(
   let activityProbeTimer: ReturnType<typeof setInterval> | null = null
   const activityProbeFollowUp = new RpcControlProbeFollowUp<WebSocket>(
     () => (state === 'connected' ? ws : null),
-    runActivityProbe,
-    (hasQueuedFollowUp) => !hasQueuedFollowUp && timedOutControlRequestIds.clear()
+    runActivityProbe
   )
   let intentionallyClosed = false
   // Consecutive auth rejections; tolerate up to AUTH_RETRY_BUDGET (issue #5200) before latching to avoid a needless re-pair.
@@ -196,7 +196,7 @@ export function connect(
   const serverPublicKey = publicKeyFromBase64(serverPublicKeyB64)
 
   const pending = new Map<string, PendingRequest>()
-  const timedOutControlRequestIds = new Set<string>()
+  const timedOutControlRequestIds = new TimedOutControlRequestIndex()
   const streamListeners = new Map<string, StreamRequest>()
   const terminalStreamListeners = new Map<number, StreamingListener>()
   const terminalStreamIdsByRequest = new Map<string, Set<number>>()
@@ -458,7 +458,7 @@ export function connect(
               if (
                 sendEncrypted({ id, deviceToken, method: stream.method, params: stream.params })
               ) {
-                stream.sent = true
+                stream.sent = 'awaiting'
               } else {
                 emitStreamError(stream, 'Connection interrupted')
                 removeStreamListener(id)
@@ -515,8 +515,8 @@ export function connect(
       }
       if (
         pending.has(response.id) ||
-        timedOutControlRequestIds.delete(response.id) ||
-        (streamListeners.has(response.id) && isStreamControlResponse(response))
+        timedOutControlRequestIds.consume(response.id) ||
+        consumeFirstStreamControlResponse(streamListeners.get(response.id))
       ) {
         controlResponseSequence++
       }
@@ -798,7 +798,7 @@ export function connect(
     const timeout = setTimeout(() => {
       timedOut = true
       pending.delete(id)
-      timedOutControlRequestIds.add(id)
+      timedOutControlRequestIds.remember(id)
       const controlResponded = controlResponseSequence > probeControlResponseSequence
       activityProbeFollowUp.finish(probeWs)
       if (controlResponded) {
@@ -892,7 +892,7 @@ export function connect(
 
   function markStreamsForReplay(): void {
     for (const [id, stream] of streamListeners) {
-      stream.sent = false
+      stream.sent = undefined
       resetTerminalStreamRoutingForRequest(id)
     }
   }
@@ -1056,7 +1056,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 (requestWs === ws) {
-            timedOutControlRequestIds.add(id)
+            timedOutControlRequestIds.remember(id)
           }
           runActivityProbe(requestWs, true)
         }, timeoutMs)
@@ -1108,7 +1108,7 @@ export function connect(
 
       if (state === 'connected') {
         if (sendEncrypted({ id, deviceToken, method, params })) {
-          stream.sent = true
+          stream.sent = 'awaiting'
         } else {
           emitStreamError(stream, 'Connection interrupted')
           removeStreamListener(id)

mobile/src/transport/rpc-stream-response-shape.ts

@@ -1,5 +1,3 @@
-import type { RpcResponse, RpcSuccess } from './types'
-
 export function isTerminalSubscribedResult(
   value: unknown
 ): value is { type: 'subscribed'; streamId: number } {
@@ -22,23 +20,12 @@ export function isStreamingSubscriptionReadyResult(
   )
 }
 
-export function isStreamControlResponse(response: RpcResponse): boolean {
-  if (!response.ok) {
-    return true
-  }
-  const result = (response as RpcSuccess).result
-  if (!result || typeof result !== 'object') {
+export function consumeFirstStreamControlResponse(
+  stream: { sent?: 'awaiting' | 'received' } | undefined
+): boolean {
+  if (stream?.sent !== 'awaiting') {
     return false
   }
-  const metadata = result as {
-    type?: unknown
-    subscriptionId?: unknown
-    streamId?: unknown
-  }
-  return (
-    metadata.type === 'end' ||
-    metadata.type === 'error' ||
-    typeof metadata.subscriptionId === 'string' ||
-    typeof metadata.streamId === 'number'
-  )
+  stream.sent = 'received'
+  return true
 }

mobile/src/transport/timed-out-control-request-index.test.ts

@@ -0,0 +1,27 @@
+import { describe, expect, it } from 'vitest'
+import { TimedOutControlRequestIndex } from './timed-out-control-request-index'
+
+describe('TimedOutControlRequestIndex', () => {
+  it('retains evidence until its bounded expiry', () => {
+    let now = 100
+    const index = new TimedOutControlRequestIndex(() => now, 50)
+    index.remember('late')
+
+    now = 149
+    expect(index.consume('late')).toBe(true)
+    index.remember('expired')
+    now = 200
+    expect(index.consume('expired')).toBe(false)
+  })
+
+  it('evicts the oldest evidence at the per-connection limit', () => {
+    const index = new TimedOutControlRequestIndex(() => 0, 100, 2)
+    index.remember('first')
+    index.remember('second')
+    index.remember('third')
+
+    expect(index.consume('first')).toBe(false)
+    expect(index.consume('second')).toBe(true)
+    expect(index.consume('third')).toBe(true)
+  })
+})

mobile/src/transport/timed-out-control-request-index.ts

@@ -0,0 +1,42 @@
+const DEFAULT_EXPIRY_MS = 60_000
+const DEFAULT_MAX_ENTRIES = 256
+
+export class TimedOutControlRequestIndex {
+  private readonly expiryById = new Map<string, number>()
+
+  constructor(
+    private readonly now: () => number = Date.now,
+    private readonly expiryMs = DEFAULT_EXPIRY_MS,
+    private readonly maxEntries = DEFAULT_MAX_ENTRIES
+  ) {}
+
+  remember(id: string): void {
+    this.prune()
+    while (this.expiryById.size >= this.maxEntries) {
+      const oldest = this.expiryById.keys().next().value
+      if (oldest === undefined) {
+        break
+      }
+      this.expiryById.delete(oldest)
+    }
+    this.expiryById.set(id, this.now() + this.expiryMs)
+  }
+
+  consume(id: string): boolean {
+    this.prune()
+    return this.expiryById.delete(id)
+  }
+
+  prune(): void {
+    const now = this.now()
+    for (const [id, expiry] of this.expiryById) {
+      if (expiry <= now) {
+        this.expiryById.delete(id)
+      }
+    }
+  }
+
+  clear(): void {
+    this.expiryById.clear()
+  }
+}