From f93338e08a6c5cad9859d7282a44f63b09c2f916 Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Thu, 6 Aug 2026 00:29:22 -0700 Subject: [PATCH] fix(execution): duplicate execution issue --- .../[id]/execute/route.async.test.ts | 2 + .../app/api/workflows/[id]/execute/route.ts | 10 ++++- .../utils/workflow-execution-utils.test.ts | 41 ++++++++++++++++++- .../utils/workflow-execution-utils.ts | 17 +++++++- apps/sim/hooks/use-execution-stream.ts | 3 +- apps/sim/lib/copilot/constants.ts | 3 ++ .../tools/client/run-tool-execution.test.ts | 33 +++++++++++++++ .../tools/client/run-tool-execution.ts | 21 +++++++++- 8 files changed, 123 insertions(+), 7 deletions(-) diff --git a/apps/sim/app/api/workflows/[id]/execute/route.async.test.ts b/apps/sim/app/api/workflows/[id]/execute/route.async.test.ts index d0a223aa0d6..d80d2f2aa31 100644 --- a/apps/sim/app/api/workflows/[id]/execute/route.async.test.ts +++ b/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() diff --git a/apps/sim/app/api/workflows/[id]/execute/route.ts b/apps/sim/app/api/workflows/[id]/execute/route.ts index d7bce0a3dbc..91a67298ed4 100644 --- a/apps/sim/app/api/workflows/[id]/execute/route.ts +++ b/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 } ) } diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.test.ts b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.test.ts index e4b75cedccc..10db82c7614 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.test.ts +++ b/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({ + 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({ diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.ts b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.ts index 013e10451d4..4d8bb03eeef 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.ts +++ b/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, diff --git a/apps/sim/hooks/use-execution-stream.ts b/apps/sim/hooks/use-execution-stream.ts index bd1ab735aec..fb156950e88 100644 --- a/apps/sim/hooks/use-execution-stream.ts +++ b/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' diff --git a/apps/sim/lib/copilot/constants.ts b/apps/sim/lib/copilot/constants.ts index 3028222f123..5102f753d51 100644 --- a/apps/sim/lib/copilot/constants.ts +++ b/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 diff --git a/apps/sim/lib/copilot/tools/client/run-tool-execution.test.ts b/apps/sim/lib/copilot/tools/client/run-tool-execution.test.ts index 873497f3407..f3b48d6533f 100644 --- a/apps/sim/lib/copilot/tools/client/run-tool-execution.test.ts +++ b/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) + }) }) diff --git a/apps/sim/lib/copilot/tools/client/run-tool-execution.ts b/apps/sim/lib/copilot/tools/client/run-tool-execution.ts index 62a7bc15110..032eb5a2447 100644 --- a/apps/sim/lib/copilot/tools/client/run-tool-execution.ts +++ b/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) {