Skip to content

Commit bb12ff2

Browse files
committed
fix(mothership): answer a workflow tool still running only once its run is readable
A run past the call's wait was answered running as soon as it was handed to the executor, before the executor wrote its run log, so an immediate workflows runs wait could find no run. The call now answers running only once the run's log exists, re-reading it each second until then or until the run settles; the dispatch hook is gone.
1 parent 686848e commit bb12ff2

3 files changed

Lines changed: 40 additions & 20 deletions

File tree

‎apps/sim/lib/mothership/tools/handlers/workflow/mutations.test.ts‎

Lines changed: 28 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -15,9 +15,14 @@ const { mocks } = vi.hoisted(() => ({
1515
executeWorkflowUseCase: vi.fn(),
1616
hasExecutionResult: vi.fn(),
1717
readAttemptedExecutionId: vi.fn(),
18+
logStatus: vi.fn(),
1819
},
1920
}))
2021

22+
vi.mock('@/lib/workflows/executor/execution-state', () => ({
23+
getWorkflowExecutionLogStatus: mocks.logStatus,
24+
}))
25+
2126
vi.mock('@/lib/mothership/application/execute-workflow-use-case', () => ({
2227
executeCopilotWorkflowUseCase: mocks.executeWorkflowUseCase,
2328
messageForCopilotWorkflowError: (error: unknown, fallback = 'Workflow operation failed') =>
@@ -445,12 +450,9 @@ describe('a workflow tool run past its wait', () => {
445450
const started = new Promise((resolve) => {
446451
settle = resolve
447452
})
448-
mocks.executeWorkflowUseCase.mockImplementationOnce(
449-
(_context: unknown, _useCase: unknown, input: { lifecycle: { onDispatch?: () => void } }) => {
450-
if (dispatch) input.lifecycle.onDispatch?.()
451-
return started
452-
}
453-
)
453+
mocks.executeWorkflowUseCase.mockReturnValueOnce(started)
454+
// The run's log exists once it started; a run still in preflight has none.
455+
mocks.logStatus.mockResolvedValue(dispatch ? 'running' : undefined)
454456
return {
455457
settle,
456458
lifecycle: () =>
@@ -493,7 +495,7 @@ describe('a workflow tool run past its wait', () => {
493495
await expect(held[0]).resolves.toMatchObject({ success: true })
494496
})
495497

496-
it('answers a run still in preflight past its wait with how it settles', async () => {
498+
it('waits on a run past its wait until its log exists, answering one that never starts with how it settles', async () => {
497499
const run = pendingRun(false)
498500
let answered = false
499501
const answer = executeRunWorkflow(
@@ -509,6 +511,25 @@ describe('a workflow tool run past its wait', () => {
509511
await expect(answer).resolves.toMatchObject({ success: false })
510512
})
511513

514+
it('answers a run still starting up running once its log exists', async () => {
515+
const run = pendingRun()
516+
mocks.logStatus.mockReset().mockResolvedValueOnce(undefined).mockResolvedValue('running')
517+
let answered = false
518+
const answer = executeRunWorkflow(
519+
{ workflowId: 'workflow-1' },
520+
{ ...context, workflowRunWaitMs: 1_000 }
521+
).then((result) => {
522+
answered = true
523+
return result
524+
})
525+
await vi.advanceTimersByTimeAsync(1_000)
526+
expect(answered).toBe(false)
527+
await vi.advanceTimersByTimeAsync(1_000)
528+
await expect(answer).resolves.toMatchObject({ output: { status: 'running' } })
529+
expect(mocks.logStatus).toHaveBeenCalledWith(run.lifecycle().executionId, 'workflow-1')
530+
run.settle({ success: true, output: {}, logs: [] })
531+
})
532+
512533
it('answers with the result when the run settles within its wait', async () => {
513534
const run = pendingRun()
514535
const answer = executeRunBlock(

‎apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts‎

Lines changed: 12 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@ import {
3939
runWorkflowFromCopilot,
4040
runWorkflowUntilBlockFromCopilot,
4141
} from '@/lib/workflows/application/run-workflow-from-copilot'
42+
import { getWorkflowExecutionLogStatus } from '@/lib/workflows/executor/execution-state'
4243
import { hasExecutionResult, readAttemptedExecutionId } from '@/executor/utils/errors'
4344
import type { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry'
4445

@@ -257,6 +258,9 @@ function copilotRunLifecycle(context: ExecutionContext) {
257258
}
258259
}
259260

261+
/** How often a run past its wait is re-read for its log before it is answered running. */
262+
const RUN_LOG_POLL_MS = 1000
263+
260264
/**
261265
* Runs a workflow tool's run and answers with its result. When the call names a wait the run
262266
* outlasts (`workflowRunWaitMs`), it answers the run still running by the execution id it gave
@@ -272,20 +276,18 @@ async function answerWithinWait(
272276
const waitMs = context.workflowRunWaitMs
273277
if (waitMs === undefined) return run(lifecycle)
274278
const executionId = lifecycle.boundExecution?.executionId ?? generateId()
275-
let markDispatched: () => void = () => {}
276-
const dispatched = new Promise<null>((resolve) => {
277-
markDispatched = () => resolve(null)
278-
})
279-
const settled = run({ ...lifecycle, executionId, onDispatch: () => markDispatched() })
279+
const settled = run({ ...lifecycle, executionId })
280280
retainSandboxExecution(settled)
281281
const stopWaiting = new AbortController()
282-
const waited = interruptibleSleep(waitMs, stopWaiting.signal).then(() => null)
282+
const waited = (ms: number) => interruptibleSleep(ms, stopWaiting.signal).then(() => null)
283283
try {
284-
const result = await Promise.race([settled, waited])
284+
let result = await Promise.race([settled, waited(waitMs)])
285+
// Answered running only once the run is readable by its id; one still starting up (or in
286+
// preflight) is waited on until its log exists or it settles.
287+
while (!result && !(await getWorkflowExecutionLogStatus(executionId, workflowId))) {
288+
result = await Promise.race([settled, waited(RUN_LOG_POLL_MS)])
289+
}
285290
if (result) return result
286-
// Only a run handed to the executor is answered running; one still in preflight settles first.
287-
const started = await Promise.race([settled, dispatched])
288-
if (started) return started
289291
} finally {
290292
stopWaiting.abort()
291293
}

‎apps/sim/lib/workflows/application/run-workflow-from-copilot.ts‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -65,8 +65,6 @@ export interface CopilotWorkflowRunLifecycle {
6565
* before it settles (a workflow tool that answers a still-running run by its id).
6666
*/
6767
executionId?: string
68-
/** Called once authorization and admission passed and the run is handed to the executor. */
69-
onDispatch?: () => void
7068
}
7169

7270
interface BaseCopilotRunInput {
@@ -309,7 +307,6 @@ async function executeCopilotRun(params: {
309307
* can already make with the id it was handed. Keep the executor call first: anything
310308
* inserted above it would be reported as a run that may exist.
311309
*/
312-
params.input.lifecycle.onDispatch?.()
313310
try {
314311
const result = await executeWorkflow(
315312
{

0 commit comments

Comments
 (0)