fix(mothership): claim under the chat lock in the Stop-versus-recovery settlement test (#8728)

* fix(mothership): claim under the chat lock in the Stop-versus-recovery settlement test

The test raced Stop's settle against a recovering controller that claimed the run without
holding the chat lock. Recovery always locks the chat, proves its lease, then claims, and
settleStoppedRunWithoutController treats a run with no lock holder as unowned. So when the
claim committed before Stop read the run, which a loaded runner does, Stop read the new
controller token, found no lock holder and cancelled a run the test had just claimed: both
sides reported success. With the lock held, the same ordering leaves the run to its
controller.

The recovering controller now locks the chat and proves its lease before claiming, as
recover-stream does, and releases the lock when done. Both orderings are asserted
deterministically before the concurrent loop, which keeps its exactly-once assertion.

* fix(mothership): guard the recovering controller's lease instead of asserting it non-null

* fix(mothership): guard the run's controller token in the recovery helper and shrink the non-null baseline

Author: waleedlatif1

apps/sim/lib/mothership/async-runs/orphaned-runs.integration.ts

@@ -565,23 +565,60 @@ describe.runIf(Boolean(redisUrl))('Chat runs no controller owns', () => {
   })
 
   it('settles a stopped run exactly once when Stop races a recovering controller claiming it', async () => {
+    /**
+     * Claims the run as recovery does: lock the chat under the run's stream, prove the lease,
+     * then claim, so the claimed run has a lock holder for as long as its controller lives.
+     */
+    const recover = async (orphan: Awaited<ReturnType<typeof admittedRun>>) => {
+      const previousToken = orphan.controllerToken
+      if (!previousToken)
+        throw new Error('Recovery needs the run it takes over to have a controller')
+      if (!(await acquirePendingChatStream(orphan.chatId, orphan.streamId, 0))) return false
+      const lease = getLocalChatStreamLease(orphan.chatId, orphan.streamId)
+      if (!lease) throw new Error('Recovery acquired the chat lock without a local lease')
+      await assertChatStreamLease(lease)
+      const claimed = await claimRunController({
+        runId: orphan.runId,
+        chatId: orphan.chatId,
+        previousToken,
+        token: lease.value,
+        recoveryBackoff: FIRST_RECOVERY,
+      })
+      if (!claimed) await releasePendingChatStream(orphan.chatId, orphan.streamId, lease)
+      return claimed
+    }
+    /** Ends a recovered controller, as its stream finishing would. */
+    const release = async (orphan: Awaited<ReturnType<typeof admittedRun>>) => {
+      const lease = getLocalChatStreamLease(orphan.chatId, orphan.streamId)
+      if (lease) await releasePendingChatStream(orphan.chatId, orphan.streamId, lease)
+    }
+
+    /** A claim committed before Stop reads the run is the interleaving a loaded runner hits. */
+    const claimedFirst = await admittedRun()
+    await stop(claimedFirst)
+    expect(await recover(claimedFirst)).toBe(true)
+    expect(await settleStoppedRunWithoutController(claimedFirst.runId)).toBe(false)
+    expect((await stored(claimedFirst.runId)).status).toBe('active')
+    await release(claimedFirst)
+
+    const stoppedFirst = await admittedRun()
+    await stop(stoppedFirst)
+    expect(await settleStoppedRunWithoutController(stoppedFirst.runId)).toBe(true)
+    expect(await recover(stoppedFirst)).toBe(false)
+    expect((await stored(stoppedFirst.runId)).status).toBe('cancelled')
+
     for (let attempt = 0; attempt < 50; attempt++) {
       const orphan = await admittedRun()
       await stop(orphan)
 
       const [claimed, stopped] = await Promise.all([
-        claimRunController({
-          runId: orphan.runId,
-          chatId: orphan.chatId,
-          previousToken: orphan.controllerToken!,
-          token: `${orphan.streamId}\n${generateId()}`,
-          recoveryBackoff: FIRST_RECOVERY,
-        }),
+        recover(orphan),
         settleStoppedRunWithoutController(orphan.runId),
       ])
 
       expect(claimed !== stopped).toBe(true)
       expect((await stored(orphan.runId)).status).toBe(stopped ? 'cancelled' : 'active')
+      await release(orphan)
     }
   })
 

scripts/check-explicit-any.baseline.json

@@ -1592,7 +1592,7 @@
     "apps/sim/lib/mothership/agent-cli/workbench-file-provenance.integration.ts": 4,
     "apps/sim/lib/mothership/agent-cli/workbench-file-provenance.ts": 2,
     "apps/sim/lib/mothership/assistant/tool-policy.test.ts": 2,
-    "apps/sim/lib/mothership/async-runs/orphaned-runs.integration.ts": 11,
+    "apps/sim/lib/mothership/async-runs/orphaned-runs.integration.ts": 10,
     "apps/sim/lib/mothership/auth/internal.ts": 1,
     "apps/sim/lib/mothership/billing/service-store.integration.ts": 3,
     "apps/sim/lib/mothership/chat/application/context.ts": 1,