more cleanup

Author: icecrasher321

apps/sim/executor/execution/executor.ts

@@ -220,7 +220,7 @@ export class DAGExecutor {
     })
     context.subflowParentMap = this.buildSubflowParentMap(dag)
 
-    const engine = this.buildExecutionPipeline(context, dag, state)
+    const engine = this.buildExecutionPipeline(context, dag, state, filteredSnapshot)
     return await engine.run()
   }
 
@@ -303,7 +303,12 @@ export class DAGExecutor {
     }
   }
 
-  private buildExecutionPipeline(context: ExecutionContext, dag: DAG, state: ExecutionState) {
+  private buildExecutionPipeline(
+    context: ExecutionContext,
+    dag: DAG,
+    state: ExecutionState,
+    snapshotState = this.contextExtensions.snapshotState
+  ) {
     const resolver = new VariableResolver(this.workflow, this.workflowVariables, state, {
       navigatePathAsync,
     })
@@ -325,8 +330,8 @@ export class DAGExecutor {
       edgeManager
     )
     edgeManager.restoreDeactivatedEdges(
-      this.contextExtensions.snapshotState?.deactivatedEdges,
-      this.contextExtensions.snapshotState?.nodesWithActivatedEdge
+      snapshotState?.deactivatedEdges,
+      snapshotState?.nodesWithActivatedEdge
     )
     const nodeOrchestrator = new NodeExecutionOrchestrator(
       dag,

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

@@ -5,6 +5,15 @@ import { describe, expect, it } from 'vitest'
 import { ExecutionState } from '@/executor/execution/state'
 
 describe('ExecutionState', () => {
+  it('returns exact suffixed cached node outputs', () => {
+    const state = new ExecutionState()
+    state.setBlockOutput('producer₍1₎', { value: 'branch-1' })
+    state.setBlockOutput('producer_loop1', { value: 'loop-1' })
+
+    expect(state.getBlockOutput('producer₍1₎')).toEqual({ value: 'branch-1' })
+    expect(state.getBlockOutput('producer_loop1')).toEqual({ value: 'loop-1' })
+  })
+
   it('prefers branch-local cloned outputs when resolving original block references', () => {
     const state = new ExecutionState()
     state.setBlockOutput('producer', { value: 'branch-0' })

apps/sim/executor/execution/state.ts

@@ -26,6 +26,7 @@ export interface LoopScope {
   condition?: string
   loopType?: 'for' | 'forEach' | 'while' | 'doWhile'
   skipFirstConditionCheck?: boolean
+  skippedAtStart?: boolean
   /** Error message if loop validation failed (e.g., exceeded max iterations) */
   validationError?: string
 }
@@ -65,7 +66,7 @@ export class ExecutionState implements BlockStateController {
   getBlockOutput(blockId: string, currentNodeId?: string): NormalizedBlockOutput | undefined {
     const normalizedId = normalizeLookupId(blockId)
     if (normalizedId !== blockId) {
-      return undefined
+      return this.blockStates.get(blockId)?.output
     }
 
     if (currentNodeId) {

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

@@ -124,4 +124,52 @@ describe('LoopOrchestrator', () => {
     )
     expect(scope.maxIterations).toBe(1)
   })
+
+  it('exits immediately when a loop was skipped at start', async () => {
+    const loopId = 'loop-1'
+    const state = createState()
+    const dag: DAG = {
+      nodes: new Map(),
+      loopConfigs: new Map([[loopId, { id: loopId, nodes: ['task-1'], loopType: 'while' }]]),
+      parallelConfigs: new Map(),
+    }
+    const resolver = {
+      resolveSingleReference: vi.fn().mockResolvedValue(1),
+    }
+    const orchestrator = new LoopOrchestrator(dag, state, 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([
+        [
+          loopId,
+          {
+            iteration: 0,
+            currentIterationOutputs: new Map(),
+            allIterationOutputs: [],
+            loopType: 'while',
+            condition: '<loop.index> > 0',
+            skippedAtStart: true,
+          },
+        ],
+      ]),
+      blockLogs: [],
+      metadata: {},
+    }
+
+    const result = await orchestrator.evaluateLoopContinuation(ctx as any, loopId)
+
+    expect(result).toMatchObject({
+      shouldContinue: false,
+      shouldExit: true,
+      selectedRoute: EDGE.LOOP_EXIT,
+      aggregatedResults: [],
+    })
+    expect(resolver.resolveSingleReference).not.toHaveBeenCalled()
+    expect(state.setBlockOutput).toHaveBeenCalledWith(loopId, { results: [] }, 0)
+  })
 })

apps/sim/executor/orchestrators/loop.ts

@@ -253,6 +253,11 @@ export class LoopOrchestrator {
       return await this.createExitResult(ctx, loopId, scope)
     }
 
+    if (scope.skippedAtStart) {
+      scope.skippedAtStart = false
+      return await this.createExitResult(ctx, loopId, scope)
+    }
+
     const iterationResults: NormalizedBlockOutput[] = []
     for (const blockOutput of scope.currentIterationOutputs.values()) {
       iterationResults.push(blockOutput)
@@ -600,6 +605,7 @@ export class LoopOrchestrator {
     if (scope.loopType === 'forEach') {
       if (!scope.items || scope.items.length === 0) {
         logger.info('ForEach loop has empty collection, skipping loop body', { loopId })
+        scope.skippedAtStart = true
         return false
       }
       return true
@@ -608,6 +614,7 @@ export class LoopOrchestrator {
     if (scope.loopType === 'for') {
       if (scope.maxIterations === 0) {
         logger.info('For loop has 0 iterations, skipping loop body', { loopId })
+        scope.skippedAtStart = true
         return false
       }
       return true
@@ -620,6 +627,7 @@ export class LoopOrchestrator {
     if (scope.loopType === 'while') {
       if (!scope.condition) {
         logger.warn('No condition defined for while loop', { loopId })
+        scope.skippedAtStart = true
         return false
       }
 
@@ -632,6 +640,7 @@ export class LoopOrchestrator {
 
       if (!result) {
         logger.info('While loop initial condition is false, skipping loop body', { loopId })
+        scope.skippedAtStart = true
       }
 
       return result

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

@@ -2,11 +2,17 @@
  * @vitest-environment node
  */
 import { beforeEach, describe, expect, it, vi } from 'vitest'
-import type { DAG } from '@/executor/dag/builder'
+import type { DAG, DAGNode } from '@/executor/dag/builder'
 import type { BlockStateWriter, ContextExtensions } from '@/executor/execution/types'
 import { ParallelOrchestrator } from '@/executor/orchestrators/parallel'
 import type { ExecutionContext } from '@/executor/types'
-import { buildBranchNodeId } from '@/executor/utils/subflow-utils'
+import {
+  buildBranchNodeId,
+  buildParallelSentinelEndId,
+  buildParallelSentinelStartId,
+  buildSentinelEndId,
+  buildSentinelStartId,
+} from '@/executor/utils/subflow-utils'
 
 const { mockCompactSubflowResults } = vi.hoisted(() => ({
   mockCompactSubflowResults: vi.fn(async (results: unknown) => results),
@@ -34,6 +40,24 @@ function createDag(): DAG {
   }
 }
 
+function createDagNode(id: string, metadata: DAGNode['metadata'] = {}): DAGNode {
+  return {
+    id,
+    block: {
+      id,
+      position: { x: 0, y: 0 },
+      config: { tool: '', params: {} },
+      inputs: {},
+      outputs: {},
+      metadata: { id: 'function', name: id },
+      enabled: true,
+    },
+    incomingEdges: new Set(),
+    outgoingEdges: new Map(),
+    metadata,
+  }
+}
+
 function createState(): BlockStateWriter {
   return {
     setBlockOutput: vi.fn(),
@@ -422,6 +446,88 @@ describe('ParallelOrchestrator', () => {
     expect(dirtySet.has(secondBranchId)).toBe(true)
   })
 
+  it('marks cloned nested loop body nodes dirty for non-zero branches', () => {
+    const dag = createDag()
+    const parallelId = 'parallel-1'
+    const loopId = 'loop-1'
+    const taskId = 'task-1'
+    const parallelStartId = buildParallelSentinelStartId(parallelId)
+    const parallelEndId = buildParallelSentinelEndId(parallelId)
+    const loopStartId = buildSentinelStartId(loopId)
+    const loopEndId = buildSentinelEndId(loopId)
+
+    dag.parallelConfigs.set(parallelId, {
+      id: parallelId,
+      nodes: [loopId],
+      count: 2,
+      parallelType: 'count',
+    })
+    dag.loopConfigs.set(loopId, {
+      id: loopId,
+      nodes: [taskId],
+      loopType: 'for',
+      iterations: 1,
+    })
+    dag.nodes.set(parallelStartId, createDagNode(parallelStartId))
+    dag.nodes.set(parallelEndId, createDagNode(parallelEndId))
+    dag.nodes.set(
+      loopStartId,
+      createDagNode(loopStartId, {
+        isSentinel: true,
+        sentinelType: 'start',
+        subflowId: loopId,
+        subflowType: 'loop',
+      })
+    )
+    dag.nodes.set(
+      taskId,
+      createDagNode(taskId, {
+        isLoopNode: true,
+        subflowId: loopId,
+        subflowType: 'loop',
+        originalBlockId: taskId,
+      })
+    )
+    dag.nodes.set(
+      loopEndId,
+      createDagNode(loopEndId, {
+        isSentinel: true,
+        sentinelType: 'end',
+        subflowId: loopId,
+        subflowType: 'loop',
+      })
+    )
+    dag.nodes.get(loopStartId)!.outgoingEdges.set(`${loopStartId}->${taskId}`, { target: taskId })
+    dag.nodes.get(taskId)!.incomingEdges.add(loopStartId)
+    dag.nodes.get(taskId)!.outgoingEdges.set(`${taskId}->${loopEndId}`, { target: loopEndId })
+    dag.nodes.get(loopEndId)!.incomingEdges.add(taskId)
+
+    const dirtySet = new Set([parallelId])
+    const orchestrator = new ParallelOrchestrator(dag, createState(), null, {})
+    orchestrator.prepareCurrentBatch(
+      createContext({
+        runFromBlockContext: { startBlockId: parallelId, dirtySet },
+        parallelExecutions: new Map([
+          [
+            parallelId,
+            {
+              parallelId,
+              totalBranches: 2,
+              batchSize: 2,
+              currentBatchStart: 0,
+              currentBatchSize: 2,
+              branchOutputs: new Map(),
+            },
+          ],
+        ]),
+      }),
+      parallelId
+    )
+
+    expect([...dirtySet]).toContain(taskId)
+    expect([...dirtySet].some((nodeId) => nodeId.startsWith(`${taskId}__clone`))).toBe(true)
+  })
+
   it('compacts accumulated outputs before scheduling later batches', async () => {
     const dag = createDag()
     const templateBranchId = buildBranchNodeId('task-1', 0)