refactor(mothership): require the recorded Stop inside the stopped-run settle

- Settling a stopped run now passes the Stop-row check as the update's own
  guard, so it cannot cancel a run nobody stopped; the separate stopped
  branch is gone and every settle derives cancelled from the Stop row.
- Chat lock ownership is read through getChatStreamLockOwners and trusted
  only when verified, instead of a second Redis read of the same keys.
- Settle transactions use the shared DbTransaction type.

Author: waleedlatif1

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

@@ -48,7 +48,7 @@ import {
   settleStoppedRunWithoutController,
   sweepOrphanedRuns,
 } from '@/lib/mothership/async-runs/orphaned-runs'
-import { updateRunStatus } from '@/lib/mothership/async-runs/repository'
+import { requestRunStop, updateRunStatus } from '@/lib/mothership/async-runs/repository'
 import { abortRun } from '@/lib/mothership/request/application/controls'
 import { claimRunController } from '@/lib/mothership/request/lifecycle/controller-ownership'
 import { chatStreamLockKey } from '@/lib/mothership/request/session/controller-lease'
@@ -160,6 +160,11 @@ describe.runIf(Boolean(redisUrl))('Chat runs no controller owns', () => {
     return { chatId, streamId, runId, controllerToken }
   }
 
+  /** Records the user's Stop the way the abort use case does before it settles anything. */
+  async function stop(run: { streamId: string; chatId: string }) {
+    await requestRunStop({ userId, workspaceId, streamId: run.streamId, chatId: run.chatId })
+  }
+
   async function stored(runId: string) {
     const [run] = await db.select().from(copilotRuns).where(eq(copilotRuns.id, runId))
     const [chat] = await db
@@ -322,8 +327,19 @@ describe.runIf(Boolean(redisUrl))('Chat runs no controller owns', () => {
     expect(run.marker).toBeNull()
   })
 
+  it('never cancels a run nobody stopped', async () => {
+    const orphan = await admittedRun({ superseded: true })
+
+    expect(await settleStoppedRunWithoutController(orphan.runId)).toBe(false)
+
+    const run = await stored(orphan.runId)
+    expect(run.status).toBe('active')
+    expect(run.marker).toBe(orphan.streamId)
+  })
+
   it('leaves a stopped run to the controller of its stream that holds the chat lock', async () => {
     const owned = await admittedRun()
+    await stop(owned)
     await redis().set(chatStreamLockKey(owned.chatId), owned.controllerToken!, 'EX', 60)
 
     try {
@@ -366,6 +382,7 @@ describe.runIf(Boolean(redisUrl))('Chat runs no controller owns', () => {
   it('settles a stopped run exactly once when Stop races its own controller finalizing', async () => {
     for (let attempt = 0; attempt < 50; attempt++) {
       const orphan = await admittedRun()
+      await stop(orphan)
 
       const [finalized, stopped] = await Promise.all([
         updateRunStatus(orphan.runId, 'complete', {}, orphan.controllerToken!),
@@ -380,6 +397,7 @@ 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 () => {
     for (let attempt = 0; attempt < 50; attempt++) {
       const orphan = await admittedRun()
+      await stop(orphan)
 
       const [claimed, stopped] = await Promise.all([
         claimRunController({

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

@@ -23,10 +23,11 @@ import {
   type SQL,
   sql,
 } from 'drizzle-orm'
+import type { DbTransaction } from '@/lib/db/types'
 import { SIM_TOOL_EXECUTION_VERSION } from '@/lib/mothership/async-runs/lifecycle'
 import { publishChatStatusChanged } from '@/lib/mothership/chat-status'
+import { getChatStreamLockOwners } from '@/lib/mothership/request/session/abort'
 import { findStreamsWithReplay } from '@/lib/mothership/request/session/buffer'
-import { findStreamsHoldingChatLock } from '@/lib/mothership/request/session/controller-lease'
 
 const logger = createLogger('OrphanedCopilotRuns')
 
@@ -110,13 +111,8 @@ const unownedRunColumns = {
   controllerToken,
 }
 
-type Transaction = Parameters<Parameters<typeof db.transaction>[0]>[0]
-
-/** A stopped run ends cancelled; the sweep ends it cancelled only if its user pressed Stop. */
-function terminalValues(reason: 'stopped' | 'orphaned' | 'legacy') {
-  if (reason === 'stopped') {
-    return { status: sql`'cancelled'::copilot_run_status`, error: sql`NULL::text` }
-  }
+/** A run ends cancelled only if its user pressed Stop, whoever settles it. */
+function terminalValues(reason: 'orphaned' | 'legacy') {
   const error = reason === 'orphaned' ? ORPHANED_RUN_ERROR : LEGACY_RUN_ERROR
   return {
     status: sql`(CASE WHEN ${stopRequested} THEN 'cancelled' ELSE 'error' END)::copilot_run_status`,
@@ -135,9 +131,8 @@ function terminalValues(reason: 'stopped' | 'orphaned' | 'legacy') {
  * touching the chat's ordering timestamp.
  */
 async function settleRuns(
-  tx: Transaction,
+  tx: DbTransaction,
   runs: UnownedRun[],
-  reason: 'stopped' | 'orphaned',
   guard: SQL | undefined
 ): Promise<UnownedRun[]> {
   if (runs.length === 0) return []
@@ -182,7 +177,7 @@ async function settleRuns(
   for (const run of runs) {
     if (run.controllerToken === null) continue
     await apply([run], eq(controllerToken, run.controllerToken), {
-      ...terminalValues(reason),
+      ...terminalValues('orphaned'),
       completedAt: sql`now()`,
       updatedAt: sql`now()`,
     })
@@ -217,6 +212,23 @@ function announceSettled(runs: UnownedRun[]): void {
   }
 }
 
+/**
+ * The streams among these whose own controller holds its chat lock, under any token: a
+ * recovering controller locks the chat before it claims the run. Throws unless the
+ * locks were read, since otherwise no stream is provably unowned.
+ */
+async function findStreamsHoldingChatLock(
+  runs: Array<{ chatId: string; streamId: string }>
+): Promise<Set<string>> {
+  const { status, ownersByChatId } = await getChatStreamLockOwners([
+    ...new Set(runs.map((run) => run.chatId)),
+  ])
+  if (status !== 'verified') throw new Error('Chat stream locks are unreadable')
+  return new Set(
+    runs.filter((run) => ownersByChatId.get(run.chatId) === run.streamId).map((run) => run.streamId)
+  )
+}
+
 /** The candidates no controller owns; leased runs are skipped when ownership is unreadable. */
 async function withoutOwners(candidates: UnownedRun[]): Promise<UnownedRun[]> {
   const leased = candidates.filter((run) => run.controllerToken !== null)
@@ -268,7 +280,7 @@ export async function sweepOrphanedRuns(): Promise<{ settledRunIds: string[] }>
 
     try {
       const unowned = await withoutOwners(candidates)
-      const settled = await db.transaction((tx) => settleRuns(tx, unowned, 'orphaned', orphanIdle))
+      const settled = await db.transaction((tx) => settleRuns(tx, unowned, orphanIdle))
       announceSettled(settled.filter((run) => run.controllerToken !== null))
       settledRunIds.push(...settled.map((run) => run.id))
     } catch (error) {
@@ -289,7 +301,8 @@ export async function sweepOrphanedRuns(): Promise<{ settledRunIds: string[] }>
 
 /**
  * Settles a stopped run as cancelled when no controller of its stream holds the chat
- * lock. A live controller observes the Stop and settles its own run.
+ * lock. A live controller observes the Stop and settles its own run. The update itself
+ * requires the recorded Stop, so this can never settle a run nobody stopped.
  */
 export async function settleStoppedRunWithoutController(runId: string): Promise<boolean> {
   const [run] = await db
@@ -299,7 +312,7 @@ export async function settleStoppedRunWithoutController(runId: string): Promise<
     .limit(1)
   if (!run?.controllerToken) return false
   if ((await findStreamsHoldingChatLock([run])).has(run.streamId)) return false
-  const settled = await db.transaction((tx) => settleRuns(tx, [run], 'stopped', undefined))
+  const settled = await db.transaction((tx) => settleRuns(tx, [run], stopRequested))
   announceSettled(settled)
   return settled.length > 0
 }

apps/sim/lib/mothership/request/session/controller-lease.ts

@@ -28,26 +28,6 @@ export async function assertChatStreamLease(lease: ChatStreamLease): Promise<voi
   }
 }
 
-/**
- * The streams among these whose own controller holds its chat lock, under any token: a
- * recovering controller locks the chat before it claims the run. Throws when the locks
- * cannot be read, since then no stream is provably unowned.
- */
-export async function findStreamsHoldingChatLock(
-  streams: Array<{ chatId: string; streamId: string }>
-): Promise<Set<string>> {
-  const redis = getRedisClient()
-  if (!redis) throw new Error('Chat stream locks are unreadable without Redis')
-  if (streams.length === 0) return new Set()
-  const values = await redis.mget(streams.map(({ chatId }) => chatStreamLockKey(chatId)))
-  const held = new Set<string>()
-  streams.forEach(({ streamId }, index) => {
-    const value = values[index]
-    if (value && streamIdFromLock(value) === streamId) held.add(streamId)
-  })
-  return held
-}
-
 /** Whether this lease still holds its chat lock; an unreadable lock counts as lost. */
 export async function holdsChatStreamLease(lease: ChatStreamLease): Promise<boolean> {
   try {