Skip to content

Commit 5e7d17a

Browse files
fix(workflows): preserve durable execution across repeated human pauses (#7798)
* fix(workflows): preserve durable execution across repeated human pauses * fix(workflows): count repeated review completion once * fix(workflows): clean resume base64 cache under the durable run Resumed blocks now write the base64 cache under the durable execution ID, so cleanup must use that ID too; the per-attempt ID matched no entries and left cache bytes and budget counters in place until TTL. --------- Co-authored-by: Waleed Latif <walif6@gmail.com>
1 parent 14c8f36 commit 5e7d17a

2 files changed

Lines changed: 212 additions & 8 deletions

File tree

apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts

Lines changed: 208 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -14,12 +14,43 @@ import { createTimeoutAbortController, getExecutionDeadlineAt } from '@/lib/core
1414
import { abortManualExecution } from '@/lib/execution/manual-cancellation'
1515
import { terminalExecutionLogFields } from '@/lib/logs/execution/cancellation'
1616

17-
const { mockReleaseExecutionSlot, mockReplaceLargeValueReferenceKeysWithClient } = vi.hoisted(
18-
() => ({
19-
mockReleaseExecutionSlot: vi.fn(),
20-
mockReplaceLargeValueReferenceKeysWithClient: vi.fn(),
21-
})
22-
)
17+
const {
18+
mockReleaseExecutionSlot,
19+
mockReplaceLargeValueReferenceKeysWithClient,
20+
mockPreprocessExecution,
21+
mockExecuteWorkflowCore,
22+
mockCleanupExecutionBase64Cache,
23+
mockResetExecutionStreamBuffer,
24+
mockInitializeExecutionStreamMeta,
25+
mockEventWriter,
26+
} = vi.hoisted(() => ({
27+
mockReleaseExecutionSlot: vi.fn(),
28+
mockPreprocessExecution: vi.fn(),
29+
mockReplaceLargeValueReferenceKeysWithClient: vi.fn(),
30+
mockExecuteWorkflowCore: vi.fn(),
31+
mockCleanupExecutionBase64Cache: vi.fn(),
32+
mockResetExecutionStreamBuffer: vi.fn(),
33+
mockInitializeExecutionStreamMeta: vi.fn(),
34+
mockEventWriter: { write: vi.fn(), writeTerminal: vi.fn(), close: vi.fn() },
35+
}))
36+
37+
vi.mock('@/lib/execution/preprocessing', () => ({ preprocessExecution: mockPreprocessExecution }))
38+
39+
vi.mock('@/lib/workflows/executor/execution-core', () => ({
40+
executeWorkflowCore: mockExecuteWorkflowCore,
41+
}))
42+
43+
vi.mock('@/lib/uploads/utils/user-file-base64.server', () => ({
44+
cleanupExecutionBase64Cache: mockCleanupExecutionBase64Cache,
45+
}))
46+
47+
vi.mock('@/lib/execution/event-buffer', () => ({
48+
createExecutionEventWriter: vi.fn(() => mockEventWriter),
49+
flushExecutionStreamReplayBuffer: vi.fn(),
50+
initializeExecutionStreamMeta: mockInitializeExecutionStreamMeta,
51+
markExecutionStreamTerminal: vi.fn(),
52+
resetExecutionStreamBuffer: mockResetExecutionStreamBuffer,
53+
}))
2354

2455
vi.mock('@/lib/billing/calculations/usage-reservation', () => ({
2556
releaseExecutionSlot: mockReleaseExecutionSlot,
@@ -1996,3 +2027,174 @@ describe('PauseResumeManager.enqueueOrStartResume admission refusals', () => {
19962027
await expect(enqueue()).rejects.toMatchObject({ retryable: false })
19972028
})
19982029
})
2030+
2031+
describe('repeated human review pauses', () => {
2032+
beforeEach(() => {
2033+
vi.clearAllMocks()
2034+
resetDbChainMock()
2035+
})
2036+
2037+
const runResumeExecution = Reflect.get(PauseResumeManager, 'runResumeExecution') as (
2038+
args: Record<string, unknown>
2039+
) => Promise<unknown>
2040+
2041+
function createRepeatedReviewResumeArgs(): Record<string, unknown> {
2042+
const seed = createSnapshotSeed()
2043+
const snapshot = JSON.parse(seed.snapshot)
2044+
snapshot.metadata = {
2045+
...snapshot.metadata,
2046+
workflowId: 'workflow-1',
2047+
executionId: 'durable-run',
2048+
useDraftState: true,
2049+
triggerType: 'manual',
2050+
}
2051+
snapshot.workflow = { blocks: [], connections: [] }
2052+
snapshot.state = { ...createExecutionState(), dagIncomingEdges: {} }
2053+
return {
2054+
reservationId: 'reservation-1',
2055+
resumeExecutionId: 'attempt-1',
2056+
pausedExecution: {
2057+
id: 'pause-1',
2058+
workflowId: 'workflow-1',
2059+
executionId: 'durable-run',
2060+
executionSnapshot: { ...seed, snapshot: JSON.stringify(snapshot) },
2061+
pausePoints: { hitl_loop0: { blockId: 'hitl', pauseKind: 'human' } },
2062+
},
2063+
contextId: 'hitl_loop0',
2064+
resumeInput: { reply: 'partial answer' },
2065+
userId: 'user-1',
2066+
}
2067+
}
2068+
2069+
it('keeps the next pause snapshot attached to the log that admission claimed', async () => {
2070+
dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'log-1', deploymentVersionId: null }])
2071+
const stopAfterSnapshot = new Error('stop after snapshot construction')
2072+
mockPreprocessExecution.mockRejectedValueOnce(stopAfterSnapshot)
2073+
2074+
await expect(runResumeExecution(createRepeatedReviewResumeArgs())).rejects.toBe(
2075+
stopAfterSnapshot
2076+
)
2077+
2078+
expect(humanInTheLoopLogger.info).toHaveBeenCalledWith(
2079+
'Created resume snapshot',
2080+
expect.objectContaining({ metadata: expect.objectContaining({ executionId: 'durable-run' }) })
2081+
)
2082+
expect(
2083+
dbChainMockFns.where.mock.calls.some(([condition]) =>
2084+
flattenMockConditions(condition).some(
2085+
(part) =>
2086+
part.type === 'eq' &&
2087+
part.left === 'workflowExecutionLogs.executionId' &&
2088+
part.right === 'durable-run'
2089+
)
2090+
)
2091+
).toBe(true)
2092+
})
2093+
2094+
it('cleans the base64 cache under the durable run that the resumed blocks wrote to', async () => {
2095+
dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'log-1', deploymentVersionId: null }])
2096+
mockPreprocessExecution.mockResolvedValueOnce({
2097+
success: true,
2098+
actorUserId: 'user-1',
2099+
executionTimeout: { async: 60_000 },
2100+
})
2101+
mockResetExecutionStreamBuffer.mockResolvedValueOnce(true)
2102+
mockInitializeExecutionStreamMeta.mockResolvedValueOnce(true)
2103+
mockEventWriter.write.mockResolvedValue({ eventId: 1 })
2104+
mockEventWriter.writeTerminal.mockResolvedValue({ eventId: 2 })
2105+
mockEventWriter.close.mockResolvedValue(undefined)
2106+
mockExecuteWorkflowCore.mockResolvedValueOnce({
2107+
success: true,
2108+
status: 'completed',
2109+
output: {},
2110+
logs: [],
2111+
metadata: { executionId: 'durable-run' },
2112+
})
2113+
2114+
await expect(runResumeExecution(createRepeatedReviewResumeArgs())).resolves.toMatchObject({
2115+
status: 'completed',
2116+
})
2117+
2118+
expect(mockExecuteWorkflowCore).toHaveBeenCalledWith(
2119+
expect.objectContaining({ includeFileBase64: true })
2120+
)
2121+
expect(mockCleanupExecutionBase64Cache).toHaveBeenCalledTimes(1)
2122+
expect(mockCleanupExecutionBase64Cache).toHaveBeenCalledWith('durable-run')
2123+
})
2124+
2125+
it('settles the answered context exactly once when the same run pauses again', async () => {
2126+
const runSpy = vi
2127+
.spyOn(PauseResumeManager as unknown as PauseResumeManagerInternals, 'runResumeExecution')
2128+
.mockResolvedValueOnce({
2129+
status: 'paused',
2130+
success: true,
2131+
metadata: { executionId: 'durable-run' },
2132+
snapshotSeed: createSnapshotSeed(),
2133+
pausePoints: [{ contextId: 'hitl_loop1', blockId: 'hitl', resumeStatus: 'paused' }],
2134+
})
2135+
const persistSpy = vi.spyOn(PauseResumeManager, 'persistPauseResult')
2136+
dbChainMockFns.limit.mockResolvedValueOnce([{ status: 'running' }]).mockResolvedValueOnce([
2137+
{
2138+
id: 'pause-1',
2139+
executionId: 'durable-run',
2140+
status: 'paused',
2141+
metadata: {},
2142+
pausePoints: {
2143+
hitl_loop0: { contextId: 'hitl_loop0', blockId: 'hitl', resumeStatus: 'resuming' },
2144+
},
2145+
},
2146+
])
2147+
const completeSpy = vi
2148+
.spyOn(
2149+
PauseResumeManager as unknown as {
2150+
markResumeCompleted: (...args: unknown[]) => Promise<void>
2151+
},
2152+
'markResumeCompleted'
2153+
)
2154+
.mockResolvedValueOnce()
2155+
const processSpy = vi.spyOn(PauseResumeManager, 'processQueuedResumes').mockResolvedValueOnce()
2156+
type Args = Parameters<typeof PauseResumeManager.startResumeExecution>[0]
2157+
try {
2158+
await PauseResumeManager.startResumeExecution({
2159+
resumeEntryId: 'entry-1',
2160+
resumeExecutionId: 'attempt-1',
2161+
contextId: 'hitl_loop0',
2162+
resumeInput: { reply: 'partial answer' },
2163+
userId: 'user-1',
2164+
pausedExecution: {
2165+
id: 'pause-1',
2166+
executionId: 'durable-run',
2167+
workflowId: 'workflow-1',
2168+
pausePoints: { hitl_loop0: { contextId: 'hitl_loop0', blockId: 'hitl' } },
2169+
executionSnapshot: createSnapshotSeed(),
2170+
metadata: {},
2171+
} as Args['pausedExecution'],
2172+
})
2173+
expect(persistSpy).toHaveBeenCalledWith(
2174+
expect.objectContaining({ executionId: 'durable-run' })
2175+
)
2176+
expect(completeSpy).toHaveBeenCalledWith(
2177+
expect.objectContaining({
2178+
parentExecutionId: 'durable-run',
2179+
})
2180+
)
2181+
expect(completeSpy.mock.calls[0][0]).not.toHaveProperty('contextId')
2182+
expect(dbChainMockFns.set).toHaveBeenCalledWith(
2183+
expect.objectContaining({
2184+
resumedCount: 1,
2185+
totalPauseCount: 2,
2186+
status: 'partially_resumed',
2187+
pausePoints: expect.objectContaining({
2188+
hitl_loop0: expect.objectContaining({ resumeStatus: 'resumed' }),
2189+
hitl_loop1: expect.objectContaining({ resumeStatus: 'paused' }),
2190+
}),
2191+
})
2192+
)
2193+
} finally {
2194+
runSpy.mockRestore()
2195+
persistSpy.mockRestore()
2196+
completeSpy.mockRestore()
2197+
processSpy.mockRestore()
2198+
}
2199+
})
2200+
})

apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -969,6 +969,7 @@ export class PauseResumeManager {
969969
}
970970

971971
if (result.status === 'paused') {
972+
/** persistPauseResult already settles the answered context and recounts the merged pauses. */
972973
await PauseResumeManager.markResumeCompleted({
973974
resumeEntryId,
974975
pausedExecutionId: pausedExecution.id,
@@ -1424,9 +1425,10 @@ export class PauseResumeManager {
14241425
})
14251426
}
14261427

1428+
/** Resume attempts have separate stream IDs; new pauses must retain the durable run ID. */
14271429
const metadata = {
14281430
...baseSnapshot.metadata,
1429-
executionId: resumeExecutionId,
1431+
executionId: parentExecutionId,
14301432
requestId: baseSnapshot.metadata.requestId,
14311433
startTime: new Date().toISOString(),
14321434
userId: effectiveUserId,
@@ -2024,7 +2026,7 @@ export class PauseResumeManager {
20242026
)
20252027
})
20262028
}
2027-
void cleanupExecutionBase64Cache(resumeExecutionId)
2029+
void cleanupExecutionBase64Cache(parentExecutionId)
20282030
}
20292031

20302032
/**

0 commit comments

Comments
 (0)