Skip to content

Commit 2243890

Browse files
authored
fix(workflows): resolve finished resume runs on the run status endpoint (#7894)
1 parent cb67145 commit 2243890

2 files changed

Lines changed: 268 additions & 9 deletions

File tree

‎apps/sim/lib/workflows/executor/execution-status.test.ts‎

Lines changed: 180 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -430,3 +430,183 @@ describe('getWorkflowExecutionStatus queue projection', () => {
430430
})
431431
})
432432
})
433+
434+
describe('getWorkflowExecutionStatus settled resume attempts', () => {
435+
const resumeInput = { ...input, executionId: 'resume-run-1' }
436+
437+
function parentLog(overrides: Record<string, unknown> = {}) {
438+
return {
439+
executionId: 'execution-1',
440+
workflowId: 'workflow-1',
441+
workspaceId: 'workspace-1',
442+
status: 'completed',
443+
level: 'info',
444+
trigger: 'api',
445+
startedAt: new Date('2026-08-05T11:00:00.000Z'),
446+
endedAt: new Date('2026-08-05T12:00:05.000Z'),
447+
totalDurationMs: 3605000,
448+
executionData: { finalOutput: { answer: 42 } },
449+
costTotal: '0.5',
450+
...overrides,
451+
}
452+
}
453+
454+
function settledAttempt(overrides: Record<string, unknown> = {}) {
455+
return {
456+
id: 'resume-entry-1',
457+
parentExecutionId: 'execution-1',
458+
status: 'completed',
459+
queuedAt: new Date('2026-08-05T12:00:00.000Z'),
460+
claimedAt: new Date('2026-08-05T12:00:01.000Z'),
461+
completedAt: new Date('2026-08-05T12:00:05.000Z'),
462+
failureReason: null,
463+
...overrides,
464+
}
465+
}
466+
467+
/** The attempt has no log of its own; the parent run is read second. */
468+
function queueSettledResume(
469+
attempt: Record<string, unknown>,
470+
log: Record<string, unknown>,
471+
pausedRows: unknown[] = []
472+
) {
473+
queueTableRows(schemaMock.workflowExecutionLogs, [])
474+
queueTableRows(schemaMock.resumeQueue, [attempt])
475+
queueTableRows(schemaMock.workflowExecutionLogs, [log])
476+
queueTableRows(schemaMock.resumeQueue, [])
477+
queueTableRows(schemaMock.pausedExecutions, pausedRows)
478+
}
479+
480+
beforeEach(() => {
481+
vi.clearAllMocks()
482+
resetDbChainMock()
483+
mockGetJob.mockResolvedValue(null)
484+
mockMaterializeForDisplayWithBlockOutputs.mockImplementation(async (executionData) => ({
485+
executionData,
486+
blockOutputs: new Map(),
487+
}))
488+
})
489+
490+
it('projects a completed resume from the run it continued, under its own run ID', async () => {
491+
queueSettledResume(settledAttempt(), parentLog())
492+
493+
const status = await getWorkflowExecutionStatus({ ...resumeInput, includeOutput: true })
494+
495+
expect(status).toEqual({
496+
executionId: 'resume-run-1',
497+
workflowId: 'workflow-1',
498+
status: 'completed',
499+
trigger: 'api',
500+
level: 'info',
501+
startedAt: '2026-08-05T12:00:01.000Z',
502+
endedAt: '2026-08-05T12:00:05.000Z',
503+
totalDurationMs: 4000,
504+
paused: null,
505+
cost: { total: 0.5 },
506+
error: null,
507+
finalOutput: { answer: 42 },
508+
blockOutputs: null,
509+
})
510+
expect(mockMaterializeForDisplayWithBlockOutputs).toHaveBeenCalledWith(
511+
expect.anything(),
512+
{ workspaceId: 'workspace-1', workflowId: 'workflow-1', executionId: 'execution-1' },
513+
[]
514+
)
515+
expect(mockGetJob).toHaveBeenCalledWith('workflow-execution:resume-run-1')
516+
})
517+
518+
it('reports the next pause when a completed resume paused the run again', async () => {
519+
queueSettledResume(settledAttempt(), parentLog({ status: 'paused', executionData: {} }), [
520+
{
521+
id: 'paused-1',
522+
status: 'partially_resumed',
523+
pausePoints: {
524+
'context-2': {
525+
contextId: 'context-2',
526+
blockId: 'block-2',
527+
pauseKind: 'human',
528+
resumeStatus: 'paused',
529+
},
530+
},
531+
metadata: {},
532+
resumedCount: 1,
533+
pausedAt: new Date('2026-08-05T12:00:04.000Z'),
534+
nextResumeAt: null,
535+
},
536+
])
537+
538+
const status = await getWorkflowExecutionStatus(resumeInput)
539+
540+
expect(status).toMatchObject({
541+
executionId: 'resume-run-1',
542+
status: 'paused',
543+
endedAt: '2026-08-05T12:00:05.000Z',
544+
paused: { contextId: 'context-2', pausedExecutionId: 'paused-1', resumedCount: 1 },
545+
})
546+
})
547+
548+
it('reports no end time while a later resume is still running the run', async () => {
549+
queueSettledResume(settledAttempt(), parentLog({ status: 'running', endedAt: null }))
550+
551+
const status = await getWorkflowExecutionStatus(resumeInput)
552+
553+
expect(status).toMatchObject({
554+
executionId: 'resume-run-1',
555+
status: 'running',
556+
startedAt: '2026-08-05T12:00:01.000Z',
557+
endedAt: null,
558+
totalDurationMs: null,
559+
})
560+
})
561+
562+
it('reports a failed resume that left the run paused as failed with its reason', async () => {
563+
queueSettledResume(
564+
settledAttempt({ status: 'failed', failureReason: 'Resume execution cancelled' }),
565+
parentLog({ status: 'paused', executionData: {} })
566+
)
567+
568+
const status = await getWorkflowExecutionStatus({ ...resumeInput, includeOutput: true })
569+
570+
expect(status).toMatchObject({
571+
executionId: 'resume-run-1',
572+
status: 'failed',
573+
level: 'error',
574+
error: 'Resume execution cancelled',
575+
endedAt: '2026-08-05T12:00:05.000Z',
576+
paused: null,
577+
finalOutput: null,
578+
blockOutputs: null,
579+
})
580+
})
581+
582+
it("prefers the run's own error when the failed resume failed the run", async () => {
583+
queueSettledResume(
584+
settledAttempt({ status: 'failed', failureReason: 'Unexpected error' }),
585+
parentLog({ status: 'failed', level: 'error', executionData: { error: 'Block 2 timed out' } })
586+
)
587+
588+
const status = await getWorkflowExecutionStatus(resumeInput)
589+
590+
expect(status).toMatchObject({ status: 'failed', error: 'Block 2 timed out' })
591+
})
592+
593+
it('reports a resume that lost to cancellation as cancelled', async () => {
594+
queueSettledResume(
595+
settledAttempt({ status: 'failed', failureReason: 'Paused execution cancelled' }),
596+
parentLog({ status: 'cancelled', executionData: {} })
597+
)
598+
599+
const status = await getWorkflowExecutionStatus(resumeInput)
600+
601+
expect(status).toMatchObject({ status: 'cancelled', level: 'info', error: null })
602+
})
603+
604+
it('returns null when the run a settled resume continued no longer exists', async () => {
605+
queueTableRows(schemaMock.workflowExecutionLogs, [])
606+
queueTableRows(schemaMock.resumeQueue, [settledAttempt()])
607+
queueTableRows(schemaMock.workflowExecutionLogs, [])
608+
queueTableRows(schemaMock.resumeQueue, [])
609+
610+
await expect(getWorkflowExecutionStatus(resumeInput)).resolves.toBeNull()
611+
})
612+
})

‎apps/sim/lib/workflows/executor/execution-status.ts‎

Lines changed: 88 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
import { db } from '@sim/db'
22
import { pausedExecutions, resumeQueue, workflowExecutionLogs } from '@sim/db/schema'
3-
import { and, eq, inArray, sql } from 'drizzle-orm'
3+
import { and, eq, sql } from 'drizzle-orm'
44
import type { WorkflowExecutionStatusResponse } from '@/lib/api/contracts/workflows'
55
import { getJobQueue } from '@/lib/core/async-jobs'
66
import type { Job } from '@/lib/core/async-jobs/types'
@@ -120,6 +120,74 @@ function projectQueueJob(
120120
}
121121
}
122122

123+
interface ResumeAttemptRow {
124+
id: string
125+
parentExecutionId: string
126+
status: string
127+
queuedAt: Date
128+
claimedAt: Date | null
129+
completedAt: Date | null
130+
failureReason: string | null
131+
}
132+
133+
type SettledResumeAttemptRow = ResumeAttemptRow & { status: 'completed' | 'failed' }
134+
135+
function isSettledResumeAttempt(
136+
attempt: ResumeAttemptRow | undefined
137+
): attempt is SettledResumeAttemptRow {
138+
return attempt?.status === 'completed' || attempt?.status === 'failed'
139+
}
140+
141+
/**
142+
* Projects a finished resume attempt as its own run resource.
143+
*
144+
* A resume never writes a log row of its own: it continues the paused run and
145+
* records under the parent's execution ID, so once its queue entry settles the
146+
* run it continued is the only durable record of what it did. The attempt
147+
* keeps its own ID and timings, and borrows the rest from that run.
148+
*
149+
* A `completed` attempt is one whose segment ran to its end — the workflow
150+
* finished, failed, or paused again — so the run's state is the answer,
151+
* including when a later resume has since moved the run on (which is why an
152+
* active run reports no end time). A `failed` attempt never finished its
153+
* segment and may have left the run paused for another attempt; it reads as
154+
* failed with its recorded reason unless the run itself was cancelled or failed
155+
* with a more specific error.
156+
*/
157+
function projectSettledResumeAttempt(
158+
executionId: string,
159+
attempt: SettledResumeAttemptRow,
160+
run: WorkflowExecutionStatusResponse
161+
): WorkflowExecutionStatusResponse {
162+
const startedAt = attempt.claimedAt ?? attempt.queuedAt
163+
const continuesInRun =
164+
attempt.status === 'completed' && (run.status === 'queued' || run.status === 'running')
165+
const endedAt = continuesInRun ? null : attempt.completedAt
166+
const resource: WorkflowExecutionStatusResponse = {
167+
...run,
168+
executionId,
169+
startedAt: startedAt.toISOString(),
170+
endedAt: endedAt?.toISOString() ?? null,
171+
totalDurationMs: endedAt ? Math.max(0, endedAt.getTime() - startedAt.getTime()) : null,
172+
}
173+
if (attempt.status === 'completed') return resource
174+
175+
const cancelled = run.status === 'cancelled'
176+
return {
177+
...resource,
178+
status: cancelled ? 'cancelled' : 'failed',
179+
level: cancelled ? 'info' : 'error',
180+
paused: null,
181+
error: cancelled
182+
? null
183+
: ((run.status === 'failed' ? run.error : null) ??
184+
attempt.failureReason ??
185+
'Resume execution failed'),
186+
finalOutput: null,
187+
blockOutputs: null,
188+
}
189+
}
190+
123191
export interface GetWorkflowExecutionStatusInput {
124192
workflowId: string
125193
executionId: string
@@ -246,25 +314,29 @@ async function readWorkflowExecutionStatus(
246314
)
247315
.limit(1)
248316

249-
const [activeResume] = await db
317+
const [resumeAttempt] = await db
250318
.select({
251319
id: resumeQueue.id,
320+
parentExecutionId: resumeQueue.parentExecutionId,
252321
status: resumeQueue.status,
253322
queuedAt: resumeQueue.queuedAt,
254323
claimedAt: resumeQueue.claimedAt,
324+
completedAt: resumeQueue.completedAt,
325+
failureReason: resumeQueue.failureReason,
255326
})
256327
.from(resumeQueue)
257328
.innerJoin(pausedExecutions, eq(resumeQueue.pausedExecutionId, pausedExecutions.id))
258329
.where(
259-
and(
260-
eq(resumeQueue.newExecutionId, executionId),
261-
eq(pausedExecutions.workflowId, workflowId),
262-
inArray(resumeQueue.status, ['pending', 'claimed'] as const)
263-
)
330+
and(eq(resumeQueue.newExecutionId, executionId), eq(pausedExecutions.workflowId, workflowId))
264331
)
265-
.orderBy(sql`case when ${resumeQueue.status} = 'claimed' then 0 else 1 end`)
332+
.orderBy(sql`case ${resumeQueue.status} when 'claimed' then 0 when 'pending' then 1 else 2 end`)
266333
.limit(1)
267334

335+
const activeResume =
336+
resumeAttempt?.status === 'pending' || resumeAttempt?.status === 'claimed'
337+
? resumeAttempt
338+
: undefined
339+
268340
const hasTerminalLog =
269341
logRow?.status === 'completed' || logRow?.status === 'failed' || logRow?.status === 'cancelled'
270342
const projectedResume = hasTerminalLog ? undefined : activeResume
@@ -304,7 +376,14 @@ async function readWorkflowExecutionStatus(
304376
}
305377
}
306378

307-
if (!logRow) return null
379+
if (!logRow) {
380+
if (!isSettledResumeAttempt(resumeAttempt)) return null
381+
const run = await readWorkflowExecutionStatus({
382+
...input,
383+
executionId: resumeAttempt.parentExecutionId,
384+
})
385+
return run ? projectSettledResumeAttempt(executionId, resumeAttempt, run) : null
386+
}
308387

309388
const [pausedRow] = await db
310389
.select({

0 commit comments

Comments
 (0)