fix(execution): duplicate execution issue

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) {