fix(execution): duplicate execution issue (#6316)

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

@@ -724,7 +724,9 @@ describe('workflow execute async route', () => {
     expect(response.status).toBe(409)
     expect(await response.json()).toEqual({
       error: 'Copilot workflow tool is already bound to another execution',
+      code: 'COPILOT_WORKFLOW_EXECUTION_CONFLICT',
     })
+    expect(mockGetAsyncToolCall).toHaveBeenCalledTimes(1)
     expect(loggingSessionMockFns.mockSetTrustedExecutionCorrelation).not.toHaveBeenCalled()
     expect(mockPreprocessExecution).not.toHaveBeenCalled()
     expect(mockExecuteWorkflowCore).not.toHaveBeenCalled()

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

@@ -27,6 +27,7 @@ import {
   getRunSegment,
   releaseWorkflowToolExecutionClaim,
 } from '@/lib/copilot/async-runs/repository'
+import { COPILOT_WORKFLOW_EXECUTION_CONFLICT_CODE } from '@/lib/copilot/constants'
 import { isWorkflowToolName, resolveWorkflowToolTargetId } from '@/lib/copilot/tools/workflow-tools'
 import { admissionRejectedResponse, tryAdmit } from '@/lib/core/admission/gate'
 import { getJobQueue, shouldExecuteInline } from '@/lib/core/async-jobs'
@@ -1213,8 +1214,15 @@ async function handleExecutePost(
     if (copilotToolCallId) {
       const boundToolCall = await claimWorkflowToolExecution(copilotToolCallId, executionId)
       if (!boundToolCall) {
+        reqLogger.warn('Rejected duplicate Copilot workflow execution', {
+          copilotToolCallId,
+          attemptedExecutionId: executionId,
+        })
         return NextResponse.json(
-          { error: 'Copilot workflow tool is already bound to another execution' },
+          {
+            error: 'Copilot workflow tool is already bound to another execution',
+            code: COPILOT_WORKFLOW_EXECUTION_CONFLICT_CODE,
+          },
           { status: 409 }
         )
       }

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.test.ts

@@ -2,16 +2,18 @@
  * @vitest-environment node
  */
 import { resetTerminalConsoleMock, terminalConsoleMockFns } from '@sim/testing'
-import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import {
   addExecutionErrorConsoleEntry,
   addHttpErrorConsoleEntry,
   createBlockEventHandlers,
+  executeWorkflowWithFullLogging,
   handleExecutionCancelledConsole,
   handleExecutionErrorConsole,
   reconcileFinalBlockLogs,
 } from '@/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils'
 import type { BlockLog } from '@/executor/types'
+import type { ExecutionStreamHttpError } from '@/hooks/use-execution-stream'
 import { useExecutionStore } from '@/stores/execution'
 
 describe('workflow-execution-utils', () => {
@@ -22,6 +24,43 @@ describe('workflow-execution-utils', () => {
     } as any)
   })
 
+  afterEach(() => {
+    vi.unstubAllGlobals()
+  })
+
+  it('classifies a duplicate Copilot claim without writing an HTTP error row', async () => {
+    vi.mocked(useExecutionStore.getState).mockReturnValue({
+      getCurrentExecutionId: vi.fn(() => 'exec-1'),
+      setActiveBlocks: vi.fn(),
+      setBlockRunStatus: vi.fn(),
+      setCurrentExecutionId: vi.fn(),
+      setEdgeRunStatus: vi.fn(),
+    } as any)
+    vi.stubGlobal(
+      'fetch',
+      vi.fn().mockResolvedValue({
+        ok: false,
+        status: 409,
+        json: vi.fn().mockResolvedValue({
+          error: 'Copilot workflow tool is already bound to another execution',
+          code: 'COPILOT_WORKFLOW_EXECUTION_CONFLICT',
+        }),
+      })
+    )
+
+    const promise = executeWorkflowWithFullLogging({
+      workflowId: 'wf-1',
+      executionId: 'exec-1',
+      copilotToolCallId: 'tool-1',
+    })
+
+    await expect(promise).rejects.toMatchObject<ExecutionStreamHttpError>({
+      httpStatus: 409,
+      code: 'COPILOT_WORKFLOW_EXECUTION_CONFLICT',
+    })
+    expect(terminalConsoleMockFns.mockAddConsole).not.toHaveBeenCalled()
+  })
+
   describe('createBlockEventHandlers', () => {
     it('skips duplicate block start rows during reconnect replay', () => {
       terminalConsoleMockFns.mockAddConsole({

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.ts

@@ -1,6 +1,8 @@
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
+import { isPlainRecord } from '@sim/utils/object'
+import { COPILOT_WORKFLOW_EXECUTION_CONFLICT_CODE } from '@/lib/copilot/constants'
 import type { SecretSafeBlockLog } from '@/lib/logs/execution/display-types'
 import type { TraceSpan } from '@/lib/logs/types'
 import type {
@@ -12,6 +14,7 @@ import type {
 import type { BlockLog, BlockState, ExecutionResult, StreamingExecution } from '@/executor/types'
 import { stripCloneSuffixes } from '@/executor/utils/subflow-utils'
 import {
+  ExecutionStreamHttpError,
   processSSEStream,
   SSEEventHandlerError,
   SSEStreamInterruptedError,
@@ -1047,8 +1050,18 @@ export async function executeWorkflowWithFullLogging(
   })
 
   if (!response.ok) {
-    const error = await response.json()
-    const errorMessage = error.error || 'Workflow run failed'
+    const error: unknown = await response.json()
+    const errorCode =
+      isPlainRecord(error) && typeof error.code === 'string' ? error.code : undefined
+    if (response.status === 409 && errorCode === COPILOT_WORKFLOW_EXECUTION_CONFLICT_CODE) {
+      throw new ExecutionStreamHttpError(
+        'Copilot workflow execution is already owned by another client',
+        response.status,
+        errorCode
+      )
+    }
+    const errorMessage =
+      isPlainRecord(error) && typeof error.error === 'string' ? error.error : 'Workflow run failed'
     addHttpErrorConsoleEntry(addConsole, {
       workflowId: wfId,
       executionId,

apps/sim/hooks/use-execution-stream.ts

@@ -27,7 +27,8 @@ const logger = createLogger('useExecutionStream')
 export class ExecutionStreamHttpError extends Error {
   constructor(
     message: string,
-    public readonly httpStatus: number
+    public readonly httpStatus: number,
+    public readonly code?: string
   ) {
     super(message)
     this.name = 'ExecutionStreamHttpError'

apps/sim/lib/copilot/constants.ts

@@ -45,6 +45,9 @@ export const MOTHERSHIP_CHAT_API_PATH = '/api/mothership/chat'
 /** POST — confirm or reject a tool call. */
 export const COPILOT_CONFIRM_API_PATH = '/api/copilot/confirm'
 
+export const COPILOT_WORKFLOW_EXECUTION_CONFLICT_CODE =
+  'COPILOT_WORKFLOW_EXECUTION_CONFLICT' as const
+
 /** Maximum entries in the in-memory SSE tool-event dedup cache. */
 export const STREAM_BUFFER_MAX_DEDUP_ENTRIES = 1_000
 

apps/sim/lib/copilot/tools/client/run-tool-execution.test.ts

@@ -9,6 +9,7 @@ const {
   executeWorkflowWithFullLogging,
   getWorkflowEntries,
   loadExecutionPointer,
+  MockExecutionStreamHttpError,
   MockSSEEventHandlerError,
   MockSSEStreamInterruptedError,
   saveExecutionPointer,
@@ -18,6 +19,16 @@ const {
   executeWorkflowWithFullLogging: vi.fn(),
   getWorkflowEntries: vi.fn(() => []),
   loadExecutionPointer: vi.fn(),
+  MockExecutionStreamHttpError: class ExecutionStreamHttpError extends Error {
+    constructor(
+      message: string,
+      public readonly httpStatus: number,
+      public readonly code?: string
+    ) {
+      super(message)
+      this.name = 'ExecutionStreamHttpError'
+    }
+  },
   MockSSEEventHandlerError: class SSEEventHandlerError extends Error {
     executionId?: string
 
@@ -63,6 +74,8 @@ vi.mock('@/stores/execution/store', () => ({
 }))
 
 vi.mock('@/hooks/use-execution-stream', () => ({
+  ExecutionStreamHttpError: MockExecutionStreamHttpError,
+  isExecutionStreamHttpError: (error: unknown) => error instanceof MockExecutionStreamHttpError,
   SSEEventHandlerError: MockSSEEventHandlerError,
   SSEStreamInterruptedError: MockSSEStreamInterruptedError,
 }))
@@ -314,4 +327,24 @@ describe('run tool execution cancellation', () => {
       })
     )
   })
+
+  it('drops a duplicate client runner without confirming or surfacing an error', async () => {
+    const fetchMock = vi.fn().mockResolvedValue({ ok: true })
+    vi.stubGlobal('fetch', fetchMock)
+    executeWorkflowWithFullLogging.mockRejectedValueOnce(
+      new MockExecutionStreamHttpError(
+        'Copilot workflow execution is already owned by another client',
+        409,
+        'COPILOT_WORKFLOW_EXECUTION_CONFLICT'
+      )
+    )
+
+    executeRunToolOnClient('tool-duplicate', 'run_workflow', { workflowId: 'wf-1' })
+
+    await vi.waitFor(() => {
+      expect(clearExecutionPointer).toHaveBeenCalledWith('wf-1')
+    })
+    expect(fetchMock).not.toHaveBeenCalled()
+    expect(setIsExecuting).toHaveBeenCalledWith('wf-1', false)
+  })
 })

apps/sim/lib/copilot/tools/client/run-tool-execution.ts

@@ -6,7 +6,10 @@ import {
   ASYNC_TOOL_CONFIRMATION_STATUS,
   type AsyncConfirmationStatus,
 } from '@/lib/copilot/async-runs/lifecycle'
-import { COPILOT_CONFIRM_API_PATH } from '@/lib/copilot/constants'
+import {
+  COPILOT_CONFIRM_API_PATH,
+  COPILOT_WORKFLOW_EXECUTION_CONFLICT_CODE,
+} from '@/lib/copilot/constants'
 import { MothershipStreamV1ToolOutcome } from '@/lib/copilot/generated/mothership-stream-v1'
 import {
   RunBlock,
@@ -19,7 +22,11 @@ import {
 } from '@/lib/copilot/tools/client/completion'
 import { getWorkflowToolCompletionMessage } from '@/lib/copilot/tools/workflow-tools'
 import { executeWorkflowWithFullLogging } from '@/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils'
-import { SSEEventHandlerError, SSEStreamInterruptedError } from '@/hooks/use-execution-stream'
+import {
+  isExecutionStreamHttpError,
+  SSEEventHandlerError,
+  SSEStreamInterruptedError,
+} from '@/hooks/use-execution-stream'
 import { useExecutionStore } from '@/stores/execution/store'
 import {
   clearExecutionPointer,
@@ -466,6 +473,16 @@ async function doExecuteRunTool(
         toolCallId,
         toolName,
       })
+    } else if (
+      isExecutionStreamHttpError(err) &&
+      err.httpStatus === 409 &&
+      err.code === COPILOT_WORKFLOW_EXECUTION_CONFLICT_CODE
+    ) {
+      logger.info('[RunTool] Ignoring duplicate client workflow execution', {
+        toolCallId,
+        toolName,
+        workflowId: targetWorkflowId,
+      })
     } else {
       const msg = toError(err).message
       if (err instanceof SSEEventHandlerError || err instanceof SSEStreamInterruptedError) {