fix(mobile): close reconnect review races

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

@@ -511,7 +511,7 @@ describe('useHostClient', () => {
     expect(connectMock.mock.calls[2]?.[0]).toEqual(updatedHost)
     expect(oldReplacement.closeMock).toHaveBeenCalledOnce()
     expect(newReplacement.sendRequest).toHaveBeenCalledWith('status.get', undefined, {
-      timeoutMs: 15_000,
+      timeoutMs: expect.any(Number),
       budgetSpansConnect: true,
       strictDeadline: true
     })

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

@@ -1,28 +1,38 @@
 import { HostProfileSchema, type HostProfile } from './types'
-import { mutateStoredHosts } from './host-store'
+import { loadHosts } from './host-store'
+import { serializeHostProfilePublication } from './host-profile-publication'
 import * as hostListLoads from './host-list-load-sharing'
 import { saveMobileRelayHostOverlay } from './mobile-relay-host-overlay-store'
 
 export class MobileRelayUpgradeHostRemovedError extends Error {}
+export class MobileRelayUpgradeHostSupersededError extends Error {}
 
-export async function saveExistingHostRelayRouting(host: HostProfile): Promise<void> {
+export async function saveExistingHostRelayRouting(
+  host: HostProfile,
+  beforePublish?: () => Promise<void>
+): Promise<void> {
   const validated = HostProfileSchema.parse(host)
-  await mutateStoredHosts((hosts) => {
-    const existing = hosts.find(({ id }) => id === validated.id)
+  await serializeHostProfilePublication(validated.id, async () => {
+    const existing = (await loadHosts()).find(({ id }) => id === validated.id)
     if (!existing || existing.publicKeyB64 !== validated.publicKeyB64) {
       throw new MobileRelayUpgradeHostRemovedError('mobile relay upgrade host was removed')
     }
-    return hosts
+    if (existing.deviceToken !== validated.deviceToken) {
+      throw new MobileRelayUpgradeHostSupersededError('mobile relay upgrade host was re-paired')
+    }
+    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?.()
+    await saveMobileRelayHostOverlay({
+      v: 2,
+      hostId: validated.id,
+      endpoints,
+      relayHostId,
+      relay
+    })
+    hostListLoads.dropSharedHostListLoad()
   })
-  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()
 }

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

@@ -2,10 +2,12 @@ import type { RpcClient } from './rpc-client'
 import { isRpcDeliveryUnknown } from './rpc-delivery-ambiguity'
 import { isLogicalClientCutoverError } from './stable-logical-rpc-client'
 
-const FORCE_RECONNECT_HEALTH_TIMEOUT_MS = 15_000
+export const FORCE_RECONNECT_TIMEOUT_MS = 15_000
 
-export async function verifyForceReconnectRpcHealth(client: RpcClient): Promise<void> {
-  const deadline = Date.now() + FORCE_RECONNECT_HEALTH_TIMEOUT_MS
+export async function verifyForceReconnectRpcHealth(
+  client: RpcClient,
+  deadline = Date.now() + FORCE_RECONNECT_TIMEOUT_MS
+): Promise<void> {
   let lastError: unknown = null
   for (;;) {
     const timeoutMs = deadline - Date.now()

mobile/src/transport/host-force-reconnect.test.ts

@@ -0,0 +1,30 @@
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { HostForceReconnectCoordinator } from './host-force-reconnect'
+
+describe('HostForceReconnectCoordinator', () => {
+  afterEach(() => vi.useRealTimers())
+
+  it('bounds a first reconnect when its same-profile open never settles', async () => {
+    vi.useFakeTimers()
+    const cancelPendingOpen = vi.fn()
+    const coordinator = new HostForceReconnectCoordinator()
+    const reconnect = coordinator.run({
+      hostId: 'host-1',
+      profileVersion: 0,
+      getEntry: () => undefined,
+      getListenerCount: () => 1,
+      removeEntry: vi.fn(),
+      cancelPendingOpen,
+      openReplacement: () => new Promise<never>(() => {})
+    })
+    const outcome = reconnect.catch((error: Error) => error.message)
+
+    expect(cancelPendingOpen).toHaveBeenCalledOnce()
+    await vi.advanceTimersByTimeAsync(14_999)
+    expect(cancelPendingOpen).toHaveBeenCalledOnce()
+    await vi.advanceTimersByTimeAsync(1)
+
+    await expect(outcome).resolves.toBe('Force Reconnect timed out')
+    expect(cancelPendingOpen).toHaveBeenCalledTimes(2)
+  })
+})

mobile/src/transport/host-force-reconnect.ts

@@ -1,5 +1,8 @@
 import type { RpcClient } from './rpc-client'
-import { verifyForceReconnectRpcHealth } from './force-reconnect-rpc-health'
+import {
+  FORCE_RECONNECT_TIMEOUT_MS,
+  verifyForceReconnectRpcHealth
+} from './force-reconnect-rpc-health'
 
 export type HostReconnectEntry = {
   client: RpcClient
@@ -47,9 +50,14 @@ export class HostForceReconnectCoordinator {
     if (pending) {
       generation += 1
       this.generations.set(operation.hostId, generation)
-      operation.cancelPendingOpen()
     }
-    const reconnect = this.replaceAndVerify(operation, generation)
+    // Why: a first reconnect must not join an ordinary open whose Keychain read may never settle.
+    operation.cancelPendingOpen()
+    const reconnect = this.replaceAndVerify(
+      operation,
+      generation,
+      Date.now() + FORCE_RECONNECT_TIMEOUT_MS
+    )
     this.pendingByHost.set(operation.hostId, {
       profileVersion: operation.profileVersion,
       generation,
@@ -66,7 +74,8 @@ export class HostForceReconnectCoordinator {
 
   private async replaceAndVerify(
     operation: HostReconnectOperation,
-    generation: number
+    generation: number,
+    deadline: number
   ): Promise<void> {
     if (this.wasCancelled(operation.hostId, generation)) {
       return
@@ -78,7 +87,7 @@ export class HostForceReconnectCoordinator {
       entry.client.close()
       operation.removeEntry()
     }
-    const fresh = await operation.openReplacement()
+    const fresh = await this.openBeforeDeadline(operation, deadline)
     if (this.wasCancelled(operation.hostId, generation)) {
       return
     }
@@ -87,7 +96,7 @@ export class HostForceReconnectCoordinator {
     }
     fresh.refCount = savedRefCount
     try {
-      await verifyForceReconnectRpcHealth(fresh.client)
+      await verifyForceReconnectRpcHealth(fresh.client, deadline)
     } catch (error) {
       if (!this.wasCancelled(operation.hostId, generation)) {
         throw error
@@ -95,6 +104,32 @@ export class HostForceReconnectCoordinator {
     }
   }
 
+  private async openBeforeDeadline(
+    operation: HostReconnectOperation,
+    deadline: number
+  ): Promise<HostReconnectEntry | null> {
+    const timeoutMs = deadline - Date.now()
+    if (timeoutMs <= 0) {
+      throw new Error('Force Reconnect timed out')
+    }
+    let timer: ReturnType<typeof setTimeout> | null = null
+    try {
+      return await Promise.race([
+        operation.openReplacement(),
+        new Promise<never>((_resolve, reject) => {
+          timer = setTimeout(() => {
+            operation.cancelPendingOpen()
+            reject(new Error('Force Reconnect timed out'))
+          }, timeoutMs)
+        })
+      ])
+    } finally {
+      if (timer) {
+        clearTimeout(timer)
+      }
+    }
+  }
+
   private generation(hostId: string): number {
     return this.generations.get(hostId) ?? 0
   }

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

@@ -0,0 +1,33 @@
+import type { HostProfile } from './types'
+
+const pendingByHost = new Map<string, Promise<void>>()
+
+export function serializeHostProfilePublication<T>(
+  hostId: string,
+  publish: () => Promise<T>
+): Promise<T> {
+  const previous = pendingByHost.get(hostId) ?? Promise.resolve()
+  const result = previous.then(publish)
+  const settled = result.then(
+    () => undefined,
+    () => undefined
+  )
+  pendingByHost.set(hostId, settled)
+  void settled.then(() => {
+    if (pendingByHost.get(hostId) === settled) {
+      pendingByHost.delete(hostId)
+    }
+  })
+  return result
+}
+
+export function publishHostProfileTransaction(
+  host: HostProfile,
+  beforeHostSave: (() => Promise<void>) | null,
+  saveHost: (host: HostProfile) => Promise<void>
+): Promise<void> {
+  return serializeHostProfilePublication(host.id, async () => {
+    await beforeHostSave?.()
+    await saveHost(host)
+  })
+}

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

@@ -45,9 +45,11 @@ import {
 } from './host-store'
 import {
   MobileRelayUpgradeHostRemovedError,
+  MobileRelayUpgradeHostSupersededError,
   saveExistingHostRelayRouting
 } from './existing-host-relay-routing'
 import { resetMobileRelayHostOverlayStoreForTests } from './mobile-relay-host-overlay-store'
+import { publishHostProfileTransaction } from './host-profile-publication'
 
 const HOSTS_STORAGE_KEY = 'orca:hosts'
 const OVERLAY_STORAGE_KEY = 'orca:mobile-relay:host-overlays:v2'
@@ -286,6 +288,74 @@ describe('host-store list mutations', () => {
     )
   })
 
+  it('rejects relay routing from a lifecycle superseded by same-host re-pairing', async () => {
+    await saveHost({ ...HOST_ONE, deviceToken: 'replacement-token' })
+    const relayHostId = 'AbCdEf0123_-xyZ9'
+
+    await expect(
+      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
+        }
+      })
+    ).rejects.toBeInstanceOf(MobileRelayUpgradeHostSupersededError)
+
+    expect(JSON.parse(storedOverlayRaw ?? '[]')).toEqual([])
+  })
+
+  it('orders an in-flight relay publication before same-host re-pair persistence', async () => {
+    let releaseOldToken: ((token: string | null) => void) | null = null
+    secureStoreMock.getItemAsync.mockImplementationOnce(
+      () =>
+        new Promise<string | null>((resolve) => {
+          releaseOldToken = resolve
+        })
+    )
+    const relayHostId = 'AbCdEf0123_-xyZ9'
+    const stalePublication = 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
+      }
+    })
+    await vi.waitFor(() => expect(secureStoreMock.getItemAsync).toHaveBeenCalledOnce())
+
+    const replacement = publishHostProfileTransaction(
+      { ...HOST_ONE, deviceToken: 'replacement-token' },
+      null,
+      saveHost
+    )
+    await Promise.resolve()
+    expect(secureStoreMock.setItemAsync).not.toHaveBeenCalled()
+
+    releaseOldToken?.('token-1')
+    await Promise.all([stalePublication, replacement])
+    expect(JSON.parse(storedOverlayRaw ?? '[]')).toEqual([])
+  })
+
   it('awaits cleanup scheduling after metadata commit', async () => {
     let resolveSchedule: (() => void) | null = null
     scheduleCleanupMock.mockReturnValue(

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

@@ -20,14 +20,19 @@ export function relayWebSocketUrl(relay: { cellUrl: string; relayHostId: string
 export async function persistRelayHost(
   host: HostProfile,
   relay: MobileRelayEndpoint,
-  saveHost: (host: HostProfile) => Promise<void>
+  saveHost: (host: HostProfile, beforePublish?: () => Promise<void>) => Promise<void>,
+  beforePublish?: () => Promise<void>
 ): Promise<HostProfile> {
   const endpoints = [
     ...(host.endpoints ?? [{ id: 'direct-primary', kind: 'lan' as const, url: host.endpoint }])
   ].filter(({ kind }) => kind !== 'relay')
   endpoints.push({ id: 'relay-primary', kind: 'relay', url: relayWebSocketUrl(relay) })
   const updated = { ...host, endpoints, relayHostId: relay.relayHostId, relay }
-  await saveHost(updated)
+  if (beforePublish) {
+    await saveHost(updated, beforePublish)
+  } else {
+    await saveHost(updated)
+  }
   return updated
 }
 

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

@@ -1,6 +1,9 @@
 import { describe, expect, it, vi } from 'vitest'
 import type { MobileRelayEndpoint } from '../../../src/shared/mobile-relay-credential-contract'
-import { MobileRelayUpgradeHostRemovedError } from './existing-host-relay-routing'
+import {
+  MobileRelayUpgradeHostRemovedError,
+  MobileRelayUpgradeHostSupersededError
+} from './existing-host-relay-routing'
 import {
   createMobileRelayDirectUpgradeJournal,
   type MobileRelayDirectUpgradeJournal
@@ -64,7 +67,9 @@ function dependencies(journal: MobileRelayDirectUpgradeJournal | null = null) {
     }),
     writeBundle: vi.fn(async () => {}),
     deleteBundle: vi.fn(async () => {}),
-    saveHost: vi.fn(async () => {}),
+    saveHost: vi.fn(async (_host: HostProfile, beforePublish?: () => Promise<void>) => {
+      await beforePublish?.()
+    }),
     randomBytes: (length: number) => new Uint8Array(length).fill(7)
   }
 }
@@ -113,11 +118,37 @@ describe('existing direct pairing relay upgrade', () => {
       reqId: journal!.reqId,
       newResumeTokenHash: journal!.pendingResumeTokenHash
     })
-    expect(deps.writeBundle).toHaveBeenCalledBefore(deps.saveHost)
+    expect(deps.saveHost).toHaveBeenCalledWith(expect.any(Object), expect.any(Function))
+    expect(deps.writeBundle).toHaveBeenCalledOnce()
     expect(result?.host.relay).toEqual(relay)
     expect(deps.clearJournal).toHaveBeenCalledWith(host.id)
   })
 
+  it('does not delete a replacement pairing bundle when the old upgrade is superseded', async () => {
+    const journal = createMobileRelayDirectUpgradeJournal(host.id, (length) =>
+      new Uint8Array(length).fill(6)
+    )
+    const committed = installed(journal)
+    const deps = dependencies(journal)
+    deps.saveHost.mockRejectedValue(
+      new MobileRelayUpgradeHostSupersededError('mobile relay upgrade host was re-paired')
+    )
+    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(MobileRelayUpgradeHostSupersededError)
+    expect(deps.writeBundle).not.toHaveBeenCalled()
+    expect(deps.deleteBundle).not.toHaveBeenCalled()
+    expect(deps.clearJournal).toHaveBeenCalledWith(host.id)
+  })
+
   it('recovers an already committed install without authorizing a second one', async () => {
     const journal = createMobileRelayDirectUpgradeJournal(host.id, (length) =>
       new Uint8Array(length).fill(3)

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

@@ -7,6 +7,7 @@ import {
 } from '../../../src/shared/mobile-relay-credential-contract'
 import {
   MobileRelayUpgradeHostRemovedError,
+  MobileRelayUpgradeHostSupersededError,
   saveExistingHostRelayRouting
 } from './existing-host-relay-routing'
 import { persistRelayHost } from './mobile-endpoint-supervisor-support'
@@ -118,15 +119,17 @@ async function publishCommitted(
       expiresAt: installed.resumeExpiresAt
     }
   })
-  // Why: the overlay must never advertise relay without its matching credential.
-  await dependencies.writeBundle(bundle)
   let updatedHost: HostProfile
   try {
-    updatedHost = await persistRelayHost(host, endpoints.relay, dependencies.saveHost)
+    updatedHost = await persistRelayHost(host, endpoints.relay, dependencies.saveHost, () =>
+      dependencies.writeBundle(bundle)
+    )
   } catch (error) {
     if (error instanceof MobileRelayUpgradeHostRemovedError) {
       await dependencies.deleteBundle(host.id)
       await dependencies.clearJournal(host.id)
+    } else if (error instanceof MobileRelayUpgradeHostSupersededError) {
+      await dependencies.clearJournal(host.id)
     }
     throw error
   }

mobile/src/transport/mobile-relay-pairing-recovery.ts

@@ -26,6 +26,7 @@ import {
 } from './mobile-relay-physical-client'
 import { createRecoveringPairingRelayCandidate } from './pairing-relay-candidate'
 import type { HostProfile, RpcResponse } from './types'
+import { publishHostProfileTransaction } from './host-profile-publication'
 
 export type MobileRelayPairingRecoveryResult = 'none' | 'recovered' | 'deferred'
 
@@ -243,10 +244,15 @@ async function publishCommitted(
   if (journal.metadata.authorizationMode !== installed.authorizationMode) {
     await dependencies.updateJournal(journal.metadata.journalId, () => reconciledJournal.metadata)
   }
-  await dependencies.writeCredentialBundle(
-    promotePairingJournalCredential({ journal: reconciledJournal, installed })
+  const host = relayHost(reconciledJournal, endpoints.relay)
+  await publishHostProfileTransaction(
+    host,
+    () =>
+      dependencies.writeCredentialBundle(
+        promotePairingJournalCredential({ journal: reconciledJournal, installed })
+      ),
+    dependencies.saveHost
   )
-  await dependencies.saveHost(relayHost(reconciledJournal, endpoints.relay))
   await dependencies.clearJournal(journal.metadata.journalId)
 }
 

mobile/src/transport/mobile-relay-rpc-connect-wait.ts

@@ -0,0 +1,32 @@
+import type { ConnectionState } from './types'
+
+export function waitForMobileRelayRpcConnected(args: {
+  getState: () => ConnectionState
+  subscribe: (listener: (state: ConnectionState) => void) => () => void
+  timeoutMs: number
+}): Promise<void> {
+  const state = args.getState()
+  if (state === 'connected') {
+    return Promise.resolve()
+  }
+  if (state === 'disconnected' || state === 'auth-failed') {
+    return Promise.reject(new Error(`relay session ${state}`))
+  }
+  return new Promise((resolve, reject) => {
+    const timer = setTimeout(() => {
+      unsubscribe()
+      reject(new Error('relay session connection timed out'))
+    }, args.timeoutMs)
+    const unsubscribe = args.subscribe((next) => {
+      if (next === 'connected') {
+        clearTimeout(timer)
+        unsubscribe()
+        resolve()
+      } else if (next === 'disconnected' || next === 'auth-failed') {
+        clearTimeout(timer)
+        unsubscribe()
+        reject(new Error(`relay session ${next}`))
+      }
+    })
+  })
+}

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

@@ -4,6 +4,8 @@ import {
   encodeBrowserScreencastFrame
 } from '../../../src/shared/browser-screencast-protocol'
 import { encodeTerminalStreamFrame, TerminalStreamOpcode } from './terminal-stream-protocol'
+import { verifyForceReconnectRpcHealth } from './force-reconnect-rpc-health'
+import { MobileE2EEAuthenticationError } from './mobile-e2ee-v2-physical-channel'
 import { isRpcDeliveryUnknown } from './rpc-delivery-ambiguity'
 
 const fakes = vi.hoisted(() => ({
@@ -335,6 +337,50 @@ describe('mobile relay RPC session', () => {
     }
   })
 
+  it('marks a Relay session auth-failed when its periodic probe is unauthorized', async () => {
+    vi.useFakeTimers()
+    try {
+      const { session } = await authenticateSession()
+
+      await vi.advanceTimersByTimeAsync(20_000)
+      const probe = fakes.sendText.mock.calls
+        .map(([payload]) => JSON.parse(payload as string) as { id: string; method: string })
+        .findLast(({ method }) => method === 'status.get')!
+      fakes.linkOptions!.onText(
+        JSON.stringify({
+          id: probe.id,
+          ok: false,
+          error: { code: 'unauthorized', message: 'Invalid device token' },
+          _meta: {}
+        })
+      )
+
+      expect(session.getFailure()).toBeInstanceOf(MobileE2EEAuthenticationError)
+      expect(session.getState()).toBe('auth-failed')
+      expect(fakes.close).toHaveBeenCalledOnce()
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('rejects Force Reconnect health when Relay authorization was revoked', async () => {
+    const { session } = await authenticateSession()
+    const verification = verifyForceReconnectRpcHealth(session)
+    await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledOnce())
+    const probe = JSON.parse(fakes.sendText.mock.calls[0]![0] as string) as { id: string }
+    fakes.linkOptions!.onText(
+      JSON.stringify({
+        id: probe.id,
+        ok: false,
+        error: { code: 'unauthorized', message: 'Invalid device token' },
+        _meta: {}
+      })
+    )
+
+    await expect(verification).rejects.toBeInstanceOf(MobileE2EEAuthenticationError)
+    expect(session.getState()).toBe('auth-failed')
+  })
+
   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

@@ -5,6 +5,7 @@ import {
 } from '../../../src/shared/mobile-relay-credential-contract'
 import { MobileRelayE2eeLink } from './mobile-relay-e2ee-link'
 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 { openRpcRequestBudget, resolvePostConnectRequestTimeout } from './rpc-request-budget'
@@ -192,6 +193,10 @@ export function connectMobileRelayRpcSession(args: {
     if (!isRpcResponse(value)) {
       return
     }
+    if (!value.ok && value.error.code === 'unauthorized') {
+      fail(new MobileE2EEAuthenticationError())
+      return
+    }
     const request = pending.get(value.id)
     if (request || timedOutControlRequestIds.delete(value.id)) {
       controlResponseSequence += 1
@@ -253,30 +258,10 @@ export function connectMobileRelayRpcSession(args: {
   }
 
   function waitForConnected(timeoutMs = requestTimeoutMs): Promise<void> {
-    if (state === 'connected') {
-      return Promise.resolve()
-    }
-    return new Promise((resolve, reject) => {
-      let timer: ReturnType<typeof setTimeout> | null = null
-      const unsubscribe = client.onStateChange((next) => {
-        if (next === 'connected') {
-          finish()
-          resolve()
-        } else if (next === 'disconnected' || next === 'auth-failed') {
-          finish()
-          reject(new Error(`relay session ${next}`))
-        }
-      })
-      timer = setTimeout(() => {
-        finish()
-        reject(new Error('relay session connection timed out'))
-      }, timeoutMs)
-      function finish(): void {
-        if (timer) {
-          clearTimeout(timer)
-        }
-        unsubscribe()
-      }
+    return waitForMobileRelayRpcConnected({
+      getState: () => state,
+      subscribe: (listener) => client.onStateChange(listener),
+      timeoutMs
     })
   }
 
@@ -309,7 +294,9 @@ export function connectMobileRelayRpcSession(args: {
     // 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)
+    if (!(error instanceof MobileE2EEAuthenticationError)) {
+      markRpcDeliveryUnknown(error)
+    }
     for (const request of pending.values()) {
       clearTimeout(request.timer)
       request.reject(error)

mobile/src/transport/pre-profile-pairing-coordinator.ts

@@ -21,6 +21,7 @@ import {
   promotePairingJournalCredential,
   writeMobileRelayCredentialBundle
 } from './mobile-relay-credential-bundle'
+import { publishHostProfileTransaction } from './host-profile-publication'
 import {
   connectMobileRelayForPairing,
   type PairingCandidateClient
@@ -28,6 +29,7 @@ import {
 import { racePairingCandidates, type PairingCandidate } from './pairing-candidate-race'
 import { resolvePairingInviteThroughDirector } from './mobile-relay-invite-director'
 import { createRecoveringPairingRelayCandidate } from './pairing-relay-candidate'
+import { relayWebSocketUrl } from './mobile-endpoint-supervisor-support'
 
 export type PreProfilePairingAttempt = {
   readonly result: Promise<{ hostId: string }>
@@ -190,7 +192,11 @@ async function runPairing(
   assertActive(isDisposed)
 
   if (!journal) {
-    await dependencies.saveHost(baseHost(offer, hostId, hostName, now))
+    await publishHostProfileTransaction(
+      baseHost(offer, hostId, hostName, now),
+      null,
+      dependencies.saveHost
+    )
     return { hostId }
   }
 
@@ -211,7 +217,11 @@ async function runPairing(
     if (winner.path !== 'direct') {
       throw new Error('relay pairing RPC unavailable after relay path authentication')
     }
-    await dependencies.saveHost(baseHost(offer, hostId, hostName, now))
+    await publishHostProfileTransaction(
+      baseHost(offer, hostId, hostName, now),
+      null,
+      dependencies.saveHost
+    )
     await dependencies.clearJournal(journal.metadata.journalId)
     return { hostId }
   }
@@ -228,9 +238,17 @@ async function runPairing(
     throw new Error('desktop returned no relay endpoint after credential install')
   }
   assertActive(isDisposed)
-  await dependencies.writeCredentialBundle(promotePairingJournalCredential({ journal, installed }))
-  await dependencies.saveHost(relayHost(journal, endpoints.relay))
-  await dependencies.clearJournal(journal.metadata.journalId)
+  const committedJournal = journal
+  const host = relayHost(committedJournal, endpoints.relay)
+  await publishHostProfileTransaction(
+    host,
+    () =>
+      dependencies.writeCredentialBundle(
+        promotePairingJournalCredential({ journal: committedJournal, installed })
+      ),
+    dependencies.saveHost
+  )
+  await dependencies.clearJournal(committedJournal.metadata.journalId)
   return { hostId }
 }
 
@@ -264,13 +282,6 @@ function relayHost(journal: MobileRelayPairingJournal, relay: MobileRelayEndpoin
   }
 }
 
-function relayWebSocketUrl(relay: MobileRelayEndpoint): string {
-  const url = new URL(relay.cellUrl)
-  url.protocol = 'wss:'
-  url.pathname = `/v1/connect/${encodeURIComponent(relay.relayHostId)}`
-  return url.toString()
-}
-
 function requireSuccess(response: RpcResponse): unknown {
   if (!response.ok) {
     throw new Error(`${response.error.code}: ${response.error.message}`)