Merge origin/main into structured-session-eviction

Three mobile conflicts, all from this stack's own extractions meeting main:

- rpc-client.ts: this branch had already moved RpcClient into rpc-client-contract
  and re-exports it; main still declares it inline. Keeping both produced a
  duplicate declaration. Dropped the inline copy and ported the one member main
  had that the contract lacked (getLastInboundAt).
- worktree-create-retry(.test).ts: the cutover-error split moved
  isLogicalClientCutoverError out of stable-logical-rpc-client, and main added two
  new imports alongside the old path. Both kept, pointed at the new module.
- StreamRequest keeps paramsForReconnect from this branch.

Root typecheck, mobile tsc, full lint gate, and the mobile suite (3755 tests) green;
pnpm-lock.yaml untouched.

mobile/src/tasks/worktree-create-retry.test.ts

@@ -1,24 +1,87 @@
-import { describe, expect, it } from 'vitest'
+import { describe, expect, it, vi } from 'vitest'
 import type { RpcClient } from '../transport/rpc-client'
+import { markRpcDeliveryUnknown } from '../transport/rpc-delivery-ambiguity'
 import { LogicalClientCutoverError } from '../transport/logical-client-cutover-error'
-import { createWorktreeWithNameRetry } from './worktree-create-retry'
+import type { ConnectionState } from '../transport/types'
+import { WORKTREE_CREATE_DEDUPE_TTL_MS } from '../../../src/shared/new-workspace/worktree-create-retry-policy'
+import {
+  createWorktreeWithNameRetry,
+  WORKTREE_CREATE_AMBIGUOUS_RECONNECT_WAIT_MS,
+  WORKTREE_CREATE_AMBIGUOUS_REPLAY_WINDOW_MS
+} from './worktree-create-retry'
+import { WORKTREE_CREATE_TIMEOUT_MS } from './workspace-create-timeout'
 
 type Attempt = { method: string; params: Record<string, unknown> }
 
+// Drives the transport state a replay has to wait on. Production couples the two:
+// a socket close rejects the pending frame AND drops the client off 'connected'.
+function connectionController(): {
+  getState: () => ConnectionState
+  onStateChange: (listener: (state: ConnectionState) => void) => () => void
+  set: (next: ConnectionState) => void
+} {
+  let state: ConnectionState = 'connected'
+  const listeners = new Set<(next: ConnectionState) => void>()
+  return {
+    getState: () => state,
+    onStateChange: (listener) => {
+      listeners.add(listener)
+      return () => listeners.delete(listener)
+    },
+    set: (next) => {
+      state = next
+      for (const listener of listeners) {
+        listener(next)
+      }
+    }
+  }
+}
+
+// Lets a parked replay reach its state wait before the test resumes the transport.
+async function flush(): Promise<void> {
+  await new Promise((resolve) => setTimeout(resolve, 0))
+}
+
 // A client whose per-call outcome is scripted: return an id, a server error
 // message, or throw (transport-level rejection, e.g. a connection-migration
 // cutover). Records every call so tests can assert on the clientMutationId.
 function scriptedClient(
-  outcomes: Array<{ id: string } | { errorMessage: string } | { throws: unknown }>,
-  attempts: Attempt[]
+  outcomes: Array<
+    | { id: string }
+    | { errorMessage: string }
+    // takesMs models how long the ambiguity took to SURFACE — a clean close is
+    // instant, a half-open socket waits out the liveness watchdog or the timeout.
+    // reconnectsAfterMs brings the transport back so the next replay can run.
+    | { throws: unknown; dropsConnection?: boolean; takesMs?: number; reconnectsAfterMs?: number }
+  >,
+  attempts: Attempt[],
+  connection?: ReturnType<typeof connectionController>,
+  // Wall-clock stamp of the last frame that really arrived. Absent = a transport that
+  // can't vouch for one, which must fall back to the send; a function models a live
+  // socket whose stamp keeps advancing.
+  lastInboundAt?: number | (() => number)
 ): RpcClient {
   let call = 0
   return {
+    getState: () => connection?.getState() ?? 'connected',
+    getLastInboundAt: () =>
+      (typeof lastInboundAt === 'function' ? lastInboundAt() : lastInboundAt) ?? null,
+    onStateChange: (listener: (state: ConnectionState) => void) =>
+      connection?.onStateChange(listener) ?? (() => {}),
     sendRequest: async (method: string, params?: unknown) => {
       attempts.push({ method, params: (params ?? {}) as Record<string, unknown> })
       const outcome = outcomes[Math.min(call, outcomes.length - 1)]!
       call += 1
       if ('throws' in outcome) {
+        if (outcome.takesMs !== undefined) {
+          await new Promise((resolve) => setTimeout(resolve, outcome.takesMs))
+        }
+        if (outcome.dropsConnection) {
+          connection?.set('reconnecting')
+          if (outcome.reconnectsAfterMs !== undefined) {
+            setTimeout(() => connection?.set('connected'), outcome.reconnectsAfterMs)
+          }
+        }
         throw outcome.throws
       }
       if ('errorMessage' in outcome) {
@@ -117,6 +180,8 @@ describe('createWorktreeWithNameRetry', () => {
     expect(attempts).toHaveLength(6)
   })
 
+  // Also the delivery-ambiguity guard: an unmarked error is a DEFINITE failure, so
+  // neither the cutover nor the replay path may re-issue it.
   it('does not treat an ordinary transport error as a cutover', async () => {
     const attempts: Attempt[] = []
     const client = scriptedClient([{ throws: new Error('Request timed out') }], attempts)
@@ -188,4 +253,407 @@ describe('createWorktreeWithNameRetry', () => {
     expect(attempts).toHaveLength(1)
     expect(attempts[0]!.params.clientMutationId).toBeUndefined()
   })
+
+  it('replays a delivery-ambiguous socket drop with the SAME key once the transport returns', async () => {
+    const attempts: Attempt[] = []
+    const connection = connectionController()
+    const client = scriptedClient(
+      [
+        {
+          throws: markRpcDeliveryUnknown(new Error('Connection interrupted')),
+          dropsConnection: true
+        },
+        { id: 'wt-drop' }
+      ],
+      attempts,
+      connection
+    )
+
+    const pending = createWorktreeWithNameRetry({
+      client,
+      baseName: 'urchin',
+      buildParams: (name) => ({ repo: 'id:r', name }),
+      supportsIdempotentCutoverRetry: true,
+      mintMutationId: () => 'key-drop'
+    })
+
+    await flush()
+    // Parked on the reconnect, not resent onto a dead socket.
+    expect(attempts).toHaveLength(1)
+
+    connection.set('connected')
+    await expect(pending).resolves.toEqual({ worktreeId: 'wt-drop', name: 'urchin' })
+    expect(attempts).toHaveLength(2)
+    // Idempotency: the host dedupes the replay against the create it may already
+    // have finished, instead of building a second worktree.
+    expect(attempts[1]!.params.clientMutationId).toBe('key-drop')
+    expect(attempts[1]!.params.name).toBe('urchin')
+  })
+
+  it('does not replay a delivery-ambiguous drop when the host lacks idempotency support', async () => {
+    const attempts: Attempt[] = []
+    const client = scriptedClient(
+      [{ throws: markRpcDeliveryUnknown(new Error('Connection interrupted')) }],
+      attempts
+    )
+    await expect(
+      createWorktreeWithNameRetry({
+        client,
+        baseName: 'limpet',
+        buildParams: (name) => ({ repo: 'id:r', name }),
+        supportsIdempotentCutoverRetry: false,
+        mintMutationId: () => 'must-not-be-used'
+      })
+    ).rejects.toThrow('Connection interrupted')
+    expect(attempts).toHaveLength(1)
+    expect(attempts[0]!.params.clientMutationId).toBeUndefined()
+  })
+
+  it('gives up after the delivery-ambiguity replay budget, still inside the dedupe TTL', async () => {
+    vi.useFakeTimers()
+    try {
+      const attempts: Attempt[] = []
+      const connection = connectionController()
+      const client = scriptedClient(
+        [
+          {
+            throws: markRpcDeliveryUnknown(new Error('Connection interrupted')),
+            dropsConnection: true,
+            reconnectsAfterMs: 1_000
+          }
+        ],
+        attempts,
+        connection
+      )
+      const startedAt = Date.now()
+      let settledAt = -1
+      const pending = createWorktreeWithNameRetry({
+        client,
+        baseName: 'barnacle',
+        buildParams: (name) => ({ repo: 'id:r', name }),
+        supportsIdempotentCutoverRetry: true,
+        mintMutationId: () => 'key-budget'
+      }).catch((error: unknown) => {
+        settledAt = Date.now()
+        throw error
+      })
+      const settled = expect(pending).rejects.toThrow('Connection interrupted')
+      await vi.advanceTimersByTimeAsync(WORKTREE_CREATE_DEDUPE_TTL_MS)
+      await settled
+      // Initial attempt + 2 replays.
+      expect(attempts).toHaveLength(3)
+      // The budget is a count, but what keeps a replay reconciling instead of building
+      // a second worktree is wall clock: every attempt has to land while the host still
+      // holds the record, with the watchdog's detection latency already deducted.
+      expect(settledAt - startedAt).toBeLessThanOrEqual(WORKTREE_CREATE_AMBIGUOUS_REPLAY_WINDOW_MS)
+    } finally {
+      vi.useRealTimers()
+    }
+    // Generous: advanceTimersByTimeAsync yields through REAL macrotasks between ticks,
+    // so vitest's default 5s real-time budget is reachable on a loaded runner.
+  }, 30_000)
+
+  it('surfaces the original ambiguity when the transport never comes back', async () => {
+    vi.useFakeTimers()
+    try {
+      const attempts: Attempt[] = []
+      const connection = connectionController()
+      const client = scriptedClient(
+        [
+          {
+            throws: markRpcDeliveryUnknown(new Error('Connection interrupted')),
+            dropsConnection: true
+          }
+        ],
+        attempts,
+        connection
+      )
+
+      const pending = createWorktreeWithNameRetry({
+        client,
+        baseName: 'anemone',
+        buildParams: (name) => ({ repo: 'id:r', name }),
+        supportsIdempotentCutoverRetry: true,
+        mintMutationId: () => 'key-stuck'
+      })
+      const settled = expect(pending).rejects.toThrow('Connection interrupted')
+      // A phone that never reconnects must not leave the Create spinner parked.
+      await vi.advanceTimersByTimeAsync(60_000)
+      await settled
+      expect(attempts).toHaveLength(1)
+    } finally {
+      vi.useRealTimers()
+    }
+    // Generous: advanceTimersByTimeAsync yields through REAL macrotasks between ticks,
+    // so vitest's default 5s real-time budget is reachable on a loaded runner.
+  }, 30_000)
+
+  it('does not replay an ambiguity that surfaced while the transport stayed connected', async () => {
+    vi.useFakeTimers()
+    try {
+      const attempts: Attempt[] = []
+      const connection = connectionController()
+      const client = scriptedClient(
+        [
+          {
+            // A silently dropped response frame leaves the socket alive, so nothing
+            // rejects until the request timeout — by then the host resolved at an
+            // unknowable point and its dedupe record is very likely gone.
+            throws: markRpcDeliveryUnknown(new Error('Request timed out: worktree.create')),
+            takesMs: WORKTREE_CREATE_TIMEOUT_MS
+          },
+          { id: 'wt-duplicate' }
+        ],
+        attempts,
+        connection,
+        // Watchdog probes keep answering throughout, so the replay window looks wide
+        // open on arrival: only the still-connected guard can stop this replay.
+        () => Date.now()
+      )
+
+      const pending = createWorktreeWithNameRetry({
+        client,
+        baseName: 'cuttlefish',
+        buildParams: (name) => ({ repo: 'id:r', name }),
+        supportsIdempotentCutoverRetry: true,
+        mintMutationId: () => 'key-expired'
+      })
+      const settled = expect(pending).rejects.toThrow('Request timed out')
+      await vi.advanceTimersByTimeAsync(WORKTREE_CREATE_TIMEOUT_MS)
+      await settled
+      // Never left 'connected', so no drop was ever detected and the staleness of the
+      // ambiguity is unbounded. Re-sending would miss the record and build a SECOND
+      // worktree — for a folder workspace, one with the very same name.
+      expect(connection.getState()).toBe('connected')
+      expect(attempts).toHaveLength(1)
+    } finally {
+      vi.useRealTimers()
+    }
+    // Generous: advanceTimersByTimeAsync yields through REAL macrotasks between ticks,
+    // so vitest's default 5s real-time budget is reachable on a loaded runner.
+  }, 30_000)
+
+  it('clamps a later reconnect wait to what is left of the replay window', async () => {
+    vi.useFakeTimers()
+    try {
+      const attempts: Attempt[] = []
+      const connection = connectionController()
+      const client = scriptedClient(
+        [
+          {
+            // Most of the window is already gone by the time the drop is reported, so the
+            // remainder — not the full wait — is what the second wait gets.
+            throws: markRpcDeliveryUnknown(new Error('Connection interrupted')),
+            dropsConnection: true,
+            takesMs: 30_000,
+            reconnectsAfterMs: 1_000
+          },
+          {
+            // Second drop, and this time the phone never comes back.
+            throws: markRpcDeliveryUnknown(new Error('Connection interrupted')),
+            dropsConnection: true
+          }
+        ],
+        attempts,
+        connection
+      )
+
+      const startedAt = Date.now()
+      let settledAt = -1
+      const pending = createWorktreeWithNameRetry({
+        client,
+        baseName: 'nautilus',
+        buildParams: (name) => ({ repo: 'id:r', name }),
+        supportsIdempotentCutoverRetry: true,
+        mintMutationId: () => 'key-clamped'
+      }).catch((error: unknown) => {
+        settledAt = Date.now()
+        throw error
+      })
+      const settled = expect(pending).rejects.toThrow('Connection interrupted')
+      await vi.advanceTimersByTimeAsync(WORKTREE_CREATE_AMBIGUOUS_REPLAY_WINDOW_MS)
+      await settled
+      expect(attempts).toHaveLength(2)
+      // The window is anchored at the FIRST ambiguity, so the second wait gets only the
+      // remainder — it must not restart a full wait and carry the replay past the record.
+      expect(settledAt - startedAt).toBe(WORKTREE_CREATE_AMBIGUOUS_REPLAY_WINDOW_MS)
+    } finally {
+      vi.useRealTimers()
+    }
+    // Generous: advanceTimersByTimeAsync yields through REAL macrotasks between ticks,
+    // so vitest's default 5s real-time budget is reachable on a loaded runner.
+  }, 30_000)
+
+  it('refuses a replay when the ambiguity surfaced long after the last inbound frame', async () => {
+    vi.useFakeTimers()
+    try {
+      const attempts: Attempt[] = []
+      const connection = connectionController()
+      const startedAt = Date.now()
+      const client = scriptedClient(
+        [
+          {
+            throws: markRpcDeliveryUnknown(new Error('Connection interrupted')),
+            dropsConnection: true,
+            reconnectsAfterMs: 1_000,
+            // The phone was backgrounded: the OS suspended JS and killed the socket, and
+            // nothing observed the drop until the app came back ten minutes later.
+            takesMs: 600_000
+          }
+        ],
+        attempts,
+        connection,
+        startedAt + 2_000
+      )
+      const pending = createWorktreeWithNameRetry({
+        client,
+        baseName: 'limpet',
+        buildParams: (name) => ({ repo: 'id:r', name }),
+        supportsIdempotentCutoverRetry: true,
+        mintMutationId: () => 'key-stale'
+      })
+      const settled = expect(pending).rejects.toThrow('Connection interrupted')
+      await vi.advanceTimersByTimeAsync(700_000)
+      await settled
+      // The host forgot this create ~9 minutes ago. A replay would not reconcile — it
+      // would build a second worktree — so the create fails instead.
+      expect(attempts).toHaveLength(1)
+    } finally {
+      vi.useRealTimers()
+    }
+    // Generous: advanceTimersByTimeAsync yields through REAL macrotasks between ticks,
+    // so vitest's default 5s real-time budget is reachable on a loaded runner.
+  }, 30_000)
+
+  it('still replays a long-running create whose socket stayed live until it dropped', async () => {
+    vi.useFakeTimers()
+    try {
+      const attempts: Attempt[] = []
+      const connection = connectionController()
+      const startedAt = Date.now()
+      const client = scriptedClient(
+        [
+          {
+            // Three minutes of clone/fetch with the socket answering probes throughout,
+            // then it drops and the watchdog reports it four seconds later.
+            throws: markRpcDeliveryUnknown(new Error('Connection interrupted')),
+            dropsConnection: true,
+            reconnectsAfterMs: 1_000,
+            takesMs: 184_000
+          },
+          { id: 'wt-long' }
+        ],
+        attempts,
+        connection,
+        startedAt + 180_000
+      )
+      const pending = createWorktreeWithNameRetry({
+        client,
+        baseName: 'kelp',
+        buildParams: (name) => ({ repo: 'id:r', name }),
+        supportsIdempotentCutoverRetry: true,
+        mintMutationId: () => 'key-long'
+      })
+      await vi.advanceTimersByTimeAsync(250_000)
+      // Anchoring at the send would refuse this — the create outran the TTL long before it
+      // went ambiguous — but the record only starts ticking when the host RESOLVES, and the
+      // live socket proves our knowledge was current.
+      expect(await pending).toEqual({ worktreeId: 'wt-long', name: 'kelp' })
+      expect(attempts).toHaveLength(2)
+      expect(attempts[0]!.params.clientMutationId).toBe('key-long')
+      expect(attempts[1]!.params.clientMutationId).toBe('key-long')
+    } finally {
+      vi.useRealTimers()
+    }
+    // Generous: advanceTimersByTimeAsync yields through REAL macrotasks between ticks,
+    // so vitest's default 5s real-time budget is reachable on a loaded runner.
+  }, 30_000)
+
+  it('does not extend the replay window when the replacement session reports fresher inbound activity', async () => {
+    vi.useFakeTimers()
+    try {
+      const attempts: Attempt[] = []
+      const connection = connectionController()
+      const startedAt = Date.now()
+      // The replacement session starts answering probes, so its inbound stamp is far
+      // newer than anything that bears on when the ORIGINAL create resolved host-side.
+      let lastInboundAt = startedAt + 1_000
+      setTimeout(() => {
+        lastInboundAt = startedAt + 45_000
+      }, 45_000)
+      const client = scriptedClient(
+        [
+          {
+            throws: markRpcDeliveryUnknown(new Error('Connection lost')),
+            dropsConnection: true,
+            takesMs: 40_000,
+            reconnectsAfterMs: 1_000
+          },
+          {
+            throws: markRpcDeliveryUnknown(new Error('Connection lost')),
+            dropsConnection: true,
+            takesMs: 20_000,
+            reconnectsAfterMs: 1_000
+          },
+          { id: 'wt-third' }
+        ],
+        attempts,
+        connection,
+        () => lastInboundAt
+      )
+
+      const pending = createWorktreeWithNameRetry({
+        client,
+        baseName: 'urchin',
+        buildParams: (name) => ({ repo: 'id:r', name }),
+        supportsIdempotentCutoverRetry: true,
+        mintMutationId: () => 'key-anchored'
+      })
+      const settled = expect(pending).rejects.toThrow('Connection lost')
+      await vi.advanceTimersByTimeAsync(120_000)
+      await settled
+      // The deadline is fixed at the FIRST ambiguity. Re-reading it here would buy another
+      // 50s of window off a stamp that says nothing about the original request.
+      expect(attempts).toHaveLength(2)
+    } finally {
+      vi.useRealTimers()
+    }
+    // Generous: advanceTimersByTimeAsync yields through REAL macrotasks between ticks,
+    // so vitest's default 5s real-time budget is reachable on a loaded runner.
+  }, 30_000)
+
+  it('does not replay a dropped transport error that was never marked delivery-unknown', async () => {
+    const attempts: Attempt[] = []
+    const connection = connectionController()
+    const client = scriptedClient(
+      // Unmarked: the frame is known NOT to have reached the wire, so the host never saw
+      // it and a replay would be a second create, not a reconciliation. The transport
+      // still leaves 'connected', so only the delivery-unknown check stops this.
+      [{ throws: new Error('Socket closed before send'), dropsConnection: true }],
+      attempts,
+      connection
+    )
+    await expect(
+      createWorktreeWithNameRetry({
+        client,
+        baseName: 'urchin',
+        buildParams: (name) => ({ repo: 'id:r', name }),
+        supportsIdempotentCutoverRetry: true,
+        mintMutationId: () => 'key-unmarked'
+      })
+    ).rejects.toThrow('Socket closed before send')
+    expect(attempts).toHaveLength(1)
+  })
+
+  it('keeps the replay window strictly inside the host dedupe TTL', () => {
+    // The window is measured from a lower bound on when the host could have resolved, so
+    // it has to leave the record room for the replay to still be in flight. Widening it
+    // to the whole TTL would let the last replay land on a record that just expired.
+    expect(WORKTREE_CREATE_AMBIGUOUS_REPLAY_WINDOW_MS).toBeGreaterThan(0)
+    expect(WORKTREE_CREATE_AMBIGUOUS_REPLAY_WINDOW_MS).toBeLessThan(WORKTREE_CREATE_DEDUPE_TTL_MS)
+    // A single wait must not be able to consume the window on its own.
+    expect(WORKTREE_CREATE_AMBIGUOUS_RECONNECT_WAIT_MS).toBeLessThan(
+      WORKTREE_CREATE_AMBIGUOUS_REPLAY_WINDOW_MS
+    )
+  })
 })

mobile/src/tasks/worktree-create-retry.ts

@@ -1,11 +1,14 @@
 import type { RpcClient } from '../transport/rpc-client'
 import type { RpcResponse, RpcSuccess } from '../transport/types'
+import { isRpcDeliveryUnknown } from '../transport/rpc-delivery-ambiguity'
+import { waitForRpcClientReconnected } from '../transport/rpc-client-reconnect-wait'
 import { isLogicalClientCutoverError } from '../transport/logical-client-cutover-error'
 import {
   CLIENT_WORKTREE_CREATE_MAX_ATTEMPTS,
   getClientWorktreeCreateCandidate,
   getGeneratedWorktreeCreateRetryCandidate,
-  isRetryableWorktreeCreateConflict
+  isRetryableWorktreeCreateConflict,
+  WORKTREE_CREATE_DEDUPE_TTL_MS
 } from '../../../src/shared/new-workspace/worktree-create-retry-policy'
 import { WORKTREE_CREATE_TIMEOUT_MS } from './workspace-create-timeout'
 
@@ -24,6 +27,29 @@ export type WorktreeCreateResult = { worktreeId: string; name: string } | { erro
 // worktree silently created.
 const WORKTREE_CREATE_CUTOVER_MAX_RETRIES = 5
 
+// Why: a connection migration is not the only way a create goes delivery-ambiguous.
+// A plain socket close (cellular flap, relay drop, the phone backgrounding and the
+// supervisor suspending a billed relay splice) rejects the in-flight frame as
+// delivery-unknown with no generation bump, and create holds the longest budget of
+// any mobile RPC — 10 minutes of clone/fetch/setup — so it is the likeliest request
+// to be caught by one. Surfacing that as a failure is wrong: the host may well have
+// finished the worktree. Replay on the same clientMutationId instead.
+const WORKTREE_CREATE_AMBIGUOUS_MAX_RETRIES = 2
+
+// Why: the host keeps a settled create's dedupe record for WORKTREE_CREATE_DEDUPE_TTL_MS
+// after it resolves; a replay that lands later misses it and the host's suffix loop
+// builds a SECOND worktree — and for folder workspaces a second one with the SAME name
+// and no collision check at all. So a replay is bounded by how stale our knowledge of the
+// host is, and that has to be measured in WALL CLOCK: iOS/Android suspend JS timers while
+// the app is backgrounded, so any ceiling derived from timer intervals — a watchdog probe
+// budget, a request timeout — can be arbitrarily wrong across a background cycle, which is
+// exactly when a create sits ambiguous for minutes.
+const WORKTREE_CREATE_REPLAY_FLIGHT_MARGIN_MS = 10_000
+export const WORKTREE_CREATE_AMBIGUOUS_REPLAY_WINDOW_MS =
+  WORKTREE_CREATE_DEDUPE_TTL_MS - WORKTREE_CREATE_REPLAY_FLIGHT_MARGIN_MS
+// Bounded so the Create spinner doesn't sit for the whole window; the deadline still caps it.
+export const WORKTREE_CREATE_AMBIGUOUS_RECONNECT_WAIT_MS = 20_000
+
 export type CreateWorktreeWithNameRetryArgs = {
   client: RpcClient
   baseName: string
@@ -78,32 +104,87 @@ export async function createWorktreeWithNameRetry(
   return { error: lastError ?? 'Failed to create workspace' }
 }
 
-// Sends worktree.create, re-issuing on a connection-migration cutover only. The
-// shared clientMutationId in `params` keeps the retry idempotent host-side.
+// Sends worktree.create, re-issuing whenever the request went delivery-ambiguous —
+// the frame reached the wire but no response came back, so the host may already have
+// built the worktree. The shared clientMutationId in `params` keeps the retry
+// idempotent host-side. A definite failure (never sent, or a server error response)
+// is returned to the caller untouched.
 async function sendWorktreeCreateResilient(
   client: RpcClient,
   params: Record<string, unknown>,
   supportsIdempotentCutoverRetry: boolean
 ): Promise<RpcResponse> {
-  for (let migrationRetry = 0; ; migrationRetry += 1) {
+  let migrationRetry = 0
+  let ambiguousRetry = 0
+  const firstSentAt = Date.now()
+  let replayDeadlineAt: number | null = null
+  for (;;) {
     try {
       return await client.sendRequest('worktree.create', params, {
         timeoutMs: WORKTREE_CREATE_TIMEOUT_MS
       })
     } catch (error) {
+      if (!supportsIdempotentCutoverRetry) {
+        throw error
+      }
+      if (isLogicalClientCutoverError(error)) {
+        if (migrationRetry >= WORKTREE_CREATE_CUTOVER_MAX_RETRIES) {
+          throw error
+        }
+        migrationRetry += 1
+        // Why: LogicalClientCutoverError is raised only after migrateTo installs an
+        // authenticated replacement, so retry immediately instead of adding UI lag.
+        continue
+      }
+      if (!isRpcDeliveryUnknown(error) || ambiguousRetry >= WORKTREE_CREATE_AMBIGUOUS_MAX_RETRIES) {
+        throw error
+      }
+      // Why: every transport path that reports a *drop* leaves 'connected' before the
+      // rejection reaches us (rpc-client.ts:675/695/1213 set state first or reject via
+      // queueMicrotask; the relay's fail() publishes synchronously). So still being
+      // 'connected' here means the socket was healthy the whole time and only the
+      // response went missing — the request-timeout path, which surfaces after
+      // WORKTREE_CREATE_TIMEOUT_MS. That says nothing about when the host actually
+      // resolved, so the dedupe record may be long gone and a replay would build a
+      // second worktree instead of reconciling. Fail the create instead.
+      if (client.getState() === 'connected') {
+        throw error
+      }
+      // Computed once: a later ambiguity reads a fresher lastInboundAt from the
+      // replacement session, which would push the deadline past the record it respects.
+      replayDeadlineAt ??= resolveReplayDeadline(client, firstSentAt)
+      const remainingWindowMs = replayDeadlineAt - Date.now()
+      if (remainingWindowMs <= 0) {
+        throw error
+      }
+      ambiguousRetry += 1
+      // Why: unlike a cutover, no replacement session exists yet — resending now
+      // would just hit the dead one, so wait for the transport to come back and
+      // surface the original ambiguity if it does not. Clamped to the window so the
+      // wait itself cannot carry the replay past the host's record.
       if (
-        !supportsIdempotentCutoverRetry ||
-        !isLogicalClientCutoverError(error) ||
-        migrationRetry >= WORKTREE_CREATE_CUTOVER_MAX_RETRIES
+        !(await waitForRpcClientReconnected(
+          client,
+          Math.min(WORKTREE_CREATE_AMBIGUOUS_RECONNECT_WAIT_MS, remainingWindowMs)
+        ))
       ) {
         throw error
       }
-      // Why: LogicalClientCutoverError is raised only after migrateTo installs an
-      // authenticated replacement, so retry immediately instead of adding UI lag.
     }
   }
 }
 
+// Latest instant the host's dedupe record is still guaranteed to exist, from the earliest
+// point the create could have resolved. lastInboundAt stamps a frame that really arrived,
+// so it stays honest across a suspension in a way a timer budget cannot; the send is the
+// fallback, and a sound floor either way, because the host cannot resolve a create it has
+// not yet received.
+function resolveReplayDeadline(client: RpcClient, firstSentAt: number): number {
+  const lastInboundAt = client.getLastInboundAt?.() ?? null
+  const anchor = lastInboundAt !== null && lastInboundAt > firstSentAt ? lastInboundAt : firstSentAt
+  return anchor + WORKTREE_CREATE_AMBIGUOUS_REPLAY_WINDOW_MS
+}
+
 function defaultWorktreeCreateMutationId(): string {
   const randomPart = Math.random().toString(36).slice(2, 10)
   return `worktree-create:${Date.now().toString(36)}:${randomPart}`

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

@@ -114,6 +114,7 @@ export function connectMobileRelayRpcSession(args: {
     getState: () => state,
     getReconnectAttempt: () => 0,
     getLastConnectedAt: () => lastConnectedAt,
+    getLastInboundAt: () => livenessWatchdog.getLastInboundAt() || null,
     onStateChange(listener) {
       stateListeners.add(listener)
       return () => stateListeners.delete(listener)

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

@@ -36,6 +36,10 @@ export type RpcClient = {
   getState: () => ConnectionState
   getReconnectAttempt: () => number
   getLastConnectedAt: () => number | null
+  // Wall-clock stamp of the last inbound frame on the current session, or null when the
+  // transport can't vouch for one. Optional so older/foreign RpcClient shapes stay valid;
+  // callers must treat absent/null as "unknown" and fall back to a safe bound.
+  getLastInboundAt?: () => number | null
   onStateChange: (listener: (state: ConnectionState) => void) => () => void
   notifyForeground: (reason?: ForegroundNudgeReason) => void
   restartAfterStructuredBackground?: () => void

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

@@ -0,0 +1,108 @@
+import { describe, expect, it, vi } from 'vitest'
+import type { RpcClient } from './rpc-client'
+import type { ConnectionState } from './types'
+import { waitForRpcClientReconnected } from './rpc-client-reconnect-wait'
+
+// notifyDuringSubscribe models a client that fires the listener synchronously from
+// onStateChange — the branch that has to survive `unsubscribe` not being assigned yet.
+function fakeClient(
+  initial: ConnectionState,
+  options: { notifyDuringSubscribe?: ConnectionState } = {}
+): { client: RpcClient; set: (next: ConnectionState) => void; listenerCount: () => number } {
+  let state = initial
+  const listeners = new Set<(next: ConnectionState) => void>()
+  const client = {
+    getState: () => state,
+    onStateChange: (listener: (next: ConnectionState) => void) => {
+      listeners.add(listener)
+      if (options.notifyDuringSubscribe) {
+        state = options.notifyDuringSubscribe
+        listener(state)
+      }
+      return () => listeners.delete(listener)
+    }
+  } as unknown as RpcClient
+  return {
+    client,
+    set: (next) => {
+      state = next
+      // Safe to delete during iteration: finish() unsubscribes the current listener.
+      for (const listener of listeners) {
+        listener(next)
+      }
+    },
+    listenerCount: () => listeners.size
+  }
+}
+
+describe('waitForRpcClientReconnected', () => {
+  it('resolves immediately when the transport is already connected', async () => {
+    const { client, listenerCount } = fakeClient('connected')
+    await expect(waitForRpcClientReconnected(client, 10_000)).resolves.toBe(true)
+    // No subscription at all on the fast path, so nothing to leak.
+    expect(listenerCount()).toBe(0)
+  })
+
+  it('resolves once the transport comes back, and unsubscribes', async () => {
+    const { client, set, listenerCount } = fakeClient('reconnecting')
+    const pending = waitForRpcClientReconnected(client, 10_000)
+    expect(listenerCount()).toBe(1)
+    set('connected')
+    await expect(pending).resolves.toBe(true)
+    expect(listenerCount()).toBe(0)
+  })
+
+  it('gives up at the timeout and unsubscribes', async () => {
+    vi.useFakeTimers()
+    try {
+      const { client, listenerCount } = fakeClient('reconnecting')
+      const pending = waitForRpcClientReconnected(client, 10_000)
+      await vi.advanceTimersByTimeAsync(10_000)
+      await expect(pending).resolves.toBe(false)
+      expect(listenerCount()).toBe(0)
+    } finally {
+      vi.useRealTimers()
+    }
+    // Generous: advanceTimersByTimeAsync yields through REAL macrotasks between ticks,
+    // so vitest's default 5s real-time budget is reachable on a loaded runner.
+  }, 30_000)
+
+  it('does not wait out the timeout when the pairing is already revoked', async () => {
+    const { client, listenerCount } = fakeClient('auth-failed')
+    await expect(waitForRpcClientReconnected(client, 10_000)).resolves.toBe(false)
+    expect(listenerCount()).toBe(0)
+  })
+
+  it('does not wait out the timeout when the pairing is revoked mid-wait', async () => {
+    vi.useFakeTimers()
+    try {
+      const { client, set, listenerCount } = fakeClient('reconnecting')
+      const pending = waitForRpcClientReconnected(client, 10_000)
+      set('auth-failed')
+      await expect(pending).resolves.toBe(false)
+      expect(listenerCount()).toBe(0)
+      // Nothing left armed that could fire after the caller moved on.
+      expect(vi.getTimerCount()).toBe(0)
+    } finally {
+      vi.useRealTimers()
+    }
+    // Generous: advanceTimersByTimeAsync yields through REAL macrotasks between ticks,
+    // so vitest's default 5s real-time budget is reachable on a loaded runner.
+  }, 30_000)
+
+  it('tears down cleanly when the state change fires synchronously during subscribe', async () => {
+    vi.useFakeTimers()
+    try {
+      const { client, listenerCount } = fakeClient('reconnecting', {
+        notifyDuringSubscribe: 'connected'
+      })
+      await expect(waitForRpcClientReconnected(client, 10_000)).resolves.toBe(true)
+      expect(listenerCount()).toBe(0)
+      expect(vi.getTimerCount()).toBe(0)
+    } finally {
+      vi.useRealTimers()
+    }
+    // Generous: advanceTimersByTimeAsync yields through REAL macrotasks between ticks,
+    // so vitest's default 5s real-time budget is reachable on a loaded runner.
+  }, 30_000)
+})

mobile/src/transport/rpc-client-reconnect-wait.ts

@@ -0,0 +1,55 @@
+import type { RpcClient } from './rpc-client'
+
+// Why: after a socket drop the direct client parks a send in waitForConnected, but a
+// relay session rejects it outright ('relay session not connected') until the supervisor
+// migrates a replacement in. A replay that fires the instant the drop lands therefore
+// burns its budget on the dead session, so callers that mean to re-issue an idempotent
+// request wait for the transport to come back first.
+//
+// Unlike waitForAuthenticated, intermediate 'disconnected'/'reconnecting' states are the
+// expected path here, not a failure. Only a revoked pairing and the timeout end the wait
+// unsuccessfully.
+export function waitForRpcClientReconnected(
+  client: RpcClient,
+  timeoutMs: number
+): Promise<boolean> {
+  const state = client.getState()
+  if (state === 'connected') {
+    return Promise.resolve(true)
+  }
+  if (state === 'auth-failed') {
+    return Promise.resolve(false)
+  }
+  return new Promise((resolve) => {
+    let settled = false
+    let unsubscribe: (() => void) | null = null
+    // Armed before subscribing so a synchronous notification during registration
+    // still finds a timer to clear.
+    const timer = setTimeout(() => finish(false), timeoutMs)
+    unsubscribe = client.onStateChange((state) => {
+      if (state === 'connected') {
+        finish(true)
+      } else if (state === 'auth-failed') {
+        // Revoked pairing never reaches 'connected', so waiting out the timeout only
+        // delays the error the caller is going to surface anyway.
+        finish(false)
+      }
+    })
+    if (settled) {
+      // The notification fired inside onStateChange, before we held the handle.
+      unsubscribe()
+      unsubscribe = null
+    }
+
+    function finish(reconnected: boolean): void {
+      if (settled) {
+        return
+      }
+      settled = true
+      clearTimeout(timer)
+      unsubscribe?.()
+      unsubscribe = null
+      resolve(reconnected)
+    }
+  })
+}

mobile/src/transport/rpc-client.ts

@@ -1128,6 +1128,10 @@ export function connect(
       return lastConnectedAt
     },
 
+    getLastInboundAt(): number | null {
+      return livenessWatchdog.getLastInboundAt() || null
+    },
+
     onStateChange(listener: (state: ConnectionState) => void): () => void {
       stateListeners.add(listener)
       return () => stateListeners.delete(listener)

mobile/src/transport/rpc-session-liveness-watchdog.ts

@@ -52,6 +52,13 @@ export class RpcSessionLivenessWatchdog {
     this.armIdle(identity)
   }
 
+  // Wall-clock stamp of the last frame that actually arrived; 0 before the first
+  // start(). Unlike a timer deadline this survives a JS suspension, so callers can
+  // tell how stale their knowledge of the peer really is.
+  getLastInboundAt(): number {
+    return this.lastInboundAt
+  }
+
   noteAuthenticatedInbound(identity: RpcSessionIdentity): void {
     if (this.identity !== identity) {
       return

mobile/src/transport/stable-logical-rpc-client.ts

@@ -147,6 +147,7 @@ export function createStableLogicalRpcClient(
     getState: () => state,
     getReconnectAttempt: () => connectionPath.reconnectAttempt(activeSession.getReconnectAttempt()),
     getLastConnectedAt: () => activeSession.getLastConnectedAt(),
+    getLastInboundAt: () => activeSession.getLastInboundAt?.() ?? null,
     onStateChange(listener) {
       stateListeners.add(listener)
       return () => stateListeners.delete(listener)