feat(mothership): answer a long workflow tool run still running, and stop detached runs on Stop

apps/sim/app/api/copilot/confirm/route.test.ts

@@ -1036,7 +1036,7 @@ describe('Copilot Confirm API Route', () => {
       status: 'background',
       message: 'Workflow execution is continuing in the background.',
       timestamp: expect.any(String),
-      data: { workflowId: 'workflow-1', executionId: 'unbound-execution' },
+      data: { workflowId: 'workflow-1', executionId: 'unbound-execution', status: 'running' },
     })
   })
 

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

@@ -14,6 +14,7 @@ import {
   closeStreamToolAdmission,
   createRunSegment,
   detachAsyncToolCall,
+  detachedWorkflowExecutionId,
   getClaimedWorkflowExecutionId,
   getUnsettledStreamSandboxProcesses,
   markAsyncToolRunning,
@@ -458,3 +459,60 @@ describe('async tool repository single-row semantics', () => {
     ).rejects.toThrow('Async tool call belongs to another run')
   })
 })
+
+describe('the workflow run a settled call left running', () => {
+  const cli = (argv: string[], exitCode: number, record: Record<string, unknown>) => ({
+    executableName: 'sim_cli',
+    args: { request: { invocation: { kind: 'cli', argv } } },
+    result: { exitCode, stdout: JSON.stringify(record), stderr: '' },
+  })
+  const running = { runId: 'run-1', workflowId: 'wf-1', status: 'running' }
+
+  it('is a workflows run the CLI answered still running', () => {
+    expect(
+      detachedWorkflowExecutionId(
+        cli(['workflows', 'run', 'wf-1', '--wait-timeout', '240'], 4, running)
+      )
+    ).toBe('run-1')
+  })
+
+  it('is a workflows run behind global flags', () => {
+    expect(
+      detachedWorkflowExecutionId(
+        cli(['--output', 'json', 'workflows', 'run', 'wf-1', '--wait-timeout', '240'], 4, running)
+      )
+    ).toBe('run-1')
+  })
+
+  it('is a workflow tool answered still running', () => {
+    expect(
+      detachedWorkflowExecutionId({
+        executableName: 'run_workflow',
+        args: { workflowId: 'wf-1' },
+        result: { status: 'running', executionId: 'exec-1', workflowId: 'wf-1' },
+      })
+    ).toBe('exec-1')
+  })
+
+  it.each([
+    [
+      'a run that settled',
+      cli(['workflows', 'run', 'wf-1'], 0, { ...running, status: 'completed' }),
+    ],
+    // Waiting on a run proves nothing about who started it; Stop must not cancel it.
+    ['a wait on a run', cli(['workflows', 'runs', 'wait', 'run-1'], 4, running)],
+    [
+      'an unreadable answer',
+      {
+        ...cli(['workflows', 'run', 'wf-1'], 4, running),
+        result: { exitCode: 4, stdout: 'not json' },
+      },
+    ],
+    [
+      'a workflow tool that settled',
+      { executableName: 'run_block', args: {}, result: { success: true, executionId: 'exec-1' } },
+    ],
+  ])('is not %s', (_label, call) => {
+    expect(detachedWorkflowExecutionId(call)).toBeUndefined()
+  })
+})

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

@@ -261,6 +261,8 @@ export interface RunStopTarget {
 /** The client workflows and sandbox commands a stopped run started, read once its admission closed. */
 export interface StoppedRunWork {
   clientWorkflowExecutionIds: string[]
+  /** Runs its calls started and left running past the call (`getDetachedWorkflowExecutions`). */
+  detachedWorkflowExecutionIds: string[]
   sandboxProcesses: StoppedSandboxProcess[]
 }
 
@@ -361,6 +363,7 @@ export async function stopLiveRunsTx(
       chatId,
       desktopDeviceId,
       clientWorkflowExecutionIds: await getUnsettledClientWorkflowExecutions(streamId, userId, tx),
+      detachedWorkflowExecutionIds: await getDetachedWorkflowExecutions(streamId, userId, tx),
       sandboxProcesses: await getUnsettledStreamSandboxProcesses(streamId, userId, tx),
     })
   }
@@ -1922,6 +1925,85 @@ 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
+ * (`WorkflowToolRunningResult`). A run the call only waited on (`workflows runs wait`) is not
+ * one: another run started it.
+ */
+export function detachedWorkflowExecutionId(call: {
+  executableName: string | null
+  args: unknown
+  result: unknown
+}): string | undefined {
+  const result = toRecordOrNull(call.result)
+  if (!result) return undefined
+  if (call.executableName === 'sim_cli') {
+    const request = toRecordOrNull(toRecordOrNull(call.args)?.request)
+    const argv = toRecordOrNull(request?.invocation)?.argv
+    // The command path, wherever global flags (`--output json`) put it, never `workflows runs`.
+    const end = Array.isArray(argv) && argv.includes('--') ? argv.indexOf('--') : undefined
+    const tokens = Array.isArray(argv) ? argv.slice(0, end) : []
+    if (!tokens.some((token, index) => token === 'workflows' && tokens[index + 1] === 'run'))
+      return undefined
+    if (result.exitCode !== 4 || typeof result.stdout !== 'string') return undefined
+    try {
+      const record = toRecordOrNull(JSON.parse(result.stdout))
+      return record?.status === 'running' && typeof record.runId === 'string'
+        ? record.runId
+        : undefined
+    } catch {
+      return undefined
+    }
+  }
+  return result.status === 'running' && typeof result.executionId === 'string'
+    ? result.executionId
+    : undefined
+}
+
+/**
+ * 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.
+ */
+export async function getDetachedWorkflowExecutions(
+  streamId: string,
+  userId: string,
+  executor: DbOrTx = db
+) {
+  return await withDbSpan(
+    TraceSpan.CopilotAsyncRunsGetRunSegment,
+    'SELECT',
+    'copilot_async_tool_calls',
+    { [TraceAttr.StreamId]: streamId, [TraceAttr.UserId]: userId },
+    async () => {
+      const rows = await executor
+        .select({
+          executableName: copilotAsyncToolCalls.executableName,
+          args: copilotAsyncToolCalls.args,
+          result: copilotAsyncToolCalls.result,
+        })
+        .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)
+          )
+        )
+      return [...new Set(rows.flatMap((row) => detachedWorkflowExecutionId(row) ?? []))]
+    }
+  )
+}
+
 export function getClaimedWorkflowExecutionId(claimedBy: string | null | undefined) {
   if (!claimedBy?.startsWith(WORKFLOW_EXECUTION_CLAIM_PREFIX)) return undefined
   const executionId = claimedBy.slice(WORKFLOW_EXECUTION_CLAIM_PREFIX.length)

apps/sim/lib/mothership/chat/payload.test.ts

@@ -34,6 +34,7 @@ import {
   ChatPayloadSchema,
   TOOL_DEADLINE_CAPABILITY,
   WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY,
+  WORKFLOW_TOOL_WAIT_TIMEOUT_CAPABILITY,
 } from '@/lib/mothership/generated/protocol'
 import { searchUsersV2Tool } from '@/tools/github/search_users'
 import { gmailListLabelsV2Tool } from '@/tools/gmail/list_labels'
@@ -430,6 +431,7 @@ describe('buildCopilotRequestPayload', () => {
         'workflow-tool-pickup',
         TOOL_DEADLINE_CAPABILITY,
         WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY,
+        WORKFLOW_TOOL_WAIT_TIMEOUT_CAPABILITY,
       ])
     }
   )
@@ -514,6 +516,7 @@ describe('buildCopilotRequestPayload', () => {
       expect(payload.clientCapabilities).toEqual([
         TOOL_DEADLINE_CAPABILITY,
         WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY,
+        WORKFLOW_TOOL_WAIT_TIMEOUT_CAPABILITY,
       ])
       if (owner === 'organization') expect(payload).not.toHaveProperty('integrationTools')
     }
@@ -854,6 +857,7 @@ describe('Assistant payload', () => {
     expect(payload.clientCapabilities).toEqual([
       TOOL_DEADLINE_CAPABILITY,
       WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY,
+      WORKFLOW_TOOL_WAIT_TIMEOUT_CAPABILITY,
     ])
     expect(payload.mode).toBe('assistant')
     expect(payload.assistantSearch).toEqual({ source: 'slack', documentIds: ['document-1'] })

apps/sim/lib/mothership/chat/payload.ts

@@ -38,6 +38,7 @@ import {
   type ModelSelection,
   TOOL_DEADLINE_CAPABILITY,
   WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY,
+  WORKFLOW_TOOL_WAIT_TIMEOUT_CAPABILITY,
 } from '@/lib/mothership/generated/protocol'
 import type { VfsSnapshotV1 } from '@/lib/mothership/generated/vfs-snapshot-v1'
 import { getToolEntry } from '@/lib/mothership/tool-executor/router'
@@ -562,13 +563,15 @@ export async function buildCopilotRequestPayload(params: BuildPayloadParams): Pr
       : {}),
     // The mounted chat view executes client-routed workflow tools (run panel UX), so the
     // UI declares that capability explicitly; headless callers omit it and the server runs
-    // those tools immediately. The deadline and run-wait tokens are this server's own: it
-    // bounds every handed call by the deadline its frame names, and its CLI lets a
-    // synchronous `workflows run` outlast `--wait-timeout`.
+    // those tools immediately. The deadline and wait tokens are this server's own: it
+    // bounds every handed call by the deadline its frame names, its CLI lets a synchronous
+    // `workflows run` outlast `--wait-timeout`, and a workflow tool's run outlasts the
+    // `waitTimeoutMs` its frame names.
     clientCapabilities: [
       ...(isAssistant ? [] : ['workflow-tool-pickup']),
       TOOL_DEADLINE_CAPABILITY,
       WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY,
+      WORKFLOW_TOOL_WAIT_TIMEOUT_CAPABILITY,
     ],
   }
 }

apps/sim/lib/mothership/generated/protocol.ts

@@ -297,6 +297,14 @@ export interface StreamToolReplay {
  */
 export interface StreamToolDeadline {
   deadlineMs?: number | undefined;
+  /**
+   * For a workflow tool (`run_workflow`, `run_workflow_until_block`, `run_from_block`,
+   * `run_block`) on a Sim that declares {@link WORKFLOW_TOOL_WAIT_TIMEOUT_CAPABILITY}: how long
+   * Sim waits for the call's run to settle once it starts waiting on it. A run still going then
+   * keeps going and the call settles with a {@link WorkflowToolRunningResult}. Absent = held
+   * until the run settles.
+   */
+  waitTimeoutMs?: number | undefined;
 }
 
 /**
@@ -322,6 +330,24 @@ export const WORKSPACE_FILE_COPY_CAPABILITY = "workspace-file-copy";
  */
 export const WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY = "workflow-run-wait-timeout";
 
+/**
+ * The `clientCapabilities` token a Sim sends when it honors `StreamToolDeadline.waitTimeoutMs`
+ * on a workflow tool call: a run that outlasts it continues and the call settles with a
+ * {@link WorkflowToolRunningResult}, so the worker waits on the run in slices.
+ */
+export const WORKFLOW_TOOL_WAIT_TIMEOUT_CAPABILITY = "workflow-tool-wait-timeout";
+
+/**
+ * A workflow tool call's output when its run outlasted the call's `waitTimeoutMs`: the run is
+ * still going under `executionId`, readable as a run of `workflowId` (`workflows runs wait`).
+ */
+export interface WorkflowToolRunningResult {
+  status: "running";
+  executionId: string;
+  workflowId: string;
+  message?: string | undefined;
+}
+
 /** POST /api/mothership — the chat request sim sends. */
 export interface ChatRequest extends StreamResponseReceipt {
   fileOwnerProtocolVersion?: 1 | undefined;
@@ -367,8 +393,9 @@ export interface ChatRequest extends StreamResponseReceipt {
    * grace windows and run tools server-side immediately. ABSENT = older/unknown caller —
    * dispatch keeps its conservative grace (deploy-skew safe: a stale tab that predates
    * this field still gets waited on). Known capability: "workflow-tool-pickup". Sim also
-   * declares its own {@link TOOL_DEADLINE_CAPABILITY}, {@link WORKSPACE_FILE_COPY_CAPABILITY} and
-   * {@link WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY} here.
+   * declares its own {@link TOOL_DEADLINE_CAPABILITY}, {@link WORKSPACE_FILE_COPY_CAPABILITY},
+   * {@link WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY} and {@link WORKFLOW_TOOL_WAIT_TIMEOUT_CAPABILITY}
+   * here.
    */
   clientCapabilities?: string[] | undefined;
   /**

apps/sim/lib/mothership/request/application/stop-delivery.ts

@@ -5,6 +5,7 @@ import { markExecutionCancelled } from '@/lib/execution/cancellation'
 import { abortManualExecution } from '@/lib/execution/manual-cancellation'
 import { settleStoppedRunWithoutController } from '@/lib/mothership/async-runs/orphaned-runs'
 import {
+  getDetachedWorkflowExecutions,
   getUnsettledClientWorkflowExecutions,
   getUnsettledStreamSandboxProcesses,
   type RunStopTarget,
@@ -34,30 +35,39 @@ interface RunStopDeliveryOptions {
 }
 
 /**
- * The client workflows and sandbox commands a stopped run started, read once its admission
- * closed. Each read is best-effort: one that fails stops nothing it would have found.
+ * The client workflows, detached workflow runs and sandbox commands a stopped run started, read
+ * once its admission closed. Each read is best-effort: one that fails stops nothing it would have
+ * found.
  */
 export async function readStoppedRunWork(
   streamId: string,
   userId: string
 ): Promise<StoppedRunWork> {
-  const [clientWorkflowExecutionIds, sandboxProcesses] = await Promise.all([
-    getUnsettledClientWorkflowExecutions(streamId, userId).catch((error) => {
-      logger.warn('Stopped stream workflow ownership could not be read', {
-        streamId,
-        error: getErrorMessage(error),
-      })
-      return []
-    }),
-    getUnsettledStreamSandboxProcesses(streamId, userId).catch((error) => {
-      logger.warn('Stopped stream sandbox ownership could not be read', {
-        streamId,
-        error: getErrorMessage(error),
-      })
-      return []
-    }),
-  ])
-  return { clientWorkflowExecutionIds, sandboxProcesses }
+  const [clientWorkflowExecutionIds, detachedWorkflowExecutionIds, sandboxProcesses] =
+    await Promise.all([
+      getUnsettledClientWorkflowExecutions(streamId, userId).catch((error) => {
+        logger.warn('Stopped stream workflow ownership could not be read', {
+          streamId,
+          error: getErrorMessage(error),
+        })
+        return []
+      }),
+      getDetachedWorkflowExecutions(streamId, userId).catch((error) => {
+        logger.warn('Stopped stream detached workflow runs could not be read', {
+          streamId,
+          error: getErrorMessage(error),
+        })
+        return []
+      }),
+      getUnsettledStreamSandboxProcesses(streamId, userId).catch((error) => {
+        logger.warn('Stopped stream sandbox ownership could not be read', {
+          streamId,
+          error: getErrorMessage(error),
+        })
+        return []
+      }),
+    ])
+  return { clientWorkflowExecutionIds, detachedWorkflowExecutionIds, sandboxProcesses }
 }
 
 /**
@@ -97,14 +107,21 @@ export async function deliverRunStop(
       error: getErrorMessage(error),
     })
   }
-  const cancelClientWorkflows = async () =>
-    Promise.all(
-      (await work).clientWorkflowExecutionIds.map(async (executionId) => {
-        /** The execution route reserved these IDs before recording ownership; no client result is trusted. */
-        abortManualExecution(executionId)
-        await markExecutionCancelled(executionId)
-      })
+  const cancelClientWorkflows = async () => {
+    const { clientWorkflowExecutionIds, detachedWorkflowExecutionIds } = await work
+    /**
+     * The execution route reserved the client IDs before recording ownership; no client result is
+     * trusted. A detached run took no call's abort signal, so only its own cancel reaches it.
+     */
+    return Promise.all(
+      [...new Set([...clientWorkflowExecutionIds, ...detachedWorkflowExecutionIds])].map(
+        async (executionId) => {
+          abortManualExecution(executionId)
+          await markExecutionCancelled(executionId)
+        }
+      )
     )
+  }
   const [settled, worker] = await Promise.all([
     waitForPendingChatStream(chatId, 8000, streamId),
     workerStop,