feat(engine): fan-out/join branch execution with crash-recoverable branch state, schema v107 (U13)

packages/engine/src/__tests__/workflow-graph-fanout.test.ts

@@ -0,0 +1,360 @@
+import { describe, expect, it, vi } from "vitest";
+import type { TaskDetail, WorkflowIr, WorkflowIrNode } from "@fusion/core";
+
+import { WorkflowGraphExecutor, type WorkflowNodeHandler } from "../workflow-graph-executor.js";
+import type {
+  WorkflowBranchPersistence,
+  WorkflowBranchProgress,
+  WorkflowBranchRunState,
+} from "../workflow-graph-branches.js";
+
+const task = { id: "FN-FANOUT" } as TaskDetail;
+const settingsOn = () => ({ experimentalFeatures: { workflowGraphExecutor: true } });
+
+/** A controllable deferred so branches can complete in any order under test control. */
+function deferred<T>() {
+  let resolve!: (v: T) => void;
+  let reject!: (e: unknown) => void;
+  const promise = new Promise<T>((res, rej) => {
+    resolve = res;
+    reject = rej;
+  });
+  return { promise, resolve, reject };
+}
+
+/** start → split → (branchA → branchB) → join → tail → end */
+function twoBranchIr(joinConfig: Record<string, unknown>): WorkflowIr {
+  return {
+    version: "v2",
+    name: "two-branch",
+    columns: [{ id: "work", name: "Work", traits: [] }],
+    nodes: [
+      { id: "start", kind: "start" },
+      { id: "split", kind: "split", column: "work" },
+      { id: "branchA", kind: "prompt", column: "work", config: { prompt: "a" } },
+      { id: "branchB", kind: "prompt", column: "work", config: { prompt: "b" } },
+      { id: "join", kind: "join", column: "work", config: joinConfig },
+      { id: "tail", kind: "prompt", column: "work", config: { prompt: "tail" } },
+      { id: "end", kind: "end" },
+    ],
+    edges: [
+      { from: "start", to: "split" },
+      { from: "split", to: "branchA" },
+      { from: "split", to: "branchB" },
+      { from: "branchA", to: "join", condition: "success" },
+      { from: "branchB", to: "join", condition: "success" },
+      { from: "join", to: "tail", condition: "success" },
+      { from: "join", to: "end", condition: "failure" },
+      { from: "tail", to: "end", condition: "success" },
+    ],
+  };
+}
+
+describe("WorkflowGraphExecutor fan-out/join (U13)", () => {
+  it("mode:all — both branches complete in any order, join fires once, advances to tail", async () => {
+    const a = deferred<void>();
+    const b = deferred<void>();
+    const tail = vi.fn(async () => ({ outcome: "success" as const }));
+    const prompt: WorkflowNodeHandler = async (node) => {
+      if (node.id === "branchA") await a.promise;
+      if (node.id === "branchB") await b.promise;
+      if (node.id === "tail") return tail();
+      return { outcome: "success" as const };
+    };
+    const executor = new WorkflowGraphExecutor({ handlers: { prompt } });
+    const run = executor.run(task, settingsOn(), twoBranchIr({ mode: "all" }));
+
+    // Complete in reverse order to prove order-independence.
+    b.resolve();
+    await Promise.resolve();
+    expect(tail).not.toHaveBeenCalled();
+    a.resolve();
+
+    const result = await run;
+    expect(result.outcome).toBe("success");
+    expect(result.visitedNodeIds).toContain("tail");
+    expect(tail).toHaveBeenCalledTimes(1);
+  });
+
+  it("mode:any with collect — first completion fires join; slower branch finishes without re-firing", async () => {
+    const slow = deferred<void>();
+    const tail = vi.fn(async () => ({ outcome: "success" as const }));
+    let slowFinished = false;
+    const prompt: WorkflowNodeHandler = async (node) => {
+      if (node.id === "branchB") {
+        await slow.promise;
+        slowFinished = true;
+      }
+      if (node.id === "tail") return tail();
+      return { outcome: "success" as const };
+    };
+    const executor = new WorkflowGraphExecutor({ handlers: { prompt } });
+    const run = executor.run(task, settingsOn(), twoBranchIr({ mode: "any", onBranchFailure: "collect" }));
+
+    // branchA resolves immediately → join fires. tail must run exactly once.
+    await Promise.resolve();
+    slow.resolve();
+    const result = await run;
+
+    expect(result.outcome).toBe("success");
+    expect(tail).toHaveBeenCalledTimes(1);
+    expect(slowFinished).toBe(true);
+  });
+
+  it("mode:any with fail-fast — slower branch is aborted via signal", async () => {
+    let aborted = false;
+    const slow = deferred<void>();
+    const prompt: WorkflowNodeHandler = async (node, ctx) => {
+      if (node.id === "branchB") {
+        ctx.signal?.addEventListener("abort", () => {
+          aborted = true;
+          slow.resolve();
+        });
+        await slow.promise;
+        if (ctx.signal?.aborted) return { outcome: "failure" as const, value: "aborted" };
+      }
+      return { outcome: "success" as const };
+    };
+    const executor = new WorkflowGraphExecutor({ handlers: { prompt } });
+    const result = await executor.run(task, settingsOn(), twoBranchIr({ mode: "any", onBranchFailure: "fail-fast" }));
+
+    expect(result.outcome).toBe("success");
+    expect(aborted).toBe(true);
+  });
+
+  it("quorum(2) of 3 — join fires on the second completion", async () => {
+    const ir: WorkflowIr = {
+      version: "v2",
+      name: "quorum",
+      columns: [{ id: "w", name: "W", traits: [] }],
+      nodes: [
+        { id: "start", kind: "start" },
+        { id: "split", kind: "split" },
+        { id: "b1", kind: "prompt", config: {} },
+        { id: "b2", kind: "prompt", config: {} },
+        { id: "b3", kind: "prompt", config: {} },
+        { id: "join", kind: "join", config: { mode: { quorum: 2 }, onBranchFailure: "collect" } },
+        { id: "tail", kind: "prompt", config: {} },
+        { id: "end", kind: "end" },
+      ],
+      edges: [
+        { from: "start", to: "split" },
+        { from: "split", to: "b1" },
+        { from: "split", to: "b2" },
+        { from: "split", to: "b3" },
+        { from: "b1", to: "join", condition: "success" },
+        { from: "b2", to: "join", condition: "success" },
+        { from: "b3", to: "join", condition: "success" },
+        { from: "join", to: "tail", condition: "success" },
+        { from: "join", to: "end", condition: "failure" },
+        { from: "tail", to: "end" },
+      ],
+    };
+    const d3 = deferred<void>();
+    const tail = vi.fn(async () => ({ outcome: "success" as const }));
+    const prompt: WorkflowNodeHandler = async (node) => {
+      if (node.id === "b3") await d3.promise;
+      if (node.id === "tail") return tail();
+      return { outcome: "success" as const };
+    };
+    const executor = new WorkflowGraphExecutor({ handlers: { prompt } });
+    const run = executor.run(task, settingsOn(), ir);
+    // b1 + b2 resolve immediately → quorum(2) satisfied without b3.
+    await Promise.resolve();
+    d3.resolve();
+    const result = await run;
+    expect(result.outcome).toBe("success");
+    expect(tail).toHaveBeenCalledTimes(1);
+  });
+
+  it("branch failure fail-fast — siblings aborted, join routes the failure edge", async () => {
+    let siblingAborted = false;
+    const slow = deferred<void>();
+    const prompt: WorkflowNodeHandler = async (node, ctx) => {
+      if (node.id === "branchA") return { outcome: "failure" as const, value: "boom" };
+      if (node.id === "branchB") {
+        ctx.signal?.addEventListener("abort", () => {
+          siblingAborted = true;
+          slow.resolve();
+        });
+        await slow.promise;
+        return { outcome: "success" as const };
+      }
+      return { outcome: "success" as const };
+    };
+    const executor = new WorkflowGraphExecutor({ handlers: { prompt } });
+    const result = await executor.run(task, settingsOn(), twoBranchIr({ mode: "all", onBranchFailure: "fail-fast" }));
+
+    expect(result.outcome).toBe("failure");
+    expect(result.visitedNodeIds).not.toContain("tail");
+    expect(siblingAborted).toBe(true);
+  });
+
+  it("branch failure collect — all branches finish; join evaluates combined outcomes", async () => {
+    const calls: string[] = [];
+    const prompt: WorkflowNodeHandler = async (node) => {
+      calls.push(node.id);
+      if (node.id === "branchA") return { outcome: "failure" as const, value: "boom" };
+      return { outcome: "success" as const };
+    };
+    const executor = new WorkflowGraphExecutor({ handlers: { prompt } });
+    const result = await executor.run(task, settingsOn(), twoBranchIr({ mode: "all", onBranchFailure: "collect" }));
+
+    // mode:all unmet (one failed) → join outcome failure; both branches ran.
+    expect(result.outcome).toBe("failure");
+    expect(calls).toContain("branchA");
+    expect(calls).toContain("branchB");
+    const branchOutcomes = result.context["node:join:branchOutcomes"] as { outcome: string }[];
+    expect(branchOutcomes.some((b) => b.outcome === "failure")).toBe(true);
+    expect(branchOutcomes.some((b) => b.outcome === "success")).toBe(true);
+  });
+
+  it("crash mid-branch resume — completed branches' nodes are NOT re-run", async () => {
+    const calls: string[] = [];
+    const store: WorkflowBranchRunState[] = [];
+    const persistence: WorkflowBranchPersistence = {
+      saveBranchState: (s) => {
+        const idx = store.findIndex((e) => e.branchId === s.branchId);
+        if (idx >= 0) store[idx] = s;
+        else store.push({ ...s });
+      },
+      loadBranchStates: () => store.map((s) => ({ ...s })),
+    };
+
+    // First run: branchA completes, branchB hangs (simulated crash before join).
+    const hang = deferred<void>();
+    const aPersisted = deferred<void>();
+    const persistenceA: WorkflowBranchPersistence = {
+      saveBranchState: (s) => {
+        persistence.saveBranchState!(s);
+        if (s.branchId === "branchA" && s.status === "completed") aPersisted.resolve();
+      },
+      loadBranchStates: persistence.loadBranchStates,
+    };
+    const prompt1: WorkflowNodeHandler = async (node) => {
+      calls.push(`run1:${node.id}`);
+      if (node.id === "branchB") await hang.promise; // never resolves this run
+      return { outcome: "success" as const };
+    };
+    const exec1 = new WorkflowGraphExecutor({ handlers: { prompt: prompt1 }, branchPersistence: persistenceA });
+    const run1 = exec1.run(task, settingsOn(), twoBranchIr({ mode: "all" }));
+    await aPersisted.promise;
+    // Don't await run1 (branchB stuck) — simulate process death by starting fresh.
+
+    expect(store.find((s) => s.branchId === "branchA")?.status).toBe("completed");
+
+    // Resume: a brand-new executor reconstructed from persisted rows.
+    const prompt2: WorkflowNodeHandler = async (node) => {
+      calls.push(`run2:${node.id}`);
+      return { outcome: "success" as const };
+    };
+    const exec2 = new WorkflowGraphExecutor({ handlers: { prompt: prompt2 }, branchPersistence: persistence });
+    const result = await exec2.run(task, settingsOn(), twoBranchIr({ mode: "all" }));
+
+    expect(result.outcome).toBe("success");
+    // branchA already completed → not re-run on resume.
+    expect(calls).not.toContain("run2:branchA");
+    // branchB re-runs (it never completed).
+    expect(calls).toContain("run2:branchB");
+    hang.resolve();
+    await run1.catch(() => {});
+  });
+
+  it("card-position invariant — no column move occurs during the parallel window", async () => {
+    // The executor never touches task.column; assert the handler context exposes
+    // the split's column to all branch nodes and the task object is untouched.
+    const columnsSeen = new Set<string | undefined>();
+    const taskColumnBefore = (task as { column?: string }).column;
+    const prompt: WorkflowNodeHandler = async (node) => {
+      if (node.id.startsWith("branch")) columnsSeen.add(node.column);
+      return { outcome: "success" as const };
+    };
+    const executor = new WorkflowGraphExecutor({ handlers: { prompt } });
+    await executor.run(task, settingsOn(), twoBranchIr({ mode: "all" }));
+    // Branch nodes live in the split's column; task position never forked.
+    expect(columnsSeen).toEqual(new Set(["work"]));
+    expect((task as { column?: string }).column).toBe(taskColumnBefore);
+  });
+
+  it("semaphore bound — branches queue, never exceeding the limit (fake semaphore)", async () => {
+    let active = 0;
+    let peak = 0;
+    const limit = 1;
+    const queue: (() => void)[] = [];
+    const fakeSemaphore = {
+      async run<T>(fn: () => Promise<T>): Promise<T> {
+        if (active >= limit) await new Promise<void>((res) => queue.push(res));
+        active += 1;
+        peak = Math.max(peak, active);
+        try {
+          return await fn();
+        } finally {
+          active -= 1;
+          queue.shift()?.();
+        }
+      },
+    };
+    const executor = new WorkflowGraphExecutor({
+      handlers: { prompt: async () => ({ outcome: "success" as const }) },
+      branchSemaphore: fakeSemaphore,
+    });
+    const result = await executor.run(task, settingsOn(), twoBranchIr({ mode: "all" }));
+    expect(result.outcome).toBe("success");
+    expect(peak).toBeLessThanOrEqual(limit);
+  });
+
+  it("nested split resolves recursively", async () => {
+    // start → split(outer) → [ branchX | split(inner) → [i1 | i2] → joinInner ] → joinOuter → end
+    const ir: WorkflowIr = {
+      version: "v2",
+      name: "nested",
+      columns: [{ id: "w", name: "W", traits: [] }],
+      nodes: [
+        { id: "start", kind: "start" },
+        { id: "outer", kind: "split" },
+        { id: "branchX", kind: "prompt", config: {} },
+        { id: "inner", kind: "split" },
+        { id: "i1", kind: "prompt", config: {} },
+        { id: "i2", kind: "prompt", config: {} },
+        { id: "joinInner", kind: "join", config: { mode: "all" } },
+        { id: "joinOuter", kind: "join", config: { mode: "all" } },
+        { id: "end", kind: "end" },
+      ],
+      edges: [
+        { from: "start", to: "outer" },
+        { from: "outer", to: "branchX" },
+        { from: "outer", to: "inner" },
+        { from: "branchX", to: "joinOuter", condition: "success" },
+        { from: "inner", to: "i1" },
+        { from: "inner", to: "i2" },
+        { from: "i1", to: "joinInner", condition: "success" },
+        { from: "i2", to: "joinInner", condition: "success" },
+        { from: "joinInner", to: "joinOuter", condition: "success" },
+        { from: "joinOuter", to: "end", condition: "success" },
+      ],
+    };
+    const calls: string[] = [];
+    const executor = new WorkflowGraphExecutor({
+      handlers: {
+        prompt: async (node) => {
+          calls.push(node.id);
+          return { outcome: "success" as const };
+        },
+      },
+    });
+    const result = await executor.run(task, settingsOn(), ir);
+    expect(result.outcome).toBe("success");
+    expect(calls).toEqual(expect.arrayContaining(["branchX", "i1", "i2"]));
+  });
+
+  it("reports live per-branch progress for the dashboard", async () => {
+    const progress: WorkflowBranchProgress[] = [];
+    const executor = new WorkflowGraphExecutor({
+      handlers: { prompt: async () => ({ outcome: "success" as const }) },
+      onBranchProgress: (p) => progress.push(p),
+    });
+    await executor.run(task, settingsOn(), twoBranchIr({ mode: "all" }));
+    expect(progress.some((p) => p.branchId === "branchA" && p.status === "completed")).toBe(true);
+    expect(progress.some((p) => p.branchId === "branchB" && p.status === "completed")).toBe(true);
+  });
+});

packages/engine/src/index.ts

@@ -25,6 +25,13 @@ export {
   type WorkflowGraphExecutorDeps,
   type WorkflowGraphExecutorResult,
 } from "./workflow-graph-executor.js";
+export {
+  runSplitJoin,
+  type WorkflowBranchPersistence,
+  type WorkflowBranchProgress,
+  type WorkflowBranchRunState,
+  type WorkflowBranchSemaphore,
+} from "./workflow-graph-branches.js";
 export {
   createDefaultNodeHandlers,
   createNoopLegacySeams,
@@ -63,6 +70,13 @@ export {
   getConflictedFiles,
   type AutostashHandle,
 } from "./merger.js";
+export {
+  registerMergeTraitHooks,
+  resolveMergePolicy,
+  type ResolvedMergePolicy,
+  type MergeFileScopeMode,
+  type MergeTraitStrategy,
+} from "./merge-trait.js";
 export {
   resolveIntegrationBranch,
   resolveIntegrationBranchSync,

packages/engine/src/workflow-graph-branches.ts

@@ -0,0 +1,337 @@
+import type { Settings, TaskDetail, WorkflowIrEdge, WorkflowIrNode } from "@fusion/core";
+import { WorkflowIrError } from "@fusion/core";
+
+import type { WorkflowNodeOutcome, WorkflowNodeResult } from "./workflow-graph-executor.js";
+
+/**
+ * Concurrent fan-out/join branch execution (U13, KTD-11, R21).
+ *
+ * When the sequential walker reaches a `split` node, every outgoing edge becomes
+ * a branch that walks concurrently up to the matching `join`. The join then
+ * synchronizes per its config (`all | any | quorum(n)`) and either fails fast
+ * (cancelling siblings via an AbortSignal) or collects all branch outcomes.
+ *
+ * This module owns ONLY the parallel window: it is handed a `runBranchNode`
+ * callback that reuses the executor's per-node retry logic, and a small set of
+ * graph lookups. The card's board position never forks — that invariant is
+ * upheld by the executor (no handler-driven column moves happen here).
+ */
+
+/** Per-branch persisted run state (ADR-0001 reconstructible). */
+export interface WorkflowBranchRunState {
+  taskId: string;
+  runId: string;
+  branchId: string;
+  /** Node the branch is currently at / last completed. */
+  currentNodeId: string;
+  status: "running" | "completed" | "failed" | "aborted";
+}
+
+/**
+ * Persistence callback surface. Kept as an injected interface so the executor
+ * stays DI-pure and fake-friendly; the SQLite-backed implementation is wired
+ * separately (see workflow_run_branches table). All methods are optional so a
+ * fully in-memory run (tests, flag-off) needs no persistence at all.
+ */
+export interface WorkflowBranchPersistence {
+  /** Idempotent upsert of a branch's progress, keyed by (taskId, runId, branchId). */
+  saveBranchState?(state: WorkflowBranchRunState): void | Promise<void>;
+  /** Load any persisted branch states for a run (used on resume). */
+  loadBranchStates?(taskId: string, runId: string): WorkflowBranchRunState[] | Promise<WorkflowBranchRunState[]>;
+}
+
+/** Minimal semaphore shape — structurally compatible with AgentSemaphore. */
+export interface WorkflowBranchSemaphore {
+  run<T>(fn: () => Promise<T>): Promise<T>;
+}
+
+/** Snapshot of a single branch's progress, surfaced for dashboard badges (U9). */
+export interface WorkflowBranchProgress {
+  branchId: string;
+  nodeId: string;
+  status: WorkflowBranchRunState["status"];
+}
+
+export interface BranchEnvironment {
+  task: TaskDetail;
+  settings: Pick<Settings, "experimentalFeatures"> | undefined;
+  runId: string;
+  nodeMap: Map<string, WorkflowIrNode>;
+  outgoingMap: Map<string, WorkflowIrEdge[]>;
+  /** Reuses the executor's executeNodeWithRetries (+ context bookkeeping). */
+  runBranchNode: (
+    node: WorkflowIrNode,
+    signal: AbortSignal,
+  ) => Promise<WorkflowNodeResult>;
+  shouldTraverseEdge: (edge: WorkflowIrEdge, source: WorkflowNodeResult) => boolean;
+  persistence?: WorkflowBranchPersistence;
+  semaphore?: WorkflowBranchSemaphore;
+  /** Reports live per-branch progress for the card (no column move). */
+  onBranchProgress?: (progress: WorkflowBranchProgress) => void;
+  /** Node IDs already completed in a prior (crashed) run — skipped on resume. */
+  completedNodeIds?: Set<string>;
+}
+
+export interface SplitJoinResult {
+  /** The join node where branches converged. */
+  joinNodeId: string;
+  /** Join outcome: success if the mode was satisfied, else failure. */
+  outcome: WorkflowNodeOutcome;
+  /** Per-branch outcomes, exposed so the join's outgoing edge conditions can read them. */
+  branchOutcomes: { branchId: string; outcome: WorkflowNodeOutcome; nodeId: string }[];
+  /** Node IDs visited across all branches (for the executor's visited list). */
+  visitedNodeIds: string[];
+}
+
+interface ResolvedJoinConfig {
+  mode: "all" | "any" | { quorum: number };
+  onBranchFailure: "fail-fast" | "collect";
+}
+
+function resolveJoinConfig(join: WorkflowIrNode): ResolvedJoinConfig {
+  const rawMode = join.config?.mode;
+  let mode: ResolvedJoinConfig["mode"] = "all";
+  if (rawMode === "all" || rawMode === "any") mode = rawMode;
+  else if (rawMode && typeof rawMode === "object" && "quorum" in rawMode) {
+    const n = (rawMode as { quorum: unknown }).quorum;
+    if (typeof n === "number" && Number.isInteger(n) && n > 0) mode = { quorum: n };
+    else throw new WorkflowIrError(`join '${join.id}' quorum must be a positive integer`);
+  }
+  const rawFail = join.config?.onBranchFailure;
+  const onBranchFailure: ResolvedJoinConfig["onBranchFailure"] =
+    rawFail === "collect" ? "collect" : "fail-fast";
+  return { mode, onBranchFailure };
+}
+
+/** How many successful completions satisfy this join mode for `branchCount` branches. */
+function requiredCompletions(mode: ResolvedJoinConfig["mode"], branchCount: number): number {
+  if (mode === "all") return branchCount;
+  if (mode === "any") return 1;
+  return Math.min(mode.quorum, branchCount);
+}
+
+/**
+ * Execute a split's branches concurrently and synchronize at the join.
+ *
+ * Returns once the join is satisfied (or definitively cannot be). Sibling
+ * branches are aborted on fail-fast via the shared AbortSignal; on collect they
+ * are awaited. Nested splits recurse: a branch walk that itself hits a `split`
+ * calls back into this function for its inner window.
+ */
+export async function runSplitJoin(
+  split: WorkflowIrNode,
+  env: BranchEnvironment,
+): Promise<SplitJoinResult> {
+  const branchEdges = (env.outgoingMap.get(split.id) ?? []).filter(
+    (e) => env.shouldTraverseEdge(e, { outcome: "success" }),
+  );
+  if (branchEdges.length === 0) {
+    throw new WorkflowIrError(`split '${split.id}' has no traversable branches`);
+  }
+
+  const join = findMatchingJoin(branchEdges[0].to, env);
+  if (!join) throw new WorkflowIrError(`split '${split.id}' has no reachable matching join`);
+  const joinConfig = resolveJoinConfig(env.nodeMap.get(join)!);
+
+  const controller = new AbortController();
+  const branchCount = branchEdges.length;
+  const required = requiredCompletions(joinConfig.mode, branchCount);
+
+  const visitedNodeIds: string[] = [];
+  const branchOutcomes: SplitJoinResult["branchOutcomes"] = [];
+  let succeeded = 0;
+  let failed = 0;
+  let settled = false;
+  let resolveJoin!: (outcome: WorkflowNodeOutcome) => void;
+  const joinReached = new Promise<WorkflowNodeOutcome>((res) => {
+    resolveJoin = res;
+  });
+
+  const settle = (outcome: WorkflowNodeOutcome): void => {
+    if (settled) return;
+    settled = true;
+    resolveJoin(outcome);
+  };
+
+  // Re-evaluate the join after each branch settles. `lastWasFailure` only
+  // affects fail-fast (one failure cancels siblings immediately).
+  const evaluateJoin = (lastWasFailure: boolean): void => {
+    if (settled) return;
+    if (joinConfig.onBranchFailure === "fail-fast" && lastWasFailure) {
+      controller.abort();
+      settle("failure");
+      return;
+    }
+    if (succeeded >= required) {
+      // Mode satisfied. Fail-fast cancels any laggards; collect lets them finish.
+      if (joinConfig.onBranchFailure === "fail-fast") controller.abort();
+      settle("success");
+      return;
+    }
+    if (succeeded + failed >= branchCount) {
+      // All branches settled but the mode is unmet.
+      settle("failure");
+    }
+  };
+
+  const branchPromises = branchEdges.map((edge) => {
+    const branchId = edge.to;
+    return walkBranch(branchId, join, env, controller.signal, visitedNodeIds)
+      .then((result) => {
+        branchOutcomes.push({ branchId, outcome: result.outcome, nodeId: result.lastNodeId });
+        if (result.outcome === "success") succeeded += 1;
+        else failed += 1;
+        evaluateJoin(result.outcome === "failure");
+      })
+      .catch((err) => {
+        // An aborted branch settles silently; any other throw fails the join.
+        if (controller.signal.aborted) {
+          branchOutcomes.push({ branchId, outcome: "failure", nodeId: branchId });
+          return;
+        }
+        failed += 1;
+        branchOutcomes.push({ branchId, outcome: "failure", nodeId: branchId });
+        evaluateJoin(true);
+        void err;
+      });
+  });
+
+  // Wait for the join to resolve, then let in-flight branches settle so collect
+  // semantics (and persistence writes) complete before we return.
+  const outcome = await joinReached;
+  await Promise.allSettled(branchPromises);
+
+  return { joinNodeId: join, outcome, branchOutcomes, visitedNodeIds };
+}
+
+interface BranchWalkResult {
+  outcome: WorkflowNodeOutcome;
+  lastNodeId: string;
+}
+
+/**
+ * Walk a single branch from `startNodeId` up to (but not including) the join.
+ * Reuses the injected per-node runner; supports nested splits by recursing into
+ * runSplitJoin. Honors the AbortSignal (fail-fast cancellation) and skips nodes
+ * already completed in a prior run (crash resume idempotency).
+ */
+async function walkBranch(
+  startNodeId: string,
+  joinId: string,
+  env: BranchEnvironment,
+  signal: AbortSignal,
+  visitedNodeIds: string[],
+): Promise<BranchWalkResult> {
+  let currentId = startNodeId;
+  let lastResult: WorkflowNodeResult = { outcome: "success" };
+
+  for (;;) {
+    if (signal.aborted) return { outcome: "failure", lastNodeId: currentId };
+    if (currentId === joinId) return { outcome: lastResult.outcome, lastNodeId: currentId };
+
+    const node = env.nodeMap.get(currentId);
+    if (!node) throw new WorkflowIrError(`Unknown workflow node: ${currentId}`);
+
+    if (node.kind === "split") {
+      // Nested split: resolve its inner window, then continue from the inner join.
+      const inner = await runSplitJoin(node, env);
+      visitedNodeIds.push(...inner.visitedNodeIds);
+      lastResult = { outcome: inner.outcome };
+      const next = nextEdge(inner.joinNodeId, env, lastResult);
+      if (!next) return { outcome: inner.outcome, lastNodeId: inner.joinNodeId };
+      currentId = next;
+      continue;
+    }
+
+    visitedNodeIds.push(currentId);
+
+    const alreadyDone = env.completedNodeIds?.has(currentId) ?? false;
+    if (alreadyDone) {
+      lastResult = { outcome: "success" };
+    } else {
+      const exec = async (): Promise<WorkflowNodeResult> => env.runBranchNode(node, signal);
+      lastResult = env.semaphore ? await env.semaphore.run(exec) : await exec();
+      env.persistence?.saveBranchState?.({
+        taskId: env.task.id,
+        runId: env.runId,
+        branchId: startNodeId,
+        currentNodeId: currentId,
+        status: lastResult.outcome === "success" ? "running" : "failed",
+      });
+      env.onBranchProgress?.({
+        branchId: startNodeId,
+        nodeId: currentId,
+        status: lastResult.outcome === "success" ? "running" : "failed",
+      });
+    }
+
+    if (lastResult.outcome === "failure") {
+      env.persistence?.saveBranchState?.({
+        taskId: env.task.id,
+        runId: env.runId,
+        branchId: startNodeId,
+        currentNodeId: currentId,
+        status: "failed",
+      });
+      return { outcome: "failure", lastNodeId: currentId };
+    }
+
+    const next = nextEdge(currentId, env, lastResult);
+    if (!next) {
+      // Dead-end before the join — treat as branch completion.
+      return { outcome: lastResult.outcome, lastNodeId: currentId };
+    }
+    if (next === joinId) {
+      env.persistence?.saveBranchState?.({
+        taskId: env.task.id,
+        runId: env.runId,
+        branchId: startNodeId,
+        currentNodeId: currentId,
+        status: "completed",
+      });
+      env.onBranchProgress?.({ branchId: startNodeId, nodeId: currentId, status: "completed" });
+      return { outcome: lastResult.outcome, lastNodeId: currentId };
+    }
+    currentId = next;
+  }
+}
+
+/** The next node along a matching outgoing edge, or undefined if none matches. */
+function nextEdge(
+  nodeId: string,
+  env: BranchEnvironment,
+  source: WorkflowNodeResult,
+): string | undefined {
+  const edges = (env.outgoingMap.get(nodeId) ?? [])
+    .filter((e) => env.shouldTraverseEdge(e, source))
+    .sort((a, b) => a.to.localeCompare(b.to));
+  return edges[0]?.to;
+}
+
+/**
+ * Find the join node a branch starting at `startNodeId` converges on. Walks
+ * forward through the (non-failure) edges; recurses one level for nested splits
+ * so balanced nesting resolves to the correct outer join.
+ */
+function findMatchingJoin(startNodeId: string, env: BranchEnvironment): string | undefined {
+  const seen = new Set<string>();
+  let currentId: string | undefined = startNodeId;
+  while (currentId && !seen.has(currentId)) {
+    seen.add(currentId);
+    const node = env.nodeMap.get(currentId);
+    if (!node) return undefined;
+    if (node.kind === "join") return currentId;
+    if (node.kind === "split") {
+      const innerJoin = findMatchingJoin(
+        (env.outgoingMap.get(currentId) ?? [])[0]?.to ?? "",
+        env,
+      );
+      if (!innerJoin) return undefined;
+      currentId = (env.outgoingMap.get(innerJoin) ?? []).find((e) => e.condition !== "failure")?.to;
+      continue;
+    }
+    const out = env.outgoingMap.get(currentId) ?? [];
+    currentId = out.find((e) => e.condition !== "failure")?.to ?? out[0]?.to;
+  }
+  return undefined;
+}

packages/engine/src/workflow-graph-executor.ts

@@ -7,6 +7,14 @@ import {
   type WorkflowCustomNodeRunner,
   type WorkflowLegacySeams,
 } from "./workflow-node-handlers.js";
+import {
+  runSplitJoin,
+  type BranchEnvironment,
+  type WorkflowBranchPersistence,
+  type WorkflowBranchProgress,
+  type WorkflowBranchRunState,
+  type WorkflowBranchSemaphore,
+} from "./workflow-graph-branches.js";
 
 export type WorkflowNodeOutcome = "success" | "failure";
 
@@ -20,6 +28,9 @@ export interface WorkflowNodeExecutionContext {
   task: TaskDetail;
   settings: Pick<Settings, "experimentalFeatures"> | undefined;
   context: Record<string, unknown>;
+  /** Set during concurrent branch execution; fail-fast aborts via this signal.
+   *  Undefined on the sequential path (zero behavior change for linear graphs). */
+  signal?: AbortSignal;
 }
 
 export type WorkflowNodeHandler = (node: WorkflowIrNode, context: WorkflowNodeExecutionContext) => Promise<WorkflowNodeResult>;
@@ -30,6 +41,15 @@ export interface WorkflowGraphExecutorDeps {
   /** Executes custom (non-seam) prompt/script/gate nodes. */
   runCustomNode?: WorkflowCustomNodeRunner;
   maxRetriesPerNode?: number;
+  /** Per-branch run-state persistence (U13). Optional — fully in-memory without it. */
+  branchPersistence?: WorkflowBranchPersistence;
+  /** Bounds concurrent branch-node execution. Omit when the semaphore is
+   *  enforced beneath runCustomNode (the session layer) to avoid double-acquire. */
+  branchSemaphore?: WorkflowBranchSemaphore;
+  /** Live per-branch progress (dashboard badges). */
+  onBranchProgress?: (progress: WorkflowBranchProgress) => void;
+  /** Stable identifier for this run, used to key persisted branch state. */
+  runId?: string;
 }
 
 export interface WorkflowGraphExecutorResult {
@@ -85,6 +105,34 @@ export class WorkflowGraphExecutor {
     const context: Record<string, unknown> = {};
     const visitedNodeIds: string[] = [];
     const inStack = new Set<string>();
+    const runId = this.deps.runId ?? `${task.id}:run`;
+
+    // On resume, completed branch nodes (from a prior crashed run) are skipped
+    // so their handlers do not re-fire (idempotency).
+    let completedNodeIds: Set<string> | undefined;
+    const persisted = await this.deps.branchPersistence?.loadBranchStates?.(task.id, runId);
+    if (persisted && persisted.length > 0) {
+      completedNodeIds = new Set(
+        persisted
+          .filter((s: WorkflowBranchRunState) => s.status === "completed")
+          .map((s) => s.currentNodeId),
+      );
+    }
+
+    // Shared branch environment: built lazily so the sequential path pays nothing.
+    const branchEnv = (): BranchEnvironment => ({
+      task,
+      settings,
+      runId,
+      nodeMap,
+      outgoingMap,
+      runBranchNode: (node, signal) => this.executeNodeWithRetries(node, task, settings, context, signal),
+      shouldTraverseEdge: (edge, source) => this.shouldTraverseEdge(edge, source),
+      persistence: this.deps.branchPersistence,
+      semaphore: this.deps.branchSemaphore,
+      onBranchProgress: this.deps.onBranchProgress,
+      completedNodeIds,
+    });
 
     const walk = async (nodeId: string): Promise<WorkflowNodeResult> => {
       const node = nodeMap.get(nodeId);
@@ -101,6 +149,23 @@ export class WorkflowGraphExecutor {
           return { outcome: "success" };
         }
 
+        if (node.kind === "split") {
+          // Concurrent fan-out: branches run in parallel up to their join, which
+          // synchronizes per its config. The card stays in the split's column for
+          // the whole window (no handler-driven move happens in here). Execution
+          // then continues sequentially from the join node.
+          const splitResult = await runSplitJoin(node, branchEnv());
+          visitedNodeIds.push(...splitResult.visitedNodeIds);
+          context[`node:${node.id}:outcome`] = splitResult.outcome;
+          context[`node:${splitResult.joinNodeId}:outcome`] = splitResult.outcome;
+          context[`node:${splitResult.joinNodeId}:branchOutcomes`] = splitResult.branchOutcomes;
+          if (!inStack.has(splitResult.joinNodeId)) visitedNodeIds.push(splitResult.joinNodeId);
+          return await traverseChildren(
+            nodeMap.get(splitResult.joinNodeId)!,
+            { outcome: splitResult.outcome },
+          );
+        }
+
         const result = await this.executeNodeWithRetries(node, task, settings, context);
         if (result.contextPatch) Object.assign(context, result.contextPatch);
         context[`node:${node.id}:outcome`] = result.outcome;
@@ -164,6 +229,7 @@ export class WorkflowGraphExecutor {
     task: TaskDetail,
     settings: Pick<Settings, "experimentalFeatures"> | undefined,
     context: Record<string, unknown>,
+    signal?: AbortSignal,
   ): Promise<WorkflowNodeResult> {
     const handler = this.handlers[node.kind];
     if (!handler) {
@@ -178,8 +244,10 @@ export class WorkflowGraphExecutor {
 
     let lastError: unknown;
     for (let attempt = 0; attempt < maxAttempts; attempt++) {
+      // Fail-fast cancellation: a branch aborted mid-retry stops re-trying.
+      if (signal?.aborted) return { outcome: "failure", value: "aborted" };
       try {
-        return await handler(node, { task, settings, context });
+        return await handler(node, { task, settings, context, signal });
       } catch (error) {
         lastError = error;
       }

packages/engine/src/workflow-graph-task-runner.ts

@@ -3,6 +3,11 @@ import { isExperimentalFeatureEnabled } from "@fusion/core";
 
 import { WorkflowGraphExecutor, type WorkflowNodeOutcome } from "./workflow-graph-executor.js";
 import type { WorkflowCustomNodeRunner, WorkflowLegacySeams } from "./workflow-node-handlers.js";
+import type {
+  WorkflowBranchPersistence,
+  WorkflowBranchProgress,
+  WorkflowBranchSemaphore,
+} from "./workflow-graph-branches.js";
 // (Both types are also used as values in the side-effect tracking wrappers below.)
 
 /**
@@ -37,6 +42,13 @@ export interface WorkflowGraphTaskRunnerDeps {
   maxRetriesPerNode?: number;
   /** Optional diagnostics hook (audit/log emission). Never throws into the run. */
   onEvent?: (event: { type: "start" | "terminal" | "fallback"; taskId: string; detail: string }) => void;
+  /** Per-branch run-state persistence + resume (U13). Additive; in-memory without it. */
+  branchPersistence?: WorkflowBranchPersistence;
+  /** Bounds concurrent branch-node execution (U13); omit when the semaphore is
+   *  enforced beneath runCustomNode at the session layer. */
+  branchSemaphore?: WorkflowBranchSemaphore;
+  /** Live per-branch progress for dashboard badges (U9/U13). */
+  onBranchProgress?: (progress: WorkflowBranchProgress) => void;
 }
 
 /**
@@ -47,8 +59,18 @@ export interface WorkflowGraphTaskRunnerDeps {
  * run the legacy pipeline; a task is never stranded by interpreter bugs.
  */
 export class WorkflowGraphTaskRunner {
+  /** Latest per-branch progress, keyed by branchId. Store/dashboard-readable
+   *  (U9 badges). Reset at the start of each run; the card never moves during a
+   *  parallel window (KTD-11) so this is purely presentational state. */
+  private readonly branchProgress = new Map<string, WorkflowBranchProgress>();
+
   public constructor(private readonly deps: WorkflowGraphTaskRunnerDeps) {}
 
+  /** Snapshot of current per-branch progress (branchId, nodeId, status). */
+  public getBranchProgress(): WorkflowBranchProgress[] {
+    return [...this.branchProgress.values()];
+  }
+
   private emit(type: "start" | "terminal" | "fallback", taskId: string, detail: string): void {
     try {
       this.deps.onEvent?.({ type, taskId, detail });
@@ -91,6 +113,7 @@ export class WorkflowGraphTaskRunner {
     }
 
     this.emit("start", task.id, definition.id);
+    this.branchProgress.clear();
 
     // Track whether any node side effects ran. A pre-run interpreter error
     // (bad IR structure, wiring) can safely fall back to the legacy pipeline;
@@ -117,6 +140,17 @@ export class WorkflowGraphTaskRunner {
         seams: wrappedSeams,
         runCustomNode: wrappedRunCustomNode,
         maxRetriesPerNode: this.deps.maxRetriesPerNode,
+        branchPersistence: this.deps.branchPersistence,
+        branchSemaphore: this.deps.branchSemaphore,
+        runId: `${task.id}:${definition.id}`,
+        onBranchProgress: (progress) => {
+          this.branchProgress.set(progress.branchId, progress);
+          try {
+            this.deps.onBranchProgress?.(progress);
+          } catch {
+            // Progress reporting must never affect the run.
+          }
+        },
       });
       const result = await executor.run(task, settings, definition.ir);
       if (!result.executed) {