fix(execution): stop terminal-race resumes before cleanup

Author: icecrasher321

apps/sim/lib/execution/cancel-workflow-execution.test.ts

@@ -17,6 +17,7 @@ const {
   mockFinalizePausedCancellationForTerminalRun,
   mockGetPausedCancellationStatus,
   mockGetActiveResumeCancellationTarget,
+  mockGetActiveResumeCancellationTargets,
   mockRollbackActiveResumeCancellation,
   mockFinalizeExecutionStream,
   mockReadExecutionMetaState,
@@ -39,6 +40,7 @@ const {
   mockFinalizePausedCancellationForTerminalRun: vi.fn(),
   mockGetPausedCancellationStatus: vi.fn(),
   mockGetActiveResumeCancellationTarget: vi.fn(),
+  mockGetActiveResumeCancellationTargets: vi.fn(),
   mockRollbackActiveResumeCancellation: vi.fn(),
   mockFinalizeExecutionStream: vi.fn(),
   mockReadExecutionMetaState: vi.fn(),
@@ -82,6 +84,8 @@ vi.mock('@/lib/workflows/executor/human-in-the-loop-manager', () => ({
     getPausedCancellationStatus: (...args: unknown[]) => mockGetPausedCancellationStatus(...args),
     getActiveResumeCancellationTarget: (...args: unknown[]) =>
       mockGetActiveResumeCancellationTarget(...args),
+    getActiveResumeCancellationTargets: (...args: unknown[]) =>
+      mockGetActiveResumeCancellationTargets(...args),
     rollbackActiveResumeCancellation: (...args: unknown[]) =>
       mockRollbackActiveResumeCancellation(...args),
   },
@@ -191,6 +195,7 @@ describe('cancelWorkflowExecution', () => {
     mockFinalizePausedCancellationForTerminalRun.mockReset().mockResolvedValue(true)
     mockGetPausedCancellationStatus.mockReset().mockResolvedValue(null)
     mockGetActiveResumeCancellationTarget.mockReset().mockResolvedValue(null)
+    mockGetActiveResumeCancellationTargets.mockReset().mockResolvedValue([])
     mockRollbackActiveResumeCancellation.mockReset().mockResolvedValue(true)
     mockFinalizeExecutionStream.mockReset().mockResolvedValue(true)
     mockReadExecutionMetaState.mockReset().mockResolvedValue({ status: 'missing' })
@@ -1650,7 +1655,7 @@ describe('cancelWorkflowExecution', () => {
     })
     expect(mockWriteTerminalEvent).not.toHaveBeenCalled()
     expect(mockCompletePausedCancellation).not.toHaveBeenCalled()
-    expect(mockFinalizePausedCancellationForTerminalRun).toHaveBeenCalledWith('ex-1', 'wf-1')
+    expect(mockFinalizePausedCancellationForTerminalRun).toHaveBeenCalledWith('ex-1', 'wf-1', [])
   })
 
   it('retries paused cancellation finalization before returning a terminal-race conflict', async () => {
@@ -1682,6 +1687,7 @@ describe('cancelWorkflowExecution', () => {
       kind: 'active_resume',
       target: ACTIVE_RESUME_TARGET,
     })
+    mockGetActiveResumeCancellationTargets.mockResolvedValue([ACTIVE_RESUME_TARGET])
     mockMarkExecutionCancelled.mockResolvedValue({ durablyRecorded: true, reason: 'recorded' })
     const returning = vi.fn().mockResolvedValue([])
     const where = vi.fn(() => ({ returning }))
@@ -1698,12 +1704,40 @@ describe('cancelWorkflowExecution', () => {
     await expect(response.json()).resolves.toEqual({
       error: 'Execution cannot be cancelled while completed',
     })
-    expect(mockFinalizePausedCancellationForTerminalRun).toHaveBeenCalledWith('ex-1', 'wf-1')
+    expect(mockFinalizePausedCancellationForTerminalRun).toHaveBeenCalledWith('ex-1', 'wf-1', [
+      'resume-entry-1',
+    ])
     expect(mockRollbackActiveResumeCancellation).not.toHaveBeenCalled()
     expect(mockClearPausedCancellationIntent).not.toHaveBeenCalled()
     expect(mockClearExecutionCancellation).not.toHaveBeenCalled()
   })
 
+  it('does not finalize a claimed resume that cannot be stopped after its parent is terminal', async () => {
+    mockGetActiveResumeCancellationTargets.mockResolvedValue([ACTIVE_RESUME_TARGET])
+    mockGetActiveResumeCancellationTarget.mockResolvedValue(ACTIVE_RESUME_TARGET)
+    dbChainMockFns.limit.mockResolvedValueOnce([
+      {
+        executionDeadlineAt: null,
+        executionOrigin: null,
+        status: 'completed',
+        workspaceId: 'workspace-1',
+      },
+    ])
+
+    const response = await POST(makeRequest(), makeParams())
+
+    expect(response.status).toBe(500)
+    await expect(response.json()).resolves.toEqual({
+      error: 'Failed to reconcile paused execution after cancellation was rejected',
+    })
+    expect(mockMarkExecutionCancelled).toHaveBeenCalledTimes(3)
+    expect(mockMarkExecutionCancelled).toHaveBeenCalledWith('resume-ex-1', {
+      executionDeadlineAt: null,
+    })
+    expect(mockFinalizePausedCancellationForTerminalRun).not.toHaveBeenCalled()
+    expect(mockReleaseExecutionSlot).not.toHaveBeenCalled()
+  })
+
   it('treats a concurrent cancellation as an idempotent success', async () => {
     mockMarkExecutionCancelled.mockResolvedValue({ durablyRecorded: true, reason: 'recorded' })
     dbChainMockFns.limit

apps/sim/lib/execution/cancel-workflow-execution.ts

@@ -316,14 +316,45 @@ async function clearPausedCancellationIntentWithRetry(
 
 async function finalizePausedCancellationForTerminalRunWithRetry(
   executionId: string,
-  workflowId: string
+  workflowId: string,
+  executionDeadlineAt: Date | null,
+  stopSummary: ExecutionStopSummary
 ): Promise<boolean> {
   for (let attempt = 1; attempt <= PAUSED_CANCELLATION_DB_ATTEMPTS; attempt++) {
     try {
-      const finalized = await PauseResumeManager.finalizePausedCancellationForTerminalRun(
+      const activeResumeTargets = await PauseResumeManager.getActiveResumeCancellationTargets(
         executionId,
         workflowId
       )
+      const stoppedResumeEntryIds: string[] = []
+      for (const target of activeResumeTargets) {
+        const stopped = await signalAndRecordActiveResumeStop({
+          workflowId,
+          executionId,
+          executionDeadlineAt,
+          target,
+          summary: stopSummary,
+        })
+        if (!stopped) break
+        stoppedResumeEntryIds.push(target.resumeEntryId)
+      }
+
+      if (stoppedResumeEntryIds.length !== activeResumeTargets.length) {
+        logger.warn('Claimed resume could not be stopped during terminal cleanup', {
+          executionId,
+          attempt,
+        })
+        if (attempt < PAUSED_CANCELLATION_DB_ATTEMPTS) {
+          await sleep(PAUSED_CANCELLATION_DB_RETRY_MS)
+        }
+        continue
+      }
+
+      const finalized = await PauseResumeManager.finalizePausedCancellationForTerminalRun(
+        executionId,
+        workflowId,
+        stoppedResumeEntryIds
+      )
       if (finalized) return true
       logger.warn('Paused cancellation terminal cleanup was rejected', {
         executionId,
@@ -771,9 +802,12 @@ export async function cancelWorkflowExecution({
     }
 
     if (execution.status !== 'running' && execution.status !== 'pending') {
+      const stopSummary = createExecutionStopSummary()
       const pausedCancellationFinalized = await finalizePausedCancellationForTerminalRunWithRetry(
         executionId,
-        workflowId
+        workflowId,
+        execution.executionDeadlineAt,
+        stopSummary
       )
       if (!pausedCancellationFinalized) throwPausedCancellationReconciliationFailed()
 
@@ -975,7 +1009,9 @@ export async function cancelWorkflowExecution({
       } else if (effectivePausedCancellationPath) {
         const pausedCancellationFinalized = await finalizePausedCancellationForTerminalRunWithRetry(
           executionId,
-          workflowId
+          workflowId,
+          execution.executionDeadlineAt,
+          stopSummary
         )
         if (!pausedCancellationFinalized) throwPausedCancellationReconciliationFailed()
       } else {

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

@@ -1288,7 +1288,9 @@ describe('PauseResumeManager paused cancellation after pause release', () => {
     queueTableRows(resumeQueue, [{ id: 'resume-entry-1' }])
 
     await expect(
-      PauseResumeManager.finalizePausedCancellationForTerminalRun('execution-1', 'workflow-1')
+      PauseResumeManager.finalizePausedCancellationForTerminalRun('execution-1', 'workflow-1', [
+        'resume-entry-1',
+      ])
     ).resolves.toBe(true)
 
     expect(dbChainMockFns.set).toHaveBeenNthCalledWith(1, {
@@ -1305,6 +1307,19 @@ describe('PauseResumeManager paused cancellation after pause release', () => {
     expect(mockReleaseExecutionSlot).toHaveBeenCalledWith('resume-entry-1')
   })
 
+  it('does not finalize or release an unconfirmed claimed resume', async () => {
+    queueTableRows(workflowExecutionLogs, [{ status: 'completed' }])
+    queueTableRows(pausedExecutions, [{ id: 'paused-exec-1', status: 'cancelling' }])
+    queueTableRows(resumeQueue, [{ id: 'resume-entry-1' }])
+
+    await expect(
+      PauseResumeManager.finalizePausedCancellationForTerminalRun('execution-1', 'workflow-1', [])
+    ).resolves.toBe(false)
+
+    expect(dbChainMockFns.set).not.toHaveBeenCalled()
+    expect(mockReleaseExecutionSlot).not.toHaveBeenCalled()
+  })
+
   it('restores only cancellation-staged queue entries while the workflow remains active', async () => {
     queueTableRows(workflowExecutionLogs, [{ status: 'running' }])
     queueTableRows(pausedExecutions, [{ id: 'paused-exec-1' }])

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

@@ -2736,10 +2736,12 @@ export class PauseResumeManager {
    * Finalizes only pause and resume state when a non-cancellation terminal
    * transition wins the parent execution race. The parent log is locked and
    * inspected but never mutated, so a late claimed resume cannot revive it.
+   * Every claimed resume must be stopped before its queue row is finalized.
    */
   static async finalizePausedCancellationForTerminalRun(
     executionId: string,
-    workflowId: string
+    workflowId: string,
+    stoppedResumeEntryIds: string[]
   ): Promise<boolean> {
     const now = new Date()
 
@@ -2791,6 +2793,11 @@ export class PauseResumeManager {
         )
         .for('update')
 
+      const stoppedResumeEntryIdSet = new Set(stoppedResumeEntryIds)
+      if (claimedResumeEntries.some((entry) => !stoppedResumeEntryIdSet.has(entry.id))) {
+        return { finalized: false, claimedResumeEntryIds: [] as string[] }
+      }
+
       if (pausedExecution.status !== 'cancelled') {
         await tx
           .update(pausedExecutions)
@@ -2886,7 +2893,18 @@ export class PauseResumeManager {
     executionId: string,
     workflowId: string
   ): Promise<ActiveResumeCancellationTarget | null> {
-    const activeResume = await execDb
+    const activeResumes = await PauseResumeManager.getActiveResumeCancellationTargets(
+      executionId,
+      workflowId
+    )
+    return activeResumes[0] ?? null
+  }
+
+  static async getActiveResumeCancellationTargets(
+    executionId: string,
+    workflowId: string
+  ): Promise<ActiveResumeCancellationTarget[]> {
+    return await execDb
       .select({
         resumeEntryId: resumeQueue.id,
         pausedExecutionId: resumeQueue.pausedExecutionId,
@@ -2904,10 +2922,6 @@ export class PauseResumeManager {
         )
       )
       .orderBy(desc(resumeQueue.claimedAt))
-      .limit(1)
-      .then((rows) => rows[0])
-
-    return activeResume ?? null
   }
 
   static async rollbackActiveResumeCancellation(