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 33dd706d8a5..da7e4a71e52 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 @@ -14,12 +14,43 @@ import { createTimeoutAbortController, getExecutionDeadlineAt } from '@/lib/core import { abortManualExecution } from '@/lib/execution/manual-cancellation' import { terminalExecutionLogFields } from '@/lib/logs/execution/cancellation' -const { mockReleaseExecutionSlot, mockReplaceLargeValueReferenceKeysWithClient } = vi.hoisted( - () => ({ - mockReleaseExecutionSlot: vi.fn(), - mockReplaceLargeValueReferenceKeysWithClient: vi.fn(), - }) -) +const { + mockReleaseExecutionSlot, + mockReplaceLargeValueReferenceKeysWithClient, + mockPreprocessExecution, + mockExecuteWorkflowCore, + mockCleanupExecutionBase64Cache, + mockResetExecutionStreamBuffer, + mockInitializeExecutionStreamMeta, + mockEventWriter, +} = vi.hoisted(() => ({ + mockReleaseExecutionSlot: vi.fn(), + mockPreprocessExecution: vi.fn(), + mockReplaceLargeValueReferenceKeysWithClient: vi.fn(), + mockExecuteWorkflowCore: vi.fn(), + mockCleanupExecutionBase64Cache: vi.fn(), + mockResetExecutionStreamBuffer: vi.fn(), + mockInitializeExecutionStreamMeta: vi.fn(), + mockEventWriter: { write: vi.fn(), writeTerminal: vi.fn(), close: vi.fn() }, +})) + +vi.mock('@/lib/execution/preprocessing', () => ({ preprocessExecution: mockPreprocessExecution })) + +vi.mock('@/lib/workflows/executor/execution-core', () => ({ + executeWorkflowCore: mockExecuteWorkflowCore, +})) + +vi.mock('@/lib/uploads/utils/user-file-base64.server', () => ({ + cleanupExecutionBase64Cache: mockCleanupExecutionBase64Cache, +})) + +vi.mock('@/lib/execution/event-buffer', () => ({ + createExecutionEventWriter: vi.fn(() => mockEventWriter), + flushExecutionStreamReplayBuffer: vi.fn(), + initializeExecutionStreamMeta: mockInitializeExecutionStreamMeta, + markExecutionStreamTerminal: vi.fn(), + resetExecutionStreamBuffer: mockResetExecutionStreamBuffer, +})) vi.mock('@/lib/billing/calculations/usage-reservation', () => ({ releaseExecutionSlot: mockReleaseExecutionSlot, @@ -1996,3 +2027,174 @@ describe('PauseResumeManager.enqueueOrStartResume admission refusals', () => { await expect(enqueue()).rejects.toMatchObject({ retryable: false }) }) }) + +describe('repeated human review pauses', () => { + beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + }) + + const runResumeExecution = Reflect.get(PauseResumeManager, 'runResumeExecution') as ( + args: Record + ) => Promise + + function createRepeatedReviewResumeArgs(): Record { + const seed = createSnapshotSeed() + const snapshot = JSON.parse(seed.snapshot) + snapshot.metadata = { + ...snapshot.metadata, + workflowId: 'workflow-1', + executionId: 'durable-run', + useDraftState: true, + triggerType: 'manual', + } + snapshot.workflow = { blocks: [], connections: [] } + snapshot.state = { ...createExecutionState(), dagIncomingEdges: {} } + return { + reservationId: 'reservation-1', + resumeExecutionId: 'attempt-1', + pausedExecution: { + id: 'pause-1', + workflowId: 'workflow-1', + executionId: 'durable-run', + executionSnapshot: { ...seed, snapshot: JSON.stringify(snapshot) }, + pausePoints: { hitl_loop0: { blockId: 'hitl', pauseKind: 'human' } }, + }, + contextId: 'hitl_loop0', + resumeInput: { reply: 'partial answer' }, + userId: 'user-1', + } + } + + it('keeps the next pause snapshot attached to the log that admission claimed', async () => { + dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'log-1', deploymentVersionId: null }]) + const stopAfterSnapshot = new Error('stop after snapshot construction') + mockPreprocessExecution.mockRejectedValueOnce(stopAfterSnapshot) + + await expect(runResumeExecution(createRepeatedReviewResumeArgs())).rejects.toBe( + stopAfterSnapshot + ) + + expect(humanInTheLoopLogger.info).toHaveBeenCalledWith( + 'Created resume snapshot', + expect.objectContaining({ metadata: expect.objectContaining({ executionId: 'durable-run' }) }) + ) + expect( + dbChainMockFns.where.mock.calls.some(([condition]) => + flattenMockConditions(condition).some( + (part) => + part.type === 'eq' && + part.left === 'workflowExecutionLogs.executionId' && + part.right === 'durable-run' + ) + ) + ).toBe(true) + }) + + it('cleans the base64 cache under the durable run that the resumed blocks wrote to', async () => { + dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'log-1', deploymentVersionId: null }]) + mockPreprocessExecution.mockResolvedValueOnce({ + success: true, + actorUserId: 'user-1', + executionTimeout: { async: 60_000 }, + }) + mockResetExecutionStreamBuffer.mockResolvedValueOnce(true) + mockInitializeExecutionStreamMeta.mockResolvedValueOnce(true) + mockEventWriter.write.mockResolvedValue({ eventId: 1 }) + mockEventWriter.writeTerminal.mockResolvedValue({ eventId: 2 }) + mockEventWriter.close.mockResolvedValue(undefined) + mockExecuteWorkflowCore.mockResolvedValueOnce({ + success: true, + status: 'completed', + output: {}, + logs: [], + metadata: { executionId: 'durable-run' }, + }) + + await expect(runResumeExecution(createRepeatedReviewResumeArgs())).resolves.toMatchObject({ + status: 'completed', + }) + + expect(mockExecuteWorkflowCore).toHaveBeenCalledWith( + expect.objectContaining({ includeFileBase64: true }) + ) + expect(mockCleanupExecutionBase64Cache).toHaveBeenCalledTimes(1) + expect(mockCleanupExecutionBase64Cache).toHaveBeenCalledWith('durable-run') + }) + + it('settles the answered context exactly once when the same run pauses again', async () => { + const runSpy = vi + .spyOn(PauseResumeManager as unknown as PauseResumeManagerInternals, 'runResumeExecution') + .mockResolvedValueOnce({ + status: 'paused', + success: true, + metadata: { executionId: 'durable-run' }, + snapshotSeed: createSnapshotSeed(), + pausePoints: [{ contextId: 'hitl_loop1', blockId: 'hitl', resumeStatus: 'paused' }], + }) + const persistSpy = vi.spyOn(PauseResumeManager, 'persistPauseResult') + dbChainMockFns.limit.mockResolvedValueOnce([{ status: 'running' }]).mockResolvedValueOnce([ + { + id: 'pause-1', + executionId: 'durable-run', + status: 'paused', + metadata: {}, + pausePoints: { + hitl_loop0: { contextId: 'hitl_loop0', blockId: 'hitl', resumeStatus: 'resuming' }, + }, + }, + ]) + const completeSpy = vi + .spyOn( + PauseResumeManager as unknown as { + markResumeCompleted: (...args: unknown[]) => Promise + }, + 'markResumeCompleted' + ) + .mockResolvedValueOnce() + const processSpy = vi.spyOn(PauseResumeManager, 'processQueuedResumes').mockResolvedValueOnce() + type Args = Parameters[0] + try { + await PauseResumeManager.startResumeExecution({ + resumeEntryId: 'entry-1', + resumeExecutionId: 'attempt-1', + contextId: 'hitl_loop0', + resumeInput: { reply: 'partial answer' }, + userId: 'user-1', + pausedExecution: { + id: 'pause-1', + executionId: 'durable-run', + workflowId: 'workflow-1', + pausePoints: { hitl_loop0: { contextId: 'hitl_loop0', blockId: 'hitl' } }, + executionSnapshot: createSnapshotSeed(), + metadata: {}, + } as Args['pausedExecution'], + }) + expect(persistSpy).toHaveBeenCalledWith( + expect.objectContaining({ executionId: 'durable-run' }) + ) + expect(completeSpy).toHaveBeenCalledWith( + expect.objectContaining({ + parentExecutionId: 'durable-run', + }) + ) + expect(completeSpy.mock.calls[0][0]).not.toHaveProperty('contextId') + expect(dbChainMockFns.set).toHaveBeenCalledWith( + expect.objectContaining({ + resumedCount: 1, + totalPauseCount: 2, + status: 'partially_resumed', + pausePoints: expect.objectContaining({ + hitl_loop0: expect.objectContaining({ resumeStatus: 'resumed' }), + hitl_loop1: expect.objectContaining({ resumeStatus: 'paused' }), + }), + }) + ) + } finally { + runSpy.mockRestore() + persistSpy.mockRestore() + completeSpy.mockRestore() + processSpy.mockRestore() + } + }) +}) 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 bca73899221..818644c1f0c 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 @@ -969,6 +969,7 @@ export class PauseResumeManager { } if (result.status === 'paused') { + /** persistPauseResult already settles the answered context and recounts the merged pauses. */ await PauseResumeManager.markResumeCompleted({ resumeEntryId, pausedExecutionId: pausedExecution.id, @@ -1424,9 +1425,10 @@ export class PauseResumeManager { }) } + /** Resume attempts have separate stream IDs; new pauses must retain the durable run ID. */ const metadata = { ...baseSnapshot.metadata, - executionId: resumeExecutionId, + executionId: parentExecutionId, requestId: baseSnapshot.metadata.requestId, startTime: new Date().toISOString(), userId: effectiveUserId, @@ -2024,7 +2026,7 @@ export class PauseResumeManager { ) }) } - void cleanupExecutionBase64Cache(resumeExecutionId) + void cleanupExecutionBase64Cache(parentExecutionId) } /**