fix(mothership): Stop only cancels detached runs still running, and a held workflow run never blocks the next turn's workbench

Author: waleedlatif1

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

@@ -11,6 +11,7 @@ import {
   copilotRequestStops,
   copilotRuns,
   user,
+  workflowExecutionLogs,
 } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { filterUndefined, toRecordOrNull } from '@sim/utils/object'
@@ -79,6 +80,10 @@ import {
   NO_SECRET_AUTHORITY,
   type TurnSecretAuthority,
 } from '@/lib/mothership/secret-mount-policy'
+import {
+  isWorkflowToolName,
+  WORKFLOW_TOOL_NAMES,
+} from '@/lib/mothership/tools/client-executed-tools'
 import {
   type DesktopToolClaimOwner,
   isDesktopToolCall,
@@ -1753,6 +1758,7 @@ export async function prepareWorkbenchAccess(
         settledAt: copilotAsyncToolCalls.executionSettledAt,
         revokedAt: copilotAsyncToolCalls.executionRevokedAt,
         toolCallId: copilotAsyncToolCalls.toolCallId,
+        executableName: copilotAsyncToolCalls.executableName,
         processes: copilotAsyncToolCalls.sandboxProcesses,
       })
       .from(copilotAsyncToolCalls)
@@ -1820,7 +1826,16 @@ export async function prepareWorkbenchAccess(
     if (processes.some((process) => process.sessionKey !== input.sessionKey))
       throw new Error('Recorded workbench ownership does not match this chat')
     return {
-      handlersPending: previous.some((row) => !row.settledAt && !row.revokedAt),
+      /**
+       * A workflow tool's handler runs a workflow, which never touches the workbench. One answered
+       * still running holds its run past its turn, so it must not hold the workbench too.
+       */
+      handlersPending: previous.some(
+        (row) =>
+          !row.settledAt &&
+          !row.revokedAt &&
+          !(row.executableName && isWorkflowToolName(row.executableName))
+      ),
       processes,
     }
   })
@@ -1925,13 +1940,6 @@ export async function getUnsettledClientWorkflowExecutions(
   )
 }
 
-const WORKFLOW_RUN_TOOL_NAMES = [
-  'run_workflow',
-  'run_workflow_until_block',
-  'run_from_block',
-  'run_block',
-] as const
-
 /**
  * The workflow execution a settled call started and left running: a `workflows run` the CLI
  * answered still running (exit 4 and its run record), or a workflow tool answered still running
@@ -1969,8 +1977,11 @@ export function detachedWorkflowExecutionId(call: {
 }
 
 /**
- * Workflow executions this run's calls started and left running past the call: what a Stop
- * cancels beyond the calls still open, since a detached run takes no call's abort signal.
+ * Workflow executions this run's calls started and left running past the call, and that are still
+ * running: what a Stop cancels beyond the calls still open, since a detached run takes no call's
+ * abort signal. A run whose log has settled or paused is left alone: a paused run waits on a
+ * person, not on this turn, and the durable cancel flag would fail its resume under the same
+ * execution id. A run with no log yet may still be starting, so it is cancelled.
  */
 export async function getDetachedWorkflowExecutions(
   streamId: string,
@@ -1983,23 +1994,45 @@ export async function getDetachedWorkflowExecutions(
     'copilot_async_tool_calls',
     { [TraceAttr.StreamId]: streamId, [TraceAttr.UserId]: userId },
     async () => {
+      const { args, result } = copilotAsyncToolCalls
+      /** Only the fields `detachedWorkflowExecutionId` reads, so no call's payload is loaded. */
       const rows = await executor
         .select({
           executableName: copilotAsyncToolCalls.executableName,
-          args: copilotAsyncToolCalls.args,
-          result: copilotAsyncToolCalls.result,
+          args: sql<unknown>`jsonb_build_object('request', jsonb_build_object('invocation', jsonb_build_object('argv', ${args} #> '{request,invocation,argv}')))`,
+          result: sql<unknown>`jsonb_build_object('exitCode', ${result} -> 'exitCode', 'stdout', ${result} -> 'stdout', 'status', ${result} -> 'status', 'executionId', ${result} -> 'executionId')`,
         })
         .from(copilotAsyncToolCalls)
         .innerJoin(copilotRuns, eq(copilotRuns.id, copilotAsyncToolCalls.runId))
         .where(
           and(
             eq(copilotRuns.streamId, streamId),
             eq(copilotRuns.userId, userId),
-            inArray(copilotAsyncToolCalls.executableName, ['sim_cli', ...WORKFLOW_RUN_TOOL_NAMES]),
-            isNotNull(copilotAsyncToolCalls.result)
+            or(
+              and(
+                eq(copilotAsyncToolCalls.executableName, 'sim_cli'),
+                sql`${result} ->> 'exitCode' = '4'`
+              ),
+              and(
+                inArray(copilotAsyncToolCalls.executableName, [...WORKFLOW_TOOL_NAMES]),
+                sql`${result} ->> 'status' = 'running'`
+              )
+            )
+          )
+        )
+      const detached = [...new Set(rows.flatMap((row) => detachedWorkflowExecutionId(row) ?? []))]
+      if (detached.length === 0) return []
+      const settled = await executor
+        .select({ executionId: workflowExecutionLogs.executionId })
+        .from(workflowExecutionLogs)
+        .where(
+          and(
+            inArray(workflowExecutionLogs.executionId, detached),
+            ne(workflowExecutionLogs.status, 'running')
           )
         )
-      return [...new Set(rows.flatMap((row) => detachedWorkflowExecutionId(row) ?? []))]
+      const notRunning = new Set(settled.map((row) => row.executionId))
+      return detached.filter((executionId) => !notRunning.has(executionId))
     }
   )
 }

apps/sim/lib/mothership/request/tools/workflow-client-settlement.integration.ts

@@ -33,11 +33,14 @@ import { eq, inArray } from 'drizzle-orm'
 import { closeRedisConnection, getRedisClient } from '@/lib/core/config/redis'
 import { SIM_TOOL_EXECUTION_VERSION } from '@/lib/mothership/async-runs/lifecycle'
 import {
+  claimToolExecution,
   claimWorkflowToolExecution,
   completeAsyncToolCall,
   detachAsyncToolCall,
   getDetachedWorkflowExecutions,
+  prepareWorkbenchAccess,
   settleClientWorkflowToolExecution,
+  upsertAsyncToolCall,
 } from '@/lib/mothership/async-runs/repository'
 import { getToolConfirmation } from '@/lib/mothership/persistence/tool-confirm'
 import { waitForWorkflowToolCompletion } from '@/lib/mothership/request/tools/client'
@@ -46,6 +49,7 @@ import {
   reportQueuedClientWorkflowTool,
   reportSettledClientWorkflowTool,
 } from '@/lib/mothership/request/tools/workflow-client-settlement'
+import { chatSandboxSessionKey } from '@/lib/mothership/tools/sandbox-session-key'
 import { PauseResumeManager } from '@/lib/workflows/executor/human-in-the-loop-manager'
 
 /** Longer than the waiter's durable poll, far shorter than the hour it used to park for. */
@@ -532,40 +536,175 @@ describe.runIf(Boolean(redisUrl))('settled client-claimed workflow tools', () =>
   })
 
   it("names the runs this run's calls started and left running, for a Stop to cancel", async () => {
-    const insertCall = async (executableName: string, args: unknown, result: unknown) =>
-      db.insert(copilotAsyncToolCalls).values({
+    const insertCall = async (executableName: string, args: unknown, result: unknown) => {
+      const toolCallId = generateId()
+      await db.insert(copilotAsyncToolCalls).values({
         runId,
-        toolCallId: generateId(),
+        toolCallId,
         toolName: executableName,
         executableName,
         args: args as Record<string, unknown>,
         result: result as Record<string, unknown>,
         status: 'completed',
       })
-    const detachedCli = generateId()
-    const detachedTool = generateId()
+      return toolCallId
+    }
     const record = (runIdValue: string, status: string) =>
       JSON.stringify({ runId: runIdValue, workflowId, status })
-    await insertCall(
-      'sim_cli',
-      { request: { invocation: { kind: 'cli', argv: ['workflows', 'run', workflowId] } } },
-      { exitCode: 4, stdout: record(detachedCli, 'running'), stderr: '' }
+    const detachedCli = generateId()
+    await executionLog(
+      await insertCall(
+        'sim_cli',
+        { request: { invocation: { kind: 'cli', argv: ['workflows', 'run', workflowId] } } },
+        { exitCode: 4, stdout: record(detachedCli, 'running'), stderr: '' }
+      ),
+      detachedCli,
+      'running'
     )
     // A wait on a run proves nothing about who started it.
     await insertCall(
       'sim_cli',
       { request: { invocation: { kind: 'cli', argv: ['workflows', 'runs', 'wait', 'other'] } } },
       { exitCode: 4, stdout: record(generateId(), 'running'), stderr: '' }
     )
+    const detachedTool = generateId()
+    await executionLog(
+      await insertCall(
+        'run_workflow',
+        { workflowId },
+        { status: 'running', executionId: detachedTool, workflowId }
+      ),
+      detachedTool,
+      'running'
+    )
+    // Answered running but not yet logged: it may still be starting, and a Stop must reach it.
+    const startingTool = generateId()
     await insertCall(
-      'run_workflow',
+      'run_from_block',
       { workflowId },
-      { status: 'running', executionId: detachedTool, workflowId }
+      { status: 'running', executionId: startingTool, workflowId }
     )
     await insertCall('run_block', { workflowId }, { success: true, executionId: generateId() })
 
     expect((await getDetachedWorkflowExecutions(streamId, userId)).sort()).toEqual(
-      [detachedCli, detachedTool].sort()
+      [detachedCli, detachedTool, startingTool].sort()
     )
   })
+
+  it.each(['pending', 'completed', 'failed', 'cancelled'] as const)(
+    'leaves a detached run whose log is %s for a Stop to skip',
+    async (logStatus) => {
+      const detachedCli = generateId()
+      const detachedTool = generateId()
+      const cliCall = generateId()
+      const toolCall = generateId()
+      await db.insert(copilotAsyncToolCalls).values([
+        {
+          runId,
+          toolCallId: cliCall,
+          toolName: 'sim_cli',
+          executableName: 'sim_cli',
+          args: {
+            request: { invocation: { kind: 'cli', argv: ['workflows', 'run', workflowId] } },
+          },
+          result: {
+            exitCode: 4,
+            stdout: JSON.stringify({ runId: detachedCli, workflowId, status: 'running' }),
+            stderr: '',
+          },
+          status: 'completed',
+        },
+        {
+          runId,
+          toolCallId: toolCall,
+          toolName: 'run_workflow',
+          executableName: 'run_workflow',
+          args: { workflowId },
+          result: { status: 'running', executionId: detachedTool, workflowId },
+          status: 'completed',
+        },
+      ])
+      // The run went on past its call and then settled, or paused at a human approval: a paused
+      // run's resume subscribes under the same id, so a cancel flag would fail it.
+      await executionLog(cliCall, detachedCli, logStatus)
+      await executionLog(toolCall, detachedTool, logStatus)
+
+      const cancellable = await getDetachedWorkflowExecutions(streamId, userId)
+      expect(cancellable).not.toContain(detachedCli)
+      expect(cancellable).not.toContain(detachedTool)
+    }
+  )
+
+  /**
+   * An earlier turn of a new chat whose `earlierTool` call is still executing under a live lease,
+   * as a workflow tool answered still running is while its run goes on, and a later turn whose
+   * own Sim call asks for the chat's workbench.
+   */
+  async function laterTurnWorkbenchAccess(earlierTool: string) {
+    const turnChatId = generateId()
+    await db.insert(copilotChats).values({
+      id: turnChatId,
+      userId,
+      workspaceId,
+      type: 'mothership',
+      conversationId: generateId(),
+    })
+    try {
+      const turn = async (startedAt: Date, toolName: string) => {
+        const turnRunId = generateId()
+        await db.insert(copilotRuns).values({
+          id: turnRunId,
+          executionId: generateId(),
+          chatId: turnChatId,
+          userId,
+          workspaceId,
+          streamId: generateId(),
+          toolExecutionVersion: SIM_TOOL_EXECUTION_VERSION,
+          status: 'active',
+          startedAt,
+          requestContext: { source: 'headless_lifecycle' },
+        })
+        const owner = {
+          toolCallId: generateId(),
+          runId: turnRunId,
+          userId,
+          ownerToken: generateId(),
+        }
+        await upsertAsyncToolCall({ runId: turnRunId, toolCallId: owner.toolCallId, toolName })
+        expect(await claimToolExecution(owner)).toMatchObject({ outcome: 'claimed' })
+        return owner
+      }
+      const earlier = await turn(new Date(Date.now() - 60_000), earlierTool)
+      await completeAsyncToolCall({
+        toolCallId: earlier.toolCallId,
+        status: 'completed',
+        result: { status: 'running', executionId: generateId(), workflowId },
+      })
+      // The earlier turn answered and ended; its call's execution still holds the run.
+      await db
+        .update(copilotRuns)
+        .set({ status: 'complete', toolAdmissionClosedAt: new Date() })
+        .where(eq(copilotRuns.id, earlier.runId))
+      const later = await turn(new Date(), 'function_execute')
+      return await prepareWorkbenchAccess({
+        ...later,
+        sessionKey: chatSandboxSessionKey(turnChatId),
+      })
+    } finally {
+      await db.delete(copilotChats).where(eq(copilotChats.id, turnChatId))
+    }
+  }
+
+  it("opens the workbench to a later turn while an earlier turn's workflow tool run goes on", async () => {
+    expect(await laterTurnWorkbenchAccess('run_workflow')).toEqual({
+      handlersPending: false,
+      processes: [],
+    })
+  })
+
+  it("keeps the workbench closed while an earlier turn's other tool is still executing", async () => {
+    expect(await laterTurnWorkbenchAccess('function_execute')).toMatchObject({
+      handlersPending: true,
+    })
+  })
 })

apps/sim/lib/mothership/tools/client-executed-tools.ts

@@ -1,14 +1,15 @@
 import { isDesktopToolCall } from '@/lib/mothership/tools/desktop-tools'
 
-const WORKFLOW_TOOL_NAMES = new Set<string>([
+/** The tools that run a workflow, in the browser or on the server. */
+export const WORKFLOW_TOOL_NAMES = [
   'run_workflow',
   'run_workflow_until_block',
   'run_block',
   'run_from_block',
-])
+] as const
 
 export function isWorkflowToolName(name: string): boolean {
-  return WORKFLOW_TOOL_NAMES.has(name)
+  return (WORKFLOW_TOOL_NAMES as readonly string[]).includes(name)
 }
 
 /**