improvement(executor): redesign executor + add start block (#1790)

* fix(billing): should allow restoring subscription (#1728) * fix(already-cancelled-sub): UI should allow restoring subscription * restore functionality fixed * fix * improvement(start): revert to start block * make it work with start block * fix start block persistence * cleanup triggers * debounce status checks * update docs * improvement(start): revert to start block * make it work with start block * fix start block persistence * cleanup triggers * debounce status checks * update docs * SSE v0.1 * v0.2 * v0.3 * v0.4 * v0.5 * v0.6 * broken checkpoint * Executor progress - everything preliminarily tested except while loops and triggers * Executor fixes * Fix var typing * Implement while loop execution * Loop and parallel result agg * Refactor v1 - loops work * Fix var resolution in for each loop * Fix while loop condition and variable resolution * Fix loop iteration counts * Fix loop badges * Clean logs * Fix variable references from start block * Fix condition block * Fix conditional convergence * Dont execute orphaned nodse * Code cleanup 1 and error surfacing * compile time try catch * Some fixes * Fix error throwing * Sentinels v1 * Fix multiple start and end nodes in loop * Edge restoration * Fix reachable nodes execution * Parallel subflows * Fix loop/parallel sentinel convergence * Loops and parallels orchestrator * Split executor * Variable resolution split * Dag phase * Refactor * Refactor * Refactor 3 * Lint + refactor * Lint + cleanup + refactor * Readability * Initial logs * Fix trace spans * Console pills for iters * Add input/output pills * Checkpoint * remove unused code * THIS IS THE COMMIT THAT CAN BREAK A LOT OF THINGS * ANOTHER BIG REFACTOR * Lint + fix tests * Fix webhook * Remove comment * Merge stash * Fix triggers? * Stuff * Fix error port * Lint * Consolidate state * Clean up some var resolution * Remove some var resolution logs * Fix chat * Fix chat triggers * Fix chat trigger fully * Snapshot refactor * Fix mcp and custom tools * Lint * Fix parallel default count and trace span overlay * Agent purple * Fix test * Fix test --------- Co-authored-by: Waleed <walif6@gmail.com> Co-authored-by: Vikhyath Mondreti <vikhyathvikku@gmail.com> Co-authored-by: Vikhyath Mondreti <vikhyath@simstudio.ai>
2026-02-16 01:15:26 -05:00 · 2025-11-02 12:21:16 -08:00
parent 7d67ae397d
commit 3bf00cbd2a
137 changed files with 8552 additions and 20440 deletions
--- a/apps/sim/executor/execution/block-executor.ts
+++ b/apps/sim/executor/execution/block-executor.ts
@@ -0,0 +1,285 @@
+import { createLogger } from '@/lib/logs/console/logger'
+import { DEFAULTS, EDGE, isSentinelBlockType } from '@/executor/consts'
+import type {
+  BlockHandler,
+  BlockLog,
+  ExecutionContext,
+  NormalizedBlockOutput,
+} from '@/executor/types'
+import type { SerializedBlock } from '@/serializer/types'
+import type { SubflowType } from '@/stores/workflows/workflow/types'
+import type { DAGNode } from '../dag/builder'
+import type { VariableResolver } from '../variables/resolver'
+import type { ExecutionState } from './state'
+import type { ContextExtensions } from './types'
+
+const logger = createLogger('BlockExecutor')
+
+export class BlockExecutor {
+  constructor(
+    private blockHandlers: BlockHandler[],
+    private resolver: VariableResolver,
+    private contextExtensions: ContextExtensions,
+    private state?: ExecutionState
+  ) {}
+
+  async execute(
+    ctx: ExecutionContext,
+    node: DAGNode,
+    block: SerializedBlock
+  ): Promise<NormalizedBlockOutput> {
+    const handler = this.findHandler(block)
+    if (!handler) {
+      throw new Error(`No handler found for block type: ${block.metadata?.id}`)
+    }
+
+    const isSentinel = isSentinelBlockType(block.metadata?.id ?? '')
+
+    let blockLog: BlockLog | undefined
+    if (!isSentinel) {
+      blockLog = this.createBlockLog(ctx, node.id, block, node)
+      ctx.blockLogs.push(blockLog)
+      this.callOnBlockStart(ctx, node, block)
+    }
+
+    const startTime = Date.now()
+    let resolvedInputs: Record<string, any> = {}
+
+    try {
+      resolvedInputs = this.resolver.resolveInputs(ctx, node.id, block.config.params, block)
+      const output = await handler.execute(ctx, block, resolvedInputs)
+
+      const isStreamingExecution =
+        output && typeof output === 'object' && 'stream' in output && 'execution' in output
+
+      let normalizedOutput: NormalizedBlockOutput
+      if (isStreamingExecution) {
+        const streamingExec = output as { stream: ReadableStream; execution: any }
+
+        if (ctx.onStream) {
+          try {
+            await ctx.onStream(streamingExec)
+          } catch (error) {
+            logger.error('Error in onStream callback', { blockId: node.id, error })
+          }
+        }
+
+        normalizedOutput = this.normalizeOutput(
+          streamingExec.execution.output || streamingExec.execution
+        )
+      } else {
+        normalizedOutput = this.normalizeOutput(output)
+      }
+
+      const duration = Date.now() - startTime
+
+      if (blockLog) {
+        blockLog.endedAt = new Date().toISOString()
+        blockLog.durationMs = duration
+        blockLog.success = true
+        blockLog.output = normalizedOutput
+      }
+
+      ctx.blockStates.set(node.id, {
+        output: normalizedOutput,
+        executed: true,
+        executionTime: duration,
+      })
+
+      if (!isSentinel) {
+        this.callOnBlockComplete(ctx, node, block, resolvedInputs, normalizedOutput, duration)
+      }
+
+      return normalizedOutput
+    } catch (error) {
+      const duration = Date.now() - startTime
+      const errorMessage = error instanceof Error ? error.message : String(error)
+
+      if (blockLog) {
+        blockLog.endedAt = new Date().toISOString()
+        blockLog.durationMs = duration
+        blockLog.success = false
+        blockLog.error = errorMessage
+      }
+
+      const errorOutput: NormalizedBlockOutput = {
+        error: errorMessage,
+      }
+
+      ctx.blockStates.set(node.id, {
+        output: errorOutput,
+        executed: true,
+        executionTime: duration,
+      })
+
+      logger.error('Block execution failed', {
+        blockId: node.id,
+        blockType: block.metadata?.id,
+        error: errorMessage,
+      })
+
+      if (!isSentinel) {
+        this.callOnBlockComplete(ctx, node, block, resolvedInputs, errorOutput, duration)
+      }
+
+      const hasErrorPort = this.hasErrorPortEdge(node)
+
+      if (hasErrorPort) {
+        logger.info('Block has error port - returning error output instead of throwing', {
+          blockId: node.id,
+          error: errorMessage,
+        })
+        return errorOutput
+      }
+
+      throw error
+    }
+  }
+
+  private findHandler(block: SerializedBlock): BlockHandler | undefined {
+    return this.blockHandlers.find((h) => h.canHandle(block))
+  }
+
+  private hasErrorPortEdge(node: DAGNode): boolean {
+    for (const [_, edge] of node.outgoingEdges) {
+      if (edge.sourceHandle === EDGE.ERROR) {
+        return true
+      }
+    }
+    return false
+  }
+
+  private createBlockLog(
+    ctx: ExecutionContext,
+    blockId: string,
+    block: SerializedBlock,
+    node: DAGNode
+  ): BlockLog {
+    let blockName = block.metadata?.name || blockId
+    let loopId: string | undefined
+    let parallelId: string | undefined
+    let iterationIndex: number | undefined
+
+    if (node?.metadata) {
+      if (node.metadata.branchIndex !== undefined && node.metadata.parallelId) {
+        blockName = `${blockName} (iteration ${node.metadata.branchIndex})`
+        iterationIndex = node.metadata.branchIndex
+        parallelId = node.metadata.parallelId
+        logger.debug('Added parallel iteration suffix', {
+          blockId,
+          parallelId,
+          branchIndex: node.metadata.branchIndex,
+          blockName,
+        })
+      } else if (node.metadata.isLoopNode && node.metadata.loopId && this.state) {
+        loopId = node.metadata.loopId
+        const loopScope = this.state.getLoopScope(loopId)
+        if (loopScope && loopScope.iteration !== undefined) {
+          blockName = `${blockName} (iteration ${loopScope.iteration})`
+          iterationIndex = loopScope.iteration
+          logger.debug('Added loop iteration suffix', {
+            blockId,
+            loopId,
+            iteration: loopScope.iteration,
+            blockName,
+          })
+        } else {
+          logger.warn('Loop scope not found for block', { blockId, loopId })
+        }
+      }
+    }
+
+    return {
+      blockId,
+      blockName,
+      blockType: block.metadata?.id || DEFAULTS.BLOCK_TYPE,
+      startedAt: new Date().toISOString(),
+      endedAt: '',
+      durationMs: 0,
+      success: false,
+      loopId,
+      parallelId,
+      iterationIndex,
+    }
+  }
+
+  private normalizeOutput(output: unknown): NormalizedBlockOutput {
+    if (output === null || output === undefined) {
+      return {}
+    }
+
+    if (typeof output === 'object' && !Array.isArray(output)) {
+      return output as NormalizedBlockOutput
+    }
+
+    return { result: output }
+  }
+
+  private callOnBlockStart(ctx: ExecutionContext, node: DAGNode, block: SerializedBlock): void {
+    const blockId = node.id
+    const blockName = block.metadata?.name || blockId
+    const blockType = block.metadata?.id || DEFAULTS.BLOCK_TYPE
+
+    const iterationContext = this.getIterationContext(node)
+
+    if (this.contextExtensions.onBlockStart) {
+      this.contextExtensions.onBlockStart(blockId, blockName, blockType, iterationContext)
+    }
+  }
+
+  private callOnBlockComplete(
+    ctx: ExecutionContext,
+    node: DAGNode,
+    block: SerializedBlock,
+    input: Record<string, any>,
+    output: NormalizedBlockOutput,
+    duration: number
+  ): void {
+    const blockId = node.id
+    const blockName = block.metadata?.name || blockId
+    const blockType = block.metadata?.id || DEFAULTS.BLOCK_TYPE
+
+    const iterationContext = this.getIterationContext(node)
+
+    if (this.contextExtensions.onBlockComplete) {
+      this.contextExtensions.onBlockComplete(
+        blockId,
+        blockName,
+        blockType,
+        {
+          input,
+          output,
+          executionTime: duration,
+        },
+        iterationContext
+      )
+    }
+  }
+
+  private getIterationContext(
+    node: DAGNode
+  ): { iterationCurrent: number; iterationTotal: number; iterationType: SubflowType } | undefined {
+    if (!node?.metadata) return undefined
+
+    if (node.metadata.branchIndex !== undefined && node.metadata.branchTotal) {
+      return {
+        iterationCurrent: node.metadata.branchIndex,
+        iterationTotal: node.metadata.branchTotal,
+        iterationType: 'parallel',
+      }
+    }
+
+    if (node.metadata.isLoopNode && node.metadata.loopId && this.state) {
+      const loopScope = this.state.getLoopScope(node.metadata.loopId)
+      if (loopScope && loopScope.iteration !== undefined && loopScope.maxIterations) {
+        return {
+          iterationCurrent: loopScope.iteration,
+          iterationTotal: loopScope.maxIterations,
+          iterationType: 'loop',
+        }
+      }
+    }
+
+    return undefined
+  }
+}
--- a/apps/sim/executor/execution/edge-manager.ts
+++ b/apps/sim/executor/execution/edge-manager.ts
@@ -0,0 +1,223 @@
+import { createLogger } from '@/lib/logs/console/logger'
+import { EDGE } from '@/executor/consts'
+import type { NormalizedBlockOutput } from '@/executor/types'
+import type { DAG, DAGNode } from '../dag/builder'
+import type { DAGEdge } from '../dag/types'
+
+const logger = createLogger('EdgeManager')
+
+export class EdgeManager {
+  private deactivatedEdges = new Set<string>()
+
+  constructor(private dag: DAG) {}
+
+  processOutgoingEdges(
+    node: DAGNode,
+    output: NormalizedBlockOutput,
+    skipBackwardsEdge = false
+  ): string[] {
+    const readyNodes: string[] = []
+    logger.debug('Processing outgoing edges', {
+      nodeId: node.id,
+      edgeCount: node.outgoingEdges.size,
+      skipBackwardsEdge,
+    })
+
+    for (const [edgeId, edge] of node.outgoingEdges) {
+      if (skipBackwardsEdge && this.isBackwardsEdge(edge.sourceHandle)) {
+        logger.debug('Skipping backwards edge', { edgeId })
+        continue
+      }
+
+      const shouldActivate = this.shouldActivateEdge(edge, output)
+      if (!shouldActivate) {
+        const isLoopEdge =
+          edge.sourceHandle === EDGE.LOOP_CONTINUE ||
+          edge.sourceHandle === EDGE.LOOP_CONTINUE_ALT ||
+          edge.sourceHandle === EDGE.LOOP_EXIT
+
+        if (!isLoopEdge) {
+          this.deactivateEdgeAndDescendants(node.id, edge.target, edge.sourceHandle)
+        }
+
+        logger.debug('Edge not activated', {
+          edgeId,
+          sourceHandle: edge.sourceHandle,
+          from: node.id,
+          to: edge.target,
+          isLoopEdge,
+          deactivatedDescendants: !isLoopEdge,
+        })
+        continue
+      }
+
+      const targetNode = this.dag.nodes.get(edge.target)
+      if (!targetNode) {
+        logger.warn('Target node not found', { target: edge.target })
+        continue
+      }
+
+      targetNode.incomingEdges.delete(node.id)
+      logger.debug('Removed incoming edge', {
+        from: node.id,
+        target: edge.target,
+        remainingIncomingEdges: targetNode.incomingEdges.size,
+      })
+
+      if (this.isNodeReady(targetNode)) {
+        logger.debug('Node ready', { nodeId: targetNode.id })
+        readyNodes.push(targetNode.id)
+      }
+    }
+
+    return readyNodes
+  }
+
+  isNodeReady(node: DAGNode): boolean {
+    if (node.incomingEdges.size === 0) {
+      return true
+    }
+
+    const activeIncomingCount = this.countActiveIncomingEdges(node)
+    if (activeIncomingCount > 0) {
+      logger.debug('Node not ready - waiting for active incoming edges', {
+        nodeId: node.id,
+        totalIncoming: node.incomingEdges.size,
+        activeIncoming: activeIncomingCount,
+      })
+      return false
+    }
+
+    logger.debug('Node ready - all remaining edges are deactivated', {
+      nodeId: node.id,
+      totalIncoming: node.incomingEdges.size,
+    })
+    return true
+  }
+
+  restoreIncomingEdge(targetNodeId: string, sourceNodeId: string): void {
+    const targetNode = this.dag.nodes.get(targetNodeId)
+    if (!targetNode) {
+      logger.warn('Cannot restore edge - target node not found', { targetNodeId })
+      return
+    }
+
+    targetNode.incomingEdges.add(sourceNodeId)
+    logger.debug('Restored incoming edge', {
+      from: sourceNodeId,
+      to: targetNodeId,
+    })
+  }
+
+  clearDeactivatedEdges(): void {
+    this.deactivatedEdges.clear()
+  }
+
+  private shouldActivateEdge(edge: DAGEdge, output: NormalizedBlockOutput): boolean {
+    const handle = edge.sourceHandle
+
+    if (handle?.startsWith(EDGE.CONDITION_PREFIX)) {
+      const conditionValue = handle.substring(EDGE.CONDITION_PREFIX.length)
+      return output.selectedOption === conditionValue
+    }
+
+    if (handle?.startsWith(EDGE.ROUTER_PREFIX)) {
+      const routeId = handle.substring(EDGE.ROUTER_PREFIX.length)
+      return output.selectedRoute === routeId
+    }
+
+    if (handle === EDGE.LOOP_CONTINUE || handle === EDGE.LOOP_CONTINUE_ALT) {
+      return output.selectedRoute === EDGE.LOOP_CONTINUE
+    }
+
+    if (handle === EDGE.LOOP_EXIT) {
+      return output.selectedRoute === EDGE.LOOP_EXIT
+    }
+
+    if (handle === EDGE.ERROR && !output.error) {
+      return false
+    }
+
+    if (handle === EDGE.SOURCE && output.error) {
+      return false
+    }
+
+    return true
+  }
+
+  private isBackwardsEdge(sourceHandle?: string): boolean {
+    return sourceHandle === EDGE.LOOP_CONTINUE || sourceHandle === EDGE.LOOP_CONTINUE_ALT
+  }
+
+  private deactivateEdgeAndDescendants(
+    sourceId: string,
+    targetId: string,
+    sourceHandle?: string
+  ): void {
+    const edgeKey = this.createEdgeKey(sourceId, targetId, sourceHandle)
+    if (this.deactivatedEdges.has(edgeKey)) {
+      return
+    }
+
+    this.deactivatedEdges.add(edgeKey)
+    const targetNode = this.dag.nodes.get(targetId)
+    if (!targetNode) return
+
+    const hasOtherActiveIncoming = this.hasActiveIncomingEdges(targetNode, sourceId)
+    if (!hasOtherActiveIncoming) {
+      logger.debug('Deactivating descendants of unreachable node', { nodeId: targetId })
+      for (const [_, outgoingEdge] of targetNode.outgoingEdges) {
+        this.deactivateEdgeAndDescendants(targetId, outgoingEdge.target, outgoingEdge.sourceHandle)
+      }
+    }
+  }
+
+  private hasActiveIncomingEdges(node: DAGNode, excludeSourceId: string): boolean {
+    for (const incomingSourceId of node.incomingEdges) {
+      if (incomingSourceId === excludeSourceId) continue
+
+      const incomingNode = this.dag.nodes.get(incomingSourceId)
+      if (!incomingNode) continue
+
+      for (const [_, incomingEdge] of incomingNode.outgoingEdges) {
+        if (incomingEdge.target === node.id) {
+          const incomingEdgeKey = this.createEdgeKey(
+            incomingSourceId,
+            node.id,
+            incomingEdge.sourceHandle
+          )
+          if (!this.deactivatedEdges.has(incomingEdgeKey)) {
+            return true
+          }
+        }
+      }
+    }
+
+    return false
+  }
+
+  private countActiveIncomingEdges(node: DAGNode): number {
+    let count = 0
+
+    for (const sourceId of node.incomingEdges) {
+      const sourceNode = this.dag.nodes.get(sourceId)
+      if (!sourceNode) continue
+
+      for (const [_, edge] of sourceNode.outgoingEdges) {
+        if (edge.target === node.id) {
+          const edgeKey = this.createEdgeKey(sourceId, edge.target, edge.sourceHandle)
+          if (!this.deactivatedEdges.has(edgeKey)) {
+            count++
+            break
+          }
+        }
+      }
+    }
+
+    return count
+  }
+
+  private createEdgeKey(sourceId: string, targetId: string, sourceHandle?: string): string {
+    return `${sourceId}-${targetId}-${sourceHandle || EDGE.DEFAULT}`
+  }
+}
--- a/apps/sim/executor/execution/engine.ts
+++ b/apps/sim/executor/execution/engine.ts
@@ -0,0 +1,201 @@
+import { createLogger } from '@/lib/logs/console/logger'
+import { BlockType } from '@/executor/consts'
+import type { ExecutionContext, ExecutionResult, NormalizedBlockOutput } from '@/executor/types'
+import type { DAG } from '../dag/builder'
+import type { NodeExecutionOrchestrator } from '../orchestrators/node'
+import type { EdgeManager } from './edge-manager'
+
+const logger = createLogger('ExecutionEngine')
+
+export class ExecutionEngine {
+  private readyQueue: string[] = []
+  private executing = new Set<Promise<void>>()
+  private queueLock = Promise.resolve()
+  private finalOutput: NormalizedBlockOutput = {}
+
+  constructor(
+    private dag: DAG,
+    private edgeManager: EdgeManager,
+    private nodeOrchestrator: NodeExecutionOrchestrator,
+    private context: ExecutionContext
+  ) {}
+
+  async run(triggerBlockId?: string): Promise<ExecutionResult> {
+    const startTime = Date.now()
+    try {
+      this.initializeQueue(triggerBlockId)
+      logger.debug('Starting execution loop', {
+        initialQueueSize: this.readyQueue.length,
+        startNodeId: triggerBlockId,
+      })
+
+      while (this.hasWork()) {
+        await this.processQueue()
+      }
+
+      logger.debug('Execution loop completed', {
+        finalOutputKeys: Object.keys(this.finalOutput),
+      })
+      await this.waitForAllExecutions()
+
+      const endTime = Date.now()
+      this.context.metadata.endTime = new Date(endTime).toISOString()
+      this.context.metadata.duration = endTime - startTime
+
+      return {
+        success: true,
+        output: this.finalOutput,
+        logs: this.context.blockLogs,
+        metadata: this.context.metadata,
+      }
+    } catch (error) {
+      const endTime = Date.now()
+      this.context.metadata.endTime = new Date(endTime).toISOString()
+      this.context.metadata.duration = endTime - startTime
+
+      const errorMessage = error instanceof Error ? error.message : String(error)
+      logger.error('Execution failed', { error: errorMessage })
+
+      const executionResult: ExecutionResult = {
+        success: false,
+        output: this.finalOutput,
+        error: errorMessage,
+        logs: this.context.blockLogs,
+        metadata: this.context.metadata,
+      }
+      const executionError = new Error(errorMessage)
+      ;(executionError as any).executionResult = executionResult
+      throw executionError
+    }
+  }
+
+  private hasWork(): boolean {
+    return this.readyQueue.length > 0 || this.executing.size > 0
+  }
+
+  private addToQueue(nodeId: string): void {
+    if (!this.readyQueue.includes(nodeId)) {
+      this.readyQueue.push(nodeId)
+      logger.debug('Added to queue', { nodeId, queueLength: this.readyQueue.length })
+    }
+  }
+
+  private addMultipleToQueue(nodeIds: string[]): void {
+    for (const nodeId of nodeIds) {
+      this.addToQueue(nodeId)
+    }
+  }
+
+  private dequeue(): string | undefined {
+    return this.readyQueue.shift()
+  }
+
+  private trackExecution(promise: Promise<void>): void {
+    this.executing.add(promise)
+    promise.finally(() => {
+      this.executing.delete(promise)
+    })
+  }
+
+  private async waitForAnyExecution(): Promise<void> {
+    if (this.executing.size > 0) {
+      await Promise.race(this.executing)
+    }
+  }
+
+  private async waitForAllExecutions(): Promise<void> {
+    await Promise.all(Array.from(this.executing))
+  }
+
+  private async withQueueLock<T>(fn: () => Promise<T> | T): Promise<T> {
+    const prevLock = this.queueLock
+    let resolveLock: () => void
+    this.queueLock = new Promise((resolve) => {
+      resolveLock = resolve
+    })
+    await prevLock
+    try {
+      return await fn()
+    } finally {
+      resolveLock!()
+    }
+  }
+
+  private initializeQueue(triggerBlockId?: string): void {
+    if (triggerBlockId) {
+      this.addToQueue(triggerBlockId)
+      return
+    }
+
+    const startNode = Array.from(this.dag.nodes.values()).find(
+      (node) =>
+        node.block.metadata?.id === BlockType.START_TRIGGER ||
+        node.block.metadata?.id === BlockType.STARTER
+    )
+    if (startNode) {
+      this.addToQueue(startNode.id)
+    } else {
+      logger.warn('No start node found in DAG')
+    }
+  }
+
+  private async processQueue(): Promise<void> {
+    while (this.readyQueue.length > 0) {
+      const nodeId = this.dequeue()
+      if (!nodeId) continue
+      const promise = this.executeNodeAsync(nodeId)
+      this.trackExecution(promise)
+    }
+
+    if (this.executing.size > 0) {
+      await this.waitForAnyExecution()
+    }
+  }
+
+  private async executeNodeAsync(nodeId: string): Promise<void> {
+    try {
+      const wasAlreadyExecuted = this.context.executedBlocks.has(nodeId)
+      const result = await this.nodeOrchestrator.executeNode(nodeId, this.context)
+      if (!wasAlreadyExecuted) {
+        await this.withQueueLock(async () => {
+          await this.handleNodeCompletion(nodeId, result.output, result.isFinalOutput)
+        })
+      } else {
+        logger.debug('Node was already executed, skipping edge processing to avoid loops', {
+          nodeId,
+        })
+      }
+    } catch (error) {
+      const errorMessage = error instanceof Error ? error.message : String(error)
+      logger.error('Node execution failed', { nodeId, error: errorMessage })
+      throw error
+    }
+  }
+
+  private async handleNodeCompletion(
+    nodeId: string,
+    output: NormalizedBlockOutput,
+    isFinalOutput: boolean
+  ): Promise<void> {
+    const node = this.dag.nodes.get(nodeId)
+    if (!node) {
+      logger.error('Node not found during completion', { nodeId })
+      return
+    }
+
+    await this.nodeOrchestrator.handleNodeCompletion(nodeId, output, this.context)
+
+    if (isFinalOutput) {
+      this.finalOutput = output
+    }
+
+    const readyNodes = this.edgeManager.processOutgoingEdges(node, output, false)
+    this.addMultipleToQueue(readyNodes)
+
+    logger.debug('Node completion handled', {
+      nodeId,
+      readyNodesCount: readyNodes.length,
+      queueSize: this.readyQueue.length,
+    })
+  }
+}
--- a/apps/sim/executor/execution/executor.ts
+++ b/apps/sim/executor/execution/executor.ts
@@ -0,0 +1,186 @@
+import { createLogger } from '@/lib/logs/console/logger'
+import type { BlockOutput } from '@/blocks/types'
+import { createBlockHandlers } from '@/executor/handlers/registry'
+import type { ExecutionContext, ExecutionResult } from '@/executor/types'
+import {
+  buildResolutionFromBlock,
+  buildStartBlockOutput,
+  resolveExecutorStartBlock,
+} from '@/executor/utils/start-block'
+import type { SerializedWorkflow } from '@/serializer/types'
+import { DAGBuilder } from '../dag/builder'
+import { LoopOrchestrator } from '../orchestrators/loop'
+import { NodeExecutionOrchestrator } from '../orchestrators/node'
+import { ParallelOrchestrator } from '../orchestrators/parallel'
+import { VariableResolver } from '../variables/resolver'
+import { BlockExecutor } from './block-executor'
+import { EdgeManager } from './edge-manager'
+import { ExecutionEngine } from './engine'
+import { ExecutionState } from './state'
+import type { ContextExtensions, WorkflowInput } from './types'
+
+const logger = createLogger('DAGExecutor')
+
+export interface DAGExecutorOptions {
+  workflow: SerializedWorkflow
+  currentBlockStates?: Record<string, BlockOutput>
+  envVarValues?: Record<string, string>
+  workflowInput?: WorkflowInput
+  workflowVariables?: Record<string, unknown>
+  contextExtensions?: ContextExtensions
+}
+
+export class DAGExecutor {
+  private workflow: SerializedWorkflow
+  private initialBlockStates: Record<string, BlockOutput>
+  private environmentVariables: Record<string, string>
+  private workflowInput: WorkflowInput
+  private workflowVariables: Record<string, unknown>
+  private contextExtensions: ContextExtensions
+  private isCancelled = false
+  private dagBuilder: DAGBuilder
+
+  constructor(options: DAGExecutorOptions) {
+    this.workflow = options.workflow
+    this.initialBlockStates = options.currentBlockStates || {}
+    this.environmentVariables = options.envVarValues || {}
+    this.workflowInput = options.workflowInput || {}
+    this.workflowVariables = options.workflowVariables || {}
+    this.contextExtensions = options.contextExtensions || {}
+    this.dagBuilder = new DAGBuilder()
+  }
+
+  async execute(workflowId: string, triggerBlockId?: string): Promise<ExecutionResult> {
+    const dag = this.dagBuilder.build(this.workflow, triggerBlockId)
+    const context = this.createExecutionContext(workflowId, triggerBlockId)
+    // Create state with shared references to context's maps/sets for single source of truth
+    const state = new ExecutionState(context.blockStates, context.executedBlocks)
+    const resolver = new VariableResolver(this.workflow, this.workflowVariables, state)
+    const loopOrchestrator = new LoopOrchestrator(dag, state, resolver)
+    const parallelOrchestrator = new ParallelOrchestrator(dag, state)
+    const allHandlers = createBlockHandlers()
+    const blockExecutor = new BlockExecutor(allHandlers, resolver, this.contextExtensions, state)
+    const edgeManager = new EdgeManager(dag)
+    const nodeOrchestrator = new NodeExecutionOrchestrator(
+      dag,
+      state,
+      blockExecutor,
+      loopOrchestrator,
+      parallelOrchestrator
+    )
+    const engine = new ExecutionEngine(dag, edgeManager, nodeOrchestrator, context)
+    return await engine.run(triggerBlockId)
+  }
+
+  cancel(): void {
+    this.isCancelled = true
+  }
+
+  async continueExecution(
+    pendingBlocks: string[],
+    context: ExecutionContext
+  ): Promise<ExecutionResult> {
+    logger.warn('Debug mode (continueExecution) is not yet implemented in the refactored executor')
+    return {
+      success: false,
+      output: {},
+      logs: context.blockLogs || [],
+      error: 'Debug mode is not yet supported in the refactored executor',
+      metadata: {
+        duration: 0,
+        startTime: new Date().toISOString(),
+      },
+    }
+  }
+
+  private createExecutionContext(workflowId: string, triggerBlockId?: string): ExecutionContext {
+    const context: ExecutionContext = {
+      workflowId,
+      workspaceId: this.contextExtensions.workspaceId,
+      executionId: this.contextExtensions.executionId,
+      isDeployedContext: this.contextExtensions.isDeployedContext,
+      blockStates: new Map(),
+      blockLogs: [],
+      metadata: {
+        startTime: new Date().toISOString(),
+        duration: 0,
+      },
+      environmentVariables: this.environmentVariables,
+      workflowVariables: this.workflowVariables,
+      decisions: {
+        router: new Map(),
+        condition: new Map(),
+      },
+      loopIterations: new Map(),
+      loopItems: new Map(),
+      completedLoops: new Set(),
+      executedBlocks: new Set(),
+      activeExecutionPath: new Set(),
+      workflow: this.workflow,
+      stream: this.contextExtensions.stream || false,
+      selectedOutputs: this.contextExtensions.selectedOutputs || [],
+      edges: this.contextExtensions.edges || [],
+      onStream: this.contextExtensions.onStream,
+      onBlockStart: this.contextExtensions.onBlockStart,
+      onBlockComplete: this.contextExtensions.onBlockComplete,
+    }
+
+    this.initializeStarterBlock(context, triggerBlockId)
+    return context
+  }
+
+  private initializeStarterBlock(context: ExecutionContext, triggerBlockId?: string): void {
+    let startResolution: ReturnType<typeof resolveExecutorStartBlock> | null = null
+
+    if (triggerBlockId) {
+      const triggerBlock = this.workflow.blocks.find((b) => b.id === triggerBlockId)
+      if (!triggerBlock) {
+        logger.error('Specified trigger block not found in workflow', {
+          triggerBlockId,
+        })
+        throw new Error(`Trigger block not found: ${triggerBlockId}`)
+      }
+
+      startResolution = buildResolutionFromBlock(triggerBlock)
+
+      if (!startResolution) {
+        logger.debug('Creating generic resolution for trigger block', {
+          triggerBlockId,
+          blockType: triggerBlock.metadata?.id,
+        })
+        startResolution = {
+          blockId: triggerBlock.id,
+          block: triggerBlock,
+          path: 'split_manual' as any,
+        }
+      }
+    } else {
+      startResolution = resolveExecutorStartBlock(this.workflow.blocks, {
+        execution: 'manual',
+        isChildWorkflow: false,
+      })
+
+      if (!startResolution?.block) {
+        logger.warn('No start block found in workflow')
+        return
+      }
+    }
+
+    const blockOutput = buildStartBlockOutput({
+      resolution: startResolution,
+      workflowInput: this.workflowInput,
+      isDeployedExecution: this.contextExtensions?.isDeployedContext === true,
+    })
+
+    context.blockStates.set(startResolution.block.id, {
+      output: blockOutput,
+      executed: true,
+      executionTime: 0,
+    })
+
+    logger.debug('Initialized start block', {
+      blockId: startResolution.block.id,
+      blockType: startResolution.block.metadata?.id,
+    })
+  }
+}
--- a/apps/sim/executor/execution/snapshot.ts
+++ b/apps/sim/executor/execution/snapshot.ts
@@ -0,0 +1,98 @@
+import type { Edge } from 'reactflow'
+import type { BlockLog, BlockState } from '@/executor/types'
+
+export interface ExecutionMetadata {
+  requestId: string
+  executionId: string
+  workflowId: string
+  workspaceId?: string
+  userId: string
+  triggerType: string
+  triggerBlockId?: string
+  useDraftState: boolean
+  startTime: string
+}
+
+export interface ExecutionCallbacks {
+  onStream?: (streamingExec: any) => Promise<void>
+  onBlockStart?: (blockId: string, blockName: string, blockType: string) => Promise<void>
+  onBlockComplete?: (
+    blockId: string,
+    blockName: string,
+    blockType: string,
+    output: any
+  ) => Promise<void>
+  onExecutorCreated?: (executor: any) => void
+}
+
+export interface SerializableExecutionState {
+  blockStates: Record<string, BlockState>
+  executedBlocks: string[]
+  blockLogs: BlockLog[]
+  decisions: {
+    router: Record<string, string>
+    condition: Record<string, string>
+  }
+  loopIterations: Record<string, number>
+  loopItems: Record<string, any>
+  completedLoops: string[]
+  loopExecutions?: Record<string, any>
+  parallelExecutions?: Record<string, any>
+  parallelBlockMapping?: Record<string, any>
+  activeExecutionPath: string[]
+  pendingQueue?: string[]
+  remainingEdges?: Edge[]
+}
+
+export class ExecutionSnapshot {
+  constructor(
+    public readonly metadata: ExecutionMetadata,
+    public readonly workflow: any,
+    public readonly input: any,
+    public readonly environmentVariables: Record<string, string>,
+    public readonly workflowVariables: Record<string, any>,
+    public readonly selectedOutputs: string[] = [],
+    public readonly state?: SerializableExecutionState
+  ) {}
+
+  toJSON(): string {
+    return JSON.stringify({
+      metadata: this.metadata,
+      workflow: this.workflow,
+      input: this.input,
+      environmentVariables: this.environmentVariables,
+      workflowVariables: this.workflowVariables,
+      selectedOutputs: this.selectedOutputs,
+      state: this.state,
+    })
+  }
+
+  static fromJSON(json: string): ExecutionSnapshot {
+    const data = JSON.parse(json)
+    return new ExecutionSnapshot(
+      data.metadata,
+      data.workflow,
+      data.input,
+      data.environmentVariables,
+      data.workflowVariables,
+      data.selectedOutputs,
+      data.state
+    )
+  }
+}
+
+// TODO: Implement pause/resume functionality
+//
+// Future implementation should include:
+// 1. executor.pause() - Captures current state mid-execution
+//    - Serialize ExecutionContext (blockStates, decisions, loops, etc) to state property
+//    - Save snapshot.toJSON() to database
+// 2. executor.resume(snapshot) - Reconstructs execution from saved state
+//    - Load snapshot from database
+//    - Restore ExecutionContext from state property
+//    - Continue execution from pendingQueue
+// 3. API endpoints:
+//    - POST /api/executions/[id]/pause
+//    - POST /api/executions/[id]/resume
+// 4. Database schema:
+//    - execution_snapshots table with snapshot JSON column
--- a/apps/sim/executor/execution/state.ts
+++ b/apps/sim/executor/execution/state.ts
@@ -0,0 +1,70 @@
+import type { NormalizedBlockOutput } from '@/executor/types'
+export interface LoopScope {
+  iteration: number
+  currentIterationOutputs: Map<string, NormalizedBlockOutput>
+  allIterationOutputs: NormalizedBlockOutput[][]
+  maxIterations?: number
+  item?: any
+  items?: any[]
+  condition?: string
+  skipFirstConditionCheck?: boolean
+}
+
+export interface ParallelScope {
+  parallelId: string
+  totalBranches: number
+  branchOutputs: Map<number, NormalizedBlockOutput[]>
+  completedCount: number
+  totalExpectedNodes: number
+}
+
+export class ExecutionState {
+  // Shared references with ExecutionContext for single source of truth
+  readonly blockStates: Map<
+    string,
+    { output: NormalizedBlockOutput; executed: boolean; executionTime: number }
+  >
+  readonly executedBlocks: Set<string>
+  readonly loopScopes = new Map<string, LoopScope>()
+  readonly parallelScopes = new Map<string, ParallelScope>()
+
+  constructor(
+    blockStates: Map<
+      string,
+      { output: NormalizedBlockOutput; executed: boolean; executionTime: number }
+    >,
+    executedBlocks: Set<string>
+  ) {
+    this.blockStates = blockStates
+    this.executedBlocks = executedBlocks
+  }
+
+  getBlockOutput(blockId: string): NormalizedBlockOutput | undefined {
+    return this.blockStates.get(blockId)?.output
+  }
+
+  setBlockOutput(blockId: string, output: NormalizedBlockOutput): void {
+    this.blockStates.set(blockId, { output, executed: true, executionTime: 0 })
+    this.executedBlocks.add(blockId)
+  }
+
+  hasExecuted(blockId: string): boolean {
+    return this.executedBlocks.has(blockId)
+  }
+
+  getLoopScope(loopId: string): LoopScope | undefined {
+    return this.loopScopes.get(loopId)
+  }
+
+  setLoopScope(loopId: string, scope: LoopScope): void {
+    this.loopScopes.set(loopId, scope)
+  }
+
+  getParallelScope(parallelId: string): ParallelScope | undefined {
+    return this.parallelScopes.get(parallelId)
+  }
+
+  setParallelScope(parallelId: string, scope: ParallelScope): void {
+    this.parallelScopes.set(parallelId, scope)
+  }
+}
--- a/apps/sim/executor/execution/types.ts
+++ b/apps/sim/executor/execution/types.ts
@@ -0,0 +1,38 @@
+import type { NormalizedBlockOutput } from '@/executor/types'
+import type { SubflowType } from '@/stores/workflows/workflow/types'
+
+export interface ContextExtensions {
+  workspaceId?: string
+  executionId?: string
+  stream?: boolean
+  selectedOutputs?: string[]
+  edges?: Array<{ source: string; target: string }>
+  isDeployedContext?: boolean
+  isChildExecution?: boolean
+  onStream?: (streamingExecution: unknown) => Promise<void>
+  onBlockStart?: (
+    blockId: string,
+    blockName: string,
+    blockType: string,
+    iterationContext?: {
+      iterationCurrent: number
+      iterationTotal: number
+      iterationType: SubflowType
+    }
+  ) => Promise<void>
+  onBlockComplete?: (
+    blockId: string,
+    blockName: string,
+    blockType: string,
+    output: { input?: any; output: NormalizedBlockOutput; executionTime: number },
+    iterationContext?: {
+      iterationCurrent: number
+      iterationTotal: number
+      iterationType: SubflowType
+    }
+  ) => Promise<void>
+}
+
+export interface WorkflowInput {
+  [key: string]: unknown
+}