diff --git a/packages/engine/src/__tests__/workflow-graph-foreach.test.ts b/packages/engine/src/__tests__/workflow-graph-foreach.test.ts
index 8734d1b0ad..86595ec1d4 100644
--- a/packages/engine/src/__tests__/workflow-graph-foreach.test.ts
+++ b/packages/engine/src/__tests__/workflow-graph-foreach.test.ts
@@ -386,7 +386,10 @@ describe("WorkflowGraphExecutor foreach (U3)", () => {
     expect(result.outcome).toBe("failure");
   });
 
-  it("parallel mode is guarded with a clear not-yet-wired failure (U10 replaces it)", async () => {
+  it("parallel mode (now worktree isolation, U10) fails cleanly without isolation wiring", async () => {
+    // U10: parallel mode defaults to worktree isolation. Without the worktree /
+    // integration deps wired, the foreach fails with a routable value rather than
+    // running shared-mode physics (which would be an unguardable concurrent-write race).
     const seams = baseSeams({
       stepExecute: async () => ({ outcome: "success", value: "step-done" }),
     });
@@ -397,7 +400,7 @@ describe("WorkflowGraphExecutor foreach (U3)", () => {
       foreachIr(singleExecuteTemplate(), { config: { mode: "parallel", concurrency: 2 } }),
     );
     expect(result.outcome).toBe("failure");
-    expect(result.context["node:fe:value"]).toBe("parallel-not-wired");
+    expect(result.context["node:fe:value"]).toBe("worktree-isolation-unwired");
   });
 
   it("getTaskSteps dep is used to read a fresh count when injected", async () => {
diff --git a/packages/engine/src/__tests__/workflow-step-parallel.test.ts b/packages/engine/src/__tests__/workflow-step-parallel.test.ts
new file mode 100644
index 0000000000..921962d085
--- /dev/null
+++ b/packages/engine/src/__tests__/workflow-step-parallel.test.ts
@@ -0,0 +1,561 @@
+import { describe, expect, it, vi } from "vitest";
+import type { TaskDetail, TaskStep, WorkflowIr, WorkflowIrNode } from "@fusion/core";
+
+import { WorkflowGraphExecutor } from "../workflow-graph-executor.js";
+import {
+  FOREACH_ACTIVE_CONTEXT_KEY,
+  type ForeachActiveContext,
+  type WorkflowLegacySeams,
+} from "../workflow-node-handlers.js";
+import {
+  IntegrationQueue,
+  type IntegrationGitOps,
+  type IntegrationProjection,
+  type IntegrationAttemptResult,
+} from "../step-integration.js";
+import type { WorkflowStepInstanceState } from "../workflow-graph-foreach.js";
+
+const settingsOn = () => ({ experimentalFeatures: { workflowGraphExecutor: true } });
+
+// ── shared test scaffolding ─────────────────────────────────────────────────
+
+/** Build a TaskDetail with a step list; dependsOn (0-indexed) per step optional. */
+function taskWithSteps(specs: Array<{ dependsOn?: number[] }> | number): TaskDetail {
+  const list: Array<{ dependsOn?: number[] }> =
+    typeof specs === "number" ? Array.from({ length: specs }, () => ({})) : specs;
+  const steps: TaskStep[] = list.map((s, i) => ({
+    name: `Step ${i + 1}`,
+    status: "pending" as const,
+    ...(s.dependsOn ? { dependsOn: s.dependsOn } : {}),
+  }));
+  return { id: "FN-PAR", steps } as unknown as TaskDetail;
+}
+
+/** Base no-op seams with an optional override. */
+function baseSeams(overrides: Partial<WorkflowLegacySeams>): WorkflowLegacySeams {
+  const ok = async () => ({ outcome: "success" as const });
+  return { planning: ok, execute: ok, review: ok, merge: ok, schedule: ok, ...overrides };
+}
+
+/** A single step-execute template. */
+function singleExecuteTemplate() {
+  return {
+    nodes: [{ id: "exec", kind: "prompt" as const, config: { seam: "step-execute" } }],
+    edges: [],
+  };
+}
+
+/** exec → step-review template (review routes approve/revise/rethink). */
+function reviewTemplate() {
+  return {
+    nodes: [
+      { id: "exec", kind: "prompt" as const, config: { seam: "step-execute" } },
+      { id: "review", kind: "step-review" as const, config: { type: "code" } },
+    ],
+    edges: [
+      { from: "exec", to: "review", condition: "success" },
+      { from: "review", to: "exec", condition: "outcome:revise", kind: "rework" as const },
+      { from: "review", to: "exec", condition: "outcome:rethink", kind: "rework" as const },
+    ],
+  };
+}
+
+function foreachIr(
+  template: { nodes: WorkflowIrNode[]; edges: WorkflowIr["edges"] },
+  config: Record<string, unknown> = {},
+): WorkflowIr {
+  return {
+    version: "v2",
+    name: "parallel-test",
+    columns: [{ id: "work", name: "Work", traits: [] }],
+    nodes: [
+      { id: "start", kind: "start" },
+      { id: "fe", kind: "foreach", config: { source: "task-steps", template, ...config } },
+      { id: "end", kind: "end" },
+    ],
+    edges: [
+      { from: "start", to: "fe" },
+      { from: "fe", to: "end", condition: "success" },
+    ],
+  };
+}
+
+/**
+ * A fake worktree+git+integration backend the executor deps are wired to. Models
+ * a per-instance branch and an ordered integration base purely in-memory; tests
+ * script which (stepIndex) integrations conflict.
+ */
+function makeFakeBackend(opts: {
+  conflictSteps?: Set<number>;
+  /** Steps whose integration conflicts only on the FIRST attempt (then succeed). */
+  conflictOnceSteps?: Set<number>;
+} = {}) {
+  const conflictSteps = opts.conflictSteps ?? new Set<number>();
+  const conflictOnceSteps = opts.conflictOnceSteps ?? new Set<number>();
+  const integrateAttempts = new Map<number, number>();
+
+  const allocations: Array<{ stepIndex: number; branchName: string; base: string | undefined }> = [];
+  const integrationOrder: number[] = [];
+  const discarded: string[] = [];
+  const released: string[] = [];
+  const doneSteps: number[] = [];
+  const instanceIntegrated: Array<{ stepIndex: number; at: string }> = [];
+  let integrationBase = "main@0";
+  let integratedCount = 0;
+  const resetBranches: string[] = [];
+
+  const gitOps: IntegrationGitOps = {
+    integrate: async (branchName, stepIndex): Promise<IntegrationAttemptResult> => {
+      const attempt = (integrateAttempts.get(stepIndex) ?? 0) + 1;
+      integrateAttempts.set(stepIndex, attempt);
+      const conflictNow =
+        conflictSteps.has(stepIndex) || (conflictOnceSteps.has(stepIndex) && attempt === 1);
+      if (conflictNow) {
+        return { kind: "conflict", conflictedFiles: [`step-${stepIndex}.ts`] };
+      }
+      integrationOrder.push(stepIndex);
+      integratedCount += 1;
+      integrationBase = `main@${integratedCount}`; // base advances on each integration
+      return { kind: "integrated", integratedAt: `t${stepIndex}` };
+    },
+    discardBranch: async (branchName) => {
+      discarded.push(branchName);
+      released.push(branchName);
+    },
+  };
+
+  const projection: IntegrationProjection = {
+    markStepDone: async (stepIndex) => {
+      doneSteps.push(stepIndex);
+    },
+    markInstanceIntegrated: async (stepIndex, at) => {
+      instanceIntegrated.push({ stepIndex, at });
+    },
+  };
+
+  return {
+    gitOps,
+    projection,
+    allocations,
+    integrationOrder,
+    discarded,
+    released,
+    doneSteps,
+    instanceIntegrated,
+    resetBranches,
+    getBase: () => integrationBase,
+    deps: {
+      allocateInstanceWorktree: async (stepIndex: number, base: string | undefined) => {
+        const branchName = `fusion/fn-par-step-${stepIndex}`;
+        allocations.push({ stepIndex, branchName, base });
+        return { worktreePath: `/wt/step-${stepIndex}`, branchName };
+      },
+      resolveIntegrationBase: async () => integrationBase,
+      integrationGitOps: gitOps,
+      integrationProjection: projection,
+    },
+  };
+}
+
+// ── IntegrationQueue state machine (TEST-FIRST) ─────────────────────────────
+
+describe("IntegrationQueue (ordered integration state machine)", () => {
+  function queueHarness(pinned: number, integrate: (b: string, i: number) => IntegrationAttemptResult) {
+    const order: number[] = [];
+    const done: number[] = [];
+    const integrated: number[] = [];
+    const discarded: string[] = [];
+    const git: IntegrationGitOps = {
+      integrate: async (b, i) => {
+        const r = integrate(b, i);
+        if (r.kind === "integrated") order.push(i);
+        return r;
+      },
+      discardBranch: async (b) => {
+        discarded.push(b);
+      },
+    };
+    const proj: IntegrationProjection = {
+      markStepDone: async (i) => {
+        done.push(i);
+      },
+      markInstanceIntegrated: async (i) => {
+        integrated.push(i);
+      },
+    };
+    const q = new IntegrationQueue(git, proj, pinned);
+    return { q, order, done, integrated, discarded };
+  }
+
+  it("integrates strictly in step order even when completion order inverts", async () => {
+    const h = queueHarness(3, () => ({ kind: "integrated", integratedAt: "t" }));
+    // Enqueue out of order: 2 first, then 0, then 1.
+    h.q.enqueue(2, "b2");
+    let outcomes = await h.q.drain();
+    expect(outcomes).toEqual([]); // 0 not ready → nothing integrates.
+    h.q.enqueue(0, "b0");
+    outcomes = await h.q.drain();
+    expect(h.order).toEqual([0]); // only 0 (1 is the next gap).
+    h.q.enqueue(1, "b1");
+    await h.q.drain();
+    expect(h.order).toEqual([0, 1, 2]); // 1 then 2 cascade.
+    expect(h.q.isDrained()).toBe(true);
+  });
+
+  it("projection-first: markStepDone precedes markInstanceIntegrated per step", async () => {
+    const events: string[] = [];
+    const git: IntegrationGitOps = {
+      integrate: async () => ({ kind: "integrated", integratedAt: "t" }),
+      discardBranch: async () => {},
+    };
+    const proj: IntegrationProjection = {
+      markStepDone: async (i) => {
+        events.push(`done:${i}`);
+      },
+      markInstanceIntegrated: async (i) => {
+        events.push(`row:${i}`);
+      },
+    };
+    const q = new IntegrationQueue(git, proj, 1);
+    q.enqueue(0, "b0");
+    await q.drain();
+    expect(events).toEqual(["done:0", "row:0"]);
+  });
+
+  it("conflict stops the drain, discards the branch, and does not mark done", async () => {
+    const h = queueHarness(2, (_b, i) =>
+      i === 0 ? { kind: "conflict", conflictedFiles: ["x"] } : { kind: "integrated", integratedAt: "t" },
+    );
+    h.q.enqueue(0, "b0");
+    h.q.enqueue(1, "b1");
+    const outcomes = await h.q.drain();
+    expect(outcomes).toEqual([{ stepIndex: 0, status: "conflict", conflictedFiles: ["x"] }]);
+    expect(h.done).toEqual([]);
+    expect(h.discarded).toContain("b0");
+    // Step 1 must NOT integrate ahead of the unresolved step 0.
+    expect(h.order).toEqual([]);
+  });
+
+  it("skip advances the cursor past a failed step", async () => {
+    const h = queueHarness(3, () => ({ kind: "integrated", integratedAt: "t" }));
+    h.q.skip(0);
+    h.q.enqueue(1, "b1");
+    h.q.enqueue(2, "b2");
+    await h.q.drain();
+    expect(h.order).toEqual([1, 2]);
+    expect(h.q.isDrained()).toBe(true);
+  });
+});
+
+// ── full U10 scenarios ──────────────────────────────────────────────────────
+
+describe("WorkflowGraphExecutor parallel/worktree foreach (U10)", () => {
+  /** Run a foreach IR with a fake backend; record the order steps START. */
+  async function runScenario(
+    task: TaskDetail,
+    config: Record<string, unknown>,
+    backend: ReturnType<typeof makeFakeBackend>,
+    overrides: Partial<{
+      semaphoreAvailability: () => number;
+      stepExecute: WorkflowLegacySeams["stepExecute"];
+      stepReview: WorkflowLegacySeams["stepReview"];
+      onReworkReset: (a: ForeachActiveContext) => void;
+      template: { nodes: WorkflowIrNode[]; edges: WorkflowIr["edges"] };
+    }> = {},
+  ) {
+    const startOrder: number[] = [];
+    const seams = baseSeams({
+      stepExecute:
+        overrides.stepExecute ??
+        (async (_t, ctx) => {
+          const active = ctx[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext;
+          startOrder.push(active.stepIndex);
+          return { outcome: "success", value: "step-done" };
+        }),
+      ...(overrides.stepReview ? { stepReview: overrides.stepReview } : {}),
+    });
+    const executor = new WorkflowGraphExecutor({
+      seams,
+      ...backend.deps,
+      ...(overrides.semaphoreAvailability ? { semaphoreAvailability: overrides.semaphoreAvailability } : {}),
+      ...(overrides.onReworkReset ? { onReworkReset: overrides.onReworkReset as never } : {}),
+    });
+    const ir = foreachIr(overrides.template ?? singleExecuteTemplate(), config);
+    const result = await executor.run(task, settingsOn(), ir);
+    return { result, startOrder };
+  }
+
+  it("diamond dep graph (0 ← 1,2 ← 3) runs 1∥2 then 3", async () => {
+    // Step 0 root; 1 and 2 depend on 0; 3 depends on 1 and 2.
+    const task = taskWithSteps([
+      { dependsOn: [] },
+      { dependsOn: [0] },
+      { dependsOn: [0] },
+      { dependsOn: [1, 2] },
+    ]);
+    const backend = makeFakeBackend();
+    const { result, startOrder } = await runScenario(
+      task,
+      { mode: "parallel", isolation: "worktree", concurrency: 4 },
+      backend,
+    );
+    expect(result.outcome).toBe("success");
+    // 0 first; 1 and 2 after 0 integrated; 3 last.
+    expect(startOrder[0]).toBe(0);
+    expect(new Set([startOrder[1], startOrder[2]])).toEqual(new Set([1, 2]));
+    expect(startOrder[3]).toBe(3);
+    // Ordered integration is step order.
+    expect(backend.integrationOrder).toEqual([0, 1, 2, 3]);
+    expect(backend.doneSteps).toEqual([0, 1, 2, 3]);
+  });
+
+  it("sequential + worktree runs one at a time with per-step branches + ordered integration", async () => {
+    const task = taskWithSteps(3);
+    const backend = makeFakeBackend();
+    const concurrentPeak = { value: 0 };
+    let active = 0;
+    const { result } = await runScenario(task, { mode: "sequential", isolation: "worktree" }, backend, {
+      stepExecute: async () => {
+        active += 1;
+        concurrentPeak.value = Math.max(concurrentPeak.value, active);
+        await Promise.resolve();
+        active -= 1;
+        return { outcome: "success", value: "step-done" };
+      },
+    });
+    expect(result.outcome).toBe("success");
+    expect(concurrentPeak.value).toBe(1); // never more than one at a time.
+    expect(backend.allocations.map((a) => a.branchName)).toEqual([
+      "fusion/fn-par-step-0",
+      "fusion/fn-par-step-1",
+      "fusion/fn-par-step-2",
+    ]);
+    expect(backend.integrationOrder).toEqual([0, 1, 2]);
+  });
+
+  it("unannotated plan stays fully sequential at concurrency 4", async () => {
+    const task = taskWithSteps(4); // no dependsOn → each implicitly depends on prev.
+    const backend = makeFakeBackend();
+    const concurrentPeak = { value: 0 };
+    const order: number[] = [];
+    let active = 0;
+    const { result } = await runScenario(
+      task,
+      { mode: "parallel", isolation: "worktree", concurrency: 4 },
+      backend,
+      {
+        stepExecute: async (_t, ctx) => {
+          const a = ctx[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext;
+          order.push(a.stepIndex);
+          active += 1;
+          concurrentPeak.value = Math.max(concurrentPeak.value, active);
+          await Promise.resolve();
+          active -= 1;
+          return { outcome: "success", value: "step-done" };
+        },
+      },
+    );
+    expect(result.outcome).toBe("success");
+    expect(concurrentPeak.value).toBe(1);
+    expect(order).toEqual([0, 1, 2, 3]);
+    expect(backend.integrationOrder).toEqual([0, 1, 2, 3]);
+  });
+
+  it("conflict between parallel steps → loser reworks on updated base and succeeds", async () => {
+    const task = taskWithSteps([{ dependsOn: [] }, { dependsOn: [] }]);
+    // Step 1 conflicts the first integration attempt, then succeeds.
+    const backend = makeFakeBackend({ conflictOnceSteps: new Set([1]) });
+    const execStarts: number[] = [];
+    const { result } = await runScenario(
+      task,
+      { mode: "parallel", isolation: "worktree", concurrency: 2 },
+      backend,
+      {
+        stepExecute: async (_t, ctx) => {
+          const a = ctx[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext;
+          execStarts.push(a.stepIndex);
+          return { outcome: "success", value: "step-done" };
+        },
+      },
+    );
+    expect(result.outcome).toBe("success");
+    // Step 1 executed twice (initial + rework after conflict).
+    expect(execStarts.filter((s) => s === 1).length).toBe(2);
+    // Both eventually integrated, in step order.
+    expect(backend.integrationOrder).toEqual([0, 1]);
+    expect(backend.doneSteps).toEqual([0, 1]);
+    // The conflicting branch was discarded before re-running.
+    expect(backend.discarded).toContain("fusion/fn-par-step-1");
+    // The rework re-allocated off the UPDATED base (after step 0 integrated).
+    const step1Allocs = backend.allocations.filter((a) => a.stepIndex === 1);
+    expect(step1Allocs.length).toBe(2);
+    expect(step1Allocs[1].base).toBe("main@1");
+  });
+
+  it("conflict rework exhaustion routes rework-exhausted", async () => {
+    const task = taskWithSteps([{ dependsOn: [] }]);
+    const backend = makeFakeBackend({ conflictSteps: new Set([0]) }); // always conflicts.
+    const { result } = await runScenario(
+      task,
+      { mode: "parallel", isolation: "worktree", concurrency: 1, maxReworkCycles: 2 },
+      backend,
+    );
+    expect(result.outcome).toBe("failure");
+    expect(result.context).toBeDefined();
+    // The foreach node's value surfaces rework-exhausted.
+    expect(result.visitedNodeIds).toContain("fe");
+    // Never marked done.
+    expect(backend.doneSteps).toEqual([]);
+  });
+
+  it("integration order is step order even when completion order inverts", async () => {
+    const task = taskWithSteps([{ dependsOn: [] }, { dependsOn: [] }, { dependsOn: [] }]);
+    const backend = makeFakeBackend();
+    // Make later steps complete FIRST by delaying step 0's execution.
+    const { result } = await runScenario(
+      task,
+      { mode: "parallel", isolation: "worktree", concurrency: 3 },
+      backend,
+      {
+        stepExecute: async (_t, ctx) => {
+          const a = ctx[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext;
+          // step 0 yields the most → completes last.
+          const delays = 2 - a.stepIndex;
+          for (let i = 0; i < delays; i++) await Promise.resolve();
+          return { outcome: "success", value: "step-done" };
+        },
+      },
+    );
+    expect(result.outcome).toBe("success");
+    expect(backend.integrationOrder).toEqual([0, 1, 2]);
+    expect(backend.doneSteps).toEqual([0, 1, 2]);
+  });
+
+  it("semaphore starvation degrades to sequential without deadlock", async () => {
+    const task = taskWithSteps([{ dependsOn: [] }, { dependsOn: [] }, { dependsOn: [] }]);
+    const backend = makeFakeBackend();
+    const concurrentPeak = { value: 0 };
+    let active = 0;
+    const { result } = await runScenario(
+      task,
+      { mode: "parallel", isolation: "worktree", concurrency: 4 },
+      backend,
+      {
+        semaphoreAvailability: () => 0, // fully starved.
+        stepExecute: async () => {
+          active += 1;
+          concurrentPeak.value = Math.max(concurrentPeak.value, active);
+          await Promise.resolve();
+          active -= 1;
+          return { outcome: "success", value: "step-done" };
+        },
+      },
+    );
+    expect(result.outcome).toBe("success");
+    expect(concurrentPeak.value).toBe(1); // forced to 1 under starvation, no deadlock.
+    expect(backend.integrationOrder).toEqual([0, 1, 2]);
+  });
+
+  it("dependency cycle at expansion fails audited", async () => {
+    // Step 1 depends on step 2 (a forward reference → cycle signature).
+    const task = taskWithSteps([{ dependsOn: [] }, { dependsOn: [2] }, { dependsOn: [] }]);
+    const backend = makeFakeBackend();
+    const { result } = await runScenario(
+      task,
+      { mode: "parallel", isolation: "worktree", concurrency: 4 },
+      backend,
+    );
+    expect(result.outcome).toBe("failure");
+    expect(backend.allocations).toEqual([]); // never expanded any instance.
+  });
+
+  it("RETHINK resets only the instance branch (branch-scoped)", async () => {
+    const task = taskWithSteps([{ dependsOn: [] }]);
+    const backend = makeFakeBackend();
+    const resetBranches: string[] = [];
+    let reviewCalls = 0;
+    const { result } = await runScenario(
+      task,
+      { mode: "parallel", isolation: "worktree", concurrency: 1 },
+      backend,
+      {
+        template: reviewTemplate(),
+        stepReview: async () => {
+          reviewCalls += 1;
+          return reviewCalls === 1 ? { verdict: "RETHINK" as const } : { verdict: "APPROVE" as const };
+        },
+        onReworkReset: (active: ForeachActiveContext) => {
+          // Branch-scoped reset: the active context carries THIS instance's branch.
+          resetBranches.push(active.branchName ?? "<none>");
+        },
+      },
+    );
+    expect(result.outcome).toBe("success");
+    expect(resetBranches).toEqual(["fusion/fn-par-step-0"]);
+    expect(backend.integrationOrder).toEqual([0]);
+  });
+
+  it("merge-blocker stays blocked until last integration (projection rule)", async () => {
+    const task = taskWithSteps([{ dependsOn: [] }, { dependsOn: [0] }]);
+    const backend = makeFakeBackend();
+    // Capture doneSteps progression: step 1 must not be done until it integrates.
+    const doneAfterStep0Integrated: number[] = [];
+    const origMarkDone = backend.projection.markStepDone;
+    backend.projection.markStepDone = async (i) => {
+      await origMarkDone(i);
+      doneAfterStep0Integrated.push(i);
+    };
+    const { result } = await runScenario(
+      task,
+      { mode: "sequential", isolation: "worktree" },
+      backend,
+    );
+    expect(result.outcome).toBe("success");
+    // done flips strictly in integration order — step 1 done ONLY after step 0.
+    expect(doneAfterStep0Integrated).toEqual([0, 1]);
+  });
+
+  it("worktree isolation without wiring fails cleanly (routable)", async () => {
+    const task = taskWithSteps(2);
+    const executor = new WorkflowGraphExecutor({ seams: baseSeams({}) });
+    const result = await executor.run(
+      task,
+      settingsOn(),
+      foreachIr(singleExecuteTemplate(), { mode: "parallel", isolation: "worktree" }),
+    );
+    expect(result.outcome).toBe("failure");
+  });
+});
+
+// ── crash-resume reconciliation ─────────────────────────────────────────────
+
+describe("worktree-isolation crash-resume reconciliation (U10)", () => {
+  it("persists branchName + awaiting-integration through the persistence hook", async () => {
+    const task = taskWithSteps([{ dependsOn: [] }]);
+    const backend = makeFakeBackend();
+    const saved: WorkflowStepInstanceState[] = [];
+    const persistence = {
+      saveInstanceState: (s: WorkflowStepInstanceState) => {
+        saved.push({ ...s });
+      },
+    };
+    const seams = baseSeams({
+      stepExecute: async () => ({ outcome: "success", value: "step-done" }),
+    });
+    const executor = new WorkflowGraphExecutor({
+      seams,
+      ...backend.deps,
+      stepInstancePersistence: persistence,
+    });
+    const result = await executor.run(
+      task,
+      settingsOn(),
+      foreachIr(singleExecuteTemplate(), { mode: "sequential", isolation: "worktree" }),
+    );
+    expect(result.outcome).toBe("success");
+    // The instance row carried branchName and reached awaiting-integration.
+    const awaiting = saved.find((s) => s.status === "awaiting-integration");
+    expect(awaiting).toBeDefined();
+    expect(awaiting?.branchName).toBe("fusion/fn-par-step-0");
+  });
+});
diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts
index 3b0ed66f7e..f3aec442f8 100644
--- a/packages/engine/src/executor.ts
+++ b/packages/engine/src/executor.ts
@@ -45,14 +45,14 @@ import {
   resolveAgentMemoryInclusionMode,
   type RunCommandResult,
 } from "@fusion/core";
-import { findWorktreeUser } from "./merger.js";
+import { findWorktreeUser, getConflictedFiles } from "./merger.js";
 import {
   runVerificationCommand,
   summarizeVerificationOutput,
   VERIFICATION_LOG_MAX_CHARS,
   type VerificationResult,
 } from "./verification-utils.js";
-import { generateWorktreeName, resolveTaskWorkingBranch } from "./worktree-names.js";
+import { canonicalStepInstanceBranchName, generateWorktreeName, resolveTaskWorkingBranch } from "./worktree-names.js";
 import { resolveTaskWorktreePath, resolveWorktreesDir } from "./worktree-paths.js";
 import { Type, type Static } from "@earendil-works/pi-ai";
 import { describeModel, promptWithFallback, compactSessionContext } from "./pi.js";
@@ -3306,6 +3306,11 @@ export class TaskExecutor {
         // Step-inversion (KTD-15, U14): code node runner — esbuild compile +
         // child-process execution with the harness contract.
         runCode: this.buildCodeNodeRunner(),
+        // Step-inversion (KTD-11, U10): worktree isolation + ordered integration +
+        // parallel scheduling. Per-instance worktrees branched off the task's main
+        // branch tip; integration rebases each branch in step order; the projection
+        // flips done-iff-integrated. Shared isolation never invokes these.
+        ...this.buildForeachWorktreeDeps(task),
       });
       let result: WorkflowGraphTaskRunResult;
       try {
@@ -3499,6 +3504,216 @@ export class TaskExecutor {
     });
   }
 
+  /**
+   * Build the worktree-isolation + ordered-integration + parallel-scheduling deps
+   * for a graph-owned foreach (KTD-11, U10). Returns the additive set the
+   * WorkflowGraphTaskRunner forwards to the foreach sub-walk:
+   *
+   *   - `allocateInstanceWorktree(i, base)` — a per-instance worktree on a
+   *     canonical `fusion/<task>-step-<i>` branch off `base` (the main tip),
+   *     created via the existing `createWorktree` path (the file-scope guard the
+   *     session machinery installs applies unchanged to anything the instance
+   *     session commits in this worktree — we do NOT bypass it);
+   *   - `resolveIntegrationBase()` — the task's main branch tip, re-read before each
+   *     (re)allocation so a rework lands on the UPDATED base;
+   *   - `integrationGitOps` — rebase the instance branch onto the main branch in
+   *     the task's MAIN worktree, fast-forward main on success; on conflict reuse
+   *     merger.ts `getConflictedFiles` (NOT reimplemented) and abort the rebase so
+   *     the next instance can integrate; `discardBranch` deletes the branch + frees
+   *     the instance worktree (pool hygiene);
+   *   - `integrationProjection` — projection-first ordering (KTD-7): `markStepDone`
+   *     flips the step `done` via `updateStep(source:"graph")` (the dependency-order
+   *     guard admits it), THEN `markInstanceIntegrated` flips the persisted row;
+   *   - `semaphoreAvailability` — the live free-slot count so parallel scheduling
+   *     clamps without hold-and-wait.
+   *
+   * Best-effort throughout: a git failure routes the foreach to a clean failure
+   * (parked for human review) rather than crashing the run.
+   */
+  private buildForeachWorktreeDeps(task: Task): {
+    allocateInstanceWorktree: (
+      stepIndex: number,
+      base: string | undefined,
+    ) => Promise<{ worktreePath: string; branchName: string }>;
+    resolveIntegrationBase: () => Promise<string | undefined>;
+    integrationGitOps: import("./step-integration.js").IntegrationGitOps;
+    integrationProjection: import("./step-integration.js").IntegrationProjection;
+    semaphoreAvailability: () => number;
+    resumeReconcile: (
+      pinned: number,
+    ) => Promise<Array<{ stepIndex: number; disposition: "integrated" | "reintegrate" | "rerun"; branchName?: string }>>;
+  } {
+    const taskId = task.id;
+    // Per-instance worktree paths, so discard can free them.
+    const instancePaths = new Map<number, string>();
+
+    const mainWorktree = async (): Promise<string> => {
+      try {
+        return (await this.store.getTask(taskId)).worktree || this.rootDir;
+      } catch {
+        return this.rootDir;
+      }
+    };
+    const mainBranch = async (): Promise<string> => {
+      try {
+        const detail = await this.store.getTask(taskId);
+        return resolveTaskWorkingBranch(detail);
+      } catch {
+        return resolveTaskWorkingBranch(task);
+      }
+    };
+
+    return {
+      resolveIntegrationBase: async (): Promise<string | undefined> => {
+        // The main branch tip (HEAD of the task's working branch in its worktree).
+        try {
+          const { stdout } = await execAsync("git rev-parse HEAD", { cwd: await mainWorktree() });
+          const sha = stdout.trim();
+          return sha.length > 0 ? sha : await mainBranch();
+        } catch {
+          return await mainBranch();
+        }
+      },
+      allocateInstanceWorktree: async (stepIndex, base): Promise<{ worktreePath: string; branchName: string }> => {
+        const branchName = canonicalStepInstanceBranchName(taskId, stepIndex);
+        const worktreePath = resolveTaskWorktreePath(
+          this.rootDir,
+          undefined,
+          `${taskId.toLowerCase()}-step-${stepIndex}`,
+        );
+        // createWorktree installs the file-scope guard (session machinery,
+        // unchanged) and branches off `base` (the integration base / updated tip).
+        const created = await this.createWorktree(branchName, worktreePath, taskId, base);
+        instancePaths.set(stepIndex, created.path);
+        return { worktreePath: created.path, branchName: created.branch };
+      },
+      integrationGitOps: {
+        integrate: async (branchName, stepIndex): Promise<import("./step-integration.js").IntegrationAttemptResult> => {
+          const cwd = await mainWorktree();
+          const target = await mainBranch();
+          try {
+            // Rebase the instance branch onto the current main tip, then ff main.
+            await execAsync(`git rebase ${target} ${branchName}`, { cwd });
+            await execAsync(`git checkout ${target}`, { cwd });
+            await execAsync(`git merge --ff-only ${branchName}`, { cwd });
+            return { kind: "integrated", integratedAt: new Date().toISOString() };
+          } catch (err) {
+            // Conflict (or other rebase failure): classify via merger helper, abort.
+            const conflictedFiles = await getConflictedFiles(cwd);
+            try {
+              await execAsync("git rebase --abort", { cwd });
+            } catch {
+              // best-effort; leave the worktree recoverable.
+            }
+            // Restore main checkout so the next instance integrates cleanly.
+            try {
+              await execAsync(`git checkout ${target}`, { cwd });
+            } catch {
+              // best-effort.
+            }
+            executorLog.warn(
+              `[step-integration] ${taskId} step ${stepIndex} branch ${branchName} conflict: ${err instanceof Error ? err.message : String(err)}`,
+            );
+            return { kind: "conflict", conflictedFiles };
+          }
+        },
+        discardBranch: async (branchName, stepIndex): Promise<void> => {
+          const cwd = await mainWorktree();
+          const path = instancePaths.get(stepIndex);
+          if (path) {
+            // Remove the instance worktree (pool hygiene). Best-effort; force so a
+            // dirty/conflicting tree is still cleaned up.
+            try {
+              await execAsync(`git worktree remove --force "${path}"`, { cwd: this.rootDir });
+            } catch {
+              // best-effort cleanup.
+            }
+            instancePaths.delete(stepIndex);
+          }
+          // Delete the (now-merged or conflicting) branch.
+          try {
+            await execAsync(`git branch -D ${branchName}`, { cwd });
+          } catch {
+            // best-effort — the branch may already be gone.
+          }
+        },
+      },
+      integrationProjection: {
+        markStepDone: async (stepIndex): Promise<void> => {
+          // Projection-first (KTD-7): graph-source write relaxes the guard to
+          // dependency order; predecessors are integrated (done) by construction.
+          await this.store.updateStep(taskId, stepIndex, "done", { source: "graph" });
+        },
+        markInstanceIntegrated: async (stepIndex, integratedAt): Promise<void> => {
+          const store = this.store as unknown as {
+            saveWorkflowRunStepInstance?: (state: WorkflowStepInstanceState) => void;
+          };
+          if (typeof store.saveWorkflowRunStepInstance !== "function") return;
+          try {
+            store.saveWorkflowRunStepInstance({
+              taskId,
+              runId: `${taskId}:run`,
+              foreachNodeId: "",
+              stepIndex,
+              pinnedStepCount: 0,
+              currentNodeId: "",
+              status: "completed",
+              reworkCount: 0,
+              branchName: canonicalStepInstanceBranchName(taskId, stepIndex),
+              integratedAt,
+            } as WorkflowStepInstanceState);
+          } catch {
+            // Persistence is additive bookkeeping — never fail the integration.
+          }
+        },
+      },
+      semaphoreAvailability: (): number => this.options.semaphore?.availableCount ?? 1,
+      resumeReconcile: async (
+        pinned,
+      ): Promise<Array<{ stepIndex: number; disposition: "integrated" | "reintegrate" | "rerun"; branchName?: string }>> => {
+        // Crash-resume reconciliation (KTD-11): reconcile each persisted instance
+        // row against branch existence. integrated → done; branch exists not
+        // integrated → re-enter the integration queue; branch missing → re-run.
+        // NOTE (handoff): this is the per-run resume seeding only; the full
+        // self-healing sweep across stale runs (recoverStaleTransitionPending
+        // analogue) is out of scope for U10.
+        const store = this.store as unknown as {
+          loadWorkflowRunStepInstances?: (taskId: string, runId: string) => WorkflowStepInstanceState[];
+        };
+        if (typeof store.loadWorkflowRunStepInstances !== "function") return [];
+        let rows: WorkflowStepInstanceState[] = [];
+        try {
+          rows = store.loadWorkflowRunStepInstances(taskId, `${taskId}:run`) ?? [];
+        } catch {
+          return [];
+        }
+        const cwd = await mainWorktree();
+        const out: Array<{ stepIndex: number; disposition: "integrated" | "reintegrate" | "rerun"; branchName?: string }> = [];
+        for (const row of rows) {
+          if (row.stepIndex < 0 || row.stepIndex >= pinned) continue;
+          if (row.status === "completed" || row.integratedAt) {
+            out.push({ stepIndex: row.stepIndex, disposition: "integrated" });
+            continue;
+          }
+          const branchName = row.branchName || canonicalStepInstanceBranchName(taskId, row.stepIndex);
+          let branchExists = false;
+          try {
+            await execAsync(`git rev-parse --verify --quiet ${branchName}`, { cwd });
+            branchExists = true;
+          } catch {
+            branchExists = false;
+          }
+          if (branchExists && row.status === "awaiting-integration") {
+            out.push({ stepIndex: row.stepIndex, disposition: "reintegrate", branchName });
+          } else {
+            out.push({ stepIndex: row.stepIndex, disposition: "rerun" });
+          }
+        }
+        return out;
+      },
+    };
+  }
+
   /**
    * RETHINK reset-on-rework (KTD-4, U5): reset the active foreach instance's step
    * to its per-step baseline before the rework edge re-enters step-execute. Drives
@@ -3509,11 +3724,18 @@ export class TaskExecutor {
    * the documented KTD-2 semantics; the git reset + step→pending are authoritative.
    */
   private async applyGraphRethinkReset(taskId: string, active: ForeachActiveContext): Promise<void> {
-    let worktreePath = this.rootDir;
-    try {
-      worktreePath = (await this.store.getTask(taskId)).worktree || this.rootDir;
-    } catch {
-      // Best-effort worktree resolution; fall back to rootDir.
+    // Worktree isolation (KTD-11): reset the instance's OWN branch/worktree only —
+    // sibling instances and the integration base are untouched, so the blast-radius
+    // guard is STRUCTURAL (skipped) in this mode. Shared isolation resets the task's
+    // main worktree and keeps the KTD-2 ancestry guard as written.
+    const branchScoped = typeof active.worktreePath === "string" && active.worktreePath.length > 0;
+    let worktreePath = active.worktreePath ?? this.rootDir;
+    if (!branchScoped) {
+      try {
+        worktreePath = (await this.store.getTask(taskId)).worktree || this.rootDir;
+      } catch {
+        // Best-effort worktree resolution; fall back to rootDir.
+      }
     }
     const liveSteps = await this.store.getTask(taskId).then((t) => t.steps).catch(() => []);
     await resetStepToBaseline(
@@ -3524,11 +3746,16 @@ export class TaskExecutor {
         // when checkpointId resolves but no session is current (KTD-2 partial path).
         sessionRef: { current: null },
         reviewType: "code",
-        blastRadiusGuard: makeAncestryBlastRadiusGuard({
-          worktreePath,
-          task: { id: taskId, steps: liveSteps },
-          stepIndex: active.stepIndex,
-        }),
+        // Branch-scoped RETHINK under worktree isolation makes the guard structural
+        // (the reset can only touch the instance's own branch); shared isolation
+        // keeps the defensive ancestry guard (KTD-2/KTD-11).
+        blastRadiusGuard: branchScoped
+          ? undefined
+          : makeAncestryBlastRadiusGuard({
+              worktreePath,
+              task: { id: taskId, steps: liveSteps },
+              stepIndex: active.stepIndex,
+            }),
       },
       { id: taskId, steps: liveSteps },
       active.stepIndex,
@@ -3815,7 +4042,11 @@ export class TaskExecutor {
           return { outcome: "failure", value: "no-active-step-instance" };
         }
         const live = await this.store.getTask(seamTask.id);
-        const worktreePath = live.worktree || this.rootDir;
+        // Worktree isolation (KTD-11, U10): run the instance's session in ITS OWN
+        // worktree when the foreach allocated one; otherwise the task's main
+        // worktree (shared isolation — unchanged). The file-scope guard the session
+        // machinery installs applies to either worktree unchanged (not bypassed).
+        const worktreePath = active.worktreePath || live.worktree || this.rootDir;
         const result = await runTaskStep(
           {
             store: this.store,
@@ -3864,7 +4095,8 @@ export class TaskExecutor {
         }
         const stepIndex = active.stepIndex;
         const detail = await this.store.getTask(seamTask.id);
-        const worktreePath = detail.worktree || this.rootDir;
+        // Worktree isolation (KTD-11): review the instance's OWN worktree when set.
+        const worktreePath = active.worktreePath || detail.worktree || this.rootDir;
         const stepName = detail.steps[stepIndex]?.name ?? `Step ${stepIndex + 1}`;
         const promptContent = detail.prompt ?? "";
         const settings = await this.store.getSettings();
diff --git a/packages/engine/src/step-integration.ts b/packages/engine/src/step-integration.ts
new file mode 100644
index 0000000000..61b1f9983a
--- /dev/null
+++ b/packages/engine/src/step-integration.ts
@@ -0,0 +1,241 @@
+/**
+ * step-integration — the ordered integration stage for worktree-isolated foreach
+ * instances (step-inversion KTD-11, U10).
+ *
+ * Under `isolation: "worktree"` each foreach step instance runs in its OWN
+ * worktree/branch off a common integration base. Completing the instance's
+ * sub-walk does NOT mark the step done — instead the instance enqueues here as
+ * `awaiting-integration`. The {@link IntegrationQueue} then lands completed
+ * branches onto the task's main branch (the integration base) **strictly in step
+ * order**: instance `i` integrates only after instances `0..i-1` are integrated
+ * or skipped. This is the single place where a worktree-isolated instance's work
+ * becomes visible on main history, so:
+ *
+ *   - **Success**: flip the projection FIRST (`updateStep(..., "done", graph)` —
+ *     the dependency-order guard admits it because predecessors are done), THEN
+ *     mark the instance row `completed`/`integratedAt` (projection-first ordering,
+ *     KTD-7: closes the merge-blocker race), THEN release the instance worktree
+ *     (pool hygiene).
+ *   - **Conflict**: discard the instance branch (release worktree), and emit
+ *     `outcome:integration-conflict` for that instance. The foreach sub-walk
+ *     routes that like a rework — re-execute the step on the UPDATED integration
+ *     base in a fresh worktree, counting against the instance's `maxReworkCycles`
+ *     budget (exhaustion → rework-exhausted as usual).
+ *
+ * All git mechanics are behind the injectable {@link IntegrationGitOps} so this
+ * module is hermetically testable with fakes. Production wires it (executor.ts)
+ * to a rebase/cherry-pick onto the task's main branch that reuses merger.ts's
+ * conflict-classification helpers (`getConflictedFiles`) for conflict detection —
+ * NOT reimplemented here.
+ *
+ * Reconcile alignment (KTD-11): `complete step N` commits live on instance
+ * branches before integration and on main history after it; the projection rule
+ * ("done iff integrated") therefore agrees with `reconcileStepsFromGitHistory`
+ * (which reads main-worktree history) by construction.
+ */
+
+import { schedulerLog } from "./logger.js";
+
+/** The outcome of attempting to integrate one instance branch onto the base. */
+export type IntegrationAttemptResult =
+  | { kind: "integrated"; integratedAt: string }
+  | { kind: "conflict"; conflictedFiles: string[] };
+
+/**
+ * Injectable git mechanics for the ordered integration stage (KTD-11). Production
+ * (executor.ts) implements these over real git — `integrate` does a rebase /
+ * cherry-pick of the instance branch onto the task's main branch and uses
+ * merger.ts's `getConflictedFiles` to detect conflicts; `discardBranch` deletes
+ * the conflicting branch. Tests inject fakes for fast, deterministic runs.
+ */
+export interface IntegrationGitOps {
+  /**
+   * Land `branchName` onto the integration base (the task's main branch) for step
+   * `stepIndex`. Returns `integrated` on a clean rebase/cherry-pick (the base now
+   * contains the step's commits), or `conflict` with the conflicting file list.
+   * MUST NOT mutate the projection or instance rows — that is the queue's job
+   * (projection-first ordering). On `conflict` the implementation MUST leave the
+   * base clean (abort the rebase) so the next instance can integrate.
+   */
+  integrate(
+    branchName: string,
+    stepIndex: number,
+  ): Promise<IntegrationAttemptResult>;
+  /**
+   * Discard a conflicting (or abandoned) instance branch and release its worktree
+   * (pool hygiene). Best-effort — never throws into the queue.
+   */
+  discardBranch(branchName: string, stepIndex: number): Promise<void>;
+}
+
+/** Projection + persistence side-effects the queue performs on a successful
+ *  integration (KTD-7 projection-first ordering). Injected so the queue stays
+ *  engine-agnostic and unit-testable. */
+export interface IntegrationProjection {
+  /**
+   * Flip the projection FIRST (KTD-7): `updateStep(taskId, stepIndex, "done")`
+   * with graph source so the dependency-order guard admits it. Awaited before the
+   * instance row flips to `completed`, closing the merge-blocker race.
+   */
+  markStepDone(stepIndex: number): Promise<void>;
+  /**
+   * Mark the instance row `completed` with `integratedAt` AFTER the projection
+   * flip (projection-first ordering). Optional — a fully in-memory run needs none.
+   */
+  markInstanceIntegrated?(stepIndex: number, integratedAt: string): Promise<void> | void;
+}
+
+/** One enqueued, completed instance awaiting ordered integration. */
+export interface PendingIntegration {
+  stepIndex: number;
+  branchName: string;
+}
+
+/** The disposition of one instance after the queue drained as far as it could. */
+export type InstanceIntegrationOutcome =
+  | { stepIndex: number; status: "integrated"; integratedAt: string }
+  | { stepIndex: number; status: "conflict"; conflictedFiles: string[] };
+
+/**
+ * Per-(task, run, foreach) ordered integration queue (KTD-11).
+ *
+ * Completed worktree-isolated instances enqueue via {@link enqueue}; the queue
+ * lands them onto the integration base **strictly in step order**. The scheduler
+ * calls {@link drain} whenever a new instance becomes available (or on each
+ * scheduler tick); `drain` integrates every contiguous run of ready instances
+ * starting at the lowest not-yet-resolved step index, stopping at the first gap
+ * (a step not yet completed) or the first conflict. Conflicts are reported back
+ * so the scheduler can route `outcome:integration-conflict` for that instance.
+ *
+ * The queue NEVER skips ahead past a gap: instance `i` integrates only after
+ * `0..i-1` are integrated or skipped, so completion-order inversion (a later step
+ * finishing first) cannot reorder integration — integration order is step order.
+ */
+export class IntegrationQueue {
+  /** Instances that have completed and are waiting to integrate, by step index. */
+  private readonly pending = new Map<number, PendingIntegration>();
+  /** Step indices whose integration is resolved (integrated OR routed conflict). */
+  private readonly resolved = new Set<number>();
+  /** The next step index eligible to integrate (advances as the queue drains). */
+  private cursor = 0;
+  /** Steps the scheduler told us to SKIP (e.g. dependency-failed) — treated as
+   *  resolved so the cursor advances past them without blocking later steps. */
+  private readonly skipped = new Set<number>();
+
+  constructor(
+    private readonly gitOps: IntegrationGitOps,
+    private readonly projection: IntegrationProjection,
+    private readonly pinnedStepCount: number,
+  ) {}
+
+  /** Enqueue a completed instance awaiting integration. Idempotent per step. */
+  enqueue(stepIndex: number, branchName: string): void {
+    if (this.resolved.has(stepIndex)) return;
+    this.pending.set(stepIndex, { stepIndex, branchName });
+  }
+
+  /**
+   * Mark a step index as skipped (resolved without integration), so the ordered
+   * cursor can advance past it. Used when an instance failed before producing a
+   * branch (the projection stays non-done; the foreach reports the failure).
+   */
+  skip(stepIndex: number): void {
+    if (this.resolved.has(stepIndex)) return;
+    this.skipped.add(stepIndex);
+    this.resolved.add(stepIndex);
+    this.advanceCursor();
+  }
+
+  /** Whether step `i` is still awaiting integration in the queue. */
+  isPending(stepIndex: number): boolean {
+    return this.pending.has(stepIndex);
+  }
+
+  /** Whether step `i` has been integrated or routed to conflict/skip. */
+  isResolved(stepIndex: number): boolean {
+    return this.resolved.has(stepIndex);
+  }
+
+  /**
+   * Integrate every contiguous ready instance starting at the cursor, in step
+   * order. Stops at the first gap (the cursor's step hasn't completed yet) or the
+   * first conflict (the conflicting step is reported and NOT marked resolved here
+   * — the scheduler routes it to rework, then re-enqueues or re-skips). Returns
+   * the per-instance outcomes produced THIS drain (callers act on conflicts).
+   */
+  async drain(): Promise<InstanceIntegrationOutcome[]> {
+    const outcomes: InstanceIntegrationOutcome[] = [];
+    for (;;) {
+      // Advance past any already-resolved/skipped steps so the cursor points at
+      // the lowest unresolved step.
+      this.advanceCursor();
+      if (this.cursor >= this.pinnedStepCount) break;
+
+      const ready = this.pending.get(this.cursor);
+      if (!ready) break; // Gap: the lowest unresolved step hasn't completed yet.
+
+      const result = await this.gitOps.integrate(ready.branchName, ready.stepIndex);
+      if (result.kind === "integrated") {
+        // Projection-first ordering (KTD-7): flip the step done BEFORE the instance
+        // row, then release the worktree (the discard path releases on conflict;
+        // here the worktree is released after a clean integration via discardBranch
+        // which the production op treats as "release, branch already merged").
+        await this.projection.markStepDone(ready.stepIndex);
+        await this.projection.markInstanceIntegrated?.(ready.stepIndex, result.integratedAt);
+        // Release the instance worktree post-integration (pool hygiene). The branch
+        // is already on the base; discardBranch in production only releases here.
+        await this.safeDiscard(ready.branchName, ready.stepIndex);
+        this.pending.delete(this.cursor);
+        this.resolved.add(this.cursor);
+        outcomes.push({
+          stepIndex: ready.stepIndex,
+          status: "integrated",
+          integratedAt: result.integratedAt,
+        });
+        this.advanceCursor();
+        continue;
+      }
+
+      // Conflict: discard the branch + release worktree, report the conflict, and
+      // STOP draining (the conflicting step is not resolved — the scheduler routes
+      // it to rework on the updated base, then re-enqueues a fresh branch or skips).
+      await this.safeDiscard(ready.branchName, ready.stepIndex);
+      this.pending.delete(this.cursor);
+      outcomes.push({
+        stepIndex: ready.stepIndex,
+        status: "conflict",
+        conflictedFiles: result.conflictedFiles,
+      });
+      break;
+    }
+    return outcomes;
+  }
+
+  /** True once every step index is resolved (integrated or skipped). */
+  isDrained(): boolean {
+    return this.resolved.size >= this.pinnedStepCount;
+  }
+
+  /** Drain-and-release any remaining pending branches (abort/cleanup path). */
+  async discardAllPending(): Promise<void> {
+    for (const { branchName, stepIndex } of this.pending.values()) {
+      await this.safeDiscard(branchName, stepIndex);
+    }
+    this.pending.clear();
+  }
+
+  private advanceCursor(): void {
+    while (this.cursor < this.pinnedStepCount && this.resolved.has(this.cursor)) {
+      this.cursor += 1;
+    }
+  }
+
+  private async safeDiscard(branchName: string, stepIndex: number): Promise<void> {
+    try {
+      await this.gitOps.discardBranch(branchName, stepIndex);
+    } catch (err) {
+      const message = err instanceof Error ? err.message : String(err);
+      schedulerLog.warn(`integration discardBranch failed for ${branchName} (step ${stepIndex}): ${message}`);
+    }
+  }
+}
diff --git a/packages/engine/src/workflow-graph-executor.ts b/packages/engine/src/workflow-graph-executor.ts
index 2f8410a24e..2f4d7c27d9 100644
--- a/packages/engine/src/workflow-graph-executor.ts
+++ b/packages/engine/src/workflow-graph-executor.ts
@@ -21,6 +21,7 @@ import {
 } from "./workflow-graph-branches.js";
 import {
   runForeach,
+  type ForeachEnvironment,
   type WorkflowStepInstancePersistence,
 } from "./workflow-graph-foreach.js";
 
@@ -97,6 +98,24 @@ export interface WorkflowGraphExecutorDeps {
    * runs (zero behavior change for non-foreach graphs).
    */
   signal?: AbortSignal;
+  /** Step-inversion (KTD-11, U10): per-instance worktree/branch allocation off the
+   *  integration base, for `isolation: "worktree"`. Absent → worktree isolation
+   *  fails cleanly (shared isolation is unaffected). */
+  allocateInstanceWorktree?: ForeachEnvironment["allocateInstanceWorktree"];
+  /** Step-inversion (KTD-11, U10): resolve the current integration base (main tip)
+   *  so reworks land on the updated base. */
+  resolveIntegrationBase?: ForeachEnvironment["resolveIntegrationBase"];
+  /** Step-inversion (KTD-11, U10): ordered-integration git mechanics (rebase /
+   *  cherry-pick + conflict detection via merger helpers). */
+  integrationGitOps?: ForeachEnvironment["integrationGitOps"];
+  /** Step-inversion (KTD-11, U10): projection-first integration writes
+   *  (updateStep done, then instance row). */
+  integrationProjection?: ForeachEnvironment["integrationProjection"];
+  /** Step-inversion (KTD-11, U10): non-blocking free-semaphore-slot accessor for
+   *  parallel scheduling (clamps concurrency without hold-and-wait). */
+  semaphoreAvailability?: ForeachEnvironment["semaphoreAvailability"];
+  /** Step-inversion (KTD-11, U10): crash-resume reconciliation hook. */
+  resumeReconcile?: ForeachEnvironment["resumeReconcile"];
 }
 
 export interface WorkflowGraphExecutorResult {
@@ -248,12 +267,19 @@ export class WorkflowGraphExecutor {
             runId,
             steps,
             context,
-            runTemplateNode: (tNode, sig) =>
-              this.executeNodeWithRetries(tNode, task, settings, context, sig),
+            runTemplateNode: (tNode, sig, contextOverride) =>
+              this.executeNodeWithRetries(tNode, task, settings, contextOverride ?? context, sig),
             shouldTraverseEdge: (edge, src) => this.shouldTraverseEdge(edge, src),
             persistence: this.deps.stepInstancePersistence,
             onReworkReset: this.deps.onReworkReset,
             signal: this.deps.signal,
+            // Worktree isolation + parallel scheduling (KTD-11, U10).
+            allocateInstanceWorktree: this.deps.allocateInstanceWorktree,
+            resolveIntegrationBase: this.deps.resolveIntegrationBase,
+            integrationGitOps: this.deps.integrationGitOps,
+            integrationProjection: this.deps.integrationProjection,
+            semaphoreAvailability: this.deps.semaphoreAvailability,
+            resumeReconcile: this.deps.resumeReconcile,
           });
           visitedNodeIds.push(...foreachResult.visitedNodeIds);
           const result: WorkflowNodeResult = {
diff --git a/packages/engine/src/workflow-graph-foreach.ts b/packages/engine/src/workflow-graph-foreach.ts
index be05b2ff99..d391aba4f7 100644
--- a/packages/engine/src/workflow-graph-foreach.ts
+++ b/packages/engine/src/workflow-graph-foreach.ts
@@ -6,6 +6,11 @@ import {
   FOREACH_ACTIVE_CONTEXT_KEY,
   type ForeachActiveContext,
 } from "./workflow-node-handlers.js";
+import {
+  IntegrationQueue,
+  type IntegrationGitOps,
+  type IntegrationProjection,
+} from "./step-integration.js";
 import { schedulerLog } from "./logger.js";
 
 /**
@@ -41,6 +46,10 @@ import { schedulerLog } from "./logger.js";
 const DEFAULT_MAX_REWORK_CYCLES = 3;
 /** Defensive cap mirroring core's validation clamp (KTD-5). */
 const MAX_REWORK_CYCLES_CAP = 10;
+/** Default parallel concurrency (KTD-3). */
+const DEFAULT_CONCURRENCY = 2;
+/** Hard cap on parallel concurrency (KTD-3). */
+const CONCURRENCY_CAP = 8;
 
 /** The foreach node's config shape this module reads (subset of WorkflowForeachConfig). */
 interface ForeachConfig {
@@ -72,12 +81,18 @@ export interface WorkflowStepInstanceState {
   pinnedStepCount: number;
   /** Template node id (NOT the materialized instance id) the instance is at. */
   currentNodeId: string;
-  status: "in-progress" | "completed" | "failed";
+  status: "in-progress" | "awaiting-integration" | "completed" | "failed";
   baselineSha?: string;
   checkpointId?: string;
   reworkCount: number;
   /** Latest authoritative step-review verdict (KTD-4/KTD-6, U5). */
   verdict?: "APPROVE" | "REVISE" | "RETHINK" | "UNAVAILABLE";
+  /** Worktree-isolation (KTD-11, U10): the instance's own branch name; null/absent
+   *  under shared isolation. */
+  branchName?: string;
+  /** Worktree-isolation (KTD-11, U10): ISO timestamp the branch integrated; absent
+   *  until the ordered integration stage lands it. */
+  integratedAt?: string;
 }
 
 export interface WorkflowStepInstancePersistence {
@@ -121,12 +136,19 @@ export interface ForeachEnvironment {
   /**
    * Runs one template node through the executor's executeNodeWithRetries (so
    * per-node maxRetries still applies inside the sub-walk). The node passed is
-   * the ORIGINAL template node; the executor reads/writes the shared context,
-   * which already carries `foreach:active` for the current instance.
+   * the ORIGINAL template node; the executor reads/writes the supplied context,
+   * which carries `foreach:active` for the current instance.
+   *
+   * `contextOverride` lets a worktree-isolated instance run on its OWN context
+   * object (KTD-11): under parallel scheduling concurrent instances must NOT share
+   * the single `foreach:active` slot, so each gets an isolated context clone. When
+   * omitted the shared `env.context` is used (shared-isolation sequential path —
+   * unchanged behavior).
    */
   runTemplateNode: (
     node: WorkflowIrNode,
     signal?: AbortSignal,
+    contextOverride?: Record<string, unknown>,
   ) => Promise<WorkflowNodeResult>;
   shouldTraverseEdge: (edge: WorkflowIrEdge, source: WorkflowNodeResult) => boolean;
   persistence?: WorkflowStepInstancePersistence;
@@ -145,6 +167,63 @@ export interface ForeachEnvironment {
   ) => void | Promise<void>;
   /** Honored between nodes (existing posture). */
   signal?: AbortSignal;
+
+  // ── Worktree isolation + parallel scheduling (KTD-11, U10) ────────────────
+  /**
+   * Allocate the instance's OWN worktree + branch off the current integration
+   * base (the task's main branch tip at instance start). Production wires this to
+   * the worktree pool / `createWorktree` with a canonical
+   * `fusion/<task>-step-<i>` branch name; tests inject a fake. Required when
+   * `isolation: "worktree"`; absent under shared isolation. The integration base
+   * is the SAME for sibling instances scheduled together (they branch from the
+   * common tip), and is the UPDATED main tip when an instance re-runs after an
+   * integration-conflict rework.
+   */
+  allocateInstanceWorktree?: (
+    stepIndex: number,
+    integrationBase: string | undefined,
+  ) => Promise<{ worktreePath: string; branchName: string }>;
+  /** Resolve the current integration base (main branch tip). Re-read before each
+   *  (re)allocation so a rework lands on the UPDATED base (KTD-11). Optional —
+   *  defaults to undefined (the allocator's own default base). */
+  resolveIntegrationBase?: () => Promise<string | undefined>;
+  /** Ordered-integration git mechanics (KTD-11). Required when `isolation:
+   *  "worktree"`; the queue uses it to land branches in step order. */
+  integrationGitOps?: IntegrationGitOps;
+  /**
+   * Projection + instance-row writes the integration stage performs on a clean
+   * integration (KTD-7 projection-first ordering). `markStepDone` flips the step
+   * `done` via `updateStep(source:"graph")`; `markInstanceIntegrated` flips the
+   * row to `completed`/`integratedAt`. Required when `isolation: "worktree"`.
+   */
+  integrationProjection?: IntegrationProjection;
+  /**
+   * Non-blocking semaphore availability accessor for parallel scheduling (KTD-11):
+   * how many slots are free for IMMEDIATE acquisition right now. The scheduler
+   * runs up to `min(concurrency, availability)` instances concurrently and
+   * degrades to fewer/sequential under contention — it NEVER hold-and-waits on
+   * slots while blocking integration (each instance acquires its own lease inside
+   * the step-execute seam like a normal session). Defaults to "unbounded" (the
+   * concurrency cap governs) when absent. A returned value ≤ 0 degrades to 1
+   * (always make forward progress; never deadlock).
+   */
+  semaphoreAvailability?: () => number;
+  /**
+   * Crash-resume reconciliation hook (KTD-11, U10). Before scheduling, the
+   * worktree scheduler reconciles each step instance against git/persistence truth:
+   *   - integrated (row `completed`/`integratedAt`)        → mark INTEGRATED (skip);
+   *   - branch exists but not integrated (`awaiting-integration`) → RE-ENTER the
+   *     integration queue (the branch is already built);
+   *   - branch missing                                     → RE-RUN the instance.
+   * Returns a per-step disposition; absent → all instances run fresh (cold start).
+   * The full self-healing sweep across runs is out of scope (handoff). `pinned` is
+   * the count this expansion pinned so the hook can reject a `pin-mismatch`.
+   */
+  resumeReconcile?: (
+    pinned: number,
+  ) =>
+    | Promise<Array<{ stepIndex: number; disposition: "integrated" | "reintegrate" | "rerun"; branchName?: string }>>
+    | Array<{ stepIndex: number; disposition: "integrated" | "reintegrate" | "rerun"; branchName?: string }>;
 }
 
 export interface ForeachRunResult {
@@ -168,6 +247,8 @@ function resolveForeachConfig(node: WorkflowIrNode): {
   template: { nodes: WorkflowIrNode[]; edges: WorkflowIrEdge[] };
   maxReworkCycles: number;
   mode: "sequential" | "parallel";
+  isolation: "shared" | "worktree";
+  concurrency: number;
 } {
   const cfg = (node.config ?? {}) as ForeachConfig;
   const template = cfg.template;
@@ -177,7 +258,57 @@ function resolveForeachConfig(node: WorkflowIrNode): {
   const raw = typeof cfg.maxReworkCycles === "number" ? cfg.maxReworkCycles : DEFAULT_MAX_REWORK_CYCLES;
   const maxReworkCycles = Math.max(1, Math.min(MAX_REWORK_CYCLES_CAP, Math.floor(raw)));
   const mode = cfg.mode === "parallel" ? "parallel" : "sequential";
-  return { template, maxReworkCycles, mode };
+  // Default isolation: worktree for parallel mode, shared for sequential (KTD-3).
+  // (Core validation rejects parallel+shared; this default mirrors that intent.)
+  const isolation: "shared" | "worktree" =
+    cfg.isolation === "worktree"
+      ? "worktree"
+      : cfg.isolation === "shared"
+      ? "shared"
+      : mode === "parallel"
+      ? "worktree"
+      : "shared";
+  const rawConc =
+    typeof cfg.concurrency === "number" && Number.isFinite(cfg.concurrency)
+      ? Math.floor(cfg.concurrency)
+      : DEFAULT_CONCURRENCY;
+  // Concurrency only meaningful in parallel mode; sequential pins to 1 (KTD-3).
+  const concurrency = mode === "parallel" ? Math.max(1, Math.min(CONCURRENCY_CAP, rawConc)) : 1;
+  return { template, maxReworkCycles, mode, isolation, concurrency };
+}
+
+/** The 0-indexed predecessor step indices instance `stepIndex` depends on. A step
+ *  with no annotation implicitly depends on the previous step (KTD-3), so an
+ *  unannotated plan is fully sequential regardless of mode. */
+function resolveDependsOn(steps: TaskStep[], stepIndex: number): number[] {
+  const deps = steps[stepIndex]?.dependsOn;
+  if (Array.isArray(deps) && deps.length > 0) return deps;
+  return stepIndex > 0 ? [stepIndex - 1] : [];
+}
+
+/**
+ * Validate the dependency DAG at expansion (KTD-3): every dependsOn index must be
+ * in range and strictly earlier (a step may only depend on lower indices), and
+ * the graph must be acyclic. Returns a refusal reason on violation, else null.
+ * Because dependsOn entries reference lower indices only, a forward-reference or
+ * self-reference is the cycle signature we reject.
+ */
+function validateDependencyDag(steps: TaskStep[]): string | null {
+  for (let i = 0; i < steps.length; i++) {
+    const deps = steps[i]?.dependsOn;
+    if (!Array.isArray(deps)) continue;
+    for (const d of deps) {
+      if (!Number.isInteger(d) || d < 0 || d >= steps.length) {
+        return `step ${i} depends on out-of-range step ${d}`;
+      }
+      if (d >= i) {
+        // A dependency on an equal/later index is the only way to form a cycle
+        // when edges always point to lower indices; reject it as an audited cycle.
+        return `dependency cycle: step ${i} depends on step ${d} (>= itself)`;
+      }
+    }
+  }
+  return null;
 }
 
 /** Find the single template entry node (no non-rework incoming edge). */
@@ -200,33 +331,21 @@ function findTemplateEntry(
   return entries[0];
 }
 
-/**
- * Expand a foreach node and run its instances sequentially in step order.
- * Returns the foreach node's aggregate outcome (KTD-3).
- */
-export async function runForeach(
+/** Compiled template state shared across instances (built once per expansion). */
+interface TemplatePlan {
+  templateById: Map<string, WorkflowIrNode>;
+  templateOutgoing: Map<string, WorkflowIrEdge[]>;
+  entry: WorkflowIrNode;
+  /** Whether the template routes an explicit `outcome:integration-conflict` edge
+   *  from any node (overrides the default rework routing — KTD-11). */
+  hasExplicitIntegrationConflictEdge: boolean;
+  templateHasStepReview: boolean;
+}
+
+function compileTemplate(
   foreachNode: WorkflowIrNode,
-  env: ForeachEnvironment,
-): Promise<ForeachRunResult> {
-  const { template, maxReworkCycles, mode } = resolveForeachConfig(foreachNode);
-
-  // U3 scope guard: parallel mode is U10. Fail cleanly with a routable outcome
-  // rather than silently running it as sequential.
-  if (mode === "parallel") {
-    return {
-      outcome: "failure",
-      value: "parallel-not-wired",
-      visitedNodeIds: [],
-    };
-  }
-
-  // Pin the count at expansion (KTD-3). Zero steps → success edge (no instances).
-  const pinnedStepCount = env.steps.length;
-  const visitedNodeIds: string[] = [];
-  if (pinnedStepCount === 0) {
-    return { outcome: "success", visitedNodeIds };
-  }
-
+  template: { nodes: WorkflowIrNode[]; edges: WorkflowIrEdge[] },
+): TemplatePlan {
   const templateById = new Map(template.nodes.map((n) => [n.id, n]));
   const templateOutgoing = new Map<string, WorkflowIrEdge[]>();
   for (const edge of template.edges) {
@@ -235,14 +354,52 @@ export async function runForeach(
     templateOutgoing.set(edge.from, list);
   }
   const entry = findTemplateEntry(template.nodes, template.edges, foreachNode.id);
-
-  // Single-authority done-marking (U6/KTD-4): when the template contains a
-  // step-review node, step-execute SUCCESS must leave the step in-progress and the
-  // review's APPROVE marks it done. Computed once and threaded into each instance.
+  const hasExplicitIntegrationConflictEdge = template.edges.some(
+    (e) => e.condition === "outcome:integration-conflict",
+  );
   const templateHasStepReview = template.nodes.some((n) => n.kind === "step-review");
+  return { templateById, templateOutgoing, entry, hasExplicitIntegrationConflictEdge, templateHasStepReview };
+}
 
-  // Sequential + shared: a runnable-set loop with concurrency 1 (U10 extends this
-  // to parallel/worktree). Instances run strictly in step order.
+/**
+ * Expand a foreach node and run its instances. Returns the foreach node's
+ * aggregate outcome (KTD-3). Dispatches on the two orthogonal axes (KTD-11):
+ *   - shared isolation (default sequential): the unchanged step-order loop —
+ *     work lands in the task's main worktree, done at step completion;
+ *   - worktree isolation (sequential OR parallel): per-instance branches off the
+ *     integration base, dependency-aware scheduling up to `min(concurrency, free
+ *     semaphore slots)`, ordered integration in step order, integration-conflict
+ *     routed as rework on the updated base.
+ */
+export async function runForeach(
+  foreachNode: WorkflowIrNode,
+  env: ForeachEnvironment,
+): Promise<ForeachRunResult> {
+  const config = resolveForeachConfig(foreachNode);
+
+  // Pin the count at expansion (KTD-3). Zero steps → success edge (no instances).
+  const pinnedStepCount = env.steps.length;
+  const visitedNodeIds: string[] = [];
+  if (pinnedStepCount === 0) {
+    return { outcome: "success", visitedNodeIds };
+  }
+
+  // Dependency-cycle / out-of-range rejection at expansion (KTD-3, audited).
+  const dagViolation = validateDependencyDag(env.steps);
+  if (dagViolation) {
+    schedulerLog.warn(
+      `foreach ${foreachNode.id} for task ${env.task.id}: ${dagViolation} — failing expansion (dependency-cycle)`,
+    );
+    return { outcome: "failure", value: "dependency-cycle", visitedNodeIds };
+  }
+
+  const plan = compileTemplate(foreachNode, config.template);
+
+  if (config.isolation === "worktree") {
+    return runForeachWorktree(foreachNode, env, config, plan, pinnedStepCount, visitedNodeIds);
+  }
+
+  // ── shared isolation (default sequential) — UNCHANGED U3 behavior ──────────
   for (let stepIndex = 0; stepIndex < pinnedStepCount; stepIndex++) {
     if (env.signal?.aborted) {
       return { outcome: "failure", value: "aborted", visitedNodeIds };
@@ -252,13 +409,13 @@ export async function runForeach(
       foreachNode,
       stepIndex,
       pinnedStepCount,
-      entry,
-      templateById,
-      templateOutgoing,
-      maxReworkCycles,
+      plan.entry,
+      plan.templateById,
+      plan.templateOutgoing,
+      config.maxReworkCycles,
       env,
       visitedNodeIds,
-      templateHasStepReview,
+      plan.templateHasStepReview,
     );
 
     if (instanceResult.outcome === "failure") {
@@ -275,6 +432,335 @@ export async function runForeach(
   return { outcome: "success", visitedNodeIds };
 }
 
+// ── Worktree isolation + dependency scheduler + ordered integration (KTD-11) ──
+
+type InstanceState = "pending" | "running" | "awaiting-integration" | "integrated" | "failed";
+
+interface WorktreeInstance {
+  stepIndex: number;
+  state: InstanceState;
+  /** Per-instance rework budget — survives integration-conflict re-executions. */
+  reworkBudget: number;
+  reworkCount: number;
+  /** The instance's allocated branch (set on each run). */
+  branchName?: string;
+  worktreePath?: string;
+  baselineSha?: string;
+  checkpointId?: string;
+  verdict?: ForeachActiveContext["verdict"];
+}
+
+/**
+ * Worktree-isolation foreach (KTD-11, U10): per-instance branches off the
+ * integration base, dependency-aware scheduling up to `min(concurrency, free
+ * semaphore slots)`, ordered integration in step order, integration-conflict
+ * routed as rework on the updated base.
+ *
+ * Scheduling: an instance is runnable when all of its `dependsOn` steps are
+ * INTEGRATED (not merely completed). The runnable set runs concurrently up to the
+ * clamped slot count; each instance acquires its own semaphore lease inside the
+ * step-execute seam (we only READ availability non-blockingly to decide how many
+ * to launch — never hold-and-wait, so a starved semaphore degrades to sequential
+ * without deadlock). Completion enqueues the branch as `awaiting-integration`;
+ * after each batch the integration queue drains in step order.
+ */
+async function runForeachWorktree(
+  foreachNode: WorkflowIrNode,
+  env: ForeachEnvironment,
+  config: ReturnType<typeof resolveForeachConfig>,
+  plan: TemplatePlan,
+  pinnedStepCount: number,
+  visitedNodeIds: string[],
+): Promise<ForeachRunResult> {
+  if (!env.allocateInstanceWorktree || !env.integrationGitOps || !env.integrationProjection) {
+    // Worktree isolation requires the full wiring; fail cleanly (routable) rather
+    // than silently running shared-mode physics.
+    return { outcome: "failure", value: "worktree-isolation-unwired", visitedNodeIds };
+  }
+
+  const instances: WorktreeInstance[] = Array.from({ length: pinnedStepCount }, (_, i) => ({
+    stepIndex: i,
+    state: "pending",
+    reworkBudget: config.maxReworkCycles,
+    reworkCount: 0,
+  }));
+
+  const queue = new IntegrationQueue(
+    env.integrationGitOps,
+    env.integrationProjection,
+    pinnedStepCount,
+  );
+
+  // Crash-resume reconciliation (KTD-11): integrated → skip; branch-exists → re-enter
+  // the integration queue; branch-missing → re-run. Cold start (no hook) runs fresh.
+  if (env.resumeReconcile) {
+    try {
+      const dispositions = await env.resumeReconcile(pinnedStepCount);
+      for (const d of dispositions) {
+        const inst = instances[d.stepIndex];
+        if (!inst) continue;
+        if (d.disposition === "integrated") {
+          inst.state = "integrated";
+        } else if (d.disposition === "reintegrate" && d.branchName) {
+          // Branch already built — enqueue for ordered integration without re-running.
+          inst.state = "awaiting-integration";
+          inst.branchName = d.branchName;
+          queue.enqueue(d.stepIndex, d.branchName);
+        }
+        // "rerun" leaves the instance pending (default).
+      }
+    } catch (err) {
+      schedulerLog.warn(
+        `foreach ${foreachNode.id} for task ${env.task.id}: resume reconcile failed, running cold: ${err instanceof Error ? err.message : String(err)}`,
+      );
+    }
+  }
+
+  const isIntegrated = (i: number): boolean => instances[i].state === "integrated";
+  const depsIntegrated = (i: number): boolean =>
+    resolveDependsOn(env.steps, i).every((d) => isIntegrated(d));
+
+  // Run one instance's sub-walk in an isolated context + freshly-allocated
+  // worktree off the CURRENT integration base. Returns awaiting-integration (with
+  // the branch) on a clean sub-walk, or a failure outcome (rework-exhausted etc.).
+  const runOneInstance = async (inst: WorktreeInstance): Promise<{ outcome: WorkflowNodeOutcome; value?: string }> => {
+    const integrationBase = env.resolveIntegrationBase
+      ? await env.resolveIntegrationBase().catch(() => undefined)
+      : undefined;
+    let allocation: { worktreePath: string; branchName: string };
+    try {
+      allocation = await env.allocateInstanceWorktree!(inst.stepIndex, integrationBase);
+    } catch (err) {
+      schedulerLog.warn(
+        `foreach ${foreachNode.id} step ${inst.stepIndex}: worktree allocation failed: ${err instanceof Error ? err.message : String(err)}`,
+      );
+      return { outcome: "failure", value: "worktree-alloc-failed" };
+    }
+    inst.branchName = allocation.branchName;
+    inst.worktreePath = allocation.worktreePath;
+
+    return runWorktreeInstanceSubWalk(foreachNode, env, plan, pinnedStepCount, visitedNodeIds, inst);
+  };
+
+  // Schedule + integrate loop.
+  for (;;) {
+    if (env.signal?.aborted) {
+      await queue.discardAllPending();
+      return { outcome: "failure", value: "aborted", visitedNodeIds };
+    }
+
+    // Runnable = pending instances whose deps are all integrated.
+    const runnable = instances.filter((i) => i.state === "pending" && depsIntegrated(i.stepIndex));
+    const running = instances.filter((i) => i.state === "running").length;
+
+    if (runnable.length > 0) {
+      // Clamp concurrency by free semaphore slots (non-blocking, KTD-11). Degrade
+      // to at least 1 so we never deadlock waiting for slots we don't hold: a
+      // starved semaphore (availability ≤ 0) still launches one at a time.
+      const rawAvail = env.semaphoreAvailability ? env.semaphoreAvailability() : config.concurrency;
+      const avail = Math.max(1, rawAvail);
+      let slots = Math.min(config.concurrency, avail) - running;
+      // Force forward progress when nothing is running (never block on slots held).
+      if (slots <= 0 && running === 0) slots = 1;
+      const toLaunch = slots > 0 ? runnable.slice(0, slots) : [];
+
+      if (toLaunch.length > 0) {
+        for (const inst of toLaunch) inst.state = "running";
+        await Promise.all(
+          toLaunch.map(async (inst) => {
+            const result = await runOneInstance(inst);
+            if (result.outcome === "success") {
+              inst.state = "awaiting-integration";
+              queue.enqueue(inst.stepIndex, inst.branchName!);
+            } else {
+              inst.state = "failed";
+              // A failed instance is skipped in the ordered queue so the cursor can
+              // advance past it; the foreach reports the failure value below.
+              queue.skip(inst.stepIndex);
+              (inst as WorktreeInstance & { failValue?: string }).failValue = result.value;
+            }
+          }),
+        );
+        continue; // re-evaluate runnable set + drain after the batch.
+      }
+    }
+
+    // Drain the integration queue in step order; route conflicts to rework.
+    const outcomes = await queue.drain();
+    const progressed = outcomes.length > 0;
+    for (const outcome of outcomes) {
+      const inst = instances[outcome.stepIndex];
+      if (outcome.status === "integrated") {
+        inst.state = "integrated";
+      } else {
+        // integration-conflict: route as rework on the UPDATED base (KTD-11). The
+        // explicit `outcome:integration-conflict` edge (if authored) is honored by
+        // the sub-walk; the DEFAULT here is the rework path (re-execute on the
+        // updated base, budget-counted). Either way we re-run the instance.
+        if (inst.reworkBudget <= 0) {
+          inst.state = "failed";
+          queue.skip(inst.stepIndex);
+          (inst as WorktreeInstance & { failValue?: string }).failValue = "rework-exhausted";
+        } else {
+          inst.reworkBudget -= 1;
+          inst.reworkCount += 1;
+          inst.state = "pending"; // re-enter scheduling; re-allocates off updated base.
+          // Default routing IS the rework path (re-execute the step on the updated
+          // base, budget-counted — KTD-11). When the template authors an explicit
+          // `outcome:integration-conflict` edge, the re-run surfaces the conflict
+          // signal in the instance context so the author's node can branch on it
+          // (the edge overrides the implicit "from entry" rework).
+          (inst as WorktreeInstance & { lastIntegrationConflict?: boolean }).lastIntegrationConflict =
+            plan.hasExplicitIntegrationConflictEdge;
+          schedulerLog.log(
+            `foreach ${foreachNode.id} step ${inst.stepIndex}: integration-conflict — reworking on updated base (budget left ${inst.reworkBudget})`,
+          );
+        }
+      }
+    }
+
+    // Terminal conditions.
+    const failed = instances.find((i) => i.state === "failed");
+    if (failed) {
+      await queue.discardAllPending();
+      const value = (failed as WorktreeInstance & { failValue?: string }).failValue ?? "instance-failed";
+      return { outcome: "failure", value, visitedNodeIds };
+    }
+    if (instances.every((i) => i.state === "integrated")) {
+      return { outcome: "success", visitedNodeIds };
+    }
+
+    // No progress and nothing runnable/running → stuck (shouldn't happen with a
+    // valid DAG, but guard against a livelock).
+    const anyActive = instances.some((i) => i.state === "running" || i.state === "awaiting-integration");
+    const anyRunnable = instances.some((i) => i.state === "pending" && depsIntegrated(i.stepIndex));
+    if (!progressed && !anyActive && !anyRunnable) {
+      await queue.discardAllPending();
+      return { outcome: "failure", value: "scheduler-stuck", visitedNodeIds };
+    }
+  }
+}
+
+/**
+ * Run one worktree-isolated instance's iterative sub-walk in an ISOLATED context
+ * (its own `foreach:active` slot — required for concurrency), against the
+ * instance's own worktree/branch. Returns success (sub-walk reached the exit;
+ * the caller enqueues the branch for ordered integration — does NOT mark done
+ * here, KTD-11) or failure (rework-exhausted / aborted / node failure).
+ *
+ * RETHINK under worktree isolation is branch-scoped (KTD-11): `onReworkReset`
+ * resets the instance's OWN branch (the executor wires it to resetStepToBaseline
+ * against this instance's worktree), so the blast-radius guard is structural.
+ */
+async function runWorktreeInstanceSubWalk(
+  foreachNode: WorkflowIrNode,
+  env: ForeachEnvironment,
+  plan: TemplatePlan,
+  pinnedStepCount: number,
+  visitedNodeIds: string[],
+  inst: WorktreeInstance,
+): Promise<{ outcome: WorkflowNodeOutcome; value?: string }> {
+  const stepIndex = inst.stepIndex;
+  // Isolated per-instance context (NOT the shared env.context) so concurrent
+  // instances never collide on the `foreach:active` slot. Seeded from the shared
+  // context so handlers still see prior walk context (read-only-ish).
+  const instanceContext: Record<string, unknown> = { ...env.context };
+  const active: ForeachActiveContext = {
+    foreachNodeId: foreachNode.id,
+    stepIndex,
+    instanceId: `${foreachNode.id}#${stepIndex}`,
+    deferDoneToReview: plan.templateHasStepReview,
+    worktreePath: inst.worktreePath,
+    branchName: inst.branchName,
+    baselineSha: inst.baselineSha,
+    checkpointId: inst.checkpointId,
+    verdict: inst.verdict,
+  };
+  instanceContext[FOREACH_ACTIVE_CONTEXT_KEY] = active;
+  // Surface a prior integration-conflict to the author's nodes when the template
+  // authored an explicit `outcome:integration-conflict` edge (KTD-11 override).
+  const lastConflict = (inst as WorktreeInstance & { lastIntegrationConflict?: boolean }).lastIntegrationConflict;
+  if (lastConflict) instanceContext["integration:conflict"] = true;
+
+  const persist = (status: WorkflowStepInstanceState["status"], currentNodeId: string): Promise<void> =>
+    persistInstanceState(env.persistence, {
+      taskId: env.task.id,
+      runId: env.runId,
+      foreachNodeId: foreachNode.id,
+      stepIndex,
+      pinnedStepCount,
+      currentNodeId,
+      status,
+      baselineSha: active.baselineSha,
+      checkpointId: active.checkpointId,
+      reworkCount: inst.reworkCount,
+      verdict: active.verdict,
+      branchName: inst.branchName,
+    });
+
+  await persist("in-progress", plan.entry.id);
+
+  let currentId = plan.entry.id;
+  let lastResult: WorkflowNodeResult = { outcome: "success" };
+
+  for (;;) {
+    if (env.signal?.aborted) {
+      await persist("failed", currentId);
+      return { outcome: "failure", value: "aborted" };
+    }
+
+    const node = plan.templateById.get(currentId);
+    if (!node) throw new WorkflowIrError(`Unknown foreach template node: ${currentId}`);
+
+    visitedNodeIds.push(instanceNodeId(foreachNode.id, stepIndex, currentId));
+
+    lastResult = await env.runTemplateNode(node, env.signal, instanceContext);
+    syncActiveFromContext(instanceContext, active);
+    // Persist captured baseline/checkpoint back onto the instance (survives rework).
+    inst.baselineSha = active.baselineSha;
+    inst.checkpointId = active.checkpointId;
+    inst.verdict = active.verdict;
+
+    if (lastResult.outcome === "failure") {
+      await persist("failed", currentId);
+      return { outcome: "failure", value: lastResult.value };
+    }
+
+    const next = chooseNextEdge(currentId, plan.templateOutgoing, lastResult, env.shouldTraverseEdge);
+    if (!next) {
+      // Sub-walk exit — work complete on the branch; AWAIT INTEGRATION (not done).
+      await persist("awaiting-integration", currentId);
+      return { outcome: "success" };
+    }
+
+    if (next.kind === "rework") {
+      if (inst.reworkBudget <= 0) {
+        await persist("failed", currentId);
+        return { outcome: "failure", value: "rework-exhausted" };
+      }
+      inst.reworkBudget -= 1;
+      inst.reworkCount += 1;
+
+      if (lastResult.value === "rethink" && env.onReworkReset) {
+        try {
+          // Branch-scoped RETHINK: resets THIS instance's branch only (KTD-11).
+          await env.onReworkReset(active, "rethink");
+          syncActiveFromContext(instanceContext, active);
+          inst.baselineSha = active.baselineSha;
+          inst.checkpointId = active.checkpointId;
+        } catch (err) {
+          schedulerLog.warn(
+            `onReworkReset failed for task ${env.task.id} foreach ${foreachNode.id} step ${stepIndex}: ${err instanceof Error ? err.message : String(err)}`,
+          );
+        }
+      }
+      await persist("in-progress", next.to);
+    }
+
+    currentId = next.to;
+  }
+}
+
 interface InstanceResult {
   outcome: WorkflowNodeOutcome;
   value?: string;
diff --git a/packages/engine/src/workflow-graph-task-runner.ts b/packages/engine/src/workflow-graph-task-runner.ts
index f20cfaeb0d..e68fea9cf8 100644
--- a/packages/engine/src/workflow-graph-task-runner.ts
+++ b/packages/engine/src/workflow-graph-task-runner.ts
@@ -14,7 +14,7 @@ import type {
   WorkflowBranchProgress,
   WorkflowBranchSemaphore,
 } from "./workflow-graph-branches.js";
-import type { WorkflowStepInstancePersistence } from "./workflow-graph-foreach.js";
+import type { ForeachEnvironment, WorkflowStepInstancePersistence } from "./workflow-graph-foreach.js";
 // (Both types are also used as values in the side-effect tracking wrappers below.)
 
 /**
@@ -69,6 +69,14 @@ export interface WorkflowGraphTaskRunnerDeps {
   /** Step-inversion (U14, KTD-15): `code` node runner. Additive; a workflow with
    *  no code node never invokes it. */
   runCode?: CodeNodeRunner;
+  /** Step-inversion (KTD-11, U10): worktree-isolation + parallel-scheduling deps.
+   *  Additive; a shared-isolation foreach never invokes them. */
+  allocateInstanceWorktree?: ForeachEnvironment["allocateInstanceWorktree"];
+  resolveIntegrationBase?: ForeachEnvironment["resolveIntegrationBase"];
+  integrationGitOps?: ForeachEnvironment["integrationGitOps"];
+  integrationProjection?: ForeachEnvironment["integrationProjection"];
+  semaphoreAvailability?: ForeachEnvironment["semaphoreAvailability"];
+  resumeReconcile?: ForeachEnvironment["resumeReconcile"];
 }
 
 /**
@@ -174,6 +182,13 @@ export class WorkflowGraphTaskRunner {
         onReworkReset: this.deps.onReworkReset,
         parseStepsDeps: this.deps.parseStepsDeps,
         runCode: this.deps.runCode,
+        // Step-inversion (KTD-11, U10): worktree isolation + parallel scheduling.
+        allocateInstanceWorktree: this.deps.allocateInstanceWorktree,
+        resolveIntegrationBase: this.deps.resolveIntegrationBase,
+        integrationGitOps: this.deps.integrationGitOps,
+        integrationProjection: this.deps.integrationProjection,
+        semaphoreAvailability: this.deps.semaphoreAvailability,
+        resumeReconcile: this.deps.resumeReconcile,
         runId: `${task.id}:${definition.id}`,
         onBranchProgress: (progress) => {
           this.branchProgress.set(progress.branchId, progress);
diff --git a/packages/engine/src/workflow-node-handlers.ts b/packages/engine/src/workflow-node-handlers.ts
index 0f5805579d..8c0ca3c2cd 100644
--- a/packages/engine/src/workflow-node-handlers.ts
+++ b/packages/engine/src/workflow-node-handlers.ts
@@ -99,6 +99,20 @@ export interface ForeachActiveContext {
    * foreach sub-walk sets this at instance entry; the step-execute seam reads it.
    */
   deferDoneToReview?: boolean;
+  /**
+   * Worktree-isolation (KTD-11, U10): the instance's OWN worktree path, branched
+   * off the integration base. Set by the foreach sub-walk at instance entry under
+   * `isolation: "worktree"`; the step-execute / step-review / RETHINK seams run
+   * against THIS path instead of the task's main worktree. Absent under
+   * `isolation: "shared"` (work lands directly in the main worktree). The file-scope
+   * guard still fires for anything the instance session commits in this worktree
+   * (the session machinery is unchanged — see executor stepExecute seam).
+   */
+  worktreePath?: string;
+  /** Worktree-isolation (KTD-11, U10): the instance's OWN branch name (e.g.
+   *  `fusion/<task>-step-<i>`). Set with {@link worktreePath}; the ordered
+   *  integration stage lands this branch onto the task's main branch. */
+  branchName?: string;
 }
 
 /**
diff --git a/packages/engine/src/worktree-names.ts b/packages/engine/src/worktree-names.ts
index 19a807495f..055a036350 100644
--- a/packages/engine/src/worktree-names.ts
+++ b/packages/engine/src/worktree-names.ts
@@ -33,6 +33,17 @@ export function canonicalFusionBranchName(taskId: string): string {
   return `fusion/${taskId.toLowerCase()}`;
 }
 
+/**
+ * Canonical per-instance branch name for a worktree-isolated foreach step
+ * (step-inversion KTD-11, U10): `fusion/<task>-step-<i>`. Deterministic from the
+ * task id + 0-based step index so crash-resume can reconstruct the branch name
+ * (and probe its existence) without persisting it separately — though the
+ * instance row also carries `branchName` for the integration/reconcile path.
+ */
+export function canonicalStepInstanceBranchName(taskId: string, stepIndex: number): string {
+  return `${canonicalFusionBranchName(taskId)}-step-${stepIndex}`;
+}
+
 export function resolveTaskWorkingBranch(task: Pick<Task, "id" | "branch" | "branchContext">): string {
   if (task.branchContext?.assignmentMode === "shared") {
     return canonicalFusionBranchName(task.id);