fix(mobile): self-heal host opens and harden session liveness (#14333)

2026-09-22 00:02:31 +00:00 · 2026-08-13 17:15:03 -07:00
parent 31fcf8e2f0
commit 54aa22b2df
41 changed files with 2127 additions and 500 deletions
@@ -785,6 +785,143 @@
      ],
      "demotionRule": "Keep experimental or demote if focused tests flake, HOST_OFFLINE can park indefinitely, direct probes serialize configured endpoints, loser cleanup leaks clients, or physical-device radio cost exceeds the measured budget."
    },
+    {
+      "id": "mobile-transport.lifecycle-liveness",
+      "title": "Wanted mobile hosts self-heal without false physical-session death",
+      "maturity": "experimental",
+      "protection": "partial",
+      "owner": "mobile-runtime",
+      "layer": "mobile-provider-and-physical-session-contract",
+      "surfaces": [
+        "mobile host acquisition",
+        "Keychain and Keystore catalog reads",
+        "direct LAN and Tailscale RPC",
+        "cloud relay RPC",
+        "foreground and network revival"
+      ],
+      "platforms": ["ios", "android", "macos", "linux", "windows"],
+      "providers": ["lan", "tailscale", "cloud-relay"],
+      "coveredPlatforms": ["macos"],
+      "coveredProviders": ["lan", "tailscale", "cloud-relay"],
+      "coverageNotes": "Deterministic React/provider tests cover catalog, missing-host, construction, cancellation, release, forget, refresh, backoff, and stale-generation outcomes. Direct sessions cover idle probing and three fair misses. Relay sessions cover zero idle polling, rate-limited foreground probes, two fair misses, authenticated activity before semantic decoding, delivery ambiguity, scheduler stalls, and write failure. Physical iOS/Android sleep, carrier/Tailscale loss, and production relay paths remain live-test gaps.",
+      "motivatingLinks": [
+        "https://linear.app/stably/issue/STA-4153/mobile-app-disconnects-frequently"
+      ],
+      "invariant": "A provider-wanted host without a logical client retries every current-generation pre-client failure until success or explicit cancellation, while release, disconnect, forget, refresh, and stale async continuations preserve their distinct resurrection policies. Direct sessions probe after authenticated silence and terminate after three fair misses. Relay sessions emit no periodic idle or demand-driven probes, use rate-limited foreground probes, and terminate after two fair misses. Authenticated traffic, scheduler stalls, and replacement identities cannot manufacture death.",
+      "oracle": "Inject rejected and never-settling catalog reads, missing profiles, synchronous construction failure, release, refresh, forget, revival, and stale completion under fake time. Require bounded 1s through 60s retry, exact current-generation publication, and no zero-owner or forgotten-client resurrection. Inject direct and relay authenticated text/binary traffic, silent probe windows, stalled clocks, replacement identities, and failed writes. Require direct three-miss tolerance; Relay zero idle and demand-driven liveness traffic, one foreground probe sequence, two fair 4s misses, a 10s voluntary rate limit, no old-session network-change probe, and delivery-unknown mutations without replay.",
+      "commands": [
+        "pnpm --dir mobile exec vitest run --root .. mobile/src/transport/client-context.test.ts mobile/src/transport/host-open-recovery.test.tsx mobile/src/transport/host-open-retry-scheduler.test.ts mobile/src/transport/rpc-session-liveness-watchdog.test.ts mobile/src/transport/rpc-session-liveness-integration.test.ts mobile/src/transport/mobile-relay-rpc-session.test.ts mobile/src/transport/mobile-relay-rpc-session-liveness.test.ts mobile/src/transport/mobile-endpoint-nudge-router.test.ts mobile/src/transport/mobile-relay-e2ee-link.test.ts mobile/src/transport/stable-logical-rpc-client.test.ts"
+      ],
+      "testFiles": [
+        "mobile/src/transport/client-context.test.ts",
+        "mobile/src/transport/host-open-recovery.test.tsx",
+        "mobile/src/transport/host-open-retry-scheduler.test.ts",
+        "mobile/src/transport/rpc-session-liveness-watchdog.test.ts",
+        "mobile/src/transport/rpc-session-liveness-integration.test.ts",
+        "mobile/src/transport/mobile-relay-rpc-session.test.ts",
+        "mobile/src/transport/mobile-relay-rpc-session-liveness.test.ts",
+        "mobile/src/transport/mobile-endpoint-nudge-router.test.ts",
+        "mobile/src/transport/mobile-relay-e2ee-link.test.ts",
+        "mobile/src/transport/stable-logical-rpc-client.test.ts"
+      ],
+      "assertionRefs": [
+        {
+          "file": "mobile/src/transport/client-context.test.ts",
+          "assertions": ["a retired close-on-release owner closes an ownerless manual reconnect"]
+        },
+        {
+          "file": "mobile/src/transport/host-open-recovery.test.tsx",
+          "assertions": [
+            "catalog, missing-host, and construction failures retry without remounting",
+            "final release and forget prevent later client publication",
+            "refresh bypasses a never-settling stale open and stale failure cannot clobber its replacement",
+            "revival expedites recovery without resetting the failure tier",
+            "a retired owner's release cannot cancel a newer owner's open",
+            "explicit reconnect demand survives a pre-client failure"
+          ]
+        },
+        {
+          "file": "mobile/src/transport/rpc-session-liveness-watchdog.test.ts",
+          "assertions": [
+            "authenticated traffic resets the idle and missed-probe clocks",
+            "two fair misses are tolerated and the third terminates exactly once",
+            "scheduler stalls and replacement identities cannot terminate the current session",
+            "continuous authenticated traffic does not churn idle timers per frame"
+          ]
+        },
+        {
+          "file": "mobile/src/transport/rpc-session-liveness-integration.test.ts",
+          "assertions": [
+            "direct handshake and probe write exceptions recover without escaping",
+            "one silent foreground window preserves the connected direct socket",
+            "authenticated terminal binary output restores direct-session liveness",
+            "a replay write failure retains every queued stream for the replacement session",
+            "liveness probe replies cannot consume the authentication retry budget"
+          ]
+        },
+        {
+          "file": "mobile/src/transport/mobile-relay-rpc-session.test.ts",
+          "assertions": [
+            "written Relay requests retain delivery-unknown failures",
+            "resume confirmation and stream routing remain authenticated"
+          ]
+        },
+        {
+          "file": "mobile/src/transport/mobile-relay-rpc-session-liveness.test.ts",
+          "assertions": [
+            "an idle Relay emits zero periodic traffic",
+            "RPCs and subscriptions after prolonged silence emit no liveness probe",
+            "foreground Relay liveness uses two fair 4-second misses and a 10-second sequence rate limit",
+            "network change does not probe the replaced Relay",
+            "a failed foreground probe write disconnects immediately"
+          ]
+        },
+        {
+          "file": "mobile/src/transport/stable-logical-rpc-client.test.ts",
+          "assertions": ["a delivery-unknown mutation is not replayed onto a replacement Relay"]
+        }
+      ],
+      "evidenceRuns": [
+        {
+          "date": "2026-08-13",
+          "runner": "local",
+          "platform": "macos",
+          "command": "pnpm --dir mobile exec vitest run --root .. mobile/src/transport/client-context.test.ts mobile/src/transport/host-open-recovery.test.tsx mobile/src/transport/host-open-retry-scheduler.test.ts mobile/src/transport/rpc-session-liveness-watchdog.test.ts mobile/src/transport/rpc-session-liveness-integration.test.ts mobile/src/transport/mobile-relay-rpc-session.test.ts mobile/src/transport/mobile-relay-rpc-session-liveness.test.ts mobile/src/transport/mobile-endpoint-nudge-router.test.ts mobile/src/transport/mobile-relay-e2ee-link.test.ts mobile/src/transport/stable-logical-rpc-client.test.ts",
+          "result": "passed",
+          "durationSeconds": 0.74,
+          "summary": "Ten focused files passed 66 deterministic provider lifecycle, direct/Relay liveness, traffic, and delivery-ambiguity tests."
+        }
+      ],
+      "runtimeBudget": {
+        "p95Seconds": 5,
+        "scope": "focused mobile provider and transport unit tests"
+      },
+      "flakeHistory": {
+        "status": "unknown",
+        "evidence": "One deterministic local focused run exists; CI and soak history are not yet available."
+      },
+      "redGreenEvidence": {
+        "status": "complete",
+        "evidence": "Current main failed six deterministic assertions: terminal catalog failure, zero-owner publication, stale failure clobber, initial direct handshake write throw, one-window direct death, and foreground probe write throw. The candidate passes the same assertions and adds relay parity, backoff, cancellation, and scheduler-fairness coverage."
+      },
+      "performanceBudget": {
+        "required": true,
+        "evidence": "Each wanted clientless host owns at most one open and one retry timer with a 60-second capped tail. Direct sessions own one idle or probe timer. Idle Relay sessions own no liveness timer and emit no periodic or demand-driven liveness traffic; foreground sequences start at most once per 10 seconds. No subprocess, wire opcode, or server-side work is introduced."
+      },
+      "promotionCriteria": [
+        "Collect 100 consecutive CI passes or 14 days of soak history.",
+        "Capture the existing final-timeout log immediately before an affected-device drop.",
+        "Run physical iOS and Android direct LAN, Tailscale, relay, background/resume, and network-handoff journeys.",
+        "Measure reconnect radio and battery impact during the 60-second catalog retry tail, direct 44-second idle liveness budget, and Relay foreground probes."
+      ],
+      "knownGaps": [
+        "No physical iOS or Android device was exercised.",
+        "No affected-device log yet proves the old one-window watchdog caused the reported live-session drop.",
+        "Carrier NAT, production Tailscale, production relay cells, and real scheduler suspension remain unmeasured.",
+        "Persistent pre-client failure metadata is not yet exposed in Home UI."
+      ],
+      "demotionRule": "Keep experimental or demote if a wanted host can park without retry, stale work can publish or notify, disconnect or forget resurrects a host, direct closes before three fair misses, Relay polls while idle or on demand, survives two fair foreground misses, authenticated traffic fails to restore health, timers leak across identity replacement, mutations replay after ambiguous delivery, or physical-device radio cost exceeds the measured budget."
+    },
    {
      "id": "desktop-relay.assignment-backpressure",
      "title": "Desktop relay drain recovery cannot amplify a director outage",
@@ -25,7 +25,7 @@ import { loadHosts, updateLastConnected } from '../../../src/transport/host-stor
 import { removeHostAndCloseClient } from '../../../src/transport/host-removal-lifecycle'
 import {
  useHostClient,
-  useCloseHost,
+  useForgetHostClient,
  useForceReconnect
 } from '../../../src/transport/client-context'
 import { useWorktreeResync } from '../../../src/transport/use-worktree-resync'
@@ -148,7 +148,7 @@ export function HostScreen({
  const repoMetadataFetchedAtRef = useRef(0)
  const newWorktreeModalRef = useRef<{ open: () => void }>(null)
  const newWorktreeModalVisibleRef = useRef(false)
-  const closeHostClient = useCloseHost()
+  const forgetHostClient = useForgetHostClient()
  const forceReconnectHost = useForceReconnect()
  const [worktrees, setWorktrees] = useState<Worktree[]>(initialCache ?? [])
  const [worktreesLoaded, setWorktreesLoaded] = useState(initialCache != null)
@@ -636,14 +636,14 @@ export function HostScreen({
      return
    }
    try {
-      await removeHostAndCloseClient(hostId, closeHostClient)
+      await removeHostAndCloseClient(hostId, forgetHostClient)
      leaveHost()
    } catch {
      // Why: removal can fail while still paired; re-open confirm (ConfirmModal closes on confirm).
      setConfirmRemoveHost(true)
      Alert.alert('Could not remove host', 'Please try again.')
    }
-  }, [hostId, leaveHost, closeHostClient])
+  }, [hostId, leaveHost, forgetHostClient])

  const navigateFromHostList = useCallback(
    (target: string) => {
@@ -25,7 +25,12 @@ import type { HomeWorktreeSummary, HostWorktreeInfo } from '../src/worktree/home
 import type { RpcClient } from '../src/transport/rpc-client'
 import { createHostConnectRefetchGate } from '../src/transport/host-connect-refetch-gate'
 import { sendSingleFlightRequest } from '../src/transport/request-single-flight'
-import { useCloseHost, useForceReconnect, usePrimeHosts } from '../src/transport/client-context'
+import {
+  useDisconnectHostClient,
+  useForceReconnect,
+  useForgetHostClient,
+  usePrimeHosts
+} from '../src/transport/client-context'
 import { useAllHostClients } from '../src/transport/use-all-host-clients'
 import {
  resolveHomeHostConnectionState,
@@ -254,7 +259,8 @@ export default function HomeScreen() {
    () => Object.fromEntries(allClients.map(({ hostId, path }) => [hostId, path])),
    [allClients]
  )
-  const closeHostClient = useCloseHost()
+  const disconnectHostClient = useDisconnectHostClient()
+  const forgetHostClient = useForgetHostClient()
  const forceReconnectHost = useForceReconnect()
  const primeHosts = usePrimeHosts()
  // Why: prime the cache with loaded HostProfiles to avoid a second serialized Keychain pass (multi-second connect latency) on cold start.
@@ -643,7 +649,7 @@ export default function HomeScreen() {
    }
    const hostToRemove = confirmRemove
    try {
-      await removeHostAndCloseClient(hostToRemove.id, closeHostClient)
+      await removeHostAndCloseClient(hostToRemove.id, forgetHostClient)
      setConfirmRemove(null)
      setHostCatalog(await loadHostCatalog())
    } catch {
@@ -949,7 +955,7 @@ export default function HomeScreen() {
            : false,
          onDismiss: () => setActionTarget(null),
          onReconnect: (hostId) => void forceReconnectHost(hostId),
-          onDisconnect: closeHostClient,
+          onDisconnect: disconnectHostClient,
          onEdit: openMobileHostEdit,
          onRemove: (host) => setConfirmRemove(host)
        })}
@@ -9,7 +9,7 @@ import {
  type PreProfilePairingAttempt
 } from '../src/transport/pre-profile-pairing-coordinator'
 import type { ConnectionLogEntry } from '../src/transport/types'
-import { useCloseHost } from '../src/transport/client-context'
+import { useRefreshHostClient } from '../src/transport/client-context'
 import { colors, spacing, radii, typography } from '../src/theme/mobile-theme'
 import { ConnectionLog } from '../src/components/ConnectionLog'
 import {
@@ -28,7 +28,7 @@ const PAIRING_OVERALL_TIMEOUT_MS = 25_000

 export default function PairConfirmScreen() {
  const router = useRouter()
-  const closeHost = useCloseHost()
+  const refreshHostClient = useRefreshHostClient()
  const insets = useSafeAreaInsets()
  const params = useLocalSearchParams<{ code?: string }>()
  const [status, setStatus] = useState<Status>('awaiting-confirm')
@@ -112,10 +112,8 @@ export default function PairConfirmScreen() {
      // Why: re-pairing the same desktop now reuses its existing host id
      // (STA-1840 dedup), so a client cached under that id from an earlier
      // pairing would keep the stale endpoint/relay. Close it so the
-      // destination screen opens a fresh client with the newly-paired
-      // profile — the removeHost() path already refreshes on re-pair, and a
-      // brand-new host has no cached entry so this is a no-op.
-      closeHost(hostId)
+      // Refresh any cached client from the newly persisted pairing profile.
+      refreshHostClient(hostId)
      const onboardingSteps = await loadMobileOnboardingSteps()
      if (!mountedRef.current) {
        return
@@ -18,7 +18,7 @@ import {
  type PreProfilePairingAttempt
 } from '../src/transport/pre-profile-pairing-coordinator'
 import type { ConnectionLogEntry, PairingOffer } from '../src/transport/types'
-import { useCloseHost } from '../src/transport/client-context'
+import { useRefreshHostClient } from '../src/transport/client-context'
 import { colors, spacing, radii, typography } from '../src/theme/mobile-theme'
 import { TextInputModal } from '../src/components/TextInputModal'
 import { ConnectionLog } from '../src/components/ConnectionLog'
@@ -47,7 +47,7 @@ function Step({ number, text }: { number: number; text: string }) {

 export default function PairScanScreen() {
  const router = useRouter()
-  const closeHost = useCloseHost()
+  const refreshHostClient = useRefreshHostClient()
  const insets = useSafeAreaInsets()
  const [permission, requestPermission] = useCameraPermissions()
  const [status, setStatus] = useState<'scanning' | 'connecting' | 'error'>('scanning')
@@ -156,10 +156,8 @@ export default function PairScanScreen() {
      // Why: re-pairing the same desktop now reuses its existing host id
      // (STA-1840 dedup), so a client cached under that id from an earlier
      // pairing would keep the stale endpoint/relay. Close it so the
-      // destination screen opens a fresh client with the newly-paired
-      // profile — the removeHost() path already refreshes on re-pair, and a
-      // brand-new host has no cached entry so this is a no-op.
-      closeHost(hostId)
+      // Refresh any cached client from the newly persisted pairing profile.
+      refreshHostClient(hostId)
      const onboardingSteps = await loadMobileOnboardingSteps()
      if (!mountedRef.current) {
        return
@@ -1,4 +1,4 @@
-import { createElement } from 'react'
+import { createElement, type ReactElement } from 'react'
 import { act, create, type ReactTestRenderer } from 'react-test-renderer'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 import type { ConnectionState } from './types'
@@ -20,7 +20,12 @@ vi.mock('./connection-revival-triggers', () => ({
  subscribeConnectionRevivalTriggers: () => () => {}
 }))

-import { RpcClientProvider, useCloseHost, useForceReconnect, useHostClient } from './client-context'
+import {
+  RpcClientProvider,
+  useDisconnectHostClient,
+  useForceReconnect,
+  useHostClient
+} from './client-context'
 import { useAllHostClients } from './use-all-host-clients'
 import { selectHomeAutoConnectHostIds } from './home-host-auto-connect'

@@ -67,25 +72,25 @@ const HOST = {

 type Harness = {
  readonly hook: ReturnType<typeof useHostClient>
-  readonly closeHost: (hostId: string) => void
+  readonly disconnectHost: (hostId: string) => void
  readonly unmount: () => void
 }

 async function renderHarness(hostId: string): Promise<Harness> {
  let hook: ReturnType<typeof useHostClient> | null = null
-  let closeHost: ((hostId: string) => void) | null = null
+  let disconnectHost: ((hostId: string) => void) | null = null
  let renderer: ReactTestRenderer | null = null

  function Probe(): null {
    hook = useHostClient(hostId)
-    closeHost = useCloseHost()
+    disconnectHost = useDisconnectHostClient()
    return null
  }

  await act(async () => {
    renderer = create(createElement(RpcClientProvider, null, createElement(Probe)))
  })
-  if (!hook || !closeHost || !renderer) {
+  if (!hook || !disconnectHost || !renderer) {
    throw new Error('harness did not render')
  }
  const mounted = renderer as ReactTestRenderer
@@ -96,11 +101,11 @@ async function renderHarness(hostId: string): Promise<Harness> {
      }
      return hook
    },
-    closeHost: (id) => {
-      if (!closeHost) {
-        throw new Error('closeHost not rendered')
+    disconnectHost: (id) => {
+      if (!disconnectHost) {
+        throw new Error('disconnectHost not rendered')
      }
-      closeHost(id)
+      disconnectHost(id)
    },
    unmount: () => mounted.unmount()
  }
@@ -203,11 +208,11 @@ describe('useHostClient', () => {
    expect(harness.hook.client).not.toBeNull()
    expect(harness.hook.state).toBe('connected')

-    // Regression (STA-1511): closeHost deletes the entry; before the fix the
+    // Regression (STA-1511): disconnect deletes the entry; before the fix the
    // hook kept handing out the closed client, so mounted screens kept
    // driving requests that could never resolve.
    await act(async () => {
-      harness.closeHost(HOST.id)
+      harness.disconnectHost(HOST.id)
    })
    expect(fake.closeMock).toHaveBeenCalled()
    expect(harness.hook.client).toBeNull()
@@ -216,6 +221,40 @@ describe('useHostClient', () => {
    harness.unmount()
  })

+  it('closes an ownerless reconnect after disconnect retires a mounted owner', async () => {
+    const initialClient = makeFakeClient('connected')
+    const replacementClient = makeFakeClient('connected')
+    connectMock.mockReturnValueOnce(initialClient).mockReturnValueOnce(replacementClient)
+    loadHostsMock.mockResolvedValue([HOST])
+    let disconnectHost: ((hostId: string) => void) | null = null
+    let reconnectHost: ((hostId: string) => Promise<void>) | null = null
+
+    function Probe(): null {
+      useAllHostClients([HOST.id], { closeUnusedOnRelease: true })
+      disconnectHost = useDisconnectHostClient()
+      reconnectHost = useForceReconnect()
+      return null
+    }
+    function App({ visible }: { visible: boolean }) {
+      return createElement(RpcClientProvider, null, visible ? createElement(Probe) : null)
+    }
+
+    let renderer: { update(element: ReactElement): void; unmount(): void } | null = null
+    await act(async () => {
+      renderer = create(createElement(App, { visible: true }))
+      await Promise.resolve()
+    })
+    await act(async () => {
+      disconnectHost?.(HOST.id)
+      await reconnectHost?.(HOST.id)
+    })
+    act(() => renderer?.update(createElement(App, { visible: false })))
+
+    expect(initialClient.closeMock).toHaveBeenCalledOnce()
+    expect(replacementClient.closeMock).toHaveBeenCalledOnce()
+    act(() => renderer?.unmount())
+  })
+
  it('reports disconnected instead of hanging when the host id is unknown', async () => {
    loadHostsMock.mockResolvedValue([])

@@ -299,10 +338,10 @@ describe('useHostClient', () => {
    connectMock.mockReturnValue(fake)
    loadHostsMock.mockReturnValue(hostLookup)

-    let closeHost: ((hostId: string) => void) | null = null
+    let disconnectHost: ((hostId: string) => void) | null = null
    let renderer: ReactTestRenderer | null = null
    function Probe(): null {
-      closeHost = useCloseHost()
+      disconnectHost = useDisconnectHostClient()
      useHostClient(HOST.id)
      return null
    }
@@ -311,11 +350,11 @@ describe('useHostClient', () => {
      renderer = create(createElement(RpcClientProvider, null, createElement(Probe)))
    })
    expect(loadHostsMock).toHaveBeenCalledOnce()
-    if (!closeHost || !resolveHosts || !renderer) {
+    if (!disconnectHost || !resolveHosts || !renderer) {
      throw new Error('pending-open harness did not initialize')
    }

-    act(() => closeHost?.(HOST.id))
+    act(() => disconnectHost?.(HOST.id))
    await act(async () => {
      resolveHosts?.([HOST])
      await hostLookup
@@ -11,12 +11,16 @@ import {
  type ReactNode
 } from 'react'
 import type { RpcClient } from './rpc-client'
-import { connectionLogStore } from './connection-log-buffer'
 import { subscribeConnectionRevivalTriggers } from './connection-revival-triggers'
 import { HostClientOpenRegistry } from './host-client-open-registry'
-import { decrementPendingAcquisition } from './host-client-acquisition-count'
 import {
-  clientActivePath,
+  HostClientAcquisitionRegistry,
+  type HostClientAcquisition
+} from './host-client-acquisition-registry'
+import { HostOpenRetryScheduler } from './host-open-retry-scheduler'
+import { openHostClientEntry, type HostClientStoreEntry } from './host-entry-opener'
+import {
+  createHostClientSelectors,
  listHostClients,
  notifyAllHostListeners,
  notifyHostStateListeners,
@@ -25,38 +29,12 @@ import {
  subscribeHostStateListener,
  type CloseEntryOptions
 } from './host-client-context-state'
-import { loadHosts } from './host-store'
-import { openHostLogicalClient } from './host-logical-client'
-import type { MobileConnectionPath } from './stable-logical-rpc-client'
 import type { ConnectionState, HostProfile } from './types'
+import type { RpcClientContextValue } from './rpc-client-context-contract'

-type StoreEntry = {
-  client: RpcClient
-  state: ConnectionState
-  refCount: number
-  unsubState: () => void
-}
-
-export type RpcClientContextValue = {
-  acquire: (hostId: string, host?: HostProfile) => RpcClient | null
-  release: (hostId: string) => void
-  releaseAndCloseIfUnused: (hostId: string) => void
-  closeIfUnused: (hostId: string) => void
-  forceReconnect: (hostId: string) => Promise<void>
-  closeHost: (hostId: string) => void
-  getState: (hostId: string) => ConnectionState
-  // null = host has no client entry and no open in flight; callers pick the default.
-  getKnownState: (hostId: string) => ConnectionState | null
-  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
-}
+type StoreEntry = HostClientStoreEntry
+export type { HostClientAcquisition } from './host-client-acquisition-registry'
+export type { RpcClientContextValue } from './rpc-client-context-contract'

 const Ctx = createContext<RpcClientContextValue | null>(null)

@@ -69,6 +47,9 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
  // Pending opens keyed by hostId so two acquire() callers in the same render don't race the host lookup.
  const pendingOpensRef = useRef(new HostClientOpenRegistry())
  const pendingAcquisitionsRef = useRef<Map<string, number>>(new Map())
+  const acquisitionsRef = useRef(new HostClientAcquisitionRegistry())
+  const manualDemandRef = useRef<Set<string>>(new Set())
+  const retrySchedulerRef = useRef<HostOpenRetryScheduler | null>(null)

  // Why: cache of already-loaded HostProfiles so openEntry can skip a second loadHosts()/Keychain pass on cold start.
  const primedHostsRef = useRef<Map<string, HostProfile>>(new Map())
@@ -79,13 +60,18 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {

  const closeEntry = useCallback((hostId: string, options: CloseEntryOptions) => {
    const entry = storeRef.current.get(hostId)
-    const acquisitionCount = entry?.refCount ?? pendingAcquisitionsRef.current.get(hostId) ?? 0
+    const acquisitionCount = acquisitionsRef.current.count(hostId)
    pendingOpensRef.current.cancel(hostId)
+    retrySchedulerRef.current?.cancel(hostId)
    if (options.preserveAcquisitions && acquisitionCount > 0) {
      pendingAcquisitionsRef.current.set(hostId, acquisitionCount)
    } else {
+      acquisitionsRef.current.clear(hostId)
      pendingAcquisitionsRef.current.delete(hostId)
    }
+    if (!options.preserveAcquisitions) {
+      manualDemandRef.current.delete(hostId)
+    }
    if (options.forgetPrimedHost) {
      primedHostsRef.current.delete(hostId)
    }
@@ -96,107 +82,59 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
    notifyAllHosts()
  }, [])

-  const closeHost = useCallback(
-    (hostId: string) => {
-      closeEntry(hostId, { forgetPrimedHost: true, preserveAcquisitions: true })
+  const openEntry = useCallback(
+    (hostId: string, allowUnowned = false): Promise<StoreEntry | null> => {
+      const retryScheduler = retrySchedulerRef.current
+      if (!retryScheduler) {
+        throw new Error('host retry scheduler not initialized')
+      }
+      return openHostClientEntry(
+        {
+          store: storeRef.current,
+          pendingOpens: pendingOpensRef.current,
+          pendingAcquisitions: pendingAcquisitionsRef.current,
+          primedHosts: primedHostsRef.current,
+          retryScheduler,
+          notifyHostState,
+          notifyAllHosts
+        },
+        hostId,
+        allowUnowned
+      ).then((entry) => {
+        if (entry) {
+          manualDemandRef.current.delete(hostId)
+        }
+        return entry
+      })
    },
-    [closeEntry]
+    []
  )

-  const openEntry = useCallback(async (hostId: string): Promise<StoreEntry | null> => {
-    const existing = pendingOpensRef.current.getActivePromise(hostId)
-    if (existing) {
-      await existing
-      return storeRef.current.get(hostId) ?? null
-    }
-    let resolve: () => void = () => {}
-    const promise = new Promise<void>((res) => {
-      resolve = res
+  if (!retrySchedulerRef.current) {
+    retrySchedulerRef.current = new HostOpenRetryScheduler({
+      canRetry: (hostId, generation) =>
+        pendingOpensRef.current.isGenerationCurrent(hostId, generation) &&
+        ((pendingAcquisitionsRef.current.get(hostId) ?? 0) > 0 ||
+          manualDemandRef.current.has(hostId)) &&
+        !storeRef.current.has(hostId) &&
+        pendingOpensRef.current.getActivePromise(hostId) === null,
+      open: (hostId) => void openEntry(hostId, manualDemandRef.current.has(hostId))
    })
-    const pendingOpen = pendingOpensRef.current.register(hostId, promise)
-    // Why: already-mounted subscribers must go amber for the Keychain read too.
-    notifyHostState(hostId, 'connecting')
-
-    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
-        }
-      }
-
-      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
-      try {
-        client = openHostLogicalClient(host, (entry) => connectionLogStore.append(hostId, entry))
-      } 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: pendingAcquisitionsRef.current.get(hostId) ?? 0,
-        unsubState
-      }
-      pendingAcquisitionsRef.current.delete(hostId)
-      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(
-    (hostId: string, host?: HostProfile): RpcClient | null => {
+    (hostId: string, acquisition: HostClientAcquisition, host?: HostProfile): RpcClient | null => {
      if (host) {
        primedHostsRef.current.set(hostId, host)
      }
+      const acquisitionCount = acquisitionsRef.current.acquire(hostId, acquisition)
      const existing = storeRef.current.get(hostId)
      if (existing) {
-        existing.refCount += 1
+        existing.refCount = acquisitionCount
        return existing.client
      }
-      pendingAcquisitionsRef.current.set(
-        hostId,
-        (pendingAcquisitionsRef.current.get(hostId) ?? 0) + 1
-      )
+      pendingAcquisitionsRef.current.set(hostId, acquisitionCount)
      // Trigger async open; returns null this tick — consumers re-call acquire() from an effect that re-runs on state changes.
      void openEntry(hostId)
      return null
@@ -209,27 +147,72 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
    []
  )

+  const refreshHostClient = useCallback(
+    (hostId: string) => {
+      closeEntry(hostId, { forgetPrimedHost: true, preserveAcquisitions: true })
+      if ((pendingAcquisitionsRef.current.get(hostId) ?? 0) > 0) {
+        void openEntry(hostId)
+      }
+    },
+    [closeEntry, openEntry]
+  )
+
+  const forgetHostClient = useCallback(
+    (hostId: string) => {
+      closeEntry(hostId, { forgetPrimedHost: true, preserveAcquisitions: false })
+    },
+    [closeEntry]
+  )
+
+  const disconnectHostClient = useCallback(
+    (hostId: string) => {
+      closeEntry(hostId, { forgetPrimedHost: false, preserveAcquisitions: false })
+    },
+    [closeEntry]
+  )
+
  // Why: no idle-close on refcount→0 — transient nav gaps flashed false 'disconnected', so keep sockets alive while foregrounded.
-  const release = useCallback((hostId: string) => {
-    const entry = storeRef.current.get(hostId)
-    if (entry) {
-      entry.refCount = Math.max(0, entry.refCount - 1)
+  const release = useCallback((hostId: string, acquisition: HostClientAcquisition) => {
+    const acquisitionCount = acquisitionsRef.current.release(hostId, acquisition)
+    if (acquisitionCount === null) {
      return
    }
-    decrementPendingAcquisition(pendingAcquisitionsRef.current, hostId)
+    const entry = storeRef.current.get(hostId)
+    if (entry) {
+      entry.refCount = acquisitionCount
+      return
+    }
+    if (acquisitionCount > 0) {
+      pendingAcquisitionsRef.current.set(hostId, acquisitionCount)
+    } else {
+      pendingAcquisitionsRef.current.delete(hostId)
+      if (!manualDemandRef.current.has(hostId)) {
+        pendingOpensRef.current.cancel(hostId)
+        retrySchedulerRef.current?.cancel(hostId)
+      }
+    }
  }, [])

  const releaseAndCloseIfUnused = useCallback(
-    (hostId: string) => {
+    (hostId: string, acquisition: HostClientAcquisition) => {
+      const acquisitionCount = acquisitionsRef.current.release(hostId, acquisition)
+      if (acquisitionCount === null) {
+        if (acquisitionsRef.current.count(hostId) === 0) {
+          closeEntry(hostId, { forgetPrimedHost: false, preserveAcquisitions: false })
+        }
+        return
+      }
      const entry = storeRef.current.get(hostId)
      if (entry) {
-        entry.refCount = Math.max(0, entry.refCount - 1)
+        entry.refCount = acquisitionCount
        if (entry.refCount === 0) {
          closeEntry(hostId, { forgetPrimedHost: false, preserveAcquisitions: false })
        }
        return
      }
-      if (decrementPendingAcquisition(pendingAcquisitionsRef.current, hostId) === 0) {
+      if (acquisitionCount > 0) {
+        pendingAcquisitionsRef.current.set(hostId, acquisitionCount)
+      } else {
        closeEntry(hostId, { forgetPrimedHost: false, preserveAcquisitions: false })
      }
    },
@@ -239,12 +222,12 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
  const closeIfUnused = useCallback(
    (hostId: string) => {
      const entry = storeRef.current.get(hostId)
-      const pendingCount = pendingAcquisitionsRef.current.get(hostId)
+      const acquisitionCount = acquisitionsRef.current.count(hostId)
      const hasPendingOpen = pendingOpensRef.current.getActivePromise(hostId) !== null
-      if (!entry && pendingCount === undefined && !hasPendingOpen) {
+      if (!entry && acquisitionCount === 0 && !hasPendingOpen) {
        return
      }
-      if ((entry?.refCount ?? pendingCount ?? 0) === 0) {
+      if (acquisitionCount === 0) {
        closeEntry(hostId, { forgetPrimedHost: false, preserveAcquisitions: false })
      }
    },
@@ -255,12 +238,15 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
    async (hostId: string) => {
      const entry = storeRef.current.get(hostId)
      // Why: ownership survives explicit close/re-pair while observers never become synthetic owners.
-      const savedRefCount = entry?.refCount ?? pendingAcquisitionsRef.current.get(hostId) ?? 0
+      const savedRefCount = acquisitionsRef.current.count(hostId)
+      manualDemandRef.current.add(hostId)
      if (entry) {
        entry.unsubState()
        entry.client.close()
        storeRef.current.delete(hostId)
      }
+      pendingOpensRef.current.cancel(hostId)
+      retrySchedulerRef.current?.cancel(hostId)
      if (savedRefCount > 0) {
        pendingAcquisitionsRef.current.set(
          hostId,
@@ -269,38 +255,13 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
      }
      // Why: Retry must read amber for the whole reopen, not grey-then-amber.
      notifyHostState(hostId, 'connecting')
-      await openEntry(hostId)
+      await openEntry(hostId, true)
    },
    [openEntry]
  )

-  // null = no entry and no open in flight; callers pick their own default.
-  const getKnownState = useCallback((hostId: string): ConnectionState | null => {
-    const entry = storeRef.current.get(hostId)
-    if (entry) {
-      return entry.state
-    }
-    // Why: the async open (a Keychain pass) predates the store entry; reading that
-    // window as 'disconnected' made every host screen flash dead on mount (S2).
-    return pendingOpensRef.current.getActivePromise(hostId) ? 'connecting' : null
-  }, [])
-
-  const getState = useCallback(
-    (hostId: string): ConnectionState => getKnownState(hostId) ?? 'disconnected',
-    [getKnownState]
-  )
-
-  const getReconnectAttempt = useCallback((hostId: string): number => {
-    return storeRef.current.get(hostId)?.client.getReconnectAttempt() ?? 0
-  }, [])
-
-  const getLastConnectedAt = useCallback((hostId: string): number | null => {
-    return storeRef.current.get(hostId)?.client.getLastConnectedAt() ?? null
-  }, [])
-
-  const getActivePath = useCallback((hostId: string): MobileConnectionPath => {
-    return clientActivePath(storeRef.current.get(hostId)?.client)
-  }, [])
+  const { getKnownState, getState, getReconnectAttempt, getLastConnectedAt, getActivePath } =
+    useMemo(() => createHostClientSelectors(storeRef.current, pendingOpensRef.current), [])

  const subscribeHostState = useCallback(
    (hostId: string, listener: (state: ConnectionState) => void) =>
@@ -321,6 +282,9 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
    const store = storeRef.current
    return () => {
      pendingOpensRef.current.cancelAll()
+      retrySchedulerRef.current?.cancelAll()
+      acquisitionsRef.current.clearAll()
+      manualDemandRef.current.clear()
      pendingAcquisitionsRef.current.clear()
      for (const [hostId] of store) {
        closeEntry(hostId, { forgetPrimedHost: true, preserveAcquisitions: false })
@@ -331,8 +295,15 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
  // Why: nudge live clients when the OS signals the link may be back so sessions recover without a restart (issue #5049).
  useEffect(() => {
    return subscribeConnectionRevivalTriggers((reason) => {
+      for (const hostId of pendingAcquisitionsRef.current.keys()) {
+        retrySchedulerRef.current?.expedite(hostId)
+      }
      for (const entry of storeRef.current.values()) {
-        entry.client.notifyForeground(reason)
+        try {
+          entry.client.notifyForeground(reason)
+        } catch {
+          // One broken physical session must not block recovery for other hosts.
+        }
      }
    })
  }, [])
@@ -344,7 +315,9 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
      releaseAndCloseIfUnused,
      closeIfUnused,
      forceReconnect,
-      closeHost,
+      refreshHostClient,
+      forgetHostClient,
+      disconnectHostClient,
      getState,
      getKnownState,
      getReconnectAttempt,
@@ -361,7 +334,9 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
      releaseAndCloseIfUnused,
      closeIfUnused,
      forceReconnect,
-      closeHost,
+      refreshHostClient,
+      forgetHostClient,
+      disconnectHostClient,
      getState,
      getKnownState,
      getReconnectAttempt,
@@ -399,6 +374,7 @@ export function useHostClient(hostId: string | undefined): {
  )
  const clientRef = useRef<RpcClient | null>(null)
  const clientHostIdRef = useRef<string | undefined>(hostId)
+  const acquisitionRef = useRef<HostClientAcquisition>({})

  useEffect(() => {
    if (!hostId) {
@@ -421,12 +397,12 @@ export function useHostClient(hostId: string | undefined): {
        clientRef.current = found.client
        force((n) => n + 1)
      } else if (!found && clientRef.current) {
-        // Why: closeHost deletes the entry with no replacement; null it so screens don't drive a dead client (STA-1511).
+        // Why: disconnect/forget deletes the entry; never retain a dead client (STA-1511).
        clientRef.current = null
        force((n) => n + 1)
      }
    })
-    const initial = ctx.acquire(hostId)
+    const initial = ctx.acquire(hostId, acquisitionRef.current)
    clientRef.current = initial
    setState(ctx.getKnownState(hostId) ?? 'connecting')
    if (initial) {
@@ -436,7 +412,7 @@ export function useHostClient(hostId: string | undefined): {
    return () => {
      cancelled = true
      unsub()
-      ctx.release(hostId)
+      ctx.release(hostId, acquisitionRef.current)
      clientRef.current = null
      clientHostIdRef.current = undefined
    }
@@ -453,9 +429,19 @@ export function useHostClient(hostId: string | undefined): {
 }

 // Why: host-store's removeHost() must close the live client but has no React-side handle; this hook bridges to it.
-export function useCloseHost(): (hostId: string) => void {
+export function useRefreshHostClient(): (hostId: string) => void {
  const ctx = useRpcClientContext()
-  return ctx.closeHost
+  return ctx.refreshHostClient
+}
+
+export function useForgetHostClient(): (hostId: string) => void {
+  const ctx = useRpcClientContext()
+  return ctx.forgetHostClient
+}
+
+export function useDisconnectHostClient(): (hostId: string) => void {
+  const ctx = useRpcClientContext()
+  return ctx.disconnectHostClient
 }

 // Why: future-proof "Connection issues — try again" affordance.
@@ -1,9 +0,0 @@
-export function decrementPendingAcquisition(pending: Map<string, number>, hostId: string): number {
-  const next = Math.max(0, (pending.get(hostId) ?? 0) - 1)
-  if (next === 0) {
-    pending.delete(hostId)
-  } else {
-    pending.set(hostId, next)
-  }
-  return next
-}
@@ -0,0 +1,38 @@
+export type HostClientAcquisition = object
+
+export class HostClientAcquisitionRegistry {
+  private readonly acquisitions = new Map<string, Set<HostClientAcquisition>>()
+
+  acquire(hostId: string, acquisition: HostClientAcquisition): number {
+    let active = this.acquisitions.get(hostId)
+    if (!active) {
+      active = new Set()
+      this.acquisitions.set(hostId, active)
+    }
+    active.add(acquisition)
+    return active.size
+  }
+
+  release(hostId: string, acquisition: HostClientAcquisition): number | null {
+    const active = this.acquisitions.get(hostId)
+    if (!active?.delete(acquisition)) {
+      return null
+    }
+    if (active.size === 0) {
+      this.acquisitions.delete(hostId)
+    }
+    return active.size
+  }
+
+  count(hostId: string): number {
+    return this.acquisitions.get(hostId)?.size ?? 0
+  }
+
+  clear(hostId: string): void {
+    this.acquisitions.delete(hostId)
+  }
+
+  clearAll(): void {
+    this.acquisitions.clear()
+  }
+}
@@ -1,4 +1,6 @@
 import type { RpcClient } from './rpc-client'
+import type { HostClientOpenRegistry } from './host-client-open-registry'
+import type { HostClientStoreEntry } from './host-entry-opener'
 import type { MobileConnectionPath, StableLogicalRpcClient } from './stable-logical-rpc-client'
 import type { ConnectionState, HostProfile } from './types'

@@ -62,6 +64,30 @@ export function primeHostProfiles(cache: Map<string, HostProfile>, hosts: HostPr
  }
 }

+export function createHostClientSelectors(
+  entries: ReadonlyMap<string, HostClientStoreEntry>,
+  pendingOpens: HostClientOpenRegistry
+) {
+  const getKnownState = (hostId: string): ConnectionState | null => {
+    const entry = entries.get(hostId)
+    if (entry) {
+      return entry.state
+    }
+    // Why: the Keychain pass predates the store entry; this window is connecting.
+    return pendingOpens.getActivePromise(hostId) ? 'connecting' : null
+  }
+  return {
+    getKnownState,
+    getState: (hostId: string): ConnectionState => getKnownState(hostId) ?? 'disconnected',
+    getReconnectAttempt: (hostId: string): number =>
+      entries.get(hostId)?.client.getReconnectAttempt() ?? 0,
+    getLastConnectedAt: (hostId: string): number | null =>
+      entries.get(hostId)?.client.getLastConnectedAt() ?? null,
+    getActivePath: (hostId: string): MobileConnectionPath =>
+      clientActivePath(entries.get(hostId)?.client)
+  }
+}
+
 export function clientActivePath(client: RpcClient | undefined): MobileConnectionPath {
  const logical = client as Partial<StableLogicalRpcClient> | undefined
  if (typeof logical?.getActivePath !== 'function') {
@@ -1,10 +1,12 @@
 export type HostClientOpenTicket = {
  cancelled: boolean
+  generation: number
  promise: Promise<void>
 }

 export class HostClientOpenRegistry {
  private readonly pending = new Map<string, HostClientOpenTicket>()
+  private readonly generations = new Map<string, number>()

  getActivePromise(hostId: string): Promise<void> | null {
    const ticket = this.pending.get(hostId)
@@ -12,7 +14,11 @@ export class HostClientOpenRegistry {
  }

  register(hostId: string, promise: Promise<void>): HostClientOpenTicket {
-    const ticket = { cancelled: false, promise }
+    const ticket = {
+      cancelled: false,
+      generation: this.advanceGeneration(hostId),
+      promise
+    }
    this.pending.set(hostId, ticket)
    return ticket
  }
@@ -25,6 +31,15 @@ export class HostClientOpenRegistry {
      // reference immediately while the ticket still cancels its continuation.
      this.pending.delete(hostId)
    }
+    this.advanceGeneration(hostId)
+  }
+
+  isCurrent(hostId: string, ticket: HostClientOpenTicket): boolean {
+    return !ticket.cancelled && this.isGenerationCurrent(hostId, ticket.generation)
+  }
+
+  isGenerationCurrent(hostId: string, generation: number): boolean {
+    return this.generations.get(hostId) === generation
  }

  deleteIfCurrent(hostId: string, ticket: HostClientOpenTicket): void {
@@ -34,9 +49,16 @@ export class HostClientOpenRegistry {
  }

  cancelAll(): void {
-    for (const ticket of this.pending.values()) {
+    for (const [hostId, ticket] of this.pending) {
      ticket.cancelled = true
+      this.advanceGeneration(hostId)
    }
    this.pending.clear()
  }
+
+  private advanceGeneration(hostId: string): number {
+    const generation = (this.generations.get(hostId) ?? 0) + 1
+    this.generations.set(hostId, generation)
+    return generation
+  }
 }
@@ -0,0 +1,141 @@
+import { connectionLogStore } from './connection-log-buffer'
+import { loadHosts } from './host-store'
+import { openHostLogicalClient } from './host-logical-client'
+import type { HostClientOpenRegistry } from './host-client-open-registry'
+import type { HostOpenRetryScheduler } from './host-open-retry-scheduler'
+import type { RpcClient } from './rpc-client'
+import type { ConnectionState, HostProfile } from './types'
+
+export type HostClientStoreEntry = {
+  client: RpcClient
+  state: ConnectionState
+  refCount: number
+  unsubState: () => void
+}
+
+type HostEntryOpenerState = {
+  store: Map<string, HostClientStoreEntry>
+  pendingOpens: HostClientOpenRegistry
+  pendingAcquisitions: Map<string, number>
+  primedHosts: Map<string, HostProfile>
+  retryScheduler: HostOpenRetryScheduler
+  notifyHostState: (hostId: string, state: ConnectionState) => void
+  notifyAllHosts: () => void
+}
+
+type HostOpenFailureCategory = 'catalog-unavailable' | 'host-not-found' | 'client-construction'
+
+export async function openHostClientEntry(
+  state: HostEntryOpenerState,
+  hostId: string,
+  allowUnowned = false
+): Promise<HostClientStoreEntry | null> {
+  const existing = state.pendingOpens.getActivePromise(hostId)
+  if (existing) {
+    await existing
+    return state.store.get(hostId) ?? null
+  }
+  let resolve: () => void = () => {}
+  const promise = new Promise<void>((res) => {
+    resolve = res
+  })
+  const ticket = state.pendingOpens.register(hostId, promise)
+  let settled = false
+  const settle = () => {
+    if (settled) {
+      return
+    }
+    settled = true
+    state.pendingOpens.deleteIfCurrent(hostId, ticket)
+    resolve()
+  }
+  const isCurrent = () => state.pendingOpens.isCurrent(hostId, ticket)
+  const isWanted = () => allowUnowned || (state.pendingAcquisitions.get(hostId) ?? 0) > 0
+  const failCurrentOpen = (category: HostOpenFailureCategory) => {
+    if (!isCurrent()) {
+      return
+    }
+    settle()
+    state.notifyHostState(hostId, 'disconnected')
+    state.notifyAllHosts()
+    const retry = state.retryScheduler.recordFailure(hostId, ticket.generation)
+    connectionLogStore.append(hostId, {
+      id: `host-open-${ticket.generation}-${Date.now()}`,
+      ts: Date.now(),
+      level: 'error',
+      message: 'Host client open failed',
+      detail: `${category}; retry ${retry.nextDelayMs}ms (failure ${retry.failureCount})`
+    })
+  }
+  state.notifyHostState(hostId, 'connecting')
+
+  try {
+    let host = state.primedHosts.get(hostId)
+    if (!host) {
+      try {
+        const hosts = await loadHosts()
+        host = hosts.find((candidate) => candidate.id === hostId)
+      } catch {
+        failCurrentOpen('catalog-unavailable')
+        return null
+      }
+      if (!host) {
+        failCurrentOpen('host-not-found')
+        return null
+      }
+    }
+    if (!isCurrent() || !isWanted()) {
+      return null
+    }
+    const published = state.store.get(hostId)
+    if (published) {
+      settle()
+      state.retryScheduler.recordSuccess(hostId)
+      return published
+    }
+
+    let client: RpcClient
+    try {
+      client = openHostLogicalClient(host, (entry) => connectionLogStore.append(hostId, entry))
+    } catch {
+      failCurrentOpen('client-construction')
+      return null
+    }
+    if (!isCurrent() || !isWanted() || state.store.has(hostId)) {
+      client.close()
+      return state.store.get(hostId) ?? null
+    }
+    const unsubState = client.onStateChange((next) => {
+      const current = state.store.get(hostId)
+      if (!current) {
+        return
+      }
+      current.state = next
+      state.notifyHostState(hostId, next)
+    })
+    const entry: HostClientStoreEntry = {
+      client,
+      state: client.getState(),
+      refCount: state.pendingAcquisitions.get(hostId) ?? 0,
+      unsubState
+    }
+    state.pendingAcquisitions.delete(hostId)
+    state.store.set(hostId, entry)
+    settle()
+    const priorFailureCount = state.retryScheduler.recordSuccess(hostId)
+    if (priorFailureCount > 0) {
+      connectionLogStore.append(hostId, {
+        id: `host-open-recovered-${ticket.generation}-${Date.now()}`,
+        ts: Date.now(),
+        level: 'success',
+        message: 'Host client recovered',
+        detail: `after ${priorFailureCount} failed open${priorFailureCount === 1 ? '' : 's'}`
+      })
+    }
+    state.notifyHostState(hostId, entry.state)
+    state.notifyAllHosts()
+    return entry
+  } finally {
+    settle()
+  }
+}
@@ -0,0 +1,388 @@
+import { createElement, type ReactElement } from 'react'
+import { act, create } from 'react-test-renderer'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import type { RpcClient } from './rpc-client'
+import type { ConnectionState } from './types'
+
+const openHostLogicalClientMock = vi.fn()
+const loadHostsMock = vi.fn()
+const revival = vi.hoisted(() => ({ callback: null as null | ((reason: 'focus') => void) }))
+
+vi.mock('./host-logical-client', () => ({
+  openHostLogicalClient: (...args: unknown[]) => openHostLogicalClientMock(...args)
+}))
+vi.mock('./host-store', () => ({
+  loadHosts: () => loadHostsMock()
+}))
+vi.mock('./connection-revival-triggers', () => ({
+  subscribeConnectionRevivalTriggers: (callback: (reason: 'focus') => void) => {
+    revival.callback = callback
+    return () => {
+      revival.callback = null
+    }
+  }
+}))
+
+import {
+  useDisconnectHostClient,
+  RpcClientProvider,
+  useForgetHostClient,
+  useForceReconnect,
+  useHostClient,
+  useRefreshHostClient
+} from './client-context'
+
+const HOST = {
+  id: 'host-1',
+  name: 'Host 1',
+  endpoint: 'ws://127.0.0.1:1',
+  deviceToken: 'token',
+  publicKeyB64: 'key',
+  lastConnected: 0
+}
+
+type MountedRenderer = {
+  unmount(): void
+  update(element: ReactElement): void
+}
+
+function fakeClient(): RpcClient {
+  return {
+    sendRequest: vi.fn(),
+    subscribe: vi.fn(() => () => {}),
+    updateTerminalSubscriptionViewport: vi.fn(),
+    getState: () => 'connected',
+    getReconnectAttempt: () => 0,
+    getLastConnectedAt: () => null,
+    onStateChange: () => () => {},
+    notifyForeground: vi.fn(),
+    close: vi.fn()
+  }
+}
+
+beforeEach(() => {
+  vi.useFakeTimers()
+  openHostLogicalClientMock.mockReset()
+  loadHostsMock.mockReset()
+  revival.callback = null
+})
+
+afterEach(() => vi.useRealTimers())
+
+describe('wanted host open recovery', () => {
+  it('recovers after a transient catalog failure without remounting', async () => {
+    const client = fakeClient()
+    loadHostsMock.mockRejectedValueOnce(new Error('catalog unavailable')).mockResolvedValue([HOST])
+    openHostLogicalClientMock.mockReturnValue(client)
+
+    let observed: { client: RpcClient | null; state: ConnectionState } | null = null
+    let renderer: MountedRenderer | null = null
+    function Probe(): null {
+      observed = useHostClient(HOST.id)
+      return null
+    }
+
+    try {
+      await act(async () => {
+        renderer = create(createElement(RpcClientProvider, null, createElement(Probe)))
+        await Promise.resolve()
+      })
+      expect(observed).toMatchObject({ client: null, state: 'disconnected' })
+      expect(loadHostsMock).toHaveBeenCalledOnce()
+
+      await act(async () => {
+        await vi.advanceTimersByTimeAsync(1_000)
+      })
+
+      expect(loadHostsMock).toHaveBeenCalledTimes(2)
+      expect(openHostLogicalClientMock).toHaveBeenCalledOnce()
+      expect(observed).toMatchObject({ client, state: 'connected' })
+    } finally {
+      act(() => renderer?.unmount())
+    }
+  })
+
+  it('recovers when a missing host appears in the catalog', async () => {
+    const client = fakeClient()
+    loadHostsMock.mockResolvedValueOnce([]).mockResolvedValueOnce([HOST])
+    openHostLogicalClientMock.mockReturnValue(client)
+
+    let observed: { client: RpcClient | null; state: ConnectionState } | null = null
+    let renderer: MountedRenderer | null = null
+    function Probe(): null {
+      observed = useHostClient(HOST.id)
+      return null
+    }
+
+    try {
+      await act(async () => {
+        renderer = create(createElement(RpcClientProvider, null, createElement(Probe)))
+        await Promise.resolve()
+      })
+      await act(async () => {
+        await vi.advanceTimersByTimeAsync(1_000)
+      })
+      expect(loadHostsMock).toHaveBeenCalledTimes(2)
+      expect(observed).toMatchObject({ client, state: 'connected' })
+    } finally {
+      act(() => renderer?.unmount())
+    }
+  })
+
+  it('recovers after synchronous client construction fails', async () => {
+    const client = fakeClient()
+    loadHostsMock.mockResolvedValue([HOST])
+    openHostLogicalClientMock.mockImplementationOnce(() => {
+      throw new Error('bad transient profile')
+    })
+    openHostLogicalClientMock.mockReturnValueOnce(client)
+
+    let observed: { client: RpcClient | null; state: ConnectionState } | null = null
+    let renderer: MountedRenderer | null = null
+    function Probe(): null {
+      observed = useHostClient(HOST.id)
+      return null
+    }
+
+    try {
+      await act(async () => {
+        renderer = create(createElement(RpcClientProvider, null, createElement(Probe)))
+        await Promise.resolve()
+      })
+      await act(async () => {
+        await vi.advanceTimersByTimeAsync(1_000)
+      })
+      expect(openHostLogicalClientMock).toHaveBeenCalledTimes(2)
+      expect(observed).toMatchObject({ client, state: 'connected' })
+    } finally {
+      act(() => renderer?.unmount())
+    }
+  })
+
+  it('expedites recovery without resetting the failure tier', async () => {
+    const client = fakeClient()
+    loadHostsMock
+      .mockRejectedValueOnce(new Error('first failure'))
+      .mockRejectedValueOnce(new Error('second failure'))
+      .mockResolvedValueOnce([HOST])
+    openHostLogicalClientMock.mockReturnValue(client)
+
+    let renderer: MountedRenderer | null = null
+    function Probe(): null {
+      useHostClient(HOST.id)
+      return null
+    }
+
+    try {
+      await act(async () => {
+        renderer = create(createElement(RpcClientProvider, null, createElement(Probe)))
+        await Promise.resolve()
+      })
+      act(() => revival.callback?.('focus'))
+      await act(async () => {
+        await Promise.resolve()
+      })
+      expect(loadHostsMock).toHaveBeenCalledTimes(2)
+
+      await vi.advanceTimersByTimeAsync(1_999)
+      expect(loadHostsMock).toHaveBeenCalledTimes(2)
+      await vi.advanceTimersByTimeAsync(1)
+      expect(loadHostsMock).toHaveBeenCalledTimes(3)
+    } finally {
+      act(() => renderer?.unmount())
+    }
+  })
+
+  it('forgets demand while a retry is waiting', async () => {
+    loadHostsMock.mockRejectedValue(new Error('catalog unavailable'))
+
+    let forgetHostClient: ((hostId: string) => void) | null = null
+    let renderer: MountedRenderer | null = null
+    function Probe(): null {
+      useHostClient(HOST.id)
+      forgetHostClient = useForgetHostClient()
+      return null
+    }
+
+    try {
+      await act(async () => {
+        renderer = create(createElement(RpcClientProvider, null, createElement(Probe)))
+        await Promise.resolve()
+      })
+      act(() => forgetHostClient?.(HOST.id))
+      await vi.advanceTimersByTimeAsync(120_000)
+      expect(loadHostsMock).toHaveBeenCalledOnce()
+    } finally {
+      act(() => renderer?.unmount())
+    }
+  })
+
+  it('does not let a retired owner cancel a newer acquisition', async () => {
+    let resolveReplacement: ((hosts: (typeof HOST)[]) => void) | null = null
+    const replacementLookup = new Promise<(typeof HOST)[]>((resolve) => {
+      resolveReplacement = resolve
+    })
+    loadHostsMock.mockResolvedValueOnce([HOST]).mockReturnValueOnce(replacementLookup)
+    const firstClient = fakeClient()
+    const replacementClient = fakeClient()
+    openHostLogicalClientMock
+      .mockReturnValueOnce(firstClient)
+      .mockReturnValueOnce(replacementClient)
+
+    let disconnectHostClient: ((hostId: string) => void) | null = null
+    function RetiredOwner(): null {
+      useHostClient(HOST.id)
+      disconnectHostClient = useDisconnectHostClient()
+      return null
+    }
+    function NewOwner(): null {
+      useHostClient(HOST.id)
+      return null
+    }
+    function App({ oldVisible, newVisible }: { oldVisible: boolean; newVisible: boolean }) {
+      return createElement(
+        RpcClientProvider,
+        null,
+        oldVisible ? createElement(RetiredOwner) : null,
+        newVisible ? createElement(NewOwner) : null
+      )
+    }
+
+    let renderer: MountedRenderer | null = null
+    try {
+      await act(async () => {
+        renderer = create(createElement(App, { oldVisible: true, newVisible: false }))
+        await Promise.resolve()
+      })
+      act(() => disconnectHostClient?.(HOST.id))
+      act(() => renderer?.update(createElement(App, { oldVisible: true, newVisible: true })))
+      act(() => renderer?.update(createElement(App, { oldVisible: false, newVisible: true })))
+
+      await act(async () => {
+        resolveReplacement?.([HOST])
+        await replacementLookup
+      })
+      expect(openHostLogicalClientMock).toHaveBeenCalledTimes(2)
+      expect(replacementClient.close).not.toHaveBeenCalled()
+    } finally {
+      act(() => renderer?.unmount())
+    }
+  })
+
+  it('keeps explicit reconnect demand through a pre-client failure', async () => {
+    loadHostsMock
+      .mockRejectedValueOnce(new Error('catalog unavailable'))
+      .mockResolvedValueOnce([HOST])
+    openHostLogicalClientMock.mockReturnValue(fakeClient())
+
+    let forceReconnect: ((hostId: string) => Promise<void>) | null = null
+    function Probe(): null {
+      forceReconnect = useForceReconnect()
+      return null
+    }
+
+    let renderer: MountedRenderer | null = null
+    try {
+      act(() => {
+        renderer = create(createElement(RpcClientProvider, null, createElement(Probe)))
+      })
+      await act(async () => {
+        await forceReconnect?.(HOST.id)
+      })
+      await act(async () => {
+        await vi.advanceTimersByTimeAsync(1_000)
+      })
+      expect(loadHostsMock).toHaveBeenCalledTimes(2)
+      expect(openHostLogicalClientMock).toHaveBeenCalledOnce()
+    } finally {
+      act(() => renderer?.unmount())
+    }
+  })
+
+  it('cancels an in-flight open after the final owner releases', async () => {
+    let resolveHosts: ((hosts: (typeof HOST)[]) => void) | null = null
+    const lookup = new Promise<(typeof HOST)[]>((resolve) => {
+      resolveHosts = resolve
+    })
+    loadHostsMock.mockReturnValue(lookup)
+    openHostLogicalClientMock.mockReturnValue(fakeClient())
+
+    function Probe(): null {
+      useHostClient(HOST.id)
+      return null
+    }
+    function App({ visible }: { visible: boolean }) {
+      return createElement(RpcClientProvider, null, visible ? createElement(Probe) : null)
+    }
+
+    let renderer: MountedRenderer | null = null
+    try {
+      act(() => {
+        renderer = create(createElement(App, { visible: true }))
+      })
+      expect(loadHostsMock).toHaveBeenCalledOnce()
+
+      act(() => renderer?.update(createElement(App, { visible: false })))
+      await act(async () => {
+        resolveHosts?.([HOST])
+        await lookup
+      })
+
+      expect(openHostLogicalClientMock).not.toHaveBeenCalled()
+    } finally {
+      act(() => renderer?.unmount())
+    }
+  })
+
+  it('ignores a canceled failure after a replacement client succeeds', async () => {
+    let rejectStale: ((error: Error) => void) | null = null
+    const staleLookup = new Promise<(typeof HOST)[]>((_, reject) => {
+      rejectStale = reject
+    })
+    loadHostsMock.mockReturnValueOnce(staleLookup).mockResolvedValueOnce([HOST])
+    const client = fakeClient()
+    openHostLogicalClientMock.mockReturnValue(client)
+
+    let refreshHostClient: ((hostId: string) => void) | null = null
+    let replacement: { client: RpcClient | null; state: ConnectionState } | null = null
+    function Primary(): null {
+      useHostClient(HOST.id)
+      refreshHostClient = useRefreshHostClient()
+      return null
+    }
+    function Replacement(): null {
+      replacement = useHostClient(HOST.id)
+      return null
+    }
+    function App({ showReplacement }: { showReplacement: boolean }) {
+      return createElement(
+        RpcClientProvider,
+        null,
+        createElement(Primary),
+        showReplacement ? createElement(Replacement) : null
+      )
+    }
+
+    let renderer: MountedRenderer | null = null
+    try {
+      act(() => {
+        renderer = create(createElement(App, { showReplacement: false }))
+      })
+      act(() => refreshHostClient?.(HOST.id))
+      await act(async () => {
+        renderer?.update(createElement(App, { showReplacement: true }))
+        await Promise.resolve()
+      })
+      expect(replacement).toMatchObject({ client, state: 'connected' })
+
+      await act(async () => {
+        rejectStale?.(new Error('stale catalog failure'))
+        await Promise.resolve()
+      })
+
+      expect(replacement).toMatchObject({ client, state: 'connected' })
+    } finally {
+      act(() => renderer?.unmount())
+    }
+  })
+})
@@ -0,0 +1,57 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { HostOpenRetryScheduler } from './host-open-retry-scheduler'
+
+describe('HostOpenRetryScheduler', () => {
+  beforeEach(() => vi.useFakeTimers())
+  afterEach(() => vi.useRealTimers())
+
+  it('advances through bounded retry tiers', async () => {
+    let generation = 1
+    const open = vi.fn()
+    const scheduler = new HostOpenRetryScheduler({
+      canRetry: (_hostId, candidate) => candidate === generation,
+      open
+    })
+
+    scheduler.recordFailure('host-1', generation)
+    await vi.advanceTimersByTimeAsync(1_000)
+    expect(open).toHaveBeenCalledOnce()
+
+    generation = 2
+    scheduler.recordFailure('host-1', generation)
+    await vi.advanceTimersByTimeAsync(1_999)
+    expect(open).toHaveBeenCalledOnce()
+    await vi.advanceTimersByTimeAsync(1)
+    expect(open).toHaveBeenCalledTimes(2)
+  })
+
+  it('expedites without forgiving the failure streak', async () => {
+    let generation = 1
+    const open = vi.fn()
+    const scheduler = new HostOpenRetryScheduler({
+      canRetry: (_hostId, candidate) => candidate === generation,
+      open
+    })
+
+    scheduler.recordFailure('host-1', generation)
+    scheduler.expedite('host-1')
+    expect(open).toHaveBeenCalledOnce()
+
+    generation = 2
+    scheduler.recordFailure('host-1', generation)
+    await vi.advanceTimersByTimeAsync(1_999)
+    expect(open).toHaveBeenCalledOnce()
+    await vi.advanceTimersByTimeAsync(1)
+    expect(open).toHaveBeenCalledTimes(2)
+  })
+
+  it('cancels retry delivery', async () => {
+    const open = vi.fn()
+    const scheduler = new HostOpenRetryScheduler({ canRetry: () => true, open })
+    scheduler.recordFailure('host-1', 1)
+    scheduler.cancel('host-1')
+
+    await vi.advanceTimersByTimeAsync(60_000)
+    expect(open).not.toHaveBeenCalled()
+  })
+})
@@ -0,0 +1,79 @@
+const RETRY_DELAYS_MS = [1_000, 2_000, 5_000, 15_000, 30_000, 60_000] as const
+
+type RetryState = {
+  failureCount: number
+  generation: number
+  timer: ReturnType<typeof setTimeout> | null
+}
+
+type HostOpenRetrySchedulerOptions = {
+  canRetry: (hostId: string, generation: number) => boolean
+  open: (hostId: string) => void
+  setTimer?: typeof setTimeout
+  clearTimer?: typeof clearTimeout
+}
+
+export class HostOpenRetryScheduler {
+  private readonly states = new Map<string, RetryState>()
+  private readonly setTimer: typeof setTimeout
+  private readonly clearTimer: typeof clearTimeout
+
+  constructor(private readonly options: HostOpenRetrySchedulerOptions) {
+    this.setTimer = options.setTimer ?? setTimeout
+    this.clearTimer = options.clearTimer ?? clearTimeout
+  }
+
+  recordFailure(hostId: string, generation: number): { failureCount: number; nextDelayMs: number } {
+    const previous = this.states.get(hostId)
+    this.clearStateTimer(previous)
+    const failureCount = (previous?.failureCount ?? 0) + 1
+    const state: RetryState = { failureCount, generation, timer: null }
+    this.states.set(hostId, state)
+    const delayIndex = Math.min(failureCount - 1, RETRY_DELAYS_MS.length - 1)
+    const nextDelayMs = RETRY_DELAYS_MS[delayIndex]
+    if (!this.options.canRetry(hostId, generation)) {
+      return { failureCount, nextDelayMs }
+    }
+    state.timer = this.setTimer(() => {
+      state.timer = null
+      if (this.states.get(hostId) === state && this.options.canRetry(hostId, generation)) {
+        this.options.open(hostId)
+      }
+    }, nextDelayMs)
+    return { failureCount, nextDelayMs }
+  }
+
+  expedite(hostId: string): void {
+    const state = this.states.get(hostId)
+    if (!state || !this.options.canRetry(hostId, state.generation)) {
+      return
+    }
+    this.clearStateTimer(state)
+    this.options.open(hostId)
+  }
+
+  recordSuccess(hostId: string): number {
+    const priorFailureCount = this.states.get(hostId)?.failureCount ?? 0
+    this.cancel(hostId)
+    return priorFailureCount
+  }
+
+  cancel(hostId: string): void {
+    this.clearStateTimer(this.states.get(hostId))
+    this.states.delete(hostId)
+  }
+
+  cancelAll(): void {
+    for (const state of this.states.values()) {
+      this.clearStateTimer(state)
+    }
+    this.states.clear()
+  }
+
+  private clearStateTimer(state: RetryState | undefined): void {
+    if (state?.timer !== null && state?.timer !== undefined) {
+      this.clearTimer(state.timer)
+      state.timer = null
+    }
+  }
+}
@@ -6,12 +6,12 @@ import { removeHost } from './host-store'

 export async function removeHostAndCloseClient(
  hostId: string,
-  closeHostClient: (hostId: string) => void
+  forgetHostClient: (hostId: string) => void
 ): Promise<void> {
  // Why: closing before the metadata commit can strand a still-paired host on
  // storage failure; closing immediately after success prevents socket leaks.
  await removeHost(hostId)
-  closeHostClient(hostId)
+  forgetHostClient(hostId)
  // Why: the notification session outlives the socket by design (it must survive
  // reconnects), so removal is the only thing that can retire it. Left behind, a
  // re-pair of the same host would inherit a watermark for a counter it never saw.
@@ -19,12 +19,10 @@ function routerFixture() {
  const router = new MobileEndpointNudgeRouter({
    logical,
    controller: { handleActiveNudge } as unknown as RelayReconnectController,
-    now: () => 20_000,
    isStopped: () => false,
    isForeground: () => foreground,
    setForeground,
    replaceRelay: vi.fn(),
-    recoverAfterDeadProbe: vi.fn(),
    scheduleDirectProbe
  })
  return { handleActiveNudge, logical, router, scheduleDirectProbe, setForeground }
@@ -38,9 +36,7 @@ describe('MobileEndpointNudgeRouter', () => {

    expect(fixture.setForeground).toHaveBeenCalledWith(true)
    expect(fixture.handleActiveNudge).toHaveBeenCalledWith(fixture.logical, 'focus')
-    expect(fixture.logical.sendRequest).toHaveBeenCalledWith('status.get', null, {
-      timeoutMs: 4000
-    })
+    expect(fixture.logical.sendRequest).not.toHaveBeenCalled()
    expect(fixture.scheduleDirectProbe).toHaveBeenCalledOnce()
  })

@@ -1,4 +1,3 @@
-import { MobileRelayFocusProbe } from './mobile-relay-focus-probe'
 import type { RelayReconnectController } from './mobile-relay-reconnect-controller'
 import type { StableLogicalRpcClient } from './stable-logical-rpc-client'
 import type { ForegroundNudgeReason } from './types'
@@ -6,28 +5,17 @@ import type { ForegroundNudgeReason } from './types'
 // Routes attention/network nudges: focus and app-resume probe a healthy relay,
 // a network change replaces it make-before-break, everything else re-enters recovery.
 export class MobileEndpointNudgeRouter {
-  private readonly focusProbe: MobileRelayFocusProbe
-
  constructor(
    private readonly args: {
      logical: StableLogicalRpcClient
      controller: RelayReconnectController
-      now: () => number
      isStopped: () => boolean
      isForeground: () => boolean
      setForeground: (foreground: boolean) => void
      replaceRelay: () => void
-      recoverAfterDeadProbe: (detail: string) => void
      scheduleDirectProbe: () => void
    }
-  ) {
-    this.focusProbe = new MobileRelayFocusProbe({
-      logical: args.logical,
-      now: args.now,
-      canProbe: () => !args.isStopped() && args.isForeground(),
-      onDead: args.recoverAfterDeadProbe
-    })
-  }
+  ) {}

  nudge(reason: ForegroundNudgeReason): void {
    const { args } = this
@@ -43,9 +31,7 @@ export class MobileEndpointNudgeRouter {
      args.setForeground(true)
    }
    const verdict = args.controller.handleActiveNudge(args.logical, reason)
-    if (verdict === 'probe') {
-      void this.focusProbe.probe()
-    } else if (verdict === 'replace') {
+    if (verdict === 'replace') {
      args.replaceRelay()
    }
    args.scheduleDirectProbe()
@@ -89,7 +89,7 @@ describe('mobile endpoint supervisor nudges', () => {
    supervisor.stop()
  })

-  it('probes a healthy relay on a focus nudge and keeps it untouched', async () => {
+  it('leaves physical relay probing to the session watchdog', async () => {
    const logical = new FakeLogicalClient('connected', 'relay')
    const deps = dependencies()
    const supervisor = new MobileEndpointSupervisor(logical, host, deps)
@@ -97,28 +97,27 @@ describe('mobile endpoint supervisor nudges', () => {

    supervisor.nudge('focus')
    await vi.advanceTimersByTimeAsync(0)
-    expect(logical.sendRequest).toHaveBeenCalledWith('status.get', null, { timeoutMs: 4000 })
+    expect(logical.sendRequest).not.toHaveBeenCalled()
    expect(logical.suspendActiveSession).not.toHaveBeenCalled()
    expect(deps.openRelay).not.toHaveBeenCalled()

-    // Repeated focus events inside the probe window coalesce into one probe.
+    // Repeated focus events never create a second supervisor-owned liveness policy.
    supervisor.nudge('focus')
    await vi.advanceTimersByTimeAsync(0)
-    expect(logical.sendRequest).toHaveBeenCalledOnce()
+    expect(logical.sendRequest).not.toHaveBeenCalled()
    supervisor.stop()
  })

-  it('suspends and re-dials when the focus probe fails', async () => {
+  it('does not suspend a relay from one focus RPC failure', async () => {
    const logical = new FakeLogicalClient('connected', 'relay')
    const deps = dependencies()
    const supervisor = new MobileEndpointSupervisor(logical, host, deps)
    await supervisor.start()

-    logical.sendRequest.mockRejectedValueOnce(new Error('relay RPC timed out: status.get'))
    supervisor.nudge('focus')
    await vi.advanceTimersByTimeAsync(0)
-    expect(logical.suspendActiveSession).toHaveBeenCalledOnce()
-    await vi.waitFor(() => expect(deps.openRelay).toHaveBeenCalledOnce())
+    expect(logical.suspendActiveSession).not.toHaveBeenCalled()
+    expect(deps.openRelay).not.toHaveBeenCalled()
    expect(logical.getState()).toBe('connected')
    supervisor.stop()
  })
@@ -60,16 +60,10 @@ export class MobileEndpointSupervisor {
    this.nudgeRouter = new MobileEndpointNudgeRouter({
      logical,
      controller: this.relayReconnect,
-      now: dependencies.now,
      isStopped: () => this.stopped,
      isForeground: () => this.foreground,
      setForeground: (foreground) => this.setForeground(foreground),
      replaceRelay: () => void this.recoverRelay(true, true),
-      recoverAfterDeadProbe: (detail) => {
-        this.logRelay('relay probe failed; recovering', detail)
-        suspendRelayIfStillConnected(this.relayReconnect, this.logical)
-        void this.recoverRelay()
-      },
      scheduleDirectProbe: () => this.directProbe.schedule(0)
    })
    this.leaseRotation = new RelayLeaseRotationTimer(dependencies, () => {
@@ -0,0 +1,62 @@
+import { describe, expect, it, vi } from 'vitest'
+
+vi.mock('./mobile-e2ee-v2-client-session', () => ({
+  MobileE2EEV2ClientSession: {
+    create: () => ({})
+  }
+}))
+
+vi.mock('./mobile-e2ee-v2-physical-channel', () => ({
+  MobileE2EEV2PhysicalChannel: class {
+    start = vi.fn()
+    handleMessage = vi.fn(async () => {})
+    sendText = vi.fn(() => true)
+    sendBinary = vi.fn(() => true)
+    dispose = vi.fn()
+  }
+}))
+
+import { MobileRelayE2eeLink } from './mobile-relay-e2ee-link'
+
+class ThrowingSocket {
+  static readonly OPEN = 1
+  readonly OPEN = ThrowingSocket.OPEN
+  readyState = ThrowingSocket.OPEN
+  bufferedAmount = 0
+  onopen: (() => void) | null = null
+  onmessage: ((event: { data: unknown }) => void) | null = null
+  onerror: (() => void) | null = null
+  onclose: ((event: { code: number }) => void) | null = null
+  send = vi.fn(() => {
+    throw new Error('relay auth write failed')
+  })
+  close = vi.fn()
+}
+
+describe('MobileRelayE2eeLink', () => {
+  it('routes the initial relay-auth write exception through link failure', () => {
+    const socket = new ThrowingSocket()
+    const onError = vi.fn()
+    new MobileRelayE2eeLink({
+      endpoint: {
+        cellUrl: 'https://relay-c1.onorca.dev',
+        relayHostId: 'AbCdEf0123_-xyZ9'
+      },
+      credential: 'credential',
+      expectedCredentialKind: 'resume',
+      deviceToken: 'device-token',
+      desktopPublicKeyB64: 'desktop-key',
+      onAuthenticated: vi.fn(),
+      onText: vi.fn(),
+      onBinary: vi.fn(),
+      onError,
+      createSocket: () => socket as unknown as WebSocket
+    })
+
+    expect(() => socket.onopen?.()).not.toThrow()
+    expect(onError).toHaveBeenCalledWith(
+      expect.objectContaining({ message: 'relay auth write failed' })
+    )
+    expect(socket.close).toHaveBeenCalledOnce()
+  })
+})
@@ -76,14 +76,18 @@ export class MobileRelayE2eeLink {

  private bindSocket(): void {
    this.socket.onopen = () => {
-      this.socket.send(
-        JSON.stringify({
-          type: 'relay-auth',
-          v: 1,
-          mode: 'connect',
-          credential: this.options.credential
-        })
-      )
+      try {
+        this.socket.send(
+          JSON.stringify({
+            type: 'relay-auth',
+            v: 1,
+            mode: 'connect',
+            credential: this.options.credential
+          })
+        )
+      } catch (error) {
+        this.fail(asError(error))
+      }
    }
    this.socket.onmessage = (event) => {
      this.inboundChain = this.inboundChain
@@ -1,53 +0,0 @@
-import {
-  isLogicalClientCutoverError,
-  type StableLogicalRpcClient
-} from './stable-logical-rpc-client'
-
-// Why: a probe must fail well before the 12s connect ceiling so a dead socket
-// still recovers in a few seconds, but outlast one slow cellular RTT.
-const PROBE_TIMEOUT_MS = 4_000
-// Why: focus fires per navigation event; one liveness answer covers them all.
-const PROBE_MIN_INTERVAL_MS = 10_000
-
-// Cheap liveness check for a focus nudge on an active relay: a healthy session
-// answers and nothing visible changes; only a dead one is handed to onDead.
-export class MobileRelayFocusProbe {
-  private lastProbeAt = 0
-
-  constructor(
-    private readonly args: {
-      logical: StableLogicalRpcClient
-      now: () => number
-      canProbe: () => boolean
-      onDead: (detail: string) => void
-    }
-  ) {}
-
-  async probe(): Promise<void> {
-    const { logical, now } = this.args
-    const at = now()
-    if (
-      !this.args.canProbe() ||
-      at - this.lastProbeAt < PROBE_MIN_INTERVAL_MS ||
-      logical.getActivePath() !== 'relay' ||
-      logical.getState() !== 'connected'
-    ) {
-      return
-    }
-    this.lastProbeAt = at
-    const generation = logical.getGeneration()
-    try {
-      await logical.sendRequest('status.get', null, { timeoutMs: PROBE_TIMEOUT_MS })
-    } catch (error) {
-      if (
-        isLogicalClientCutoverError(error) ||
-        !this.args.canProbe() ||
-        logical.getGeneration() !== generation ||
-        logical.getActivePath() !== 'relay'
-      ) {
-        return
-      }
-      this.args.onDead(String((error as Error).message).slice(0, 80))
-    }
-  }
-}
@@ -0,0 +1,173 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+const fakes = vi.hoisted(() => ({
+  linkOptions: null as null | {
+    onHello(value: unknown): void
+    onAuthenticated(): void
+    onText(value: string): void
+    onBinary(value: Uint8Array): void
+  },
+  sendText: vi.fn(() => true),
+  close: vi.fn()
+}))
+
+vi.mock('./mobile-relay-e2ee-link', () => ({
+  MobileRelayE2eeLink: class {
+    constructor(options: NonNullable<typeof fakes.linkOptions>) {
+      fakes.linkOptions = options
+    }
+    sendText = fakes.sendText
+    close = fakes.close
+  }
+}))
+
+import { connectMobileRelayRpcSession } from './mobile-relay-rpc-session'
+
+const relay = {
+  v: 1 as const,
+  directorUrl: 'https://relay.onorca.dev',
+  cellUrl: 'https://relay-c1.onorca.dev',
+  assignmentEpoch: 7,
+  relayHostId: 'AbCdEf0123_-xyZ9',
+  e2eeFraming: 2 as const
+}
+
+async function authenticateSession() {
+  const session = connectMobileRelayRpcSession({
+    relay,
+    resumeToken: 'resume-secret',
+    resumeCredentialVersion: 3,
+    resumeConfirmReqId: 'confirm-1',
+    deviceToken: 'device-token',
+    desktopPublicKeyB64: 'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=',
+    requestTimeoutMs: 30_000
+  })
+  fakes.linkOptions!.onHello({
+    type: 'relay-hello',
+    ok: true,
+    credentialKind: 'resume',
+    leaseExpiresAt: Date.now() + 60_000,
+    acceptedCredentialVersion: 3,
+    acceptedAs: 'current',
+    resumeExpiresAt: Date.now() + 300_000
+  })
+  fakes.linkOptions!.onAuthenticated()
+  await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledOnce())
+  const confirmation = sentRequests()[0]!
+  fakes.linkOptions!.onText(
+    JSON.stringify({
+      id: confirmation.id,
+      ok: true,
+      result: {
+        v: 1,
+        relay,
+        resumeConfirmation: {
+          v: 1,
+          reqId: 'confirm-1',
+          currentVersion: 3,
+          acceptedAs: 'current',
+          renewed: true,
+          resumeExpiresAt: Date.now() + 300_000
+        }
+      },
+      _meta: { runtimeId: 'runtime-1' }
+    })
+  )
+  await vi.waitFor(() => expect(session.getState()).toBe('connected'))
+  fakes.sendText.mockClear()
+  return session
+}
+
+function sentRequests(): Array<{ id: string; method: string }> {
+  return fakes.sendText.mock.calls.map(
+    ([value]) => JSON.parse(value as string) as { id: string; method: string }
+  )
+}
+
+describe('mobile relay RPC session liveness', () => {
+  beforeEach(() => {
+    vi.useFakeTimers()
+    vi.clearAllMocks()
+    fakes.linkOptions = null
+    fakes.sendText.mockReturnValue(true)
+  })
+  afterEach(() => vi.useRealTimers())
+
+  it('sends no periodic traffic while an authenticated relay is idle', async () => {
+    const session = await authenticateSession()
+
+    await vi.advanceTimersByTimeAsync(60_000)
+
+    expect(fakes.sendText).not.toHaveBeenCalled()
+    expect(session.getState()).toBe('connected')
+    session.close()
+  })
+
+  it('disconnects after two fair foreground misses', async () => {
+    const session = await authenticateSession()
+
+    session.notifyForeground('focus')
+    expect(sentRequests().map(({ method }) => method)).toEqual(['status.get'])
+    await vi.advanceTimersByTimeAsync(4_000)
+    expect(session.getState()).toBe('connected')
+    expect(fakes.sendText).toHaveBeenCalledTimes(2)
+    await vi.advanceTimersByTimeAsync(4_000)
+
+    expect(session.getState()).toBe('disconnected')
+    expect(fakes.close).toHaveBeenCalledOnce()
+  })
+
+  it('rate-limits foreground sequences without suppressing a retry', async () => {
+    const session = await authenticateSession()
+    session.notifyForeground('focus')
+    const firstProbe = sentRequests()[0]!
+    fakes.linkOptions!.onText(
+      JSON.stringify({ id: firstProbe.id, ok: true, result: {}, _meta: { runtimeId: 'r1' } })
+    )
+
+    session.notifyForeground('focus')
+    await vi.advanceTimersByTimeAsync(9_999)
+    session.notifyForeground('app-resume')
+    expect(fakes.sendText).toHaveBeenCalledOnce()
+    await vi.advanceTimersByTimeAsync(1)
+    session.notifyForeground('focus')
+
+    expect(fakes.sendText).toHaveBeenCalledTimes(2)
+    session.close()
+  })
+
+  it('does not probe the old relay on a network change', async () => {
+    const session = await authenticateSession()
+
+    session.notifyForeground('network-change')
+
+    expect(fakes.sendText).not.toHaveBeenCalled()
+    session.close()
+  })
+
+  it('does not probe when work follows prolonged inbound silence', async () => {
+    const session = await authenticateSession()
+    await vi.advanceTimersByTimeAsync(60_000)
+
+    const pending = session.sendRequest('terminal.send', { terminal: 'term', text: 'hi' })
+    const outcome = pending.catch(() => undefined)
+    session.subscribe('terminal.subscribe', { terminal: 'term' }, vi.fn())
+    await vi.advanceTimersByTimeAsync(0)
+
+    expect(sentRequests().map(({ method }) => method)).toEqual([
+      'terminal.send',
+      'terminal.subscribe'
+    ])
+    session.close()
+    await outcome
+  })
+
+  it('fails immediately when a liveness probe cannot be written', async () => {
+    const session = await authenticateSession()
+    fakes.sendText.mockReturnValue(false)
+
+    session.notifyForeground('focus')
+
+    expect(session.getState()).toBe('disconnected')
+    expect(fakes.close).toHaveBeenCalledOnce()
+  })
+})
@@ -1,4 +1,4 @@
-import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import {
  BrowserScreencastOpcode,
  encodeBrowserScreencastFrame
@@ -103,6 +103,7 @@ describe('mobile relay RPC session', () => {
    fakes.linkOptions = null
    fakes.sendText.mockReturnValue(true)
  })
+  afterEach(() => vi.useRealTimers())

  it('requires exact resume observations and confirms by request ID before becoming connected', async () => {
    const { session, confirmationRequest } = await authenticateSession()
@@ -9,9 +9,14 @@ import { MobileE2EEAuthenticationError } from './mobile-e2ee-v2-physical-channel
 import { markRpcDeliveryUnknown } from './rpc-delivery-ambiguity'
 import { openRpcRequestBudget, resolvePostConnectRequestTimeout } from './rpc-request-budget'
 import { isRpcResponse } from './rpc-response-shape'
+import { RpcSessionLivenessWatchdog } from './rpc-session-liveness-watchdog'
 import type { RpcClient } from './rpc-client'
 import type { ConnectionState, RpcResponse } from './types'

+const RELAY_PROBE_TIMEOUT_MS = 4_000
+const RELAY_MISSED_PROBE_LIMIT = 2
+const RELAY_FOREGROUND_PROBE_MIN_INTERVAL_MS = 10_000
+
 type PendingRequest = {
  resolve: (response: RpcResponse) => void
  reject: (error: Error) => void
@@ -48,6 +53,7 @@ export function connectMobileRelayRpcSession(args: {
  let resumeConfirmation: DeviceResumeConfirmed | null = null
  let failure: Error | null = null
  let closed = false
+  const livenessIdentity = {}
  const streams = new MobileRelayRpcStreams({
    nextId,
    sendFrame,
@@ -74,8 +80,14 @@ export function connectMobileRelayRpcSession(args: {
      publishState('handshaking')
    },
    onAuthenticated: () => void confirmResume(),
-    onText: handleText,
-    onBinary: handleBinary,
+    onText: (plaintext) => {
+      livenessWatchdog.noteAuthenticatedInbound(livenessIdentity)
+      handleText(plaintext)
+    },
+    onBinary: (plaintext) => {
+      livenessWatchdog.noteAuthenticatedInbound(livenessIdentity)
+      handleBinary(plaintext)
+    },
    onError: fail
  })

@@ -103,12 +115,17 @@ export function connectMobileRelayRpcSession(args: {
      stateListeners.add(listener)
      return () => stateListeners.delete(listener)
    },
-    notifyForeground: () => {},
+    notifyForeground: (reason) => {
+      if (state === 'connected' && reason !== 'network-change') {
+        livenessWatchdog.probeNow(livenessIdentity)
+      }
+    },
    close() {
      if (closed) {
        return
      }
      closed = true
+      livenessWatchdog.stop(livenessIdentity)
      link.close()
      rejectPending(new Error('Client closed'))
      streams.clear()
@@ -119,6 +136,16 @@ export function connectMobileRelayRpcSession(args: {
    getResumeConfirmation: () => resumeConfirmation,
    getFailure: () => failure
  }
+  const livenessWatchdog = new RpcSessionLivenessWatchdog({
+    transport: 'relay',
+    idleProbeMs: null,
+    probeTimeoutMs: RELAY_PROBE_TIMEOUT_MS,
+    missedProbeLimit: RELAY_MISSED_PROBE_LIMIT,
+    voluntaryProbeMinIntervalMs: RELAY_FOREGROUND_PROBE_MIN_INTERVAL_MS,
+    sendProbe: () =>
+      state === 'connected' && sendFrame({ id: nextId(), method: 'status.get', params: undefined }),
+    terminate: () => fail(new Error('relay session liveness timeout'))
+  })
  return client

  async function confirmResume(): Promise<void> {
@@ -139,6 +166,7 @@ export function connectMobileRelayRpcSession(args: {
      resumeConfirmation = result.resumeConfirmation
      resumeExpiresAt = result.resumeConfirmation.resumeExpiresAt
      lastConnectedAt = Date.now()
+      livenessWatchdog.start(livenessIdentity)
      publishState('connected')
    } catch (error) {
      fail(asError(error))
@@ -242,6 +270,7 @@ export function connectMobileRelayRpcSession(args: {
    }
    closed = true
    failure = error
+    livenessWatchdog.stop(livenessIdentity)
    link.close()
    rejectPending(error)
    publishState(error instanceof MobileE2EEAuthenticationError ? 'auth-failed' : 'disconnected')
@@ -57,8 +57,10 @@ export class MobileRelayRpcStreams {
    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) {
+          if (!this.options.sendFrame({ id, method, params: stream.params })) {
+            this.fail(id, stream, 'Connection interrupted')
+          }
        }
      })
      .catch((error: unknown) => {
@@ -114,8 +116,7 @@ export class MobileRelayRpcStreams {
    }
    handleTerminalBinaryFrame(bytes, {
      terminalSnapshots: this.terminalSnapshots,
-      getListener: (streamId) => this.terminalListeners.get(streamId),
-      recordValidatedInboundTraffic: () => {}
+      getListener: (streamId) => this.terminalListeners.get(streamId)
    })
  }

@@ -1,87 +0,0 @@
-import type { ConnectionState } from './types'
-
-// Why: RN auto-pongs pings natively, so JS needs an app-level probe to detect half-open sockets.
-const ACTIVITY_PROBE_INTERVAL_MS = 20_000
-const ACTIVITY_PROBE_TIMEOUT_MS = 8_000
-
-export type RpcActivityProbeDependencies = {
-  getState: () => ConnectionState
-  getSocket: () => WebSocket | null
-  // Counts only validated inbound traffic, so a malformed frame cannot pass for liveness.
-  getInboundSequence: () => number
-  nextId: () => string
-  registerPending: (id: string, onSettled: () => void) => void
-  clearPending: (id: string) => void
-  sendProbe: (id: string) => boolean
-  forceReconnect: (socket: WebSocket) => void
-}
-
-export type RpcActivityProbe = {
-  start: () => void
-  stop: () => void
-  run: () => void
-}
-
-export function createRpcActivityProbe(
-  dependencies: RpcActivityProbeDependencies
-): RpcActivityProbe {
-  let timer: ReturnType<typeof setInterval> | null = null
-  let inFlight = false
-
-  function stop(): void {
-    if (timer) {
-      clearInterval(timer)
-      timer = null
-    }
-  }
-
-  function run(): void {
-    const probeWs = dependencies.getSocket()
-    if (dependencies.getState() !== 'connected' || !probeWs || inFlight) {
-      return
-    }
-    inFlight = true
-    const id = dependencies.nextId()
-    const probeInboundSequence = dependencies.getInboundSequence()
-    let timedOut = false
-    const timeout = setTimeout(() => {
-      timedOut = true
-      inFlight = false
-      dependencies.clearPending(id)
-      // Why: any validated frame answered for the socket, even if this probe did not.
-      if (dependencies.getInboundSequence() > probeInboundSequence) {
-        return
-      }
-      console.log('[net] activity-probe TIMEOUT — forcing reconnect', {
-        state: dependencies.getState()
-      })
-      // Why: stale probe timers must not close a replacement socket.
-      if (probeWs === dependencies.getSocket() && probeWs.readyState === WebSocket.OPEN) {
-        dependencies.forceReconnect(probeWs)
-      }
-    }, ACTIVITY_PROBE_TIMEOUT_MS)
-
-    dependencies.registerPending(id, () => {
-      if (timedOut) {
-        return
-      }
-      inFlight = false
-      clearTimeout(timeout)
-    })
-
-    if (!dependencies.sendProbe(id)) {
-      inFlight = false
-      clearTimeout(timeout)
-      dependencies.clearPending(id)
-    }
-  }
-
-  return {
-    run,
-    stop,
-    start(): void {
-      stop()
-      timer = setInterval(run, ACTIVITY_PROBE_INTERVAL_MS)
-    }
-  }
-}
@@ -0,0 +1,28 @@
+import type { HostClientAcquisition } from './host-client-acquisition-registry'
+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,
+    acquisition: HostClientAcquisition,
+    host?: HostProfile
+  ) => RpcClient | null
+  release: (hostId: string, acquisition: HostClientAcquisition) => void
+  releaseAndCloseIfUnused: (hostId: string, acquisition: HostClientAcquisition) => void
+  closeIfUnused: (hostId: string) => void
+  forceReconnect: (hostId: string) => Promise<void>
+  refreshHostClient: (hostId: string) => void
+  forgetHostClient: (hostId: string) => void
+  disconnectHostClient: (hostId: string) => void
+  getState: (hostId: string) => ConnectionState
+  getKnownState: (hostId: string) => ConnectionState | null
+  getReconnectAttempt: (hostId: string) => number
+  getLastConnectedAt: (hostId: string) => number | null
+  getActivePath: (hostId: string) => MobileConnectionPath
+  subscribeHostState: (hostId: string, listener: (state: ConnectionState) => void) => () => void
+  getAllClients: () => { hostId: string; client: RpcClient }[]
+  subscribeAllHosts: (listener: () => void) => () => void
+  primeHosts: (hosts: HostProfile[]) => void
+}
@@ -151,14 +151,13 @@ describe.runIf(RUN_LIVE)('live foreground recovery (issue #5049)', () => {
      // comes back to the foreground.
      blackhole = true
      c.notifyForeground()
-      // Foreground probe budget is 8s; the interval probe alone would take
-      // up to 28s. Allow scheduling slack but stay well under 28s.
+      // Three fair probe windows tolerate transient mobile/Tailscale stalls.
      const detectMs = await waitFor(
        'half-open detected',
-        15_000,
+        32_000,
        () => c.getState() !== 'connected'
      )
-      expect(detectMs).toBeLessThan(12_000)
+      expect(detectMs).toBeLessThan(30_000)

      blackhole = false
      await waitFor('recovered after link healed', 15_000, () => c.getState() === 'connected')
@@ -18,18 +18,15 @@ function encodeFrame(opcode: TerminalStreamOpcode, streamId: number, payload: un
 describe('handleTerminalBinaryFrame', () => {
  it('routes terminal metadata frames to the stream listener', () => {
    const listener = vi.fn()
-    const recordValidatedInboundTraffic = vi.fn()

    handleTerminalBinaryFrame(
      encodeFrame(TerminalStreamOpcode.Metadata, 42, { cwd: '/repo/src' }),
      {
        terminalSnapshots: new Map(),
-        getListener: (streamId) => (streamId === 42 ? listener : undefined),
-        recordValidatedInboundTraffic
+        getListener: (streamId) => (streamId === 42 ? listener : undefined)
      }
    )

-    expect(recordValidatedInboundTraffic).toHaveBeenCalledTimes(1)
    expect(listener).toHaveBeenCalledWith({ type: 'metadata', streamId: 42, cwd: '/repo/src' })
  })
 })
@@ -16,7 +16,6 @@ type StreamingListener = (result: unknown) => void
 type TerminalBinaryFrameOptions = {
  terminalSnapshots: Map<number, TerminalSnapshotState>
  getListener: (streamId: number) => StreamingListener | undefined
-  recordValidatedInboundTraffic: () => void
 }

 export function handleTerminalBinaryFrame(
@@ -29,11 +28,9 @@ export function handleTerminalBinaryFrame(
  }
  const listener = options.getListener(frame.streamId)
  if (!listener) {
-    options.recordValidatedInboundTraffic()
    return
  }
  if (frame.opcode === TerminalStreamOpcode.Output) {
-    options.recordValidatedInboundTraffic()
    listener({
      type: 'data',
      streamId: frame.streamId,
@@ -46,7 +43,6 @@ export function handleTerminalBinaryFrame(
    if (!meta) {
      return
    }
-    options.recordValidatedInboundTraffic()
    options.terminalSnapshots.set(frame.streamId, {
      streamId: frame.streamId,
      meta,
@@ -55,7 +51,6 @@ export function handleTerminalBinaryFrame(
    return
  }
  if (frame.opcode === TerminalStreamOpcode.SnapshotChunk) {
-    options.recordValidatedInboundTraffic()
    const snapshot = options.terminalSnapshots.get(frame.streamId)
    if (!snapshot) {
      return
@@ -64,7 +59,6 @@ export function handleTerminalBinaryFrame(
    return
  }
  if (frame.opcode === TerminalStreamOpcode.SnapshotEnd) {
-    options.recordValidatedInboundTraffic()
    const snapshot = options.terminalSnapshots.get(frame.streamId)
    if (!snapshot) {
      return
@@ -84,7 +78,6 @@ export function handleTerminalBinaryFrame(
    if (!meta) {
      return
    }
-    options.recordValidatedInboundTraffic()
    listener({
      ...meta,
      type: 'resized',
@@ -97,7 +90,6 @@ export function handleTerminalBinaryFrame(
    if (!meta) {
      return
    }
-    options.recordValidatedInboundTraffic()
    listener({
      ...meta,
      type: 'metadata',
@@ -106,7 +98,6 @@ export function handleTerminalBinaryFrame(
    return
  }
  if (frame.opcode === TerminalStreamOpcode.Error) {
-    options.recordValidatedInboundTraffic()
    listener({
      type: 'error',
      streamId: frame.streamId,
@@ -687,7 +687,7 @@ describe('mobile rpc-client connection timeout', () => {
      client.close()
    })

-    it('reaps a half-open socket within 8s of foreground', async () => {
+    it('reaps a half-open socket after three fair foreground probe windows', async () => {
      const client = connect('ws://desktop.invalid', 'token', 'server-key')
      const socket = mockSockets[0]!
      openAndAuthenticate(socket)
@@ -696,9 +696,8 @@ describe('mobile rpc-client connection timeout', () => {
      client.notifyForeground()
      expect(sentRequests(socket, 'status.get')).toHaveLength(1)

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

      await vi.advanceTimersByTimeAsync(500)
      openAndAuthenticate(mockSockets[mockSockets.length - 1]!)
@@ -714,10 +713,9 @@ describe('mobile rpc-client connection timeout', () => {
      socket.emitCloseOnClose = false

      client.notifyForeground()
-      await vi.advanceTimersByTimeAsync(8_000)
+      await vi.advanceTimersByTimeAsync(24_000)

      expect(socket.close).toHaveBeenCalledTimes(1)
-      expect(client.getState()).toBe('reconnecting')
      socket.onclose?.()
      expect(client.getState()).toBe('reconnecting')

@@ -740,7 +738,8 @@ describe('mobile rpc-client connection timeout', () => {
      client.notifyForeground()
      client.notifyForeground()
      expect(sentRequests(socket, 'status.get')).toHaveLength(1)
-      await vi.advanceTimersByTimeAsync(8_000)
+      await vi.advanceTimersByTimeAsync(24_000)
+      expect(sentRequests(socket, 'status.get')).toHaveLength(3)
      expect(socket.close).toHaveBeenCalledTimes(1)
      expect(client.getState()).toBe('reconnecting')

@@ -748,6 +747,7 @@ describe('mobile rpc-client connection timeout', () => {
      expect(mockSockets).toHaveLength(2)
      client.close()
    })
+
    it('keeps a healthy connection when the foreground probe is answered', async () => {
      const { client, socket } = connectAuthenticated()

@@ -782,13 +782,14 @@ describe('mobile rpc-client connection timeout', () => {

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

      await vi.advanceTimersByTimeAsync(12_000)
      expect(sentRequests(socket, 'status.get')).toHaveLength(2)
      await vi.advanceTimersByTimeAsync(7_999)
      expect(socket.close).not.toHaveBeenCalled()
      await vi.advanceTimersByTimeAsync(1)
+      expect(socket.close).not.toHaveBeenCalled()
+      await vi.advanceTimersByTimeAsync(16_000)
      expect(socket.close).toHaveBeenCalled()
      expect(client.getState()).toBe('reconnecting')

@@ -853,18 +854,29 @@ describe('mobile rpc-client connection timeout', () => {
      client.close()
    })

-    it('does not count malformed or undecryptable inbound payloads as probe activity', async () => {
+    it('counts authenticated unknown payloads before semantic decoding', async () => {
      const { client, socket } = connectAuthenticated()

      client.notifyForeground()
-      socket.receive('undecryptable')
      socket.receive('encrypted:{"unexpected":true}')
      socket.receive('encrypted:{"id":"rpc-incomplete","ok":true}')
      socket.receive(new Uint8Array([0xff, 0x00, 0x01]))

      await vi.advanceTimersByTimeAsync(8_000)
+      expect(socket.close).not.toHaveBeenCalled()
+      expect(client.getState()).toBe('connected')
+
+      client.close()
+    })
+
+    it('does not count undecryptable payloads as probe activity', async () => {
+      const { client, socket } = connectAuthenticated()
+
+      client.notifyForeground()
+      socket.receive('undecryptable')
+
+      await vi.advanceTimersByTimeAsync(24_000)
      expect(socket.close).toHaveBeenCalled()
-      expect(client.getState()).toBe('reconnecting')

      client.close()
    })
@@ -37,7 +37,14 @@ import {
 import { markRpcDeliveryUnknown } from './rpc-delivery-ambiguity'
 import { openRpcRequestBudget, resolvePostConnectRequestTimeout } from './rpc-request-budget'
 import { isRpcResponse } from './rpc-response-shape'
-import { createRpcActivityProbe } from './rpc-client-activity-probe'
+import {
+  isStreamingSubscriptionReadyResult,
+  isTerminalSubscribedResult
+} from './rpc-subscription-result-shapes'
+import {
+  RpcSessionLivenessWatchdog,
+  type RpcSessionIdentity
+} from './rpc-session-liveness-watchdog'
 import { isStaleForegroundDial } from './rpc-stale-dial'
 import { websocketPayloadToUint8 } from './websocket-payload-bytes'

@@ -129,6 +136,7 @@ const CONNECT_TIMEOUT_MS = 12_000
 const HANDSHAKE_TIMEOUT_MS = 5_000
 // Why: RN may not expose WebSocket.readyState constants, but the CONNECTING protocol value (0) is stable across runtimes.
 const WEBSOCKET_CONNECTING_STATE = 0
+const LIVENESS_REQUEST_ID_PREFIX = 'mobile-liveness-'

 export type ConnectOptions = {
  onStateChange?: (state: ConnectionState) => void
@@ -177,7 +185,7 @@ export function connect(
  let lastConnectedAt: number | null = null
  // Why: cheap diagnostics for RN/OkHttp process-state poisoning (retry cadence, inbound traffic, close timing).
  let lastInboundAt: number | null = null
-  let inboundSequence = 0
+  let livenessIdentity: (RpcSessionIdentity & { socket: WebSocket }) | null = null
  let lastWsClosedAt: number | null = null
  let wsConstructionCounter = 0
  let dialStartedAt = 0
@@ -315,6 +323,7 @@ export function connect(
    const openingWs = ws
    let openingWsAuthenticated = false
    let openingWsLastInboundAt: number | null = null
+    let openingLivenessIdentity: (RpcSessionIdentity & { socket: WebSocket }) | null = null

    // Why: RN can leave opens pending forever on flaky handoffs — force reconnect if onopen never arrives.
    connectTimer = setTimeout(() => {
@@ -351,7 +360,12 @@ export function connect(
        type: 'e2ee_hello',
        publicKeyB64: publicKeyToBase64(ephemeral.publicKey)
      })
-      openingWs.send(hello)
+      try {
+        openingWs.send(hello)
+      } catch {
+        closeAndSynthesize(openingWs)
+        return
+      }
      emitLog('info', 'Sent e2ee_hello', 'Awaiting server e2ee_ready')

      sharedKey = deriveSharedKey(ephemeral.secretKey, serverPublicKey)
@@ -419,9 +433,11 @@ export function connect(
              streamCount: streamListeners.size
            })
            openingWsAuthenticated = true
+            openingLivenessIdentity = { socket: openingWs }
+            livenessIdentity = openingLivenessIdentity
+            livenessWatchdog.start(openingLivenessIdentity)
            setState('connected')
            emitLog('success', 'Authenticated', 'Channel ready for RPC')
-            activityProbe.start()
            for (const [id, stream] of streamListeners) {
              if (stream.cancelled) {
                removeStreamListener(id)
@@ -441,8 +457,9 @@ export function connect(
              ) {
                stream.sent = true
              } else {
-                emitStreamError(stream, 'Connection interrupted')
-                removeStreamListener(id)
+                // The failed write already starts recovery; retain every stream for replay.
+                markStreamsForReplay()
+                break
              }
            }
          } else if (msg.type === 'e2ee_error' || (!msg.ok && msg.error?.code === 'unauthorized')) {
@@ -473,6 +490,9 @@ export function connect(
        if (!plaintextBytes) {
          return
        }
+        if (openingLivenessIdentity) {
+          livenessWatchdog.noteAuthenticatedInbound(openingLivenessIdentity)
+        }
        handleBinaryFrame(plaintextBytes)
        return
      }
@@ -481,6 +501,9 @@ export function connect(
      if (plaintext === null) {
        return
      }
+      if (openingLivenessIdentity) {
+        livenessWatchdog.noteAuthenticatedInbound(openingLivenessIdentity)
+      }

      let response: unknown
      try {
@@ -491,8 +514,9 @@ export function connect(
      if (!isRpcResponse(response)) {
        return
      }
-      recordValidatedInboundTraffic()
-
+      if (response.id.startsWith(LIVENESS_REQUEST_ID_PREFIX)) {
+        return
+      }
      // Why: a mid-session unauthorized may be transient (issue #5200) — handleAuthRejection retries before latching auth-failed.
      if (!response.ok && response.error.code === 'unauthorized') {
        handleAuthRejection('Unauthorized — pairing may be revoked')
@@ -641,7 +665,10 @@ export function connect(
    pendingBrowserScreencastRequestId = null
    markStreamsForReplay()
    clearHandshakeTimer()
-    activityProbe.stop()
+    if (livenessIdentity?.socket === closedWs) {
+      livenessWatchdog.stop(livenessIdentity)
+      livenessIdentity = null
+    }
    if (intentionallyClosed) {
      console.log('[net] handleSocketClosed — intentional close')
      setState('disconnected')
@@ -672,6 +699,10 @@ export function connect(

  // Why: an auth rejection may be transient (issue #5200) — retry up to AUTH_RETRY_BUDGET times before latching auth-failed.
  function handleAuthRejection(reason: string, preserveRecovery = false): void {
+    if (livenessIdentity) {
+      livenessWatchdog.stop(livenessIdentity)
+      livenessIdentity = null
+    }
    authRejectionCount++
    if (authRejectionCount < AUTH_RETRY_BUDGET) {
      console.log('[net] auth rejected — retrying handshake', {
@@ -877,21 +908,15 @@ export function connect(
    }
  }

-  function recordValidatedInboundTraffic(): void {
-    inboundSequence++
-  }
-
  function handleBinaryFrame(bytes: Uint8Array): void {
    const browserFrame = decodeBrowserScreencastFrame(bytes)
    if (browserFrame) {
-      recordValidatedInboundTraffic()
      handleBrowserBinaryFrame(browserFrame)
      return
    }
    handleTerminalBinaryFrame(bytes, {
      terminalSnapshots,
-      getListener: (streamId) => terminalStreamListeners.get(streamId),
-      recordValidatedInboundTraffic
+      getListener: (streamId) => terminalStreamListeners.get(streamId)
    })
  }

@@ -908,8 +933,16 @@ export function connect(

  function sendEncrypted(request: unknown): boolean {
    if (ws && ws.readyState === WebSocket.OPEN && sharedKey) {
-      ws.send(encrypt(JSON.stringify(request), sharedKey))
-      return true
+      const sendingWs = ws
+      try {
+        sendingWs.send(encrypt(JSON.stringify(request), sharedKey))
+        return true
+      } catch {
+        if (ws === sendingWs) {
+          closeAndSynthesize(sendingWs)
+        }
+        return false
+      }
    }
    console.log('[net] sendEncrypted FAILED — channel not ready', {
      hasWs: !!ws,
@@ -955,15 +988,23 @@ export function connect(
    }
  }

-  const activityProbe = createRpcActivityProbe({
-    getState: () => state,
-    getSocket: () => ws,
-    getInboundSequence: () => inboundSequence,
-    nextId,
-    registerPending: (id, onSettled) => pending.set(id, { resolve: onSettled, reject: onSettled }),
-    clearPending: (id) => pending.delete(id),
-    sendProbe: (id) => sendEncrypted({ id, deviceToken, method: 'status.get' }),
-    forceReconnect: closeAndSynthesize
+  const livenessWatchdog = new RpcSessionLivenessWatchdog({
+    transport: 'direct',
+    sendProbe: (identity) => {
+      if (identity !== livenessIdentity || state !== 'connected') {
+        return false
+      }
+      return sendEncrypted({
+        id: `${LIVENESS_REQUEST_ID_PREFIX}${nextId()}`,
+        deviceToken,
+        method: 'status.get'
+      })
+    },
+    terminate: (identity) => {
+      if (identity === livenessIdentity && livenessIdentity.socket === ws) {
+        closeAndSynthesize(livenessIdentity.socket)
+      }
+    }
  })

  openConnection()
@@ -1119,10 +1160,11 @@ export function connect(
        return
      }
      if (state === 'connected') {
-        // Why: OS can kill the TCP path while backgrounded without onclose; probe now to detect the half-open socket in ≤8s (issue #5049).
+        // Why: resume probes now; three fair misses detect a half-open socket within 24s.
        console.log('[net] foreground — probing live connection')
-        activityProbe.start()
-        activityProbe.run()
+        if (livenessIdentity) {
+          livenessWatchdog.probeNow(livenessIdentity)
+        }
        return
      }
      const dialing = ws
@@ -1157,7 +1199,10 @@ export function connect(
      }
      clearConnectTimer()
      clearHandshakeTimer()
-      activityProbe.stop()
+      if (livenessIdentity) {
+        livenessWatchdog.stop(livenessIdentity)
+        livenessIdentity = null
+      }
      if (ws) {
        ws.close()
        ws = null
@@ -1169,25 +1214,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,188 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { connect } from './rpc-client'
+import { encodeTerminalStreamFrame, TerminalStreamOpcode } from './terminal-stream-protocol'
+
+vi.mock('./e2ee', () => ({
+  generateKeyPair: () => ({
+    publicKey: new Uint8Array(32),
+    secretKey: new Uint8Array(32)
+  }),
+  deriveSharedKey: () => new Uint8Array(32),
+  publicKeyFromBase64: () => new Uint8Array(32),
+  publicKeyToBase64: () => 'client-public-key',
+  encrypt: (plaintext: string) => `encrypted:${plaintext}`,
+  decrypt: (raw: string) => raw.replace(/^encrypted:/, ''),
+  decryptBytes: (bytes: Uint8Array) => bytes
+}))
+
+class MockWebSocket {
+  static readonly CONNECTING = 0
+  static readonly OPEN = 1
+  static readonly CLOSED = 3
+
+  readonly CONNECTING = MockWebSocket.CONNECTING
+  readonly OPEN = MockWebSocket.OPEN
+  readonly CLOSED = MockWebSocket.CLOSED
+  readyState = MockWebSocket.CONNECTING
+  onopen: (() => void) | null = null
+  onclose: (() => void) | null = null
+  onmessage: ((event: { data: unknown }) => void) | null = null
+  onerror: (() => void) | null = null
+  throwOnSend = false
+  throwForMethod: string | null = null
+  readonly sent: string[] = []
+  close = vi.fn(() => {
+    this.readyState = MockWebSocket.CLOSED
+    this.onclose?.()
+  })
+
+  constructor() {
+    sockets.push(this)
+  }
+
+  send(payload: string): void {
+    if (
+      this.throwOnSend ||
+      (this.throwForMethod && payload.includes(`"method":"${this.throwForMethod}"`))
+    ) {
+      throw new Error('socket send failed')
+    }
+    this.sent.push(payload)
+  }
+
+  authenticate(): void {
+    this.readyState = MockWebSocket.OPEN
+    this.onopen?.()
+    this.onmessage?.({ data: JSON.stringify({ type: 'e2ee_ready' }) })
+    this.onmessage?.({ data: 'encrypted:{"type":"e2ee_authenticated"}' })
+  }
+}
+
+const sockets: MockWebSocket[] = []
+const originalWebSocket = globalThis.WebSocket
+
+beforeEach(() => {
+  vi.useFakeTimers()
+  sockets.length = 0
+  globalThis.WebSocket = MockWebSocket as unknown as typeof WebSocket
+})
+
+afterEach(() => {
+  vi.useRealTimers()
+  globalThis.WebSocket = originalWebSocket
+})
+
+describe('physical session liveness', () => {
+  it('recovers when the initial handshake write races socket teardown', () => {
+    const client = connect('ws://desktop.invalid', 'token', 'server-key')
+    const socket = sockets[0]!
+    socket.throwOnSend = true
+
+    try {
+      expect(() => socket.authenticate()).not.toThrow()
+      expect(socket.close).toHaveBeenCalledOnce()
+      expect(client.getState()).toBe('reconnecting')
+    } finally {
+      client.close()
+    }
+  })
+
+  it('tolerates the first fair silent foreground-probe window', async () => {
+    const client = connect('ws://desktop.invalid', 'token', 'server-key')
+    const socket = sockets[0]!
+    socket.authenticate()
+
+    try {
+      client.notifyForeground()
+      await vi.advanceTimersByTimeAsync(8_000)
+
+      expect(socket.close).not.toHaveBeenCalled()
+      expect(client.getState()).toBe('connected')
+    } finally {
+      client.close()
+    }
+  })
+
+  it('counts authenticated terminal binary output as liveness', async () => {
+    const client = connect('ws://desktop.invalid', 'token', 'server-key')
+    const socket = sockets[0]!
+    socket.authenticate()
+    client.notifyForeground()
+    socket.onmessage?.({
+      data: encodeTerminalStreamFrame({
+        opcode: TerminalStreamOpcode.Output,
+        streamId: 42,
+        seq: 1,
+        payload: new TextEncoder().encode('hello')
+      })
+    })
+    await Promise.resolve()
+    await vi.advanceTimersByTimeAsync(8_000)
+
+    expect(socket.close).not.toHaveBeenCalled()
+    expect(client.getState()).toBe('connected')
+    client.close()
+  })
+
+  it('turns a probe write exception into session recovery', () => {
+    const client = connect('ws://desktop.invalid', 'token', 'server-key')
+    const socket = sockets[0]!
+    socket.authenticate()
+    socket.throwOnSend = true
+
+    try {
+      expect(() => client.notifyForeground()).not.toThrow()
+      expect(socket.close).toHaveBeenCalledOnce()
+      expect(client.getState()).toBe('reconnecting')
+    } finally {
+      client.close()
+    }
+  })
+
+  it('retains every queued stream when replay write fails', async () => {
+    const client = connect('ws://desktop.invalid', 'token', 'server-key')
+    const first = sockets[0]!
+    const disposeFirst = client.subscribe('terminal.subscribe', { terminal: 'term-1' }, vi.fn())
+    const disposeSecond = client.subscribe('terminal.subscribe', { terminal: 'term-2' }, vi.fn())
+    first.throwForMethod = 'terminal.subscribe'
+
+    first.authenticate()
+    expect(client.getState()).toBe('reconnecting')
+    await vi.advanceTimersByTimeAsync(500)
+
+    const replacement = sockets[1]!
+    replacement.authenticate()
+    expect(
+      replacement.sent.filter((payload) => payload.includes('terminal.subscribe'))
+    ).toHaveLength(2)
+
+    disposeFirst()
+    disposeSecond()
+    client.close()
+  })
+
+  it('does not let probe replies consume the authentication retry budget', () => {
+    const client = connect('ws://desktop.invalid', 'token', 'server-key')
+    const socket = sockets[0]!
+    socket.authenticate()
+
+    for (let index = 0; index < 3; index++) {
+      client.notifyForeground()
+      const request = socket.sent
+        .map((payload) => payload.replace(/^encrypted:/, ''))
+        .map((payload) => JSON.parse(payload) as { id?: string; method?: string })
+        .findLast((payload) => payload.method === 'status.get')
+      socket.onmessage?.({
+        data: `encrypted:${JSON.stringify({
+          id: request?.id,
+          ok: false,
+          error: { code: 'unauthorized', message: 'Unauthorized' }
+        })}`
+      })
+    }
+
+    expect(client.getState()).toBe('connected')
+    expect(sockets).toHaveLength(1)
+    client.close()
+  })
+})
@@ -0,0 +1,145 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import {
+  LIVENESS_IDLE_MS,
+  LIVENESS_PROBE_TIMEOUT_MS,
+  RpcSessionLivenessWatchdog
+} from './rpc-session-liveness-watchdog'
+
+describe('RpcSessionLivenessWatchdog', () => {
+  beforeEach(() => vi.useFakeTimers())
+  afterEach(() => vi.useRealTimers())
+
+  function fixture() {
+    const sendProbe = vi.fn(() => true)
+    const terminate = vi.fn()
+    const watchdog = new RpcSessionLivenessWatchdog({
+      transport: 'direct',
+      sendProbe,
+      terminate,
+      now: Date.now
+    })
+    const identity = {}
+    watchdog.start(identity)
+    return { identity, sendProbe, terminate, watchdog }
+  }
+
+  it('probes only after authenticated-inbound idle', async () => {
+    const { identity, sendProbe, watchdog } = fixture()
+    await vi.advanceTimersByTimeAsync(LIVENESS_IDLE_MS - 1)
+    expect(sendProbe).not.toHaveBeenCalled()
+
+    watchdog.noteAuthenticatedInbound(identity)
+    await vi.advanceTimersByTimeAsync(LIVENESS_IDLE_MS - 1)
+    expect(sendProbe).not.toHaveBeenCalled()
+    await vi.advanceTimersByTimeAsync(1)
+    expect(sendProbe).toHaveBeenCalledOnce()
+  })
+
+  it('requires three fair consecutive misses', async () => {
+    const { identity, terminate, watchdog } = fixture()
+    watchdog.probeNow(identity)
+
+    await vi.advanceTimersByTimeAsync(LIVENESS_PROBE_TIMEOUT_MS * 2)
+    expect(terminate).not.toHaveBeenCalled()
+    await vi.advanceTimersByTimeAsync(LIVENESS_PROBE_TIMEOUT_MS)
+    expect(terminate).toHaveBeenCalledOnce()
+    expect(terminate).toHaveBeenCalledWith(identity)
+  })
+
+  it('authenticated activity resets suspicion', async () => {
+    const { identity, terminate, watchdog } = fixture()
+    watchdog.probeNow(identity)
+    await vi.advanceTimersByTimeAsync(LIVENESS_PROBE_TIMEOUT_MS)
+    watchdog.noteAuthenticatedInbound(identity)
+    await vi.advanceTimersByTimeAsync(LIVENESS_IDLE_MS + LIVENESS_PROBE_TIMEOUT_MS * 2)
+    expect(terminate).not.toHaveBeenCalled()
+  })
+
+  it('does not churn timers during continuous authenticated traffic', async () => {
+    const setTimer = vi.fn(setTimeout)
+    const sendProbe = vi.fn(() => true)
+    const watchdog = new RpcSessionLivenessWatchdog({
+      transport: 'direct',
+      sendProbe,
+      terminate: vi.fn(),
+      now: Date.now,
+      setTimer
+    })
+    const identity = {}
+    watchdog.start(identity)
+
+    await vi.advanceTimersByTimeAsync(10_000)
+    for (let index = 0; index < 100; index++) {
+      watchdog.noteAuthenticatedInbound(identity)
+    }
+    expect(setTimer).toHaveBeenCalledOnce()
+
+    await vi.advanceTimersByTimeAsync(10_000)
+    expect(sendProbe).not.toHaveBeenCalled()
+    expect(setTimer).toHaveBeenCalledTimes(2)
+  })
+
+  it('does not charge a scheduler-stalled probe window', () => {
+    let now = 0
+    let callback: (() => void) | null = null
+    const terminate = vi.fn()
+    const watchdog = new RpcSessionLivenessWatchdog({
+      transport: 'relay',
+      idleProbeMs: null,
+      probeTimeoutMs: 4_000,
+      missedProbeLimit: 2,
+      sendProbe: () => true,
+      terminate,
+      now: () => now,
+      setTimer: (next) => {
+        callback = next
+        return 1 as unknown as ReturnType<typeof setTimeout>
+      },
+      clearTimer: () => {}
+    })
+    const identity = {}
+    watchdog.start(identity)
+    watchdog.probeNow(identity)
+    now = 8_000
+    callback?.()
+    now += 4_000
+    callback?.()
+    expect(terminate).not.toHaveBeenCalled()
+  })
+
+  it('invalidates late callbacks after identity replacement', () => {
+    const callbacks: (() => void)[] = []
+    const terminate = vi.fn()
+    const watchdog = new RpcSessionLivenessWatchdog({
+      transport: 'direct',
+      sendProbe: () => true,
+      terminate,
+      now: () => 0,
+      setTimer: (callback) => {
+        callbacks.push(callback)
+        return callbacks.length as unknown as ReturnType<typeof setTimeout>
+      },
+      clearTimer: () => {}
+    })
+    const first = {}
+    const replacement = {}
+    watchdog.start(first)
+    watchdog.probeNow(first)
+    watchdog.start(replacement)
+    callbacks.forEach((callback) => callback())
+    expect(terminate).not.toHaveBeenCalled()
+  })
+
+  it('terminates immediately when a probe cannot be written', () => {
+    const terminate = vi.fn()
+    const identity = {}
+    const watchdog = new RpcSessionLivenessWatchdog({
+      transport: 'direct',
+      sendProbe: () => false,
+      terminate
+    })
+    watchdog.start(identity)
+    watchdog.probeNow(identity)
+    expect(terminate).toHaveBeenCalledWith(identity)
+  })
+})
@@ -0,0 +1,189 @@
+export const LIVENESS_IDLE_MS = 20_000
+export const LIVENESS_PROBE_TIMEOUT_MS = 8_000
+export const MISSED_PROBE_LIMIT = 3
+
+export type RpcSessionIdentity = object
+
+type WatchdogOptions = {
+  transport: 'direct' | 'relay'
+  sendProbe: (identity: RpcSessionIdentity) => boolean
+  terminate: (identity: RpcSessionIdentity) => void
+  idleProbeMs?: number | null
+  probeTimeoutMs?: number
+  missedProbeLimit?: number
+  voluntaryProbeMinIntervalMs?: number
+  now?: () => number
+  setTimer?: typeof setTimeout
+  clearTimer?: typeof clearTimeout
+}
+
+export class RpcSessionLivenessWatchdog {
+  private identity: RpcSessionIdentity | null = null
+  private timer: ReturnType<typeof setTimeout> | null = null
+  private probing = false
+  private missedProbes = 0
+  private lastInboundAt = 0
+  private lastVoluntaryProbeAt: number | null = null
+  private readonly idleProbeMs: number | null
+  private readonly probeTimeoutMs: number
+  private readonly missedProbeLimit: number
+  private readonly voluntaryProbeMinIntervalMs: number
+  private readonly now: () => number
+  private readonly setTimer: typeof setTimeout
+  private readonly clearTimer: typeof clearTimeout
+
+  constructor(private readonly options: WatchdogOptions) {
+    this.idleProbeMs = options.idleProbeMs === undefined ? LIVENESS_IDLE_MS : options.idleProbeMs
+    this.probeTimeoutMs = options.probeTimeoutMs ?? LIVENESS_PROBE_TIMEOUT_MS
+    this.missedProbeLimit = options.missedProbeLimit ?? MISSED_PROBE_LIMIT
+    this.voluntaryProbeMinIntervalMs = options.voluntaryProbeMinIntervalMs ?? 0
+    this.now = options.now ?? Date.now
+    this.setTimer = options.setTimer ?? setTimeout
+    this.clearTimer = options.clearTimer ?? clearTimeout
+  }
+
+  start(identity: RpcSessionIdentity): void {
+    this.clearActiveTimer()
+    this.identity = identity
+    this.probing = false
+    this.missedProbes = 0
+    this.lastInboundAt = this.now()
+    this.lastVoluntaryProbeAt = null
+    this.armIdle(identity)
+  }
+
+  noteAuthenticatedInbound(identity: RpcSessionIdentity): void {
+    if (this.identity !== identity) {
+      return
+    }
+    this.lastInboundAt = this.now()
+    if (this.missedProbes > 0) {
+      console.log('[net] activity-probe recovered', {
+        transport: this.options.transport,
+        priorMissedProbes: this.missedProbes
+      })
+    }
+    if (!this.probing && this.missedProbes === 0) {
+      return
+    }
+    this.missedProbes = 0
+    this.probing = false
+    this.armIdle(identity)
+  }
+
+  probeNow(identity: RpcSessionIdentity): void {
+    if (this.identity !== identity || this.probing) {
+      return
+    }
+    const now = this.now()
+    if (
+      this.lastVoluntaryProbeAt !== null &&
+      now - this.lastVoluntaryProbeAt < this.voluntaryProbeMinIntervalMs
+    ) {
+      return
+    }
+    this.lastVoluntaryProbeAt = now
+    this.startProbe(identity)
+  }
+
+  stop(identity: RpcSessionIdentity): void {
+    if (this.identity !== identity) {
+      return
+    }
+    this.clearActiveTimer()
+    this.identity = null
+    this.probing = false
+    this.missedProbes = 0
+    this.lastInboundAt = 0
+    this.lastVoluntaryProbeAt = null
+  }
+
+  private armIdle(identity: RpcSessionIdentity, delayMs = this.idleProbeMs): void {
+    this.clearActiveTimer()
+    if (delayMs === null) {
+      return
+    }
+    this.timer = this.setTimer(() => {
+      this.timer = null
+      if (this.identity !== identity) {
+        return
+      }
+      const idleMs = this.now() - this.lastInboundAt
+      if (this.idleProbeMs !== null && idleMs < this.idleProbeMs) {
+        this.armIdle(identity, Math.max(1, this.idleProbeMs - Math.max(0, idleMs)))
+      } else {
+        this.startProbe(identity)
+      }
+    }, delayMs)
+  }
+
+  private startProbe(identity: RpcSessionIdentity): void {
+    if (this.identity !== identity) {
+      return
+    }
+    this.clearActiveTimer()
+    this.probing = true
+    const sentAt = this.now()
+    let sent = false
+    try {
+      sent = this.options.sendProbe(identity)
+    } catch {
+      sent = false
+    }
+    if (!sent) {
+      this.terminateCurrent(identity)
+      return
+    }
+    this.timer = this.setTimer(() => this.handleProbeTimeout(identity, sentAt), this.probeTimeoutMs)
+  }
+
+  private handleProbeTimeout(identity: RpcSessionIdentity, sentAt: number): void {
+    this.timer = null
+    if (this.identity !== identity) {
+      return
+    }
+    const elapsedMs = this.now() - sentAt
+    if (elapsedMs < 0 || elapsedMs > this.probeTimeoutMs * 1.5) {
+      console.log('[net] activity-probe unfair window skipped', {
+        transport: this.options.transport,
+        elapsedMs,
+        timeoutMs: this.probeTimeoutMs
+      })
+      this.startProbe(identity)
+      return
+    }
+    this.missedProbes += 1
+    if (this.missedProbes >= this.missedProbeLimit) {
+      this.terminateCurrent(identity)
+      return
+    }
+    console.log('[net] activity-probe timeout tolerated', {
+      transport: this.options.transport,
+      missedProbes: this.missedProbes,
+      missedProbeLimit: this.missedProbeLimit
+    })
+    this.startProbe(identity)
+  }
+
+  private terminateCurrent(identity: RpcSessionIdentity): void {
+    if (this.identity !== identity) {
+      return
+    }
+    this.clearActiveTimer()
+    this.identity = null
+    this.probing = false
+    console.log('[net] activity-probe TIMEOUT — forcing reconnect', {
+      transport: this.options.transport,
+      missedProbes: this.missedProbes,
+      missedProbeLimit: this.missedProbeLimit
+    })
+    this.options.terminate(identity)
+  }
+
+  private clearActiveTimer(): void {
+    if (this.timer !== null) {
+      this.clearTimer(this.timer)
+      this.timer = null
+    }
+  }
+}
@@ -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'
+  )
+}
@@ -554,7 +554,7 @@ describe('settings host client lifecycle', () => {
    }

    const originalClient = clients.get(replacedHostId)?.[0]
-    act(() => context?.closeHost(replacedHostId))
+    act(() => context?.refreshHostClient(replacedHostId))
    expect(originalClient?.closeMock).toHaveBeenCalledOnce()

    await act(async () => {
@@ -146,8 +146,9 @@ describe('stable logical RPC client', () => {
    await expect(client.sendRequest('status.get')).resolves.toEqual(success('next'))
  })

-  it('lets the physical close settle in-flight requests on suspend, preserving delivery marks', async () => {
+  it('preserves delivery ambiguity without replaying a mutation after relay replacement', async () => {
    const session = new FakeSession('connected')
+    const replacement = new FakeSession('connected')
    const inFlight = deferred<RpcResponse>()
    session.sendRequest.mockReturnValue(inFlight.promise)
    // Mirror the real physical contract: close() rejects post-write pendings
@@ -161,8 +162,10 @@ describe('stable logical RPC client', () => {

    await expect(request).rejects.toBe(closeError)
    await expect(request.catch((error: unknown) => isRpcDeliveryUnknown(error))).resolves.toBe(true)
-    // New requests while suspended still fail definitively before any write.
-    await expect(client.sendRequest('status.get')).rejects.toThrow('Client suspended')
+    await client.migrateTo(replacement, 'relay')
+    expect(replacement.sendRequest).not.toHaveBeenCalled()
+    replacement.sendRequest.mockResolvedValue(success('next'))
+    await expect(client.sendRequest('status.get')).resolves.toEqual(success('next'))
  })

  it('lets the physical close settle in-flight requests on close, keeping pre-write failures definite', async () => {
@@ -3,6 +3,7 @@ import type { RpcClient } from './rpc-client'
 import type { MobileConnectionPath } from './stable-logical-rpc-client'
 import type { ConnectionState } from './types'
 import { useRpcClientContext } from './client-context'
+import type { HostClientAcquisition } from './client-context'

 type UseAllHostClientsOptions = {
  autoConnectHostIds?: readonly string[]
@@ -24,6 +25,7 @@ export function useAllHostClients(hostIds: string[], options?: UseAllHostClients
  )
  const [tick, setTick] = useState(0)
  const acquiredHostIdsRef = useRef<Set<string>>(new Set())
+  const acquisitionsRef = useRef<Map<string, HostClientAcquisition>>(new Map())
  const hostUnsubscribesRef = useRef<Map<string, () => void>>(new Map())
  const closeUnusedRef = useRef(closeUnusedOnRelease)

@@ -42,10 +44,17 @@ export function useAllHostClients(hostIds: string[], options?: UseAllHostClients
      }
      hostUnsubscribesRef.current.clear()
      for (const id of acquiredHostIds) {
+        const acquisition = acquisitionsRef.current.get(id)
+        if (!acquisition) {
+          if (closeUnusedRef.current) {
+            ctx.closeIfUnused(id)
+          }
+          continue
+        }
        if (closeUnusedRef.current) {
-          ctx.releaseAndCloseIfUnused(id)
+          ctx.releaseAndCloseIfUnused(id, acquisition)
        } else {
-          ctx.release(id)
+          ctx.release(id, acquisition)
        }
      }
      if (closeUnusedRef.current) {
@@ -56,6 +65,7 @@ export function useAllHostClients(hostIds: string[], options?: UseAllHostClients
        }
      }
      acquiredHostIdsRef.current.clear()
+      acquisitionsRef.current.clear()
    }
  }, [ctx])

@@ -82,16 +92,26 @@ export function useAllHostClients(hostIds: string[], options?: UseAllHostClients

    for (const id of acquiredHostIdsRef.current) {
      if (!nextAcquiredHostIds.has(id)) {
-        if (closeUnusedOnRelease) {
-          ctx.releaseAndCloseIfUnused(id)
-        } else {
-          ctx.release(id)
+        const acquisition = acquisitionsRef.current.get(id)
+        if (!acquisition) {
+          if (closeUnusedOnRelease) {
+            ctx.closeIfUnused(id)
+          }
+          continue
        }
+        if (closeUnusedOnRelease) {
+          ctx.releaseAndCloseIfUnused(id, acquisition)
+        } else {
+          ctx.release(id, acquisition)
+        }
+        acquisitionsRef.current.delete(id)
      }
    }
    for (const id of nextAcquiredHostIds) {
      if (!acquiredHostIdsRef.current.has(id)) {
-        ctx.acquire(id)
+        const acquisition = {}
+        acquisitionsRef.current.set(id, acquisition)
+        ctx.acquire(id, acquisition)
      }
    }
    if (closeUnusedOnRelease) {