From b078d0d1101d99afcc1d15ba34cb90a1553a8885 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 16 Sep 2026 15:25:57 -0700 Subject: [PATCH] fix(workflows): resolve finished resume runs on the run status endpoint --- .../executor/execution-status.test.ts | 180 ++++++++++++++++++ .../workflows/executor/execution-status.ts | 97 +++++++++- 2 files changed, 268 insertions(+), 9 deletions(-) diff --git a/apps/sim/lib/workflows/executor/execution-status.test.ts b/apps/sim/lib/workflows/executor/execution-status.test.ts index e593b9614fc..1a086357676 100644 --- a/apps/sim/lib/workflows/executor/execution-status.test.ts +++ b/apps/sim/lib/workflows/executor/execution-status.test.ts @@ -430,3 +430,183 @@ describe('getWorkflowExecutionStatus queue projection', () => { }) }) }) + +describe('getWorkflowExecutionStatus settled resume attempts', () => { + const resumeInput = { ...input, executionId: 'resume-run-1' } + + function parentLog(overrides: Record = {}) { + return { + executionId: 'execution-1', + workflowId: 'workflow-1', + workspaceId: 'workspace-1', + status: 'completed', + level: 'info', + trigger: 'api', + startedAt: new Date('2026-08-05T11:00:00.000Z'), + endedAt: new Date('2026-08-05T12:00:05.000Z'), + totalDurationMs: 3605000, + executionData: { finalOutput: { answer: 42 } }, + costTotal: '0.5', + ...overrides, + } + } + + function settledAttempt(overrides: Record = {}) { + return { + id: 'resume-entry-1', + parentExecutionId: 'execution-1', + status: 'completed', + queuedAt: new Date('2026-08-05T12:00:00.000Z'), + claimedAt: new Date('2026-08-05T12:00:01.000Z'), + completedAt: new Date('2026-08-05T12:00:05.000Z'), + failureReason: null, + ...overrides, + } + } + + /** The attempt has no log of its own; the parent run is read second. */ + function queueSettledResume( + attempt: Record, + log: Record, + pausedRows: unknown[] = [] + ) { + queueTableRows(schemaMock.workflowExecutionLogs, []) + queueTableRows(schemaMock.resumeQueue, [attempt]) + queueTableRows(schemaMock.workflowExecutionLogs, [log]) + queueTableRows(schemaMock.resumeQueue, []) + queueTableRows(schemaMock.pausedExecutions, pausedRows) + } + + beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + mockGetJob.mockResolvedValue(null) + mockMaterializeForDisplayWithBlockOutputs.mockImplementation(async (executionData) => ({ + executionData, + blockOutputs: new Map(), + })) + }) + + it('projects a completed resume from the run it continued, under its own run ID', async () => { + queueSettledResume(settledAttempt(), parentLog()) + + const status = await getWorkflowExecutionStatus({ ...resumeInput, includeOutput: true }) + + expect(status).toEqual({ + executionId: 'resume-run-1', + workflowId: 'workflow-1', + status: 'completed', + trigger: 'api', + level: 'info', + startedAt: '2026-08-05T12:00:01.000Z', + endedAt: '2026-08-05T12:00:05.000Z', + totalDurationMs: 4000, + paused: null, + cost: { total: 0.5 }, + error: null, + finalOutput: { answer: 42 }, + blockOutputs: null, + }) + expect(mockMaterializeForDisplayWithBlockOutputs).toHaveBeenCalledWith( + expect.anything(), + { workspaceId: 'workspace-1', workflowId: 'workflow-1', executionId: 'execution-1' }, + [] + ) + expect(mockGetJob).toHaveBeenCalledWith('workflow-execution:resume-run-1') + }) + + it('reports the next pause when a completed resume paused the run again', async () => { + queueSettledResume(settledAttempt(), parentLog({ status: 'paused', executionData: {} }), [ + { + id: 'paused-1', + status: 'partially_resumed', + pausePoints: { + 'context-2': { + contextId: 'context-2', + blockId: 'block-2', + pauseKind: 'human', + resumeStatus: 'paused', + }, + }, + metadata: {}, + resumedCount: 1, + pausedAt: new Date('2026-08-05T12:00:04.000Z'), + nextResumeAt: null, + }, + ]) + + const status = await getWorkflowExecutionStatus(resumeInput) + + expect(status).toMatchObject({ + executionId: 'resume-run-1', + status: 'paused', + endedAt: '2026-08-05T12:00:05.000Z', + paused: { contextId: 'context-2', pausedExecutionId: 'paused-1', resumedCount: 1 }, + }) + }) + + it('reports no end time while a later resume is still running the run', async () => { + queueSettledResume(settledAttempt(), parentLog({ status: 'running', endedAt: null })) + + const status = await getWorkflowExecutionStatus(resumeInput) + + expect(status).toMatchObject({ + executionId: 'resume-run-1', + status: 'running', + startedAt: '2026-08-05T12:00:01.000Z', + endedAt: null, + totalDurationMs: null, + }) + }) + + it('reports a failed resume that left the run paused as failed with its reason', async () => { + queueSettledResume( + settledAttempt({ status: 'failed', failureReason: 'Resume execution cancelled' }), + parentLog({ status: 'paused', executionData: {} }) + ) + + const status = await getWorkflowExecutionStatus({ ...resumeInput, includeOutput: true }) + + expect(status).toMatchObject({ + executionId: 'resume-run-1', + status: 'failed', + level: 'error', + error: 'Resume execution cancelled', + endedAt: '2026-08-05T12:00:05.000Z', + paused: null, + finalOutput: null, + blockOutputs: null, + }) + }) + + it("prefers the run's own error when the failed resume failed the run", async () => { + queueSettledResume( + settledAttempt({ status: 'failed', failureReason: 'Unexpected error' }), + parentLog({ status: 'failed', level: 'error', executionData: { error: 'Block 2 timed out' } }) + ) + + const status = await getWorkflowExecutionStatus(resumeInput) + + expect(status).toMatchObject({ status: 'failed', error: 'Block 2 timed out' }) + }) + + it('reports a resume that lost to cancellation as cancelled', async () => { + queueSettledResume( + settledAttempt({ status: 'failed', failureReason: 'Paused execution cancelled' }), + parentLog({ status: 'cancelled', executionData: {} }) + ) + + const status = await getWorkflowExecutionStatus(resumeInput) + + expect(status).toMatchObject({ status: 'cancelled', level: 'info', error: null }) + }) + + it('returns null when the run a settled resume continued no longer exists', async () => { + queueTableRows(schemaMock.workflowExecutionLogs, []) + queueTableRows(schemaMock.resumeQueue, [settledAttempt()]) + queueTableRows(schemaMock.workflowExecutionLogs, []) + queueTableRows(schemaMock.resumeQueue, []) + + await expect(getWorkflowExecutionStatus(resumeInput)).resolves.toBeNull() + }) +}) diff --git a/apps/sim/lib/workflows/executor/execution-status.ts b/apps/sim/lib/workflows/executor/execution-status.ts index 873dc5182ed..edac4c8649f 100644 --- a/apps/sim/lib/workflows/executor/execution-status.ts +++ b/apps/sim/lib/workflows/executor/execution-status.ts @@ -1,6 +1,6 @@ import { db } from '@sim/db' import { pausedExecutions, resumeQueue, workflowExecutionLogs } from '@sim/db/schema' -import { and, eq, inArray, sql } from 'drizzle-orm' +import { and, eq, sql } from 'drizzle-orm' import type { WorkflowExecutionStatusResponse } from '@/lib/api/contracts/workflows' import { getJobQueue } from '@/lib/core/async-jobs' import type { Job } from '@/lib/core/async-jobs/types' @@ -120,6 +120,74 @@ function projectQueueJob( } } +interface ResumeAttemptRow { + id: string + parentExecutionId: string + status: string + queuedAt: Date + claimedAt: Date | null + completedAt: Date | null + failureReason: string | null +} + +type SettledResumeAttemptRow = ResumeAttemptRow & { status: 'completed' | 'failed' } + +function isSettledResumeAttempt( + attempt: ResumeAttemptRow | undefined +): attempt is SettledResumeAttemptRow { + return attempt?.status === 'completed' || attempt?.status === 'failed' +} + +/** + * Projects a finished resume attempt as its own run resource. + * + * A resume never writes a log row of its own: it continues the paused run and + * records under the parent's execution ID, so once its queue entry settles the + * run it continued is the only durable record of what it did. The attempt + * keeps its own ID and timings, and borrows the rest from that run. + * + * A `completed` attempt is one whose segment ran to its end — the workflow + * finished, failed, or paused again — so the run's state is the answer, + * including when a later resume has since moved the run on (which is why an + * active run reports no end time). A `failed` attempt never finished its + * segment and may have left the run paused for another attempt; it reads as + * failed with its recorded reason unless the run itself was cancelled or failed + * with a more specific error. + */ +function projectSettledResumeAttempt( + executionId: string, + attempt: SettledResumeAttemptRow, + run: WorkflowExecutionStatusResponse +): WorkflowExecutionStatusResponse { + const startedAt = attempt.claimedAt ?? attempt.queuedAt + const continuesInRun = + attempt.status === 'completed' && (run.status === 'queued' || run.status === 'running') + const endedAt = continuesInRun ? null : attempt.completedAt + const resource: WorkflowExecutionStatusResponse = { + ...run, + executionId, + startedAt: startedAt.toISOString(), + endedAt: endedAt?.toISOString() ?? null, + totalDurationMs: endedAt ? Math.max(0, endedAt.getTime() - startedAt.getTime()) : null, + } + if (attempt.status === 'completed') return resource + + const cancelled = run.status === 'cancelled' + return { + ...resource, + status: cancelled ? 'cancelled' : 'failed', + level: cancelled ? 'info' : 'error', + paused: null, + error: cancelled + ? null + : ((run.status === 'failed' ? run.error : null) ?? + attempt.failureReason ?? + 'Resume execution failed'), + finalOutput: null, + blockOutputs: null, + } +} + export interface GetWorkflowExecutionStatusInput { workflowId: string executionId: string @@ -246,25 +314,29 @@ async function readWorkflowExecutionStatus( ) .limit(1) - const [activeResume] = await db + const [resumeAttempt] = await db .select({ id: resumeQueue.id, + parentExecutionId: resumeQueue.parentExecutionId, status: resumeQueue.status, queuedAt: resumeQueue.queuedAt, claimedAt: resumeQueue.claimedAt, + completedAt: resumeQueue.completedAt, + failureReason: resumeQueue.failureReason, }) .from(resumeQueue) .innerJoin(pausedExecutions, eq(resumeQueue.pausedExecutionId, pausedExecutions.id)) .where( - and( - eq(resumeQueue.newExecutionId, executionId), - eq(pausedExecutions.workflowId, workflowId), - inArray(resumeQueue.status, ['pending', 'claimed'] as const) - ) + and(eq(resumeQueue.newExecutionId, executionId), eq(pausedExecutions.workflowId, workflowId)) ) - .orderBy(sql`case when ${resumeQueue.status} = 'claimed' then 0 else 1 end`) + .orderBy(sql`case ${resumeQueue.status} when 'claimed' then 0 when 'pending' then 1 else 2 end`) .limit(1) + const activeResume = + resumeAttempt?.status === 'pending' || resumeAttempt?.status === 'claimed' + ? resumeAttempt + : undefined + const hasTerminalLog = logRow?.status === 'completed' || logRow?.status === 'failed' || logRow?.status === 'cancelled' const projectedResume = hasTerminalLog ? undefined : activeResume @@ -304,7 +376,14 @@ async function readWorkflowExecutionStatus( } } - if (!logRow) return null + if (!logRow) { + if (!isSettledResumeAttempt(resumeAttempt)) return null + const run = await readWorkflowExecutionStatus({ + ...input, + executionId: resumeAttempt.parentExecutionId, + }) + return run ? projectSettledResumeAttempt(executionId, resumeAttempt, run) : null + } const [pausedRow] = await db .select({