Skip to content

Commit 607b58c

Browse files
committed
fix(mothership): re-send an undelivered workflow tool completion while another run owns the workflow
A run whose completion report failed keeps it in session storage for recovery to re-send. Recovery refused to touch the call whenever another run tool owned the workflow, which the queue now makes common, so the stored report was never delivered. Re-sending a stored report needs nothing from the workflow; recovery now does it first, and only skips the pointer lookup and cleanup that belong to the run that owns the workflow. Also makes the abandoned-reconnect test track the running flag, so it fails without the stale-state clearing it covers.
1 parent fd8ffd0 commit 607b58c

2 files changed

Lines changed: 60 additions & 11 deletions

File tree

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

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -840,6 +840,18 @@ describe('run tool execution cancellation', () => {
840840
)
841841
let reconnectOpen = true
842842
isReconnectStreamOpen.mockImplementation(() => reconnectOpen)
843+
const running = new Map<string, boolean>()
844+
setIsExecuting.mockImplementation((workflowId: string, value: boolean) => {
845+
running.set(workflowId, value)
846+
})
847+
getWorkflowExecution.mockImplementation((workflowId: string) => ({
848+
isExecuting: running.get(workflowId) ?? false,
849+
}))
850+
onTestFinished(() => {
851+
setIsExecuting.mockReset()
852+
getWorkflowExecution.mockReset()
853+
getWorkflowExecution.mockImplementation(() => ({ isExecuting: false }))
854+
})
843855
executeWorkflowWithFullLogging.mockRejectedValueOnce(
844856
new MockSSEStreamInterruptedError('Execution stream interrupted', 'exec-abandoned')
845857
)
@@ -852,6 +864,7 @@ describe('run tool execution cancellation', () => {
852864
// The editor's reconnect re-attaches, then the user navigates away and it is
853865
// cancelled, leaving the execution marked current in the store.
854866
setCurrentExecutionId('wf-1', 'exec-abandoned')
867+
setIsExecuting('wf-1', true)
855868
serverStatus = 'completed'
856869
await vi.waitFor(() =>
857870
expect(isReconnectStreamOpen).toHaveBeenCalledWith('wf-1', 'exec-abandoned')
@@ -868,6 +881,36 @@ describe('run tool execution cancellation', () => {
868881
])
869882
})
870883

884+
it('re-sends an undelivered completion while a later call owns the workflow', async () => {
885+
const { finish, launchedToolCallIds } = holdExecutions()
886+
let confirmsFailFor: string | null = 'tool-a'
887+
vi.stubGlobal(
888+
'fetch',
889+
vi.fn(async (_url: string, init?: RequestInit) =>
890+
confirmsFailFor && String(init?.body).includes(`"toolCallId":"${confirmsFailFor}"`)
891+
? { ok: false, status: 500 }
892+
: { ok: true }
893+
)
894+
)
895+
896+
executeRunToolOnClient('tool-a', 'run_block', { workflowId: 'wf-1', blockId: 'block-1' })
897+
executeRunToolOnClient('tool-b', 'run_block', { workflowId: 'wf-1', blockId: 'block-2' })
898+
await vi.waitFor(() => expect(launchedToolCallIds()).toEqual(['tool-a']))
899+
const firstExecutionId = executeWorkflowWithFullLogging.mock.calls[0][0].executionId
900+
finish('tool-a')
901+
await vi.waitFor(() => expect(launchedToolCallIds()).toEqual(['tool-a', 'tool-b']))
902+
903+
confirmsFailFor = null
904+
await expect(bindRunToolToExecution('tool-a', 'wf-1')).resolves.toBe(true)
905+
906+
const delivered = confirmBodies()
907+
.filter((body) => body.toolCallId === 'tool-a')
908+
.at(-1)
909+
expect(delivered).toMatchObject({ status: 'success', executionId: firstExecutionId })
910+
finish('tool-b')
911+
await vi.waitFor(() => expect(isRunToolActiveForWorkflow('wf-1')).toBe(false))
912+
})
913+
871914
it('treats a waiting call as already owned by this tab when the chat recovers it', async () => {
872915
const { finish, launchedToolCallIds } = holdExecutions()
873916

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

Lines changed: 17 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -423,18 +423,14 @@ export async function bindRunToolToExecution(
423423
})
424424
return true
425425
}
426-
if (existingToolCallId && existingToolCallId !== toolCallId) {
427-
logger.warn('[RunTool] Recovery skipped: another run tool is already active', {
428-
workflowId,
429-
toolCallId,
430-
existingToolCallId,
431-
})
432-
return false
433-
}
434-
435-
const pointer = await loadExecutionPointer(workflowId).catch(() => null)
426+
// Another run owns the workflow's pointer, so it says nothing about this call.
427+
const otherRunOwnsWorkflow =
428+
existingToolCallId !== undefined || interruptedExecutionIdByWorkflowId.has(workflowId)
436429
const pendingCompletion = loadPendingCompletionReport(toolCallId)
437430
if (pendingCompletion) {
431+
const pointer = otherRunOwnsWorkflow
432+
? null
433+
: await loadExecutionPointer(workflowId).catch(() => null)
438434
const executionId = pendingCompletion.executionId ?? pointer?.executionId
439435
logger.info('[RunTool] Recovery re-sending pending completion report', {
440436
workflowId,
@@ -453,7 +449,7 @@ export async function bindRunToolToExecution(
453449
executionId
454450
)
455451
clearPendingCompletionReport(toolCallId)
456-
if (pendingCompletion.clearExecutionPointerAfterReport) {
452+
if (pendingCompletion.clearExecutionPointerAfterReport && !otherRunOwnsWorkflow) {
457453
await clearExecutionPointer(workflowId)
458454
}
459455
} catch (error) {
@@ -467,6 +463,16 @@ export async function bindRunToolToExecution(
467463
return true
468464
}
469465

466+
if (otherRunOwnsWorkflow) {
467+
logger.warn('[RunTool] Recovery skipped: another run owns this workflow', {
468+
workflowId,
469+
toolCallId,
470+
existingToolCallId,
471+
})
472+
return false
473+
}
474+
475+
const pointer = await loadExecutionPointer(workflowId).catch(() => null)
470476
if (!pointer?.executionId) {
471477
logger.info('[RunTool] Recovery skipped: no tab-local execution pointer', {
472478
workflowId,

0 commit comments

Comments
 (0)