improvement(executor): clean up nested subflow implementation

- Fix wireSentinelEdges to use LOOP_EXIT handle for nested loop terminals
- Extract buildExecutionPipeline to deduplicate orchestrator wiring
- Replace two-phase init with constructor injection for Loop/ParallelOrchestrator
- Remove dead code: shouldExecuteLoopNode, resolveForEachItems, isLoopNode, isParallelNode, isSubflowBlockType
- Deduplicate currentItem resolution in ParallelResolver via resolveCurrentItem
- Type getDistributionItems param as SerializedParallel instead of any
- Demote verbose per-reference logger.info to logger.debug in evaluateWhileCondition
- Add loop-in-parallel wiring test in edges.test.ts

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: waleedlatif1

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/terminal/utils.ts

@@ -125,14 +125,6 @@ export function isEventFromEditableElement(e: KeyboardEvent): boolean {
   return false
 }
 
-/**
- * Checks if a block type is a subflow (loop or parallel)
- */
-export function isSubflowBlockType(blockType: string): boolean {
-  const lower = blockType?.toLowerCase() || ''
-  return lower === 'loop' || lower === 'parallel'
-}
-
 /**
  * Node type for the tree structure
  */

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

@@ -1574,5 +1574,123 @@ describe('EdgeConstructor', () => {
         Array.from(funcNode.outgoingEdges.values()).some((e) => e.target === parallelSentinelEnd)
       ).toBe(true)
     })
+
+    it('should wire loop-in-parallel with correct exit handles', () => {
+      const outerParallelId = 'outer-parallel'
+      const innerLoopId = 'inner-loop'
+      const functionId = 'func-1'
+
+      const outerSentinelStart = `parallel-${outerParallelId}-sentinel-start`
+      const outerSentinelEnd = `parallel-${outerParallelId}-sentinel-end`
+      const innerSentinelStart = `loop-${innerLoopId}-sentinel-start`
+      const innerSentinelEnd = `loop-${innerLoopId}-sentinel-end`
+
+      const dag = createMockDAG([
+        outerSentinelStart,
+        outerSentinelEnd,
+        innerSentinelStart,
+        innerSentinelEnd,
+        functionId,
+      ])
+
+      dag.nodes.get(outerSentinelStart)!.metadata = {
+        isSentinel: true,
+        isParallelSentinel: true,
+        sentinelType: 'start',
+        parallelId: outerParallelId,
+      }
+      dag.nodes.get(outerSentinelEnd)!.metadata = {
+        isSentinel: true,
+        isParallelSentinel: true,
+        sentinelType: 'end',
+        parallelId: outerParallelId,
+      }
+      dag.nodes.get(innerSentinelStart)!.metadata = {
+        isSentinel: true,
+        sentinelType: 'start',
+        loopId: innerLoopId,
+      }
+      dag.nodes.get(innerSentinelEnd)!.metadata = {
+        isSentinel: true,
+        sentinelType: 'end',
+        loopId: innerLoopId,
+      }
+
+      const innerLoop: SerializedLoop = {
+        id: innerLoopId,
+        nodes: [functionId],
+        iterations: 3,
+        loopType: 'for',
+      }
+
+      dag.loopConfigs.set(innerLoopId, innerLoop)
+      dag.parallelConfigs.set(outerParallelId, {
+        id: outerParallelId,
+        nodes: [innerLoopId],
+        count: 2,
+        parallelType: 'count',
+      })
+
+      const workflow = createMockWorkflow(
+        [createMockBlock(functionId), createMockBlock(innerLoopId, 'loop')],
+        [
+          {
+            source: outerParallelId,
+            target: innerLoopId,
+            sourceHandle: 'parallel-start-source',
+          },
+          {
+            source: innerLoopId,
+            target: functionId,
+            sourceHandle: 'loop-start-source',
+          },
+        ],
+        { [innerLoopId]: innerLoop }
+      )
+
+      const edgeConstructor = new EdgeConstructor()
+      edgeConstructor.execute(
+        workflow,
+        dag,
+        new Set([innerLoopId]),
+        new Set([functionId]),
+        new Set([outerParallelId, innerLoopId, functionId]),
+        new Map()
+      )
+
+      // Outer sentinel-start → inner loop sentinel-start
+      const outerStartNode = dag.nodes.get(outerSentinelStart)!
+      const edgeToInnerStart = Array.from(outerStartNode.outgoingEdges.values()).find(
+        (e) => e.target === innerSentinelStart
+      )
+      expect(edgeToInnerStart).toBeDefined()
+
+      // Inner loop sentinel-end → outer parallel sentinel-end with loop_exit handle
+      const innerEndNode = dag.nodes.get(innerSentinelEnd)!
+      const edgeToOuterEnd = Array.from(innerEndNode.outgoingEdges.values()).find(
+        (e) => e.target === outerSentinelEnd
+      )
+      expect(edgeToOuterEnd).toBeDefined()
+      expect(edgeToOuterEnd!.sourceHandle).toBe('loop_exit')
+
+      // Inner loop back-edge: sentinel-end → sentinel-start with loop_continue handle
+      const backEdge = Array.from(innerEndNode.outgoingEdges.values()).find(
+        (e) => e.target === innerSentinelStart
+      )
+      expect(backEdge).toBeDefined()
+      expect(backEdge!.sourceHandle).toBe('loop_continue')
+
+      // Inner loop wiring: sentinel-start → function
+      const innerStartNode = dag.nodes.get(innerSentinelStart)!
+      expect(
+        Array.from(innerStartNode.outgoingEdges.values()).some((e) => e.target === functionId)
+      ).toBe(true)
+
+      // Function → inner loop sentinel-end
+      const funcNode = dag.nodes.get(functionId)!
+      expect(
+        Array.from(funcNode.outgoingEdges.values()).some((e) => e.target === innerSentinelEnd)
+      ).toBe(true)
+    })
   })
 })

apps/sim/executor/execution/executor.ts

@@ -71,24 +71,7 @@ export class DAGExecutor {
     const { context, state } = this.createExecutionContext(workflowId, triggerBlockId)
     context.subflowParentMap = this.buildSubflowParentMap(dag)
 
-    const resolver = new VariableResolver(this.workflow, this.workflowVariables, state)
-    const loopOrchestrator = new LoopOrchestrator(dag, state, resolver)
-    loopOrchestrator.setContextExtensions(this.contextExtensions)
-    const parallelOrchestrator = new ParallelOrchestrator(dag, state)
-    parallelOrchestrator.setResolver(resolver)
-    parallelOrchestrator.setContextExtensions(this.contextExtensions)
-    const allHandlers = createBlockHandlers()
-    const blockExecutor = new BlockExecutor(allHandlers, resolver, this.contextExtensions, state)
-    const edgeManager = new EdgeManager(dag)
-    loopOrchestrator.setEdgeManager(edgeManager)
-    const nodeOrchestrator = new NodeExecutionOrchestrator(
-      dag,
-      state,
-      blockExecutor,
-      loopOrchestrator,
-      parallelOrchestrator
-    )
-    const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
+    const engine = this.buildExecutionPipeline(context, dag, state)
     return await engine.run(triggerBlockId)
   }
 
@@ -213,26 +196,36 @@ export class DAGExecutor {
     })
     context.subflowParentMap = this.buildSubflowParentMap(dag)
 
+    const engine = this.buildExecutionPipeline(context, dag, state)
+    return await engine.run()
+  }
+
+  private buildExecutionPipeline(context: ExecutionContext, dag: DAG, state: ExecutionState) {
     const resolver = new VariableResolver(this.workflow, this.workflowVariables, state)
-    const loopOrchestrator = new LoopOrchestrator(dag, state, resolver)
-    loopOrchestrator.setContextExtensions(this.contextExtensions)
-    const parallelOrchestrator = new ParallelOrchestrator(dag, state)
-    parallelOrchestrator.setResolver(resolver)
-    parallelOrchestrator.setContextExtensions(this.contextExtensions)
     const allHandlers = createBlockHandlers()
     const blockExecutor = new BlockExecutor(allHandlers, resolver, this.contextExtensions, state)
     const edgeManager = new EdgeManager(dag)
-    loopOrchestrator.setEdgeManager(edgeManager)
+    const loopOrchestrator = new LoopOrchestrator(
+      dag,
+      state,
+      resolver,
+      this.contextExtensions,
+      edgeManager
+    )
+    const parallelOrchestrator = new ParallelOrchestrator(
+      dag,
+      state,
+      resolver,
+      this.contextExtensions
+    )
     const nodeOrchestrator = new NodeExecutionOrchestrator(
       dag,
       state,
       blockExecutor,
       loopOrchestrator,
       parallelOrchestrator
     )
-    const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
-
-    return await engine.run()
+    return new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
   }
 
   private createExecutionContext(

apps/sim/executor/orchestrators/loop.ts

@@ -42,23 +42,14 @@ export interface LoopContinuationResult {
 }
 
 export class LoopOrchestrator {
-  private edgeManager: EdgeManager | null = null
-  private contextExtensions: ContextExtensions | null = null
-
   constructor(
     private dag: DAG,
     private state: BlockStateController,
-    private resolver: VariableResolver
+    private resolver: VariableResolver,
+    private contextExtensions: ContextExtensions | null = null,
+    private edgeManager: EdgeManager | null = null
   ) {}
 
-  setContextExtensions(contextExtensions: ContextExtensions): void {
-    this.contextExtensions = contextExtensions
-  }
-
-  setEdgeManager(edgeManager: EdgeManager): void {
-    this.edgeManager = edgeManager
-  }
-
   initializeLoopScope(ctx: ExecutionContext, loopId: string): LoopScope {
     const loopConfig = this.dag.loopConfigs.get(loopId) as SerializedLoop | undefined
     if (!loopConfig) {
@@ -103,7 +94,7 @@ export class LoopOrchestrator {
         scope.loopType = 'forEach'
         let items: any[]
         try {
-          items = this.resolveForEachItems(ctx, loopConfig.forEachItems)
+          items = resolveArrayInput(ctx, loopConfig.forEachItems, this.resolver)
         } catch (error) {
           const errorMessage = `ForEach loop resolution failed: ${error instanceof Error ? error.message : String(error)}`
           logger.error(errorMessage, { loopId, forEachItems: loopConfig.forEachItems })
@@ -359,9 +350,8 @@ export class LoopOrchestrator {
     for (const nodeId of loopConfig.nodes) {
       if (this.dag.loopConfigs.has(nodeId)) {
         ctx.loopExecutions?.delete(nodeId)
-        // Delete cloned loop variants (e.g., inner-loop__obranch-N)
-        // Only delete clone entries from subflowParentMap, never the original
-        // structural entries which are needed for SSE iteration context.
+        // Delete cloned loop variants (e.g., inner-loop__obranch-N) but not original
+        // subflowParentMap entries which are needed for SSE iteration context.
         if (ctx.loopExecutions) {
           const prefix = `${nodeId}__obranch-`
           for (const key of ctx.loopExecutions.keys()) {
@@ -400,11 +390,7 @@ export class LoopOrchestrator {
    */
   private deleteParallelScopeAndClones(parallelId: string, ctx: ExecutionContext): void {
     ctx.parallelExecutions?.delete(parallelId)
-    // Do NOT delete the original entry from subflowParentMap — it is a static
-    // structural mapping needed for SSE iteration context on subsequent loop
-    // iterations. Only delete __obranch-N clone entries below.
-
-    // Delete any cloned scopes (e.g., inner-parallel__obranch-1)
+    // Delete cloned scopes (__obranch-N) but not original subflowParentMap entries
     if (ctx.parallelExecutions) {
       const prefix = `${parallelId}__obranch-`
       for (const key of ctx.parallelExecutions.keys()) {
@@ -415,7 +401,6 @@ export class LoopOrchestrator {
       }
     }
 
-    // Recurse into nested subflows within this parallel
     const parallelConfig = this.dag.parallelConfigs.get(parallelId)
     if (parallelConfig?.nodes) {
       for (const nodeId of parallelConfig.nodes) {
@@ -483,17 +468,14 @@ export class LoopOrchestrator {
         for (const id of this.collectAllLoopNodeIds(nodeId, visited)) {
           result.add(id)
         }
-        // Also collect cloned loop variants (e.g., loop-1__obranch-N)
         this.collectClonedSubflowNodes(nodeId, result, visited)
       } else if (this.dag.parallelConfigs.has(nodeId)) {
         for (const id of this.collectAllParallelNodeIds(nodeId, visited)) {
           result.add(id)
         }
-        // Also collect cloned parallel variants
         this.collectClonedSubflowNodes(nodeId, result, visited)
       } else {
         result.add(nodeId)
-        // Collect ALL dynamic branch nodes (not just branch 0)
         this.collectAllBranchNodes(nodeId, result)
       }
     }
@@ -605,23 +587,19 @@ export class LoopOrchestrator {
       return true
     }
 
-    // for: skip if maxIterations is 0
     if (scope.loopType === 'for') {
       if (scope.maxIterations === 0) {
         logger.info('For loop has 0 iterations, skipping loop body', { loopId })
-        // Set empty output for the loop
         this.state.setBlockOutput(loopId, { results: [] }, DEFAULTS.EXECUTION_TIME)
         return false
       }
       return true
     }
 
-    // doWhile: always execute at least once
     if (scope.loopType === 'doWhile') {
       return true
     }
 
-    // while: check condition before first iteration
     if (scope.loopType === 'while') {
       if (!scope.condition) {
         logger.warn('No condition defined for while loop', { loopId })
@@ -641,10 +619,6 @@ export class LoopOrchestrator {
     return true
   }
 
-  shouldExecuteLoopNode(_ctx: ExecutionContext, _nodeId: string, _loopId: string): boolean {
-    return true
-  }
-
   private async evaluateWhileCondition(
     ctx: ExecutionContext,
     condition: string,
@@ -663,10 +637,9 @@ export class LoopOrchestrator {
 
       const evaluatedCondition = replaceValidReferences(condition, (match) => {
         const resolved = this.resolver.resolveSingleReference(ctx, '', match, scope)
-        logger.info('Resolved variable reference in loop condition', {
+        logger.debug('Resolved variable reference in loop condition', {
           reference: match,
           resolvedValue: resolved,
-          resolvedType: typeof resolved,
         })
         if (resolved !== undefined) {
           if (typeof resolved === 'boolean' || typeof resolved === 'number') {
@@ -721,8 +694,4 @@ export class LoopOrchestrator {
       return false
     }
   }
-
-  private resolveForEachItems(ctx: ExecutionContext, items: any): any[] {
-    return resolveArrayInput(ctx, items, this.resolver)
-  }
 }

apps/sim/executor/orchestrators/node.ts

@@ -56,14 +56,6 @@ export class NodeExecutionOrchestrator {
       this.loopOrchestrator.initializeLoopScope(ctx, loopId)
     }
 
-    if (loopId && !this.loopOrchestrator.shouldExecuteLoopNode(ctx, nodeId, loopId)) {
-      return {
-        nodeId,
-        output: {},
-        isFinalOutput: false,
-      }
-    }
-
     const parallelId = node.metadata.parallelId
     if (parallelId && !this.parallelOrchestrator.getParallelScope(ctx, parallelId)) {
       const parallelConfig = this.dag.parallelConfigs.get(parallelId)