fix(mothership): harden abandoned-run takeover

- Add the `mothership-abandoned-run-takeover-disabled` AppConfig kill switch,
  checked once per scan; off by default, so takeovers run unless it is on.
- Stop a controller whose chat lock lapsed while every renewal threw, so a
  controller partitioned from Redis cannot keep driving a run taken over.
- Never revoke another call awaiting its re-run from a sibling's wait or
  workbench access; the sibling's workbench access still stops that call's
  dead attempt's commands.
- Treat a workspace access loss found inside recovery like an authorization
  denial, by error type; other recovery failures stay recoverable.
- Cover organization Assistant takeover for a member and a removed member.

Author: waleedlatif1

apps/sim/lib/core/config/feature-flags.ts

@@ -120,6 +120,13 @@ const FEATURE_FLAGS = {
       'targeting. Off-AppConfig falls back to KNOWLEDGE_MEMBER_ACCESS.',
     fallback: 'KNOWLEDGE_MEMBER_ACCESS',
   },
+  'mothership-abandoned-run-takeover-disabled': {
+    description:
+      'Kill switch for Sim taking over Chat runs whose controller died (a deploy or a crash) and ' +
+      're-running their interrupted tool. Global on/off only; on stops every takeover scan. ' +
+      'Off by default, so takeovers run unless this is turned on.',
+    fallback: null,
+  },
 } satisfies Record<string, FeatureFlagDefinition>
 
 /**

apps/sim/lib/desktop/executor/bound-turn.integration.ts

@@ -511,7 +511,7 @@ describe.runIf(Boolean(redisUrl))("a turn bound to a desktop's background execut
 
       await lapse(toolCallId, 'lease')
       /** A Sim tool waiting on the same run sweeps its expired executions; this one is not its. */
-      await revokeExpiredSimToolExecutions({ runId: run.runId, userId })
+      await revokeExpiredSimToolExecutions({ runId: run.runId, userId, toolCallId: 'waiting-call' })
       expect((await storedCall(toolCallId)).status).toBe('running')
       await answer
       const late = await desktop.complete(toolCallId, executionToken, { output: 'built' })

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

@@ -24,6 +24,7 @@ import {
   like,
   lt,
   ne,
+  not,
   notInArray,
   or,
   type SQL,
@@ -757,6 +758,18 @@ const interruptedSimExecution = and(
   lt(copilotAsyncToolCalls.executionAttempts, SIM_TOOL_EXECUTION_MAX_ATTEMPTS)
 )
 
+/**
+ * An interrupted execution a takeover may still run again: its run still admits tools. Another
+ * call's wait or workbench access must not revoke it as interrupted before that re-run.
+ */
+const rerunnableSimExecution = and(
+  interruptedSimExecution,
+  sql`EXISTS (SELECT 1 FROM ${copilotRuns} r WHERE r.id = ${copilotAsyncToolCalls.runId} AND r.tool_admission_closed_at IS NULL AND r.status NOT IN (${sql.join(
+    TERMINAL_RUN_STATUSES.map((status) => sql`${status}`),
+    sql`, `
+  )}))`
+)
+
 /**
  * Sim claims a call it is about to execute under an execution lease; a terminal result never
  * releases it. The claim takes a call's first execution, or re-runs one whose owner died
@@ -994,12 +1007,23 @@ export async function hasLiveSimToolExecution(runId: string): Promise<boolean> {
   return !!live
 }
 
-/** Revocation ends local execution authority; recorded remote commands remain independently unsettled. */
-export async function revokeExpiredSimToolExecutions(input: { runId: string; userId: string }) {
+/**
+ * Revocation ends local execution authority; recorded remote commands remain independently
+ * unsettled. The waiting call's own lapsed execution is revoked; another call's is spared while
+ * a takeover may still run it again.
+ */
+export async function revokeExpiredSimToolExecutions(input: {
+  runId: string
+  userId: string
+  toolCallId: string
+}) {
   return db.transaction((tx) =>
     revokeExpiredExecutions(
       tx,
-      sql`EXISTS (SELECT 1 FROM ${copilotRuns} r WHERE r.id = ${copilotAsyncToolCalls.runId} AND r.id = ${input.runId}::uuid AND r.user_id = ${input.userId})`
+      and(
+        sql`EXISTS (SELECT 1 FROM ${copilotRuns} r WHERE r.id = ${copilotAsyncToolCalls.runId} AND r.id = ${input.runId}::uuid AND r.user_id = ${input.userId})`,
+        or(eq(copilotAsyncToolCalls.toolCallId, input.toolCallId), not(rerunnableSimExecution))
+      )
     )
   )
 }
@@ -1358,7 +1382,11 @@ export async function prepareWorkbenchAccess(
       )
     await revokeExpiredExecutions(
       tx,
-      sql`EXISTS (SELECT 1 FROM ${copilotRuns} r WHERE r.id = ${copilotAsyncToolCalls.runId} AND r.chat_id = ${owner.chatId} AND r.user_id = ${input.userId})`
+      and(
+        sql`EXISTS (SELECT 1 FROM ${copilotRuns} r WHERE r.id = ${copilotAsyncToolCalls.runId} AND r.chat_id = ${owner.chatId} AND r.user_id = ${input.userId})`,
+        /** This run was just verified to admit tools; every older run's admission just closed. */
+        not(and(eq(copilotAsyncToolCalls.runId, input.runId), interruptedSimExecution))
+      )
     )
     const previous = await tx
       .select({
@@ -1393,6 +1421,20 @@ export async function prepareWorkbenchAccess(
      * A re-run's earlier attempts may have left commands running on the remote workbench, which
      * outlive the Sim process that started them; they are stopped like any earlier call's.
      */
+    /** A sibling awaiting its re-run has no live handler, but its dead attempt's commands may still run. */
+    const rerunnableSiblings = await tx
+      .select({
+        toolCallId: copilotAsyncToolCalls.toolCallId,
+        processes: copilotAsyncToolCalls.sandboxProcesses,
+      })
+      .from(copilotAsyncToolCalls)
+      .where(
+        and(
+          eq(copilotAsyncToolCalls.runId, input.runId),
+          ne(copilotAsyncToolCalls.toolCallId, input.toolCallId),
+          interruptedSimExecution
+        )
+      )
     const earlierAttempts = {
       toolCallId: input.toolCallId,
       processes: Object.fromEntries(
@@ -1401,7 +1443,7 @@ export async function prepareWorkbenchAccess(
         )
       ),
     }
-    const processes = [...previous, earlierAttempts].flatMap((row) =>
+    const processes = [...previous, ...rerunnableSiblings, earlierAttempts].flatMap((row) =>
       Object.entries(row.processes).flatMap(([id, process]) =>
         process.settled
           ? []
@@ -1594,9 +1636,10 @@ export async function claimWorkflowToolExecution(
                 /** Only the server re-runs a workflow whose Sim executor died; no browser resumes it. */
                 executor === 'sim' ? interruptedSimExecution : undefined,
                 /**
-                 * A server claim whose claimant died before starting the execution. Starting is
-                 * fenced by `claimToolExecution`, so a claimant still alive and this one cannot
-                 * both run it: the second to start observes the first's result.
+                 * A server claim that never started, such as one whose claimant died between
+                 * claiming and starting. Nothing here checks the claimant is gone: what keeps
+                 * the call to one execution is `claimToolExecution`, the only start, so if the
+                 * earlier claimant is alive whichever starts second observes the first's result.
                  */
                 executor === 'sim'
                   ? and(

apps/sim/lib/mothership/persistence/tool-confirm/index.ts

@@ -156,7 +156,7 @@ export async function waitForToolConfirmation(
     const checkDurableConfirmation = async (source: 'subscribe' | 'pubsub' | 'poll') => {
       if (options.executionScope) {
         try {
-          await revokeExpiredSimToolExecutions(options.executionScope)
+          await revokeExpiredSimToolExecutions({ ...options.executionScope, toolCallId })
         } catch (error) {
           logger.warn('Could not verify retained tool execution ownership', {
             toolCallId,

apps/sim/lib/mothership/persistence/tool-confirm/tool-confirm.test.ts

@@ -308,7 +308,10 @@ describe('copilot orchestrator persistence', () => {
       })
       await vi.advanceTimersByTimeAsync(5_000)
       expect(await waiting).toMatchObject({ status: 'error', message: 'Outcome unknown' })
-      expect(revokeExpiredSimToolExecutions).toHaveBeenCalledWith(scope)
+      expect(revokeExpiredSimToolExecutions).toHaveBeenCalledWith({
+        ...scope,
+        toolCallId: 'tool-1',
+      })
     } finally {
       vi.useRealTimers()
     }