Skip to content

Commit 203cccf

Browse files
icecrasher321claude
andcommitted
fix(execution): keep terminal reconnect off runs a Sim run tool owns
The terminal's reconnect effect treated "execution pointer present and the current execution id matches" as an orphaned run and claimed it. A Chat run tool creates exactly that state before the server has acknowledged the run, and opening the workflow tab in Chat re-runs the effect mid-run, so the reconnect GET raced the execute POST's buffer init, got a 404, and logged "Execution state is no longer available after reconnect" as a Run Error on a run that succeeded. It also tore down the live run's store state and cleared the pointer the tool keeps for reload recovery. The run tool now exposes its ownership (isRunToolActiveForWorkflow) and the reconnect effect skips a workflow whose run it owns, leaving the pointer in place. When the tool gives up an interrupted run it notifies subscribeToRunToolRelease subscribers and the hook re-arms its reconnect, so the terminal re-attaches from the last persisted event the way the manual run path already does on interruption. Runs the tool observed to completion never notify, so a failed completion report still leaves the pointer for bindRunToolToExecution to re-report after a reload. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
1 parent 51d2118 commit 203cccf

4 files changed

Lines changed: 298 additions & 1 deletion

File tree

‎apps/sim/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-workflow-execution.test.tsx‎

Lines changed: 161 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,12 +19,15 @@ const {
1919
mockFetch,
2020
mockHandleExecutionCancelledConsole,
2121
mockHandleExecutionErrorConsole,
22+
mockIsExecutionStreamHttpError,
23+
mockIsRunToolActiveForWorkflow,
2224
mockLoadExecutionPointer,
2325
mockReconnect,
2426
mockRequestJson,
2527
mockResolveStartCandidates,
2628
mockSelectBestTrigger,
2729
mockUploadInternalFileSession,
30+
runToolReleaseListeners,
2831
terminalStoreState,
2932
workflowBlocks,
3033
workflowStoreState,
@@ -101,12 +104,15 @@ const {
101104
mockFetch: vi.fn(),
102105
mockHandleExecutionCancelledConsole: vi.fn(),
103106
mockHandleExecutionErrorConsole: vi.fn(),
107+
mockIsExecutionStreamHttpError: vi.fn(() => false),
108+
mockIsRunToolActiveForWorkflow: vi.fn(() => false),
104109
mockLoadExecutionPointer: vi.fn(),
105110
mockReconnect: vi.fn(),
106111
mockRequestJson: vi.fn(),
107112
mockResolveStartCandidates: vi.fn(),
108113
mockSelectBestTrigger: vi.fn(),
109114
mockUploadInternalFileSession: vi.fn(),
115+
runToolReleaseListeners: new Set<(workflowId: string) => void>(),
110116
terminalStoreState,
111117
workflowBlocks,
112118
workflowStoreState,
@@ -125,6 +131,16 @@ vi.mock('@/lib/api/client/request', () => ({
125131
requestJson: mockRequestJson,
126132
}))
127133

134+
vi.mock('@/lib/copilot/tools/client/run-tool-execution', () => ({
135+
isRunToolActiveForWorkflow: mockIsRunToolActiveForWorkflow,
136+
subscribeToRunToolRelease: (listener: (workflowId: string) => void) => {
137+
runToolReleaseListeners.add(listener)
138+
return () => {
139+
runToolReleaseListeners.delete(listener)
140+
}
141+
},
142+
}))
143+
128144
vi.mock('@/lib/api/contracts/workflows', () => ({
129145
cancelWorkflowExecutionContract: {},
130146
workflowLogContract: {},
@@ -214,7 +230,7 @@ vi.mock('@/hooks/use-execution-stream', () => {
214230
class SSEStreamInterruptedError extends Error {}
215231

216232
return {
217-
isExecutionStreamHttpError: () => false,
233+
isExecutionStreamHttpError: mockIsExecutionStreamHttpError,
218234
SSEEventHandlerError,
219235
SSEStreamInterruptedError,
220236
useExecutionStream: () => ({
@@ -419,6 +435,8 @@ function resetWorkflowExecutionTestState() {
419435
mockBeginScopedExecution.mockReset().mockReturnValue({})
420436
mockAdoptScopedExecution.mockReset().mockReturnValue(undefined)
421437
mockEndScopedExecution.mockReset().mockReturnValue(true)
438+
mockIsExecutionStreamHttpError.mockReset().mockReturnValue(false)
439+
mockIsRunToolActiveForWorkflow.mockReset().mockReturnValue(false)
422440
mockLoadExecutionPointer.mockReset().mockResolvedValue(null)
423441
mockReconnect.mockReset().mockResolvedValue(undefined)
424442
mockResolveStartCandidates.mockReset().mockReturnValue([])
@@ -430,6 +448,36 @@ function resetWorkflowExecutionTestState() {
430448
executionStoreState.getWorkflowExecution.mockReturnValue(idleExecution)
431449
executionStoreState.getCurrentExecutionId.mockReturnValue(null)
432450
workflowStoreState.edges.length = 0
451+
runToolReleaseListeners.clear()
452+
}
453+
454+
/**
455+
* The store and pointer state a Sim run tool leaves behind the moment it starts
456+
* a run, before the server has acknowledged it: this is what the reconnect
457+
* flow reads as an orphaned run.
458+
*/
459+
function primeRunToolOwnedExecution() {
460+
terminalStoreState._hasHydrated = true
461+
executionStoreState.getWorkflowExecution.mockReturnValue({
462+
...executionStoreState.getWorkflowExecution(),
463+
status: 'running',
464+
isExecuting: true,
465+
currentExecutionId: 'execution-1',
466+
})
467+
executionStoreState.getCurrentExecutionId.mockReturnValue('execution-1')
468+
mockLoadExecutionPointer.mockResolvedValue({
469+
workflowId: 'workflow-1',
470+
executionId: 'execution-1',
471+
lastEventId: 0,
472+
})
473+
}
474+
475+
/** The reconnect endpoint's answer while the run's buffer does not exist yet. */
476+
function rejectReconnectWithMissingRunBuffer() {
477+
mockIsExecutionStreamHttpError.mockReturnValue(true)
478+
mockReconnect.mockRejectedValue(
479+
Object.assign(new Error('Reconnect failed (404)'), { httpStatus: 404 })
480+
)
433481
}
434482

435483
describe('useWorkflowExecution lifecycle ownership', () => {
@@ -594,6 +642,118 @@ describe('useWorkflowExecution lifecycle ownership', () => {
594642
unmount()
595643
})
596644

645+
it('logs a Run Error when a reconnect for an unowned pointer finds no run buffer', async () => {
646+
primeRunToolOwnedExecution()
647+
rejectReconnectWithMissingRunBuffer()
648+
649+
const { unmount } = renderWorkflowExecutionHook()
650+
await act(async () => {
651+
await Promise.resolve()
652+
await Promise.resolve()
653+
})
654+
655+
expect(mockReconnect).toHaveBeenCalledTimes(1)
656+
expect(mockHandleExecutionErrorConsole.mock.calls[0]).toEqual(
657+
expect.arrayContaining([
658+
expect.objectContaining({
659+
executionId: 'execution-1',
660+
error: 'Execution state is no longer available after reconnect',
661+
}),
662+
])
663+
)
664+
expect(executionStoreState.setCurrentExecutionId).toHaveBeenCalledWith('workflow-1', null)
665+
expect(executionStoreState.setIsExecuting).toHaveBeenCalledWith('workflow-1', false)
666+
expect(mockClearExecutionPointer).toHaveBeenCalledWith('workflow-1')
667+
668+
unmount()
669+
})
670+
671+
it('leaves a run owned by a client run tool to its live stream instead of reconnecting', async () => {
672+
/*
673+
* Same state as above, but a Sim run tool in this tab still owns the run.
674+
* Its live stream is the source of truth, so reconnecting here would race
675+
* the run's own start (the 404 above, logged as a Run Error mid-run), tear
676+
* down the live run's store state, and clear the pointer the tool keeps
677+
* for reload recovery.
678+
*/
679+
primeRunToolOwnedExecution()
680+
rejectReconnectWithMissingRunBuffer()
681+
mockIsRunToolActiveForWorkflow.mockReturnValue(true)
682+
683+
const { unmount } = renderWorkflowExecutionHook()
684+
await act(async () => {
685+
await Promise.resolve()
686+
await Promise.resolve()
687+
})
688+
689+
expect(mockIsRunToolActiveForWorkflow).toHaveBeenCalledWith('workflow-1')
690+
expect(mockReconnect).not.toHaveBeenCalled()
691+
expect(mockHandleExecutionErrorConsole).not.toHaveBeenCalled()
692+
expect(mockClearExecutionPointer).not.toHaveBeenCalled()
693+
expect(executionStoreState.setCurrentExecutionId).not.toHaveBeenCalled()
694+
expect(executionStoreState.setIsExecuting).not.toHaveBeenCalled()
695+
expect(executionStoreState.setActiveBlocks).not.toHaveBeenCalled()
696+
697+
unmount()
698+
})
699+
700+
it('reconnects once the client run tool releases a run whose stream dropped', async () => {
701+
primeRunToolOwnedExecution()
702+
mockIsRunToolActiveForWorkflow.mockReturnValue(true)
703+
704+
const { unmount } = renderWorkflowExecutionHook()
705+
await act(async () => {
706+
await Promise.resolve()
707+
await Promise.resolve()
708+
})
709+
expect(mockReconnect).not.toHaveBeenCalled()
710+
expect(runToolReleaseListeners.size).toBeGreaterThan(0)
711+
712+
/*
713+
* What the run tool leaves behind when it gives the run up: no current
714+
* execution, not executing, ownership released, and the pointer still
715+
* carrying the last event it persisted.
716+
*/
717+
executionStoreState.getWorkflowExecution.mockReturnValue({
718+
...executionStoreState.getWorkflowExecution(),
719+
status: 'idle',
720+
isExecuting: false,
721+
currentExecutionId: null,
722+
})
723+
executionStoreState.getCurrentExecutionId.mockReturnValue(null)
724+
mockLoadExecutionPointer.mockResolvedValue({
725+
workflowId: 'workflow-1',
726+
executionId: 'execution-1',
727+
lastEventId: 5,
728+
})
729+
mockIsRunToolActiveForWorkflow.mockReturnValue(false)
730+
await act(async () => {
731+
for (const listener of runToolReleaseListeners) listener('workflow-2')
732+
await Promise.resolve()
733+
await Promise.resolve()
734+
})
735+
expect(mockReconnect).not.toHaveBeenCalled()
736+
737+
await act(async () => {
738+
for (const listener of runToolReleaseListeners) listener('workflow-1')
739+
await Promise.resolve()
740+
await Promise.resolve()
741+
})
742+
743+
expect(mockReconnect).toHaveBeenCalledTimes(1)
744+
expect(mockReconnect).toHaveBeenCalledWith(
745+
expect.objectContaining({
746+
workflowId: 'workflow-1',
747+
executionId: 'execution-1',
748+
fromEventId: 5,
749+
})
750+
)
751+
expect(mockClearExecutionPointer).not.toHaveBeenCalled()
752+
753+
unmount()
754+
expect(runToolReleaseListeners.size).toBe(0)
755+
})
756+
597757
it('does not let delayed debug completion reset a replacement execution', async () => {
598758
const debugPersistenceExecution = {}
599759
const replacementPersistenceExecution = {}

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

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,10 @@ import {
2121
workflowLogContract,
2222
workflowStateSchema,
2323
} from '@/lib/api/contracts/workflows'
24+
import {
25+
isRunToolActiveForWorkflow,
26+
subscribeToRunToolRelease,
27+
} from '@/lib/copilot/tools/client/run-tool-execution'
2428
import type { SecretSafeBlockLog } from '@/lib/logs/execution/display-types'
2529
import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
2630
import { processStreamingBlockLogs } from '@/lib/tokenization'
@@ -2386,6 +2390,14 @@ export function useWorkflowExecution() {
23862390
[activeWorkflowId, setExecutionResult, tryStartExecution]
23872391
)
23882392

2393+
useEffect(() => {
2394+
if (!activeWorkflowId) return
2395+
return subscribeToRunToolRelease((workflowId) => {
2396+
if (workflowId !== activeWorkflowId) return
2397+
setReconnectAttemptNonce((nonce) => nonce + 1)
2398+
})
2399+
}, [activeWorkflowId])
2400+
23892401
useEffect(() => {
23902402
if (!activeWorkflowId || !hasHydrated) return
23912403
if (activeReconnections.has(activeWorkflowId)) return
@@ -2404,6 +2416,13 @@ export function useWorkflowExecution() {
24042416
}
24052417

24062418
const runReconnect = async () => {
2419+
if (isRunToolActiveForWorkflow(reconnectWorkflowId)) {
2420+
logger.info('Reconnection skipped; a client run tool owns this workflow run', {
2421+
workflowId: reconnectWorkflowId,
2422+
})
2423+
return
2424+
}
2425+
24072426
let executionId: string | undefined
24082427
let fromEventId = 0
24092428

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

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -115,7 +115,9 @@ import {
115115
cancelRunToolExecution,
116116
executeRunToolOnClient,
117117
isRunToolActiveForId,
118+
isRunToolActiveForWorkflow,
118119
reportManualRunToolStop,
120+
subscribeToRunToolRelease,
119121
} from './run-tool-execution'
120122

121123
describe('run tool execution cancellation', () => {
@@ -150,6 +152,83 @@ describe('run tool execution cancellation', () => {
150152
expect(capturedSignal?.aborted).toBe(true)
151153
})
152154

155+
it('owns the workflow for exactly as long as the client run is in flight', async () => {
156+
executeWorkflowWithFullLogging.mockImplementationOnce(async (options: any) => {
157+
await new Promise((_, reject) => {
158+
options.abortSignal.addEventListener(
159+
'abort',
160+
() => reject(new DOMException('Aborted', 'AbortError')),
161+
{ once: true }
162+
)
163+
})
164+
})
165+
let ownedWhenPointerSaved: boolean | undefined
166+
saveExecutionPointer.mockImplementationOnce(() => {
167+
ownedWhenPointerSaved = isRunToolActiveForWorkflow('wf-1')
168+
})
169+
expect(isRunToolActiveForWorkflow('wf-1')).toBe(false)
170+
171+
executeRunToolOnClient('tool-1', 'run_workflow', { workflowId: 'wf-1' })
172+
await Promise.resolve()
173+
const ownedWhileInFlight = isRunToolActiveForWorkflow('wf-1')
174+
const otherWorkflowOwnedWhileInFlight = isRunToolActiveForWorkflow('wf-2')
175+
176+
cancelRunToolExecution('wf-1')
177+
await vi.waitFor(() => expect(clearExecutionPointer).toHaveBeenCalledWith('wf-1'))
178+
179+
expect(ownedWhenPointerSaved).toBe(true)
180+
expect(ownedWhileInFlight).toBe(true)
181+
expect(otherWorkflowOwnedWhileInFlight).toBe(false)
182+
expect(saveExecutionPointer).toHaveBeenCalledWith(
183+
expect.objectContaining({ workflowId: 'wf-1', lastEventId: 0 })
184+
)
185+
expect(isRunToolActiveForWorkflow('wf-1')).toBe(false)
186+
})
187+
188+
it('releases an interrupted run to reconnect subscribers only after giving up ownership', async () => {
189+
const ownedAtRelease: boolean[] = []
190+
const listener = vi.fn((workflowId: string) => {
191+
ownedAtRelease.push(isRunToolActiveForWorkflow(workflowId))
192+
})
193+
const unsubscribe = subscribeToRunToolRelease(listener)
194+
executeWorkflowWithFullLogging.mockRejectedValueOnce(
195+
new MockSSEEventHandlerError('Block handler failed on event 7', 'exec-1')
196+
)
197+
198+
try {
199+
executeRunToolOnClient('tool-1', 'run_workflow', { workflowId: 'wf-1' })
200+
await vi.waitFor(() => expect(listener).toHaveBeenCalledWith('wf-1'))
201+
202+
expect(listener).toHaveBeenCalledTimes(1)
203+
expect(ownedAtRelease).toEqual([false])
204+
expect(setIsExecuting).toHaveBeenCalledWith('wf-1', false)
205+
expect(setCurrentExecutionId).toHaveBeenCalledWith('wf-1', null)
206+
expect(setIsExecuting.mock.invocationCallOrder.at(-1)).toBeLessThan(
207+
listener.mock.invocationCallOrder[0]
208+
)
209+
expect(clearExecutionPointer).not.toHaveBeenCalled()
210+
} finally {
211+
unsubscribe()
212+
}
213+
})
214+
215+
it('does not release a run it observed to completion, even when the report fails', async () => {
216+
const listener = vi.fn()
217+
const unsubscribe = subscribeToRunToolRelease(listener)
218+
vi.stubGlobal('fetch', vi.fn().mockResolvedValue({ ok: false, status: 500 }))
219+
executeWorkflowWithFullLogging.mockResolvedValueOnce({ success: true })
220+
221+
try {
222+
executeRunToolOnClient('tool-1', 'run_workflow', { workflowId: 'wf-1' })
223+
await vi.waitFor(() => expect(isRunToolActiveForWorkflow('wf-1')).toBe(false))
224+
225+
expect(listener).not.toHaveBeenCalled()
226+
expect(clearExecutionPointer).not.toHaveBeenCalled()
227+
} finally {
228+
unsubscribe()
229+
}
230+
})
231+
153232
it('can report a manual stop using the explicit toolCallId override', async () => {
154233
const fetchMock = vi.fn().mockResolvedValue({ ok: true })
155234
vi.stubGlobal('fetch', fetchMock)

0 commit comments

Comments
 (0)