fix(workflows): confirm every active resume stop

Author: j15z

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

@@ -803,6 +803,7 @@ describe('cancelWorkflowExecution', () => {
     mockMarkExecutionCancelled
       .mockResolvedValueOnce({ durablyRecorded: false, reason: 'redis_unavailable' })
       .mockResolvedValueOnce({ durablyRecorded: true, reason: 'recorded' })
+      .mockResolvedValueOnce({ durablyRecorded: true, reason: 'recorded' })
     mockCompletePausedCancellation.mockResolvedValueOnce(true)
 
     const response = await POST(makeRequest(), makeParams())
@@ -822,7 +823,10 @@ describe('cancelWorkflowExecution', () => {
     expect(mockMarkExecutionCancelled).toHaveBeenNthCalledWith(1, 'resume-ex-1', {
       executionDeadlineAt: null,
     })
-    expect(mockMarkExecutionCancelled).toHaveBeenNthCalledWith(2, 'resume-ex-2', {
+    expect(mockMarkExecutionCancelled).toHaveBeenNthCalledWith(2, 'resume-ex-1', {
+      executionDeadlineAt: null,
+    })
+    expect(mockMarkExecutionCancelled).toHaveBeenNthCalledWith(3, 'resume-ex-2', {
       executionDeadlineAt: null,
     })
     expect(mockCompletePausedCancellation).toHaveBeenCalledWith('ex-1', 'wf-1')
@@ -842,6 +846,7 @@ describe('cancelWorkflowExecution', () => {
       .mockResolvedValueOnce({ durablyRecorded: false, reason: 'redis_unavailable' })
       .mockResolvedValueOnce({ durablyRecorded: false, reason: 'redis_unavailable' })
       .mockResolvedValueOnce({ durablyRecorded: true, reason: 'recorded' })
+      .mockResolvedValueOnce({ durablyRecorded: true, reason: 'recorded' })
     mockCompletePausedCancellation.mockResolvedValueOnce(true)
 
     const response = await POST(makeRequest(), makeParams())
@@ -864,12 +869,53 @@ describe('cancelWorkflowExecution', () => {
     expect(mockMarkExecutionCancelled).toHaveBeenNthCalledWith(2, 'resume-ex-1', {
       executionDeadlineAt: null,
     })
-    expect(mockMarkExecutionCancelled).toHaveBeenNthCalledWith(3, 'resume-ex-2', {
+    expect(mockMarkExecutionCancelled).toHaveBeenNthCalledWith(3, 'resume-ex-1', {
+      executionDeadlineAt: null,
+    })
+    expect(mockMarkExecutionCancelled).toHaveBeenNthCalledWith(4, 'resume-ex-2', {
       executionDeadlineAt: null,
     })
     expect(mockCompletePausedCancellation).toHaveBeenCalledWith('ex-1', 'wf-1')
   })
 
+  it('does not treat replacement queue cancellation as confirmation of the original stop', async () => {
+    mockStagePausedCancellation
+      .mockResolvedValueOnce({ kind: 'active_resume', target: ACTIVE_RESUME_TARGET })
+      .mockResolvedValueOnce({
+        kind: 'active_resume',
+        target: REPLACEMENT_ACTIVE_RESUME_TARGET,
+      })
+    mockGetActiveResumeCancellationTarget
+      .mockResolvedValueOnce(REPLACEMENT_ACTIVE_RESUME_TARGET)
+      .mockResolvedValueOnce(REPLACEMENT_ACTIVE_RESUME_TARGET)
+    mockRollbackActiveResumeCancellation.mockResolvedValueOnce(false)
+    mockCancelByExecution.mockResolvedValue(1)
+    mockMarkExecutionCancelled
+      .mockResolvedValueOnce({ durablyRecorded: false, reason: 'redis_unavailable' })
+      .mockResolvedValueOnce({ durablyRecorded: false, reason: 'redis_unavailable' })
+      .mockResolvedValueOnce({ durablyRecorded: true, reason: 'recorded' })
+
+    const response = await POST(makeRequest(), makeParams())
+
+    expect(response.status).toBe(200)
+    await expect(response.json()).resolves.toMatchObject({
+      success: false,
+      pausedCancelled: false,
+      reason: 'active_resume_signal_failed',
+    })
+    expect(mockMarkExecutionCancelled).toHaveBeenNthCalledWith(1, 'resume-ex-1', {
+      executionDeadlineAt: null,
+    })
+    expect(mockMarkExecutionCancelled).toHaveBeenNthCalledWith(2, 'resume-ex-1', {
+      executionDeadlineAt: null,
+    })
+    expect(mockMarkExecutionCancelled).toHaveBeenNthCalledWith(3, 'resume-ex-2', {
+      executionDeadlineAt: null,
+    })
+    expect(mockWriteTerminalEvent).not.toHaveBeenCalled()
+    expect(mockCompletePausedCancellation).not.toHaveBeenCalled()
+  })
+
   it('returns success when a paused HITL execution is cancelled directly in the database', async () => {
     mockStagePausedCancellation.mockResolvedValue({ kind: 'idle' })
     mockCompletePausedCancellation.mockResolvedValue(true)

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

@@ -181,11 +181,14 @@ async function didActiveResumeStop(
   target: ActiveResumeCancellationTarget,
   signal: ExecutionStopSignalResult
 ): Promise<boolean> {
-  if (signal.accepted) return true
+  if (signal.cancellation.durablyRecorded || signal.locallyAborted) return true
   const currentTarget = await PauseResumeManager.getActiveResumeCancellationTarget(
     executionId,
     workflowId
   )
+  if (currentTarget?.resumeEntryId === target.resumeEntryId && signal.queueJobsCancelled > 0) {
+    return true
+  }
   if (currentTarget && currentTarget.resumeEntryId !== target.resumeEntryId) {
     logger.warn('A replacement resume became active while cancellation was staged', {
       executionId,
@@ -707,6 +710,7 @@ export async function cancelWorkflowExecution({
       pausedCancellationStage.kind === 'active_resume' ? pausedCancellationStage.target : null
     let activeResumeEntryId = activeResumeTarget?.resumeEntryId ?? null
     let activeResumeSignalAccepted = false
+    const activeResumeTargetsNeedingStopConfirmation: ActiveResumeCancellationTarget[] = []
 
     if (activeResumeTarget && !isWorkflowGroupExecution) {
       activeResumeSignalAccepted = await signalAndRecordActiveResumeStop({
@@ -728,6 +732,7 @@ export async function cancelWorkflowExecution({
           await clearStopSignalMarkers(stopSummary)
           return activeResumeSignalFailureResult(executionId, stopSummary)
         }
+        activeResumeTargetsNeedingStopConfirmation.push(activeResumeTarget)
       }
     } else if (!effectivePausedCancellationPath && !isWorkflowGroupExecution) {
       const signal = await signalExecutionStop({
@@ -778,6 +783,7 @@ export async function cancelWorkflowExecution({
               await clearStopSignalMarkers(stopSummary)
               return activeResumeSignalFailureResult(executionId, stopSummary)
             }
+            activeResumeTargetsNeedingStopConfirmation.push(activeResumeTarget)
           }
         } else if (!effectivePausedCancellationPath) {
           return {
@@ -934,6 +940,19 @@ export async function cancelWorkflowExecution({
 
     let pauseReconciliationFailed = false
     let activeResumeSignalFailed = false
+    for (const target of activeResumeTargetsNeedingStopConfirmation) {
+      const stopConfirmed = await signalAndRecordActiveResumeStop({
+        workflowId,
+        executionId,
+        executionDeadlineAt: execution.executionDeadlineAt,
+        target,
+        summary: stopSummary,
+      })
+      if (target.resumeEntryId === activeResumeEntryId) {
+        activeResumeSignalAccepted = stopConfirmed
+      }
+      activeResumeSignalFailed = activeResumeSignalFailed || !stopConfirmed
+    }
     if (isWorkflowGroupExecution) {
       if (activeResumeTarget) {
         activeResumeSignalAccepted = await signalAndRecordActiveResumeStop({
@@ -975,9 +994,7 @@ export async function cancelWorkflowExecution({
               summary: stopSummary,
             })
           }
-          activeResumeSignalFailed = !activeResumeSignalAccepted
-        } else {
-          activeResumeSignalFailed = false
+          activeResumeSignalFailed = activeResumeSignalFailed || !activeResumeSignalAccepted
         }
       }
     } catch (error) {