fix hitl cases

Author: icecrasher321

apps/sim/executor/dag/construction/edges.test.ts

@@ -220,6 +220,36 @@ describe('EdgeConstructor', () => {
       })
       expect(dag.nodes.get(sourceEndId)!.incomingEdges).not.toContain(sourceStartId)
     })
+
+    it('wires terminal top-level loop start exit through its own sentinel end', () => {
+      const loopId = 'loop-1'
+      const taskId = 'task-1'
+      const loopStartId = `loop-${loopId}-sentinel-start`
+      const loopEndId = `loop-${loopId}-sentinel-end`
+      const dag = createMockDAG([loopStartId, loopEndId, taskId])
+      dag.loopConfigs.set(loopId, { id: loopId, nodes: [taskId], iterations: 1 })
+      const workflow = createMockWorkflow(
+        [createMockBlock(loopId, 'loop'), createMockBlock(taskId)],
+        []
+      )
+
+      edgeConstructor.execute(
+        workflow,
+        dag,
+        new Set(),
+        new Set([taskId]),
+        new Set([taskId]),
+        new Map()
+      )
+
+      const loopStartTargets = Array.from(dag.nodes.get(loopStartId)!.outgoingEdges.values())
+      expect(loopStartTargets).toContainEqual({
+        target: loopEndId,
+        sourceHandle: 'loop_exit',
+        targetHandle: undefined,
+      })
+      expect(dag.nodes.get(loopEndId)!.incomingEdges).not.toContain(loopStartId)
+    })
   })
 
   describe('Condition block edge wiring', () => {
@@ -530,8 +560,9 @@ describe('EdgeConstructor', () => {
 
       // Sentinel start should have edge to node in loop (it's a start node - no incoming from loop)
       const sentinelStartNode = dag.nodes.get(sentinelStartId)!
-      expect(sentinelStartNode.outgoingEdges.size).toBe(1)
-      const startEdge = Array.from(sentinelStartNode.outgoingEdges.values())[0]
+      const startEdge = Array.from(sentinelStartNode.outgoingEdges.values()).find(
+        (edge) => edge.target === nodeInLoopId
+      )
       expect(startEdge.target).toBe(nodeInLoopId)
 
       // Node in loop should have edge to sentinel end (it's a terminal node - no outgoing to loop)
@@ -589,7 +620,10 @@ describe('EdgeConstructor', () => {
 
       // Sentinel start should have edges to both nodes (both are start nodes)
       const sentinelStartNode = dag.nodes.get(sentinelStartId)!
-      expect(sentinelStartNode.outgoingEdges.size).toBe(2)
+      const bodyStartEdges = Array.from(sentinelStartNode.outgoingEdges.values()).filter(
+        (edge) => edge.target === node1Id || edge.target === node2Id
+      )
+      expect(bodyStartEdges).toHaveLength(2)
 
       // Both nodes should have edges to sentinel end (both are terminal nodes)
       const node1 = dag.nodes.get(node1Id)!
@@ -791,7 +825,7 @@ describe('EdgeConstructor', () => {
 
         const loop1StartNode = dag.nodes.get(loop1SentinelStart)!
         const earlyExitEdges = Array.from(loop1StartNode.outgoingEdges.values()).filter(
-          (e) => e.target === loop2SentinelStart && e.sourceHandle === 'loop_exit'
+          (e) => e.target === loop1SentinelEnd && e.sourceHandle === 'loop_exit'
         )
         expect(earlyExitEdges.length).toBeGreaterThan(0)
       })
@@ -953,7 +987,7 @@ describe('EdgeConstructor', () => {
 
         const loopStartNode = dag.nodes.get(loopSentinelStart)!
         const earlyExitEdges = Array.from(loopStartNode.outgoingEdges.values()).filter(
-          (e) => e.target === parallelSentinelStart && e.sourceHandle === 'loop_exit'
+          (e) => e.target === loopSentinelEnd && e.sourceHandle === 'loop_exit'
         )
         expect(earlyExitEdges.length).toBeGreaterThan(0)
       })

apps/sim/executor/dag/construction/edges.ts

@@ -236,16 +236,15 @@ export class EdgeConstructor {
         continue
       }
 
-      let loopSentinelStartId: string | undefined
-
       if (sourceIsLoopBlock) {
         const sentinelEndId = buildSentinelEndId(originalSource)
-        loopSentinelStartId = buildSentinelStartId(originalSource)
+        const loopSentinelStartId = buildSentinelStartId(originalSource)
         if (!dag.nodes.has(sentinelEndId) || !dag.nodes.has(loopSentinelStartId)) {
           continue
         }
         source = sentinelEndId
         sourceHandle = EDGE.LOOP_EXIT
+        this.addSubflowStartExitBypass(dag, originalSource)
       }
 
       if (targetIsLoopBlock) {
@@ -286,11 +285,6 @@ export class EdgeConstructor {
         continue
       }
 
-      const sourceLoopNodes = dag.loopConfigs.get(originalSource)?.nodes
-      if (loopSentinelStartId && !sourceLoopNodes?.includes(originalTarget)) {
-        this.addEdge(dag, loopSentinelStartId, target, EDGE.LOOP_EXIT, targetHandle)
-      }
-
       if (!this.isEdgeReachable(source, target, reachableBlocks, dag)) {
         continue
       }
@@ -326,6 +320,8 @@ export class EdgeConstructor {
         continue
       }
 
+      this.addSubflowStartExitBypass(dag, loopId)
+
       const { startNodes, terminalNodes } = this.findLoopBoundaryNodes(nodes, dag)
 
       for (const startNodeId of startNodes) {
@@ -367,6 +363,8 @@ export class EdgeConstructor {
         continue
       }
 
+      this.addSubflowStartExitBypass(dag, parallelId)
+
       const { entryNodes, terminalNodes } = this.findParallelBoundaryNodes(nodes, dag)
 
       for (const entryNodeId of entryNodes) {

apps/sim/executor/execution/edge-manager.ts

@@ -141,6 +141,10 @@ export class EdgeManager {
     this.nodesWithActivatedEdge = new Set(activatedNodeIds ?? [])
   }
 
+  markNodeWithActivatedEdge(nodeId: string): void {
+    this.nodesWithActivatedEdge.add(nodeId)
+  }
+
   /**
    * Clear deactivated edges for a set of nodes (used when restoring loop state for next iteration).
    *

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

@@ -9,6 +9,7 @@ vi.mock('@/lib/execution/cancellation', () => ({
 }))
 
 import { isExecutionCancelled, isRedisCancellationEnabled } from '@/lib/execution/cancellation'
+import { EDGE } from '@/executor/constants'
 import type { DAG, DAGNode } from '@/executor/dag/builder'
 import type { EdgeManager } from '@/executor/execution/edge-manager'
 import type { NodeExecutionOrchestrator } from '@/executor/orchestrators/node'
@@ -89,6 +90,9 @@ function createMockEdgeManager(
     restoreIncomingEdge: vi.fn(),
     clearDeactivatedEdges: vi.fn(),
     clearDeactivatedEdgesForNodes: vi.fn(),
+    getDeactivatedEdges: vi.fn(() => []),
+    getNodesWithActivatedEdge: vi.fn(() => []),
+    markNodeWithActivatedEdge: vi.fn(),
   } as unknown as MockEdgeManager
 }
 
@@ -130,7 +134,9 @@ describe('ExecutionEngine', () => {
       endNode.incomingEdges.add('start')
 
       const dag = createMockDAG([startNode, endNode])
-      const context = createMockContext()
+      const context = createMockContext({
+        decisions: { router: new Map(), condition: new Map() },
+      })
       const edgeManager = createMockEdgeManager((node) => {
         if (node.id === 'start') return ['end']
         return []
@@ -147,7 +153,9 @@ describe('ExecutionEngine', () => {
     it('should mark execution as successful when completed without cancellation', async () => {
       const startNode = createMockNode('start', 'starter')
       const dag = createMockDAG([startNode])
-      const context = createMockContext()
+      const context = createMockContext({
+        decisions: { router: new Map(), condition: new Map() },
+      })
       const edgeManager = createMockEdgeManager()
       const nodeOrchestrator = createMockNodeOrchestrator()
 
@@ -173,12 +181,36 @@ describe('ExecutionEngine', () => {
       const nodeOrchestrator = createMockNodeOrchestrator()
 
       const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
-      const result = await engine.run()
+      const result = await engine.run('hitl')
 
       expect(result.success).toBe(true)
       expect(nodeOrchestrator.executionCount).toBe(0)
     })
 
+    it('marks resumed pause edge targets as activated before readiness checks', async () => {
+      const targetNode = createMockNode('join', 'function')
+      targetNode.incomingEdges.add('pause-block')
+      targetNode.incomingEdges.add('condition-block')
+      const dag = createMockDAG([targetNode])
+      const context = createMockContext({
+        metadata: {
+          executionId: 'test-execution',
+          startTime: new Date().toISOString(),
+          pendingBlocks: [],
+          remainingEdges: [{ source: 'pause-block', target: 'join' }],
+        } as any,
+      })
+      const edgeManager = createMockEdgeManager(() => [])
+      vi.mocked(edgeManager.isNodeReady).mockReturnValue(false)
+      const nodeOrchestrator = createMockNodeOrchestrator()
+
+      const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
+      await engine.run()
+
+      expect(edgeManager.markNodeWithActivatedEdge).toHaveBeenCalledWith('join')
+      expect(nodeOrchestrator.executeNode).not.toHaveBeenCalled()
+    })
+
     it('should execute all nodes in a multi-node workflow', async () => {
       const nodes = [
         createMockNode('start', 'starter'),
@@ -207,6 +239,73 @@ describe('ExecutionEngine', () => {
       expect(result.success).toBe(true)
       expect(nodeOrchestrator.executionCount).toBe(4)
     })
+
+    it('records paused block completion before returning paused result', async () => {
+      const node = createMockNode('hitl', 'function')
+      const dag = createMockDAG([node])
+      const context = createMockContext({
+        decisions: { router: new Map(), condition: new Map() },
+      })
+      const edgeManager = createMockEdgeManager()
+      const nodeOrchestrator = createMockNodeOrchestrator()
+      const pauseOutput = {
+        response: { status: 'paused' },
+        _pauseMetadata: {
+          contextId: 'pause-1',
+          blockId: 'hitl',
+          response: { status: 'paused' },
+          timestamp: new Date().toISOString(),
+          pauseKind: 'hitl',
+        },
+      }
+      vi.mocked(nodeOrchestrator.executeNode).mockResolvedValue({
+        nodeId: 'hitl',
+        output: pauseOutput,
+        isFinalOutput: false,
+      })
+
+      const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
+      const result = await engine.run('hitl')
+
+      expect(result.status).toBe('paused')
+      expect(nodeOrchestrator.handleNodeCompletion).toHaveBeenCalledWith(
+        context,
+        'hitl',
+        pauseOutput
+      )
+    })
+
+    it('does not stop run-until execution on parallel batch continuation', async () => {
+      const parallelEnd = createMockNode('parallel-end', 'parallel')
+      const nextNode = createMockNode('next', 'function')
+      parallelEnd.outgoingEdges.set('continue', {
+        target: 'next',
+        sourceHandle: EDGE.PARALLEL_CONTINUE,
+      })
+      nextNode.incomingEdges.add('parallel-end')
+      const dag = createMockDAG([parallelEnd, nextNode])
+      const context = createMockContext({
+        stopAfterBlockId: 'parallel-end',
+        decisions: { router: new Map(), condition: new Map() },
+      })
+      const edgeManager = createMockEdgeManager((node) =>
+        node.id === 'parallel-end' ? ['next'] : []
+      )
+      const nodeOrchestrator = createMockNodeOrchestrator()
+      vi.mocked(nodeOrchestrator.executeNode).mockImplementation(async (_ctx, nodeId) => ({
+        nodeId,
+        output:
+          nodeId === 'parallel-end'
+            ? { selectedRoute: EDGE.PARALLEL_CONTINUE }
+            : { result: 'done' },
+        isFinalOutput: nodeId === 'next',
+      }))
+
+      const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
+      await engine.run('parallel-end')
+
+      expect(nodeOrchestrator.executeNode).toHaveBeenCalledWith(context, 'next')
+    })
   })
 
   describe('Cancellation via AbortSignal', () => {

apps/sim/executor/execution/engine.ts

@@ -1,7 +1,7 @@
 import { createLogger, type Logger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { isExecutionCancelled, isRedisCancellationEnabled } from '@/lib/execution/cancellation'
-import { BlockType } from '@/executor/constants'
+import { BlockType, EDGE } from '@/executor/constants'
 import type { DAG } from '@/executor/dag/builder'
 import type { EdgeManager } from '@/executor/execution/edge-manager'
 import { serializePauseSnapshot } from '@/executor/execution/snapshot-serializer'
@@ -303,6 +303,9 @@ export class ExecutionEngine {
         if (targetNode) {
           const hadEdge = targetNode.incomingEdges.has(edge.source)
           targetNode.incomingEdges.delete(edge.source)
+          if (hadEdge) {
+            this.edgeManager.markNodeWithActivatedEdge(targetNode.id)
+          }
 
           if (this.edgeManager.isNodeReady(targetNode)) {
             this.execLogger.info('Node became ready after edge removal', { nodeId: targetNode.id })
@@ -412,6 +415,8 @@ export class ExecutionEngine {
     }
 
     if (output._pauseMetadata) {
+      await this.nodeOrchestrator.handleNodeCompletion(this.context, nodeId, output)
+
       const pauseMetadata = output._pauseMetadata
       this.pausedBlocks.set(pauseMetadata.contextId, pauseMetadata)
       this.context.metadata.status = 'paused'
@@ -439,8 +444,9 @@ export class ExecutionEngine {
     if (this.context.stopAfterBlockId === nodeId) {
       // For loop/parallel sentinels, only stop if the subflow has fully exited (all iterations done)
       // shouldContinue: true means more iterations, shouldExit: true means loop is done
-      const shouldContinueLoop = output.shouldContinue === true
-      if (!shouldContinueLoop) {
+      const shouldContinue =
+        output.shouldContinue === true || output.selectedRoute === EDGE.PARALLEL_CONTINUE
+      if (!shouldContinue) {
         this.execLogger.info('Stopping execution after target block', { nodeId })
         this.stoppedEarlyFlag = true
         return

apps/sim/executor/orchestrators/loop.test.ts

@@ -81,4 +81,47 @@ describe('LoopOrchestrator', () => {
     expect(parallelStart.incomingEdges.has(loopStartId)).toBe(true)
     expect(parallelStart.incomingEdges.has(parallelEndId)).toBe(false)
   })
+
+  it('resolves forEach collections with the loop start sentinel scope', async () => {
+    const loopId = 'loop-1'
+    const dag: DAG = {
+      nodes: new Map(),
+      loopConfigs: new Map([
+        [
+          loopId,
+          {
+            id: loopId,
+            nodes: ['task-1'],
+            loopType: 'forEach',
+            forEachItems: '<Producer.items>',
+          },
+        ],
+      ]),
+      parallelConfigs: new Map(),
+    }
+    const resolver = {
+      resolveSingleReference: vi.fn().mockResolvedValue(['item-1']),
+    }
+    const orchestrator = new LoopOrchestrator(dag, createState(), resolver as any, {}, {
+      clearDeactivatedEdgesForNodes: vi.fn(),
+    } as unknown as EdgeManager)
+    const ctx = {
+      workflowId: 'workflow-1',
+      workspaceId: 'workspace-1',
+      executionId: 'execution-1',
+      userId: 'user-1',
+      loopExecutions: new Map(),
+      blockLogs: [],
+      metadata: {},
+    }
+
+    const scope = await orchestrator.initializeLoopScope(ctx as any, loopId)
+
+    expect(resolver.resolveSingleReference).toHaveBeenCalledWith(
+      expect.any(Object),
+      'loop-loop-1-sentinel-start',
+      '<Producer.items>'
+    )
+    expect(scope.maxIterations).toBe(1)
+  })
 })

apps/sim/executor/orchestrators/loop.ts

@@ -139,7 +139,12 @@ export class LoopOrchestrator {
         }
         let items: any[]
         try {
-          items = await resolveArrayInputAsync(ctx, loopConfig.forEachItems, this.resolver)
+          items = await resolveArrayInputAsync(
+            ctx,
+            loopConfig.forEachItems,
+            this.resolver,
+            buildSentinelStartId(loopId)
+          )
         } catch (error) {
           const errorMessage = `ForEach loop resolution failed: ${toError(error).message}`
           logger.error(errorMessage, { loopId, forEachItems: loopConfig.forEachItems })