diff --git a/docs/plans/2026-06-03-002-feat-workflow-interpreter-cutover-plan.md b/docs/plans/2026-06-03-002-feat-workflow-interpreter-cutover-plan.md
new file mode 100644
index 0000000000..98ad6b1c33
--- /dev/null
+++ b/docs/plans/2026-06-03-002-feat-workflow-interpreter-cutover-plan.md
@@ -0,0 +1,82 @@
+---
+title: "feat: Workflow interpreter cutover — graph owns the full lifecycle"
+type: feat
+status: active
+date: 2026-06-03
+depth: deep
+origin: docs/plans/2026-06-03-001-feat-executable-custom-workflows-node-editor-plan.md (Deferred Track)
+---
+
+# feat: Workflow interpreter cutover — graph owns the full lifecycle
+
+## Summary
+
+Promote `WorkflowGraphExecutor` from a flag-off no-op to the authoritative driver of a task's lifecycle, so a custom workflow graph can **replace** planning → execute → review → merge, not just inject steps around it. The user has explicitly waived the FN-4359 reliability freeze for this track; the parity invariants below remain the correctness bar regardless.
+
+The existing scaffold already provides: graph walking with cycle detection, `success`/`failure`/`outcome:*` edge conditions, per-node retries, a `WorkflowLegacySeams` interface (`execute/review/merge/schedule`), the `workflowGraphExecutor` experimental flag, and the dual-observe parity machinery. The MVP (plan 001) provides: persisted workflow definitions, the node editor, and per-task selection. What's missing is **real seam implementations**, **custom-node handlers** (the default handlers throw for non-seam prompt/script nodes), and the **entry point** that routes a graph-selected task through the interpreter.
+
+---
+
+## Scope Boundaries
+
+### In scope
+- M-A: Interpreter foundation — custom-node handlers + a `WorkflowGraphTaskRunner` with injected seams, fully tested with fakes.
+- M-B: Real seam wiring — delegate execute/review/merge to the legacy engine implementations via narrow injected callbacks from `TaskExecutor`/`ProjectEngine`.
+- M-C: Flag-gated entry point — graph-selected tasks route through the interpreter when `experimentalFeatures.workflowGraphExecutor` is on; legacy fallback on any interpreter error.
+- M-D: Parity + graduation — dual-observe on real runs, drive drift to zero, then default the flag on for graph-selected tasks.
+
+### Deferred to Follow-Up Work
+- Removing the legacy hardcoded pipeline (FN-5719 Phase 4) — only after M-D proves parity in the field.
+- Planning/triage as a replaceable seam (the `schedule` seam exists but triage replacement needs the planning subsystem mapped first).
+- Parallel branches (fan-out) executing concurrently — the walker is sequential; concurrency within a graph is a later extension.
+
+---
+
+## Key Technical Decisions
+
+- **KTD-1 — Seams delegate, never reimplement.** The `execute`/`review`/`merge` seam implementations call the same engine functions the legacy path uses (agent session machinery, `reviewStep`, the auto-merge queue). The interpreter owns *sequencing*; the engine keeps owning *mechanics* (worktrees, leases, file-scope guard, squash contract, self-healing). This is the enqueue-only posture from DAG ADR-0001 applied to seams.
+- **KTD-2 — Custom nodes run on the WorkflowStep machinery.** Non-seam prompt/script/gate nodes execute via the same prompt-session/script/verdict machinery as workflow steps (proven, readonly-tool-policy aware), invoked through an injected `runCustomNode` callback — the interpreter stays engine-agnostic and unit-testable with fakes.
+- **KTD-3 — Legacy fallback on interpreter error (M-C).** Any thrown error from the interpreter path (not a graph-routed `failure` edge) falls back to the legacy pipeline for that task and emits an audit event. No task is ever stranded by interpreter bugs.
+- **KTD-4 — Column transitions are seam side-effects.** `execute` seam entry → `in-progress`, review handoff → `in-review`, merge success → `done` — performed by the delegated engine code itself (KTD-1), so board invariants (FN-5147 terminal-until-merged, hard-cancel) are preserved by construction.
+- **KTD-5 — Invariant bar (from FN-5719 / workflow-steps.md):** `FileScopeViolationError` guard, squash/merge contract, `autoMerge:false` terminal-until-merged, `moveTask(in-progress→todo)` hard-cancel, resume-limbo non-oscillation. Parity is machine-checked via the existing `compareWorkflowRunObservations` / `compareWorkflowRunAudits`.
+
+---
+
+## Implementation Units
+
+### U1. Custom-node handlers (unblock non-seam nodes)
+**Goal:** Default handlers route seam-configured prompt/script nodes to seams and **custom** prompt/script nodes to an injected runner instead of throwing.
+**Files:** `packages/engine/src/workflow-node-handlers.ts`, `packages/engine/src/__tests__/workflow-node-handlers.test.ts`.
+**Approach:** `createDefaultNodeHandlers(seams, runCustomNode)` — `resolveSeam` returns undefined (not throw) for non-seam nodes; gate nodes keep the context-gate behavior but also support prompt/script-backed gates via `runCustomNode` with `gateMode` semantics.
+**Test scenarios:** seam node → seam called; custom prompt node → runner called with node config; custom script node → runner; gate node with context expectation → pass/fail; gate node with prompt config → runner verdict drives outcome; unknown seam string → error.
+
+### U2. WorkflowGraphTaskRunner (engine-agnostic orchestration)
+**Goal:** A runner that loads a task's selected workflow IR, runs `WorkflowGraphExecutor` with injected seams + custom-node runner, and maps the terminal outcome to a lifecycle disposition (`completed` / `failed` / `fell-back`).
+**Files:** `packages/engine/src/workflow-graph-task-runner.ts` (new), `packages/engine/src/__tests__/workflow-graph-task-runner.test.ts` (new).
+**Approach:** Pure DI: `{ store, seams, runCustomNode, settings }`. Reads selection via `store.getTaskWorkflowSelection` + `getWorkflowDefinition`. Falls back (disposition `fell-back`) when flag off, no selection, or IR load fails. Audit events for start/terminal/fallback.
+**Test scenarios:** full graph run order with fake seams (execute→review→merge sequencing via the builtin IR); custom pre-merge node runs between start and execute when authored that way; failure edge routes to end with `failed`; thrown seam error → `fell-back`; flag off → `fell-back`; gate failure blocks merge seam.
+
+### U3. Real seam implementations (M-B)
+**Goal:** `createEngineSeams(executor, projectEngine)` delegating to real engine entry points.
+**Files:** `packages/engine/src/workflow-engine-seams.ts` (new), executor/project-engine narrow accessor methods as needed, `packages/engine/src/__tests__/workflow-engine-seams.test.ts`.
+**Approach:** execute → the executor's implementation-phase entry for an already-claimed task; review → `reviewStep` path with verdict mapped to `outcome:*`; merge → enqueue on the auto-merge queue and await terminal merge outcome. Each seam returns `WorkflowNodeResult` with `value` carrying verdict/outcome tokens for edge conditions.
+**Execution note:** characterization-first — capture the legacy call sequence for one task end-to-end before extracting accessors.
+
+### U4. Flag-gated entry point + fallback (M-C)
+**Goal:** Graph-selected tasks route through the runner from `TaskExecutor.execute`; interpreter errors fall back to legacy mid-flight where safe, else fail the task through existing recovery.
+**Files:** `packages/engine/src/executor.ts` (top-of-execute branch only), `packages/engine/src/__tests__/workflow-graph-entry.test.ts`.
+
+### U5. Dual-observe parity on real runs + graduation (M-D)
+**Goal:** Enable `workflowInterpreterDualObserve` for graph-selected tasks, record drift, fix until `{agree:true}` is sustained, then flip `workflowGraphExecutor` default for selected-workflow tasks.
+**Files:** `packages/engine/src/workflow-parity-observer.ts` call-site wiring, settings default change, docs update.
+
+---
+
+## Risks
+
+- **TaskExecutor.execute is ~3k lines of intertwined state.** Mitigation: U3 extracts *accessors*, never moves logic; U4 touches only a top-of-function branch; characterization tests first.
+- **Self-healing/recovery assume the legacy shape.** Mitigation: KTD-3 fallback + parity observation before authority; recovery paths treat interpreter tasks as legacy until M-D.
+- **Merge queue awaiting from inside a graph walk** could deadlock with the executor's own lifecycle. Mitigation: merge seam enqueues and resolves on the queue's completion callback (same contract the legacy handoff uses), never polls.
+
+## Sources
+Plan 001 (Deferred Track), `workflow-graph-executor.ts`, `workflow-node-handlers.ts`, `workflow-parity-observer.ts`, `docs/rfcs/FN-5719-decouple-executor-merger.md`, `docs/dag/adr-0001-dag-orchestration.md`, `docs/workflow-steps.md`.
diff --git a/packages/engine/src/__tests__/workflow-graph-task-runner.test.ts b/packages/engine/src/__tests__/workflow-graph-task-runner.test.ts
new file mode 100644
index 0000000000..fb44f27e6b
--- /dev/null
+++ b/packages/engine/src/__tests__/workflow-graph-task-runner.test.ts
@@ -0,0 +1,211 @@
+import { describe, expect, it } from "vitest";
+import type { Settings, TaskDetail, WorkflowDefinition, WorkflowIr } from "@fusion/core";
+
+import { WorkflowGraphTaskRunner, type WorkflowGraphRunnerStore } from "../workflow-graph-task-runner.js";
+import type { WorkflowNodeResult } from "../workflow-graph-executor.js";
+
+const task = { id: "FN-9001" } as TaskDetail;
+const flagOn = { experimentalFeatures: { workflowGraphExecutor: true } } as unknown as Pick<
+  Settings,
+  "experimentalFeatures"
+>;
+const flagOff = { experimentalFeatures: {} } as unknown as Pick<Settings, "experimentalFeatures">;
+
+/** start → lint(custom) → execute → review → merge → notify(custom) → end, with seam failure edges to end. */
+function fullLifecycleIr(): WorkflowIr {
+  return {
+    version: "v1",
+    name: "full",
+    nodes: [
+      { id: "start", kind: "start" },
+      { id: "lint", kind: "prompt", config: { prompt: "lint it" } },
+      { id: "execute", kind: "prompt", config: { seam: "execute" } },
+      { id: "review", kind: "prompt", config: { seam: "review" } },
+      { id: "merge", kind: "prompt", config: { seam: "merge" } },
+      { id: "notify", kind: "script", config: { scriptName: "notify" } },
+      { id: "zend", kind: "end" },
+    ],
+    edges: [
+      { from: "start", to: "lint" },
+      { from: "lint", to: "execute", condition: "success" },
+      { from: "execute", to: "review", condition: "success" },
+      { from: "review", to: "merge", condition: "success" },
+      { from: "merge", to: "notify", condition: "success" },
+      { from: "notify", to: "zend", condition: "success" },
+      { from: "execute", to: "zend", condition: "failure" },
+      { from: "review", to: "zend", condition: "failure" },
+      { from: "merge", to: "zend", condition: "failure" },
+    ],
+  };
+}
+
+function definition(ir: WorkflowIr): WorkflowDefinition {
+  return {
+    id: "WF-001",
+    name: "Full lifecycle",
+    description: "",
+    ir,
+    layout: {},
+    createdAt: "2026-06-03T00:00:00.000Z",
+    updatedAt: "2026-06-03T00:00:00.000Z",
+  };
+}
+
+function storeWith(def: WorkflowDefinition | undefined, workflowId = "WF-001"): WorkflowGraphRunnerStore {
+  return {
+    getTaskWorkflowSelection: () => (def ? { workflowId, stepIds: [] } : undefined),
+    getWorkflowDefinition: async () => def,
+  };
+}
+
+function recordingSeams(calls: string[], overrides: Partial<Record<string, WorkflowNodeResult>> = {}) {
+  const seam = (name: string) => async (): Promise<WorkflowNodeResult> => {
+    calls.push(name);
+    return overrides[name] ?? { outcome: "success" };
+  };
+  return {
+    execute: seam("execute"),
+    review: seam("review"),
+    merge: seam("merge"),
+    schedule: seam("schedule"),
+  };
+}
+
+describe("WorkflowGraphTaskRunner (CU-U2)", () => {
+  it("runs the full lifecycle in graph order: custom → execute → review → merge → custom", async () => {
+    const calls: string[] = [];
+    const runner = new WorkflowGraphTaskRunner({
+      store: storeWith(definition(fullLifecycleIr())),
+      seams: recordingSeams(calls),
+      runCustomNode: async (node) => {
+        calls.push(`custom:${node.id}`);
+        return { outcome: "success" };
+      },
+    });
+
+    const result = await runner.run(task, flagOn);
+
+    expect(result.disposition).toBe("completed");
+    expect(calls).toEqual(["custom:lint", "execute", "review", "merge", "custom:notify"]);
+    expect(result.visitedNodeIds).toEqual(["start", "lint", "execute", "review", "merge", "notify"]);
+  });
+
+  it("a failing seam terminates the run as failed without running later nodes", async () => {
+    const calls: string[] = [];
+    const runner = new WorkflowGraphTaskRunner({
+      store: storeWith(definition(fullLifecycleIr())),
+      seams: recordingSeams(calls, { review: { outcome: "failure", value: "REVISE" } }),
+      runCustomNode: async (node) => {
+        calls.push(`custom:${node.id}`);
+        return { outcome: "success" };
+      },
+    });
+
+    const result = await runner.run(task, flagOn);
+
+    expect(result.disposition).toBe("failed");
+    expect(calls).toEqual(["custom:lint", "execute", "review"]);
+    expect(calls).not.toContain("merge");
+  });
+
+  it("a failing custom gate before execute blocks the whole pipeline", async () => {
+    const calls: string[] = [];
+    const runner = new WorkflowGraphTaskRunner({
+      store: storeWith(definition(fullLifecycleIr())),
+      seams: recordingSeams(calls),
+      runCustomNode: async (node) => {
+        calls.push(`custom:${node.id}`);
+        return node.id === "lint" ? { outcome: "failure", value: "lint-failed" } : { outcome: "success" };
+      },
+    });
+
+    const result = await runner.run(task, flagOn);
+
+    expect(result.disposition).toBe("failed");
+    expect(calls).toEqual(["custom:lint"]);
+  });
+
+  it("falls back when the flag is off", async () => {
+    const runner = new WorkflowGraphTaskRunner({
+      store: storeWith(definition(fullLifecycleIr())),
+      seams: recordingSeams([]),
+      runCustomNode: async () => ({ outcome: "success" }),
+    });
+    const result = await runner.run(task, flagOff);
+    expect(result.disposition).toBe("fell-back");
+    expect(result.reason).toBe("flag-off");
+  });
+
+  it("falls back when the task has no workflow selection", async () => {
+    const runner = new WorkflowGraphTaskRunner({
+      store: storeWith(undefined),
+      seams: recordingSeams([]),
+      runCustomNode: async () => ({ outcome: "success" }),
+    });
+    const result = await runner.run(task, flagOn);
+    expect(result).toMatchObject({ disposition: "fell-back", reason: "no-selection" });
+  });
+
+  it("falls back when the selected workflow no longer exists", async () => {
+    const store: WorkflowGraphRunnerStore = {
+      getTaskWorkflowSelection: () => ({ workflowId: "WF-404", stepIds: [] }),
+      getWorkflowDefinition: async () => undefined,
+    };
+    const runner = new WorkflowGraphTaskRunner({
+      store,
+      seams: recordingSeams([]),
+      runCustomNode: async () => ({ outcome: "success" }),
+    });
+    const result = await runner.run(task, flagOn);
+    expect(result.disposition).toBe("fell-back");
+    expect(result.reason).toMatch(/workflow-missing/);
+  });
+
+  it("falls back (never strands the task) when the interpreter throws", async () => {
+    // Malformed graph: edge references unknown node → WorkflowIrError inside run().
+    const badIr: WorkflowIr = {
+      version: "v1",
+      name: "bad",
+      nodes: [
+        { id: "start", kind: "start" },
+        { id: "end", kind: "end" },
+      ],
+      edges: [{ from: "start", to: "ghost" }],
+    };
+    const events: string[] = [];
+    const runner = new WorkflowGraphTaskRunner({
+      store: storeWith(definition(badIr)),
+      seams: recordingSeams([]),
+      runCustomNode: async () => ({ outcome: "success" }),
+      onEvent: (e) => events.push(e.type),
+    });
+    const result = await runner.run(task, flagOn);
+    expect(result.disposition).toBe("fell-back");
+    expect(result.reason).toMatch(/interpreter-error/);
+    expect(events).toContain("fallback");
+  });
+
+  it("exposes node outcomes in the shared context for downstream consumers", async () => {
+    const runner = new WorkflowGraphTaskRunner({
+      store: storeWith(definition(fullLifecycleIr())),
+      seams: recordingSeams([]),
+      runCustomNode: async () => ({ outcome: "success", value: "APPROVE" }),
+    });
+    const result = await runner.run(task, flagOn);
+    expect(result.context?.["node:lint:outcome"]).toBe("success");
+    expect(result.context?.["node:lint:value"]).toBe("APPROVE");
+  });
+
+  it("onEvent diagnostics failures never affect the run", async () => {
+    const runner = new WorkflowGraphTaskRunner({
+      store: storeWith(definition(fullLifecycleIr())),
+      seams: recordingSeams([]),
+      runCustomNode: async () => ({ outcome: "success" }),
+      onEvent: () => {
+        throw new Error("diagnostics boom");
+      },
+    });
+    const result = await runner.run(task, flagOn);
+    expect(result.disposition).toBe("completed");
+  });
+});
diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts
index 71bed8fbb2..cd827692d9 100644
--- a/packages/engine/src/index.ts
+++ b/packages/engine/src/index.ts
@@ -24,9 +24,17 @@ export {
 export {
   createDefaultNodeHandlers,
   createNoopLegacySeams,
+  type WorkflowCustomNodeRunner,
   type WorkflowLegacySeams,
   type WorkflowSeamName,
 } from "./workflow-node-handlers.js";
+export {
+  WorkflowGraphTaskRunner,
+  type WorkflowGraphRunDisposition,
+  type WorkflowGraphRunnerStore,
+  type WorkflowGraphTaskRunResult,
+  type WorkflowGraphTaskRunnerDeps,
+} from "./workflow-graph-task-runner.js";
 export { collectTaskEvaluationEvidence } from "./evaluator-evidence.js";
 export { Scheduler, type SchedulerOptions } from "./scheduler.js";
 export { MeshLeaseManager, type MeshLeaseManagerOptions, type LeaseRecoveryContext } from "./mesh-lease-manager.js";
diff --git a/packages/engine/src/workflow-graph-task-runner.ts b/packages/engine/src/workflow-graph-task-runner.ts
new file mode 100644
index 0000000000..8cde380019
--- /dev/null
+++ b/packages/engine/src/workflow-graph-task-runner.ts
@@ -0,0 +1,117 @@
+import type { Settings, TaskDetail, WorkflowDefinition } from "@fusion/core";
+import { isExperimentalFeatureEnabled } from "@fusion/core";
+
+import { WorkflowGraphExecutor, type WorkflowNodeOutcome } from "./workflow-graph-executor.js";
+import type { WorkflowCustomNodeRunner, WorkflowLegacySeams } from "./workflow-node-handlers.js";
+
+/**
+ * Terminal disposition of an interpreter-driven task run.
+ * - "completed"  — the graph ran to its end node successfully.
+ * - "failed"     — the graph ran and terminated on a failure outcome.
+ * - "fell-back"  — the interpreter did not (or could not) own this task;
+ *                  the caller must run the legacy pipeline instead.
+ */
+export type WorkflowGraphRunDisposition = "completed" | "failed" | "fell-back";
+
+export interface WorkflowGraphTaskRunResult {
+  disposition: WorkflowGraphRunDisposition;
+  outcome?: WorkflowNodeOutcome;
+  visitedNodeIds: string[];
+  /** Why the runner fell back (flag-off, no-selection, workflow-missing, interpreter-error). */
+  reason?: string;
+  /** Shared graph context after the run (node outcomes/values). */
+  context?: Record<string, unknown>;
+}
+
+/** The minimal store surface the runner needs — keeps tests fake-friendly. */
+export interface WorkflowGraphRunnerStore {
+  getTaskWorkflowSelection(taskId: string): { workflowId: string; stepIds: string[] } | undefined;
+  getWorkflowDefinition(id: string): Promise<WorkflowDefinition | undefined>;
+}
+
+export interface WorkflowGraphTaskRunnerDeps {
+  store: WorkflowGraphRunnerStore;
+  seams: WorkflowLegacySeams;
+  runCustomNode: WorkflowCustomNodeRunner;
+  maxRetriesPerNode?: number;
+  /** Optional diagnostics hook (audit/log emission). Never throws into the run. */
+  onEvent?: (event: { type: "start" | "terminal" | "fallback"; taskId: string; detail: string }) => void;
+}
+
+/**
+ * Drives a task's lifecycle from its selected workflow graph. The runner owns
+ * SEQUENCING only — seam nodes delegate to the legacy engine implementations
+ * (execute/review/merge), custom nodes run via the injected runner. Any
+ * interpreter-level error yields a "fell-back" disposition so the caller can
+ * run the legacy pipeline; a task is never stranded by interpreter bugs.
+ */
+export class WorkflowGraphTaskRunner {
+  public constructor(private readonly deps: WorkflowGraphTaskRunnerDeps) {}
+
+  private emit(type: "start" | "terminal" | "fallback", taskId: string, detail: string): void {
+    try {
+      this.deps.onEvent?.({ type, taskId, detail });
+    } catch {
+      // Diagnostics must never affect the run.
+    }
+  }
+
+  private fallBack(taskId: string, reason: string): WorkflowGraphTaskRunResult {
+    this.emit("fallback", taskId, reason);
+    return { disposition: "fell-back", reason, visitedNodeIds: [] };
+  }
+
+  public async run(
+    task: TaskDetail,
+    settings: Pick<Settings, "experimentalFeatures"> | undefined,
+  ): Promise<WorkflowGraphTaskRunResult> {
+    if (!isExperimentalFeatureEnabled(settings, "workflowGraphExecutor")) {
+      return this.fallBack(task.id, "flag-off");
+    }
+
+    let selection: { workflowId: string; stepIds: string[] } | undefined;
+    try {
+      selection = this.deps.store.getTaskWorkflowSelection(task.id);
+    } catch (err) {
+      return this.fallBack(task.id, `selection-error: ${err instanceof Error ? err.message : String(err)}`);
+    }
+    if (!selection) {
+      return this.fallBack(task.id, "no-selection");
+    }
+
+    let definition: WorkflowDefinition | undefined;
+    try {
+      definition = await this.deps.store.getWorkflowDefinition(selection.workflowId);
+    } catch (err) {
+      return this.fallBack(task.id, `workflow-load-error: ${err instanceof Error ? err.message : String(err)}`);
+    }
+    if (!definition) {
+      return this.fallBack(task.id, `workflow-missing: ${selection.workflowId}`);
+    }
+
+    this.emit("start", task.id, definition.id);
+    try {
+      const executor = new WorkflowGraphExecutor({
+        seams: this.deps.seams,
+        runCustomNode: this.deps.runCustomNode,
+        maxRetriesPerNode: this.deps.maxRetriesPerNode,
+      });
+      const result = await executor.run(task, settings, definition.ir);
+      if (!result.executed) {
+        return this.fallBack(task.id, "not-executed");
+      }
+      const disposition: WorkflowGraphRunDisposition = result.outcome === "success" ? "completed" : "failed";
+      this.emit("terminal", task.id, `${definition.id}:${disposition}`);
+      return {
+        disposition,
+        outcome: result.outcome,
+        visitedNodeIds: result.visitedNodeIds,
+        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)}`);
+    }
+  }
+}