Skip to content

Commit 293dc5b

Browse files
icecrasher321icecrasher321
authored
fix(execution): duplicate execution issue (#6316)
1 parent 377702f commit 293dc5b

8 files changed

Lines changed: 123 additions & 7 deletions

File tree

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -724,7 +724,9 @@ describe('workflow execute async route', () => {
724724
expect(response.status).toBe(409)
725725
expect(await response.json()).toEqual({
726726
error: 'Copilot workflow tool is already bound to another execution',
727+
code: 'COPILOT_WORKFLOW_EXECUTION_CONFLICT',
727728
})
729+
expect(mockGetAsyncToolCall).toHaveBeenCalledTimes(1)
728730
expect(loggingSessionMockFns.mockSetTrustedExecutionCorrelation).not.toHaveBeenCalled()
729731
expect(mockPreprocessExecution).not.toHaveBeenCalled()
730732
expect(mockExecuteWorkflowCore).not.toHaveBeenCalled()

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

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@ import {
2727
getRunSegment,
2828
releaseWorkflowToolExecutionClaim,
2929
} from '@/lib/copilot/async-runs/repository'
30+
import { COPILOT_WORKFLOW_EXECUTION_CONFLICT_CODE } from '@/lib/copilot/constants'
3031
import { isWorkflowToolName, resolveWorkflowToolTargetId } from '@/lib/copilot/tools/workflow-tools'
3132
import { admissionRejectedResponse, tryAdmit } from '@/lib/core/admission/gate'
3233
import { getJobQueue, shouldExecuteInline } from '@/lib/core/async-jobs'
@@ -1213,8 +1214,15 @@ async function handleExecutePost(
12131214
if (copilotToolCallId) {
12141215
const boundToolCall = await claimWorkflowToolExecution(copilotToolCallId, executionId)
12151216
if (!boundToolCall) {
1217+
reqLogger.warn('Rejected duplicate Copilot workflow execution', {
1218+
copilotToolCallId,
1219+
attemptedExecutionId: executionId,
1220+
})
12161221
return NextResponse.json(
1217-
{ error: 'Copilot workflow tool is already bound to another execution' },
1222+
{
1223+
error: 'Copilot workflow tool is already bound to another execution',
1224+
code: COPILOT_WORKFLOW_EXECUTION_CONFLICT_CODE,
1225+
},
12181226
{ status: 409 }
12191227
)
12201228
}

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

Lines changed: 40 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,16 +2,18 @@
22
* @vitest-environment node
33
*/
44
import { resetTerminalConsoleMock, terminalConsoleMockFns } from '@sim/testing'
5-
import { beforeEach, describe, expect, it, vi } from 'vitest'
5+
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
66
import {
77
addExecutionErrorConsoleEntry,
88
addHttpErrorConsoleEntry,
99
createBlockEventHandlers,
10+
executeWorkflowWithFullLogging,
1011
handleExecutionCancelledConsole,
1112
handleExecutionErrorConsole,
1213
reconcileFinalBlockLogs,
1314
} from '@/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils'
1415
import type { BlockLog } from '@/executor/types'
16+
import type { ExecutionStreamHttpError } from '@/hooks/use-execution-stream'
1517
import { useExecutionStore } from '@/stores/execution'
1618

1719
describe('workflow-execution-utils', () => {
@@ -22,6 +24,43 @@ describe('workflow-execution-utils', () => {
2224
} as any)
2325
})
2426

27+
afterEach(() => {
28+
vi.unstubAllGlobals()
29+
})
30+
31+
it('classifies a duplicate Copilot claim without writing an HTTP error row', async () => {
32+
vi.mocked(useExecutionStore.getState).mockReturnValue({
33+
getCurrentExecutionId: vi.fn(() => 'exec-1'),
34+
setActiveBlocks: vi.fn(),
35+
setBlockRunStatus: vi.fn(),
36+
setCurrentExecutionId: vi.fn(),
37+
setEdgeRunStatus: vi.fn(),
38+
} as any)
39+
vi.stubGlobal(
40+
'fetch',
41+
vi.fn().mockResolvedValue({
42+
ok: false,
43+
status: 409,
44+
json: vi.fn().mockResolvedValue({
45+
error: 'Copilot workflow tool is already bound to another execution',
46+
code: 'COPILOT_WORKFLOW_EXECUTION_CONFLICT',
47+
}),
48+
})
49+
)
50+
51+
const promise = executeWorkflowWithFullLogging({
52+
workflowId: 'wf-1',
53+
executionId: 'exec-1',
54+
copilotToolCallId: 'tool-1',
55+
})
56+
57+
await expect(promise).rejects.toMatchObject<ExecutionStreamHttpError>({
58+
httpStatus: 409,
59+
code: 'COPILOT_WORKFLOW_EXECUTION_CONFLICT',
60+
})
61+
expect(terminalConsoleMockFns.mockAddConsole).not.toHaveBeenCalled()
62+
})
63+
2564
describe('createBlockEventHandlers', () => {
2665
it('skips duplicate block start rows during reconnect replay', () => {
2766
terminalConsoleMockFns.mockAddConsole({

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

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
import { createLogger } from '@sim/logger'
22
import { toError } from '@sim/utils/errors'
33
import { generateId } from '@sim/utils/id'
4+
import { isPlainRecord } from '@sim/utils/object'
5+
import { COPILOT_WORKFLOW_EXECUTION_CONFLICT_CODE } from '@/lib/copilot/constants'
46
import type { SecretSafeBlockLog } from '@/lib/logs/execution/display-types'
57
import type { TraceSpan } from '@/lib/logs/types'
68
import type {
@@ -12,6 +14,7 @@ import type {
1214
import type { BlockLog, BlockState, ExecutionResult, StreamingExecution } from '@/executor/types'
1315
import { stripCloneSuffixes } from '@/executor/utils/subflow-utils'
1416
import {
17+
ExecutionStreamHttpError,
1518
processSSEStream,
1619
SSEEventHandlerError,
1720
SSEStreamInterruptedError,
@@ -1047,8 +1050,18 @@ export async function executeWorkflowWithFullLogging(
10471050
})
10481051

10491052
if (!response.ok) {
1050-
const error = await response.json()
1051-
const errorMessage = error.error || 'Workflow run failed'
1053+
const error: unknown = await response.json()
1054+
const errorCode =
1055+
isPlainRecord(error) && typeof error.code === 'string' ? error.code : undefined
1056+
if (response.status === 409 && errorCode === COPILOT_WORKFLOW_EXECUTION_CONFLICT_CODE) {
1057+
throw new ExecutionStreamHttpError(
1058+
'Copilot workflow execution is already owned by another client',
1059+
response.status,
1060+
errorCode
1061+
)
1062+
}
1063+
const errorMessage =
1064+
isPlainRecord(error) && typeof error.error === 'string' ? error.error : 'Workflow run failed'
10521065
addHttpErrorConsoleEntry(addConsole, {
10531066
workflowId: wfId,
10541067
executionId,

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,8 @@ const logger = createLogger('useExecutionStream')
2727
export class ExecutionStreamHttpError extends Error {
2828
constructor(
2929
message: string,
30-
public readonly httpStatus: number
30+
public readonly httpStatus: number,
31+
public readonly code?: string
3132
) {
3233
super(message)
3334
this.name = 'ExecutionStreamHttpError'

‎apps/sim/lib/copilot/constants.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,9 @@ export const MOTHERSHIP_CHAT_API_PATH = '/api/mothership/chat'
4545
/** POST — confirm or reject a tool call. */
4646
export const COPILOT_CONFIRM_API_PATH = '/api/copilot/confirm'
4747

48+
export const COPILOT_WORKFLOW_EXECUTION_CONFLICT_CODE =
49+
'COPILOT_WORKFLOW_EXECUTION_CONFLICT' as const
50+
4851
/** Maximum entries in the in-memory SSE tool-event dedup cache. */
4952
export const STREAM_BUFFER_MAX_DEDUP_ENTRIES = 1_000
5053

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

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ const {
99
executeWorkflowWithFullLogging,
1010
getWorkflowEntries,
1111
loadExecutionPointer,
12+
MockExecutionStreamHttpError,
1213
MockSSEEventHandlerError,
1314
MockSSEStreamInterruptedError,
1415
saveExecutionPointer,
@@ -18,6 +19,16 @@ const {
1819
executeWorkflowWithFullLogging: vi.fn(),
1920
getWorkflowEntries: vi.fn(() => []),
2021
loadExecutionPointer: vi.fn(),
22+
MockExecutionStreamHttpError: class ExecutionStreamHttpError extends Error {
23+
constructor(
24+
message: string,
25+
public readonly httpStatus: number,
26+
public readonly code?: string
27+
) {
28+
super(message)
29+
this.name = 'ExecutionStreamHttpError'
30+
}
31+
},
2132
MockSSEEventHandlerError: class SSEEventHandlerError extends Error {
2233
executionId?: string
2334

@@ -63,6 +74,8 @@ vi.mock('@/stores/execution/store', () => ({
6374
}))
6475

6576
vi.mock('@/hooks/use-execution-stream', () => ({
77+
ExecutionStreamHttpError: MockExecutionStreamHttpError,
78+
isExecutionStreamHttpError: (error: unknown) => error instanceof MockExecutionStreamHttpError,
6679
SSEEventHandlerError: MockSSEEventHandlerError,
6780
SSEStreamInterruptedError: MockSSEStreamInterruptedError,
6881
}))
@@ -314,4 +327,24 @@ describe('run tool execution cancellation', () => {
314327
})
315328
)
316329
})
330+
331+
it('drops a duplicate client runner without confirming or surfacing an error', async () => {
332+
const fetchMock = vi.fn().mockResolvedValue({ ok: true })
333+
vi.stubGlobal('fetch', fetchMock)
334+
executeWorkflowWithFullLogging.mockRejectedValueOnce(
335+
new MockExecutionStreamHttpError(
336+
'Copilot workflow execution is already owned by another client',
337+
409,
338+
'COPILOT_WORKFLOW_EXECUTION_CONFLICT'
339+
)
340+
)
341+
342+
executeRunToolOnClient('tool-duplicate', 'run_workflow', { workflowId: 'wf-1' })
343+
344+
await vi.waitFor(() => {
345+
expect(clearExecutionPointer).toHaveBeenCalledWith('wf-1')
346+
})
347+
expect(fetchMock).not.toHaveBeenCalled()
348+
expect(setIsExecuting).toHaveBeenCalledWith('wf-1', false)
349+
})
317350
})

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

Lines changed: 19 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,10 @@ import {
66
ASYNC_TOOL_CONFIRMATION_STATUS,
77
type AsyncConfirmationStatus,
88
} from '@/lib/copilot/async-runs/lifecycle'
9-
import { COPILOT_CONFIRM_API_PATH } from '@/lib/copilot/constants'
9+
import {
10+
COPILOT_CONFIRM_API_PATH,
11+
COPILOT_WORKFLOW_EXECUTION_CONFLICT_CODE,
12+
} from '@/lib/copilot/constants'
1013
import { MothershipStreamV1ToolOutcome } from '@/lib/copilot/generated/mothership-stream-v1'
1114
import {
1215
RunBlock,
@@ -19,7 +22,11 @@ import {
1922
} from '@/lib/copilot/tools/client/completion'
2023
import { getWorkflowToolCompletionMessage } from '@/lib/copilot/tools/workflow-tools'
2124
import { executeWorkflowWithFullLogging } from '@/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils'
22-
import { SSEEventHandlerError, SSEStreamInterruptedError } from '@/hooks/use-execution-stream'
25+
import {
26+
isExecutionStreamHttpError,
27+
SSEEventHandlerError,
28+
SSEStreamInterruptedError,
29+
} from '@/hooks/use-execution-stream'
2330
import { useExecutionStore } from '@/stores/execution/store'
2431
import {
2532
clearExecutionPointer,
@@ -466,6 +473,16 @@ async function doExecuteRunTool(
466473
toolCallId,
467474
toolName,
468475
})
476+
} else if (
477+
isExecutionStreamHttpError(err) &&
478+
err.httpStatus === 409 &&
479+
err.code === COPILOT_WORKFLOW_EXECUTION_CONFLICT_CODE
480+
) {
481+
logger.info('[RunTool] Ignoring duplicate client workflow execution', {
482+
toolCallId,
483+
toolName,
484+
workflowId: targetWorkflowId,
485+
})
469486
} else {
470487
const msg = toError(err).message
471488
if (err instanceof SSEEventHandlerError || err instanceof SSEStreamInterruptedError) {

0 commit comments

Comments
 (0)