orchestratectl 0.1.5

//! The live, end-to-end code-pipeline driver (design.md §6 call diagram — the
//! FIRST bold-to-live wiring, breakdown T5 walking skeleton).
//!
//! This is the additive `orchestratectl pipeline run` command: it runs one
//! single feature through the whole loop —
//! **spec(Opus) → code[claude-deepseek] → floor-gate → verify(Opus) → merge** —
//! reusing every landed piece behind the seam:
//!
//! - the [`CodeHarness`](crate::harness::CodeHarness) adapters (the code stage
//!   is [`ClaudeHarness::deepseek`](crate::harness::claude::ClaudeHarness::deepseek));
//! - the deterministic [`floor`](crate::floor) as the hard merge gate (design §4);
//! - the `plan.json` v2 types + validator ([`octl_core::plan`], design §13);
//! - the [`DecisionEnvelope`](crate::pipeline::DecisionEnvelope) audit record
//!   (design §2), stamped with the tier that made each call.
//!
//! # Additive, not a `run create`
//!
//! This command does **not** create an orchestratectl run, append events, or
//! touch the supervisor / reducer / lock layer — it keeps its own scratch state
//! (intent.md, plan.json, transcripts, git worktrees) under a work dir and emits
//! a structured [`PipelineReport`](crate::pipeline::live::PipelineReport). That keeps it clear of the five
//! state-integrity invariants (no new raw event-append path) while it proves the
//! whole system end to end.
//!
//! # Scope
//!
//! Beyond the original happy-path skeleton, this now wires the **bounded
//! verify→triage→fix loop** (design §7 TRIGGER_RE_SPEC, §8 RE_CODE_CHUNK): a
//! floor-blocked chunk or a failed verify is fed back as a RE_CODE_CHUNK
//! re-brief and MUST re-verify before it can close; a SPEC-FLAW verdict emits
//! TRIGGER_RE_SPEC (a new `plan.v(N+1)` + a DAG-diff deciding which chunks revert
//! to Pending). The loop is bounded **hard** by the deterministic circuit-breakers
//! of [`fixloop::FixLoopConfig`](crate::pipeline::live::fixloop::FixLoopConfig) (design §9) so it can never loop on judgment
//! alone. With the breakers set to [`OFF`](crate::pipeline::live::fixloop::FixLoopConfig::OFF) the driver
//! reverts to the original walking-skeleton behaviour (the first failure is
//! terminal), which is how the pre-loop tests stay meaningful. The floor stays
//! the hard merge gate throughout (design §4/§14): a chunk or feature it blocks is
//! never merged.
//!
//! Tier promotion and the finer per-finding triage of design §8 (DISCUSS /
//! SPIN_OFF / DROP) remain deferred to follow-ups.

pub mod breakers;
pub mod fixloop;
pub mod git;
pub mod providers;

use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use std::time::{Duration, Instant};

use octl_core::plan::{self, Acceptance, Chunk, Plan, Tier};
use serde::Serialize;

use crate::error::CliError;
use crate::floor::{
    self, evaluate_floor, BaselineSnapshot, CheckRun, FloorInputs, FloorVerdict, RunSnapshot,
};
use crate::harness::{
    CancelToken, Check as HarnessCheck, ChunkOutcome, ChunkRequest, CodeHarness, Usage,
};
use crate::output::{self, OutputFormat, OutputSpec};
use crate::pipeline::{
    route_proposal, Action, ChunkState, ChunkStatus, Coordinator, CoordinatorProposal, Decider,
    DeciderVerdict, DecisionContext, DecisionEnvelope, DecisionTier, DecisionTrigger, Finding,
    FindingVerdict, Severity,
};

use breakers::{failure_fingerprint, ResourceBudget, ResourceMeter};
use fixloop::{next_tier, FixLoopConfig};
use git::MergeOutcome;
use providers::{
    SpecContext, SpecProvider, VerifyContext, VerifyDisposition, VerifyJudgment, VerifyProvider,
};

/// A failure in the live pipeline. Mapped to a [`CliError`] at the command
/// boundary; each variant carries a stable code for the error envelope.
#[derive(Debug, thiserror::Error)]
pub enum PipelineError {
    /// A git shell-out failed.
    #[error("git error: {0}")]
    Git(String),
    /// Bad input / setup precondition (repo, branch, workdir).
    #[error("setup error: {0}")]
    Setup(String),
    /// The spec stage could not be driven to a candidate plan.
    #[error("spec stage failed: {0}")]
    Spec(String),
    /// The spec produced a plan that failed the T2 validator (even after retry).
    #[error("plan invalid: {0}")]
    PlanInvalid(String),
    /// The verify stage could not be driven to a verdict.
    #[error("verify stage failed: {0}")]
    Verify(String),
    /// The floor's capture layer could not collect what it needs to judge.
    #[error("floor capture error: {0}")]
    Floor(String),
    /// The code harness could not produce a result at all.
    #[error("harness error: {0}")]
    Harness(String),
    /// An I/O failure writing scratch state.
    #[error("io error: {0}")]
    Io(String),
}

impl PipelineError {
    /// A stage-scoped error (`spec`/`verify`) carrying the underlying message.
    fn stage(stage: &str, message: impl Into<String>) -> Self {
        match stage {
            "spec" => PipelineError::Spec(message.into()),
            _ => PipelineError::Verify(message.into()),
        }
    }

    /// Append contextual detail to the error's message while preserving its variant
    /// (and therefore its stable [`code`](Self::code)). Used to surface a co-occurring
    /// wave fault — e.g. a sibling build thread that ALSO panicked — on the dominant
    /// hard error, without collapsing the typed error into a free-form string (which
    /// would drop the variant discriminant and the stable code the envelope carries).
    fn with_note(self, note: impl AsRef<str>) -> Self {
        let n = note.as_ref();
        match self {
            PipelineError::Git(m) => PipelineError::Git(format!("{m} — {n}")),
            PipelineError::Setup(m) => PipelineError::Setup(format!("{m} — {n}")),
            PipelineError::Spec(m) => PipelineError::Spec(format!("{m} — {n}")),
            PipelineError::PlanInvalid(m) => PipelineError::PlanInvalid(format!("{m} — {n}")),
            PipelineError::Verify(m) => PipelineError::Verify(format!("{m} — {n}")),
            PipelineError::Floor(m) => PipelineError::Floor(format!("{m} — {n}")),
            PipelineError::Harness(m) => PipelineError::Harness(format!("{m} — {n}")),
            PipelineError::Io(m) => PipelineError::Io(format!("{m} — {n}")),
        }
    }

    /// Stable error code for the CLI error envelope.
    fn code(&self) -> &'static str {
        match self {
            PipelineError::Git(_) => "git_error",
            PipelineError::Setup(_) => "setup_error",
            PipelineError::Spec(_) => "spec_failed",
            PipelineError::PlanInvalid(_) => "plan_invalid",
            PipelineError::Verify(_) => "verify_failed",
            PipelineError::Floor(_) => "floor_error",
            PipelineError::Harness(_) => "harness_error",
            PipelineError::Io(_) => "io_error",
        }
    }
}

impl From<floor::FloorError> for PipelineError {
    fn from(e: floor::FloorError) -> Self {
        PipelineError::Floor(e.to_string())
    }
}

impl From<PipelineError> for CliError {
    fn from(e: PipelineError) -> Self {
        let code = e.code();
        // A bad plan or bad setup is the caller's problem (User); everything
        // else is a system/IO/tooling failure the caller cannot fix by
        // re-phrasing input.
        match e {
            PipelineError::PlanInvalid(_) | PipelineError::Setup(_) => {
                CliError::user(code, e.to_string())
            }
            _ => CliError::system(code, e.to_string()),
        }
    }
}

/// A hard failure of [`run_pipeline_tiered`], pairing the dominant
/// [`PipelineError`] with the run's accumulated [`PipelineReport`] when the run got
/// far enough to accumulate auditable state (issue
/// `pipeline-hard-failure-carries-report`).
///
/// The `error` is the terminal signal and drives the CLI exit code / envelope (its
/// stable [`PipelineError::code`] is preserved — a hard failure is NEVER downgraded to
/// a success). The `report` is `Some` once the run passed the spec stage and could
/// preserve floor-green / blocked wave siblings (state-integrity invariant 5): it
/// carries their `branch_preserved` entries so `cmd_run` can render the invariant-5
/// audit on the failure path instead of silently dropping it. A pre-plan failure (bad
/// repo/branch, spec failure) carries `report: None` — there is no chunk state to
/// audit yet.
#[derive(Debug)]
pub struct PipelineFailure {
    /// The dominant typed error (drives the exit code + stable envelope `code`).
    pub error: PipelineError,
    /// The accumulated report, when the run reached a state worth auditing. Boxed so
    /// the common (report-less) `Err` stays small — the report is a large struct and
    /// an unboxed `Err`-variant that size is what `clippy::result_large_err` guards.
    pub report: Option<Box<PipelineReport>>,
}

impl std::fmt::Display for PipelineFailure {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        self.error.fmt(f)
    }
}

impl std::error::Error for PipelineFailure {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        Some(&self.error)
    }
}

/// A pre-plan failure carries no report — the `?`-sites in the setup/spec prologue
/// (before any chunk state accrues) propagate through this conversion.
impl From<PipelineError> for PipelineFailure {
    fn from(error: PipelineError) -> Self {
        PipelineFailure {
            error,
            report: None,
        }
    }
}

// NOTE: there is deliberately NO `From<PipelineFailure> for CliError`. The exit
// code / envelope mapping is the inner `PipelineError`'s, and the report must be
// RENDERED before the failure is turned into a `CliError` — `cmd_run` destructures
// the `PipelineFailure`, emits the report, then converts `error` explicitly. A
// blanket `PipelineFailure -> CliError` would let a bare `?` silently drop the audit
// report on some future caller (`pipeline-hard-failure-carries-report` review).

/// Fully-resolved configuration for one pipeline run.
pub struct PipelineConfig {
    /// Git repository to operate on (a path inside it; the toplevel is derived).
    pub repo: PathBuf,
    /// The intent text (already resolved from a string or a file).
    pub intent: String,
    /// Branch the feature forks from and (on success) merges back to.
    pub source_branch: String,
    /// Optional file-scope hint passed to the spec stage.
    pub files: Vec<PathBuf>,
    /// Optional slug override (else derived from the intent).
    pub slug: Option<String>,
    /// Base `cargo test` invocation the floor enumerates + runs per-binary for
    /// its structured, target-qualified test snapshot (default `cargo test`).
    pub test_cmd: String,
    /// Base `cargo clippy` invocation the floor captures via
    /// `--message-format=json` (any `--message-format` is forced to `json`;
    /// default `cargo clippy`).
    pub clippy_cmd: String,
    /// Scratch root for worktrees + artifacts (intent.md, plan.json, transcripts).
    pub workdir: PathBuf,
    /// How many out-of-scope files the floor tolerates before failing file-scope.
    pub file_scope_slack: usize,
    /// Keep worktrees/branches after the run (skip teardown) for debugging.
    pub keep: bool,
    /// Optional per-chunk wall-clock ceiling for the code harness.
    pub chunk_timeout: Option<Duration>,
    /// How many independent chunks in one dependency wave may **build**
    /// concurrently (design §6 VAIHE 2). `1` (the default) keeps the proven
    /// strictly-sequential path — each chunk forks off the moving `feat/<slug>`
    /// tip and merges before the next starts. A value `> 1` schedules the
    /// no-dependency-path chunks of a wave to build in parallel worktrees off a
    /// shared base, then merges them in a deterministic order with the floor
    /// re-checked at each merge. Bounded further at runtime by the §9
    /// process-count budget (never reinvents a limiter).
    pub max_build_concurrency: usize,
    /// Circuit-breaker bounds for the verify→triage→fix loop (design §9). Use
    /// [`FixLoopConfig::OFF`] for the v1 "first failure is terminal" behaviour.
    pub fix_loop: FixLoopConfig,
    /// Deterministic resource ceilings (design §9): cost/token, wall-time,
    /// process-count, storage, and repeated-identical-failure. Force the loop to
    /// abort regardless of convergence when crossed. Use
    /// [`ResourceBudget::UNLIMITED`] to disable every resource breaker.
    pub budget: ResourceBudget,
}

/// `serde` `skip_serializing_if` predicate for a `bool` that stays out of the
/// output when `false` (there is no built-in for `&bool`).
#[allow(clippy::trivially_copy_pass_by_ref)]
fn is_false(b: &bool) -> bool {
    !*b
}

/// One chunk's outcome in the report.
#[derive(Debug, Clone, Serialize)]
pub struct ChunkReport {
    /// Chunk id.
    pub id: String,
    /// Chunk title.
    pub title: String,
    /// Starting tier (wire name).
    pub tier: String,
    /// Harness outcome: `committed | no_change | failed | timeout | cancelled`.
    pub outcome: String,
    /// Whether the floor passed (`None` when the chunk produced no commit to gate).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub floor_passed: Option<bool>,
    /// The full floor verdict, when the floor ran.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub floor: Option<FloorVerdict>,
    /// Whether the chunk was merged into the integration branch.
    pub merged: bool,
    /// The chunk's own resulting commit (the harness-produced, floor-gated oid),
    /// when it committed. This is the AUTHORED commit and is preserved verbatim
    /// even after a provenance rollback replays the chunk (item E): it names the
    /// exact tree the floor gated, whereas `merge_commit` tracks the current
    /// on-branch commit, which a replay rewrites.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub commit: Option<String>,
    /// The integration-branch commit that folded the chunk in, when merged. On the
    /// initial merge this is the no-ff merge commit (distinct from `commit`, the
    /// chunk's own tip). After a provenance rollback replays the chunk it is the
    /// LINEAR replay commit on the rebuilt branch — so it always names where the
    /// chunk currently lives on `feat/<slug>`, while `commit` keeps the authored
    /// oid. See `replayed`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub merge_commit: Option<String>,
    /// Whether this chunk's on-branch commit is a REPLAY (a provenance rollback
    /// cherry-picked it onto a rebuilt integration branch) rather than its original
    /// no-ff merge (item E). When true, `commit` is the authored/floor-gated oid and
    /// `merge_commit` is the replayed oid, whose tree may differ from the gated one
    /// (the feature-floor re-check at the tip is the safety net). Skipped in the
    /// audit output unless set, so an un-replayed chunk's report is unchanged.
    #[serde(skip_serializing_if = "is_false", default)]
    pub replayed: bool,
    /// A failure/blocked reason, when not merged.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub reason: Option<String>,
    /// The preserved branch name, when the chunk was kept for inspection
    /// (state-integrity invariant 5: unmerged work is preserved).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub branch_preserved: Option<String>,
}

/// The verify stage's result in the report.
#[derive(Debug, Clone, Serialize)]
pub struct VerifyReport {
    /// Whether every executable acceptance check passed (the mechanical half).
    pub acceptance_checks_passed: bool,
    /// Whether the LLM judged product-vs-intent a pass.
    pub judged_passed: bool,
    /// The combined verdict (`acceptance_checks_passed && judged_passed`).
    pub passed: bool,
    /// One-line judge summary.
    pub summary: String,
    /// Judge findings (recorded, not looped on in v1).
    pub findings: Vec<String>,
}

/// The structured summary the command emits (text + `--json`).
#[derive(Debug, Clone, Serialize)]
pub struct PipelineReport {
    /// Feature slug.
    pub slug: String,
    /// Source branch.
    pub source_branch: String,
    /// Integration branch.
    pub integration_branch: String,
    /// Intent revision (always 1 in the v1 skeleton — no re-spec).
    pub intent_rev: u32,
    /// Plan revision (always 1 in the v1 skeleton).
    pub plan_rev: u32,
    /// Number of chunks in the plan.
    pub chunk_count: usize,
    /// Per-chunk outcome + floor verdict.
    pub chunks: Vec<ChunkReport>,
    /// Verify result, when the pipeline reached the verify stage.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub verify: Option<VerifyReport>,
    /// The feature-level floor verdict at the tip, when the final re-check ran —
    /// so a `floor_blocked` status names exactly which gate failed, rather than
    /// hiding it in the verify summary.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub feature_floor: Option<FloorVerdict>,
    /// Whether the feature merged into the source branch.
    pub merged: bool,
    /// The final commit on the source branch, when merged.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub final_commit: Option<String>,
    /// Overall status: `merged | floor_blocked | verify_failed | chunk_failed |
    /// rollback_conflict | …`. `rollback_conflict` means a provenance rollback
    /// could not cleanly replay a kept chunk onto the rebuilt integration branch;
    /// the branch was restored intact and the run terminated (item B).
    pub status: String,
    /// Decision envelopes recording the tier that made each call (design §2).
    pub decisions: Vec<DecisionEnvelope>,
    /// Number of `RE_CODE_CHUNK` re-briefs performed across the whole run (design
    /// §8) — both code-stage floor re-codes and verify-driven fix re-codes.
    pub recode_count: u32,
    /// Number of `PROMOTE_TIER` promotions performed across the whole run (design
    /// §3): a repeat-failing chunk re-run at a higher model tier.
    pub promote_count: u32,
    /// Number of `TRIGGER_RE_SPEC` events (design §7). `plan_rev` equals `1 +
    /// respec_count`.
    pub respec_count: u32,
    /// Set when a deterministic circuit-breaker stopped the loop (design §9),
    /// naming which ceiling tripped. Its presence means the run terminated on a
    /// breaker rather than converging or hitting a plain terminal state.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub circuit_breaker: Option<String>,
    /// The accumulated per-run resource tally (design §9 cost instrumentation):
    /// total tokens/cost metered from the harness [`Usage`], agent-invocation
    /// count, and peak scratch-storage bytes. Present on every run so the spend is
    /// auditable whether or not a breaker tripped.
    pub resources: ResourceMeter,
    /// A terminal failure reason, when the pipeline could not complete the loop.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub failure: Option<String>,
}

/// Derive a filesystem-/branch-safe slug from the intent's first non-empty line.
/// Lowercased, non-alphanumerics collapsed to single hyphens, trimmed, capped —
/// and guaranteed non-empty (falls back to `feature`) so it satisfies the plan
/// validator's `feature.slug` and forms a valid `feat/<slug>` branch.
#[must_use]
pub fn slugify(intent: &str) -> String {
    let seed = intent.lines().find(|l| !l.trim().is_empty()).unwrap_or("");
    let mut slug = String::new();
    let mut prev_hyphen = false;
    for ch in seed.chars() {
        if ch.is_ascii_alphanumeric() {
            slug.push(ch.to_ascii_lowercase());
            prev_hyphen = false;
        } else if !prev_hyphen && !slug.is_empty() {
            slug.push('-');
            prev_hyphen = true;
        }
    }
    let slug = slug.trim_matches('-');
    let slug: String = slug.chars().take(48).collect();
    let slug = slug.trim_matches('-').to_string();
    if slug.is_empty() {
        "feature".to_string()
    } else {
        slug
    }
}

/// Resolve the `--intent` argument: if it names an existing file (or is prefixed
/// with `@`), read that file; otherwise treat it as the intent text verbatim.
///
/// # Errors
///
/// Returns [`PipelineError::Setup`] when a named file cannot be read, or when the
/// resolved intent is empty.
pub fn resolve_intent(raw: &str) -> Result<String, PipelineError> {
    let text = if let Some(path) = raw.strip_prefix('@') {
        std::fs::read_to_string(path)
            .map_err(|e| PipelineError::Setup(format!("could not read intent file {path}: {e}")))?
    } else if Path::new(raw).is_file() {
        std::fs::read_to_string(raw)
            .map_err(|e| PipelineError::Setup(format!("could not read intent file {raw}: {e}")))?
    } else {
        raw.to_string()
    };
    if text.trim().is_empty() {
        return Err(PipelineError::Setup("intent is empty".to_string()));
    }
    Ok(text)
}

/// Order chunks so every chunk appears after all of its `deps` (a stable
/// topological sort). The plan validator guarantees the graph is acyclic and all
/// deps resolve, so this always succeeds; ties break on the plan's declared
/// order for determinism.
fn topo_order(chunks: &[Chunk]) -> Vec<usize> {
    use std::collections::HashMap;
    let index: HashMap<&str, usize> = chunks
        .iter()
        .enumerate()
        .map(|(i, c)| (c.id.as_str(), i))
        .collect();
    let mut done = vec![false; chunks.len()];
    let mut order = Vec::with_capacity(chunks.len());
    // Repeatedly emit the first not-yet-done chunk whose deps are all done.
    while order.len() < chunks.len() {
        let mut progressed = false;
        for (i, c) in chunks.iter().enumerate() {
            if done[i] {
                continue;
            }
            let ready = c
                .deps
                .iter()
                .all(|d| index.get(d.as_str()).is_some_and(|&j| done[j]));
            if ready {
                done[i] = true;
                order.push(i);
                progressed = true;
            }
        }
        if !progressed {
            // Unreachable for a validated (acyclic) plan; emit the remainder in
            // declared order rather than loop forever.
            for (i, _) in chunks.iter().enumerate() {
                if !done[i] {
                    order.push(i);
                    done[i] = true;
                }
            }
        }
    }
    order
}

/// Convert a plan check into the harness `Check` the agent runs as its own
/// self-check (the authoritative floor check is run separately by
/// [`floor::runner`]). Synthesizes a stable id since `plan::Check` carries none.
fn to_harness_check(i: usize, c: &plan::Check) -> HarnessCheck {
    HarnessCheck {
        id: format!("chk-{i}"),
        desc: c.desc.clone(),
        run: c.run.clone(),
        timeout: None,
    }
}

/// Capture a [`RunSnapshot`] (tests + clippy) in `dir` using the configured
/// commands.
///
/// A **fresh** `CARGO_TARGET_DIR` is allocated per capture
/// (`floor-capture-hardening-round-2` item 1 / F4), and the test and clippy
/// passes get **separate** dirs. Because every `capture_snapshot` call (fork
/// baseline, each chunk tip, the feature tip) gets its own dirs, baseline and tip
/// never share a warm cache — closing the shared-`target/` clippy bypass where a
/// warm cache re-emits zero warnings. Separating the test and clippy dirs *within*
/// a snapshot means clippy can never reuse artifacts the `cargo test --no-run`
/// pass warmed and skip re-linting them; the floor does not depend on clippy's
/// internal cache-fingerprint behaviour to re-emit its warnings. The dirs live in
/// the system temp area (outside the worktree, so they never perturb the
/// file-scope diff) and are removed when the guards drop at the end of this
/// function.
///
/// When `dir` is a cargo workspace (a `Cargo.toml` is present), the capture is
/// additionally judged against **independent** trusted metadata
/// (`floor-capture-hardening-round-3` items 2–4): a forged custom harness
/// (`harness = false` on a test-producing target) is rejected before anything
/// runs (item 3), the captured enumeration must cover the metadata-expected
/// test-target set and be non-empty when tests exist (item 2), and doctests are
/// captured in a `--doc` pass and folded in (item 4). A non-cargo `dir` (only
/// reachable from unit-test fixtures — a real pipeline `dir` is always a cargo
/// workspace, and if its `Cargo.toml` vanished `cargo test` would itself fail
/// closed) skips these metadata steps.
fn capture_snapshot(cfg: &PipelineConfig, dir: &Path) -> Result<RunSnapshot, PipelineError> {
    let alloc = |what| {
        tempfile::TempDir::new().map_err(move |e| {
            PipelineError::from(floor::FloorError::Capture {
                what,
                message: format!("could not allocate a floor target dir: {e}"),
            })
        })
    };
    let test_target_dir = alloc("tests")?;
    let clippy_target_dir = alloc("clippy")?;

    // Trusted, independent metadata (only when `dir` is a real cargo workspace).
    let meta = if dir.join("Cargo.toml").exists() {
        let m = floor::metadata::load(dir)?;
        // Item 3: reject a forged custom harness before trusting any capture.
        floor::metadata::reject_forged_harness(&m)?;
        Some(m)
    } else {
        None
    };

    let mut tests =
        floor::runner::capture_test_snapshot(&cfg.test_cmd, dir, test_target_dir.path())?;

    if let Some(meta) = &meta {
        // Item 2: the captured enumeration must cover the metadata-expected set
        // (absolute, so a compromised/empty baseline is caught — not just a
        // baseline-relative narrowing).
        floor::metadata::verify_enumeration(meta, &tests.targets)?;
        // Item 4: capture doctests into the same snapshot (reuses the test target
        // dir — same ref, so warming is fine and never crosses baseline/tip).
        floor::runner::capture_doctests(dir, test_target_dir.path(), meta, &mut tests)?;
    }

    let clippy =
        floor::runner::capture_clippy_snapshot(&cfg.clippy_cmd, dir, clippy_target_dir.path())?;
    Ok(RunSnapshot {
        tests,
        clippy,
        coverage: None,
    })
}

/// Build a decision envelope stamped with the deciding tier (design §2).
fn envelope(
    actor: &str,
    tier: DecisionTier,
    reason: impl Into<String>,
    inputs: Vec<String>,
    model: impl Into<String>,
    prompt_version: impl Into<String>,
) -> DecisionEnvelope {
    DecisionEnvelope {
        actor: actor.to_string(),
        input_artifacts: inputs,
        reason: reason.into(),
        decision_tier: tier,
        model: model.into(),
        prompt_version: prompt_version.into(),
    }
}

/// Resolves the [`CodeHarness`] a chunk runs on at a given [`Tier`] (design §3
/// adaptive promotion). `PROMOTE_TIER` re-runs a stuck chunk at a higher tier, so
/// the code stage selects its harness by the chunk's **current** (possibly
/// promoted) tier rather than a single fixed adapter.
pub trait TierHarness {
    /// The harness to run a chunk at `tier` on.
    fn harness(&self, tier: Tier) -> &dyn CodeHarness;

    /// The next tier up the ladder from `tier` that this resolver actually has a
    /// distinct harness for, or `None` at the ceiling — or when there is no ladder.
    /// Promotion consults THIS, not the abstract [`Tier`] enum, so a single-harness
    /// resolver never "promotes" a chunk onto the very same adapter (which would
    /// burn budget and mislabel the reported tier for no behavioural change).
    /// Defaults to the full `code → mid → high` ladder.
    fn next_tier(&self, tier: Tier) -> Option<Tier> {
        next_tier(tier)
    }
}

/// A [`TierHarness`] that returns ONE harness for every tier — the behaviour when
/// no per-tier ladder is configured (and what the 4-arg [`run_pipeline`] wraps its
/// single injected harness in). A promotion still bumps the recorded tier and
/// re-runs the chunk, but on the same adapter; the live command wires a real
/// per-tier ladder ([`LiveTierHarness`]) so a promoted chunk runs on a stronger
/// model, while unit tests use this to exercise the promotion control flow
/// deterministically.
pub struct SingleTierHarness<'a>(pub &'a dyn CodeHarness);

impl TierHarness for SingleTierHarness<'_> {
    fn harness(&self, _tier: Tier) -> &dyn CodeHarness {
        self.0
    }
    /// One harness for every tier means there is nothing to promote *to*: a
    /// single-harness resolver reports no higher tier, so `run_pipeline` (which
    /// wraps its one injected harness here) never promotes regardless of
    /// `max_promotions` — the pre-tiering behaviour is preserved exactly.
    fn next_tier(&self, _tier: Tier) -> Option<Tier> {
        None
    }
}

/// The live per-tier ladder (design §3/§10): cheap `claude-deepseek flash` for the
/// base tier, `claude-deepseek pro` for mid, and ambient Opus `claude` for high —
/// so a promoted chunk actually runs on a stronger model. Constructed by
/// [`cmd_run`]; the three adapters self-source their own credentials (no secret is
/// read here).
struct LiveTierHarness {
    code: crate::harness::claude::ClaudeHarness,
    mid: crate::harness::claude::ClaudeHarness,
    high: crate::harness::claude::ClaudeHarness,
}

impl TierHarness for LiveTierHarness {
    fn harness(&self, tier: Tier) -> &dyn CodeHarness {
        match tier {
            Tier::Code => &self.code,
            Tier::Mid => &self.mid,
            Tier::High => &self.high,
        }
    }
}

/// The live loop's fast **coordinator** (design §3 "coordinator (PM) … fast/cheap,
/// stateless fn"): the deterministic supervisor control flow *is* the coordinator.
/// It never *generates* proposals from context — the live loop already knows the
/// action each decision point implies — so [`coordinate`](Coordinator::coordinate)
/// is unused; the type exists only to carry the coordinator-tier envelope metadata
/// (actor / model / prompt version) into the shared [`route_proposal`] routing, so
/// routine live decisions are stamped by the SAME path the scaffold uses.
struct LiveCoordinator;

impl Coordinator for LiveCoordinator {
    fn coordinate(&self, _ctx: &DecisionContext) -> Vec<CoordinatorProposal> {
        Vec::new()
    }
    fn model(&self) -> String {
        "coordinator".to_string()
    }
    fn prompt_version(&self) -> String {
        "v1".to_string()
    }
}

/// A `'static` [`LiveCoordinator`] so the live loop can hold a `&'a dyn Coordinator`
/// without a lifetime shorter than the borrowed `cfg`. The coordinator is a ZST
/// with no state, so one shared instance is correct.
static LIVE_COORDINATOR: LiveCoordinator = LiveCoordinator;

/// The live **decider** seam (design §0.2/§2): the consequential-decision authority
/// the fast coordinator defers to. In the live loop the consequential proposals are
/// ALREADY backed by an Opus stage — `DECLARE_CONVERGED` ⟵ verify(Opus) passed +
/// the deterministic floor green, `TRIGGER_RE_SPEC` ⟵ verify(Opus)'s SPEC-FLAW
/// verdict + the Opus re-plan — so this decider **confirms** each proposal and
/// records Opus provenance, giving `decision_tier` an honest decider-tier stamp.
///
/// It is a deliberate seam, not a second Opus round-trip: it is where a distinct
/// second-opinion Opus decider drops in, and where the sequenced circuit-breaker
/// layer (`pipeline-circuit-breakers`) forces an `ESCALATE` override — the live
/// loop already honours a returned [`Action::Escalate`] at both consequential
/// decision points, so the breaker layer needs no further control-flow hook here.
struct LiveDecider {
    /// The Opus model backing the consequential decision (verify/spec are Opus in
    /// the live path), recorded on the decider-tier envelope.
    model: String,
}

impl Decider for LiveDecider {
    fn decide_consequential(
        &self,
        _ctx: &DecisionContext,
        proposed: &CoordinatorProposal,
    ) -> DeciderVerdict {
        DeciderVerdict {
            action: proposed.action.clone(),
            reason: proposed.reason.clone(),
            input_artifacts: proposed.input_artifacts.clone(),
        }
    }
    fn model(&self) -> String {
        self.model.clone()
    }
    fn prompt_version(&self) -> String {
        "v1".to_string()
    }
}

/// Project the live run's chunk state into the [`DecisionContext`] the shared
/// [`route_proposal`] routing (and any decider it defers to) reads. The live loop
/// tracks a coarser [`LiveChunkStatus`]; map it into the scaffold's
/// [`ChunkStatus`], and carry each chunk's **current** (possibly promoted) tier so
/// a decider sees how far a chunk has already been escalated.
fn live_decision_ctx(run: &Run, plan: &Plan, trigger: DecisionTrigger) -> DecisionContext {
    let chunks = plan
        .chunks
        .iter()
        .map(|c| {
            let status = match run.chunk_status.get(&c.id) {
                Some(LiveChunkStatus::Merged) => ChunkStatus::AwaitingVerify,
                _ => ChunkStatus::Pending,
            };
            let tier = run.chunk_tier.get(&c.id).copied().unwrap_or(c.tier);
            (c.id.clone(), ChunkState { status, tier })
        })
        .collect();
    DecisionContext {
        run_id: format!("pipeline-{}", run.slug),
        plan_rev: plan.plan_rev,
        intent_rev: plan.intent_rev,
        chunks,
        trigger,
    }
}

/// Internal running state threaded through the driver's stages so teardown and
/// the report can see what was created.
struct Run<'a> {
    cfg: &'a PipelineConfig,
    /// The fast coordinator whose metadata stamps routine live decisions (design
    /// §3). A `'static` ZST — the live control flow supplies the proposals.
    coordinator: &'a dyn Coordinator,
    /// The decider the shared routing defers every *consequential* live decision to
    /// (design §0.2). Injected so tests can spy on / override it.
    decider: &'a dyn Decider,
    repo: PathBuf,
    slug: String,
    integration_branch: String,
    integration_wt: PathBuf,
    fork_commit: String,
    decisions: Vec<DecisionEnvelope>,
    chunk_reports: Vec<ChunkReport>,
    /// The feature-level floor verdict, once the final re-check runs.
    feature_floor: Option<FloorVerdict>,
    /// Set when the code stage stopped a chunk short; names the terminal status
    /// (`chunk_floor_blocked` vs `chunk_failed` vs `chunk_merge_conflict`, or
    /// `circuit_breaker` once a chunk's re-code budget is exhausted).
    code_block_status: Option<&'static str>,
    /// The kept chunk a provenance rollback could not replay onto the rebuilt
    /// integration branch (item B). Set on the `rollback_conflict` terminal path so
    /// [`finalize`] can name the offending chunk in the report's failure reason
    /// instead of surfacing a bare `PipelineError::Git`.
    rollback_conflict: Option<String>,
    /// Per-chunk lifecycle across the fix loop (design §7). Seeded Pending from
    /// the plan; a chunk becomes `Merged` when it lands on `feat/<slug>`, and is
    /// reset to `Pending` by a `RE_CODE_CHUNK` re-brief or a re-spec DAG-diff.
    chunk_status: BTreeMap<String, LiveChunkStatus>,
    /// Each chunk's **current** model tier (design §3). Seeded from the plan's
    /// declared `chunk.tier`; a `PROMOTE_TIER` bumps it up the ladder so the code
    /// stage re-runs the chunk on a stronger harness.
    chunk_tier: BTreeMap<String, Tier>,
    /// How many times each chunk has already been promoted (design §3), bounded by
    /// [`FixLoopConfig::max_promotions`].
    chunk_promotions: BTreeMap<String, u32>,
    /// Per-chunk merge provenance for the currently-merged chunks (item 1):
    /// enough to replay each kept-done chunk's own commits onto a rebuilt
    /// integration branch. Recorded when a chunk merges, dropped when it is
    /// reverted / removed, and updated in place by a provenance rebuild.
    chunk_provenance: BTreeMap<String, ChunkProvenance>,
    /// Monotonic merge-order counter stamped onto each [`ChunkProvenance`] so a
    /// rollback replays kept chunks in their original stacking order.
    merge_seq: u64,
    /// Cumulative `RE_CODE_CHUNK` floor re-codes per `(plan_rev, chunk_id, tier)`
    /// (item 2). Unlike the per-code-stage-visit counter (which resets every time
    /// the chunk re-enters the code stage from a verify iteration / re-spec), this
    /// does NOT reset across visits, so the nominal `max_recode_per_chunk` cannot be
    /// exceeded across visits at a given tier. Keying by tier preserves the design
    /// §3 promotion semantics (a promoted chunk earns a FRESH re-code budget at the
    /// new tier) while still capping re-codes at each individual tier — without the
    /// tier in the key, a cumulative-exhausted chunk would cascade one attempt per
    /// tier up the ladder instead of getting a real budget at the stronger model.
    chunk_recode_total: BTreeMap<(u32, String, &'static str), u32>,
    /// Chunk (worktree, branch) pairs preserved because they were not merged.
    preserved: Vec<(PathBuf, String)>,
    /// Total `RE_CODE_CHUNK` re-briefs (design §8), for the report + breaker audit.
    recode_count: u32,
    /// Total `PROMOTE_TIER` promotions across the run (design §3), for the report.
    promote_count: u32,
    /// Total `TRIGGER_RE_SPEC` events (design §7).
    respec_count: u32,
    /// Set when a circuit-breaker stopped the loop (design §9).
    circuit_breaker: Option<String>,
    /// Live per-run resource accounting (design §9): tokens/cost metered from the
    /// harness [`Usage`], agent-invocation count, peak storage, and the
    /// identical-failure fingerprints. The deterministic breakers read this.
    meter: ResourceMeter,
    /// Wall-clock start, for the wall-time breaker (design §9). Held here so the
    /// breaker check is a pure function of a measured [`Duration`].
    started: Instant,
    merged_to_source: bool,
}

/// Enough provenance about a merged chunk to replay its content onto a rebuilt
/// integration branch (item 1: provenance-aware rollback). A chunk's own commits
/// live in the range `base..commit`; cherry-picking that range replays exactly the
/// chunk's change (not its ancestry), so a rollback that resets `feat/<slug>` to
/// the fork and replays only the kept-done chunks drops the reverted/removed
/// chunks' code instead of leaving it stranded on the branch.
#[derive(Debug, Clone)]
struct ChunkProvenance {
    /// The integration tip the chunk forked from (exclusive lower bound of the
    /// replay range).
    base: String,
    /// The chunk's own floor-gated tip commit (inclusive upper bound).
    commit: String,
    /// Original merge order, so a rollback replays kept chunks in the order they
    /// were first stacked onto the integration branch.
    order: u64,
}

/// A chunk's lifecycle in the live fix loop (a coarse projection of design §7's
/// chunk states, sufficient for the skeleton). `NeedsReverify` is modelled by
/// resetting to `Pending`: a re-coded chunk is re-run *and* re-verified.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum LiveChunkStatus {
    /// Not yet coded, reverted by a re-code / re-spec, or awaiting a re-run.
    Pending,
    /// Committed and merged into the integration branch.
    Merged,
}

impl Drop for Run<'_> {
    /// Teardown runs unconditionally when the `Run` goes out of scope — on every
    /// success AND every error path — so no `?` early-return can leak a worktree
    /// or branch (the manual-teardown-per-return approach reviewers flagged as
    /// leaky). [`teardown`] is idempotent-safe and honours `--keep`.
    fn drop(&mut self) {
        teardown(self);
    }
}

/// Run the whole pipeline for one feature and return the structured report.
///
/// The stages ([`SpecProvider`], [`CodeHarness`], [`VerifyProvider`]) are
/// injected so the orchestration logic is unit-testable with deterministic
/// stubs; the live command wires the real Claude/deepseek implementations.
///
/// This 4-arg form runs every chunk on the ONE injected `code` harness (no tier
/// ladder) and defers consequential decisions to a confirming in-process decider
/// — the pre-tiering behaviour. For adaptive tier promotion + a spy-able decider
/// seam use [`run_pipeline_tiered`].
///
/// # Errors
///
/// Returns a [`PipelineError`] for a hard failure (bad repo/branch, an
/// undriveable stage, a git or floor-capture failure). A *floor block* or a
/// failed *verify* is NOT an error — it is a completed run whose
/// [`PipelineReport::status`] records the block; the report is still returned so
/// the caller can inspect the per-chunk floor verdicts.
pub fn run_pipeline(
    cfg: &PipelineConfig,
    spec: &dyn SpecProvider,
    code: &dyn CodeHarness,
    verify: &dyn VerifyProvider,
) -> Result<PipelineReport, PipelineFailure> {
    let resolver = SingleTierHarness(code);
    // The confirming decider preserves the pre-tiering behaviour: a consequential
    // decision is stamped decider-tier and its proposed action is recorded as-is.
    let decider = crate::pipeline::ScriptedDecider::confirming();
    run_pipeline_tiered(cfg, spec, &resolver, verify, &decider)
}

/// The tiered entry point (design §0.2/§3): runs each chunk on the tier the
/// `harnesses` resolver picks for its **current** (possibly promoted) tier, and
/// defers every consequential decision to the injected `decider`. A repeat-failing
/// chunk is re-run at a higher tier (`PROMOTE_TIER`) before the repeated-failure
/// breaker gives up. Routine decisions never touch the decider.
///
/// # Errors
///
/// As [`run_pipeline`].
pub fn run_pipeline_tiered(
    cfg: &PipelineConfig,
    spec: &dyn SpecProvider,
    harnesses: &dyn TierHarness,
    verify: &dyn VerifyProvider,
    decider: &dyn Decider,
) -> Result<PipelineReport, PipelineFailure> {
    // --- 1. Setup: validate repo/branch, fork the integration branch, snapshot
    //        the baseline, write intent.md. Every fallible step after the branch
    //        is created runs under `run`, whose `Drop` guarantees teardown. ---
    let repo = git::toplevel(&cfg.repo)?;
    // The source MUST be a real local branch — a tag / remote-tracking ref /
    // `HEAD` would resolve but then `git worktree add` and the final merge would
    // target a detached or non-updatable ref and the run would "merge" nowhere.
    if !git::branch_exists(&repo, &cfg.source_branch) {
        return Err(PipelineError::Setup(format!(
            "source `{}` is not a local branch (tags, remotes, and HEAD are rejected)",
            cfg.source_branch
        ))
        .into());
    }
    let source_commit = git::resolve_commit(&repo, &cfg.source_branch)?;

    // A caller-supplied slug is slugified too — never trusted verbatim into a
    // branch name / filesystem path (an unsanitised `../x` would traverse).
    let slug = cfg
        .slug
        .as_deref()
        .map_or_else(|| slugify(&cfg.intent), slugify);
    let integration_branch = format!("feat/{slug}");
    if git::branch_exists(&repo, &integration_branch) {
        return Err(PipelineError::Setup(format!(
            "integration branch `{integration_branch}` already exists; refusing to reuse it"
        ))
        .into());
    }

    std::fs::create_dir_all(&cfg.workdir).map_err(|e| {
        PipelineError::Io(format!(
            "could not create workdir {}: {e}",
            cfg.workdir.display()
        ))
    })?;
    std::fs::write(cfg.workdir.join("intent.md"), &cfg.intent)
        .map_err(|e| PipelineError::Io(format!("could not write intent.md: {e}")))?;

    // Fork the integration branch from the pinned source OID (not the mutable
    // branch name), so the whole run is anchored to one commit even if the source
    // branch moves under us. The fork commit IS the source commit by construction.
    git::create_branch(&repo, &integration_branch, &source_commit)?;
    let integration_wt = cfg.workdir.join("integration");

    let mut run = Run {
        cfg,
        coordinator: &LIVE_COORDINATOR,
        decider,
        repo: repo.clone(),
        slug: slug.clone(),
        integration_branch: integration_branch.clone(),
        integration_wt: integration_wt.clone(),
        fork_commit: source_commit.clone(),
        decisions: Vec::new(),
        chunk_reports: Vec::new(),
        feature_floor: None,
        code_block_status: None,
        rollback_conflict: None,
        chunk_status: BTreeMap::new(),
        chunk_tier: BTreeMap::new(),
        chunk_promotions: BTreeMap::new(),
        chunk_provenance: BTreeMap::new(),
        merge_seq: 0,
        chunk_recode_total: BTreeMap::new(),
        preserved: Vec::new(),
        recode_count: 0,
        promote_count: 0,
        respec_count: 0,
        circuit_breaker: None,
        meter: ResourceMeter::new(),
        started: Instant::now(),
        merged_to_source: false,
    };

    git::worktree_add(&repo, &integration_wt, &integration_branch)?;
    // Prove the worktree really is at the fork OID and clean *before* capture
    // (`floor-capture-hardening-round-3` item 5): a matching OID recorded on the
    // baseline does not by itself prove the captured files were that OID, so
    // verify `HEAD == fork_commit` on a clean tree first — else the baseline's
    // provenance would be a claim, not a fact.
    verify_capture_ref(&integration_wt, &run.fork_commit)?;
    let baseline_snapshot = capture_snapshot(cfg, &integration_wt)?;
    // Pin the baseline to the immutable fork OID (not the mutable `feat/<slug>`
    // ref) and fingerprint the toolchain it was captured with
    // (`floor-capture-trust-model` item 5).
    let baseline = BaselineSnapshot::new(
        format!("{integration_branch}@fork"),
        run.fork_commit.clone(),
        floor::runner::rustc_version(&integration_wt),
        baseline_snapshot,
    );

    // --- 2. Spec (Opus): produce + validate the initial plan (retry once). ---
    let mut plan =
        produce_and_validate_plan(&mut run, spec, &baseline.to_plan_baseline(), 1, None)?;
    // T5 evaluator gate: the plan's baseline must match the live one (and the live
    // baseline's own provenance must be well-formed) — item 5.
    gate_plan_baseline(&baseline, &plan)?;
    // (spec invocations are metered inside produce_and_validate_plan.)
    // Discard any side effect the (headless, permission-skipped) spec stage left
    // in the worktree: spec is a planner, so chunks must fork from a pristine
    // fork commit, not from spec's stray edits.
    git::restore_to(&run.integration_wt, &run.fork_commit)?;
    write_plan(&run, &plan)?;
    run.decisions.push(envelope(
        "spec",
        DecisionTier::Decider,
        format!("produced plan with {} chunk(s)", plan.chunks.len()),
        vec![
            format!("intent_rev:1"),
            format!("baseline:{}", baseline.r#ref),
        ],
        spec.model(),
        spec.prompt_version(),
    ));
    // Seed every chunk Pending (design §7): the code stage advances them to Merged.
    run.chunk_status = plan
        .chunks
        .iter()
        .map(|c| (c.id.clone(), LiveChunkStatus::Pending))
        .collect();
    // Seed each chunk's current tier from its plan-declared tier (design §3): a
    // PROMOTE_TIER bumps it up the ladder from here.
    run.chunk_tier = plan.chunks.iter().map(|c| (c.id.clone(), c.tier)).collect();

    // Per-chunk verify findings to fold into the next code-stage re-brief (design
    // §8 RE_CODE_CHUNK). Populated when a FIX verdict targets chunks; the code
    // stage consumes them, so it is cleared after each pass.
    let mut pending_findings: BTreeMap<String, Vec<String>> = BTreeMap::new();
    // Per-chunk prior diff to seed the next code-stage re-brief (item L): a
    // verify-FIX rollback captures each re-code target's reverted attempt here so
    // the re-run does not lose the code the rollback dropped. Consumed and cleared
    // alongside `pending_findings`.
    let mut pending_prior_diff: BTreeMap<String, String> = BTreeMap::new();

    // --- 3-5. The bounded verify→triage→fix loop (design §7/§8), stopped hard by
    //          the deterministic circuit-breakers of §9. ---
    //
    // Everything past the spec stage runs inside this fallible section. A hard
    // `PipelineError` here — most importantly a hard failure in a concurrent wave,
    // after floor-green / blocked siblings have been preserved (invariant 5) — is
    // paired with the run's accumulated report on the way out, so the
    // `branch_preserved` audit is SURFACED on the failure path instead of being
    // dropped when `run` is torn down (issue `pipeline-hard-failure-carries-report`).
    // The exit code stays the error's — the failure is not downgraded.
    let body: Result<PipelineReport, PipelineError> = (|| {
        let mut fix_iter = 0u32;
        let outcome = loop {
            // Deterministic resource breakers (design §9) at the round boundary, BEFORE
            // spending another code/verify cycle: refresh the storage measurement, then
            // trip on any crossed ceiling (cost/token/wall-time/process/storage).
            // Supervisor-owned — the orchestrator is never consulted about a breaker.
            refresh_storage(&mut run);
            if let Some(msg) = resource_breach(&run) {
                run.circuit_breaker = Some(msg);
                break LoopExit::Terminal {
                    verify: None,
                    status: "circuit_breaker",
                };
            }

            // CODE STAGE over the Pending chunks, each with its own bounded RE_CODE
            // re-brief loop (design §8). Already-Merged chunks are skipped.
            run_code_stage(
                &mut run,
                &plan,
                harnesses,
                &baseline,
                &pending_findings,
                &pending_prior_diff,
            )?;
            pending_findings.clear();
            pending_prior_diff.clear();
            if run.circuit_breaker.is_some() {
                // A chunk exhausted its re-code budget — the repeated-failure breaker
                // tripped (design §9). Stop; the failing chunk is preserved.
                break LoopExit::Terminal {
                    verify: None,
                    status: "circuit_breaker",
                };
            }
            if !all_merged(&run, &plan) {
                // A chunk could not be merged and re-code was off / not applicable:
                // terminal at the specific status the code stage recorded (the floor
                // is the hard gate — no merge, design §4/§14).
                let status = run.code_block_status.unwrap_or("chunk_failed");
                break LoopExit::Terminal {
                    verify: None,
                    status,
                };
            }

            // VERIFY on the feature tip (design §6 VAIHE 3). Capture the floor-gated
            // tip BEFORE verify runs, then restore to it afterwards — verify runs
            // headless with skipped permissions, so a verify-time commit or untracked
            // write must never become the tip and smuggle content past the floor
            // (`restore_to` hard-resets AND cleans untracked files).
            let feat_tip = git::head(&run.integration_wt)?;
            let (verify_report, disposition) = run_verify_stage(&mut run, &plan, verify)?;
            git::restore_to(&run.integration_wt, &feat_tip)?;
            // The verify invocation spent process/wall-time budget — check the breakers
            // BEFORE acting on its verdict, so a verify that crossed a ceiling aborts
            // rather than converging-and-merging or launching a re-spec (design §9).
            if let Some(msg) = resource_breach(&run) {
                run.circuit_breaker = Some(msg);
                break LoopExit::Terminal {
                    verify: Some(verify_report),
                    status: "circuit_breaker",
                };
            }
            if verify_report.passed {
                break LoopExit::Converged {
                    verify: verify_report,
                    feat_tip,
                };
            }

            // Verify failed → triage, bounded by the fix-iteration breaker (design §9).
            if fix_iter >= run.cfg.fix_loop.max_fix_iterations {
                // With the bound at 0 no fix was ever attempted → the v1 terminal
                // `verify_failed`; otherwise the loop tried and the breaker trips.
                let status = if run.cfg.fix_loop.max_fix_iterations == 0 {
                    "verify_failed"
                } else {
                    run.circuit_breaker = Some(format!(
                        "verify still failing after {} fix iteration(s)",
                        run.cfg.fix_loop.max_fix_iterations
                    ));
                    "circuit_breaker"
                };
                break LoopExit::Terminal {
                    verify: Some(verify_report),
                    status,
                };
            }
            fix_iter += 1;

            match disposition {
                VerifyDisposition::Fix | VerifyDisposition::FixChunks { .. } => {
                    let targets = resolve_fix_targets(&disposition, &plan, &run);
                    if targets.is_empty() {
                        // No chunk to re-code (e.g. nothing merged yet) — cannot make
                        // progress on a fix, so this is a terminal verify failure.
                        break LoopExit::Terminal {
                            verify: Some(verify_report),
                            status: "verify_failed",
                        };
                    }
                    // Provenance-aware rollback (item 1): rebuild the integration branch
                    // keeping every merged chunk EXCEPT the re-code targets AND their
                    // transitive dependents, so a dropped chunk's downstream code cannot
                    // linger on a tree that no longer contains what it was authored
                    // against. Targets get a re-code decision + the verify findings;
                    // dependents are simply reverted (their inputs changed under them).
                    let seeds: BTreeSet<String> = targets.iter().cloned().collect();
                    let affected = dependent_closure(&plan, &seeds);
                    let keep: BTreeSet<String> = run
                        .chunk_status
                        .iter()
                        .filter(|(id, s)| {
                            **s == LiveChunkStatus::Merged && !affected.contains(id.as_str())
                        })
                        .map(|(id, _)| id.clone())
                        .collect();
                    // Item L: capture each re-code target's provenance diff BEFORE the
                    // rollback drops it, so its re-run carries the reverted attempt's code
                    // into the re-brief (the code-stage `prior_diff` only covers in-visit
                    // floor retries, not a verify-driven rollback). Best-effort — a diff
                    // failure just means no carried diff for that target.
                    let mut captured_diffs: BTreeMap<String, String> = BTreeMap::new();
                    for id in &targets {
                        if let Some(prov) = run.chunk_provenance.get(id) {
                            if let Ok(d) = git::diff(&run.integration_wt, &prov.base, &prov.commit)
                            {
                                captured_diffs.insert(id.clone(), d);
                            }
                        }
                    }
                    // Rebuild the integration branch keeping only the untouched merged
                    // chunks. A replay conflict is terminal (item B): name the chunk in a
                    // `rollback_conflict` report instead of erroring — the branch is
                    // restored intact. Audit/loop state is mutated only AFTER a clean
                    // rebuild, so a conflict-abort leaves the kept chunks' reports intact.
                    match rebuild_integration(&mut run, &keep)? {
                        RebuildOutcome::Rebuilt => {}
                        RebuildOutcome::Conflict { chunk_id } => {
                            run.rollback_conflict = Some(chunk_id);
                            break LoopExit::Terminal {
                                verify: Some(verify_report),
                                status: "rollback_conflict",
                            };
                        }
                    }
                    // Drop reports for the chunks whose code we are rolling back, so a
                    // mid-re-code abort cannot leave a stale "merged" report behind (the
                    // re-run upserts a fresh one). Mirrors the re-spec path's retain.
                    run.chunk_reports
                        .retain(|r| !affected.contains(r.id.as_str()));
                    for id in &targets {
                        record_recode_decision(
                            &mut run,
                            &plan,
                            id,
                            &verify_report.findings,
                            "verify FIX",
                        );
                        pending_findings.insert(id.clone(), verify_report.findings.clone());
                        if let Some(d) = captured_diffs.remove(id) {
                            pending_prior_diff.insert(id.clone(), d);
                        }
                    }
                    // Revert every affected chunk (targets + dependents) to Pending so
                    // the code stage re-runs them off the rebuilt tip in dependency order.
                    for id in &affected {
                        run.chunk_status
                            .insert(id.clone(), LiveChunkStatus::Pending);
                    }
                    // Loop back → the code stage re-runs the reverted chunks, then the
                    // loop re-verifies (design §8: FIX-class MUST re-verify before close).
                }
                VerifyDisposition::SpecFlaw { reason, chunk_ids } => {
                    if run.respec_count >= run.cfg.fix_loop.max_respec {
                        // Re-spec off (0) → terminal verify failure; else breaker trip.
                        let status = if run.cfg.fix_loop.max_respec == 0 {
                            "verify_failed"
                        } else {
                            run.circuit_breaker = Some(format!(
                                "re-spec budget exhausted after {} re-spec(s)",
                                run.cfg.fix_loop.max_respec
                            ));
                            "circuit_breaker"
                        };
                        break LoopExit::Terminal {
                            verify: Some(verify_report),
                            status,
                        };
                    }
                    // TRIGGER_RE_SPEC (design §7): route the consequential decision to
                    // the decider, produce plan.v(N+1), DAG-diff which chunks revert to
                    // Pending, then loop back to the code stage. If the decider declined
                    // the re-spec (ESCALATE override) the loop hands up instead.
                    match trigger_re_spec(
                        &mut run,
                        spec,
                        &plan,
                        &reason,
                        &chunk_ids,
                        &verify_report.findings,
                        &baseline,
                    )? {
                        ReSpecOutcome::Replanned(new_plan) => plan = *new_plan,
                        ReSpecOutcome::Escalated => {
                            break LoopExit::Terminal {
                                verify: Some(verify_report),
                                status: "escalated",
                            };
                        }
                        ReSpecOutcome::RollbackConflict { chunk_id } => {
                            // The re-spec's provenance rollback could not replay a kept
                            // chunk (item B): terminate with a report naming it, branch
                            // restored intact.
                            run.rollback_conflict = Some(chunk_id);
                            break LoopExit::Terminal {
                                verify: Some(verify_report),
                                status: "rollback_conflict",
                            };
                        }
                    }
                    // Loop back → code stage re-runs reverted chunks, then re-verifies.
                }
            }
        };

        // Resolve the loop outcome into a report.
        let (verify_report, feat_tip) = match outcome {
            LoopExit::Terminal { verify, status } => {
                return Ok(finalize(&run, &plan, verify, false, None, status));
            }
            LoopExit::Converged { verify, feat_tip } => (verify, feat_tip),
        };

        // --- 5. Merge: re-check the feature floor at the tip, then merge → source. ---
        let declared: Vec<PathBuf> = union_declared_files(&plan);
        let feature_floor = evaluate_feature_floor(&run, &plan, &baseline, &declared, &feat_tip)?;
        run.feature_floor = Some(feature_floor.clone());

        if !feature_floor.passed() {
            // Floor regressed at the tip — do NOT merge (design §4/§14).
            return Ok(finalize(
                &run,
                &plan,
                Some(verify_report),
                false,
                None,
                "floor_blocked",
            ));
        }

        // The consequential ship judgment: the fast coordinator PROPOSES
        // DECLARE_CONVERGED (verify passed + the deterministic floor is green) and the
        // shared tiered routing defers it to the decider (design §0.2/§2). The decider
        // CONFIRMS — or overrides to ESCALATE (the seam the circuit-breaker layer
        // forces later): an escalation stops short of the merge.
        // The trigger is the passed verify report (no outstanding findings) — the
        // evidence the ship decision rests on.
        let converge_ctx = live_decision_ctx(
            &run,
            &plan,
            DecisionTrigger::VerifyReport {
                report_id: format!("verify-plan-v{}", plan.plan_rev),
                findings: Vec::new(),
            },
        );
        let (converge_action, converge_env) = route_proposal(
            run.coordinator,
            run.decider,
            &converge_ctx,
            CoordinatorProposal {
                action: Action::DeclareConverged,
                reason: "declared converged: verify passed and the feature floor is green"
                    .to_string(),
                input_artifacts: vec![
                    format!("feat:{feat_tip}"),
                    format!("source:{source_commit}"),
                ],
            },
        );
        run.decisions.push(converge_env);
        if !matches!(converge_action, Action::DeclareConverged) {
            // The decider declined to ship (an ESCALATE override, or any non-converge
            // verdict): do NOT merge — the feature is handed up rather than landed.
            return Ok(finalize(
                &run,
                &plan,
                Some(verify_report),
                false,
                None,
                "escalated",
            ));
        }

        // The merge mechanics are routine coordination (supervisor-tier), gated by
        // the decider decision above — kept as a SEPARATE envelope so the tier split
        // is honest (the merge is not itself an Opus judgment).
        match merge_feature_to_source(&run, &feat_tip)? {
            MergeOutcome::Merged { commit } => {
                run.merged_to_source = true;
                run.decisions.push(envelope(
                    "supervisor",
                    DecisionTier::Coordinator,
                    format!("merged {feat_tip} into {}", run.cfg.source_branch),
                    vec![
                        format!("feat:{feat_tip}"),
                        format!("source:{source_commit}"),
                    ],
                    "supervisor",
                    "v1",
                ));
                Ok(finalize(
                    &run,
                    &plan,
                    Some(verify_report),
                    true,
                    Some(commit),
                    "merged",
                ))
            }
            MergeOutcome::Conflict { details } => {
                // The floor was green, but the source branch moved underneath us and
                // the merge conflicts. Report it (preserve the integration branch);
                // it is not a crash — the caller resolves and re-runs.
                let mut vr = verify_report;
                vr.summary = format!("{} (source merge conflicted: {details})", vr.summary);
                Ok(finalize(
                    &run,
                    &plan,
                    Some(vr),
                    false,
                    None,
                    "merge_conflict",
                ))
            }
        }
    })();

    // A hard `PipelineError` past the spec stage keeps its typed variant + stable
    // code (the exit signal), but now carries the run's accumulated report so the
    // preserved wave siblings (invariant 5) are auditable on the failure path. The
    // report is built from the SAME `run` state the successful terminals finalize
    // from — `run` outlives this call, so its `Drop`-time teardown still runs after
    // the report is captured, exactly as on the `Ok` path.
    body.map_err(|error| {
        let mut report = finalize(&run, &plan, None, false, None, "pipeline_error");
        report.failure = Some(error.to_string());
        PipelineFailure {
            error,
            report: Some(Box::new(report)),
        }
    })
}

/// The outcome of the bounded fix loop: either a terminal state (a breaker trip,
/// a floor block, or a failed verify with the fix budget spent) or convergence
/// (verify passed, ready to merge at `feat_tip`).
enum LoopExit {
    /// The loop stopped without converging; carries the report status and the
    /// verify report, if the loop reached verify.
    Terminal {
        verify: Option<VerifyReport>,
        status: &'static str,
    },
    /// Verify passed; merge the feature at this tip.
    Converged {
        verify: VerifyReport,
        feat_tip: String,
    },
}

/// Whether every chunk in the current plan is Merged (design §7). The fix loop
/// proceeds to verify only when the whole plan is on the integration branch.
fn all_merged(run: &Run, plan: &Plan) -> bool {
    plan.chunks
        .iter()
        .all(|c| run.chunk_status.get(&c.id) == Some(&LiveChunkStatus::Merged))
}

/// The chunks a FIX verdict should re-code (design §8). Only **merged** chunks
/// are eligible (verify runs after the whole plan is on `feat`, so a Pending
/// chunk is not a re-code target). An explicit, in-plan `FixChunks` list is
/// honoured verbatim (deduplicated, plan order); a bare `Fix` falls back to every
/// merged chunk — the coarse default, since the judge did not attribute the
/// failure. An explicit list that names ONLY unknown/unmerged chunks yields an
/// empty target set (the caller then treats it as a terminal verify failure)
/// rather than silently exploding to an all-chunk re-code from one bad id.
fn resolve_fix_targets(disp: &VerifyDisposition, plan: &Plan, run: &Run) -> Vec<String> {
    let is_merged = |id: &str| run.chunk_status.get(id) == Some(&LiveChunkStatus::Merged);
    let all_merged: Vec<String> = plan
        .chunks
        .iter()
        .filter(|c| is_merged(&c.id))
        .map(|c| c.id.clone())
        .collect();
    match disp {
        VerifyDisposition::FixChunks { chunk_ids } => {
            // Keep only merged, in-plan ids, in plan order, deduplicated. An empty
            // result is returned as-is (NOT widened to every chunk).
            plan.chunks
                .iter()
                .map(|c| c.id.clone())
                .filter(|id| is_merged(id) && chunk_ids.iter().any(|c| c == id))
                .collect()
        }
        _ => all_merged,
    }
}

/// The transitive-dependent closure of `seeds` within `plan`: `seeds` plus every
/// chunk that (transitively) declares one of them in `deps`. A verify-FIX rollback
/// that drops a chunk MUST also drop everything downstream of it — keeping a
/// dependent while removing its dependency would replay the dependent onto a tree
/// missing what it was authored against (a cherry-pick conflict or, worse, a
/// silently-broken feature). This mirrors the re-spec path's DAG-diff dirtiness
/// propagation ([`fixloop::dag_diff`]) for the verify-FIX path.
fn dependent_closure(plan: &Plan, seeds: &BTreeSet<String>) -> BTreeSet<String> {
    let mut affected = seeds.clone();
    loop {
        let mut grew = false;
        for c in &plan.chunks {
            if affected.contains(&c.id) {
                continue;
            }
            if c.deps.iter().any(|d| affected.contains(d)) {
                affected.insert(c.id.clone());
                grew = true;
            }
        }
        if !grew {
            break;
        }
    }
    affected
}

/// Record a `RE_CODE_CHUNK` decision as the T4 [`Action`] primitive, routed
/// through the shared tiered seam ([`route_proposal`]). `RE_CODE_CHUNK` is
/// **routine**, so the coordinator emits it directly (coordinator-tier) and the
/// decider is **not** consulted (design §0.2 — the cost win). `findings` are the
/// verify/floor findings folded into the re-brief.
fn record_recode_decision(
    run: &mut Run,
    plan: &Plan,
    chunk_id: &str,
    findings: &[String],
    source: &str,
) {
    let action = Action::ReCodeChunk {
        chunk_id: chunk_id.to_string(),
        findings: findings
            .iter()
            .enumerate()
            .map(|(i, f)| Finding {
                id: format!("{chunk_id}-f{i}"),
                summary: f.clone(),
                verdict: FindingVerdict::Fix,
                severity: Severity::Medium,
            })
            .collect(),
    };
    let ctx = live_decision_ctx(
        run,
        plan,
        DecisionTrigger::ChunkCommitted {
            chunk_id: chunk_id.to_string(),
        },
    );
    let (_, env) = route_proposal(
        run.coordinator,
        run.decider,
        &ctx,
        CoordinatorProposal {
            action,
            reason: format!("re-code chunk {chunk_id} ({source})"),
            input_artifacts: vec![format!("chunk:{chunk_id}")],
        },
    );
    run.recode_count = run.recode_count.saturating_add(1);
    run.decisions.push(env);
}

/// The outcome of a `TRIGGER_RE_SPEC` attempt after the decider seam ruled on it.
enum ReSpecOutcome {
    /// The decider confirmed the re-spec; the loop continues on the new plan.
    /// Boxed — a [`Plan`] is large and the [`Escalated`](ReSpecOutcome::Escalated)
    /// variant carries nothing, so boxing keeps the enum small.
    Replanned(Box<Plan>),
    /// The decider overrode the re-spec with an ESCALATE (or any non-re-spec
    /// verdict): the loop hands the feature up instead of re-planning.
    Escalated,
    /// The new plan was produced, but the provenance rollback that rebuilds the
    /// integration branch to the re-spec's kept-done set could not replay the named
    /// kept chunk (item B). The branch was restored intact; the loop terminates with
    /// a `rollback_conflict` report.
    RollbackConflict { chunk_id: String },
}

/// The tier a repeat-failing chunk should be promoted to (design §3), or `None`
/// when it may not promote: its promotion budget ([`FixLoopConfig::max_promotions`])
/// is spent, or the resolver has no higher tier to run it on. Consulting the
/// resolver (not the abstract [`Tier`] enum) is what stops a single-harness
/// resolver from "promoting" onto the same adapter.
fn promotion_target(
    run: &Run,
    harnesses: &dyn TierHarness,
    chunk_id: &str,
    current_tier: Tier,
) -> Option<Tier> {
    let used = run.chunk_promotions.get(chunk_id).copied().unwrap_or(0);
    if used >= run.cfg.fix_loop.max_promotions {
        return None;
    }
    harnesses.next_tier(current_tier)
}

/// `PROMOTE_TIER` (design §3): bump a stuck chunk to `promoted`, record the routine
/// decision through the shared tiered seam (routine → coordinator-tier, the decider
/// is NOT consulted), and update the promotion bookkeeping. `promoted` is the tier
/// [`promotion_target`] returned.
fn promote_chunk(run: &mut Run, plan: &Plan, chunk_id: &str, current_tier: Tier, promoted: Tier) {
    let ctx = live_decision_ctx(
        run,
        plan,
        DecisionTrigger::ChunkCommitted {
            chunk_id: chunk_id.to_string(),
        },
    );
    let (_, env) = route_proposal(
        run.coordinator,
        run.decider,
        &ctx,
        CoordinatorProposal {
            action: Action::PromoteTier {
                chunk_id: chunk_id.to_string(),
                tier: promoted,
            },
            reason: format!(
                "promote chunk {chunk_id} {} → {} (repeat-fail)",
                current_tier.wire_name(),
                promoted.wire_name()
            ),
            input_artifacts: vec![format!("chunk:{chunk_id}")],
        },
    );
    run.decisions.push(env);
    run.chunk_tier.insert(chunk_id.to_string(), promoted);
    let used = run
        .chunk_promotions
        .entry(chunk_id.to_string())
        .or_insert(0);
    *used = used.saturating_add(1);
    run.promote_count = run.promote_count.saturating_add(1);
}

/// `TRIGGER_RE_SPEC` (design §7): record the decision, ask the spec provider for a
/// new plan revision against the flaw reason, DAG-diff old→new to decide which
/// chunks revert to Pending, apply that to the run's chunk state, persist the new
/// plan revision, and return it.
fn trigger_re_spec(
    run: &mut Run,
    spec: &dyn SpecProvider,
    old_plan: &Plan,
    reason: &str,
    forced: &[String],
    findings: &[String],
    baseline: &BaselineSnapshot,
) -> Result<ReSpecOutcome, PipelineError> {
    let new_rev = old_plan.plan_rev.saturating_add(1);

    // Route the consequential TRIGGER_RE_SPEC through the shared tiered seam: it is
    // deferred to the decider (design §0.2/§2), whose verdict is recorded. A
    // confirming decider ratifies it (the Opus re-plan below is the authority); an
    // ESCALATE override stops the re-spec before any new plan is produced. The
    // verify SPEC-FLAW findings are carried into the context as the evidence the
    // decider rules on (so the seam is not blind).
    let trigger_findings: Vec<Finding> = findings
        .iter()
        .enumerate()
        .map(|(i, f)| Finding {
            id: format!("respec-f{i}"),
            summary: f.clone(),
            verdict: FindingVerdict::SpecFlaw,
            severity: Severity::High,
        })
        .collect();
    let ctx = live_decision_ctx(
        run,
        old_plan,
        DecisionTrigger::VerifyReport {
            report_id: format!("respec-v{new_rev}"),
            findings: trigger_findings,
        },
    );
    let (action, env) = route_proposal(
        run.coordinator,
        run.decider,
        &ctx,
        CoordinatorProposal {
            action: Action::TriggerReSpec {
                reason: reason.to_string(),
                chunk_ids: forced.to_vec(),
            },
            reason: format!("re-spec to plan.v{new_rev}: {reason}"),
            input_artifacts: vec![format!("plan:{}", old_plan.plan_rev)],
        },
    );
    run.decisions.push(env);
    // Execute the decider's RECORDED verdict, not the original proposal: the decider
    // may soften/retarget the re-spec (a different reason or forced-chunk set), and
    // the pipeline must do what the audit trail says it did. A non-re-spec verdict
    // (ESCALATE override) hands the feature up before any new plan is produced.
    let (reason, forced): (String, Vec<String>) = match action {
        Action::TriggerReSpec { reason, chunk_ids } => (reason, chunk_ids),
        _ => return Ok(ReSpecOutcome::Escalated),
    };

    // Produce plan.v(N+1). The spec stage runs headless in the integration
    // worktree; restore it to the current tip afterward so a planner's stray edit
    // never bleeds into the code stage.
    let feat_tip = git::head(&run.integration_wt)?;
    let old_raw = serde_json::to_value(old_plan)
        .map_err(|e| PipelineError::Io(format!("could not serialize prior plan: {e}")))?;
    let new_plan = produce_and_validate_plan(
        run,
        spec,
        &baseline.to_plan_baseline(),
        new_rev,
        Some((&old_raw, reason.as_str())),
    )?;
    // T5 evaluator gate on the re-spec'd plan too (item 5).
    gate_plan_baseline(baseline, &new_plan)?;
    // (the re-spec invocation is metered inside produce_and_validate_plan.)
    git::restore_to(&run.integration_wt, &feat_tip)?;

    // DAG-diff old→new: which chunks revert to Pending vs. stay Done (design §7).
    let merged: BTreeSet<String> = run
        .chunk_status
        .iter()
        .filter(|(_, s)| **s == LiveChunkStatus::Merged)
        .map(|(id, _)| id.clone())
        .collect();
    let diff = fixloop::dag_diff(old_plan, &new_plan, &merged, &forced);

    // Build the NEW plan's chunk state as LOCALS first; nothing on `run` is mutated
    // until the provenance rollback below succeeds. Rebuilding the chunk-status map:
    // removed chunks drop out (with their reports); kept-done chunks stay Merged;
    // everything else is Pending.
    let mut status = BTreeMap::new();
    for id in &diff.kept_done {
        status.insert(id.clone(), LiveChunkStatus::Merged);
    }
    for id in &diff.revert_to_pending {
        status.insert(id.clone(), LiveChunkStatus::Pending);
    }
    // Rebuild each chunk's tier for the NEW plan (design §3): a kept-done chunk
    // keeps whatever tier it converged at; a reverted / brand-new chunk resets to
    // the new plan's declared tier and its promotion count clears — the re-coded
    // chunk earns its own fresh promotion budget.
    let keep: BTreeSet<&str> = diff.kept_done.iter().map(String::as_str).collect();
    let mut new_tier = BTreeMap::new();
    let mut new_promotions = BTreeMap::new();
    for c in &new_plan.chunks {
        if keep.contains(c.id.as_str()) {
            let tier = run.chunk_tier.get(&c.id).copied().unwrap_or(c.tier);
            new_tier.insert(c.id.clone(), tier);
            if let Some(&n) = run.chunk_promotions.get(&c.id) {
                new_promotions.insert(c.id.clone(), n);
            }
        } else {
            new_tier.insert(c.id.clone(), c.tier);
        }
    }

    // Provenance-aware rollback (item 1): rebuild the integration branch from the
    // fork replaying ONLY the kept-done chunks, so a reverted chunk's stale code and
    // a removed chunk's code are both dropped instead of stranded on `feat/<slug>`.
    // This is done BEFORE any `run` state is repointed to the new plan (item B
    // transactional-audit fix): a replay conflict must leave the run describing the
    // OLD plan that still matches the restored branch — otherwise `finalize` would
    // emit a report for the new plan while the branch holds the old chunks. On a
    // conflict, touch nothing and hand the offending chunk id up for a
    // `rollback_conflict` report (the branch is restored intact regardless).
    let kept_ids: BTreeSet<String> = diff.kept_done.iter().cloned().collect();
    if let RebuildOutcome::Conflict { chunk_id } = rebuild_integration(run, &kept_ids)? {
        return Ok(ReSpecOutcome::RollbackConflict { chunk_id });
    }

    // The rebuild committed — NOW apply the new plan's state to the run, so the
    // audit records only advance to the new revision once the branch actually did.
    run.chunk_status = status;
    run.chunk_tier = new_tier;
    run.chunk_promotions = new_promotions;
    // The cumulative re-code budget is keyed by plan_rev, so the new revision gets a
    // fresh budget; drop the prior revision's entries so the map can't accumulate
    // across many re-specs.
    run.chunk_recode_total
        .retain(|(rev, _, _), _| *rev >= new_rev);
    // Drop reports for chunks no longer in the plan (removed) or about to be
    // re-coded (reverted) — the re-run upserts a fresh report for the latter. (The
    // rollback already refreshed the KEPT chunks' reports in place.)
    run.chunk_reports.retain(|r| keep.contains(r.id.as_str()));
    // The re-spec has now taken effect on both the branch and the run state.
    run.respec_count = run.respec_count.saturating_add(1);

    write_plan(run, &new_plan)?;
    run.decisions.push(envelope(
        "spec",
        DecisionTier::Decider,
        format!(
            "re-spec plan.v{new_rev}: {} chunk(s) revert to pending, {} kept done",
            diff.revert_to_pending.len(),
            diff.kept_done.len()
        ),
        vec![format!("plan:{new_rev}"), format!("intent_rev:1")],
        spec.model(),
        spec.prompt_version(),
    ));
    Ok(ReSpecOutcome::Replanned(Box::new(new_plan)))
}

/// Persist the plan for a revision to the workdir: `plan.json` always (the
/// current plan) plus `plan.v{N}.json` for the revision, so the immutable
/// per-revision history is retained for audit (design §7).
fn write_plan(run: &Run, plan: &Plan) -> Result<(), PipelineError> {
    let plan_json = serde_json::to_string_pretty(plan)
        .map_err(|e| PipelineError::Io(format!("could not serialize plan.json: {e}")))?;
    std::fs::write(run.cfg.workdir.join("plan.json"), &plan_json)
        .map_err(|e| PipelineError::Io(format!("could not write plan.json: {e}")))?;
    std::fs::write(
        run.cfg
            .workdir
            .join(format!("plan.v{}.json", plan.plan_rev)),
        &plan_json,
    )
    .map_err(|e| PipelineError::Io(format!("could not write plan revision: {e}")))?;
    Ok(())
}

/// Bounded number of spec attempts: the initial produce plus repair re-prompts.
/// Keeps the pre-existing count (design §6 VAIHE 1 — bounded re-spec).
const MAX_PLAN_ATTEMPTS: u32 = 2;

/// Filename the last invalid plan is persisted under (in the workdir) when the
/// spec stage exhausts its attempts, so a human can inspect what the model
/// actually produced.
const INVALID_PLAN_FILE: &str = "plan.invalid.json";

/// Ask the spec provider for a plan, normalize the authoritative fields
/// (feature/baseline/versions) over its output, and validate with the T2
/// validator. On a validation failure this runs a **repair loop** (design §6
/// VAIHE 1): it re-prompts the spec model with the exact validator error and the
/// invalid JSON it produced, so the model corrects precisely that error rather
/// than re-guessing blind. The parse stays strict — the driver never patches a
/// missing field server-side; the model must produce valid output. Bounded to
/// [`MAX_PLAN_ATTEMPTS`]; on exhaustion the last invalid plan is persisted to the
/// workdir ([`INVALID_PLAN_FILE`]) and the error surfaces the last validator
/// message. Returns the validated [`Plan`].
fn produce_and_validate_plan(
    run: &mut Run,
    spec: &dyn SpecProvider,
    baseline: &plan::Baseline,
    plan_rev: u32,
    respec: Option<(&serde_json::Value, &str)>,
) -> Result<Plan, PipelineError> {
    let ctx = SpecContext {
        intent: &run.cfg.intent,
        slug: &run.slug,
        source_branch: &run.cfg.source_branch,
        integration_branch: &run.integration_branch,
        files: &run.cfg.files,
        worktree: &run.integration_wt,
        baseline,
    };

    // Carried across attempts so a repair re-prompt can feed back the exact
    // validator error and the raw JSON the model produced. `last_raw` holds the
    // model's own output (pre-normalize) — what it must correct — and is what we
    // persist on exhaustion.
    let mut last_raw: Option<serde_json::Value> = None;
    let mut last_err: Option<String> = None;

    for attempt in 0..MAX_PLAN_ATTEMPTS {
        // The first attempt produces the candidate (a re-spec if `respec` is set,
        // else a fresh plan); every later attempt is a VALIDATOR repair re-prompt
        // carrying the previous error + invalid JSON forward (after attempt 0 both
        // `last_raw` and `last_err` are always set). A re-spec that produces an
        // invalid plan is thus still repaired the same bounded way.
        let produced = match (attempt, &last_raw, &last_err) {
            (0, _, _) => match respec {
                Some((prev, reason)) => spec.respec_plan(&ctx, prev, reason),
                None => spec.produce_plan(&ctx),
            },
            (_, Some(invalid), Some(err)) => spec.repair_plan(&ctx, invalid, err),
            // Unreachable: fall back to a fresh produce rather than panic.
            _ => spec.produce_plan(&ctx),
        };
        // Count EVERY spec provider invocation toward the process-count breaker
        // (design §9) — including each validator-repair re-prompt, and even a call
        // that then errored. Metering here (rather than once at the call site) is
        // why the two callers no longer meter the spec stage themselves.
        run.meter.record_agent_run(None);
        let raw = match produced {
            Ok(raw) => raw,
            // The spec provider itself failed (spawn/timeout/parse). If a prior
            // attempt already produced an invalid plan, persist it so the failure
            // is still inspectable, then propagate the transport error.
            Err(e) => {
                let _ = persist_invalid_plan(run, last_raw.as_ref());
                return Err(e);
            }
        };
        let normalized = normalize_plan(raw.clone(), run, baseline, plan_rev);
        match plan::parse_and_validate_plan(&normalized) {
            Ok(p) => return Ok(p),
            Err(e) => {
                last_err = Some(e.to_string());
                last_raw = Some(raw);
            }
        }
    }

    // Exhausted: persist the last invalid plan so a human can inspect it (right
    // now nothing was kept), then fail with the last validator message.
    let last_err = last_err.unwrap_or_else(|| "no plan produced".to_string());
    let persisted = persist_invalid_plan(run, last_raw.as_ref());
    Err(PipelineError::PlanInvalid(format!(
        "spec produced an invalid plan after {MAX_PLAN_ATTEMPTS} attempt(s): {last_err}{persisted}"
    )))
}

/// The T5 evaluator's baseline gate (`floor-capture-hardening-round-3` item 5):
/// require the plan's persisted `baseline` to match the live supervisor baseline
/// via [`BaselineSnapshot::verify_plan_baseline`]. This is the first (and only)
/// caller of `verify_plan_baseline`, which was groundwork until now. Beyond
/// catching a plan whose baseline was captured at a different commit / toolchain /
/// enumeration, it fails the run closed when the live baseline's own provenance is
/// unprovable — a malformed `commit_oid` or an `"unknown"` toolchain (rustc probe
/// failed) — rather than proceeding to gate merges against a baseline it cannot
/// vouch for.
fn gate_plan_baseline(live: &BaselineSnapshot, plan: &Plan) -> Result<(), PipelineError> {
    live.verify_plan_baseline(&plan.baseline).map_err(|e| {
        PipelineError::PlanInvalid(format!(
            "plan baseline does not match the live baseline: {e}"
        ))
    })
}

/// Prove a worktree is at `expected_oid` on a clean tree before a provenance
/// capture (`floor-capture-hardening-round-3` item 5). A `BaselineSnapshot`
/// records the OID it *claims* to have captured; this makes that claim a fact by
/// checking `HEAD == expected_oid` and that there are no uncommitted changes, so a
/// dirty or detached-elsewhere worktree cannot be captured under a pinned OID that
/// does not describe its files.
fn verify_capture_ref(worktree: &Path, expected_oid: &str) -> Result<(), PipelineError> {
    let head = git::head(worktree)?;
    if head != expected_oid {
        return Err(PipelineError::Setup(format!(
            "capture worktree HEAD {head} != expected fork OID {expected_oid}; refusing to capture a baseline under an unverified OID"
        )));
    }
    if !git::is_clean(worktree)? {
        return Err(PipelineError::Setup(format!(
            "capture worktree {} has uncommitted changes; refusing to capture a baseline whose files are not the pinned OID {expected_oid}",
            worktree.display()
        )));
    }
    Ok(())
}

/// Best-effort write of the last invalid plan to `<workdir>/plan.invalid.json`
/// and return a suffix naming the file for the error message (or a note that it
/// could not be persisted). Never fails the run — the primary error is the
/// validation failure, and losing the artifact must not mask it.
fn persist_invalid_plan(run: &Run, raw: Option<&serde_json::Value>) -> String {
    let Some(raw) = raw else {
        return String::new();
    };
    let path = run.cfg.workdir.join(INVALID_PLAN_FILE);
    let body = serde_json::to_string_pretty(raw).unwrap_or_else(|_| raw.to_string());
    match std::fs::write(&path, body) {
        Ok(()) => format!(" (raw invalid plan written to {})", path.display()),
        Err(e) => format!(" (could not persist invalid plan: {e})"),
    }
}

/// Overwrite the supervisor-owned fields on a spec-produced plan value so the
/// contract's identity/baseline/version fields are authoritative regardless of
/// what the model emitted — the model is trusted only for `chunks`/`acceptance`
/// (design §1: intent + baseline are orchestrator-owned, not spec-writable).
fn normalize_plan(
    raw: serde_json::Value,
    run: &Run,
    baseline: &plan::Baseline,
    plan_rev: u32,
) -> serde_json::Value {
    use serde_json::json;
    let mut obj = match raw {
        serde_json::Value::Object(m) => m,
        // Not an object → leave it; the validator will reject it clearly.
        other => return other,
    };
    obj.insert(
        "schema_version".to_string(),
        json!(plan::PLAN_SCHEMA_VERSION),
    );
    obj.insert("plan_rev".to_string(), json!(plan_rev));
    obj.insert("intent_rev".to_string(), json!(1));
    obj.insert(
        "feature".to_string(),
        json!({
            "slug": run.slug,
            "source_branch": run.cfg.source_branch,
            "integration_branch": run.integration_branch,
        }),
    );
    // Inject the FULL supervisor-owned baseline, including the provenance fields
    // (`commit_oid` / `toolchain` / `enumerated_targets_hash`) — not just the ref
    // and two content hashes (`floor-capture-hardening-round-3` item 5). The
    // persisted `plan.json` therefore carries the provenance the T5 evaluator
    // (`gate_plan_baseline`) re-verifies; omitting them would make a persisted plan
    // fail `verify_plan_baseline`'s fail-closed-on-empty-provenance guard.
    obj.insert(
        "baseline".to_string(),
        json!({
            "ref": baseline.r#ref,
            "commit_oid": baseline.commit_oid,
            "toolchain": baseline.toolchain,
            "test_passlist_hash": baseline.test_passlist_hash,
            "clippy_warnings_hash": baseline.clippy_warnings_hash,
            "enumerated_targets_hash": baseline.enumerated_targets_hash,
        }),
    );
    serde_json::Value::Object(obj)
}

/// The union of every chunk's `files_touched`, de-duplicated — the declared
/// scope the feature-level floor gates against.
fn union_declared_files(plan: &Plan) -> Vec<PathBuf> {
    let mut seen = std::collections::BTreeSet::new();
    for chunk in &plan.chunks {
        for f in &chunk.files_touched {
            seen.insert(PathBuf::from(f));
        }
    }
    seen.into_iter().collect()
}

/// Run the plan's still-Pending chunks in dependency order, each through its own
/// bounded `RE_CODE` re-brief loop (design §6 VAIHE 2 + §8). For each chunk: fork a
/// worktree off the current integration tip, drive the harness, gate the floor,
/// and merge on green. A floor-blocked / harness-failed attempt is re-briefed
/// with its findings and retried up to
/// [`max_recode_per_chunk`](FixLoopConfig::max_recode_per_chunk) times; once that
/// budget is exhausted the repeated-failure circuit-breaker (design §9) stops the
/// stage and the last failing attempt is preserved. Already-Merged chunks (from a
/// prior iteration or a re-spec's kept-done set) are skipped. On any block the
/// stage stops (the floor is the hard gate); the caller inspects
/// [`Run::circuit_breaker`] / [`Run::code_block_status`].
/// The first resource ceiling crossed so far, if any (design §9), as the
/// `circuit_breaker` message. Pure over the run's meter + measured wall-clock —
/// supervisor-owned, never gated on the orchestrator. Storage uses the last value
/// [`refresh_storage`] observed (refreshed at each round boundary).
fn resource_breach(run: &Run) -> Option<String> {
    run.meter.breach(&run.cfg.budget, run.started.elapsed())
}

/// Re-measure the scratch-workdir size into the meter (design §9 storage ceiling)
/// so the next [`resource_breach`] sees the current disk footprint. Called at each
/// round boundary — cheap enough there, too heavy to run per attempt. A no-op when
/// the storage breaker is off.
fn refresh_storage(run: &mut Run) {
    if run.cfg.budget.max_storage_bytes.is_some() {
        let bytes = breakers::dir_size_bytes(&run.cfg.workdir);
        run.meter.observe_storage_bytes(bytes);
    }
}

/// Provenance-aware rollback (item 1, design §7): rebuild the integration branch
/// from the fork, replaying ONLY the kept-done chunks' commits. Every merged chunk
/// NOT in `keep` (a re-code target, a re-spec-reverted chunk, or a re-spec-removed
/// chunk) has its content dropped, so a re-code no longer stacks on top of stale
/// work and a removed chunk's code no longer lingers on `feat/<slug>`.
///
/// Mechanics: hard-reset the integration worktree to the immutable fork commit,
/// then cherry-pick each kept chunk's own commit range (`base..commit`) back in
/// original merge order. Cherry-picking the RANGE replays exactly that chunk's
/// change (not its ancestry), so omitting a chunk drops precisely its content. The
/// kept chunks' provenance is repointed to the replayed oids and their
/// merged-chunk reports gain the replayed on-branch commit in `merge_commit` +
/// the `replayed` flag, while the report's authored `commit` is kept verbatim
/// (item E): a linear replay's tree may differ from the originally-gated one, so
/// the audit distinguishes the authored oid from the current on-branch oid.
///
/// A successful provenance rebuild's deferred mutations: the new per-chunk
/// provenance map, and the `(chunk_id, replayed_commit)` report-oid updates to
/// apply once every kept chunk has replayed cleanly.
type ReplayResult = (BTreeMap<String, ChunkProvenance>, Vec<(String, String)>);

/// The outcome of a provenance rebuild ([`rebuild_integration`]).
enum RebuildOutcome {
    /// Every kept chunk replayed cleanly; the integration branch is rebuilt and
    /// the run's provenance/reports are updated.
    Rebuilt,
    /// A kept chunk could not be replayed onto the rebuilt tip; the branch was
    /// restored intact and the run should terminate with a `rollback_conflict`
    /// report naming this chunk (item B). NOT a `PipelineError` — a conflict is a
    /// pipeline outcome the orchestrator gets a structured report for, not a crash.
    Conflict { chunk_id: String },
}

/// One step's result inside the transactional replay closure.
enum Replayed {
    /// All kept chunks replayed; carries the deferred provenance + report updates.
    Done(ReplayResult),
    /// The named chunk conflicted; the caller restores the intact tip and reports.
    Conflict(String),
}

/// The durable-provenance ref for one chunk of a run (item G):
/// `refs/pipeline/prov/<slug>/<chunk>`. The run is identified by its feature slug
/// (already ref-safe — lowercased alnum + hyphens from [`slugify`]); the chunk id is
/// used verbatim.
///
/// The chunk id is safe as a trailing ref component even though the plan validator's
/// `[A-Za-z0-9_.-]` alphabet is looser than git's ref grammar (which forbids `..`, a
/// trailing `.`, or a `.lock` suffix): a chunk only HAS provenance to pin once it
/// merged, and merging first created its own worktree branch `<slug>/chunk-<id>`. That
/// branch creation already rejects any `<id>` that would make `chunk-<id>` an illegal
/// ref component — the same dot rules — so a pinnable chunk's id is guaranteed to form
/// a legal `refs/pipeline/prov/<slug>/<id>` too. A legitimately dotted id like `a.b`
/// (a single mid-dot, legal) is pinned verbatim so the ref name matches the id.
fn provenance_ref(slug: &str, chunk_id: &str) -> String {
    format!("refs/pipeline/prov/{slug}/{chunk_id}")
}

/// The ref-namespace prefix (with trailing `/`) holding a run's provenance refs, so
/// teardown can enumerate and prune exactly this run's refs (item G).
fn provenance_ref_prefix(slug: &str) -> String {
    format!("refs/pipeline/prov/{slug}/")
}

/// The durable authored oid to pin for a kept chunk (item G). Prefer the chunk's
/// REPORT `commit` — the original floor-gated authored oid, preserved verbatim across
/// replays (item E) — over [`ChunkProvenance::commit`], which a prior rollback's
/// successful rebuild rewrites to the *replayed* oid. Pinning the report oid keeps the
/// stable authored anchor (the one `report.chunks[*].commit` references) alive across
/// REPEATED rollbacks, instead of repointing the ref to a transient replay and
/// orphaning the original authored commit. Falls back to the provenance oid if no
/// report commit is recorded (defensive — a merged chunk always has one).
fn authored_commit_for<'b>(run: &'b Run, id: &str, prov: &'b ChunkProvenance) -> &'b str {
    run.chunk_reports
        .iter()
        .find(|r| r.id == id)
        .and_then(|r| r.commit.as_deref())
        .unwrap_or(&prov.commit)
}

/// Pin each kept chunk's authored commit under its [`provenance_ref`] (item G).
/// Best-effort per ref: a pin that fails (a transient git/hook/disk error) is skipped
/// rather than aborting the rollback. The rollback itself is data-safe WITHOUT the
/// pins (under default git the authored commits stay reachable for the run's
/// lifetime), so failing the whole rollback for a belt-and-suspenders anchor would be
/// a worse outcome than proceeding — the exposure it closes is real only against an
/// external aggressive `git gc --prune=now` (issue: nil under default config).
fn pin_provenance_refs(run: &Run, kept: &[(String, ChunkProvenance)]) {
    for (id, prov) in kept {
        let oid = authored_commit_for(run, id, prov);
        let _ = git::update_ref(&run.repo, &provenance_ref(&run.slug, id), oid);
    }
}

/// The rebuild is **transactional**: the integration worktree's pre-rebuild tip is
/// captured first, and on ANY failure (a cherry-pick conflict, or a hard git error
/// mid-replay) the branch is restored to that intact tip before returning — so a
/// half-rebuilt branch is never left behind and the `Drop` teardown preserves the
/// REAL prior work (state-integrity invariant 5). In-memory audit state (chunk
/// reports, provenance) is likewise mutated only after every replay succeeds, so a
/// failed rollback leaves the run's records consistent with the (restored) branch.
///
/// A cherry-pick conflict (a kept chunk that no longer applies onto the rebuilt
/// tip) yields [`RebuildOutcome::Conflict`] naming the chunk, so the loop can
/// terminate with a `rollback_conflict` [`PipelineReport`] rather than a bare
/// [`PipelineError::Git`] (item B) — the run stops with the (restored, intact)
/// integration branch preserved either way. A genuine git failure mid-replay is
/// still surfaced as a hard error.
fn rebuild_integration(
    run: &mut Run,
    keep: &BTreeSet<String>,
) -> Result<RebuildOutcome, PipelineError> {
    // Fail closed: every id we intend to keep MUST have recorded provenance. A keep
    // id without provenance would be silently dropped from the rebuilt branch while
    // `chunk_status` still says Merged — resetting the branch BEFORE catching that
    // would be data loss, so check first, before touching git.
    let missing: Vec<&String> = keep
        .iter()
        .filter(|id| !run.chunk_provenance.contains_key(id.as_str()))
        .collect();
    if !missing.is_empty() {
        return Err(PipelineError::Git(format!(
            "provenance rollback refused: kept chunk(s) lack merge provenance: {missing:?}"
        )));
    }

    // The kept chunks' provenance, in the order they were originally stacked.
    let mut kept: Vec<(String, ChunkProvenance)> = run
        .chunk_provenance
        .iter()
        .filter(|(id, _)| keep.contains(id.as_str()))
        .map(|(id, p)| (id.clone(), p.clone()))
        .collect();
    kept.sort_by_key(|(_, p)| p.order);

    // Item G (durable provenance refs): pin each kept chunk's authored commit under
    // `refs/pipeline/prov/<run>/<chunk>` BEFORE the replay closure resets the branch
    // to the fork. After that reset the authored OIDs are reachable only via the
    // object DB (the replayed chunks get fresh cherry-picked OIDs), so an external
    // aggressive `git gc --prune=now` racing the rollback could sweep them; a ref is
    // a first-class GC root that survives. Done here, before ANY reset, to keep the
    // pin atomic relative to the branch reset. Best-effort: a pin failure (e.g. a
    // chunk id that is not a valid ref component) does not abort the rollback —
    // object-DB reachability still holds for the run's lifetime under default git.
    pin_provenance_refs(run, &kept);

    // The intact pre-rebuild tip, to restore on any failure.
    let original_tip = git::head(&run.integration_wt)?;

    // Do all the fallible git work in a closure so ONE restore path handles every
    // early exit; report/provenance mutation is deferred until it fully succeeds.
    // `Replayed::Conflict(id)` signals a replay conflict (restore + report);
    // `Replayed::Done(..)` carries the new provenance and report updates.
    let replay = || -> Result<Replayed, PipelineError> {
        // Reset to the fork, discarding every merged chunk's commits (kept ones are
        // replayed below). `restore_to` hard-resets AND cleans untracked files.
        git::restore_to(&run.integration_wt, &run.fork_commit)?;
        let mut new_prov: BTreeMap<String, ChunkProvenance> = BTreeMap::new();
        // (chunk_id, replayed_commit) report updates, applied only on full success.
        let mut report_updates: Vec<(String, String)> = Vec::new();
        for (id, prov) in &kept {
            let base = git::head(&run.integration_wt)?;
            match git::cherry_pick(&run.integration_wt, &prov.base, &prov.commit)? {
                MergeOutcome::Merged { commit } => {
                    report_updates.push((id.clone(), commit.clone()));
                    new_prov.insert(
                        id.clone(),
                        ChunkProvenance {
                            base,
                            commit,
                            order: prov.order,
                        },
                    );
                }
                // Conflict → signal the caller to restore + report, naming the chunk.
                MergeOutcome::Conflict { .. } => return Ok(Replayed::Conflict(id.clone())),
            }
        }
        Ok(Replayed::Done((new_prov, report_updates)))
    };

    match replay() {
        Ok(Replayed::Done((new_prov, report_updates))) => {
            // Commit the transaction: point each kept chunk's report at its replayed
            // on-branch commit (via `merge_commit`) and flag it replayed, keeping the
            // authored `commit` oid intact (item E); then swap in the new provenance
            // (dropped chunks are gone).
            for (id, commit) in report_updates {
                for r in &mut run.chunk_reports {
                    if r.id == id {
                        r.merge_commit = Some(commit.clone());
                        r.replayed = true;
                    }
                }
            }
            run.chunk_provenance = new_prov;
            Ok(RebuildOutcome::Rebuilt)
        }
        Ok(Replayed::Conflict(chunk_id)) => {
            // Conflict: restore the intact branch, leave in-memory state untouched,
            // and hand the offending chunk id back for a `rollback_conflict` report.
            // The restore MUST succeed to honour the report's "restored intact"
            // claim — if it fails (disk full, a rejecting hook, repo corruption), the
            // branch is in an unknown half-rebuilt state, so surface a hard error
            // instead of a `Conflict` that would falsely assert preservation.
            restore_intact(&run.integration_wt, &original_tip)?;
            Ok(RebuildOutcome::Conflict { chunk_id })
        }
        Err(e) => {
            // A hard git failure mid-replay: restore the intact branch, then surface.
            // A failed restore is chained onto the original error rather than masking
            // it — either way this is a hard error, but the message must not claim
            // the branch was preserved when it was not.
            if let Err(restore_err) = restore_intact(&run.integration_wt, &original_tip) {
                return Err(PipelineError::Git(format!(
                    "provenance rollback failed ({e}) AND restoring the integration branch to its intact tip {original_tip} also failed ({restore_err}); the branch may be in an unknown state"
                )));
            }
            Err(e)
        }
    }
}

/// Hard-reset the integration worktree back to `original_tip` and VERIFY the reset
/// landed (HEAD matches, worktree clean). A rollback's transactional/preservation
/// guarantee rests entirely on this restore succeeding, so — unlike the best-effort
/// teardown resets elsewhere — its failure is surfaced as an error rather than
/// swallowed, so no caller can report "restored intact" over an unknown branch.
fn restore_intact(worktree: &Path, original_tip: &str) -> Result<(), PipelineError> {
    git::restore_to(worktree, original_tip)?;
    let head = git::head(worktree)?;
    if head != original_tip {
        return Err(PipelineError::Git(format!(
            "rollback restore did not land: HEAD is {head}, expected {original_tip}"
        )));
    }
    if !git::is_clean(worktree)? {
        return Err(PipelineError::Git(
            "rollback restore left the integration worktree dirty".to_string(),
        ));
    }
    Ok(())
}

fn run_code_stage(
    run: &mut Run,
    plan: &Plan,
    harnesses: &dyn TierHarness,
    baseline: &BaselineSnapshot,
    pending_findings: &BTreeMap<String, Vec<String>>,
    pending_prior_diff: &BTreeMap<String, String>,
) -> Result<(), PipelineError> {
    // Process the still-Pending chunks in dependency **waves** (design §6 VAIHE 2):
    // wave `i` is the set of chunks whose deps are all satisfied (already Merged or
    // built in an earlier wave), so a wave's members have no dependency PATH between
    // them and can build concurrently; later waves still serialise on their dep
    // edges. With `max_build_concurrency == 1` (the default) each wave is drained
    // one chunk at a time off the moving `feat/<slug>` tip — byte-for-byte the
    // proven strictly-sequential path.
    let waves = ready_waves(plan, &run.chunk_status);
    for wave in &waves {
        let pending: Vec<usize> = wave
            .iter()
            .copied()
            .filter(|&i| run.chunk_status.get(&plan.chunks[i].id) != Some(&LiveChunkStatus::Merged))
            .collect();
        if pending.is_empty() {
            continue;
        }
        let k = effective_build_concurrency(run, pending.len());
        if k <= 1 {
            for &idx in &pending {
                process_chunk_sequential(
                    run,
                    plan,
                    idx,
                    harnesses,
                    baseline,
                    pending_findings,
                    pending_prior_diff,
                )?;
                // Any block / breaker the chunk hit set `code_block_status` (the floor
                // is the hard gate): stop the whole stage, exactly as the flat loop did.
                if run.code_block_status.is_some() {
                    return Ok(());
                }
            }
        } else {
            run_wave_concurrent(
                run,
                plan,
                &pending,
                harnesses,
                baseline,
                k,
                pending_findings,
                pending_prior_diff,
            )?;
            if run.code_block_status.is_some() {
                return Ok(());
            }
        }
    }
    Ok(())
}

/// The effective build concurrency for a wave of `wave_len` ready chunks: the
/// configured [`max_build_concurrency`](PipelineConfig::max_build_concurrency),
/// clamped to the wave size and to the **remaining §9 process-count budget** — so
/// concurrency reuses the existing resource breaker rather than reinventing a
/// limiter. With no process ceiling the budget clamp is inactive; the value is
/// never below 1.
///
/// This bounds how many builds are DISPATCHED against the budget known at wave
/// start, not the final tally: the dispatched builds each meter afterwards and a
/// re-code can spend several agent runs, so a wave can still overshoot the ceiling
/// within itself — the §9 breaker (checked at the wave boundary and the round
/// boundary) is what actually stops the run once the metered spend crosses the cap.
fn effective_build_concurrency(run: &Run, wave_len: usize) -> usize {
    let mut k = run.cfg.max_build_concurrency.min(wave_len).max(1);
    if let Some(cap) = run.cfg.budget.max_processes {
        // `processes` already counts spec + earlier chunk/verify runs; leave at least
        // one slot so a wave always makes progress (the breaker still trips later if
        // the run genuinely exceeds the ceiling).
        let used = run.meter.processes;
        let remaining = (cap as usize).saturating_sub(used as usize).max(1);
        k = k.min(remaining);
    }
    k
}

/// Drain ONE chunk to a terminal outcome the strictly-sequential way (design §6
/// VAIHE 2 default): fork off the current moving `feat/<slug>` tip, drive the
/// bounded `RE_CODE` re-brief + adaptive-promotion loop (design §8/§3), merge on
/// green. A block / breaker sets [`Run::code_block_status`] (+ preserves the
/// attempt) and returns; the caller stops the stage. This is the original flat
/// code-stage body, unchanged — the wave dispatcher just calls it per chunk.
#[allow(clippy::too_many_arguments)]
fn process_chunk_sequential(
    run: &mut Run,
    plan: &Plan,
    idx: usize,
    harnesses: &dyn TierHarness,
    baseline: &BaselineSnapshot,
    pending_findings: &BTreeMap<String, Vec<String>>,
    pending_prior_diff: &BTreeMap<String, String>,
) -> Result<(), PipelineError> {
    let chunk = &plan.chunks[idx];
    // Seed the re-brief with any verify findings the fix loop routed to this
    // chunk (a verify-driven RE_CODE_CHUNK). This seed PERSISTS across
    // floor-retry attempts: a floor failure mid re-code appends its findings
    // rather than erasing the verify context (why the chunk is being re-coded
    // at all), so the model never "forgets" the original fix on attempt 2.
    let verify_seed: Vec<String> = pending_findings.get(&chunk.id).cloned().unwrap_or_default();
    let mut findings: Vec<String> = verify_seed.clone();
    // The prior failed attempt's diff, carried into the next re-brief so a
    // torn-down retry does not lose the failing work (item 3: re-code amnesia).
    // Seeded from a verify-FIX rollback's captured pre-rollback diff (item L) so
    // the very FIRST re-run after a rollback also carries the reverted code, not
    // just the in-visit floor retries.
    let mut prior_diff: Option<String> = pending_prior_diff.get(&chunk.id).cloned();
    // Two counters (see design §3 + §8): `recode` is the per-tier re-code
    // attempt (1-based) the re-code budget bounds — it RESETS to 1 on a
    // promotion so each tier gets its own fresh budget. `seq` is a monotonic
    // attempt id that NEVER resets; it names the attempt's worktree/branch, so
    // a promoted re-run can never collide with a superseded lower-tier attempt's
    // branch even if that attempt's cleanup failed (the names stay distinct).
    let mut recode = 1u32;
    let mut seq = 1u32;
    loop {
        let current_tier = run.chunk_tier.get(&chunk.id).copied().unwrap_or(chunk.tier);
        match attempt_chunk(
            run,
            plan,
            chunk,
            harnesses,
            current_tier,
            baseline,
            seq,
            &findings,
            prior_diff.as_deref(),
        )? {
            ChunkAttempt::Merged {
                verdict,
                base,
                commit,
                merge_commit,
            } => {
                run.decisions.push(envelope(
                    "supervisor",
                    DecisionTier::Coordinator,
                    format!("chunk {} floor green — merged", chunk.id),
                    vec![format!("chunk:{}", chunk.id), format!("commit:{commit}")],
                    "supervisor",
                    "v1",
                ));
                // Record merge provenance BEFORE moving `commit`/`base` into the
                // report, so a later rollback can replay this chunk (item 1).
                run.merge_seq += 1;
                run.chunk_provenance.insert(
                    chunk.id.clone(),
                    ChunkProvenance {
                        base,
                        commit: commit.clone(),
                        order: run.merge_seq,
                    },
                );
                upsert_chunk_report(
                    run,
                    ChunkReport {
                        id: chunk.id.clone(),
                        title: chunk.title.clone(),
                        tier: current_tier.wire_name().to_string(),
                        outcome: "committed".to_string(),
                        floor_passed: Some(true),
                        floor: Some(verdict),
                        merged: true,
                        commit: Some(commit),
                        merge_commit: Some(merge_commit),
                        replayed: false,
                        reason: None,
                        branch_preserved: None,
                    },
                );
                run.chunk_status
                    .insert(chunk.id.clone(), LiveChunkStatus::Merged);
                // A resource ceiling the merge's own spend crossed stops the
                // stage here (design §9). This is a post-attempt backstop, not an
                // atomic gate: this chunk already landed on `feat/<slug>` (its
                // work is preserved there — the feature never reaches source, and
                // teardown keeps the integration branch), and further chunks/
                // rounds are what the breaker prevents.
                refresh_storage(run);
                if let Some(msg) = resource_breach(run) {
                    run.circuit_breaker = Some(msg);
                    run.code_block_status = Some("circuit_breaker");
                    return Ok(());
                }
                break;
            }
            ChunkAttempt::Blocked {
                outcome,
                diff: attempt_diff,
                status,
                reason,
                findings: attempt_findings,
                floor,
                floor_passed,
                recodable,
                wt,
                branch,
            } => {
                // Deterministic resource circuit-breakers (design §9), checked
                // BEFORE spending another attempt on this chunk — the whole point
                // of §9 (never gated on the model's judgment). (a) repeated-
                // identical-failure: the SAME block (chunk + status + findings)
                // recurring to the ceiling aborts instead of grinding the re-code
                // budget on an unchanging failure. (b) any resource ceiling the
                // attempt just metered crossed (cost/token/process/wall-time). On
                // either, preserve the attempt (state-integrity invariant 5) and
                // stop the stage at a `circuit_breaker` terminal.
                let fp = failure_fingerprint(
                    &chunk.id,
                    current_tier.wire_name(),
                    status,
                    &attempt_findings,
                );
                let recurrence = run.meter.record_failure(&fp);
                // Refresh the storage measurement here too (not only at the round
                // boundary) so intra-code-stage disk growth can trip the breaker
                // before the next round; a no-op when the storage breaker is off.
                refresh_storage(run);
                if let Some(msg) = run
                    .cfg
                    .budget
                    .identical_failure_breach(recurrence)
                    .or_else(|| resource_breach(run))
                {
                    run.circuit_breaker = Some(msg.clone());
                    run.code_block_status = Some("circuit_breaker");
                    run.decisions.push(envelope(
                        "supervisor",
                        DecisionTier::Coordinator,
                        format!(
                            "chunk {} stopped by circuit-breaker — preserved, not merged ({msg})",
                            chunk.id
                        ),
                        vec![format!("chunk:{}", chunk.id)],
                        "supervisor",
                        "v1",
                    ));
                    push_blocked_chunk(
                        run,
                        chunk,
                        outcome,
                        floor,
                        floor_passed,
                        reason,
                        &wt,
                        &branch,
                    );
                    return Ok(());
                }

                // Re-code while the chunk is re-codable and BOTH budgets hold
                // (design §8): the per-tier `recode` counter (1-based; re-code N
                // is allowed iff N ≤ max_recode_per_chunk), AND the cumulative
                // per-`(plan_rev, chunk, tier)` budget (item 2). The cumulative
                // counter does NOT reset when the chunk re-enters the code stage
                // from a later verify iteration / re-spec, so a chunk cannot be
                // floor re-coded past the nominal bound across visits at a given
                // tier. (A re-spec bumps `plan_rev` → a genuinely new budget; a
                // promotion changes `current_tier` → a fresh per-tier budget, so a
                // promoted stronger model is not starved by the lower tier's spent
                // budget — while each individual tier is still capped.)
                let recode_key = (plan.plan_rev, chunk.id.clone(), current_tier.wire_name());
                let recode_total = run
                    .chunk_recode_total
                    .get(&recode_key)
                    .copied()
                    .unwrap_or(0);
                if recodable
                    && recode <= run.cfg.fix_loop.max_recode_per_chunk
                    && recode_total < run.cfg.fix_loop.max_recode_per_chunk
                {
                    record_recode_decision(
                        run,
                        plan,
                        &chunk.id,
                        &attempt_findings,
                        "floor re-code",
                    );
                    *run.chunk_recode_total.entry(recode_key).or_insert(0) += 1;
                    // The floor-failed attempt is superseded — drop its
                    // worktree + branch to make room for the re-brief (it is
                    // not mergeable work; the final failed attempt, if the
                    // budget runs out, IS preserved below). Carry its diff into
                    // the next re-brief before the worktree is gone (item 3).
                    let _ = git::worktree_remove(&run.repo, &wt);
                    let _ = git::delete_branch(&run.repo, &branch, true);
                    // Keep the LAST committed diff: a later attempt that produced
                    // no commit (harness no-change / timeout) must not erase the
                    // failing code a prior attempt did commit.
                    if attempt_diff.is_some() {
                        prior_diff = attempt_diff;
                    }
                    // Persist the verify seed, append this attempt's floor
                    // findings (see `verify_seed` above).
                    findings = verify_seed
                        .iter()
                        .cloned()
                        .chain(attempt_findings)
                        .collect();
                    recode += 1;
                    seq += 1;
                    continue;
                }

                // Re-code budget exhausted at this tier. Before giving up,
                // adaptive promotion (design §3): a repeat-failing chunk is
                // re-run at the NEXT model tier the resolver offers. Bounded by
                // `max_promotions` and the top of the ladder; only a re-codable
                // block promotes (a merge conflict is not the model's fault).
                if let Some(promoted) = recodable
                    .then(|| promotion_target(run, harnesses, &chunk.id, current_tier))
                    .flatten()
                {
                    promote_chunk(run, plan, &chunk.id, current_tier, promoted);
                    // The failed attempt at the old tier is superseded — drop it.
                    // The promoted re-run gets a fresh per-tier re-code budget
                    // (`recode = 1`) but a NEW monotonic `seq`, so its
                    // worktree/branch never collide with the just-dropped attempt.
                    let _ = git::worktree_remove(&run.repo, &wt);
                    let _ = git::delete_branch(&run.repo, &branch, true);
                    if attempt_diff.is_some() {
                        prior_diff = attempt_diff;
                    }
                    findings = verify_seed
                        .iter()
                        .cloned()
                        .chain(attempt_findings)
                        .collect();
                    recode = 1;
                    seq += 1;
                    continue;
                }

                // Terminal for this chunk. Distinguish a breaker trip (we tried
                // re-codes / promotions and exhausted them) from the v1
                // first-failure block (re-code off, or a non-re-codable outcome
                // like a conflict).
                let promotions = run.chunk_promotions.get(&chunk.id).copied().unwrap_or(0);
                if recodable && (run.cfg.fix_loop.max_recode_per_chunk > 0 || promotions > 0) {
                    // Report the CUMULATIVE floor re-code count for this chunk in
                    // this plan revision (item O), not just the per-visit `seq`
                    // (which resets each time the chunk re-enters the code stage
                    // from a later verify iteration / re-spec). Sum ACROSS TIERS,
                    // not just the current one, so a chunk exhausted after a
                    // promotion doesn't read "0 re-codes at tier X" while the
                    // lower tiers actually spent the budget — the total reflects
                    // the real effort, and `promotions` shows how it was spread.
                    let cumulative_recodes: u32 = run
                        .chunk_recode_total
                        .iter()
                        .filter(|((rev, id, _tier), _)| *rev == plan.plan_rev && id == &chunk.id)
                        .map(|(_, n)| *n)
                        .sum();
                    run.circuit_breaker = Some(format!(
                            "chunk {} still blocked after {seq} attempt(s) this visit ({cumulative_recodes} cumulative floor re-code(s) across visits, {promotions} promotion(s)): {reason}",
                            chunk.id,
                        ));
                    run.code_block_status = Some("circuit_breaker");
                } else {
                    run.code_block_status = Some(status);
                }
                run.decisions.push(envelope(
                    "supervisor",
                    DecisionTier::Coordinator,
                    format!(
                        "chunk {} blocked — preserved, not merged ({reason})",
                        chunk.id
                    ),
                    vec![format!("chunk:{}", chunk.id)],
                    "supervisor",
                    "v1",
                ));
                push_blocked_chunk(
                    run,
                    chunk,
                    outcome,
                    floor,
                    floor_passed,
                    reason,
                    &wt,
                    &branch,
                );
                return Ok(());
            }
        }
    }
    Ok(())
}

/// Partition the plan's not-yet-`Merged` chunks into dependency **waves** for the
/// concurrent scheduler (design §6 VAIHE 2). Wave `i` is the maximal set of
/// still-Pending chunks whose deps are ALL satisfied — either already `Merged`
/// (kept-done, or merged in an earlier code-stage visit) or emitted in an earlier
/// wave. A wave's members therefore have no dependency PATH between them (they can
/// build concurrently); chunks in later waves still serialise on their dep edges.
///
/// Deterministic: within a wave chunks keep the plan's declared order (the same
/// tie-break as [`topo_order`]), so the downstream merge order is stable and
/// reproducible. Returns indices into `plan.chunks`. A validated (acyclic) plan
/// always fully partitions; a would-be cycle / unmet dep (unreachable for a
/// validated plan) is emitted as one final wave in declared order rather than
/// looping forever — mirroring `topo_order`'s own no-progress fallback.
fn ready_waves(plan: &Plan, status: &BTreeMap<String, LiveChunkStatus>) -> Vec<Vec<usize>> {
    let is_merged = |id: &str| status.get(id) == Some(&LiveChunkStatus::Merged);
    let mut remaining: Vec<usize> = plan
        .chunks
        .iter()
        .enumerate()
        .filter(|(_, c)| !is_merged(&c.id))
        .map(|(i, _)| i)
        .collect();
    // A dep is "satisfied" once its chunk is merged or has landed in an earlier wave.
    let mut done: BTreeSet<String> = plan
        .chunks
        .iter()
        .filter(|c| is_merged(&c.id))
        .map(|c| c.id.clone())
        .collect();
    let mut waves: Vec<Vec<usize>> = Vec::new();
    while !remaining.is_empty() {
        let wave: Vec<usize> = remaining
            .iter()
            .copied()
            .filter(|&i| plan.chunks[i].deps.iter().all(|d| done.contains(d)))
            .collect();
        if wave.is_empty() {
            // No chunk became ready — a cycle or an unmet external dep, which a
            // validated plan never has. Emit the remainder as one final wave in
            // declared order instead of spinning forever.
            waves.push(remaining.clone());
            break;
        }
        for &i in &wave {
            done.insert(plan.chunks[i].id.clone());
        }
        remaining.retain(|i| !wave.contains(i));
        waves.push(wave);
    }
    waves
}

/// The outcome of building ONE wave chunk off the shared wave base (design §6
/// VAIHE 2): a floor-green built commit ready for the deterministic merge phase, or
/// a terminal block after the chunk exhausted its floor re-code budget. Mirrors
/// [`BuildAttempt`] but carries the per-chunk audit the main thread folds back
/// (metered usage, the findings behind each floor re-code) — a build thread holds
/// no `&mut Run`, so it accumulates these and the folder replays them deterministically.
struct WaveBuildResult {
    /// Index into `plan.chunks`.
    idx: usize,
    /// The tier the chunk built at (no in-wave promotion — see the module note).
    tier: Tier,
    /// One entry per harness invocation (design §9 metering), folded in order.
    usages: Vec<Option<Usage>>,
    /// The findings behind each floor re-code the build spent, replayed as
    /// `RE_CODE_CHUNK` decisions on the main thread so the audit + `recode_count`
    /// match the sequential path.
    recode_findings: Vec<Vec<String>>,
    outcome: WaveBuildOutcome,
}

/// The deterministic identity (worktree + branch) of the attempt a wave worker is
/// CURRENTLY building. A worker registers this in the shared artifact map BEFORE each
/// [`build_and_gate`] call so that, if the attempt commits real work and then the
/// worker hard-errors or panics, the branch it committed on is still named for the
/// invariant-5 audit fold — [`WaveJob::Error`]/[`WaveJob::Panicked`] discard the
/// [`WaveBuildResult`], so without this the branch would be orphaned (issue
/// `wave-terminal-worker-own-artifact-unaudited`). Cleared on a clean terminal return
/// (the `WaveBuildResult` then carries the identity, folded normally).
struct WaveArtifact {
    wt: PathBuf,
    branch: String,
    /// The branch's tip at registration time, BEFORE this attempt's `build_and_gate`
    /// ran — `None` when the deterministic branch did not yet exist. The audit only
    /// attributes committed work to this worker when the tip ADVANCED past this: a
    /// stale branch left by a prior interrupted run (which `worktree_add` refuses to
    /// recreate, failing the attempt) must not be mis-reported as this worker's work.
    initial_tip: Option<String>,
}

/// A wave chunk's build result (see [`WaveBuildResult`]).
enum WaveBuildOutcome {
    /// Floor green off the shared wave base; the deterministic merge phase merges
    /// `commit` (re-checking the floor at the moved tip).
    Built {
        verdict: FloorVerdict,
        base: String,
        commit: String,
        wt: PathBuf,
        branch: String,
    },
    /// Terminal block; the worktree/branch are preserved (state-integrity invariant 5).
    Blocked {
        outcome: &'static str,
        status: &'static str,
        reason: String,
        floor: Option<FloorVerdict>,
        floor_passed: Option<bool>,
        recodable: bool,
        wt: PathBuf,
        branch: String,
    },
}

/// What one wave-build worker handed back to the folding main thread. A worker
/// holds no `&mut Run`, so it returns one of these into the shared results map and
/// the fold replays it deterministically. The `Panicked` variant exists because the
/// worker wraps its build in [`std::panic::catch_unwind`]: a panic must NOT unwind
/// out of the wave scope (which would strand every SIBLING chunk's floor-green
/// worktree — the fold never runs to preserve them, violating invariant 5), so it
/// is caught and surfaced here as a terminal marker the fold turns into a clean
/// stage-stop after preserving the siblings.
enum WaveJob {
    /// The build completed (floor-green or a terminal block) — fold it normally.
    Done(WaveBuildResult),
    /// An orderly hard [`PipelineError`] from the build thread. Propagated by the
    /// fold exactly as the sequential path propagates its own hard errors (`Run::Drop`
    /// tears down); sibling preservation on this path is a tracked follow-up.
    Error(PipelineError),
    /// The build thread panicked; the worker's `catch_unwind` caught it so the wave
    /// scope does not re-panic. The fold preserves every sibling build and stops the
    /// stage at a terminal block naming the crashed chunk. Carries the panic payload
    /// message ([`panic_message`]) so the audit/circuit-breaker records WHY it crashed
    /// instead of a context-free "a thread panicked".
    Panicked(String),
}

/// Best-effort human-readable text of a caught panic payload (from
/// [`std::panic::catch_unwind`]). Panic payloads are almost always `&str` (a string
/// literal `panic!`) or `String` (a formatted one); anything else is reported
/// generically. Used to give the wave's terminal decision a real reason.
fn panic_message(payload: &(dyn std::any::Any + Send)) -> String {
    payload
        .downcast_ref::<&str>()
        .map(|s| (*s).to_string())
        .or_else(|| payload.downcast_ref::<String>().cloned())
        .unwrap_or_else(|| "unknown panic payload".to_string())
}

/// Build ONE wave chunk off the shared `wave_base` through a bounded floor re-code
/// loop (design §6 VAIHE 2 + §8) WITHOUT merging. Thread-safe: it takes only shared
/// (`Sync`) inputs and touches no `&mut Run`, so `k` of these run concurrently in a
/// wave; the git worktree-metadata mutations serialise on `git_lock` while the slow
/// harness runs stay parallel. Returns the built commit or a terminal block, plus
/// the per-chunk audit for the main thread to fold in.
///
/// Adaptive tier promotion (design §3) is intentionally NOT done in the build thread
/// — a promotion mutates shared run state, which a worker holding no `&mut Run`
/// cannot touch. A wave chunk that exhausts its per-tier re-code budget blocks
/// (preserved). Promotion still happens, just deferred to the main thread: on
/// build-phase exhaustion [`run_wave_concurrent`] re-queues the blocked chunk into
/// the sequential drain ([`process_chunk_sequential`]) off the moved tip, which
/// exercises promotion — so `--max-build-concurrency > 1` no longer terminally
/// blocks a promotable chunk that `1` would have promoted-and-merged
/// (`immoderately-dirty-cushion`). The deterministic FLOOR gate is identical to the
/// sequential path — it is the shared [`build_and_gate`].
#[allow(clippy::too_many_arguments)]
fn build_chunk_in_wave(
    cfg: &PipelineConfig,
    repo: &Path,
    slug: &str,
    wave_base: &str,
    plan_rev: u32,
    idx: usize,
    chunk: &Chunk,
    tier: Tier,
    harness: &dyn CodeHarness,
    baseline: &BaselineSnapshot,
    verify_seed: &[String],
    seed_prior_diff: Option<&str>,
    recode_budget: u32,
    git_lock: &Mutex<()>,
    artifacts: &Mutex<BTreeMap<usize, WaveArtifact>>,
) -> Result<WaveBuildResult, PipelineError> {
    let mut findings: Vec<String> = verify_seed.to_vec();
    let mut prior_diff: Option<String> = seed_prior_diff.map(str::to_string);
    let mut seq = 1u32;
    let mut recode = 1u32;
    let mut usages: Vec<Option<Usage>> = Vec::new();
    let mut recode_findings: Vec<Vec<String>> = Vec::new();
    loop {
        // Register the deterministic identity this attempt is ABOUT to build under,
        // BEFORE `build_and_gate` creates (and the harness may commit on) the branch.
        // If the harness commits real work and then hard-errors or panics, `?`/unwind
        // leaves without a `WaveBuildResult`, but the fold can still name + audit this
        // branch from the registry (invariant 5, issue
        // `wave-terminal-worker-own-artifact-unaudited`). A clean terminal return
        // clears the entry (the result carries the identity from there).
        {
            let (wt, branch) = chunk_attempt_names(cfg, slug, &chunk.id, seq);
            // Capture the branch's tip BEFORE this attempt runs (read-only git, safe to
            // run concurrently — no `git_lock` needed). `None` in the normal fresh case;
            // `Some(oid)` only for a stale branch a prior interrupted run left behind,
            // which the audit then refuses to attribute to this worker.
            let initial_tip = if git::branch_exists(repo, &branch) {
                git::resolve_commit(repo, &branch).ok()
            } else {
                None
            };
            artifacts
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .insert(
                    idx,
                    WaveArtifact {
                        wt,
                        branch,
                        initial_tip,
                    },
                );
        }
        let (built, usage) = build_and_gate(
            cfg,
            repo,
            slug,
            wave_base,
            plan_rev,
            chunk,
            harness,
            baseline,
            seq,
            &findings,
            prior_diff.as_deref(),
            Some(git_lock),
        )?;
        usages.push(usage);
        match built {
            BuildAttempt::Built {
                verdict,
                base,
                commit,
                wt,
                branch,
            } => {
                // Clean terminal return: the `WaveBuildResult` carries the identity, so
                // clear the in-flight marker — the fold preserves this via the result,
                // not the artifact registry.
                artifacts
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .remove(&idx);
                return Ok(WaveBuildResult {
                    idx,
                    tier,
                    usages,
                    recode_findings,
                    outcome: WaveBuildOutcome::Built {
                        verdict,
                        base,
                        commit,
                        wt,
                        branch,
                    },
                });
            }
            BuildAttempt::Blocked {
                outcome,
                diff,
                status,
                reason,
                findings: attempt_findings,
                floor,
                floor_passed,
                recodable,
                wt,
                branch,
            } => {
                // Re-code while re-codable and the (cumulative) budget holds. The
                // superseded attempt's worktree/branch are dropped under the git lock;
                // its diff is carried into the next re-brief (item 3: re-code amnesia).
                if recodable && recode <= recode_budget {
                    {
                        let _g = git_lock
                            .lock()
                            .unwrap_or_else(std::sync::PoisonError::into_inner);
                        let _ = git::worktree_remove(repo, &wt);
                        let _ = git::delete_branch(repo, &branch, true);
                    }
                    if diff.is_some() {
                        prior_diff = diff;
                    }
                    recode_findings.push(attempt_findings.clone());
                    findings = verify_seed
                        .iter()
                        .cloned()
                        .chain(attempt_findings)
                        .collect();
                    recode += 1;
                    seq += 1;
                    continue;
                }
                // `attempt_findings` drove the (now-exhausted) re-code loop; a
                // terminal block is never re-briefed, so it is dropped here.
                drop(attempt_findings);
                // Clean terminal return (the `WaveBuildResult` carries the identity):
                // clear the in-flight marker so the fold preserves this via the result.
                artifacts
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .remove(&idx);
                return Ok(WaveBuildResult {
                    idx,
                    tier,
                    usages,
                    recode_findings,
                    outcome: WaveBuildOutcome::Blocked {
                        outcome,
                        status,
                        reason,
                        floor,
                        floor_passed,
                        recodable,
                        wt,
                        branch,
                    },
                });
            }
        }
    }
}

/// Run ONE dependency wave with concurrent builds + a deterministic serial merge
/// (design §6 VAIHE 2). `pending` are the wave's still-Pending chunk indices (no
/// dependency path between them); `k > 1` is the effective concurrency.
///
/// Phase 1 — **concurrent build**: every wave chunk builds off the SAME shared base
/// (the integration tip at wave start) in its own worktree, `k` at a time
/// ([`build_chunk_in_wave`], work-stealing over a shared queue). The per-chunk audit
/// is folded back on this thread in deterministic `pending` order (metering,
/// re-code decisions, cumulative re-code budget), so the recorded state never
/// depends on thread timing.
///
/// Phase 2 — **deterministic serial merge**: the floor-green chunks merge into
/// `feat/<slug>` in `pending` order, and the floor is **re-checked at each merge**
/// (the build was gated against the STALE shared base; a sibling merged first may
/// have moved the tip). A merge conflict, or a floor regression at the moved tip,
/// triggers the design's **rebase-and-fix**: discard the stale build and rebuild the
/// chunk off the moved tip via the sequential drain (which re-forks off the new tip,
/// re-gates, re-merges). Any block there stops the whole stage, exactly as the
/// sequential path does.
///
/// Phase 3 — **build-phase exhaustion → sequential drain**: a chunk that exhausted
/// its floor re-code budget during the concurrent build blocked WITHOUT a tier
/// promotion (the worker holds no `&mut Run`). Rather than terminally block — which
/// would make a promotable chunk fail at `> 1` where `1` succeeds — each such chunk
/// is re-queued into the same sequential drain off the moved tip, which exercises
/// adaptive promotion (design §3, `immoderately-dirty-cushion`). Its preserved
/// build-phase attempt (kept before Phase 2 as invariant-5 protection against a
/// merge-phase early return) is first reconciled so the re-run's fresh outcome is
/// the only one — no orphan worktree, no double preservation.
#[allow(clippy::too_many_arguments)]
fn run_wave_concurrent(
    run: &mut Run,
    plan: &Plan,
    pending: &[usize],
    harnesses: &dyn TierHarness,
    baseline: &BaselineSnapshot,
    k: usize,
    pending_findings: &BTreeMap<String, Vec<String>>,
    pending_prior_diff: &BTreeMap<String, String>,
) -> Result<(), PipelineError> {
    let wave_base = git::head(&run.integration_wt)?;
    let plan_rev = plan.plan_rev;
    let max_recode = run.cfg.fix_loop.max_recode_per_chunk;

    // Snapshot every per-chunk build input on the main thread — the build threads
    // hold no `&Run`. The re-code budget is the sequential path's cumulative bound:
    // `max_recode_per_chunk` minus what earlier visits already spent at this tier.
    struct Job<'a> {
        idx: usize,
        chunk: &'a Chunk,
        tier: Tier,
        harness: &'a dyn CodeHarness,
        verify_seed: Vec<String>,
        prior_diff: Option<String>,
        recode_budget: u32,
    }
    let jobs: Vec<Job> = pending
        .iter()
        .map(|&idx| {
            let chunk = &plan.chunks[idx];
            let tier = run.chunk_tier.get(&chunk.id).copied().unwrap_or(chunk.tier);
            let spent = run
                .chunk_recode_total
                .get(&(plan_rev, chunk.id.clone(), tier.wire_name()))
                .copied()
                .unwrap_or(0);
            Job {
                idx,
                chunk,
                tier,
                harness: harnesses.harness(tier),
                verify_seed: pending_findings.get(&chunk.id).cloned().unwrap_or_default(),
                prior_diff: pending_prior_diff.get(&chunk.id).cloned(),
                recode_budget: max_recode.saturating_sub(spent),
            }
        })
        .collect();

    let cfg = run.cfg;
    let repo = run.repo.clone();
    let slug = run.slug.clone();
    let git_lock: Mutex<()> = Mutex::new(());
    let queue: Mutex<VecDeque<usize>> = Mutex::new((0..jobs.len()).collect());
    let results: Mutex<BTreeMap<usize, WaveJob>> = Mutex::new(BTreeMap::new());
    // The in-flight artifact identity of each worker, keyed by chunk idx. A worker
    // clears its entry on a clean terminal return; an entry that SURVIVES the scope
    // belongs to a worker that hard-errored or panicked mid-attempt — its branch (which
    // may hold committed work) is not named by any `WaveBuildResult`, so the fold audits
    // it (invariant 5, `wave-terminal-worker-own-artifact-unaudited`).
    let artifacts: Mutex<BTreeMap<usize, WaveArtifact>> = Mutex::new(BTreeMap::new());

    std::thread::scope(|scope| {
        for _ in 0..k {
            scope.spawn(|| loop {
                let next = {
                    let mut q = queue
                        .lock()
                        .unwrap_or_else(std::sync::PoisonError::into_inner);
                    q.pop_front()
                };
                let Some(job_i) = next else { break };
                let job = &jobs[job_i];
                // Catch a panic INSIDE the worker so it never unwinds out of the wave
                // scope: an escaping panic re-panics on the scope's join, skipping the
                // fold — every sibling chunk's floor-green worktree would then go
                // un-preserved (invariant 5). `AssertUnwindSafe` is sound here: the
                // only shared state a re-code touches (the `git_lock`, its poison
                // flag) is already accessed through `PoisonError::into_inner`, so a
                // poisoned lock after a panic is handled, not observed as torn state.
                let res = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                    build_chunk_in_wave(
                        cfg,
                        &repo,
                        &slug,
                        &wave_base,
                        plan_rev,
                        job.idx,
                        job.chunk,
                        job.tier,
                        job.harness,
                        baseline,
                        &job.verify_seed,
                        job.prior_diff.as_deref(),
                        job.recode_budget,
                        &git_lock,
                        &artifacts,
                    )
                }));
                let outcome = match res {
                    Ok(Ok(r)) => WaveJob::Done(r),
                    Ok(Err(e)) => WaveJob::Error(e),
                    Err(panic) => WaveJob::Panicked(panic_message(panic.as_ref())),
                };
                results
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .insert(job.idx, outcome);
            });
        }
    });

    let mut results = results
        .into_inner()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    // Any entry that survived the scope belongs to a worker that hard-errored or
    // panicked mid-attempt (a clean return clears it): the branch it may have committed
    // on is named by no `WaveBuildResult`. Audited below on the terminal paths.
    let artifacts = artifacts
        .into_inner()
        .unwrap_or_else(std::sync::PoisonError::into_inner);

    // Fold the per-chunk audit deterministically (pending order), splitting the
    // outcomes into the floor-green builds (to merge) and the terminal blocks. A
    // worker that hard-errored or panicked is folded separately (below) so the loop
    // still visits every OTHER chunk's result — a mid-fold `?` would skip them.
    let mut built: Vec<WaveBuildResult> = Vec::new();
    let mut blocked: Vec<WaveBuildResult> = Vec::new();
    let mut hard_error: Option<PipelineError> = None;
    // (chunk id, panic payload message) of the FIRST worker that panicked, if any.
    let mut panicked: Option<(String, String)> = None;
    for &idx in pending {
        let r = match results
            .remove(&idx)
            .expect("every wave job records a result")
        {
            WaveJob::Done(r) => r,
            WaveJob::Error(e) => {
                // Audit this worker's OWN artifact HERE, keyed on its ACTUAL terminal
                // classification, so the reason string is per-worker correct even when a
                // hard error and a panic co-occur on different chunks (auditing in the
                // dominant-terminal block below would stamp every survivor with one
                // reason). `artifacts` still holds the crasher's in-flight identity — a
                // clean return would have cleared it (invariant 5,
                // `wave-terminal-worker-own-artifact-unaudited`).
                if let Some(artifact) = artifacts.get(&idx) {
                    audit_terminal_worker_artifact(
                        run,
                        &plan.chunks[idx],
                        &wave_base,
                        artifact,
                        "terminal wave worker hard-errored after committing — branch preserved, contents unaudited (floor never gated this attempt)",
                    );
                }
                if hard_error.is_none() {
                    hard_error = Some(e);
                }
                continue;
            }
            WaveJob::Panicked(msg) => {
                if let Some(artifact) = artifacts.get(&idx) {
                    audit_terminal_worker_artifact(
                        run,
                        &plan.chunks[idx],
                        &wave_base,
                        artifact,
                        "terminal wave worker panicked after committing — branch preserved, contents unaudited (floor never gated this attempt)",
                    );
                }
                if panicked.is_none() {
                    panicked = Some((plan.chunks[idx].id.clone(), msg));
                }
                continue;
            }
        };
        for u in &r.usages {
            run.meter.record_agent_run(u.as_ref());
        }
        let chunk = &plan.chunks[idx];
        let recode_key = (plan_rev, chunk.id.clone(), r.tier.wire_name());
        for f in &r.recode_findings {
            record_recode_decision(run, plan, &chunk.id, f, "floor re-code");
            *run.chunk_recode_total
                .entry(recode_key.clone())
                .or_insert(0) += 1;
        }
        match r.outcome {
            WaveBuildOutcome::Built { .. } => built.push(r),
            WaveBuildOutcome::Blocked { .. } => blocked.push(r),
        }
    }

    // A hard error DOMINATES the wave's terminal outcome: it propagates as `Err`, so
    // the run exits non-zero with a typed, coded error — as on the sequential hard-error
    // path. This precedence is deliberate: `cmd_run` renders a circuit-breaker `Ok`
    // report with exit code 0, so downgrading a hard infrastructure failure to `Ok`
    // merely because a sibling ALSO panicked would hide the failure from CI / supervisors
    // (a co-occurring panic must not make the failure quieter). A co-occurring panic is
    // still surfaced in the returned error (via `with_note`, which preserves the error's
    // variant + code), never silently swallowed.
    //
    // Before returning, PRESERVE the floor-green (`built`) and blocked siblings
    // (state-integrity invariant 5): they committed work that never merged. Their
    // worktrees/branches already survive on disk (`teardown` never deletes chunk
    // branches), but recording each as a `branch_preserved` `ChunkReport` is what makes
    // that preservation AUDITABLE — `run_pipeline_tiered` now pairs this hard `Err` with
    // the accumulated report (`pipeline-hard-failure-carries-report`), so `cmd_run`
    // surfaces the preserved siblings on the failure path rather than dropping them.
    //
    // Each terminal worker's OWN artifact was already audited above, in the fold's
    // `WaveJob::Error`/`Panicked` arms, keyed on its actual classification (so a panic
    // co-occurring with a hard error keeps its own reason string). Here we only preserve
    // the floor-green (`built`) and blocked SIBLINGS that never crashed.
    if let Some(e) = hard_error {
        for r in blocked.iter().chain(built.iter()) {
            preserve_wave_build(run, &plan.chunks[r.idx], r);
        }
        return match panicked {
            Some((chunk_id, msg)) => Err(e.with_note(format!(
                "a sibling build thread also panicked on chunk {chunk_id}: {msg}"
            ))),
            None => Err(e),
        };
    }

    // A build worker panicked with NO co-occurring hard error. Its per-worker
    // `catch_unwind` kept the panic from unwinding the wave scope (an escaping panic
    // re-panics on the scope join, skipping this fold — every sibling's floor-green
    // worktree would then go un-preserved, invariant 5). This path returns a graceful
    // circuit-breaker `Ok` report, so preservation IS surfaced: record every
    // committed-but-unmerged SIBLING (the committed-but-`blocked` ones and the
    // floor-green `built` ones) so the panic can't strand real work, then stop the stage
    // at a terminal block naming the crashed chunk. The panicked worker's OWN committed
    // branch was already audited above in the fold's `WaveJob::Panicked` arm (a branch
    // with no commit beyond the wave base is left as the `--keep`-recoverable residual it
    // always was — invariant 5, `wave-terminal-worker-own-artifact-unaudited`).
    if let Some((chunk_id, msg)) = panicked {
        for r in blocked.iter().chain(built.iter()) {
            preserve_wave_build(run, &plan.chunks[r.idx], r);
        }
        run.circuit_breaker = Some(format!(
            "wave build thread panicked on chunk {chunk_id} — sibling builds preserved, stage stopped: {msg}"
        ));
        run.code_block_status = Some("circuit_breaker");
        run.decisions.push(envelope(
            "supervisor",
            DecisionTier::Coordinator,
            format!(
                "chunk {chunk_id} build thread panicked — sibling builds preserved, stage stopped: {msg}"
            ),
            vec![format!("chunk:{chunk_id}")],
            "supervisor",
            "v1",
        ));
        return Ok(());
    }

    // Non-terminal path: preserve every build-phase block's worktree/branch NOW
    // (state-integrity invariant 5): they hold committed work, and a later early
    // return from the merge phase (a rebase-and-fix that itself blocks) must never
    // strand it. The terminal stage-stop status is set after the good chunks merge,
    // below. The floor-green `built` chunks are NOT preserved here — they proceed to
    // the merge phase, which preserves any that a later early return leaves unmerged.
    for r in &blocked {
        preserve_wave_build(run, &plan.chunks[r.idx], r);
    }

    // A resource ceiling crossed by the wave's builds stops before ANY merge (design
    // §9): preserve every un-merged BUILT worktree (invariant 5; blocked ones were
    // preserved just above) and terminate.
    refresh_storage(run);
    if let Some(msg) = resource_breach(run) {
        run.circuit_breaker = Some(msg.clone());
        run.code_block_status = Some("circuit_breaker");
        run.decisions.push(envelope(
            "supervisor",
            DecisionTier::Coordinator,
            format!("wave stopped by circuit-breaker before merge — builds preserved ({msg})"),
            vec![format!("plan_rev:{plan_rev}")],
            "supervisor",
            "v1",
        ));
        for r in &built {
            preserve_wave_build(run, &plan.chunks[r.idx], r);
        }
        return Ok(());
    }

    // Phase 2 — deterministic serial merge with a floor re-check at each merge, and
    // rebase-and-fix on a conflict / floor regression at the moved tip. Drained from
    // a queue (not a `for` over a moved Vec) so that EVERY early return can first
    // preserve the still-un-merged builds left in the queue (state-integrity
    // invariant 5): a rebase-and-fix that itself blocks, or a post-merge breaker,
    // must never strand a later floor-green build's committed worktree.
    let mut pending_merges: VecDeque<WaveBuildResult> = built.into();
    while let Some(r) = pending_merges.pop_front() {
        let idx = r.idx;
        let tier = r.tier;
        // The build-time verdict was gated against the STALE shared base; the merge
        // phase re-checks the floor at the moved tip, so it is deliberately dropped.
        let (base, commit, wt, branch) = match r.outcome {
            WaveBuildOutcome::Built {
                verdict: _,
                base,
                commit,
                wt,
                branch,
            } => (base, commit, wt, branch),
            WaveBuildOutcome::Blocked { .. } => unreachable!("split above keeps only Built here"),
        };
        let chunk = &plan.chunks[idx];
        let pre_tip = git::head(&run.integration_wt)?;
        match git::merge_no_ff(
            &run.integration_wt,
            &commit,
            &format!("pipeline: merge chunk {}", chunk.id),
        )? {
            MergeOutcome::Merged {
                commit: merge_commit,
            } => {
                // Re-check the floor at the moved tip (design §6 VAIHE 2: a chunk
                // merges ONLY IF the floor still holds). The no-ff merge introduced
                // exactly this chunk's changes relative to `pre_tip`.
                let post_tip = git::head(&run.integration_wt)?;
                let changed = floor::git::changed_files(&run.integration_wt, &pre_tip, &post_tip)?;
                let verdict2 = gate_chunk(
                    run.cfg,
                    &run.repo,
                    chunk,
                    &run.integration_wt,
                    &pre_tip,
                    &changed,
                    baseline,
                )?;
                if verdict2.passed() {
                    finalize_wave_merge(
                        run,
                        chunk,
                        tier,
                        verdict2,
                        base,
                        commit,
                        merge_commit,
                        &wt,
                        &branch,
                    );
                    // Post-merge resource backstop (design §9), mirroring the
                    // sequential Merged arm: the merge's own spend may have crossed a
                    // ceiling. The chunk is already on feat (preserved); stop here.
                    refresh_storage(run);
                    if let Some(msg) = resource_breach(run) {
                        run.circuit_breaker = Some(msg);
                        run.code_block_status = Some("circuit_breaker");
                        preserve_pending_merges(run, plan, &pending_merges);
                        return Ok(());
                    }
                } else {
                    // Floor regressed against a sibling merged earlier this wave →
                    // undo the merge and rebase-and-fix off the moved tip.
                    git::restore_to(&run.integration_wt, &pre_tip)?;
                    let _ = git::worktree_remove(&run.repo, &wt);
                    let _ = git::delete_branch(&run.repo, &branch, true);
                    run.decisions.push(envelope(
                        "supervisor",
                        DecisionTier::Coordinator,
                        format!(
                            "chunk {} floor regressed at merge — rebase&fix off moved tip ({})",
                            chunk.id,
                            fixloop::floor_findings(&verdict2).join("; ")
                        ),
                        vec![format!("chunk:{}", chunk.id)],
                        "supervisor",
                        "v1",
                    ));
                    process_chunk_sequential(
                        run,
                        plan,
                        idx,
                        harnesses,
                        baseline,
                        pending_findings,
                        pending_prior_diff,
                    )?;
                    if run.code_block_status.is_some() {
                        preserve_pending_merges(run, plan, &pending_merges);
                        return Ok(());
                    }
                }
            }
            MergeOutcome::Conflict { details } => {
                // The stale build cannot land on the moved tip → discard it and
                // rebase-and-fix: rebuild the chunk off the current tip (design §6).
                let _ = git::worktree_remove(&run.repo, &wt);
                let _ = git::delete_branch(&run.repo, &branch, true);
                run.decisions.push(envelope(
                    "supervisor",
                    DecisionTier::Coordinator,
                    format!(
                        "chunk {} merge conflict — rebase&fix off moved tip: {details}",
                        chunk.id
                    ),
                    vec![format!("chunk:{}", chunk.id)],
                    "supervisor",
                    "v1",
                ));
                process_chunk_sequential(
                    run,
                    plan,
                    idx,
                    harnesses,
                    baseline,
                    pending_findings,
                    pending_prior_diff,
                )?;
                if run.code_block_status.is_some() {
                    preserve_pending_merges(run, plan, &pending_merges);
                    return Ok(());
                }
            }
        }
    }

    // Phase 3 — build-phase exhaustion → sequential drain (design §3,
    // `immoderately-dirty-cushion`). A chunk that exhausted its floor re-code budget
    // during the concurrent build blocked WITHOUT a tier promotion — the build worker
    // holds no `&mut Run`, so it can't promote. Rather than terminally block the
    // stage (which would make a promotable chunk fail at `--max-build-concurrency > 1`
    // where `1` promotes-and-merges), re-queue each such chunk into the sequential
    // drain off the moved tip, exactly as the merge phase re-queues a conflict /
    // floor-regression. `process_chunk_sequential` re-forks off the current tip: it
    // spends one attempt at the (already-exhausted) tier, finds the cumulative
    // re-code budget spent, and promotes to the next tier — the adaptive promotion
    // the wave build deferred. A genuine block / breaker there stops the whole stage
    // (the sequential drain preserves its OWN fresh attempt and sets the terminal
    // status); any not-yet-drained blocked chunk stays preserved from before the
    // merge phase, so its committed work is never stranded (invariant 5).
    for r in blocked {
        let idx = r.idx;
        let chunk_id = plan.chunks[idx].id.clone();
        // Defensive guard: a NON-re-codable build-phase block cannot be rescued by a
        // re-run (re-coding the same chunk would not change the outcome), so keep it
        // terminal + preserved (from before the merge phase) instead of spending a
        // wasted sequential attempt that would just block again. Every block from
        // `build_chunk_in_wave` is re-codable today — its only `Blocked` source,
        // `build_and_gate`, never emits a merge conflict (the sole non-re-codable
        // outcome, which lives in the merge step) — so this guards a future path, at
        // no cost to the promotable case.
        if let WaveBuildOutcome::Blocked {
            recodable: false,
            status,
            ..
        } = &r.outcome
        {
            run.code_block_status = Some(status);
            run.decisions.push(envelope(
                "supervisor",
                DecisionTier::Coordinator,
                format!(
                    "chunk {chunk_id} blocked in wave (not re-codable) — preserved, not merged"
                ),
                vec![format!("chunk:{chunk_id}")],
                "supervisor",
                "v1",
            ));
            return Ok(());
        }
        // Reconcile the build-phase attempt we preserved before the merge phase: it
        // was invariant-5 protection against a merge-phase early return, but now we
        // re-fork the chunk off the moved tip, so drop that stale attempt (worktree +
        // branch), un-record it from `run.preserved`, AND remove its now-dangling
        // blocked report — no orphan worktree, no double preservation, no audit report
        // pointing at a force-deleted branch.
        reconcile_preserved_wave_build(run, &plan.chunks[idx], &r);
        run.decisions.push(envelope(
            "supervisor",
            DecisionTier::Coordinator,
            format!(
                "chunk {chunk_id} exhausted the wave re-code budget — re-queued to the sequential drain for tier promotion"
            ),
            vec![format!("chunk:{chunk_id}")],
            "supervisor",
            "v1",
        ));
        process_chunk_sequential(
            run,
            plan,
            idx,
            harnesses,
            baseline,
            pending_findings,
            pending_prior_diff,
        )?;
        if run.code_block_status.is_some() {
            return Ok(());
        }
    }
    Ok(())
}

/// Record a floor-green wave build's merge into `feat/<slug>` (design §6 VAIHE 2):
/// mirrors the sequential [`process_chunk_sequential`] `Merged` arm — decision
/// envelope, merge provenance (for a later rollback), one-per-chunk report, and the
/// `Merged` status — then drops the chunk worktree/branch. `verdict` is the floor
/// re-checked at the MOVED tip, so the reported floor reflects what actually shipped.
#[allow(clippy::too_many_arguments)]
fn finalize_wave_merge(
    run: &mut Run,
    chunk: &Chunk,
    tier: Tier,
    verdict: FloorVerdict,
    base: String,
    commit: String,
    merge_commit: String,
    wt: &Path,
    branch: &str,
) {
    run.decisions.push(envelope(
        "supervisor",
        DecisionTier::Coordinator,
        format!("chunk {} floor green — merged (wave)", chunk.id),
        vec![format!("chunk:{}", chunk.id), format!("commit:{commit}")],
        "supervisor",
        "v1",
    ));
    run.merge_seq += 1;
    run.chunk_provenance.insert(
        chunk.id.clone(),
        ChunkProvenance {
            base,
            commit: commit.clone(),
            order: run.merge_seq,
        },
    );
    upsert_chunk_report(
        run,
        ChunkReport {
            id: chunk.id.clone(),
            title: chunk.title.clone(),
            // The tier the build actually ran at (from the build result), not a
            // re-read of `run.chunk_tier` — so if in-wave promotion is ever added the
            // report can't disagree with the tier that produced this commit.
            tier: tier.wire_name().to_string(),
            outcome: "committed".to_string(),
            floor_passed: Some(true),
            floor: Some(verdict),
            merged: true,
            commit: Some(commit),
            merge_commit: Some(merge_commit),
            replayed: false,
            reason: None,
            branch_preserved: None,
        },
    );
    run.chunk_status
        .insert(chunk.id.clone(), LiveChunkStatus::Merged);
    let _ = git::worktree_remove(&run.repo, wt);
    let _ = git::delete_branch(&run.repo, branch, true);
}

/// Preserve every still-un-merged build left in the merge queue when the merge
/// phase stops early (a rebase-and-fix that blocked, or a post-merge breaker):
/// their worktrees hold floor-green committed work that never reached feat, so
/// state-integrity invariant 5 forbids dropping them on the way out.
fn preserve_pending_merges(run: &mut Run, plan: &Plan, pending: &VecDeque<WaveBuildResult>) {
    for r in pending {
        preserve_wave_build(run, &plan.chunks[r.idx], r);
    }
}

/// Preserve an un-merged wave build (a terminal block, or a build stopped by a
/// wave-boundary breaker) — its worktree/branch hold committed work that never
/// reached feat, so state-integrity invariant 5 forbids dropping it.
fn preserve_wave_build(run: &mut Run, chunk: &Chunk, r: &WaveBuildResult) {
    let (outcome, floor, floor_passed, reason, wt, branch) = match &r.outcome {
        WaveBuildOutcome::Blocked {
            outcome,
            floor,
            floor_passed,
            reason,
            wt,
            branch,
            ..
        } => (
            *outcome,
            floor.clone(),
            *floor_passed,
            reason.clone(),
            wt.as_path(),
            branch.as_str(),
        ),
        WaveBuildOutcome::Built {
            verdict,
            wt,
            branch,
            ..
        } => (
            "committed",
            Some(verdict.clone()),
            Some(true),
            "wave build preserved before merge".to_string(),
            wt.as_path(),
            branch.as_str(),
        ),
    };
    push_blocked_chunk(run, chunk, outcome, floor, floor_passed, reason, wt, branch);
}

/// Audit a wave worker's OWN artifact after it hard-errored or panicked mid-attempt
/// (invariant 5, `wave-terminal-worker-own-artifact-unaudited`). The worker discarded
/// its [`WaveBuildResult`] on the way out (`WaveJob::Error`/`Panicked`), so nothing
/// else names the deterministic branch it may have COMMITTED real work on. If that
/// branch exists AND holds commits beyond the wave `base`, record an audit-only
/// `branch_preserved` [`ChunkReport`] — the floor NEVER gated the crashing attempt, so
/// `floor`/`floor_passed` stay `None` and the report does not vouch for the contents —
/// and keep the worktree/branch on disk (teardown never deletes chunk branches).
///
/// A branch that never committed anything beyond `base` (or was never created — a panic
/// before `worktree_add`) is nothing invariant 5 protects: it is left untouched as the
/// `--keep`-recoverable residual the panic/error path always produced, and NOT reported
/// (an audit entry would falsely imply preserved committed work). Likewise a branch
/// whose tip never advanced past `artifact.initial_tip` was NOT this worker's work — a
/// stale branch a prior interrupted run left behind, which `worktree_add` refused to
/// recreate — so it is not mis-attributed here. A `commits_ahead_of` error is treated
/// conservatively as "has work" so a transient git failure never drops a possibly-
/// committed branch.
fn audit_terminal_worker_artifact(
    run: &mut Run,
    chunk: &Chunk,
    base: &str,
    artifact: &WaveArtifact,
    reason: &str,
) {
    if !git::branch_exists(&run.repo, &artifact.branch) {
        return;
    }
    // The branch ref is the durable artifact (the worktree may be dirty after a mid-run
    // panic); read the committed tip from it, not from the worktree.
    let commit = git::resolve_commit(&run.repo, &artifact.branch).ok();
    // Only attribute work THIS attempt produced: if the tip is unchanged from what was
    // there when the worker registered, this attempt never advanced the branch (a stale
    // pre-existing branch whose `worktree_add` failed the attempt). Guarded on
    // `commit.is_some()` so a resolve failure still falls through to conservative
    // preservation below rather than being read as "unchanged".
    if commit.is_some() && commit == artifact.initial_tip {
        return;
    }
    let ahead = git::commits_ahead_of(&run.repo, base, &artifact.branch).unwrap_or(1);
    if ahead == 0 {
        return;
    }
    run.preserved
        .push((artifact.wt.clone(), artifact.branch.clone()));
    upsert_chunk_report(
        run,
        ChunkReport {
            id: chunk.id.clone(),
            title: chunk.title.clone(),
            tier: chunk.tier.wire_name().to_string(),
            outcome: "committed".to_string(),
            floor_passed: None,
            floor: None,
            merged: false,
            commit,
            merge_commit: None,
            replayed: false,
            reason: Some(reason.to_string()),
            branch_preserved: Some(artifact.branch.clone()),
        },
    );
}

/// Undo the invariant-5 preservation of a build-phase block we are about to re-drive
/// through the sequential drain (`immoderately-dirty-cushion`). The block was
/// preserved before the merge phase as protection against a merge-phase early
/// return; now that the merge phase completed and we re-fork the chunk off the moved
/// tip, that stale attempt must be reconciled so it isn't left orphaned. Three
/// things go together: drop its worktree + branch, remove its `run.preserved` entry,
/// and remove its now-dangling blocked `ChunkReport`. Force-deleting the branch is
/// safe — it holds a superseded attempt the sequential re-run replaces (the same
/// treatment the sequential path gives its own superseded re-code / promotion
/// attempts, and the same the merge phase gives a conflicting build).
///
/// Removing the report — not leaving it for the re-run's upsert to overwrite —
/// matters for the hard-error window: `process_chunk_sequential` re-records a report
/// on every ORDERLY return (merge or its own fresh block), but a hard `PipelineError`
/// propagates and tears the run down. Clearing it here means that teardown never
/// leaves an audit report claiming a branch is preserved when this function just
/// force-deleted it — bringing the hard-error window to exact parity with the merge
/// phase's rebase-and-fix (whose conflicting build was never preserved to begin with).
fn reconcile_preserved_wave_build(run: &mut Run, chunk: &Chunk, r: &WaveBuildResult) {
    let (wt, branch) = match &r.outcome {
        WaveBuildOutcome::Blocked { wt, branch, .. }
        | WaveBuildOutcome::Built { wt, branch, .. } => (wt.clone(), branch.clone()),
    };
    let _ = git::worktree_remove(&run.repo, &wt);
    let _ = git::delete_branch(&run.repo, &branch, true);
    run.preserved
        .retain(|(w, b)| w.as_path() != wt.as_path() || b.as_str() != branch.as_str());
    run.chunk_reports.retain(|rep| rep.id != chunk.id);
}

/// The result of one chunk attempt: a green-floor merge, or a block (with the
/// findings a re-code would fold into the next brief, and whether the outcome is
/// re-codable at all — a merge conflict is not).
enum ChunkAttempt {
    /// Floor green and merged into the integration branch.
    Merged {
        /// The floor verdict at the gated commit.
        verdict: FloorVerdict,
        /// The integration tip the chunk forked from (its provenance replay base).
        base: String,
        /// The chunk's own resulting commit oid.
        commit: String,
        /// The integration-branch merge commit.
        merge_commit: String,
    },
    /// The attempt did not merge; the worktree/branch are the (as-yet un-torn-down)
    /// attempt state.
    Blocked {
        /// Report `outcome` label (`committed`/`no_change`/`failed`/…).
        outcome: &'static str,
        /// The failed attempt's committed diff (`base..head`), when it committed —
        /// carried into the next re-brief so a torn-down retry keeps the failing
        /// work (item 3: re-code amnesia). `None` when the attempt never committed
        /// (a harness no-change / failure / timeout).
        diff: Option<String>,
        /// Terminal status if this is the final attempt and re-code is off.
        status: &'static str,
        /// Human-readable block reason.
        reason: String,
        /// Findings folded into a re-code re-brief (floor violations, or the
        /// harness failure reason).
        findings: Vec<String>,
        /// The floor verdict, when the block came from the floor gate.
        floor: Option<FloorVerdict>,
        /// Whether the floor passed (`Some(true)` only on a merge conflict).
        floor_passed: Option<bool>,
        /// Whether this outcome can be retried by a re-code (a merge conflict
        /// cannot — re-coding the same chunk would not resolve a moved tip here).
        recodable: bool,
        /// The attempt's worktree (preserved on a terminal block).
        wt: PathBuf,
        /// The attempt's branch (preserved on a terminal block).
        branch: String,
    },
}

/// Run ONE attempt of a chunk: fork an attempt worktree off the current
/// integration tip, drive the harness with the (possibly re-briefed) brief,
/// validate the harness's claimed commit against real git state, gate the floor,
/// and — on green — merge the exact gated oid into the integration branch. Every
/// integrity check (lying commit, empty diff, rewritten history, dirty worktree)
/// is preserved from the pre-loop driver; a failure returns a re-codable
/// [`ChunkAttempt::Blocked`].
#[allow(clippy::too_many_arguments)]
fn attempt_chunk(
    run: &mut Run,
    plan: &Plan,
    chunk: &Chunk,
    harnesses: &dyn TierHarness,
    current_tier: Tier,
    baseline: &BaselineSnapshot,
    seq: u32,
    findings: &[String],
    prior_diff: Option<&str>,
) -> Result<ChunkAttempt, PipelineError> {
    let base_commit = git::head(&run.integration_wt)?;
    // Build + floor-gate the attempt off the current integration tip (design §6
    // VAIHE 2). The build/gate is shared with the concurrent wave path — the ONE
    // floor gate — so the sequential path stays byte-for-byte the same.
    let (built, usage) = build_and_gate(
        run.cfg,
        &run.repo,
        &run.slug,
        &base_commit,
        plan.plan_rev,
        chunk,
        harnesses.harness(current_tier),
        baseline,
        seq,
        findings,
        prior_diff,
        None,
    )?;
    // Meter this agent invocation into the per-run tally (design §9 cost
    // instrumentation): fold in the harness-reported Usage and count the process.
    // Done for EVERY outcome — a timeout/cancel/no-change still spent tokens.
    run.meter.record_agent_run(usage.as_ref());

    let (verdict, base, commit, chunk_wt, chunk_branch) = match built {
        BuildAttempt::Blocked {
            outcome,
            diff,
            status,
            reason,
            findings,
            floor,
            floor_passed,
            recodable,
            wt,
            branch,
        } => {
            return Ok(ChunkAttempt::Blocked {
                outcome,
                diff,
                status,
                reason,
                findings,
                floor,
                floor_passed,
                recodable,
                wt,
                branch,
            })
        }
        BuildAttempt::Built {
            verdict,
            base,
            commit,
            wt,
            branch,
        } => (verdict, base, commit, wt, branch),
    };

    // Floor green → supervisor-side merge of the EXACT gated commit oid (not the
    // mutable branch name — a stray child could have advanced the branch).
    match git::merge_no_ff(
        &run.integration_wt,
        &commit,
        &format!("pipeline: merge chunk {}", chunk.id),
    )? {
        MergeOutcome::Merged {
            commit: merge_commit,
        } => {
            // Merged into feat → drop the chunk worktree + branch. Force-delete:
            // the branch is provably merged into the integration branch (its work
            // is preserved there), but `git branch -d` checks against the repo's
            // ambient HEAD — which is not `feat/<slug>` — and so would refuse and
            // leak the branch, colliding when a later fix-loop iteration re-runs
            // the same chunk.
            let _ = git::worktree_remove(&run.repo, &chunk_wt);
            let _ = git::delete_branch(&run.repo, &chunk_branch, true);
            Ok(ChunkAttempt::Merged {
                verdict,
                base,
                commit,
                merge_commit,
            })
        }
        MergeOutcome::Conflict { details } => Ok(ChunkAttempt::Blocked {
            outcome: "committed",
            // A merge conflict is not re-codable, so its diff is never re-briefed.
            diff: None,
            status: "chunk_merge_conflict",
            reason: format!("chunk merge conflict: {details}"),
            findings: vec![format!("chunk merge conflict: {details}")],
            floor: Some(verdict),
            floor_passed: Some(true),
            recodable: false,
            wt: chunk_wt,
            branch: chunk_branch,
        }),
    }
}

/// The deterministic worktree + branch names for one build attempt of a chunk
/// (design §6). The first attempt (seq 1) keeps the bare chunk name; every later
/// attempt — re-code OR promotion — suffixes `-a{seq}` with the MONOTONIC sequence,
/// so no two attempts (across any tier) ever share a worktree/branch name, even if a
/// superseded attempt's cleanup failed. Shared by [`build_and_gate`] (which CREATES
/// them) and the wave worker ([`build_chunk_in_wave`], which PRE-REGISTERS them so a
/// commit-then-panic/error attempt is still named for the invariant-5 audit) so the
/// naming can never drift between the two.
fn chunk_attempt_names(
    cfg: &PipelineConfig,
    slug: &str,
    chunk_id: &str,
    seq: u32,
) -> (PathBuf, String) {
    let suffix = if seq == 1 {
        String::new()
    } else {
        format!("-a{seq}")
    };
    let branch = format!("{slug}/chunk-{chunk_id}{suffix}");
    let wt = cfg.workdir.join(format!("chunk-{chunk_id}{suffix}"));
    (wt, branch)
}

/// One build-and-gate attempt of a chunk, WITHOUT merging (design §6 VAIHE 2):
/// fork an attempt worktree off `base_commit`, drive the harness with the
/// (possibly re-briefed) brief, validate the claimed commit against real git state
/// (lying commit, no-advance, rewritten history, non-linear merge, dirty tree,
/// empty diff), and gate the deterministic floor (§4). Returns a floor-green
/// [`BuildAttempt::Built`] (the caller merges the exact gated oid) or a re-codable
/// [`BuildAttempt::Blocked`], plus the harness [`Usage`] for the caller to meter.
///
/// It is the SINGLE build/gate implementation: [`attempt_chunk`] (sequential,
/// forking off the moving tip) and the concurrent wave build (forking every wave
/// chunk off one shared base, then merging deterministically) both go through it,
/// so the integrity checks + floor gate can never drift between the two paths.
///
/// `git_lock`, when supplied, serialises the `git worktree add` metadata mutation
/// against sibling wave-build threads — concurrent `worktree add` on one repo can
/// race on `.git/worktrees`. The slow harness `run_chunk` stays OUTSIDE the lock so
/// independent chunks' agents genuinely run in parallel.
#[allow(clippy::too_many_arguments)]
fn build_and_gate(
    cfg: &PipelineConfig,
    repo: &Path,
    slug: &str,
    base_commit: &str,
    plan_rev: u32,
    chunk: &Chunk,
    harness: &dyn CodeHarness,
    baseline: &BaselineSnapshot,
    seq: u32,
    findings: &[String],
    prior_diff: Option<&str>,
    git_lock: Option<&Mutex<()>>,
) -> Result<(BuildAttempt, Option<Usage>), PipelineError> {
    let (chunk_wt, chunk_branch) = chunk_attempt_names(cfg, slug, &chunk.id, seq);

    {
        // Serialise ONLY the worktree-add metadata mutation when running concurrently
        // (a poisoned lock is fine to reuse — no invariant is guarded by the guard
        // itself, only mutual exclusion of the git call).
        let _guard = git_lock.map(|m| m.lock().unwrap_or_else(std::sync::PoisonError::into_inner));
        git::worktree_add_new_branch(repo, &chunk_wt, &chunk_branch, base_commit)?;
    }

    let checks: Vec<HarnessCheck> = chunk
        .checks
        .iter()
        .enumerate()
        .map(|(i, c)| to_harness_check(i, c))
        .collect();
    let req = ChunkRequest {
        run_id: format!("pipeline-{slug}"),
        chunk_id: chunk.id.clone(),
        attempt_id: format!("a{seq}"),
        worktree_path: chunk_wt.clone(),
        base_commit: base_commit.to_string(),
        plan_rev: plan_rev.to_string(),
        brief: fixloop::rebrief(&chunk.brief, findings, prior_diff),
        checks,
        files: chunk.files_touched.iter().map(PathBuf::from).collect(),
        timeout: cfg.chunk_timeout,
    };

    let cancel = CancelToken::new();
    let result = harness
        .run_chunk(&req, &cancel)
        .map_err(|e| PipelineError::Harness(e.to_string()))?;
    let usage = result.usage.clone();

    // A harness failure (no change / failed / timeout / cancelled) is re-codable;
    // its reason is the sole re-brief finding.
    let harness_block = |outcome: &'static str, reason: String| BuildAttempt::Blocked {
        outcome,
        // A harness that never produced a commit has no diff to carry forward.
        diff: None,
        status: "chunk_failed",
        findings: vec![reason.clone()],
        reason,
        floor: None,
        floor_passed: None,
        recodable: true,
        wt: chunk_wt.clone(),
        branch: chunk_branch.clone(),
    };

    let commit = match &result.outcome {
        ChunkOutcome::Committed { commit } => commit.clone(),
        ChunkOutcome::NoChange => {
            return Ok((
                harness_block("no_change", "chunk produced no commit".to_string()),
                usage,
            ))
        }
        ChunkOutcome::Failed { reason } => {
            return Ok((harness_block("failed", reason.clone()), usage))
        }
        ChunkOutcome::Timeout => {
            return Ok((
                harness_block("timeout", "chunk timed out".to_string()),
                usage,
            ))
        }
        ChunkOutcome::Cancelled => {
            return Ok((
                harness_block("cancelled", "chunk cancelled".to_string()),
                usage,
            ))
        }
    };

    // Validate the harness's claimed commit against real git state BEFORE the
    // floor — an adapter that lies (reports a commit but left HEAD unmoved,
    // committed an empty/rewritten tree, or left the passing work uncommitted)
    // must not slip a merge past the floor. These are re-codable failures.
    let head = git::head(&chunk_wt)?;
    if head != commit {
        return Ok((
            harness_block(
                "failed",
                format!("harness reported commit {commit} but worktree HEAD is {head}"),
            ),
            usage,
        ));
    }
    if head == base_commit {
        return Ok((
            harness_block(
                "no_change",
                "harness reported a commit but HEAD did not advance".to_string(),
            ),
            usage,
        ));
    }
    if !git::is_ancestor(&chunk_wt, base_commit, &head)? {
        return Ok((
            harness_block(
                "failed",
                format!(
                    "chunk commit {head} is not a descendant of its base {base_commit} (history rewritten)"
                ),
            ),
            usage,
        ));
    }
    // Reject a non-linear chunk history (item F): the provenance rollback replays a
    // kept chunk with `git cherry-pick base..commit`, which cannot replay a merge
    // commit without `-m` (it aborts, which the rollback would misread as a
    // conflict). A chunk that merged something into its own range must be flattened
    // by the harness before it can be a safe, replayable unit — treat it as a
    // re-codable failure here rather than letting it become an un-replayable merged
    // chunk later.
    if git::range_has_merge(&chunk_wt, base_commit, &head)? {
        return Ok((
            harness_block(
                "failed",
                format!(
                    "chunk history {base_commit}..{head} contains a merge commit; the provenance rollback replays a linear range and cannot cherry-pick a merge — the chunk must be a linear sequence of commits"
                ),
            ),
            usage,
        ));
    }
    if !git::is_clean(&chunk_wt)? {
        return Ok((
            harness_block(
                "failed",
                "chunk worktree has uncommitted changes after the commit".to_string(),
            ),
            usage,
        ));
    }
    let changed = floor::git::changed_files(&chunk_wt, base_commit, &head)?;
    if changed.is_empty() {
        return Ok((
            harness_block("no_change", "committed chunk has an empty diff".to_string()),
            usage,
        ));
    }

    let verdict = gate_chunk(cfg, repo, chunk, &chunk_wt, base_commit, &changed, baseline)?;
    if !verdict.passed() {
        // Floor blocked → re-codable, with the floor violations as findings.
        // Capture the failing diff now, BEFORE the worktree is torn down for the
        // retry, so the re-brief carries the code this attempt actually produced
        // (item 3: re-code amnesia). Best-effort — a diff failure must not mask the
        // floor block, so it degrades to no carried diff.
        let diff = git::diff(&chunk_wt, base_commit, &head).ok();
        return Ok((
            BuildAttempt::Blocked {
                outcome: "committed",
                diff,
                status: "chunk_floor_blocked",
                reason: "floor gate failed".to_string(),
                findings: fixloop::floor_findings(&verdict),
                floor: Some(verdict),
                floor_passed: Some(false),
                recodable: true,
                wt: chunk_wt,
                branch: chunk_branch,
            },
            usage,
        ));
    }

    Ok((
        BuildAttempt::Built {
            verdict,
            base: base_commit.to_string(),
            commit: head,
            wt: chunk_wt,
            branch: chunk_branch,
        },
        usage,
    ))
}

/// The result of one [`build_and_gate`] attempt: a floor-green built commit (not
/// yet merged) or a block. Mirrors [`ChunkAttempt`] minus the merge outcomes — the
/// caller owns the merge, so a `chunk_merge_conflict` is produced there, not here.
enum BuildAttempt {
    /// Floor green; the exact `commit` is ready for the caller to merge.
    Built {
        verdict: FloorVerdict,
        /// The base the chunk forked from (its provenance replay base).
        base: String,
        /// The chunk's own floor-gated tip commit oid.
        commit: String,
        /// The attempt's worktree (torn down by the caller once merged).
        wt: PathBuf,
        /// The attempt's branch.
        branch: String,
    },
    /// The attempt did not produce a mergeable commit; fields mirror
    /// [`ChunkAttempt::Blocked`].
    Blocked {
        outcome: &'static str,
        diff: Option<String>,
        status: &'static str,
        reason: String,
        findings: Vec<String>,
        floor: Option<FloorVerdict>,
        floor_passed: Option<bool>,
        recodable: bool,
        wt: PathBuf,
        branch: String,
    },
}

/// Push a chunk report, replacing any existing report for the same chunk id
/// (a chunk can be re-run across fix-loop iterations — the latest outcome wins,
/// so the report stays one-per-chunk).
fn upsert_chunk_report(run: &mut Run, report: ChunkReport) {
    run.chunk_reports.retain(|r| r.id != report.id);
    run.chunk_reports.push(report);
}

/// Record a chunk that did not merge and mark its worktree/branch preserved for
/// inspection (state-integrity invariant 5). Upserts so a re-run's terminal block
/// replaces any earlier report for the chunk.
#[allow(clippy::too_many_arguments)]
fn push_blocked_chunk(
    run: &mut Run,
    chunk: &Chunk,
    outcome: &str,
    floor: Option<FloorVerdict>,
    floor_passed: Option<bool>,
    reason: String,
    chunk_wt: &Path,
    chunk_branch: &str,
) {
    run.preserved
        .push((chunk_wt.to_path_buf(), chunk_branch.to_string()));
    upsert_chunk_report(
        run,
        ChunkReport {
            id: chunk.id.clone(),
            title: chunk.title.clone(),
            tier: chunk.tier.wire_name().to_string(),
            outcome: outcome.to_string(),
            floor_passed,
            floor,
            merged: false,
            commit: None,
            merge_commit: None,
            replayed: false,
            reason: Some(reason),
            branch_preserved: Some(chunk_branch.to_string()),
        },
    );
}

/// Evaluate the per-chunk floor (design §4): the chunk's own checks pass, no
/// baseline regression / new clippy / test-gaming, and the changed files stay in
/// scope. Test/clippy regressions are judged against the fork baseline; the
/// assertion-density signal is judged against the chunk's own **base commit**
/// (the current integration tip), not the fork — so a later chunk that guts a
/// test an earlier chunk added is caught, instead of hiding behind the fork's
/// lower count. File-scope is against the chunk's `files_touched`.
///
/// Takes `cfg` + `repo` directly (not `&Run`) so it can run inside a concurrent
/// wave-build thread that holds no `&Run` (design §6 VAIHE 2) — the floor gate is
/// the one hard correctness gate and MUST be byte-for-byte the same on the
/// sequential and the concurrent path.
fn gate_chunk(
    cfg: &PipelineConfig,
    repo: &Path,
    chunk: &Chunk,
    chunk_wt: &Path,
    base_commit: &str,
    changed: &[PathBuf],
    baseline: &BaselineSnapshot,
) -> Result<FloorVerdict, PipelineError> {
    let check_results: Vec<CheckRun> = floor::runner::run_checks(&chunk.checks, chunk_wt);
    let current = capture_snapshot(cfg, chunk_wt)?;
    let declared: Vec<PathBuf> = chunk.files_touched.iter().map(PathBuf::from).collect();
    let baseline_assertions = floor::runner::assertion_counts_at_ref(repo, base_commit, &declared)?;
    let current_assertions = floor::runner::assertion_counts_on_disk(chunk_wt, &declared);

    let inputs = FloorInputs {
        baseline: &baseline.snapshot,
        current: &current,
        check_results: &check_results,
        declared_files: &declared,
        changed_files: changed,
        baseline_assertions: &baseline_assertions,
        current_assertions: &current_assertions,
        file_scope_slack: cfg.file_scope_slack,
    };
    Ok(evaluate_floor(&inputs))
}

/// Run the plan's executable acceptance checks, then ask the verify provider to
/// judge product-vs-intent (design §6 VAIHE 3). Returns the verify report. The
/// feature-floor re-check re-runs the acceptance checks itself (on the pristine,
/// gated tip) rather than reusing these results, so a verify-time mutation can
/// never leave a stale-green acceptance result behind the final gate.
fn run_verify_stage(
    run: &mut Run,
    plan: &Plan,
    verify: &dyn VerifyProvider,
) -> Result<(VerifyReport, VerifyDisposition), PipelineError> {
    let acceptance_checks: Vec<plan::Check> = plan
        .acceptance
        .iter()
        .filter_map(acceptance_to_check)
        .collect();
    let acceptance_results = floor::runner::run_checks(&acceptance_checks, &run.integration_wt);
    let acceptance_checks_passed = acceptance_results.iter().all(|r| r.passed);

    let judgment: VerifyJudgment = verify.verify(&VerifyContext {
        intent: &run.cfg.intent,
        plan,
        worktree: &run.integration_wt,
        acceptance_results: &acceptance_results,
    })?;
    // Count the verify agent invocation toward the process-count breaker (design
    // §9). Verify does not surface Usage through its trait, so no tokens/cost are
    // added — a documented follow-up (spec/verify token accounting).
    run.meter.record_agent_run(None);

    run.decisions.push(envelope(
        "verify",
        DecisionTier::Decider,
        format!(
            "acceptance checks {}, judge {}",
            if acceptance_checks_passed {
                "passed"
            } else {
                "FAILED"
            },
            if judgment.passed { "passed" } else { "FAILED" }
        ),
        vec![format!("plan:{}", plan.plan_rev)],
        verify.model(),
        verify.prompt_version(),
    ));

    // The combined verdict is mechanical (acceptance checks) ∧ judged. The judge's
    // disposition (FIX vs SPEC-FLAW) only carries a signal when the JUDGE failed;
    // if the judge passed but an acceptance check failed, there is no SPEC-FLAW
    // signal, so fall back to a bare FIX (re-code, don't re-spec).
    let passed = acceptance_checks_passed && judgment.passed;
    let disposition = if judgment.passed {
        VerifyDisposition::Fix
    } else {
        judgment.disposition.clone()
    };

    // Build the re-brief findings. Feed the FAILED acceptance checks in as
    // mechanical findings (a judge-passed / check-failed verdict would otherwise
    // re-code with no context and just reproduce the same output → NoChange →
    // breaker). And guarantee at least one finding on any failure, so the
    // RE_CODE_CHUNK re-brief always differs from the original brief.
    let mut findings = judgment.findings;
    for r in acceptance_results.iter().filter(|r| !r.passed) {
        findings.push(format!("acceptance check failed: {} (`{}`)", r.desc, r.run));
    }
    if !passed && findings.is_empty() {
        findings.push(format!(
            "verify failed without specific findings: {}. Review the implementation against the intent and correct it.",
            summary_or_default(&judgment.summary)
        ));
    }

    Ok((
        VerifyReport {
            acceptance_checks_passed,
            judged_passed: judgment.passed,
            passed,
            summary: judgment.summary,
            findings,
        },
        disposition,
    ))
}

/// The judge summary, or a placeholder when it is blank — so a synthetic
/// fallback finding is never an empty sentence.
fn summary_or_default(summary: &str) -> &str {
    if summary.trim().is_empty() {
        "(no summary)"
    } else {
        summary
    }
}

/// Convert an executable `acceptance` item into a runnable [`plan::Check`];
/// LLM-judged `assertion` items have no command and yield `None`.
fn acceptance_to_check(a: &Acceptance) -> Option<plan::Check> {
    match a {
        Acceptance::Check {
            desc,
            run,
            cwd,
            expect_exit,
        } => Some(plan::Check {
            desc: desc.clone(),
            run: run.clone(),
            cwd: cwd.clone(),
            expect_exit: *expect_exit,
            extra: serde_json::Map::new(),
        }),
        Acceptance::Assertion { .. } => None,
    }
}

/// The feature-level floor re-check before the final merge (design §4: the floor
/// is re-checked at the tip). Scoped to the whole feature: the acceptance checks
/// are re-run FRESH on the current (restored, gated) tip, changed files are
/// `fork..feat`, declared files are the union. The assertion-density baseline is
/// the fork (the whole feature is judged against the pre-feature state).
fn evaluate_feature_floor(
    run: &Run,
    plan: &Plan,
    baseline: &BaselineSnapshot,
    declared: &[PathBuf],
    feat_tip: &str,
) -> Result<FloorVerdict, PipelineError> {
    let acceptance_checks: Vec<plan::Check> = plan
        .acceptance
        .iter()
        .filter_map(acceptance_to_check)
        .collect();
    let check_results = floor::runner::run_checks(&acceptance_checks, &run.integration_wt);
    let current = capture_snapshot(run.cfg, &run.integration_wt)?;
    let changed = floor::git::changed_files(&run.integration_wt, &run.fork_commit, feat_tip)?;
    let baseline_assertions =
        floor::runner::assertion_counts_at_ref(&run.repo, &run.fork_commit, declared)?;
    let current_assertions = floor::runner::assertion_counts_on_disk(&run.integration_wt, declared);
    let inputs = FloorInputs {
        baseline: &baseline.snapshot,
        current: &current,
        check_results: &check_results,
        declared_files: declared,
        changed_files: &changed,
        baseline_assertions: &baseline_assertions,
        current_assertions: &current_assertions,
        file_scope_slack: run.cfg.file_scope_slack,
    };
    Ok(evaluate_floor(&inputs))
}

/// Merge the exact floor-gated `feat_tip` oid into the source branch (design §6
/// VAIHE 4). Merges the OID (not the mutable branch name) in the worktree that
/// has the source branch checked out (verified clean) when there is one;
/// otherwise materializes a throwaway worktree, merges, and removes it. Returns
/// the merge [`MergeOutcome`] so the driver can report a conflict rather than
/// crash (the source branch may have moved after the floor turned green).
fn merge_feature_to_source(run: &Run, feat_tip: &str) -> Result<MergeOutcome, PipelineError> {
    let message = format!(
        "pipeline: merge {} into {}",
        run.integration_branch, run.cfg.source_branch
    );
    if let Some(src_wt) = git::worktree_for_branch(&run.repo, &run.cfg.source_branch)? {
        if !git::is_clean(&src_wt)? {
            return Err(PipelineError::Setup(format!(
                "source branch `{}` worktree {} is dirty; cannot merge",
                run.cfg.source_branch,
                src_wt.display()
            )));
        }
        git::merge_no_ff(&src_wt, feat_tip, &message)
    } else {
        // Source branch not checked out anywhere: materialize a scratch worktree.
        let src_wt = run.cfg.workdir.join("source-merge");
        git::worktree_add(&run.repo, &src_wt, &run.cfg.source_branch)?;
        let out = git::merge_no_ff(&src_wt, feat_tip, &message);
        let _ = git::worktree_remove(&run.repo, &src_wt);
        out
    }
}

/// Build the final report from the accumulated run state.
fn finalize(
    run: &Run,
    plan: &Plan,
    verify: Option<VerifyReport>,
    merged: bool,
    final_commit: Option<String>,
    status: &str,
) -> PipelineReport {
    let failure = match status {
        "merged" => None,
        "chunk_floor_blocked" => {
            Some("a chunk failed the deterministic floor; the feature was not merged".to_string())
        }
        "chunk_merge_conflict" => {
            Some("a chunk floor-passed but conflicted merging into the integration branch".to_string())
        }
        "chunk_failed" => {
            Some("a chunk failed to produce a mergeable commit (harness failure / no change / timeout)".to_string())
        }
        "verify_failed" => {
            Some("verify judged the product does not match intent (or an acceptance check failed); not merged".to_string())
        }
        "floor_blocked" => Some("the feature floor regressed at the tip; not merged".to_string()),
        "rollback_conflict" => Some(format!(
            "the provenance rollback could not cleanly replay kept chunk{} onto the rebuilt integration branch; not merged — the integration branch was restored intact and its work preserved",
            run.rollback_conflict
                .as_ref()
                .map(|c| format!(" `{c}`"))
                .unwrap_or_default()
        )),
        "escalated" => Some(
            "the decider escalated a consequential decision (declined to converge or re-spec) and handed the feature up; not merged — see the decision log for the reason".to_string(),
        ),
        "merge_conflict" => {
            Some("the feature floor was green but the source branch moved and the merge conflicted".to_string())
        }
        "circuit_breaker" => Some(
            run.circuit_breaker
                .clone()
                .unwrap_or_else(|| "a circuit-breaker stopped the fix loop; not merged".to_string()),
        ),
        other => Some(other.to_string()),
    };
    PipelineReport {
        slug: run.slug.clone(),
        source_branch: run.cfg.source_branch.clone(),
        integration_branch: run.integration_branch.clone(),
        intent_rev: 1,
        plan_rev: plan.plan_rev,
        chunk_count: plan.chunks.len(),
        chunks: run.chunk_reports.clone(),
        verify,
        feature_floor: run.feature_floor.clone(),
        merged,
        final_commit,
        status: status.to_string(),
        decisions: run.decisions.clone(),
        recode_count: run.recode_count,
        promote_count: run.promote_count,
        respec_count: run.respec_count,
        circuit_breaker: run.circuit_breaker.clone(),
        resources: run.meter.clone(),
        failure,
    }
}

/// Tear down the scratch worktrees/branches (design §6 VAIHE 4 teardown), gated
/// on the terminal outcome (state-integrity invariant 5): the integration
/// worktree is always removed, but the integration branch is deleted only when
/// the feature merged to source; a preserved (unmerged) chunk keeps both its
/// worktree and branch. `--keep` skips teardown entirely.
fn teardown(run: &Run) {
    if run.cfg.keep {
        return;
    }
    // Remove the integration worktree (its work is merged or abandoned).
    let _ = git::worktree_remove(&run.repo, &run.integration_wt);

    // Delete the integration branch only when it holds no unmerged work: either
    // it merged to source (redundant), or — on an early failure (bad spec) — it
    // never accumulated a chunk commit beyond the source branch. If a chunk did
    // merge into it but the feature never reached source, PRESERVE it: those are
    // real, unmerged commits (state-integrity invariant 5, source-relative check).
    let safe_to_delete = run.merged_to_source
        || git::commits_ahead_of(&run.repo, &run.cfg.source_branch, &run.integration_branch)
            .is_ok_and(|n| n == 0);
    // Item G: prune this run's durable provenance refs on the SAME gate as branch
    // deletion, and only once the integration branch they backstop is actually gone.
    // The refs exist to protect the kept chunks' authored OIDs across the destructive
    // rollback window (a reset that orphans them from every branch tip). That window
    // closes when the run ends: on a `safe_to_delete` outcome (merged to source, or an
    // early failure that never accumulated work) the run is done and the anchors are
    // intentionally RELEASED — this deliberately ends durability for the authored OIDs
    // (which a rollback made unreachable from `source_branch`, so they may be gc'd
    // afterwards; `report.chunks[*].commit` is a historical oid, not a promise that it
    // still resolves post-teardown). On the PRESERVED (unmerged) path the whole block
    // is skipped, so a preserved branch's refs stay exactly as long as its branch does.
    //
    // The prune is gated on the branch actually being deleted: if `delete_branch`
    // fails (e.g. the branch is checked out elsewhere) the branch still roots its
    // on-branch commits, and dropping the refs then would strip gc protection from the
    // now-orphaned authored OIDs while the branch lingers — so keep the refs too.
    if safe_to_delete {
        // `-d` refuses to drop an unmerged branch, so even a race here fails closed
        // rather than losing work.
        let branch_deleted =
            git::delete_branch(&run.repo, &run.integration_branch, run.merged_to_source).is_ok();
        if branch_deleted {
            if let Ok(refs) = git::refs_under(&run.repo, &provenance_ref_prefix(&run.slug)) {
                for r in &refs {
                    let _ = git::delete_ref(&run.repo, r);
                }
            }
        }
    }
    // Preserved chunk worktrees/branches are intentionally left in place — they
    // hold unmerged work (invariant 5). Nothing else to remove.
}

// --- CLI entry --------------------------------------------------------------

/// Parsed `pipeline run` arguments, kept independent of clap so the wiring can
/// be exercised without the parser.
pub struct PipelineRunConfig {
    /// The raw `--intent` value (string or file path / `@file`).
    pub intent: String,
    /// The source branch.
    pub source_branch: String,
    /// Optional file-scope hints.
    pub files: Vec<PathBuf>,
    /// Optional slug override.
    pub slug: Option<String>,
    /// Optional repo path (default: cwd).
    pub repo: Option<PathBuf>,
    /// Optional test-capture command override.
    pub test_cmd: Option<String>,
    /// Optional clippy-capture command override.
    pub clippy_cmd: Option<String>,
    /// Optional workdir override.
    pub workdir: Option<PathBuf>,
    /// File-scope slack.
    pub file_scope_slack: usize,
    /// Keep worktrees after the run.
    pub keep: bool,
    /// Optional per-chunk timeout (seconds).
    pub chunk_timeout_secs: Option<u64>,
    /// Max concurrent chunk builds per dependency wave (design §6 VAIHE 2). `None`
    /// (or `< 1`) uses the sequential default (1).
    pub max_build_concurrency: Option<usize>,
    /// Max `RE_CODE` re-attempts per chunk in the code stage (design §8/§9).
    /// `None` uses the live default.
    pub max_recode_per_chunk: Option<u32>,
    /// Max verify→fix cycles (design §8/§9). `None` uses the live default.
    pub max_fix_iterations: Option<u32>,
    /// Max `TRIGGER_RE_SPEC` events (design §7/§9). `None` uses the live default.
    pub max_respec: Option<u32>,
    /// Max `PROMOTE_TIER` promotions per chunk (design §3). `None` uses the live
    /// default; `0` disables adaptive promotion.
    pub max_promotions: Option<u32>,
    /// Resource circuit-breaker: cost ceiling in USD (design §9). `None` uses the
    /// live default; `0` disables the cost breaker.
    pub max_cost_usd: Option<f64>,
    /// Resource circuit-breaker: total-token ceiling (design §9). `None` uses the
    /// live default; `0` disables it.
    pub max_total_tokens: Option<u64>,
    /// Resource circuit-breaker: wall-time ceiling in seconds (design §9). `None`
    /// uses the live default; `0` disables it.
    pub max_wall_time_secs: Option<u64>,
    /// Resource circuit-breaker: max agent invocations (design §9). `None` uses the
    /// live default; `0` disables it.
    pub max_processes: Option<u32>,
    /// Resource circuit-breaker: scratch-storage ceiling in MiB (design §9). `None`
    /// uses the live default; `0` disables it.
    pub max_storage_mb: Option<u64>,
    /// Resource circuit-breaker: identical-failure recurrence ceiling (design §9).
    /// `None` uses the live default; `0` disables it.
    pub max_identical_failures: Option<u32>,
}

/// Resolve a `u64` resource ceiling from an optional CLI override: an explicit
/// `0` disables the breaker (`None`), any other value overrides, and an absent
/// flag falls back to `default` (design §9: `0` = off, uniform across breakers).
fn resolve_u64_ceiling(user: Option<u64>, default: Option<u64>) -> Option<u64> {
    match user {
        Some(0) => None,
        Some(v) => Some(v),
        None => default,
    }
}

/// [`resolve_u64_ceiling`] for a `u32` ceiling.
fn resolve_u32_ceiling(user: Option<u32>, default: Option<u32>) -> Option<u32> {
    match user {
        Some(0) => None,
        Some(v) => Some(v),
        None => default,
    }
}

/// [`resolve_u64_ceiling`] for a USD (`f64`) ceiling: only a finite, positive value
/// enables the cost breaker. A non-finite (`NaN`/`inf` — both of which `f64::parse`
/// accepts) or non-positive value disables it (`None`), never leaves it enabled but
/// impossible to trip. `PipelineRunConfig` is constructable from code, so this guard
/// is the authoritative one even though the CLI could also validate.
fn resolve_f64_ceiling(user: Option<f64>, default: Option<f64>) -> Option<f64> {
    match user {
        Some(v) if v.is_finite() && v > 0.0 => Some(v),
        Some(_) => None,
        None => default,
    }
}

/// `pipeline run` entry point: resolve config, wire the LIVE Claude/deepseek
/// stages (design §10: spec/verify = `claude` Opus, code = `claude-deepseek`),
/// run the pipeline, and emit the report envelope.
pub fn cmd_run(
    cfg: &PipelineRunConfig,
    spec: &OutputSpec,
    warnings: &[String],
) -> Result<(), CliError> {
    let intent = resolve_intent(&cfg.intent)?;
    let repo = cfg
        .repo
        .clone()
        .unwrap_or_else(|| std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")));
    let slug_preview = cfg.slug.clone().unwrap_or_else(|| slugify(&intent));
    let workdir = cfg.workdir.clone().unwrap_or_else(|| {
        std::env::temp_dir()
            .join("octl-pipeline")
            .join(&slug_preview)
    });

    let pcfg = PipelineConfig {
        repo,
        intent,
        source_branch: cfg.source_branch.clone(),
        files: cfg.files.clone(),
        slug: cfg.slug.clone(),
        test_cmd: cfg
            .test_cmd
            .clone()
            .unwrap_or_else(|| "cargo test".to_string()),
        clippy_cmd: cfg
            .clippy_cmd
            .clone()
            .unwrap_or_else(|| "cargo clippy".to_string()),
        workdir,
        file_scope_slack: cfg.file_scope_slack,
        keep: cfg.keep,
        chunk_timeout: cfg.chunk_timeout_secs.map(Duration::from_secs),
        // Sequential by default (1); a `0`/`None` collapses to 1 so the field is
        // always a valid concurrency bound the wave scheduler can `min` against.
        max_build_concurrency: cfg.max_build_concurrency.filter(|&n| n > 1).unwrap_or(1),
        fix_loop: {
            // The verify→triage→fix loop is ON by default for the live command
            // (design §7/§8), bounded by the §9 breakers; each bound is
            // individually overridable.
            let d = FixLoopConfig::live_default();
            FixLoopConfig {
                max_recode_per_chunk: cfg.max_recode_per_chunk.unwrap_or(d.max_recode_per_chunk),
                max_fix_iterations: cfg.max_fix_iterations.unwrap_or(d.max_fix_iterations),
                max_respec: cfg.max_respec.unwrap_or(d.max_respec),
                max_promotions: cfg.max_promotions.unwrap_or(d.max_promotions),
            }
        },
        budget: {
            // The deterministic resource breakers are ON by default for the live
            // command (design §9), each ceiling individually overridable; a supplied
            // `0` disables that one breaker (`None` in the resolved budget). Wall-time
            // is expressed in seconds and storage in MiB on the CLI; both round-trip
            // through the resolve helper in those units, then convert.
            let d = ResourceBudget::live_default();
            ResourceBudget {
                max_cost_usd: resolve_f64_ceiling(cfg.max_cost_usd, d.max_cost_usd),
                max_total_tokens: resolve_u64_ceiling(cfg.max_total_tokens, d.max_total_tokens),
                max_wall_time: resolve_u64_ceiling(
                    cfg.max_wall_time_secs,
                    d.max_wall_time.map(|w| w.as_secs()),
                )
                .map(Duration::from_secs),
                max_processes: resolve_u32_ceiling(cfg.max_processes, d.max_processes),
                max_storage_bytes: resolve_u64_ceiling(
                    cfg.max_storage_mb,
                    d.max_storage_bytes.map(|b| b / (1024 * 1024)),
                )
                .map(|mb| mb.saturating_mul(1024 * 1024)),
                max_identical_failures: resolve_u32_ceiling(
                    cfg.max_identical_failures,
                    d.max_identical_failures,
                ),
            }
        },
    };

    // LIVE stages: spec/verify on ambient-login `claude` (Opus). The code stage is
    // a per-tier ladder (design §3/§10): cheap `claude-deepseek flash` at the base,
    // `claude-deepseek pro` at mid, and ambient Opus `claude` at high — so a
    // PROMOTE_TIER re-run actually escalates the model. Every adapter self-sources
    // its own credentials (no secret is read or hardcoded here).
    let spec_provider = providers::ClaudeSpecProvider;
    let verify_provider = providers::ClaudeVerifyProvider;
    use crate::harness::claude::ClaudeHarness;
    let harnesses = LiveTierHarness {
        code: ClaudeHarness::deepseek("flash"),
        mid: ClaudeHarness::deepseek("pro"),
        high: ClaudeHarness::claude(Some("opus".to_string())),
    };
    // The consequential-decision authority (design §0.2/§2): verify/spec are Opus in
    // the live path, so the decider records that provenance on decider-tier
    // envelopes. The routine coordination decisions never reach it.
    let decider = LiveDecider {
        model: verify_provider.model(),
    };

    let report = match run_pipeline_tiered(
        &pcfg,
        &spec_provider,
        &harnesses,
        &verify_provider,
        &decider,
    ) {
        Ok(report) => report,
        Err(PipelineFailure { error, report }) => {
            // A hard failure past the spec stage still exits NON-ZERO (the error's
            // stable code drives the envelope), but first render the accumulated
            // report so the invariant-5 audit — every `branch_preserved` sibling
            // that committed work but never merged — is visible on the failure path
            // rather than lost (`pipeline-hard-failure-carries-report`). A pre-plan
            // failure carries no report; there is nothing to audit yet.
            if let Some(report) = report {
                // Best-effort: the report goes to stdout, but the DOMINANT signal is
                // the error (its stable code + non-zero exit, emitted by the dispatcher
                // to stderr). A failure to render the report must NOT replace the
                // pipeline error's exit code — so swallow an emit error here rather than
                // `?`-propagating it (which would exit with the emit error's code and
                // hide the real failure). See `pipeline-hard-failure-carries-report`.
                match spec.format {
                    OutputFormat::Json | OutputFormat::Jsonl => {
                        if let Err(e) = output::emit_envelope(&report, spec, warnings) {
                            eprintln!("warning: could not render failure report: {}", e.message);
                        }
                    }
                    OutputFormat::Text => {
                        print_report(&report);
                        output::emit_text_warnings(warnings);
                    }
                }
            }
            return Err(error.into());
        }
    };

    match spec.format {
        OutputFormat::Json | OutputFormat::Jsonl => output::emit_envelope(&report, spec, warnings)?,
        OutputFormat::Text => {
            print_report(&report);
            output::emit_text_warnings(warnings);
        }
    }
    Ok(())
}

/// Render a byte count in the largest binary unit that keeps it ≥ 1 (e.g. `2.0
/// GiB`, `512 B`), for the human-readable resource line.
fn human_bytes(bytes: u64) -> String {
    const UNITS: [&str; 5] = ["B", "KiB", "MiB", "GiB", "TiB"];
    let mut v = bytes as f64;
    let mut unit = 0;
    while v >= 1024.0 && unit < UNITS.len() - 1 {
        v /= 1024.0;
        unit += 1;
    }
    if unit == 0 {
        format!("{bytes} B")
    } else {
        format!("{v:.1} {}", UNITS[unit])
    }
}

/// Render the human-readable pipeline summary (`--output text`).
fn print_report(r: &PipelineReport) {
    println!("pipeline {} — {}", r.slug, r.status);
    println!(
        "  source: {}  integration: {}",
        r.source_branch, r.integration_branch
    );
    println!("  chunks: {}", r.chunk_count);
    for c in &r.chunks {
        let floor = match c.floor_passed {
            Some(true) => "floor:green",
            Some(false) => "floor:BLOCKED",
            None => "floor:-",
        };
        let merged = if c.merged { "merged" } else { "not-merged" };
        println!(
            "    [{}] {} — {} {} {}",
            c.id, c.title, c.outcome, floor, merged
        );
        if let Some(reason) = &c.reason {
            println!("        reason: {}", output::escape_one_line(reason));
        }
        // Surface the invariant-5 audit: a preserved (unmerged, committed) chunk names
        // the branch its work was kept on, so a hard-failure report shows exactly where
        // each sibling's work survives.
        if let Some(branch) = &c.branch_preserved {
            println!("        preserved: {}", output::escape_one_line(branch));
        }
    }
    if let Some(v) = &r.verify {
        println!(
            "  verify: {} (acceptance-checks: {}, judge: {}) — {}",
            if v.passed { "passed" } else { "FAILED" },
            v.acceptance_checks_passed,
            v.judged_passed,
            output::escape_one_line(&v.summary)
        );
    }
    if r.recode_count > 0 || r.respec_count > 0 || r.promote_count > 0 {
        println!(
            "  fix loop: {} re-code(s), {} promotion(s), {} re-spec(s) → plan.v{}",
            r.recode_count, r.promote_count, r.respec_count, r.plan_rev
        );
    }
    match (&r.merged, &r.final_commit) {
        (true, Some(commit)) => println!("  merged → {} @ {}", r.source_branch, commit),
        _ => println!("  merged: no"),
    }
    {
        let res = &r.resources;
        println!(
            "  resources: {} token(s), ${:.4}, {} agent invocation(s), {} scratch storage",
            res.total_tokens,
            res.cost_usd,
            res.processes,
            human_bytes(res.storage_bytes)
        );
    }
    if let Some(cb) = &r.circuit_breaker {
        println!("  circuit-breaker: {}", output::escape_one_line(cb));
    }
    if let Some(f) = &r.failure {
        println!("  failure: {}", output::escape_one_line(f));
    }
    for d in &r.decisions {
        println!(
            "  decision[{}] {}: {}",
            match d.decision_tier {
                DecisionTier::Coordinator => "coordinator",
                DecisionTier::Decider => "decider",
            },
            d.actor,
            output::escape_one_line(&d.reason)
        );
    }
}

#[cfg(test)]
mod tests;