feat(engine): graph interpreter owns the full task lifecycle behind the flag (CU-U3, CU-U4)

Real engine seams: execute delegates to the legacy implementation phase via a completion interceptor that stops execute() at the implementation-complete boundary (no double review/merge); review performs the in-review handoff; merge resolves through ProjectEngine.onMerge over the same serialized merge queue (wired via a late-bound setMergeRequester, mirroring setMergeEnqueuer). Custom graph nodes run on the proven WorkflowStep machinery (readonly tool policy, verdict parsing). Adds a 'planning' seam to the vocabulary (no-op for pre-specified tasks; custom planning is a prompt node today). Entry point: execute() routes graph-selected tasks through the runner when experimentalFeatures.workflowGraphExecutor is on, with process-wide routing claims (FN-4811 posture), duplicate-dispatch dropping, pre-run errors falling back to legacy, and mid-run errors parking the task in review (never re-running the implementation, never stranding the task). Flag off by default: all 587 executor tests pass unchanged.
2026-06-03 11:07:30 -07:00
parent 446879fae0
commit 44e77fda5c
8 changed files with 327 additions and 9 deletions
--- a/packages/engine/src/tests/workflow-graph-executor-parity.test.ts
+++ b/packages/engine/src/tests/workflow-graph-executor-parity.test.ts
@@ -33,6 +33,7 @@ describe("WorkflowGraphExecutor interpreter-parity", () => {
  it("matches legacy execute-review-merge success path", async () => {
    const events: string[] = [];
    const seams: WorkflowLegacySeams = {
+      planning: async () => ({ outcome: "success" }),
      execute: async () => ({ outcome: "success" }),
      review: async () => ({ outcome: "success" }),
      merge: async () => ({ outcome: "success" }),
@@ -53,6 +54,7 @@ describe("WorkflowGraphExecutor interpreter-parity", () => {

  it("routes file-scope-like merge failure parity", async () => {
    const seams: WorkflowLegacySeams = {
+      planning: async () => ({ outcome: "success" }),
      execute: async () => ({ outcome: "success" }),
      review: async () => ({ outcome: "success" }),
      merge: async () => ({ outcome: "failure", value: "FileScopeViolationError" }),
@@ -67,6 +69,7 @@ describe("WorkflowGraphExecutor interpreter-parity", () => {

  it("preserves autoMerge:false terminal in-review semantics via review failure", async () => {
    const seams: WorkflowLegacySeams = {
+      planning: async () => ({ outcome: "success" }),
      execute: async () => ({ outcome: "success" }),
      review: async () => ({ outcome: "failure", value: "manual-merge-required" }),
      merge: async () => ({ outcome: "success" }),
@@ -80,6 +83,7 @@ describe("WorkflowGraphExecutor interpreter-parity", () => {

  it("matches self-healing parity by routing deterministic failure outcomes", async () => {
    const seams: WorkflowLegacySeams = {
+      planning: async () => ({ outcome: "success" }),
      execute: async () => ({ outcome: "failure", value: "recoverable" }),
      review: vi.fn(async () => ({ outcome: "success" as const })),
      merge: vi.fn(async () => ({ outcome: "success" as const })),
@@ -94,6 +98,7 @@ describe("WorkflowGraphExecutor interpreter-parity", () => {

  it("matches moveTask hard-cancel behavior by halting downstream seams", async () => {
    const seams: WorkflowLegacySeams = {
+      planning: async () => ({ outcome: "success" }),
      execute: async () => ({ outcome: "failure", value: "hard-cancel" }),
      review: vi.fn(async () => ({ outcome: "success" as const })),
      merge: vi.fn(async () => ({ outcome: "success" as const })),
--- a/packages/engine/src/tests/workflow-graph-task-runner.test.ts
+++ b/packages/engine/src/tests/workflow-graph-task-runner.test.ts
@@ -64,6 +64,7 @@ function recordingSeams(calls: string[], overrides: Partial<Record<string, Workf
    return overrides[name] ?? { outcome: "success" };
  };
  return {
+    planning: seam("planning"),
    execute: seam("execute"),
    review: seam("review"),
    merge: seam("merge"),
@@ -185,6 +186,39 @@ describe("WorkflowGraphTaskRunner (CU-U2)", () => {
    expect(events).toContain("fallback");
  });

+  it("an interpreter error AFTER side effects terminates as failed, not fell-back", async () => {
+    // Cycle reached only after custom nodes execute: re-running legacy would
+    // repeat the implementation, so the runner must not signal fallback.
+    const cyclicIr: WorkflowIr = {
+      version: "v1",
+      name: "cyclic",
+      nodes: [
+        { id: "start", kind: "start" },
+        { id: "a", kind: "prompt", config: { prompt: "a" } },
+        { id: "b", kind: "prompt", config: { prompt: "b" } },
+        { id: "end", kind: "end" },
+      ],
+      edges: [
+        { from: "start", to: "a", condition: "success" },
+        { from: "a", to: "b", condition: "success" },
+        { from: "b", to: "a", condition: "success" },
+      ],
+    };
+    const calls: string[] = [];
+    const runner = new WorkflowGraphTaskRunner({
+      store: storeWith(definition(cyclicIr)),
+      seams: recordingSeams(calls),
+      runCustomNode: async (node) => {
+        calls.push(`custom:${node.id}`);
+        return { outcome: "success" };
+      },
+    });
+    const result = await runner.run(task, flagOn);
+    expect(calls.length).toBeGreaterThan(0);
+    expect(result.disposition).toBe("failed");
+    expect(result.reason).toMatch(/interpreter-error/);
+  });
+
  it("exposes node outcomes in the shared context for downstream consumers", async () => {
    const runner = new WorkflowGraphTaskRunner({
      store: storeWith(definition(fullLifecycleIr())),
--- a/packages/engine/src/tests/workflow-node-handlers.test.ts
+++ b/packages/engine/src/tests/workflow-node-handlers.test.ts
@@ -9,6 +9,7 @@ const node = (kind: WorkflowIrNode["kind"], seam?: string): WorkflowIrNode => ({
 describe("workflow node handlers", () => {
  it("dispatches prompt node to matching seam", async () => {
    const seams = {
+      planning: vi.fn(async () => ({ outcome: "success" as const })),
      execute: vi.fn(async () => ({ outcome: "success" as const })),
      review: vi.fn(async () => ({ outcome: "success" as const })),
      merge: vi.fn(async () => ({ outcome: "success" as const })),
@@ -22,6 +23,7 @@ describe("workflow node handlers", () => {

  it("dispatches script node to matching seam", async () => {
    const seams = {
+      planning: vi.fn(async () => ({ outcome: "success" as const })),
      execute: vi.fn(async () => ({ outcome: "success" as const })),
      review: vi.fn(async () => ({ outcome: "success" as const })),
      merge: vi.fn(async () => ({ outcome: "success" as const })),
@@ -34,6 +36,7 @@ describe("workflow node handlers", () => {

  it("gate returns failure when expected context value does not match", async () => {
    const handlers = createDefaultNodeHandlers({
+      planning: async () => ({ outcome: "success" }),
      execute: async () => ({ outcome: "success" }),
      review: async () => ({ outcome: "success" }),
      merge: async () => ({ outcome: "success" }),
@@ -49,6 +52,7 @@ describe("workflow node handlers", () => {
  });

  const noopSeams = () => ({
+    planning: vi.fn(async () => ({ outcome: "success" as const })),
    execute: vi.fn(async () => ({ outcome: "success" as const })),
    review: vi.fn(async () => ({ outcome: "success" as const })),
    merge: vi.fn(async () => ({ outcome: "success" as const })),
--- a/packages/engine/src/executor.ts
+++ b/packages/engine/src/executor.ts
@@ -8,8 +8,11 @@ const execAsync = promisify(exec);
 import { delimiter, isAbsolute, join, relative, resolve as resolvePath } from "node:path";
 import { existsSync, realpathSync } from "node:fs";
 import { readFile, rm, writeFile } from "node:fs/promises";
-import type { TaskStore, Task, TaskDetail, TaskTokenUsage, StepStatus, Settings, WorkflowStep, MissionStore, Slice, AgentState, AgentCapability, RunMutationContext, AgentHeartbeatConfig, Agent, AgentMemoryInclusionMode, ProjectSettings } from "@fusion/core";
-import { RetryStormError, TaskDeletedError, serializeRetryStormError } from "@fusion/core";
+import type { TaskStore, Task, TaskDetail, TaskTokenUsage, StepStatus, Settings, WorkflowStep, MissionStore, Slice, AgentState, AgentCapability, RunMutationContext, AgentHeartbeatConfig, Agent, AgentMemoryInclusionMode, ProjectSettings, MergeResult, WorkflowIrNode } from "@fusion/core";
+import { RetryStormError, TaskDeletedError, serializeRetryStormError, isExperimentalFeatureEnabled } from "@fusion/core";
+import { WorkflowGraphTaskRunner, type WorkflowGraphTaskRunResult } from "./workflow-graph-task-runner.js";
+import type { WorkflowLegacySeams } from "./workflow-node-handlers.js";
+import type { WorkflowNodeResult } from "./workflow-graph-executor.js";
 import {
  ApprovalRequestStore,
  buildExecutionMemoryInstructions,
@@ -3149,7 +3152,222 @@ export class TaskExecutor {
   * a task that already has `task.worktree` set, the existing path is used
   * as-is. Branches remain task-scoped (`fusion/{task-id}`).
   */
+  // ── Workflow graph interpreter (cutover M-B/M-C) ─────────────────────────
+  //
+  // When `experimentalFeatures.workflowGraphExecutor` is enabled and a task has
+  // a selected custom workflow, the graph runner owns lifecycle SEQUENCING:
+  // custom prompt/script/gate nodes run via the WorkflowStep machinery, and the
+  // planning/execute/review/merge seam nodes delegate to the legacy engine
+  // implementations. Any interpreter-level error falls back to the legacy
+  // pipeline — a task is never stranded by interpreter bugs.
+
+  /** Completion interceptors for graph-driven tasks: when present for a task,
+   *  execute() stops at the implementation-complete boundary (no workflow
+   *  steps, no review handoff) and hands control back to the graph runner.
+   *  Doubles as the re-entrancy guard for graph routing. */
+  private graphCompletionInterceptors = new Map<string, (info: { modifiedFiles: string[] }) => void>();
+
+  /** Tasks currently being orchestrated by the graph runner. Process-wide for
+   *  the same reason as executingTaskLock (FN-4811): duplicate execute()
+   *  invocations can arrive from different TaskExecutor instances in one
+   *  process (engine restart race, hybrid runtimes), and the graph runner does
+   *  not hold the executing-task lock between seams. */
+  private get graphRouting(): Set<string> {
+    return TaskExecutor.processWideGraphRouting;
+  }
+
+  private static processWideGraphRouting = new Set<string>();
+
+  /** Wired by the runtime to ProjectEngine.onMerge — resolves with the merge outcome. */
+  private mergeRequester?: (taskId: string) => Promise<MergeResult>;
+
+  setMergeRequester(requestMerge: (taskId: string) => Promise<MergeResult>): void {
+    this.mergeRequester = requestMerge;
+  }
+
+  /**
+   * Route a task through the workflow graph interpreter when eligible.
+   * Returns true when the graph owned the task to a terminal disposition
+   * (completed or failed); false when the legacy pipeline should run.
+   */
+  private async maybeExecuteWorkflowGraph(task: Task): Promise<boolean> {
+    // Claim synchronously before any await so concurrent execute() calls for
+    // the same task cannot both enter graph routing (mirrors executingTaskLock).
+    this.graphRouting.add(task.id);
+    try {
+      let settings: Settings;
+      try {
+        settings = await this.store.getSettings();
+      } catch {
+        return false;
+      }
+      if (!isExperimentalFeatureEnabled(settings, "workflowGraphExecutor")) return false;
+      if (typeof this.store.getTaskWorkflowSelection !== "function") return false;
+
+      let selection: { workflowId: string; stepIds: string[] } | undefined;
+      try {
+        selection = this.store.getTaskWorkflowSelection(task.id);
+      } catch {
+        return false;
+      }
+      if (!selection) return false;
+
+      const runner = new WorkflowGraphTaskRunner({
+        store: this.store,
+        seams: this.createGraphSeams(settings),
+        runCustomNode: (node, nodeTask) => this.runGraphCustomNode(node, nodeTask, settings),
+        onEvent: (event) => executorLog.log(`[workflow-graph] ${event.type} ${event.taskId}: ${event.detail}`),
+      });
+      const detail = await this.store.getTask(task.id);
+      const result = await runner.run(detail, settings);
+      if (result.disposition === "fell-back") {
+        executorLog.log(`[workflow-graph] ${task.id} fell back to legacy pipeline: ${result.reason}`);
+        return false;
+      }
+      if (result.disposition === "failed") {
+        await this.handleGraphFailure(task, result);
+      }
+      return true;
+    } finally {
+      this.graphRouting.delete(task.id);
+    }
+  }
+
+  /**
+   * Run ONLY the implementation phase of execute() for a graph-driven task —
+   * full legacy setup plus the agent session up to fn_task_done. The registered
+   * interceptor makes execute() stop at the completion boundary instead of
+   * running workflow steps and the review handoff.
+   */
+  private async runImplementationPhase(task: Task): Promise<{ taskDone: boolean; modifiedFiles: string[] }> {
+    let captured: { taskDone: boolean; modifiedFiles: string[] } = { taskDone: false, modifiedFiles: [] };
+    this.graphCompletionInterceptors.set(task.id, (info) => {
+      captured = { taskDone: true, modifiedFiles: info.modifiedFiles };
+    });
+    try {
+      await this.execute(task);
+    } finally {
+      this.graphCompletionInterceptors.delete(task.id);
+    }
+    return captured;
+  }
+
+  /** Seam implementations delegating to the legacy engine (KTD-1: delegate, never reimplement). */
+  private createGraphSeams(_settings: Settings): WorkflowLegacySeams {
+    return {
+      // Built-in triage/spec generation runs upstream of the interpreter today,
+      // so planning is a no-op for already-specified tasks. Custom planning
+      // behavior is expressed as a custom prompt node before the execute seam.
+      planning: async () => ({ outcome: "success", value: "pre-specified" }),
+      execute: async (seamTask) => {
+        const result = await this.runImplementationPhase(seamTask as Task);
+        return result.taskDone
+          ? { outcome: "success", value: "implemented" }
+          : { outcome: "failure", value: "implementation-incomplete" };
+      },
+      review: async (seamTask) => {
+        // The legacy "review" stage is the in-review handoff: per-step AI review
+        // already ran during implementation (fn_review_step), and the in-review
+        // column is the staging state the merge queue consumes.
+        const live = await this.store.getTask(seamTask.id);
+        await this.persistTokenUsage(seamTask.id);
+        await this.handoffTaskToReview(live, "workflow-graph-review");
+        return { outcome: "success", value: "in-review" };
+      },
+      merge: async (seamTask) => {
+        if (!this.mergeRequester) {
+          return { outcome: "failure", value: "merge-unavailable" };
+        }
+        const result = await this.mergeRequester(seamTask.id);
+        if (result.merged || result.noOp) {
+          return { outcome: "success", value: result.noOp ? "merge-noop" : "merged" };
+        }
+        return { outcome: "failure", value: result.reason ?? result.error ?? "merge-failed" };
+      },
+      schedule: async () => ({ outcome: "success" }),
+    };
+  }
+
+  /** Run a custom (non-seam) graph node on the proven WorkflowStep machinery. */
+  private async runGraphCustomNode(
+    node: WorkflowIrNode,
+    nodeTask: TaskDetail,
+    settings: Settings,
+  ): Promise<WorkflowNodeResult> {
+    const cfg = node.config ?? {};
+    const live = await this.store.getTask(nodeTask.id);
+    const worktreePath = live.worktree || this.rootDir;
+    const scriptName = typeof cfg.scriptName === "string" && cfg.scriptName.trim() ? cfg.scriptName : undefined;
+    const mode: "prompt" | "script" = node.kind === "script" || (node.kind === "gate" && scriptName) ? "script" : "prompt";
+    const now = new Date().toISOString();
+    const step: WorkflowStep = {
+      id: `graph:${node.id}`,
+      name: typeof cfg.name === "string" && cfg.name.trim() ? cfg.name : node.id,
+      description: typeof cfg.description === "string" ? cfg.description : "",
+      mode,
+      phase: "pre-merge",
+      gateMode: node.kind === "gate" || cfg.gateMode === "gate" ? "gate" : "advisory",
+      prompt: typeof cfg.prompt === "string" ? cfg.prompt : "",
+      toolMode: cfg.toolMode === "coding" ? "coding" : "readonly",
+      scriptName,
+      enabled: true,
+      createdAt: now,
+      updatedAt: now,
+      ...(typeof cfg.modelProvider === "string" && typeof cfg.modelId === "string"
+        ? { modelProvider: cfg.modelProvider, modelId: cfg.modelId }
+        : {}),
+    };
+
+    const outcome = mode === "script"
+      ? await this.executeScriptWorkflowStep(live, step, worktreePath, settings)
+      : await this.executeWorkflowStep(live, step, worktreePath, settings);
+
+    const blocking = step.gateMode === "gate";
+    // Script-mode outcomes carry no structured verdict; prompt-mode may.
+    const verdict = (outcome as { verdict?: string }).verdict;
+    return {
+      outcome: outcome.success || !blocking ? "success" : "failure",
+      value: verdict ?? (outcome.success ? "passed" : "failed"),
+    };
+  }
+
+  /** Terminal failure of a graph run: record the error and park the task in
+   *  review so a human can act — never leave it invisible in in-progress. */
+  private async handleGraphFailure(task: Task, result: WorkflowGraphTaskRunResult): Promise<void> {
+    const failedNode = result.visitedNodeIds[result.visitedNodeIds.length - 1];
+    const message = `Workflow graph terminated with failure at node '${failedNode ?? "unknown"}'`;
+    executorLog.warn(`${task.id}: ${message}`);
+    try {
+      await this.store.logEntry(task.id, message, undefined, this.getRunContextFor(task.id));
+      await this.store.updateTask(task.id, { error: message }, this.getRunContextFor(task.id));
+      const live = await this.store.getTask(task.id);
+      if (live.column === "in-progress") {
+        await this.persistTokenUsage(task.id);
+        await this.handoffTaskToReview(live, "workflow-graph-failed");
+      }
+    } catch (err) {
+      executorLog.error(
+        `${task.id}: failed to park graph-failed task: ${err instanceof Error ? err.message : String(err)}`,
+      );
+    }
+  }
+
  async execute(task: Task): Promise<void> {
+    // Workflow graph interpreter routing (cutover M-C): graph-selected tasks
+    // are orchestrated by the interpreter. The execute seam re-enters this
+    // method with a completion interceptor registered (which claims the task
+    // lock normally), so routing is skipped for that inner invocation.
+    if (!this.graphCompletionInterceptors.has(task.id)) {
+      if (this.graphRouting.has(task.id)) {
+        // Duplicate dispatch while the graph runner owns this task — drop it,
+        // mirroring the executingTaskLock duplicate-invocation behavior.
+        executorLog.log(`execute() called for ${task.id} while graph routing is active — skipping duplicate`);
+        return;
+      }
+      const graphOwned = await this.maybeExecuteWorkflowGraph(task);
+      if (graphOwned) return;
+    }
+
    // FN-4811 follow-up (FN-4814/FN-4809/FN-4811 production failure): claim a
    // PROCESS-WIDE lock synchronously before any other work. Per-instance
    // `this.executing` was insufficient in production because two execute()
@@ -4555,6 +4773,17 @@ export class TaskExecutor {
              executorLog.log(`${task.id}: captured ${modifiedFiles.length} modified files`);
            }

+            // Graph-driven completion (interpreter cutover): the workflow graph
+            // owns workflow steps, review handoff, and merge from here — stop
+            // at the implementation-complete boundary and hand control back.
+            const graphCompletion = this.graphCompletionInterceptors.get(task.id);
+            if (graphCompletion) {
+              this.clearCompletedTaskWatchdog(task.id);
+              executorLog.log(`✓ ${task.id} implementation complete — graph interpreter owns the remaining lifecycle`);
+              graphCompletion({ modifiedFiles });
+              return;
+            }
+
            this.scheduleCompletedTaskWatchdog(task.id, "task completion");
            if (await this.shouldDeferCompletionForGlobalPause(task.id, "before workflow steps after task completion")) {
              return;
--- a/packages/engine/src/project-engine.ts
+++ b/packages/engine/src/project-engine.ts
@@ -353,6 +353,9 @@ export class ProjectEngine {
    this.runtime.setMergeActiveClearer?.((taskId) => {
      this.mergeActive.delete(taskId);
    });
+    // Workflow-graph interpreter merge seam: resolves with the merge outcome
+    // through the same serialized merge queue the legacy pipeline uses.
+    this.runtime.setMergeRequester?.((taskId) => this.onMerge(taskId));
  }

  getActiveMergeTaskId(): string | null {
--- a/packages/engine/src/runtimes/in-process-runtime.ts
+++ b/packages/engine/src/runtimes/in-process-runtime.ts
@@ -131,6 +131,7 @@ export class InProcessRuntime
   * before `start()` via `setMergeEnqueuer`.
   */
  private mergeEnqueuer?: (taskId: string) => boolean;
+  private mergeRequester?: (taskId: string) => Promise<import("@fusion/core").MergeResult>;
  private clearMergeActive?: (taskId: string) => void;
  private activeMergeTaskIdProvider?: () => string | null;
  /** Tracks whether startup recovery was intentionally deferred due to pause state. */
@@ -495,6 +496,9 @@ export class InProcessRuntime
        this.config.workingDirectory,
        executorOptions
      );
+      if (this.mergeRequester) {
+        this.executor.setMergeRequester(this.mergeRequester);
+      }

      this.worktreePool.setInvariantViolationHandler((violation: PoolInvariantViolation) => {
        void (async () => {
@@ -1042,6 +1046,16 @@ export class InProcessRuntime
    this.mergeEnqueuer = enqueueMerge;
  }

+  /**
+   * Wire the workflow-graph merge seam to ProjectEngine.onMerge. Late-bindable:
+   * forwards immediately when the executor already exists, and is re-applied at
+   * executor construction during start().
+   */
+  setMergeRequester(requestMerge: (taskId: string) => Promise<import("@fusion/core").MergeResult>): void {
+    this.mergeRequester = requestMerge;
+    this.executor?.setMergeRequester(requestMerge);
+  }
+
  setMergeActiveClearer(clearMergeActive: (taskId: string) => void): void {
    this.clearMergeActive = clearMergeActive;
  }
--- a/packages/engine/src/workflow-graph-task-runner.ts
+++ b/packages/engine/src/workflow-graph-task-runner.ts
@@ -3,6 +3,7 @@ import { isExperimentalFeatureEnabled } from "@fusion/core";

 import { WorkflowGraphExecutor, type WorkflowNodeOutcome } from "./workflow-graph-executor.js";
 import type { WorkflowCustomNodeRunner, WorkflowLegacySeams } from "./workflow-node-handlers.js";
+// (Both types are also used as values in the side-effect tracking wrappers below.)

 /**
 * Terminal disposition of an interpreter-driven task run.
@@ -90,10 +91,29 @@ export class WorkflowGraphTaskRunner {
    }

    this.emit("start", task.id, definition.id);
+
+    // Track whether any node side effects ran. A pre-run interpreter error
+    // (bad IR structure, wiring) can safely fall back to the legacy pipeline;
+    // a mid-run error cannot — re-running legacy would repeat the implementation
+    // session — so it terminates as "failed" for the caller to park instead.
+    let sideEffectsRan = false;
+    const seams = this.deps.seams;
+    const wrappedSeams: WorkflowLegacySeams = {
+      planning: (t, c) => ((sideEffectsRan = true), seams.planning(t, c)),
+      execute: (t, c) => ((sideEffectsRan = true), seams.execute(t, c)),
+      review: (t, c) => ((sideEffectsRan = true), seams.review(t, c)),
+      merge: (t, c) => ((sideEffectsRan = true), seams.merge(t, c)),
+      schedule: (t, c) => ((sideEffectsRan = true), seams.schedule(t, c)),
+    };
+    const wrappedRunCustomNode: WorkflowCustomNodeRunner = (node, t, c) => {
+      sideEffectsRan = true;
+      return this.deps.runCustomNode(node, t, c);
+    };
+
    try {
      const executor = new WorkflowGraphExecutor({
-        seams: this.deps.seams,
-        runCustomNode: this.deps.runCustomNode,
+        seams: wrappedSeams,
+        runCustomNode: wrappedRunCustomNode,
        maxRetriesPerNode: this.deps.maxRetriesPerNode,
      });
      const result = await executor.run(task, settings, definition.ir);
@@ -109,9 +129,13 @@ export class WorkflowGraphTaskRunner {
        context: result.context,
      };
    } catch (err) {
-      // Interpreter-level error (bad IR, handler wiring, etc.) — never strand
-      // the task; the caller runs the legacy pipeline.
-      return this.fallBack(task.id, `interpreter-error: ${err instanceof Error ? err.message : String(err)}`);
+      const reason = `interpreter-error: ${err instanceof Error ? err.message : String(err)}`;
+      if (sideEffectsRan) {
+        // Too late to fall back — the caller parks the task for human review.
+        this.emit("terminal", task.id, `${definition.id}:failed (${reason})`);
+        return { disposition: "failed", outcome: "failure", reason, visitedNodeIds: [] };
+      }
+      return this.fallBack(task.id, reason);
    }
  }
 }
--- a/packages/engine/src/workflow-node-handlers.ts
+++ b/packages/engine/src/workflow-node-handlers.ts
@@ -3,9 +3,13 @@ import type { TaskDetail, WorkflowIrNode } from "@fusion/core";

 import type { WorkflowNodeHandler, WorkflowNodeResult } from "./workflow-graph-executor.js";

-export type WorkflowSeamName = "execute" | "review" | "merge" | "schedule";
+export type WorkflowSeamName = "planning" | "execute" | "review" | "merge" | "schedule";

 export interface WorkflowLegacySeams {
+  /** Planning/spec stage. Built-in triage runs upstream of the interpreter
+   *  today, so the default engine seam is a no-op for already-specified tasks;
+   *  custom planning behavior is expressed as a custom prompt node. */
+  planning: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
  execute: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
  review: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
  merge: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
@@ -27,7 +31,7 @@ export type WorkflowCustomNodeRunner = (
 export function resolveSeamName(node: { config?: Record<string, unknown> }): WorkflowSeamName | undefined {
  const seam = node.config?.seam;
  if (seam === undefined) return undefined;
-  if (seam === "execute" || seam === "review" || seam === "merge" || seam === "schedule") {
+  if (seam === "planning" || seam === "execute" || seam === "review" || seam === "merge" || seam === "schedule") {
    return seam;
  }
  throw new WorkflowIrError(`Unsupported workflow seam: ${String(seam)}`);
@@ -99,6 +103,7 @@ export const gateNodeHandler: WorkflowNodeHandler = createGateHandler();
 export function createNoopLegacySeams(): WorkflowLegacySeams {
  const success = async (): Promise<WorkflowNodeResult> => ({ outcome: "success" });
  return {
+    planning: success,
    execute: success,
    review: success,
    merge: success,