From b6501d0e05a009edd025e7e47232456b0e6d325d Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 29 Sep 2026 09:37:30 -0700 Subject: [PATCH 1/6] fix(tables): settle a resume whose pause cannot be saved as failed, and never fail a completed run A resumed run that paused but whose pause state could not be persisted failed its log yet returned a paused result, so the cell showed paused and the resume entry was marked completed. It now throws after failing the log, so the attempt settles as failed and reports execution_failed. markResumeFailed also rewrote a completed log as failed when a step after a completed run threw. A completed run's outcome now stands, and an already failed log keeps its original end time. --- .../human-in-the-loop-manager.test.ts | 73 +++++++++++++++++++ .../executor/human-in-the-loop-manager.ts | 50 ++++++------- 2 files changed, 98 insertions(+), 25 deletions(-) diff --git a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts index 95205271c98..b3fb9e30c32 100644 --- a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts +++ b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts @@ -22,6 +22,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { createTimeoutAbortController, getExecutionDeadlineAt } from '@/lib/core/execution-limits' import { abortManualExecution } from '@/lib/execution/manual-cancellation' import { terminalExecutionLogFields } from '@/lib/logs/execution/cancellation' +import { LoggingSession } from '@/lib/logs/execution/logging-session' const { mockExecuteWorkflowCore, @@ -265,6 +266,8 @@ describe('what a failed resume did to its paused execution', () => { { logStatus: 'running', pauseStatus: 'paused', executionFailed: true }, { logStatus: 'cancelled', pauseStatus: 'paused', executionFailed: false }, { logStatus: 'running', pauseStatus: 'cancelling', executionFailed: false }, + { logStatus: 'failed', pauseStatus: 'paused', executionFailed: true }, + { logStatus: 'completed', pauseStatus: 'paused', executionFailed: false }, ])( 'reports the execution failed: $executionFailed for a $logStatus log and $pauseStatus pause', async ({ logStatus, pauseStatus, executionFailed }) => { @@ -276,6 +279,23 @@ describe('what a failed resume did to its paused execution', () => { } ) + it.each([ + { logStatus: 'completed', updated: [resumeQueue] }, + { logStatus: 'failed', updated: [resumeQueue, pausedExecutions] }, + { logStatus: 'running', updated: [resumeQueue, pausedExecutions, workflowExecutionLogs] }, + ])( + 'leaves a $logStatus execution log as it is when a resume fails late', + async ({ logStatus, updated }) => { + queueTableRows(workflowExecutionLogs, [{ status: logStatus }]) + queueTableRows(pausedExecutions, [{ status: 'paused' }]) + const managerInternals = PauseResumeManager as unknown as PauseResumeManagerInternals + + await managerInternals.markResumeFailed(attemptArgs) + + expect(dbChainMockFns.update.mock.calls.map(([table]) => table)).toEqual(updated) + } + ) + /** Resume args that collect every outcome the manager reports. */ function argsReportingOutcomes(onAttemptFailed?: () => Promise) { const outcomes: FailedResumeOutcome[] = [] @@ -313,6 +333,59 @@ describe('what a failed resume did to its paused execution', () => { } ) + describe('when the resumed run pauses but its pause cannot be saved', () => { + const spies: { mockRestore: () => void }[] = [] + + function pauseRun(options: { snapshotSeed?: unknown; persistError?: Error }) { + const managerInternals = PauseResumeManager as unknown as PauseResumeManagerInternals + spies.push( + vi.spyOn(managerInternals, 'runResumeExecution').mockResolvedValueOnce({ + success: true, + status: 'paused', + output: {}, + logs: [], + pausePoints: [], + snapshotSeed: options.snapshotSeed, + metadata: { executionId: 'parent-execution-1', duration: 1, startTime: 'start' }, + }), + vi.spyOn(managerInternals, 'markResumeFailed').mockResolvedValueOnce(true), + vi.spyOn(LoggingSession, 'markExecutionAsFailed').mockResolvedValueOnce(), + vi.spyOn(PauseResumeManager, 'processQueuedResumes').mockResolvedValueOnce() + ) + if (options.persistError) { + spies.push( + vi + .spyOn(PauseResumeManager, 'persistPauseResult') + .mockRejectedValueOnce(options.persistError) + ) + } + } + + afterEach(() => { + for (const spy of spies.splice(0)) spy.mockRestore() + }) + + it('fails the attempt when the pause state cannot be persisted', async () => { + pauseRun({ snapshotSeed: createSnapshotSeed(), persistError: new Error('lock timeout') }) + const { outcomes, args } = argsReportingOutcomes() + + await expect(PauseResumeManager.startResumeExecution(args)).rejects.toThrow( + 'Failed to persist pause state: lock timeout' + ) + expect(outcomes).toEqual(['execution_failed']) + }) + + it('fails the attempt when the paused run has no snapshot seed', async () => { + pauseRun({}) + const { outcomes, args } = argsReportingOutcomes() + + await expect(PauseResumeManager.startResumeExecution(args)).rejects.toThrow( + 'Missing snapshot seed for paused execution' + ) + expect(outcomes).toEqual(['execution_failed']) + }) + }) + describe('when the resumed run fails', () => { const rawError = new Error('Block failed') const spies: { mockRestore: () => void }[] = [] diff --git a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts index 732240b12f9..b22559848ff 100644 --- a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts +++ b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts @@ -924,18 +924,22 @@ export class PauseResumeManager { }) if (result.status === 'paused') { + /** + * A pause that cannot be saved fails the execution. Fail the log with the + * reason, then throw so the attempt settles as failed below. + */ const effectiveExecutionId = result.metadata?.executionId ?? resumeExecutionId - if (!result.snapshotSeed) { - logger.error('Missing snapshot seed for paused resume execution', { - resumeExecutionId, - }) + const failPause = async (message: string, cause?: unknown): Promise => { await LoggingSession.markExecutionAsFailed( effectiveExecutionId, - 'Missing snapshot seed for paused execution', + message, undefined, pausedExecution.workflowId ) - await releaseExecutionSlot(resumeEntryId) + throw new Error(message, { cause }) + } + if (!result.snapshotSeed) { + await failPause('Missing snapshot seed for paused execution') } else { try { await PauseResumeManager.persistPauseResult({ @@ -947,19 +951,10 @@ export class PauseResumeManager { executorUserId: result.metadata?.userId, }) } catch (pauseError) { - logger.error( - 'Failed to persist pause result for resumed execution', - projectResolvedSecretDiagnosticError(pauseError, undefined, { - resumeExecutionId, - }) - ) - await LoggingSession.markExecutionAsFailed( - effectiveExecutionId, + await failPause( `Failed to persist pause state: ${toError(pauseError).message}`, - undefined, - pausedExecution.workflowId + pauseError ) - await releaseExecutionSlot(resumeEntryId) } } } else { @@ -2268,6 +2263,9 @@ export class PauseResumeManager { return false } + /** The run completed before a later step threw; its outcome stands. */ + if (executionLog?.status === 'completed') return false + await tx .update(pausedExecutions) .set({ @@ -2278,15 +2276,17 @@ export class PauseResumeManager { if (pausedExecution?.status === 'cancelling') return false - await tx - .update(workflowExecutionLogs) - .set(terminalExecutionLogFields('failed', now)) - .where( - and( - eq(workflowExecutionLogs.executionId, args.parentExecutionId), - sql`${workflowExecutionLogs.status} != 'cancelled'` + if (executionLog?.status !== 'failed') { + await tx + .update(workflowExecutionLogs) + .set(terminalExecutionLogFields('failed', now)) + .where( + and( + eq(workflowExecutionLogs.executionId, args.parentExecutionId), + sql`${workflowExecutionLogs.status} != 'cancelled'` + ) ) - ) + } return true }) From 6be9ba0261ef7230f952fb1c831c5ff761f0d10c Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 29 Sep 2026 09:48:34 -0700 Subject: [PATCH 2/6] fix(tables): settle the cell as completed when a completed resume's later step throws Leaving a completed log alone reported no outcome, so the cell stayed on its last running state. markResumeFailed now reports what the attempt left the execution as, and a run that completed before a later step threw settles the cell as completed. The pause point is still marked failed as before. --- apps/sim/background/resume-execution.ts | 21 ++++++--- .../resume-governed-subject.test.ts | 13 ++++++ .../human-in-the-loop-manager.test.ts | 46 ++++++++++--------- .../executor/human-in-the-loop-manager.ts | 26 +++++------ 4 files changed, 64 insertions(+), 42 deletions(-) diff --git a/apps/sim/background/resume-execution.ts b/apps/sim/background/resume-execution.ts index db86c516b1a..1d95aedbbe4 100644 --- a/apps/sim/background/resume-execution.ts +++ b/apps/sim/background/resume-execution.ts @@ -364,19 +364,26 @@ async function buildResumeCellWriters( /** * A resume that throws never reaches the terminal write in * {@link runResumeAndCellTerminal}, which would leave the cell showing its last - * partial `running` state. Mirror what the failed attempt did to the execution: - * a pause that stayed resumable goes back to paused, and a failed execution - * fails the cell. + * partial `running` state. Mirror what the failed attempt left the execution + * as: a pause that stayed resumable goes back to paused, a failed execution + * fails the cell, and a run that completed before a later step threw + * completes it. */ async function writeFailedResumeCellTerminal( writers: CellWriters, outcome: FailedResumeOutcome, error: unknown ): Promise { - if (outcome === 'pause_retained') { - await writers.writeCellTerminal('paused', null) - } else { - await writers.writeCellTerminal('error', getErrorMessage(error, 'Resume execution failed')) + switch (outcome) { + case 'pause_retained': + await writers.writeCellTerminal('paused', null) + return + case 'execution_completed': + await writers.writeCellTerminal('completed', null) + return + case 'execution_failed': + await writers.writeCellTerminal('error', getErrorMessage(error, 'Resume execution failed')) + return } } diff --git a/apps/sim/background/resume-governed-subject.test.ts b/apps/sim/background/resume-governed-subject.test.ts index d7589d6eb45..117847ded5b 100644 --- a/apps/sim/background/resume-governed-subject.test.ts +++ b/apps/sim/background/resume-governed-subject.test.ts @@ -253,6 +253,19 @@ describe('resuming a paused table cell', () => { }) }, 20_000) + it('marks the cell completed when the run completed before a later step failed', async () => { + const bookkeepingFailure = new Error('Database unavailable') + failResume('execution_completed', bookkeepingFailure) + + await expect(executeResumeJob(PAYLOAD)).rejects.toBe(bookkeepingFailure) + + expect(lastCellExecutionState()).toMatchObject({ + status: 'completed', + executionId: 'parent-execution-1', + error: null, + }) + }, 20_000) + it('puts the cell back to paused when the pause stayed resumable', async () => { const admissionRefusal = new Error('Execution can no longer be resumed') failResume('pause_retained', admissionRefusal) diff --git a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts index b3fb9e30c32..5b21107d277 100644 --- a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts +++ b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts @@ -86,7 +86,7 @@ if (!humanInTheLoopLogger) { } interface PauseResumeManagerInternals { - markResumeFailed: (...args: unknown[]) => Promise + markResumeFailed: (...args: unknown[]) => Promise runResumeExecution: (...args: unknown[]) => Promise } @@ -263,24 +263,24 @@ describe('what a failed resume did to its paused execution', () => { ) it.each([ - { logStatus: 'running', pauseStatus: 'paused', executionFailed: true }, - { logStatus: 'cancelled', pauseStatus: 'paused', executionFailed: false }, - { logStatus: 'running', pauseStatus: 'cancelling', executionFailed: false }, - { logStatus: 'failed', pauseStatus: 'paused', executionFailed: true }, - { logStatus: 'completed', pauseStatus: 'paused', executionFailed: false }, + { logStatus: 'running', pauseStatus: 'paused', outcome: 'execution_failed' }, + { logStatus: 'cancelled', pauseStatus: 'paused', outcome: undefined }, + { logStatus: 'running', pauseStatus: 'cancelling', outcome: undefined }, + { logStatus: 'failed', pauseStatus: 'paused', outcome: 'execution_failed' }, + { logStatus: 'completed', pauseStatus: 'paused', outcome: 'execution_completed' }, ])( - 'reports the execution failed: $executionFailed for a $logStatus log and $pauseStatus pause', - async ({ logStatus, pauseStatus, executionFailed }) => { + 'reports $outcome for a $logStatus log and $pauseStatus pause', + async ({ logStatus, pauseStatus, outcome }) => { queueTableRows(workflowExecutionLogs, [{ status: logStatus }]) queueTableRows(pausedExecutions, [{ status: pauseStatus }]) const managerInternals = PauseResumeManager as unknown as PauseResumeManagerInternals - await expect(managerInternals.markResumeFailed(attemptArgs)).resolves.toBe(executionFailed) + await expect(managerInternals.markResumeFailed(attemptArgs)).resolves.toBe(outcome) } ) it.each([ - { logStatus: 'completed', updated: [resumeQueue] }, + { logStatus: 'completed', updated: [resumeQueue, pausedExecutions] }, { logStatus: 'failed', updated: [resumeQueue, pausedExecutions] }, { logStatus: 'running', updated: [resumeQueue, pausedExecutions, workflowExecutionLogs] }, ])( @@ -348,7 +348,7 @@ describe('what a failed resume did to its paused execution', () => { snapshotSeed: options.snapshotSeed, metadata: { executionId: 'parent-execution-1', duration: 1, startTime: 'start' }, }), - vi.spyOn(managerInternals, 'markResumeFailed').mockResolvedValueOnce(true), + vi.spyOn(managerInternals, 'markResumeFailed').mockResolvedValueOnce('execution_failed'), vi.spyOn(LoggingSession, 'markExecutionAsFailed').mockResolvedValueOnce(), vi.spyOn(PauseResumeManager, 'processQueuedResumes').mockResolvedValueOnce() ) @@ -390,13 +390,14 @@ describe('what a failed resume did to its paused execution', () => { const rawError = new Error('Block failed') const spies: { mockRestore: () => void }[] = [] - function failRun(options: { executionFailed: boolean; queuedResumesError?: Error }) { + function failRun(options: { + outcome: FailedResumeOutcome | undefined + queuedResumesError?: Error + }) { const managerInternals = PauseResumeManager as unknown as PauseResumeManagerInternals spies.push( vi.spyOn(managerInternals, 'runResumeExecution').mockRejectedValueOnce(rawError), - vi - .spyOn(managerInternals, 'markResumeFailed') - .mockResolvedValueOnce(options.executionFailed), + vi.spyOn(managerInternals, 'markResumeFailed').mockResolvedValueOnce(options.outcome), options.queuedResumesError ? vi .spyOn(PauseResumeManager, 'processQueuedResumes') @@ -410,12 +411,13 @@ describe('what a failed resume did to its paused execution', () => { }) it.each([ - { executionFailed: true, reported: ['execution_failed'] }, - { executionFailed: false, reported: [] }, + { outcome: 'execution_failed' as const, reported: ['execution_failed'] }, + { outcome: 'execution_completed' as const, reported: ['execution_completed'] }, + { outcome: undefined, reported: [] }, ])( - 'reports $reported when it failed the execution: $executionFailed', - async ({ executionFailed, reported }) => { - failRun({ executionFailed }) + 'reports $reported when the attempt left the execution $outcome', + async ({ outcome, reported }) => { + failRun({ outcome }) const { outcomes, args } = argsReportingOutcomes() await expect(PauseResumeManager.startResumeExecution(args)).rejects.toBe(rawError) @@ -424,7 +426,7 @@ describe('what a failed resume did to its paused execution', () => { ) it('reports the outcome before draining queued resumes, which may throw', async () => { - failRun({ executionFailed: true, queuedResumesError: new Error('queue drain failed') }) + failRun({ outcome: 'execution_failed', queuedResumesError: new Error('queue drain failed') }) const { outcomes, args } = argsReportingOutcomes() await expect(PauseResumeManager.startResumeExecution(args)).rejects.toThrow( @@ -434,7 +436,7 @@ describe('what a failed resume did to its paused execution', () => { }) it('rethrows the run failure when the outcome handler fails', async () => { - failRun({ executionFailed: true }) + failRun({ outcome: 'execution_failed' }) const { args } = argsReportingOutcomes(async () => { throw new Error('Database unavailable') }) diff --git a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts index b22559848ff..68d0754fa88 100644 --- a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts +++ b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts @@ -149,11 +149,12 @@ class ResumeAdmissionError extends Error { } /** - * What a failed resume attempt did to its paused execution, as reported by the - * transaction that settled the attempt: the pause stayed resumable, or the - * resumed run failed the execution. + * What a failed resume attempt left its paused execution as, read from the + * transaction that settled the attempt: the pause stayed resumable, the + * resumed run failed the execution, or the run completed before a later step + * of the attempt threw. */ -export type FailedResumeOutcome = 'pause_retained' | 'execution_failed' +export type FailedResumeOutcome = 'pause_retained' | 'execution_failed' | 'execution_completed' /** Matches the paused execution mode to the deployment recorded on its durable root log. */ export function requireResumeDeploymentVersion( @@ -1032,14 +1033,13 @@ export class PauseResumeManager { }) if (pauseResumable) outcome = 'pause_retained' } else { - const executionFailed = await PauseResumeManager.markResumeFailed({ + outcome = await PauseResumeManager.markResumeFailed({ resumeEntryId, pausedExecutionId: pausedExecution.id, parentExecutionId: pausedExecution.executionId, contextId, failureReason: message, }) - if (executionFailed) outcome = 'execution_failed' } if (outcome && onAttemptFailed) { await onAttemptFailed(outcome, error).catch((hookError: unknown) => { @@ -2228,7 +2228,7 @@ export class PauseResumeManager { parentExecutionId: string contextId: string failureReason: string - }): Promise { + }): Promise<'execution_failed' | 'execution_completed' | undefined> { const now = new Date() return execDb.transaction(async (tx) => { @@ -2260,12 +2260,9 @@ export class PauseResumeManager { .set({ status: 'cancelled', updatedAt: now, nextResumeAt: null }) .where(eq(pausedExecutions.id, args.pausedExecutionId)) } - return false + return undefined } - /** The run completed before a later step threw; its outcome stands. */ - if (executionLog?.status === 'completed') return false - await tx .update(pausedExecutions) .set({ @@ -2274,7 +2271,10 @@ export class PauseResumeManager { }) .where(eq(pausedExecutions.id, args.pausedExecutionId)) - if (pausedExecution?.status === 'cancelling') return false + if (pausedExecution?.status === 'cancelling') return undefined + + /** The run completed before a later step of the attempt threw; its outcome stands. */ + if (executionLog?.status === 'completed') return 'execution_completed' if (executionLog?.status !== 'failed') { await tx @@ -2288,7 +2288,7 @@ export class PauseResumeManager { ) } - return true + return 'execution_failed' }) } From ccce75f87cce5ee4ecd25eac57930f1a7f662718 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 29 Sep 2026 10:08:03 -0700 Subject: [PATCH 3/6] fix(tables): continue the cascade after a completed resume whose later step threw A completed run's cell is completed, so its downstream workflow groups still start; the failure is still rethrown. A pause that cannot be saved now throws a stable message with the underlying error on cause, so API callers never see internal persistence details. --- apps/sim/background/resume-execution.ts | 35 ++++++++++++++++--- .../resume-governed-subject.test.ts | 10 ++++++ .../human-in-the-loop-manager.test.ts | 7 ++-- .../executor/human-in-the-loop-manager.ts | 14 ++++---- 4 files changed, 51 insertions(+), 15 deletions(-) diff --git a/apps/sim/background/resume-execution.ts b/apps/sim/background/resume-execution.ts index 1d95aedbbe4..b448e261b04 100644 --- a/apps/sim/background/resume-execution.ts +++ b/apps/sim/background/resume-execution.ts @@ -178,13 +178,36 @@ export async function executeResumeJob(payload: ResumeExecutionPayload, signal?: cellContext.rowId, parentExecutionId, async () => { + let completedBeforeFailure = false const result = await runResumeAndCellTerminal( payload, pausedExecution, writers, attemptSignal, - attemptTimeoutController - ) + attemptTimeoutController, + () => { + completedBeforeFailure = true + } + ).catch(async (error: unknown) => { + /** + * The run completed and only a later step of the attempt threw, so its + * cell is completed: continue the cascade as a completed run would, and + * still surface the failure. + */ + if (completedBeforeFailure) { + await continueCascadeAfterResume(cellContext, billingAttribution, attemptSignal).catch( + (cascadeError: unknown) => { + logger.error( + 'Failed to continue the cascade after a completed resume', + projectResolvedSecretDiagnosticError(cascadeError, undefined, { + resumeExecutionId, + }) + ) + } + ) + } + throw error + }) if (result.status === 'paused' || result.status === 'cancelled') return result await continueCascadeAfterResume(cellContext, billingAttribution, attemptSignal) return result @@ -392,7 +415,8 @@ async function runResumeAndCellTerminal( pausedExecution: Awaited>, writers: CellWriters, signal: AbortSignal | undefined, - timeoutController: ReturnType + timeoutController: ReturnType, + onCompletedBeforeFailure?: () => void ): Promise>> { if (!pausedExecution) throw new Error('Paused execution missing — already nulled by caller') const result = await PauseResumeManager.startResumeExecution({ @@ -403,7 +427,10 @@ async function runResumeAndCellTerminal( resumeInput: payload.resumeInput, userId: payload.userId, onBlockComplete: writers.cellOnBlockComplete, - onAttemptFailed: (outcome, error) => writeFailedResumeCellTerminal(writers, outcome, error), + onAttemptFailed: async (outcome, error) => { + if (outcome === 'execution_completed') onCompletedBeforeFailure?.() + await writeFailedResumeCellTerminal(writers, outcome, error) + }, abortSignal: signal, }) diff --git a/apps/sim/background/resume-governed-subject.test.ts b/apps/sim/background/resume-governed-subject.test.ts index 117847ded5b..55150a15bcf 100644 --- a/apps/sim/background/resume-governed-subject.test.ts +++ b/apps/sim/background/resume-governed-subject.test.ts @@ -264,6 +264,16 @@ describe('resuming a paused table cell', () => { executionId: 'parent-execution-1', error: null, }) + expect(mocks.runRowCascadeLoop).toHaveBeenCalledTimes(1) + }, 20_000) + + it('does not continue the cascade when the resume failed the execution', async () => { + const runFailure = new Error('Block failed') + failResume('execution_failed', runFailure) + + await expect(executeResumeJob(PAYLOAD)).rejects.toBe(runFailure) + + expect(mocks.runRowCascadeLoop).not.toHaveBeenCalled() }, 20_000) it('puts the cell back to paused when the pause stayed resumable', async () => { diff --git a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts index 5b21107d277..b5b182b2ef4 100644 --- a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts +++ b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts @@ -369,9 +369,10 @@ describe('what a failed resume did to its paused execution', () => { pauseRun({ snapshotSeed: createSnapshotSeed(), persistError: new Error('lock timeout') }) const { outcomes, args } = argsReportingOutcomes() - await expect(PauseResumeManager.startResumeExecution(args)).rejects.toThrow( - 'Failed to persist pause state: lock timeout' - ) + await expect(PauseResumeManager.startResumeExecution(args)).rejects.toMatchObject({ + message: 'Failed to persist pause state', + cause: new Error('lock timeout'), + }) expect(outcomes).toEqual(['execution_failed']) }) diff --git a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts index 68d0754fa88..63c32790d41 100644 --- a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts +++ b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts @@ -927,17 +927,18 @@ export class PauseResumeManager { if (result.status === 'paused') { /** * A pause that cannot be saved fails the execution. Fail the log with the - * reason, then throw so the attempt settles as failed below. + * reason, then throw so the attempt settles as failed below. The thrown + * message stays stable; the underlying error rides on `cause`. */ const effectiveExecutionId = result.metadata?.executionId ?? resumeExecutionId - const failPause = async (message: string, cause?: unknown): Promise => { + const failPause = async (reason: string, cause?: unknown): Promise => { await LoggingSession.markExecutionAsFailed( effectiveExecutionId, - message, + cause === undefined ? reason : `${reason}: ${toError(cause).message}`, undefined, pausedExecution.workflowId ) - throw new Error(message, { cause }) + throw new Error(reason, { cause }) } if (!result.snapshotSeed) { await failPause('Missing snapshot seed for paused execution') @@ -952,10 +953,7 @@ export class PauseResumeManager { executorUserId: result.metadata?.userId, }) } catch (pauseError) { - await failPause( - `Failed to persist pause state: ${toError(pauseError).message}`, - pauseError - ) + await failPause('Failed to persist pause state', pauseError) } } } else { From 3e15700c0a45910507b7b542653ea96d86a47859 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 29 Sep 2026 10:08:54 -0700 Subject: [PATCH 4/6] fix(tables): keep server-side logging for a resumed pause that cannot be saved --- .../sim/lib/workflows/executor/human-in-the-loop-manager.ts | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts index 63c32790d41..82f64214ac3 100644 --- a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts +++ b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts @@ -932,6 +932,12 @@ export class PauseResumeManager { */ const effectiveExecutionId = result.metadata?.executionId ?? resumeExecutionId const failPause = async (reason: string, cause?: unknown): Promise => { + logger.error( + reason, + cause === undefined + ? { resumeExecutionId } + : projectResolvedSecretDiagnosticError(cause, undefined, { resumeExecutionId }) + ) await LoggingSession.markExecutionAsFailed( effectiveExecutionId, cause === undefined ? reason : `${reason}: ${toError(cause).message}`, From 70755c385e551e8272331a5a344672918a7b24d4 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 29 Sep 2026 10:24:07 -0700 Subject: [PATCH 5/6] fix(tables): continue the cascade only once the completed cell is saved --- apps/sim/background/resume-execution.ts | 3 ++- .../background/resume-governed-subject.test.ts | 17 +++++++++++++++-- 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/apps/sim/background/resume-execution.ts b/apps/sim/background/resume-execution.ts index b448e261b04..f6a8fa9b7cd 100644 --- a/apps/sim/background/resume-execution.ts +++ b/apps/sim/background/resume-execution.ts @@ -428,8 +428,9 @@ async function runResumeAndCellTerminal( userId: payload.userId, onBlockComplete: writers.cellOnBlockComplete, onAttemptFailed: async (outcome, error) => { - if (outcome === 'execution_completed') onCompletedBeforeFailure?.() await writeFailedResumeCellTerminal(writers, outcome, error) + /** Only a cell saved as completed may start its downstream groups. */ + if (outcome === 'execution_completed') onCompletedBeforeFailure?.() }, abortSignal: signal, }) diff --git a/apps/sim/background/resume-governed-subject.test.ts b/apps/sim/background/resume-governed-subject.test.ts index 55150a15bcf..80dbfebd60d 100644 --- a/apps/sim/background/resume-governed-subject.test.ts +++ b/apps/sim/background/resume-governed-subject.test.ts @@ -226,7 +226,10 @@ describe('resuming a paused table cell', () => { return payload?.executionState } - /** Fails the resume the way the manager does: settle, report the outcome, rethrow. */ + /** + * Fails the resume the way the manager does: settle, report the outcome (a + * failing handler is logged, never rethrown), rethrow the attempt's error. + */ function failResume(outcome: FailedResumeOutcome, error: Error) { mocks.startResumeExecution.mockImplementationOnce( async ({ @@ -234,7 +237,7 @@ describe('resuming a paused table cell', () => { }: { onAttemptFailed?: (outcome: FailedResumeOutcome, error: unknown) => Promise }) => { - await onAttemptFailed?.(outcome, error) + await onAttemptFailed?.(outcome, error).catch(() => undefined) throw error } ) @@ -267,6 +270,16 @@ describe('resuming a paused table cell', () => { expect(mocks.runRowCascadeLoop).toHaveBeenCalledTimes(1) }, 20_000) + it('does not continue the cascade when the completed cell could not be saved', async () => { + const bookkeepingFailure = new Error('Database unavailable') + failResume('execution_completed', bookkeepingFailure) + mocks.writeWorkflowGroupState.mockRejectedValueOnce(new Error('Cell write failed')) + + await expect(executeResumeJob(PAYLOAD)).rejects.toBe(bookkeepingFailure) + + expect(mocks.runRowCascadeLoop).not.toHaveBeenCalled() + }, 20_000) + it('does not continue the cascade when the resume failed the execution', async () => { const runFailure = new Error('Block failed') failResume('execution_failed', runFailure) From 4fef9c4eb769e4b622e8cd3285a9fb29edafe4e5 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 29 Sep 2026 10:33:59 -0700 Subject: [PATCH 6/6] test(tables): assert which downstream groups a resume's cascade started --- .../background/resume-governed-subject.test.ts | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/apps/sim/background/resume-governed-subject.test.ts b/apps/sim/background/resume-governed-subject.test.ts index 80dbfebd60d..a689b1e78b0 100644 --- a/apps/sim/background/resume-governed-subject.test.ts +++ b/apps/sim/background/resume-governed-subject.test.ts @@ -220,6 +220,16 @@ describe('resuming a paused table cell', () => { }, 20_000) describe('when the resume throws', () => { + /** Downstream groups the row's cascade started after the resume. */ + let startedGroups: string[] + + beforeEach(() => { + startedGroups = [] + mocks.runRowCascadeLoop.mockImplementation(async (payload: { groupId: string }) => { + startedGroups.push(payload.groupId) + }) + }) + /** The execution state the last cell write persisted. */ function lastCellExecutionState() { const [, payload] = mocks.writeWorkflowGroupState.mock.calls.at(-1) ?? [] @@ -267,7 +277,7 @@ describe('resuming a paused table cell', () => { executionId: 'parent-execution-1', error: null, }) - expect(mocks.runRowCascadeLoop).toHaveBeenCalledTimes(1) + expect(startedGroups).toEqual([NEXT_GROUP.id]) }, 20_000) it('does not continue the cascade when the completed cell could not be saved', async () => { @@ -277,7 +287,7 @@ describe('resuming a paused table cell', () => { await expect(executeResumeJob(PAYLOAD)).rejects.toBe(bookkeepingFailure) - expect(mocks.runRowCascadeLoop).not.toHaveBeenCalled() + expect(startedGroups).toEqual([]) }, 20_000) it('does not continue the cascade when the resume failed the execution', async () => { @@ -286,7 +296,7 @@ describe('resuming a paused table cell', () => { await expect(executeResumeJob(PAYLOAD)).rejects.toBe(runFailure) - expect(mocks.runRowCascadeLoop).not.toHaveBeenCalled() + expect(startedGroups).toEqual([]) }, 20_000) it('puts the cell back to paused when the pause stayed resumable', async () => {