fix(executor): preserve stop-after across pause and resume (#8776)

* fix(executor): preserve stop-after across pause and resume

* fix(executor): retain stop targets while sibling branches pause

Author: waleedlatif1

.github/scripts/http-e2e.sh

@@ -122,13 +122,15 @@ case "$group" in
   # Self-hosted: hosted billing admits a run only through a Redis usage reservation.
   stop-after)
     export NEXT_PUBLIC_FORCE_HOSTED=false
+    export CRON_SECRET=stop-after-e2e-local-cron-secret
     export INTERNAL_API_SECRET=stop-after-http-ci-local-secret-at-least-32-characters
     export DB_TX_TRIPWIRE=throw
     export NEXT_PUBLIC_CHAT_DISABLED=true
     start_app stop-after 3018 workflow record-http-status
     STOP_AFTER_E2E_BASE_URL="$NEXT_PUBLIC_APP_URL" \
     STOP_AFTER_E2E_DATABASE_URL="$DATABASE_URL" \
     STOP_AFTER_E2E_REPORT_PATH="$report_dir/stop-after-http-report.json" \
+    STOP_AFTER_E2E_CRON_SECRET="$CRON_SECRET" \
       bun run test:workflow-stop-after:e2e
     ;;
 

apps/sim/executor/execution/engine.test.ts

@@ -1258,6 +1258,53 @@ describe('ExecutionEngine', () => {
       expect(executed).toEqual(['start', 'condition', 'stop'])
     })
 
+    it.each([
+      ['approval', 'stop'],
+      ['stop', 'approval'],
+    ])(
+      'completes without retaining a sibling pause when %s finishes before %s',
+      async (first, second) => {
+        const { engine, executed } = buildRun(
+          {
+            version: '1',
+            blocks: [
+              block('start', BlockType.STARTER),
+              block('approval'),
+              block('stop'),
+              block('after'),
+            ],
+            connections: [
+              { source: 'start', target: first },
+              { source: 'start', target: second },
+              { source: 'stop', target: 'after' },
+            ],
+            loops: {},
+            parallels: {},
+          },
+          'stop',
+          {
+            approval: {
+              _pauseMetadata: {
+                contextId: 'approval',
+                blockId: 'approval',
+                response: {},
+                timestamp: new Date().toISOString(),
+                pauseKind: 'human',
+              },
+            },
+          }
+        )
+
+        const result = await engine.run('start')
+
+        expect(result.success).toBe(true)
+        expect(result.status).not.toBe('paused')
+        expect(result.pausePoints).toBeUndefined()
+        expect(result.snapshotSeed).toBeUndefined()
+        expect(executed).toEqual(['start', first, second])
+      }
+    )
+
     it('runs a join reached through one taken and one skipped branch', async () => {
       const { engine, executed } = buildRun(
         {

apps/sim/executor/execution/engine.ts

@@ -128,8 +128,22 @@ export class ExecutionEngine {
         throw this.executionError
       }
 
+      if (this.stopBlockReached) {
+        this.pausedBlocks.clear()
+        this.context.metadata.pausePoints = []
+        this.context.metadata.status = 'completed'
+        this.context.metadata.stopAfterBlockReached = true
+      }
+
       /** A pause keeps a run whose stop block can still run; one proven unreachable fails. */
-      if (!this.cancelledFlag && (this.stopBlockUnreachable || this.pausedBlocks.size === 0)) {
+      const waitingForPausedBranch =
+        this.context.metadata.resumeFromSnapshot === true &&
+        this.context.metadata.resumeHasPendingPauses === true &&
+        !this.responseOutputLocked
+      if (
+        !this.cancelledFlag &&
+        (this.stopBlockUnreachable || (this.pausedBlocks.size === 0 && !waitingForPausedBranch))
+      ) {
         this.assertStopBlockReached()
       }
 
@@ -293,6 +307,19 @@ export class ExecutionEngine {
   }
 
   private initializeQueue(triggerBlockId?: string): void {
+    const stopBlockId = this.context.stopAfterBlockId
+    if (
+      this.context.metadata.resumeFromSnapshot &&
+      this.context.metadata.stopAfterBlockReached &&
+      stopBlockId
+    ) {
+      this.stopBlockReached = true
+      this.stoppedEarlyFlag = true
+      const state = this.context.blockStates.get(stopBlockId)
+      if (state) this.setFinalOutput(stopBlockId, state.output)
+      return
+    }
+
     if (this.context.runFromBlockContext) {
       const { startBlockId } = this.context.runFromBlockContext
       this.execLogger.info('Initializing queue for run-from-block mode', {

apps/sim/executor/execution/snapshot-serializer.ts

@@ -307,6 +307,7 @@ function buildExecutionSnapshot(
         : undefined,
     /** Preserve the run-level agent-events opt-in across HITL pause/resume. */
     agentEvents: metadataFromContext?.agentEvents === true ? true : undefined,
+    stopAfterBlockId: metadataFromContext?.stopAfterBlockId,
   }
 
   const snapshot = new ExecutionSnapshot(

apps/sim/executor/execution/types.ts

@@ -59,6 +59,12 @@ export interface ExecutionMetadata {
   pendingBlocks?: string[]
   resumeFromSnapshot?: boolean
   resumeTerminalNoop?: boolean
+  /** Other durable pauses remain after the current context resumes. */
+  resumeHasPendingPauses?: boolean
+  /** Original workflow block ID, before loop/parallel sentinel resolution. */
+  stopAfterBlockId?: string
+  /** A resumed pause completed the stop target before the engine starts. */
+  stopAfterBlockReached?: boolean
   credentialAccountUserId?: string
   workflowStateOverride?: {
     blocks: Record<string, any>

apps/sim/executor/types.ts

@@ -383,6 +383,10 @@ interface ExecutionMetadata {
   useDraftState?: boolean
   resumeFromSnapshot?: boolean
   resumeTerminalNoop?: boolean
+  /** Other durable pauses remain after the current context resumes. */
+  resumeHasPendingPauses?: boolean
+  /** Stop-after completion retires outstanding pause contexts when settling a resume. */
+  stopAfterBlockReached?: boolean
   executionMode?: 'sync' | 'stream' | 'async'
   /**
    * Run-level agent-events opt-in (see the snapshot ExecutionMetadata).

apps/sim/lib/workflows/executor/execution-core.ts

@@ -631,7 +631,6 @@ async function executeWorkflowCoreImpl(
     abortSignal,
     includeFileBase64,
     base64MaxBytes,
-    stopAfterBlockId,
     runFromBlock,
     resumeDeploymentVersionId,
     draftState,
@@ -647,6 +646,8 @@ async function executeWorkflowCoreImpl(
     throw new Error(`Execution metadata missing workspaceId for workflow ${workflowId}`)
   }
   const resumeFromSnapshot = metadata.resumeFromSnapshot === true
+  const stopAfterBlockId =
+    options.stopAfterBlockId ?? (resumeFromSnapshot ? metadata.stopAfterBlockId : undefined)
   if (!resumeFromSnapshot && resumeDeploymentVersionId !== undefined) {
     throw new Error('Deployment version authority can only be supplied for a resumed execution')
   }
@@ -1236,7 +1237,14 @@ async function executeWorkflowCoreImpl(
       ...(workflowInputResolvedSecretTraceProvenance
         ? { workflowInputResolvedSecretTraceProvenance }
         : {}),
-      metadata,
+      metadata: {
+        ...metadata,
+        stopAfterBlockId,
+        stopAfterBlockReached:
+          resumeFromSnapshot &&
+          stopAfterBlockId !== undefined &&
+          snapshot.state?.completedPauseContexts?.includes(stopAfterBlockId) === true,
+      },
       startRunMetadata,
       abortSignal,
       includeFileBase64,

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

@@ -993,6 +993,7 @@ export class PauseResumeManager {
             pausedExecutionId: pausedExecution.id,
             parentExecutionId: pausedExecution.executionId,
             contextId,
+            stopAfterBlockReached: result.metadata?.stopAfterBlockReached === true,
           })
         }
       }
@@ -1478,6 +1479,9 @@ export class PauseResumeManager {
       isClientSession: baseSnapshot.metadata.isClientSession,
       resumeFromSnapshot: true,
       resumeTerminalNoop: stateCopy?.resumeTerminalNoop === true,
+      resumeHasPendingPauses: Object.entries(pausePoints).some(
+        ([id, point]) => id !== contextId && point.resumeStatus !== 'resumed'
+      ),
     }
 
     const resumeSnapshot = new ExecutionSnapshot(
@@ -2095,6 +2099,7 @@ export class PauseResumeManager {
     pausedExecutionId?: string
     parentExecutionId?: string
     contextId?: string
+    stopAfterBlockReached?: boolean
   }): Promise<void> {
     const { resumeEntryId, pausedExecutionId, parentExecutionId, contextId } = args
     const now = new Date()
@@ -2178,6 +2183,45 @@ export class PauseResumeManager {
 
       if (!pausedExecution || !contextId) return
 
+      if (args.stopAfterBlockReached) {
+        await tx
+          .update(pausedExecutions)
+          .set({
+            pausePoints: sql`(
+              SELECT COALESCE(jsonb_object_agg(point.key, point.value), '{}'::jsonb)
+              FROM jsonb_each(jsonb_set(
+                ${updatePausePointResumeStateSql(contextId, 'resumed')},
+                ARRAY[${contextId}, 'resumedAt'],
+                ${JSON.stringify(now.toISOString())}::jsonb
+              )) AS point
+              WHERE point.value->>'resumeStatus' = 'resumed'
+            )`,
+            totalPauseCount: sql`${pausedExecutions.resumedCount} + 1`,
+            resumedCount: sql`${pausedExecutions.resumedCount} + 1`,
+            status: 'fully_resumed',
+            nextResumeAt: null,
+            automaticResumeRetryCount: 0,
+            metadata: sql`${pausedExecutions.metadata} - 'automaticResumeWaiting'`,
+            updatedAt: now,
+          })
+          .where(eq(pausedExecutions.id, targetPausedExecutionId))
+        await tx
+          .update(resumeQueue)
+          .set({
+            status: 'failed',
+            completedAt: now,
+            failureReason: 'Execution stopped after its target block',
+          })
+          .where(
+            and(
+              eq(resumeQueue.parentExecutionId, targetParentExecutionId),
+              eq(resumeQueue.pausedExecutionId, targetPausedExecutionId),
+              eq(resumeQueue.status, 'pending')
+            )
+          )
+        return
+      }
+
       await tx
         .update(pausedExecutions)
         .set({