dot fixes

Author: Sg312

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

@@ -11,6 +11,7 @@ import {
   completeAsyncToolCall,
   detachAsyncToolCall,
   getClaimedWorkflowExecutionId,
+  markAsyncToolRunning,
   recordToolPermissionDecision,
   releaseWorkflowToolExecutionClaim,
   replaceTerminalAsyncToolCallResult,
@@ -162,6 +163,29 @@ describe('async tool repository single-row semantics', () => {
     await expect(claimWorkflowToolExecution('workflow-tool', 'execution-2')).resolves.toBeNull()
   })
 
+  it('overwrites a workflow execution claim once the sim path starts running it', async () => {
+    // The server-side fallback claims `workflow:<id>` and then immediately runs
+    // the tool, whose executor re-marks the row as running under 'sim-stream'.
+    // The claim value is therefore NOT durable identity — only its
+    // `claimedBy IS NULL` precondition is load-bearing, since that is what keeps
+    // a late browser locked out. Pinning this so nobody builds on reading it back.
+    dbChainMockFns.returning.mockResolvedValueOnce([
+      {
+        toolCallId: 'workflow-tool',
+        status: 'running',
+        claimedBy: 'sim-stream',
+      },
+    ])
+
+    const result = await markAsyncToolRunning('workflow-tool', 'sim-stream')
+
+    expect(result).toMatchObject({ claimedBy: 'sim-stream' })
+    expect(dbChainMockFns.set).toHaveBeenCalledWith(
+      expect.objectContaining({ claimedBy: 'sim-stream' })
+    )
+    expect(getClaimedWorkflowExecutionId('sim-stream')).toBeUndefined()
+  })
+
   it('releases a matching pre-start workflow claim without changing its lifecycle status', async () => {
     dbChainMockFns.returning.mockResolvedValueOnce([
       {

apps/sim/lib/copilot/constants.ts

@@ -36,6 +36,19 @@ export const TOOL_WATCHDOG_RESUME_GRACE_MS = 30_000
 /** Timeout for the client-side streaming response handler (60 min). */
 export const STREAM_TIMEOUT_MS = 3_600_000
 
+/**
+ * How long a workflow tool call waits for a browser to pick it up before the
+ * server runs it itself.
+ *
+ * Workflow tools are client-routed, but the only thing that starts one is the
+ * mounted chat view — a call frame that arrives while the user is on a
+ * different chat is never dispatched by anyone, and the turn used to park for
+ * the full STREAM_TIMEOUT_MS. The real pickup path (stream frame -> execute
+ * POST -> claim) lands in ~1-3s, so 30s is an order of magnitude of headroom
+ * and cannot steal work from a live tab.
+ */
+export const COPILOT_WORKFLOW_TOOL_CLIENT_GRACE_MS = 30_000
+
 /** SessionStorage key for persisting active stream metadata across page reloads. */
 export const STREAM_STORAGE_KEY = 'copilot_active_stream'
 

apps/sim/lib/copilot/request/handlers/handlers.test.ts

@@ -15,10 +15,16 @@ const { isSimExecuted, executeTool, ensureHandlersRegistered, toolRequiresApprov
   })
 )
 
-const { upsertAsyncToolCall, markAsyncToolRunning, completeAsyncToolCall } = vi.hoisted(() => ({
+const {
+  upsertAsyncToolCall,
+  markAsyncToolRunning,
+  completeAsyncToolCall,
+  claimWorkflowToolExecution,
+} = vi.hoisted(() => ({
   upsertAsyncToolCall: vi.fn(),
   markAsyncToolRunning: vi.fn(),
   completeAsyncToolCall: vi.fn(),
+  claimWorkflowToolExecution: vi.fn().mockResolvedValue(null),
 }))
 
 const { waitForClientToolCompletion, waitForToolCompletion, waitForWorkflowToolCompletion } =
@@ -56,6 +62,7 @@ vi.mock('@/lib/copilot/async-runs/repository', () => ({
   upsertAsyncToolCall,
   markAsyncToolRunning,
   completeAsyncToolCall,
+  claimWorkflowToolExecution,
 }))
 
 vi.mock('@/lib/copilot/request/tools/client', () => ({
@@ -578,11 +585,13 @@ describe('sse-handlers tool lifecycle', () => {
     await sleep(0)
     await Promise.allSettled(context.pendingToolPromises.values())
 
+    // The waiter always receives a signal now: the server fallback needs a
+    // handle to cancel its own wait if it ends up running the tool itself.
     expect(waitForWorkflowToolCompletion).toHaveBeenCalledWith({
       toolCallId: 'tool-background',
       workflowId: 'workflow-1',
       timeoutMs: 1000,
-      abortSignal: undefined,
+      abortSignal: expect.any(AbortSignal),
       registry: execContext.resolvedSecretTraceRegistry,
     })
     expect(onEvent).toHaveBeenCalledWith(
@@ -644,6 +653,116 @@ describe('sse-handlers tool lifecycle', () => {
     )
   })
 
+  it('runs a workflow tool server-side when no browser picks it up', async () => {
+    // Nobody claims it, the wait expires, and the server wins the claim.
+    waitForWorkflowToolCompletion.mockResolvedValue(null)
+    claimWorkflowToolExecution.mockResolvedValueOnce({ toolCallId: 'tool-unclaimed' })
+    executeTool.mockResolvedValueOnce({ success: true, output: { ran: 'on-server' } })
+    const onEvent = vi.fn()
+
+    await sseHandlers.tool(
+      {
+        type: MothershipStreamV1EventType.tool,
+        payload: {
+          toolCallId: 'tool-unclaimed',
+          toolName: 'run_workflow',
+          arguments: { workflowId: 'workflow-1' },
+          executor: MothershipStreamV1ToolExecutor.client,
+          mode: MothershipStreamV1ToolMode.async,
+          phase: MothershipStreamV1ToolPhase.call,
+        },
+      } satisfies StreamEvent,
+      context,
+      execContext,
+      { onEvent, interactive: true, timeout: 1 }
+    )
+
+    await Promise.allSettled(context.pendingToolPromises.values())
+
+    // Regression guard: the wait sets the call to 'executing' before parking,
+    // and executeToolAndReport short-circuits anything already 'executing'. If
+    // the handoff stops resetting the status, executeTool is never reached and
+    // the workflow silently does not run.
+    expect(executeTool).toHaveBeenCalled()
+    expect(executeTool.mock.calls.at(-1)?.[0]).toBe('run_workflow')
+    // The claimed execution id must reach the handler so the run is attributable.
+    expect(executeTool.mock.calls.at(-1)?.[2]?.boundWorkflowExecutionId).toBeTruthy()
+
+    const workflowResults = onEvent.mock.calls
+      .map(([event]) => event)
+      .filter(
+        (event) =>
+          event?.type === MothershipStreamV1EventType.tool &&
+          event.payload?.toolCallId === 'tool-unclaimed' &&
+          event.payload?.phase === MothershipStreamV1ToolPhase.result
+      )
+    // Exactly one result, from the sim path — no client-flavored duplicate on top.
+    expect(workflowResults).toHaveLength(1)
+    expect(workflowResults[0].payload.executor).toBe(MothershipStreamV1ToolExecutor.sim)
+  })
+
+  it('claims and runs the same workflow when the call omits an explicit workflowId', async () => {
+    // The waiter resolves the target via resolveWorkflowToolTargetId(args, ctx)
+    // while the handler resolves it as params.workflowId || context.workflowId.
+    // If those two ever diverge, the fallback would claim one workflow and run
+    // another.
+    waitForWorkflowToolCompletion.mockResolvedValue(null)
+    claimWorkflowToolExecution.mockResolvedValueOnce({ toolCallId: 'tool-implicit-workflow' })
+    executeTool.mockResolvedValueOnce({ success: true, output: {} })
+
+    await sseHandlers.tool(
+      {
+        type: MothershipStreamV1EventType.tool,
+        payload: {
+          toolCallId: 'tool-implicit-workflow',
+          toolName: 'run_workflow',
+          arguments: {},
+          executor: MothershipStreamV1ToolExecutor.client,
+          mode: MothershipStreamV1ToolMode.async,
+          phase: MothershipStreamV1ToolPhase.call,
+        },
+      } satisfies StreamEvent,
+      context,
+      execContext,
+      { onEvent: vi.fn(), interactive: true, timeout: 1 }
+    )
+
+    await Promise.allSettled(context.pendingToolPromises.values())
+
+    expect(waitForWorkflowToolCompletion).toHaveBeenCalledWith(
+      expect.objectContaining({ workflowId: 'workflow-1' })
+    )
+    expect(executeTool.mock.calls.at(-1)?.[2]?.workflowId).toBe('workflow-1')
+  })
+
+  it('does not run a workflow tool server-side when a browser holds the claim', async () => {
+    waitForWorkflowToolCompletion.mockResolvedValue(null)
+    claimWorkflowToolExecution.mockResolvedValueOnce(null)
+    executeTool.mockClear()
+    const onEvent = vi.fn()
+
+    await sseHandlers.tool(
+      {
+        type: MothershipStreamV1EventType.tool,
+        payload: {
+          toolCallId: 'tool-claimed-elsewhere',
+          toolName: 'run_workflow',
+          arguments: { workflowId: 'workflow-1' },
+          executor: MothershipStreamV1ToolExecutor.client,
+          mode: MothershipStreamV1ToolMode.async,
+          phase: MothershipStreamV1ToolPhase.call,
+        },
+      } satisfies StreamEvent,
+      context,
+      execContext,
+      { onEvent, interactive: true, timeout: 1 }
+    )
+
+    await Promise.allSettled(context.pendingToolPromises.values())
+
+    expect(executeTool).not.toHaveBeenCalled()
+  })
+
   it('waits for the desktop client when a static VFS read is explicitly user-local', async () => {
     waitForClientToolCompletion.mockResolvedValueOnce({
       status: 'success',

apps/sim/lib/copilot/request/handlers/tool.ts

@@ -2,9 +2,12 @@ import { isBrowserToolName } from '@sim/browser-protocol'
 import { createLogger } from '@sim/logger'
 import { isTerminalToolName } from '@sim/terminal-protocol'
 import { getErrorMessage, toError } from '@sim/utils/errors'
-import type { AsyncCompletionSignal } from '@/lib/copilot/async-runs/lifecycle'
+import type {
+  AsyncCompletionSignal,
+  AsyncTerminalCompletionSnapshot,
+} from '@/lib/copilot/async-runs/lifecycle'
 import { upsertAsyncToolCall } from '@/lib/copilot/async-runs/repository'
-import { STREAM_TIMEOUT_MS } from '@/lib/copilot/constants'
+import { COPILOT_WORKFLOW_TOOL_CLIENT_GRACE_MS, STREAM_TIMEOUT_MS } from '@/lib/copilot/constants'
 import {
   MothershipStreamV1AsyncToolRecordStatus,
   type MothershipStreamV1ToolCallDescriptor,
@@ -24,17 +27,15 @@ import {
 } from '@/lib/copilot/request/session'
 import { markToolResultSeen, wasToolResultSeen } from '@/lib/copilot/request/sse-utils'
 import { setTerminalToolCallState } from '@/lib/copilot/request/tool-call-state'
-import {
-  waitForClientToolCompletion,
-  waitForWorkflowToolCompletion,
-} from '@/lib/copilot/request/tools/client'
+import { waitForClientToolCompletion } from '@/lib/copilot/request/tools/client'
 import { sealClientToolContext } from '@/lib/copilot/request/tools/client-completion-seal.server'
 import { executeToolAndReport } from '@/lib/copilot/request/tools/executor'
 import {
   runGatedToolExecution,
   TOOL_AWAITING_APPROVAL_STATUS,
   toolCallNeedsApproval,
 } from '@/lib/copilot/request/tools/permission'
+import { raceWorkflowToolClientPickup } from '@/lib/copilot/request/tools/workflow-client-fallback'
 import type {
   ExecutionContext,
   OrchestratorOptions,
@@ -671,9 +672,11 @@ async function dispatchToolExecution(
 ): Promise<void> {
   const scopeLabel = scope === 'subagent' ? 'subagent ' : ''
 
-  const fireToolExecution = (): Promise<AsyncCompletionSignal> => {
+  const fireToolExecution = (
+    execContextOverride?: ExecutionContext
+  ): Promise<AsyncCompletionSignal> => {
     return (async () => {
-      return executeToolAndReport(toolCallId, context, execContext, options)
+      return executeToolAndReport(toolCallId, context, execContextOverride ?? execContext, options)
     })().catch((err) => {
       logger.error(`Parallel ${scopeLabel}tool execution failed`, {
         toolCallId,
@@ -754,23 +757,58 @@ async function dispatchToolExecution(
         ...(context.runId ? { [TraceAttr.RunId]: context.runId } : {}),
       },
       async (span) => {
-        const completion = isWorkflowToolName(toolName)
-          ? await waitForWorkflowToolCompletion({
-              toolCallId,
-              workflowId: resolveWorkflowToolTargetId(args, execContext.workflowId),
-              timeoutMs: timeoutMs ?? STREAM_TIMEOUT_MS,
-              abortSignal: options.abortSignal,
-              registry: execContext.resolvedSecretTraceRegistry,
-            })
-          : await waitForClientToolCompletion({
-              toolCallId,
-              runId: context.runId,
-              userId: execContext.userId,
-              timeoutMs,
-              abortSignal: options.abortSignal,
-              registry: execContext.resolvedSecretTraceRegistry,
-            })
-        span.setAttribute(TraceAttr.ToolCompletionReceived, completion !== undefined)
+        let completion: AsyncTerminalCompletionSnapshot | null
+        if (isWorkflowToolName(toolName)) {
+          const race = await raceWorkflowToolClientPickup({
+            toolCallId,
+            workflowId: resolveWorkflowToolTargetId(args, execContext.workflowId),
+            timeoutMs: timeoutMs ?? STREAM_TIMEOUT_MS,
+            graceMs: COPILOT_WORKFLOW_TOOL_CLIENT_GRACE_MS,
+            abortSignal: options.abortSignal,
+            registry: execContext.resolvedSecretTraceRegistry,
+            runOnServer: (boundExecutionId) => {
+              // `executeToolAndReportInner` short-circuits a call that is
+              // already 'executing' — which is exactly what this wait set it to
+              // before parking. Hand it back the state it dispatches from.
+              toolCall.status = 'pending'
+              return fireToolExecution({
+                ...execContext,
+                boundWorkflowExecutionId: boundExecutionId,
+              })
+            },
+          })
+
+          if (race.winner === 'sim') {
+            // `executeToolAndReport` already emitted its own `executor: sim`
+            // result and marked it seen, so the client-completion bookkeeping
+            // below must not run again on top of it.
+            span.setAttribute(TraceAttr.ToolExecutor, MothershipStreamV1ToolExecutor.sim)
+            if (race.signal) {
+              span.setAttribute(TraceAttr.ToolOutcome, race.signal.status)
+            }
+            return (
+              race.signal ?? {
+                status: MothershipStreamV1ToolOutcome.error,
+                message: 'Tool completion missing',
+                data: { error: 'Tool completion missing' },
+              }
+            )
+          }
+          completion = race.completion ?? null
+        } else {
+          completion = await waitForClientToolCompletion({
+            toolCallId,
+            runId: context.runId,
+            userId: execContext.userId,
+            timeoutMs,
+            abortSignal: options.abortSignal,
+            registry: execContext.resolvedSecretTraceRegistry,
+          })
+        }
+        span.setAttribute(TraceAttr.ToolExecutor, MothershipStreamV1ToolExecutor.client)
+        // Both waiters resolve `T | null`, never undefined — comparing against
+        // undefined made this a constant `true` and hid every timeout.
+        span.setAttribute(TraceAttr.ToolCompletionReceived, completion !== null)
         if (completion) {
           span.setAttribute(TraceAttr.ToolOutcome, completion.status)
         }