fix(mothership): claim a recovering run only once its takeover can start

Recovery claimed the run before reading its replay, resolving billing and
checking access. A takeover that then failed released the lock but had already
spent an attempt and refreshed the run, so an exhausted run whose takeover kept
failing there stayed active, and the refreshed row kept the orphan sweep away.

Claim last: a takeover that fails before it starts leaves the run untouched,
and an exhausted claim always reaches the terminal path.

Author: waleedlatif1

apps/sim/lib/mothership/request/application/recover-stream.ts

@@ -90,25 +90,6 @@ export const readChatStream = defineAuthorizedChatUseCase({
     if (!(await acquirePendingChatStream(chatId, run.streamId, 0))) return run
     const lease = getLocalChatStreamLease(chatId, run.streamId)!
     try {
-      await assertChatStreamLease(lease)
-      if (
-        !(await claimRunController({
-          runId: run.id,
-          chatId,
-          previousToken: saved.controllerToken,
-          token: lease.value,
-          recoveryBackoff: plan.backoff,
-        }))
-      ) {
-        await releasePendingChatStream(chatId, run.streamId, lease)
-        return (await getLatestRunForStream(run.streamId, userId)) ?? run
-      }
-      logger.info('Claimed stream run for recovery', {
-        runId: run.id,
-        streamId: run.streamId,
-        attempt: plan.backoff.attempts,
-        exhausted: plan.kind === 'exhausted',
-      })
       if (isHosted && !config.data.billingAdmission)
         throw new OrchestrationError(
           'forbidden',
@@ -145,6 +126,30 @@ export const readChatStream = defineAuthorizedChatUseCase({
       const recoveredEvents = ringIntact ? events : []
       const lastEvent = recoveredEvents.at(-1)
       const resumeSeq = lastEvent ? lastEvent.seq : ((await getLatestSeq(run.streamId)) ?? 0)
+      /**
+       * Claim last: everything before it can fail without touching the run, so a takeover
+       * that cannot start neither spends the recovery budget nor refreshes the run, and
+       * an exhausted claim always reaches the terminal path below.
+       */
+      await assertChatStreamLease(lease)
+      if (
+        !(await claimRunController({
+          runId: run.id,
+          chatId,
+          previousToken: saved.controllerToken,
+          token: lease.value,
+          recoveryBackoff: plan.backoff,
+        }))
+      ) {
+        await releasePendingChatStream(chatId, run.streamId, lease)
+        return (await getLatestRunForStream(run.streamId, userId)) ?? run
+      }
+      logger.info('Claimed stream run for recovery', {
+        runId: run.id,
+        streamId: run.streamId,
+        attempt: plan.backoff.attempts,
+        exhausted: plan.kind === 'exhausted',
+      })
       const requestId = typeof saved?.requestId === 'string' ? saved.requestId : generateId()
       const completion = {
         chatId,

apps/sim/lib/mothership/request/session/recovery-storm.integration.ts

@@ -2,8 +2,9 @@
  * How often reconnecting to an orphaned Chat run re-POSTs it to the worker, against real
  * Redis and PostgreSQL. The production reconnect route, stream recovery, chat lifecycle
  * and finalization run unmodified; a local HTTP server stands in for the worker. Faults
- * are injected at one seam, the leased replay append: an append that fails once or every
- * time, or a lease lost with no successor to take over.
+ * are injected at two seams: the leased replay append (an append that fails once or every
+ * time, or a lease lost with no successor to take over) and a recovering controller's read
+ * of the replay ring.
  */
 import { authMock, authMockFns } from '@sim/testing/mocks/auth.mock'
 import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'
@@ -19,6 +20,8 @@ const { redisUrl, inheritedEnv, worker, faults } = await vi.hoisted(async () =>
           frames: 'any_tool' | 'tool_result'
           effect: 'throw' | 'throw_once' | 'lose_lease'
         },
+    /** Fails a recovering controller's read of the whole replay ring. */
+    recoveryRead: false,
   }
   const worker = {
     posts: [] as Array<{ path: string; at: number }>,
@@ -88,6 +91,12 @@ vi.mock('@/lib/mothership/request/session/buffer', async (importOriginal) => {
       if (hit && fault.effect === 'lose_lease') await getRedisClient()!.del(lease.key)
       return actual.appendEvents(...args)
     },
+    readEvents: async (...args: Parameters<typeof actual.readEvents>) => {
+      const [, afterCursor] = args
+      if (faults.recoveryRead && afterCursor === '0')
+        throw new Error('simulated Redis read failure')
+      return actual.readEvents(...args)
+    },
   }
 })
 
@@ -420,6 +429,39 @@ describe.runIf(Boolean(redisUrl))('reconnecting to an orphaned Chat run', () =>
     expect(stored.recoveryBackoff).toMatchObject({ attempts: 1 })
   }, 30_000)
 
+  it('leaves the run untouched when a takeover fails before it starts, then ends it once exhausted', async () => {
+    worker.mode = 'frames'
+    const claimedAt = Date.now() - 1_000
+    const recoveryBackoff = { attempts: MAX_RECOVERY_ATTEMPTS, claimedAt, notBefore: claimedAt }
+    const run = await orphanedRun({ recoveryBackoff })
+    const { controllerToken } = (await storedRun(run.runId)).requestContext as {
+      controllerToken: string
+    }
+    faults.recoveryRead = true
+    try {
+      const { turns } = await reconnect(run, { windowMs: 2_000 })
+
+      expect(turns).toHaveLength(0)
+      expect(await storedRun(run.runId)).toMatchObject({
+        status: 'active',
+        requestContext: { controllerToken },
+        recoveryBackoff,
+      })
+    } finally {
+      faults.recoveryRead = false
+    }
+
+    const { turns, stops } = await reconnect(run, { windowMs: 10_000 })
+
+    expect(turns).toHaveLength(0)
+    expect(stops).toHaveLength(1)
+    expect(await storedRun(run.runId)).toMatchObject({
+      status: 'error',
+      error: new StreamRecoveryExhaustedError().userMessage,
+      recoveryBackoff: { attempts: MAX_RECOVERY_ATTEMPTS + 1 },
+    })
+  }, 30_000)
+
   it('never takes over a parked run whose controller still holds the chat lock', async () => {
     worker.mode = 'frames'
     const run = await orphanedRun({ status: 'paused_waiting_for_tool' })