fix(executor): capture BlockExecutor locally so finally drains its own instance

Previously this.blockExecutor was overwritten on every buildExecutionPipeline
call. Concurrent or re-entrant execute()/executeFromBlock() calls would have
their finally block drain the wrong instance, allowing the first execution's
block events to land after its terminal event. Returning { engine, blockExecutor }
and capturing both locally makes the drain pinned to the same instance the
engine.run() ran against.

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

Author: waleedlatif1

apps/sim/executor/execution/executor.ts

@@ -53,7 +53,6 @@ export class DAGExecutor {
   private contextExtensions: ContextExtensions
   private dagBuilder: DAGBuilder
   private execLogger: Logger
-  private blockExecutor: BlockExecutor | null = null
 
   constructor(options: DAGExecutorOptions) {
     this.workflow = options.workflow
@@ -80,11 +79,11 @@ export class DAGExecutor {
     const { context, state } = this.createExecutionContext(workflowId, triggerBlockId)
     context.subflowParentMap = this.buildSubflowParentMap(dag)
 
-    const engine = this.buildExecutionPipeline(context, dag, state)
+    const { engine, blockExecutor } = this.buildExecutionPipeline(context, dag, state)
     try {
       return await engine.run(triggerBlockId)
     } finally {
-      await this.blockExecutor?.awaitPendingCallbacks()
+      await blockExecutor.awaitPendingCallbacks()
     }
   }
 
@@ -211,19 +210,18 @@ export class DAGExecutor {
     })
     context.subflowParentMap = this.buildSubflowParentMap(dag)
 
-    const engine = this.buildExecutionPipeline(context, dag, state)
+    const { engine, blockExecutor } = this.buildExecutionPipeline(context, dag, state)
     try {
       return await engine.run()
     } finally {
-      await this.blockExecutor?.awaitPendingCallbacks()
+      await blockExecutor.awaitPendingCallbacks()
     }
   }
 
   private buildExecutionPipeline(context: ExecutionContext, dag: DAG, state: ExecutionState) {
     const resolver = new VariableResolver(this.workflow, this.workflowVariables, state)
     const allHandlers = createBlockHandlers()
     const blockExecutor = new BlockExecutor(allHandlers, resolver, this.contextExtensions, state)
-    this.blockExecutor = blockExecutor
     const edgeManager = new EdgeManager(dag)
     const loopOrchestrator = new LoopOrchestrator(
       dag,
@@ -245,7 +243,8 @@ export class DAGExecutor {
       loopOrchestrator,
       parallelOrchestrator
     )
-    return new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
+    const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
+    return { engine, blockExecutor }
   }
 
   private createExecutionContext(