fusion/packages/engine/src/workflow-graph-foreach.ts

import type { TaskDetail, TaskStep, WorkflowIrEdge, WorkflowIrNode } from "@fusion/core";
import { WorkflowIrError, instanceNodeId, resolveMaxReworkCycles } from "@fusion/core";

import type { WorkflowNodeOutcome, WorkflowNodeResult } from "./workflow-graph-executor.js";
import {
  FOREACH_ACTIVE_CONTEXT_KEY,
  INTEGRATION_CONFLICT_CONTEXT_KEY,
  type ForeachActiveContext,
} from "./workflow-node-handlers.js";
import {
  IntegrationQueue,
  type IntegrationGitOps,
  type IntegrationProjection,
} from "./step-integration.js";
import { schedulerLog } from "./logger.js";

/**
 * Foreach region expansion + instance sub-walk (step-inversion KTD-3/KTD-5, U3).
 *
 * When the sequential walker reaches a `foreach` node it does NOT recurse through
 * the main `walk` (whose `inStack` cycle detector intentionally throws on any
 * back-edge). Instead it hands control here, which:
 *
 *   - reads `Task.steps[]` and pins the count at expansion time;
 *   - for each step `i` in order, runs the inline template subgraph as an
 *     **iterative region sub-walk** (a `for(;;)` over `currentId`, modeled on
 *     `walkBranch` in workflow-graph-branches.ts), from the template entry to its
 *     exit, materializing deterministic instance node ids
 *     `<foreachId>#<i>:<templateNodeId>` purely as walk state (the IR/nodeMap are
 *     never mutated);
 *   - permits `kind: "rework"` edges as the only legal cycles — each traversal
 *     decrements a per-instance budget seeded from `config.maxReworkCycles`
 *     (default 3, defensively clamped to ≤10); exhaustion emits the
 *     `outcome:rework-exhausted` outcome from the foreach node;
 *   - threads the active instance under the reserved `foreach:active` context key
 *     so template handlers (step-execute now; step-review in U5) know which step
 *     they operate on, clearing it on instance exit;
 *   - honors the abort signal between nodes (existing posture).
 *
 * Only sequential + shared physics are implemented here (concurrency 1). The
 * scheduler is intentionally a runnable-set loop running one instance at a time
 * so U10 can extend it to parallel/worktree without restructuring. Parallel mode
 * is guarded to a clean failure (U10 replaces it).
 */

// Rework budget default + clamp live in @fusion/core (DEFAULT_MAX_REWORK_CYCLES /
// MAX_REWORK_CYCLES_CAP / resolveMaxReworkCycles) so the foreach sub-walk and the
// top-level PR review loop (U6) share one definition and cannot drift.
/** 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 {
  source?: unknown;
  maxReworkCycles?: number;
  mode?: "sequential" | "parallel";
  concurrency?: number;
  isolation?: "shared" | "worktree";
  template?: { nodes: WorkflowIrNode[]; edges: WorkflowIrEdge[] };
}

/**
 * Narrow persistence hook for foreach instance run-state (KTD-6, U3 stub).
 *
 * The real SQLite-backed adapter lands in U4 (executor half); this interface is
 * shaped so that wiring is a pure additive change. All methods are optional and
 * default to no-ops — the sub-walk calls them at instance start / completion /
 * each rework pass, but a fully in-memory run (tests, flag-off, pre-U4 store)
 * needs none of them. Instance identity is deterministic
 * (`<foreachNodeId>#<stepIndex>`), so a future resume can seed the sub-walk
 * position directly from a loaded `currentNodeId` + `reworkCount` (KTD-6) —
 * this hook is the seam where that seeding will plug in.
 */
export interface WorkflowStepInstanceState {
  taskId: string;
  runId: string;
  foreachNodeId: string;
  stepIndex: number;
  pinnedStepCount: number;
  /** Template node id (NOT the materialized instance id) the instance is at. */
  currentNodeId: string;
  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 {
  /** Idempotent upsert keyed by (taskId, runId, foreachNodeId, stepIndex). */
  saveInstanceState?(state: WorkflowStepInstanceState): void | Promise<void>;
  /** Load any persisted instance states for a run (used on resume — U4). */
  loadInstanceStates?(
    taskId: string,
    runId: string,
  ): WorkflowStepInstanceState[] | Promise<WorkflowStepInstanceState[]>;
  /** Prune stale instance rows for a task, keeping only `keepRunId` (U4). */
  clearStaleInstanceStates?(taskId: string, keepRunId: string): void | Promise<void>;
}

/**
 * Await a persistence call inside a guard so a Promise-returning impl cannot
 * escape as an unhandled rejection, and a persistence failure never kills
 * instance execution (log-and-continue). Mirrors `persistBranchState`.
 */
async function persistInstanceState(
  persistence: WorkflowStepInstancePersistence | undefined,
  state: WorkflowStepInstanceState,
): Promise<void> {
  try {
    await persistence?.saveInstanceState?.(state);
  } catch (err) {
    const message = err instanceof Error ? err.message : String(err);
    schedulerLog.warn(
      `saveInstanceState failed for task ${state.taskId} run ${state.runId} foreach ${state.foreachNodeId} step ${state.stepIndex}: ${message}`,
    );
  }
}

export interface ForeachEnvironment {
  task: TaskDetail;
  runId: string;
  /** Fresh step list (KTD-3: read at expansion, count pinned). */
  steps: TaskStep[];
  /** The shared walk context; the active-instance key is threaded in/out of it. */
  context: Record<string, unknown>;
  /**
   * 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 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;
  /**
   * RETHINK reset-on-rework hook (KTD-4, U5). Invoked BEFORE re-entering the
   * instance's step-execute node when the rework edge being traversed was
   * triggered by an `outcome:rethink` (the verdict that resets to baseline). The
   * production wiring (executor.ts) calls `resetStepToBaseline` with the
   * instance's persisted `baselineSha`/`checkpointId`; tests inject a fake. Other
   * rework outcomes (e.g. `revise`) do NOT call this — they revise in place
   * (today's REVISE semantics). Optional with a no-op default.
   */
  onReworkReset?: (
    active: ForeachActiveContext,
    reason: string,
  ) => 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 }>;
  /**
   * Optional task-level log sink (FIX 4 — context gap). When an integration
   * conflict routes a step instance to rework, the re-running agent has no record
   * of WHY its base changed. Wiring this to the engine's task log path
   * (`store.logEntry`) writes a visible entry so the rework is explicable. Best
   * effort: a logging failure must never affect the run. Absent under tests / a
   * foreach env without a logging dep (the conflict still logs to schedulerLog).
   */
  logTaskEntry?: (summary: string, detail?: string) => void | Promise<void>;
}

export interface ForeachRunResult {
  /** Foreach node outcome: success when all instances completed; otherwise the
   *  routed outcome value (e.g. "rework-exhausted") with a failure outcome unless
   *  the caller routes it. */
  outcome: WorkflowNodeOutcome;
  /** Outcome value for `outcome:` edge routing (e.g. "rework-exhausted"). */
  value?: string;
  /** Materialized instance node ids visited, for the executor's visited list. */
  visitedNodeIds: string[];
}

// `instanceNodeId` now lives in `@fusion/core` (column-agent plan KTD-2) so the
// instance-id format has exactly one owner. Re-exported here (the imported binding)
// for back-compat with any local callers; the format is unchanged.
export { instanceNodeId };

/** Resolve the foreach config, validating the bits this module relies on. */
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;
  if (!template || !Array.isArray(template.nodes) || !Array.isArray(template.edges)) {
    throw new WorkflowIrError(`foreach node '${node.id}' has no template subgraph`);
  }
  const maxReworkCycles = resolveMaxReworkCycles(cfg.maxReworkCycles);
  const mode = cfg.mode === "parallel" ? "parallel" : "sequential";
  // 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). */
function findTemplateEntry(
  nodes: WorkflowIrNode[],
  edges: WorkflowIrEdge[],
  foreachId: string,
): WorkflowIrNode {
  const incoming = new Map<string, number>();
  for (const edge of edges) {
    if (edge.kind === "rework") continue;
    incoming.set(edge.to, (incoming.get(edge.to) ?? 0) + 1);
  }
  const entries = nodes.filter((n) => (incoming.get(n.id) ?? 0) === 0);
  if (entries.length !== 1) {
    throw new WorkflowIrError(
      `foreach node '${foreachId}' template must have exactly one entry node (found ${entries.length})`,
    );
  }
  return entries[0];
}

/** 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,
  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) {
    const list = templateOutgoing.get(edge.from) ?? [];
    list.push(edge);
    templateOutgoing.set(edge.from, list);
  }
  const entry = findTemplateEntry(template.nodes, template.edges, foreachNode.id);
  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 };
}

/**
 * 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 };
    }

    /**
     * Engine restart resume semantics: shared-isolation foreach replays the
     * graph from the foreach node, but task steps already persisted as terminal
     * must not re-run their step-execute instance handlers.
     */
    const stepStatus = env.steps[stepIndex]?.status;
    if (stepStatus === "done" || stepStatus === "skipped") {
      schedulerLog.log(
        `foreach ${foreachNode.id} for task ${env.task.id}: skipping step ${stepIndex} — already ${stepStatus}`,
      );
      continue;
    }

    const instanceResult = await runInstance(
      foreachNode,
      stepIndex,
      pinnedStepCount,
      plan.entry,
      plan.templateById,
      plan.templateOutgoing,
      config.maxReworkCycles,
      env,
      visitedNodeIds,
      plan.templateHasStepReview,
    );

    if (instanceResult.outcome === "failure") {
      // Rework exhaustion routes a dedicated outcome; other failures propagate.
      return {
        outcome: "failure",
        value: instanceResult.value,
        visitedNodeIds,
      };
    }
  }

  // All instances completed → foreach success edge (KTD-3).
  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,
    // Single source of truth for the instance-row identity (KTD-6/KTD-11): the
    // REAL runId + foreachNodeId the sub-walk persisted rows under, so integration
    // flips the SAME row instead of writing an orphan.
    { runId: env.runId, foreachNodeId: foreachNode.id },
  );

  // 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));

  // Track instances whose worktree/branch has been released so a terminal sweep
  // (and the integration queue) never double-discards. The integration queue
  // releases integrated/conflicted branches itself; this set covers the branches
  // we release directly (failed / aborted / stuck instances that allocated a
  // worktree but the queue never resolved). discardBranch is best-effort either
  // way, but the guard keeps the release "exactly once" contract honest (FIX 2).
  const released = new Set<number>();
  const releaseInstanceWorktree = async (inst: WorktreeInstance): Promise<void> => {
    if (!inst.branchName) return; // never allocated.
    if (released.has(inst.stepIndex)) return;
    released.add(inst.stepIndex);
    try {
      await env.integrationGitOps!.discardBranch(inst.branchName, inst.stepIndex);
    } catch (err) {
      schedulerLog.warn(
        `foreach ${foreachNode.id} step ${inst.stepIndex}: worktree release failed: ${err instanceof Error ? err.message : String(err)}`,
      );
    }
  };
  // Terminal cleanup: release every allocated instance the integration queue did
  // NOT own. The queue itself releases `integrated` (post-integration discard) and
  // `awaiting-integration` (via discardAllPending on the abort/fail/stuck paths)
  // and conflicted branches (marked in `released`). This sweep covers the
  // remainder — `failed` / `running` / `pending` instances that allocated a
  // worktree — so failure / rework-exhausted / abort / scheduler-stuck never leak
  // an allocated worktree+branch. The `released` guard keeps it exactly once.
  const releaseUnresolvedAllocated = async (): Promise<void> => {
    await Promise.all(
      instances
        .filter(
          (i) =>
            i.branchName &&
            i.state !== "integrated" &&
            i.state !== "awaiting-integration",
        )
        .map((i) => releaseInstanceWorktree(i)),
    );
  };

  // 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;
    // Fresh allocation: this instance now holds a live, un-released worktree again
    // (a prior conflict/rework may have flagged it released). Clear the marker so a
    // later failure of THIS run releases the new worktree exactly once.
    released.delete(inst.stepIndex);

    return runWorktreeInstanceSubWalk(foreachNode, env, plan, pinnedStepCount, visitedNodeIds, inst);
  };

  // Schedule + integrate loop.
  for (;;) {
    if (env.signal?.aborted) {
      await queue.discardAllPending();
      await releaseUnresolvedAllocated();
      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. Release
              // its allocated worktree+branch immediately (FIX 2 — the queue never
              // sees a failed instance, so it would otherwise leak).
              queue.skip(inst.stepIndex);
              await releaseInstanceWorktree(inst);
              (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: the queue's drain() already discarded the
        // conflicting branch (safeDiscard) before reporting the conflict, so the
        // branch is released — record it so the terminal sweep never double-discards.
        released.add(inst.stepIndex);
        // 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;
          const conflictedFiles =
            outcome.status === "conflict" && Array.isArray(outcome.conflictedFiles)
              ? outcome.conflictedFiles
              : [];
          schedulerLog.log(
            `foreach ${foreachNode.id} step ${inst.stepIndex}: integration-conflict — reworking on updated base (budget left ${inst.reworkBudget})`,
          );
          // Task-level audit so the re-running agent sees WHY its base moved
          // (FIX 4). Best-effort; a logging failure never affects the run.
          try {
            const filesNote = conflictedFiles.length > 0 ? ` (files: ${conflictedFiles.join(", ")})` : "";
            await env.logTaskEntry?.(
              `integration conflict on step ${inst.stepIndex}: reworking on updated base${filesNote}`,
              conflictedFiles.length > 0 ? `Conflicted files:\n${conflictedFiles.map((f) => `- ${f}`).join("\n")}` : undefined,
            );
          } catch { /* log is best-effort */ }
        }
      }
    }

    // Terminal conditions.
    const failed = instances.find((i) => i.state === "failed");
    if (failed) {
      await queue.discardAllPending();
      await releaseUnresolvedAllocated();
      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();
      await releaseUnresolvedAllocated();
      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).
  // Consume the one-shot signal here: clear it on the instance so a LATER clean
  // rework (e.g. a fresh rethink with no conflict) does not re-surface a stale
  // conflict on the seeded context.
  const conflictHolder = inst as WorktreeInstance & { lastIntegrationConflict?: boolean };
  if (conflictHolder.lastIntegrationConflict) {
    instanceContext[INTEGRATION_CONFLICT_CONTEXT_KEY] = true;
    conflictHolder.lastIntegrationConflict = false;
  }

  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;
}

/**
 * Run one foreach instance (step `stepIndex`) as an iterative region sub-walk.
 * Threads `foreach:active` into the shared context on entry and clears it on
 * exit. Rework edges loop `currentId` back, bounded by the per-instance budget.
 */
async function runInstance(
  foreachNode: WorkflowIrNode,
  stepIndex: number,
  pinnedStepCount: number,
  entry: WorkflowIrNode,
  templateById: Map<string, WorkflowIrNode>,
  templateOutgoing: Map<string, WorkflowIrEdge[]>,
  maxReworkCycles: number,
  env: ForeachEnvironment,
  visitedNodeIds: string[],
  templateHasStepReview: boolean,
): Promise<InstanceResult> {
  // Per-instance rework budget (KTD-5) — NOT shared across instances.
  let reworkBudget = maxReworkCycles;
  let reworkCount = 0;

  // Active-instance context (KTD-3). baselineSha/checkpointId start undefined and
  // are captured by step-execute (U3) into this same object so later template
  // nodes (step-review/reset, U5) can read them. deferDoneToReview tells the
  // step-execute seam to leave the step in-progress when a review will decide
  // done-ness (U6/KTD-4).
  const active: ForeachActiveContext = {
    foreachNodeId: foreachNode.id,
    stepIndex,
    instanceId: `${foreachNode.id}#${stepIndex}`,
    deferDoneToReview: templateHasStepReview,
  };
  env.context[FOREACH_ACTIVE_CONTEXT_KEY] = active;

  await persistInstanceState(env.persistence, {
    taskId: env.task.id,
    runId: env.runId,
    foreachNodeId: foreachNode.id,
    stepIndex,
    pinnedStepCount,
    currentNodeId: entry.id,
    status: "in-progress",
    baselineSha: active.baselineSha,
    checkpointId: active.checkpointId,
    reworkCount,
    verdict: active.verdict,
  });

  try {
    let currentId = entry.id;
    let lastResult: WorkflowNodeResult = { outcome: "success" };

    for (;;) {
      if (env.signal?.aborted) {
        await persistInstanceState(env.persistence, {
          taskId: env.task.id,
          runId: env.runId,
          foreachNodeId: foreachNode.id,
          stepIndex,
          pinnedStepCount,
          currentNodeId: currentId,
          status: "failed",
          baselineSha: active.baselineSha,
          checkpointId: active.checkpointId,
          reworkCount,
          verdict: active.verdict,
        });
        return { outcome: "failure", value: "aborted" };
      }

      const node = 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);
      // step-execute (and U5 nodes) write captured baseline/checkpoint into the
      // active context via their contextPatch; mirror them onto `active` so the
      // reserved key stays the single source of truth for later nodes.
      syncActiveFromContext(env.context, active);

      if (lastResult.outcome === "failure") {
        await persistInstanceState(env.persistence, {
          taskId: env.task.id,
          runId: env.runId,
          foreachNodeId: foreachNode.id,
          stepIndex,
          pinnedStepCount,
          currentNodeId: currentId,
          status: "failed",
          baselineSha: active.baselineSha,
          checkpointId: active.checkpointId,
          reworkCount,
          verdict: active.verdict,
        });
        return { outcome: "failure", value: lastResult.value };
      }

      // Pick the next edge. Rework edges are the only legal back-edges.
      const next = chooseNextEdge(currentId, templateOutgoing, lastResult, env.shouldTraverseEdge);
      if (!next) {
        // No outgoing edge matched → template exit reached. Instance complete.
        await persistInstanceState(env.persistence, {
          taskId: env.task.id,
          runId: env.runId,
          foreachNodeId: foreachNode.id,
          stepIndex,
          pinnedStepCount,
          currentNodeId: currentId,
          status: "completed",
          baselineSha: active.baselineSha,
          checkpointId: active.checkpointId,
          reworkCount,
          verdict: active.verdict,
        });
        return { outcome: "success" };
      }

      if (next.kind === "rework") {
        if (reworkBudget <= 0) {
          // Budget exhausted (KTD-5): emit rework-exhausted from the foreach node.
          await persistInstanceState(env.persistence, {
            taskId: env.task.id,
            runId: env.runId,
            foreachNodeId: foreachNode.id,
            stepIndex,
            pinnedStepCount,
            currentNodeId: currentId,
            status: "failed",
            baselineSha: active.baselineSha,
            checkpointId: active.checkpointId,
            reworkCount,
            verdict: active.verdict,
          });
          return { outcome: "failure", value: "rework-exhausted" };
        }
        reworkBudget -= 1;
        reworkCount += 1;

        // RETHINK reset-on-rework (KTD-4, U5): when the rework edge was triggered
        // by an `outcome:rethink` verdict, reset the step to its per-step baseline
        // (git reset + session rewind + step→pending) BEFORE re-entering the
        // step-execute node. REVISE-driven rework revises in place — no reset.
        if (lastResult.value === "rethink" && env.onReworkReset) {
          try {
            await env.onReworkReset(active, "rethink");
            // The reset may have rewound the session; re-sync captured state.
            syncActiveFromContext(env.context, active);
          } catch (err) {
            const message = err instanceof Error ? err.message : String(err);
            schedulerLog.warn(
              `onReworkReset failed for task ${env.task.id} foreach ${foreachNode.id} step ${stepIndex}: ${message}`,
            );
          }
        }

        await persistInstanceState(env.persistence, {
          taskId: env.task.id,
          runId: env.runId,
          foreachNodeId: foreachNode.id,
          stepIndex,
          pinnedStepCount,
          currentNodeId: next.to,
          status: "in-progress",
          baselineSha: active.baselineSha,
          checkpointId: active.checkpointId,
          reworkCount,
          verdict: active.verdict,
        });
      }

      currentId = next.to;
    }
  } finally {
    // Clear the active-instance context on exit (KTD-3): absent outside instances.
    delete env.context[FOREACH_ACTIVE_CONTEXT_KEY];
  }
}

/** Sync baseline/checkpoint a handler wrote into the shared `foreach:active`
 *  context object back onto our local `active` snapshot. Handlers that patch the
 *  reserved key (step-execute) update the SAME object reference, but a handler
 *  could replace it via contextPatch — re-read defensively. */
function syncActiveFromContext(
  context: Record<string, unknown>,
  active: ForeachActiveContext,
): void {
  const fromContext = context[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext | undefined;
  if (fromContext && fromContext !== active) {
    active.baselineSha = fromContext.baselineSha ?? active.baselineSha;
    active.checkpointId = fromContext.checkpointId ?? active.checkpointId;
    active.verdict = fromContext.verdict ?? active.verdict;
    // Keep the canonical object reference stable for later nodes.
    context[FOREACH_ACTIVE_CONTEXT_KEY] = active;
  }
}

/**
 * Choose the single next edge from `nodeId`. A rework edge wins only when no
 * non-rework edge matches the outcome (rework is the explicit loop-back, not a
 * primary forward edge); among matching forward edges the lowest `to` id wins
 * (deterministic, mirrors walkBranch/traverseChildren ordering).
 */
function chooseNextEdge(
  nodeId: string,
  templateOutgoing: Map<string, WorkflowIrEdge[]>,
  source: WorkflowNodeResult,
  shouldTraverseEdge: (edge: WorkflowIrEdge, source: WorkflowNodeResult) => boolean,
): WorkflowIrEdge | undefined {
  const edges = (templateOutgoing.get(nodeId) ?? []).filter((e) => shouldTraverseEdge(e, source));
  if (edges.length === 0) return undefined;
  const forward = edges.filter((e) => e.kind !== "rework").sort((a, b) => a.to.localeCompare(b.to));
  if (forward.length > 0) return forward[0];
  const rework = edges.filter((e) => e.kind === "rework").sort((a, b) => a.to.localeCompare(b.to));
  return rework[0];
}