fix(executor): workflow abort has to send abort signal to route for correct state update (#2571)

apps/sim/executor/execution/engine.ts

@@ -39,7 +39,7 @@ export class ExecutionEngine {
       this.initializeQueue(triggerBlockId)
 
       while (this.hasWork()) {
-        if (this.context.isCancelled && this.executing.size === 0) {
+        if (this.context.abortSignal?.aborted && this.executing.size === 0) {
           break
         }
         await this.processQueue()
@@ -54,7 +54,7 @@ export class ExecutionEngine {
       this.context.metadata.endTime = new Date(endTime).toISOString()
       this.context.metadata.duration = endTime - startTime
 
-      if (this.context.isCancelled) {
+      if (this.context.abortSignal?.aborted) {
         return {
           success: false,
           output: this.finalOutput,
@@ -75,7 +75,7 @@ export class ExecutionEngine {
       this.context.metadata.endTime = new Date(endTime).toISOString()
       this.context.metadata.duration = endTime - startTime
 
-      if (this.context.isCancelled) {
+      if (this.context.abortSignal?.aborted) {
         return {
           success: false,
           output: this.finalOutput,
@@ -234,7 +234,7 @@ export class ExecutionEngine {
 
   private async processQueue(): Promise<void> {
     while (this.readyQueue.length > 0) {
-      if (this.context.isCancelled) {
+      if (this.context.abortSignal?.aborted) {
         break
       }
       const nodeId = this.dequeue()

apps/sim/executor/execution/executor.ts

@@ -37,7 +37,6 @@ export class DAGExecutor {
   private workflowInput: WorkflowInput
   private workflowVariables: Record<string, unknown>
   private contextExtensions: ContextExtensions
-  private isCancelled = false
   private dagBuilder: DAGBuilder
 
   constructor(options: DAGExecutorOptions) {
@@ -54,13 +53,6 @@ export class DAGExecutor {
     const dag = this.dagBuilder.build(this.workflow, triggerBlockId, savedIncomingEdges)
     const { context, state } = this.createExecutionContext(workflowId, triggerBlockId)
 
-    // Link cancellation flag to context
-    Object.defineProperty(context, 'isCancelled', {
-      get: () => this.isCancelled,
-      enumerable: true,
-      configurable: true,
-    })
-
     const resolver = new VariableResolver(this.workflow, this.workflowVariables, state)
     const loopOrchestrator = new LoopOrchestrator(dag, state, resolver)
     loopOrchestrator.setContextExtensions(this.contextExtensions)
@@ -82,10 +74,6 @@ export class DAGExecutor {
     return await engine.run(triggerBlockId)
   }
 
-  cancel(): void {
-    this.isCancelled = true
-  }
-
   async continueExecution(
     _pendingBlocks: string[],
     context: ExecutionContext
@@ -180,6 +168,7 @@ export class DAGExecutor {
       onStream: this.contextExtensions.onStream,
       onBlockStart: this.contextExtensions.onBlockStart,
       onBlockComplete: this.contextExtensions.onBlockComplete,
+      abortSignal: this.contextExtensions.abortSignal,
     }
 
     if (this.contextExtensions.resumeFromSnapshot) {

apps/sim/executor/execution/snapshot.ts

@@ -34,7 +34,6 @@ export interface ExecutionCallbacks {
     blockType: string,
     output: any
   ) => Promise<void>
-  onExecutorCreated?: (executor: any) => void
 }
 
 export interface SerializableExecutionState {

apps/sim/executor/execution/types.ts

@@ -22,6 +22,11 @@ export interface ContextExtensions {
   dagIncomingEdges?: Record<string, string[]>
   snapshotState?: SerializableExecutionState
   metadata?: ExecutionMetadata
+  /**
+   * AbortSignal for cancellation support.
+   * When aborted, the execution should stop gracefully.
+   */
+  abortSignal?: AbortSignal
   onStream?: (streamingExecution: unknown) => Promise<void>
   onBlockStart?: (
     blockId: string,