feat(mship): async runs

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

@@ -459,7 +459,7 @@ describe('Copilot Confirm API Route', () => {
     getAsyncToolCall.mockResolvedValue({
       ...existingRow,
       toolName: 'run_workflow',
-      args: { workflowId: 'workflow-1' },
+      args: { workflowId: 'workflow-1', async: true },
       status: 'running',
       claimedBy: null,
     })
@@ -469,6 +469,7 @@ describe('Copilot Confirm API Route', () => {
         toolCallId: 'tool-call-123',
         status: 'error',
         message: 'untrusted client detail',
+        data: { code: 'ASYNC_WORKFLOW_DEPLOYMENT_STALE' },
       })
     )
 
@@ -477,14 +478,47 @@ describe('Copilot Confirm API Route', () => {
     expect(completeAsyncToolCall).toHaveBeenCalledWith({
       toolCallId: 'tool-call-123',
       status: 'failed',
-      result: { success: false, workflowId: 'workflow-1' },
-      error: 'Workflow execution failed.',
+      result: {
+        success: false,
+        workflowId: 'workflow-1',
+        code: 'ASYNC_WORKFLOW_DEPLOYMENT_STALE',
+        error: 'Async execution requires the current workflow to match its deployed version',
+      },
+      error: 'Async execution requires the current workflow to match its deployed version',
     })
     expect(JSON.stringify(publishToolConfirmation.mock.calls)).not.toContain(
       'untrusted client detail'
     )
   })
 
+  it('discards an unknown async workflow preflight failure', async () => {
+    getAsyncToolCall.mockResolvedValue({
+      ...existingRow,
+      toolName: 'run_workflow',
+      args: { workflowId: 'workflow-1', async: true },
+      status: 'running',
+      claimedBy: null,
+    })
+
+    const response = await POST(
+      createMockPostRequest({
+        toolCallId: 'tool-call-123',
+        status: 'error',
+        message: 'untrusted client detail',
+        data: { code: 'UNTRUSTED_CLIENT_CODE' },
+      })
+    )
+
+    expect(response.status).toBe(200)
+    expect(completeAsyncToolCall).toHaveBeenCalledWith({
+      toolCallId: 'tool-call-123',
+      status: 'failed',
+      result: { success: false, workflowId: 'workflow-1' },
+      error: 'Workflow execution failed.',
+    })
+    expect(JSON.stringify(publishToolConfirmation.mock.calls)).not.toContain('untrusted')
+  })
+
   it('downgrades an unverifiable success from a stale client to a structural failure', async () => {
     getAsyncToolCall.mockResolvedValue({
       ...existingRow,

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

@@ -2,6 +2,7 @@ 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 { isPlainRecord } from '@sim/utils/object'
 import { type NextRequest, NextResponse } from 'next/server'
 import { copilotConfirmContract } from '@/lib/api/contracts/copilot'
 import { parseRequest, validationErrorResponse } from '@/lib/api/server'
@@ -38,7 +39,9 @@ import {
   sealClientToolCompletion,
 } from '@/lib/copilot/request/tools/client-completion-seal.server'
 import {
+  type AsyncWorkflowDeploymentError,
   createStructuralWorkflowToolCompletionData,
+  getAsyncWorkflowDeploymentError,
   getWorkflowToolCompletionExecutionId,
   getWorkflowToolCompletionMessage,
   getWorkflowToolConfirmationStatus,
@@ -278,6 +281,7 @@ export const POST = withRouteHandler((req: NextRequest) => {
 
         let effectiveStatus = status
         let executionId = submittedExecutionId
+        let deploymentError: AsyncWorkflowDeploymentError | undefined
 
         if (isWorkflowTool) {
           const claimedExecutionId = getClaimedWorkflowExecutionId(existing.claimedBy)
@@ -329,16 +333,28 @@ export const POST = withRouteHandler((req: NextRequest) => {
           } else {
             executionId = undefined
           }
+
+          if (
+            effectiveStatus === ASYNC_TOOL_CONFIRMATION_STATUS.error &&
+            executionId === undefined &&
+            existing.toolName === 'run_workflow' &&
+            isPlainRecord(existing.args) &&
+            existing.args.async === true
+          ) {
+            deploymentError = getAsyncWorkflowDeploymentError(data)
+          }
         }
 
         span.setAttribute(TraceAttr.ToolConfirmationStatus, effectiveStatus)
         const projected = isWorkflowTool
           ? {
-              message: getWorkflowToolCompletionMessage(effectiveStatus),
+              message:
+                deploymentError?.message ?? getWorkflowToolCompletionMessage(effectiveStatus),
               data: createStructuralWorkflowToolCompletionData(
                 effectiveStatus,
                 workflowId,
-                executionId
+                executionId,
+                deploymentError
               ),
             }
           : {

apps/sim/app/api/workflows/[id]/execute/route.async.test.ts

@@ -29,6 +29,7 @@ import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
 import { AsyncJobEnqueueError } from '@/lib/core/async-jobs/types'
 import { getRemainingExecutionMs } from '@/lib/core/execution-limits'
 import { INTERNAL_EXECUTION_DEADLINE_HEADER } from '@/lib/execution/execution-deadline-header'
+import { WORKFLOW_NOT_DEPLOYED_CODE } from '@/lib/execution/preprocessing'
 import {
   PRIVATE_SECRET_PROVENANCE_BUNDLE_V1,
   PRIVATE_SECRET_PROVENANCE_FIELD,
@@ -39,6 +40,7 @@ const {
   mockAssertBillingAttributionSnapshot,
   mockClaimExecutionId,
   mockClaimWorkflowToolExecution,
+  mockCheckNeedsRedeployment,
   mockEnqueue,
   mockExecuteWorkflowJob,
   mockExecuteWorkflowCore,
@@ -67,6 +69,7 @@ const {
   }),
   mockClaimExecutionId: vi.fn(),
   mockClaimWorkflowToolExecution: vi.fn(),
+  mockCheckNeedsRedeployment: vi.fn(),
   mockEnqueue: vi.fn().mockResolvedValue('job-123'),
   mockExecuteWorkflowJob: vi.fn(),
   mockExecuteWorkflowCore: vi.fn(),
@@ -118,6 +121,10 @@ vi.mock('@/lib/workflows/utils', () => workflowsUtilsMock)
 
 vi.mock('@/lib/execution/preprocessing', () => executionPreprocessingMock)
 
+vi.mock('@/app/api/workflows/utils', () => ({
+  checkNeedsRedeployment: mockCheckNeedsRedeployment,
+}))
+
 vi.mock('@/lib/workflows/persistence/utils', () => workflowsPersistenceUtilsMock)
 
 vi.mock('@/lib/workflows/executor/execution-core', () => ({
@@ -415,6 +422,7 @@ describe('workflow execute async route', () => {
       toolCallId: 'copilot-tool-1',
       claimedBy: 'workflow:execution-123',
     })
+    mockCheckNeedsRedeployment.mockResolvedValue(false)
     mockHasDurableExecutionOwner.mockResolvedValue(false)
     mockGetAsyncToolCall.mockReset().mockResolvedValue({
       toolCallId: 'copilot-tool-1',
@@ -875,6 +883,94 @@ describe('workflow execute async route', () => {
     expect(mockExecuteWorkflowCore).not.toHaveBeenCalled()
   })
 
+  it('queues a bound Copilot workflow execution asynchronously', async () => {
+    const request = createBoundCopilotExecutionRequest({
+      stream: false,
+      triggerBlockId: 'trigger-async',
+    })
+    request.headers.set('X-Execution-Mode', 'async')
+
+    const response = await POST(request, {
+      params: Promise.resolve({ id: 'workflow-1' }),
+    })
+
+    expect(response.status).toBe(202)
+    expect(mockClaimWorkflowToolExecution).toHaveBeenCalledWith('copilot-tool-1', 'execution-123')
+    expect(mockPreprocessExecution).toHaveBeenCalledWith(
+      expect.objectContaining({ checkDeployment: true, executionType: 'async' })
+    )
+    expect(loggingSessionMockFns.mockSetTrustedExecutionCorrelation).toHaveBeenCalledWith({
+      executionId: 'execution-123',
+      requestId: 'req-12345678',
+      source: 'workflow',
+      workflowId: 'workflow-1',
+      triggerType: 'copilot',
+      copilotToolCallId: 'copilot-tool-1',
+    })
+    expect(mockEnqueue).toHaveBeenCalledWith(
+      'workflow-execution',
+      expect.objectContaining({
+        executionId: 'execution-123',
+        triggerBlockId: 'trigger-async',
+        correlation: expect.objectContaining({ copilotToolCallId: 'copilot-tool-1' }),
+      }),
+      expect.any(Object)
+    )
+  })
+
+  it('rejects a bound async run when the deployed workflow is stale', async () => {
+    mockCheckNeedsRedeployment.mockResolvedValueOnce(true)
+    const request = createBoundCopilotExecutionRequest({ stream: false })
+    request.headers.set('X-Execution-Mode', 'async')
+
+    const response = await POST(request, {
+      params: Promise.resolve({ id: 'workflow-1' }),
+    })
+
+    expect(response.status).toBe(409)
+    await expect(response.json()).resolves.toEqual({
+      error: 'Async execution requires the current workflow to match its deployed version',
+      code: 'ASYNC_WORKFLOW_DEPLOYMENT_STALE',
+    })
+    expect(mockClaimWorkflowToolExecution).toHaveBeenCalledWith('copilot-tool-1', 'execution-123')
+    expect(mockReleaseExecutionSlot).toHaveBeenCalledWith('execution-123')
+    expect(mockReleaseWorkflowToolExecutionClaim).toHaveBeenCalledWith(
+      'copilot-tool-1',
+      'execution-123'
+    )
+    expect(mockEnqueue).not.toHaveBeenCalled()
+  })
+
+  it('rejects a bound async run when the workflow has not been deployed', async () => {
+    mockPreprocessExecution.mockResolvedValueOnce({
+      success: false,
+      error: {
+        message: 'Workflow is not deployed',
+        statusCode: 403,
+        code: WORKFLOW_NOT_DEPLOYED_CODE,
+      },
+    })
+    const request = createBoundCopilotExecutionRequest({ stream: false })
+    request.headers.set('X-Execution-Mode', 'async')
+
+    const response = await POST(request, {
+      params: Promise.resolve({ id: 'workflow-1' }),
+    })
+
+    expect(response.status).toBe(403)
+    await expect(response.json()).resolves.toEqual({
+      error: 'Async execution requires the workflow to be deployed first',
+      code: 'ASYNC_WORKFLOW_DEPLOYMENT_MISSING',
+    })
+    expect(mockReleaseExecutionSlot).toHaveBeenCalledWith('execution-123')
+    expect(mockReleaseWorkflowToolExecutionClaim).toHaveBeenCalledWith(
+      'copilot-tool-1',
+      'execution-123'
+    )
+    expect(mockEnqueue).not.toHaveBeenCalled()
+    expect(mockCheckNeedsRedeployment).not.toHaveBeenCalled()
+  })
+
   it.each([
     [
       'cancelled',