nornir 0.4.21

//! Release pipeline orchestrator.
//!
//! Walks the workspace dep graph in build order and, for each repo,
//! runs the release gates (build → test → publish) while recording
//! provenance into the warehouse's `release_lineage` table.
//!
//! The recorded lineage — per repo: git SHA, dirty flag, gate status,
//! published versions, plus the dep-graph snapshot id that pinned the
//! plan — is enough to reconstruct *exactly which sources shipped
//! together*. That's the rollback story: pick a release_id from Urðr,
//! get back the per-repo SHAs, `git checkout` them, and you have the
//! tree as it was at release time.
//!
//! Cargo subprocess use here follows the same rationale documented in
//! [`super::publish::run_cargo_publish`] — cargo owns build/test/publish;
//! we don't reimplement it. Everything else (git inspection,
//! warehouse writes) stays pure Rust via `gix` and `iceberg-rust`.

use std::collections::{BTreeMap, VecDeque};
use std::io::{BufRead, BufReader};
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::{Arc, Mutex};
use std::time::Instant;

use anyhow::{Context, Result};
use arrow::array::{
    Array, BooleanArray, Int32Array, RecordBatch, StringArray, TimestampMicrosecondArray,
};
use chrono::Utc;
use futures::TryStreamExt;
use iceberg::Catalog;
use iceberg::arrow::schema_to_arrow_schema;
use uuid::Uuid;

use crate::warehouse::dep_graph::WorkspaceGraph;
use crate::warehouse::iceberg::{self as wh_iceberg, IcebergWarehouse};
use crate::warehouse::release_events::{ReleaseEventEmitter, phase, status};
use crate::workspace::descriptor::WorkspaceDescriptor;

/// The upstream components `repo` waited on in the build order — the
/// dep-graph producers (`from == repo` edges point consumer→producer, so
/// `to` is the dependency). De-duplicated + sorted for a stable
/// `depends_on` list. Empty `Vec` for a root with no in-workspace deps.
fn depends_on_for(graph: &WorkspaceGraph, repo: &str) -> Vec<String> {
    let mut deps: Vec<String> = graph
        .dependencies_of(repo)
        .iter()
        .map(|e| e.to.clone())
        .filter(|d| d != repo)
        .collect();
    deps.sort();
    deps.dedup();
    deps
}

/// Options controlling a single release run.
#[derive(Debug, Clone)]
pub struct PipelineOptions {
    /// If true, skip the publish gate entirely (don't even call cargo
    /// with `--dry-run`). Lineage row records `dry_run = true` and
    /// `published_versions_csv` stays empty.
    pub dry_run: bool,
    /// Dep-graph snapshot this release is pinned to. Use
    /// [`crate::warehouse::dep_graph::record_dep_graph`] first, then
    /// pass the returned UUID here.
    pub dep_graph_snapshot_id: Uuid,
    /// If true (default), refuse to start the release when any repo
    /// has uncommitted changes or untracked files. Rationale: nornir
    /// only remembers git SHAs as the identity of shipped code; a
    /// dirty tree means the SHA doesn't fully describe what would
    /// ship, so the release can't be safely rolled back later. The
    /// fix is always one of: `git commit` the work, or
    /// `git reset --hard <sha>` to the last clean release SHA (look
    /// it up via [`query_release_history`]).
    pub refuse_if_dirty: bool,
    /// Optional progress writer. When set, `run_pipeline` emits live
    /// `ReleaseEvent`s into the underlying ndjson file so the egui
    /// viz / `nornir-server` SSE can render the run live.
    pub progress: Option<crate::release::progress::ProgressWriter>,
    /// Optional gate configuration. When present, per-repo gates that
    /// need config (`no_regression`, `guard_intact`) are evaluated as
    /// release-blocking checks. Absent ⇒ those gates are skipped
    /// (back-compat: existing callers that only smoke-test build/test).
    pub gate_config: Option<PipelineGateConfig>,
}

/// Config for the release-blocking gates that need more than the
/// workspace descriptor: regression history keys off per-repo
/// [`Gates`], and `guard_intact` needs the workspace root to load the
/// recorded guard manifest.
#[derive(Debug, Clone)]
pub struct PipelineGateConfig {
    /// Workspace root (parent of `workspace_holger`); where
    /// `.nornir/guard-manifest.json` lives.
    pub workspace_root: PathBuf,
    /// Per-repo gate flags, keyed by repo name (from `nornir.toml`).
    pub repo_gates: BTreeMap<String, crate::config::Gates>,
}

impl Default for PipelineOptions {
    fn default() -> Self {
        Self {
            dry_run: true,
            dep_graph_snapshot_id: Uuid::nil(),
            refuse_if_dirty: true,
            progress: None,
            gate_config: None,
        }
    }
}

/// Whether a recorded per-repo `gate_status` represents success. Single
/// source of truth shared by the loop-abort check and the final-ok
/// rollup, so they can never diverge (the historical BUG#1 was exactly
/// such a divergence — only `failed_test` aborted/flipped, so
/// `failed_bench` shipped silently).
pub(crate) fn gate_succeeded(status: &str) -> bool {
    matches!(status, "succeeded" | "succeeded_dry_run")
}

#[derive(Debug, Clone)]
pub struct RepoGitState {
    pub sha: String,
    pub branch: String,
    pub dirty: bool,
}

#[derive(Debug, Clone)]
pub struct RepoReleaseRecord {
    pub repo: String,
    pub build_order_idx: usize,
    pub git: RepoGitState,
    pub gate_status: String,
    pub tests_passed: u32,
    pub tests_failed: u32,
    pub published_versions: Vec<(String, String)>,
    /// Urðr time-machine pins captured by the release run's capture phase
    /// (`release_lineage` cols 15/16). `None` when the projection wasn't
    /// captured (e.g. `--skip-snapshot`, or a build with no binary for dwarf).
    pub tantivy_snapshot_id: Option<String>,
    pub dwarf_snapshot_id: Option<String>,
}

#[derive(Debug, Clone)]
pub struct ReleaseReport {
    pub release_id: Uuid,
    pub workspace_name: String,
    /// Dep-graph snapshot pinned at release time. `Uuid::nil()` for
    /// pre-1.x rows that didn't record it.
    pub dep_graph_snapshot_id: Uuid,
    pub repos: Vec<RepoReleaseRecord>,
}

/// Run the pipeline. Returns a [`ReleaseReport`] AND persists the
/// same data into the warehouse before returning.
///
/// Stops at the first repo whose test gate fails — but every repo
/// processed up to (and including) the failure is recorded.
pub async fn run_pipeline(
    wh: &IcebergWarehouse,
    desc: &WorkspaceDescriptor,
    graph: &WorkspaceGraph,
    opts: &PipelineOptions,
) -> Result<ReleaseReport> {
    let release_id = Uuid::new_v4();
    let workspace_name = desc.workspace.name.clone();
    let order = graph.build_order().context("derive build order")?;

    // Release-op DAG emitter. Keyed by the run's release_id so every event row
    // groups under one run; `seq` is minted per-emit. Emits are best-effort +
    // non-fatal (see `ReleaseEventEmitter`) so a logging-row failure never
    // aborts the release. The whole run is bracketed by a `run/start`/`run/end`.
    let events = ReleaseEventEmitter::new(release_id.to_string());
    events
        .emit_async(
            wh, &workspace_name, "", "run", phase::START, status::RUNNING,
            &format!("order: {}", order.join(" → ")), None, None,
        )
        .await;

    let sources: BTreeMap<String, PathBuf> = crate::workspace::resolve::resolve_sources(desc)
        .context("resolve workspace sources")?;

    let mut records: Vec<RepoReleaseRecord> = Vec::new();

    use crate::release::progress::{now, ReleaseEvent};
    if let Some(p) = &opts.progress {
        p.emit(&ReleaseEvent::RunStart {
            ts: now(),
            run_id: release_id.to_string(),
            workspace: workspace_name.clone(),
        });
    }

    // Pre-flight: collect git state for every repo. If any is dirty
    // and `refuse_if_dirty` is set, abort before running any cargo —
    // the SHA on a dirty tree doesn't fully describe what would
    // ship, so the release would be unrollable.
    let mut pre: BTreeMap<String, RepoGitState> = BTreeMap::new();
    for repo in &order {
        let root = sources
            .get(repo)
            .ok_or_else(|| anyhow::anyhow!("no source path for repo `{repo}` in descriptor"))?;
        let git = read_git_state(root)
            .with_context(|| format!("read git state for {}", root.display()))?;
        pre.insert(repo.clone(), git);
    }
    if opts.refuse_if_dirty {
        let dirty: Vec<&String> = order.iter().filter(|r| pre[*r].dirty).collect();
        if !dirty.is_empty() {
            let detail = dirty
                .iter()
                .map(|r| format!("  - {r} (HEAD={})", &pre[*r].sha[..12]))
                .collect::<Vec<_>>()
                .join("\n");
            anyhow::bail!(
                "release refused: {} repo(s) have uncommitted changes:\n{detail}\n\
                 commit the work, or `git reset --hard <last-clean-sha>` to roll back, \
                 then re-run. (Set refuse_if_dirty=false to override.)",
                dirty.len()
            );
        }
    }

    for (idx, repo) in order.iter().enumerate() {
        let root = sources.get(repo).expect("checked above");
        let git = pre.remove(repo).expect("checked above");

        println!(
            "\n\u{2501}\u{2501}\u{2501} [{i}/{n}] {repo} \u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}",
            i = idx + 1,
            n = order.len(),
        );
        println!(
            "📍 sha={} branch={} dirty={}",
            &git.sha[..git.sha.len().min(12)],
            git.branch,
            if git.dirty { "yes" } else { "no" }
        );

        if let Some(p) = &opts.progress {
            p.emit(&ReleaseEvent::RepoStart {
                ts: now(),
                repo: repo.clone(),
                sha: git.sha.clone(),
            });
            p.emit(&ReleaseEvent::PhaseStart {
                ts: now(),
                repo: repo.clone(),
                phase: "test".to_string(),
            });
        }

        // Dep-graph producers this repo waited on (the DAG `depends_on` edges).
        let deps = depends_on_for(graph, repo);

        let phase_start = Instant::now();
        events
            .emit_async(
                wh, repo, repo, "test", phase::START, status::RUNNING, "",
                Some(deps.clone()), None,
            )
            .await;
        let (tests_passed, tests_failed, test_ok) =
            run_cargo_test(root, Some(repo), opts.progress.as_ref())
                .with_context(|| format!("cargo test for repo `{repo}`"))?;
        println!(
            "   \u{2192} test totals: {tests_passed} passed, {tests_failed} failed ({:.1}s) {}",
            phase_start.elapsed().as_secs_f32(),
            if test_ok { "\u{2713}" } else { "\u{2717} FAILED" }
        );
        events
            .emit_async(
                wh, repo, repo, "test", phase::END,
                if test_ok { status::OK } else { status::FAIL },
                &format!("{tests_passed} passed, {tests_failed} failed"),
                Some(deps.clone()),
                Some(phase_start.elapsed().as_millis() as i64),
            )
            .await;
        if let Some(p) = &opts.progress {
            p.emit(&ReleaseEvent::PhaseEnd {
                ts: now(),
                repo: repo.clone(),
                phase: "test".to_string(),
                ok: test_ok,
                duration_ms: phase_start.elapsed().as_millis() as u64,
            });
        }

        let bench_start = Instant::now();
        events
            .emit_async(
                wh, repo, repo, "bench", phase::START, status::RUNNING, "",
                Some(deps.clone()), None,
            )
            .await;
        // `bench_status` is one of: "skipped" | "ok" (passed) or a fatal
        // "failed_*" status. A fatal status both aborts the run and flips
        // final_ok (via `gate_succeeded`). Note: a run is persisted as a
        // future regression baseline ONLY if it passes every bench gate —
        // we must not let a red/regressed/errored run poison the history.
        let bench_status = match run_bench_example(root) {
            Ok(None) => {
                println!("\u{26A1} bench: skipped (no examples/nornir-bench.rs)");
                "skipped".to_string()
            }
            Ok(Some(run)) => {
                let n_res = run.results.len();
                let n_failed = run.tests.iter().filter(|t| !t.passed).count();
                println!(
                    "\u{26A1} bench ({:.1}s): {} result(s), {} test fail(s)",
                    bench_start.elapsed().as_secs_f32(),
                    n_res,
                    n_failed,
                );
                for r in &run.results {
                    let mut kv: Vec<String> = r
                        .metrics
                        .iter()
                        .filter_map(|(k, v)| v.as_f64().map(|f| format!("{k}={f:.2}")))
                        .collect();
                    kv.sort();
                    println!("   \u{2022} {:30} {}", r.name, kv.join(" "));
                }
                if n_failed > 0 {
                    for t in run.tests.iter().filter(|t| !t.passed) {
                        println!("   \u{2717} test {}", t.name);
                    }
                    // Bench unit tests failed → fatal. Do NOT persist (a
                    // red run must never become a regression baseline).
                    "failed_bench".to_string()
                } else {
                    // Regression gate (if configured) runs BEFORE the
                    // append, so the candidate can't be compared against
                    // itself (which would make every drop 0%).
                    let regression = match &opts.gate_config {
                        Some(gc) if gc.repo_gates.get(repo).map(|g| g.no_regression).unwrap_or(false) => {
                            let g = gc.repo_gates.get(repo).expect("checked");
                            let pct = if g.max_regression_pct > 0.0 { g.max_regression_pct } else { 10.0 };
                            let hist = wh
                                .query_bench_runs_async(&crate::warehouse::BenchFilter {
                                    repo: Some(repo.clone()),
                                    machine: Some(run.machine.clone()),
                                    limit: None,
                                })
                                .await
                                .with_context(|| format!("read bench history for `{repo}`"))?;
                            crate::release::gate::no_regression_against(&run, &hist, pct).err()
                        }
                        _ => None,
                    };
                    if let Some(e) = regression {
                        println!("   \u{2717} regression gate: {e:#}");
                        "failed_regression".to_string()
                    } else {
                        wh.append_bench_run_async(repo, &run)
                            .await
                            .with_context(|| format!("persist bench run for `{repo}`"))?;
                        println!("   \u{2713} persisted into bench_runs/bench_results");
                        "ok".to_string()
                    }
                }
            }
            Err(e) => {
                // Bench failed to even execute → fatal. The configured
                // release evidence could not be produced; correctness-first
                // means we must not ship.
                println!("\u{26A1} bench: \u{2717} error: {e:#}");
                "failed_bench_exec".to_string()
            }
        };

        let bench_ev_status = match bench_status.as_str() {
            "ok" => status::OK,
            "skipped" => status::WARN,
            _ => status::FAIL,
        };
        events
            .emit_async(
                wh, repo, repo, "bench", phase::END, bench_ev_status, &bench_status,
                Some(deps.clone()),
                Some(bench_start.elapsed().as_millis() as i64),
            )
            .await;

        // guard_intact gate: the recorded-vs-current tamper check. Only
        // evaluated for repos that opt in via `[gates].guard_intact` and
        // when a gate_config (workspace root + manifest) is available.
        let guard_failure: Option<String> = match &opts.gate_config {
            Some(gc) if gc.repo_gates.get(repo).map(|g| g.guard_intact).unwrap_or(false) => {
                match crate::guard::read_manifest(&gc.workspace_root) {
                    Ok(manifest) => crate::guard::intact(&gc.workspace_root, &manifest)
                        .err()
                        .map(|e| format!("{e:#}")),
                    Err(e) => Some(format!("guard manifest unavailable: {e:#}")),
                }
            }
            _ => None,
        };

        let bench_ok = matches!(bench_status.as_str(), "ok" | "skipped");

        let (gate_status, published) = if !test_ok {
            ("failed_test".to_string(), Vec::new())
        } else if !bench_ok {
            // bench_status already holds the specific fatal reason.
            (bench_status.clone(), Vec::new())
        } else if let Some(reason) = &guard_failure {
            println!("   \u{2717} guard_intact gate: {reason}");
            ("failed_guard_intact".to_string(), Vec::new())
        } else if opts.dry_run {
            ("succeeded_dry_run".to_string(), Vec::new())
        } else {
            // Real publish path is intentionally NOT wired here: the
            // per-repo publish_order TOML lives outside the workspace
            // descriptor (see release/publish.rs) and we don't want
            // to push to crates.io as a side-effect of a smoke-test.
            // When that's wired, switch this arm to call publish_all
            // and populate `published`.
            ("succeeded".to_string(), Vec::new())
        };

        let repo_ok = gate_succeeded(&gate_status);

        // Gate outcome event: a single `gate/end` row carrying the per-repo
        // `gate_status` (the same string `release_lineage` records), so the DAG
        // shows exactly where a partial release stalled. `start`+`end` collapse
        // to one boundary here — the gate rollup is instantaneous.
        events
            .emit_async(
                wh, repo, repo, "gate", phase::END,
                if repo_ok { status::OK } else { status::FAIL },
                &gate_status, Some(deps.clone()), None,
            )
            .await;

        // Historize the knowledge map (syn symbols + call edges + git-heat) for
        // this repo at its release SHA — so a release captures the code-fact
        // projection, not just bench/lineage. Best-effort + non-fatal: a capture
        // hiccup must never fail an otherwise-green release. (Index/dwarf/vector
        // capture are tracked separately — plan.md §"Fat-client completion".)
        if repo_ok && !opts.dry_run {
            let root_owned = root.clone();
            let repo_owned = repo.clone();
            match tokio::task::spawn_blocking(move || {
                crate::knowledge::scan_all(&root_owned, &repo_owned)
            })
            .await
            {
                Ok(Ok(scan)) => {
                    let s = wh.append_symbol_scan_async(&scan.symbols).await;
                    let g = wh.append_git_heat_scan_async(&scan.git).await;
                    if s.is_ok() && g.is_ok() {
                        println!("   \u{2713} historized knowledge map (symbols + git-heat)");
                    } else {
                        eprintln!(
                            "   \u{26A0} knowledge capture partial for `{repo}`: {:?} {:?}",
                            s.err(),
                            g.err()
                        );
                    }
                }
                Ok(Err(e)) => eprintln!("   \u{26A0} knowledge scan skipped for `{repo}`: {e:#}"),
                Err(e) => eprintln!("   \u{26A0} knowledge scan task failed for `{repo}`: {e}"),
            }
        }

        records.push(RepoReleaseRecord {
            repo: repo.clone(),
            build_order_idx: idx,
            git,
            gate_status: gate_status.clone(),
            tests_passed,
            tests_failed,
            published_versions: published,
            // The library orchestrator historizes the knowledge map but not the
            // index/dwarf; the rich CLI path (`run_release_run`) pins those.
            tantivy_snapshot_id: None,
            dwarf_snapshot_id: None,
        });

        if let Some(p) = &opts.progress {
            p.emit(&ReleaseEvent::RepoEnd {
                ts: now(),
                repo: repo.clone(),
                ok: repo_ok,
            });
        }

        // Any non-success status aborts the remaining build order: a
        // dependency failed its gate, so dependents can't be validated
        // against it. (Historically only "failed_test" broke here —
        // BUG#1 — letting bench failures ship silently.)
        if !repo_ok {
            break;
        }
    }

    persist_lineage(wh, release_id, &workspace_name, &opts.dep_graph_snapshot_id, &records, opts.dry_run)
        .await?;

    let final_ok = records.iter().all(|r| gate_succeeded(&r.gate_status));
    events
        .emit_async(
            wh, &workspace_name, "", "run", phase::END,
            if final_ok { status::OK } else { status::FAIL },
            &format!("{} repo(s) processed", records.len()),
            None, None,
        )
        .await;
    if let Some(p) = &opts.progress {
        p.emit(&ReleaseEvent::RunEnd {
            ts: now(),
            run_id: release_id.to_string(),
            ok: final_ok,
        });
    }

    Ok(ReleaseReport {
        release_id,
        workspace_name,
        dep_graph_snapshot_id: opts.dep_graph_snapshot_id,
        repos: records,
    })
}

pub fn read_git_state(root: &Path) -> Result<RepoGitState> {
    let repo = gix::open(root).with_context(|| format!("gix::open {}", root.display()))?;
    let head = repo.head().context("read HEAD")?;
    let sha = head
        .id()
        .ok_or_else(|| anyhow::anyhow!("HEAD has no commit (unborn?)"))?
        .to_string();
    let branch = match repo.head_name()? {
        Some(name) => name.shorten().to_string(),
        None => "(detached)".to_string(),
    };
    let dirty = is_dirty(&repo);
    Ok(RepoGitState { sha, branch, dirty })
}

/// Cheap dirty check using `gix status`. We swallow errors as
/// `false` — provenance shouldn't fail on an exotic repo layout.
fn is_dirty(repo: &gix::Repository) -> bool {
    let Ok(platform) = repo.status(gix::progress::Discard) else {
        return false;
    };
    let Ok(iter) = platform.into_index_worktree_iter(Vec::<gix::bstr::BString>::new()) else {
        return false;
    };
    iter.filter_map(Result::ok).next().is_some()
}

/// Returns `(passed, failed, ok)` where `ok = failed == 0 && cargo
/// itself exited cleanly`. Parsing is line-oriented over stdout —
/// cargo's "test result: ok. N passed; M failed" summary.
///
/// `progress`, when set, receives structured events for every
/// `Running ...` binary and every per-test pass/fail. The live egui
/// pane and `nornir-server`'s `/release/progress` SSE both consume
/// this stream. Pass `None` for headless / unit-test calls.
pub fn run_cargo_test(
    root: &Path,
    repo_for_events: Option<&str>,
    progress: Option<&crate::release::progress::ProgressWriter>,
) -> Result<(u32, u32, bool)> {
    use crate::release::progress::{now, ReleaseEvent};

    let repo_label = repo_for_events.unwrap_or("").to_string();
    let emit = |ev: ReleaseEvent| {
        if let Some(p) = progress { p.emit(&ev); }
    };

    // Heavy tests (`#[ignore]`'d: container integration, large corpora, the
    // workspace×function viz matrix) run by DEFAULT now — they ARE the point of
    // a release gate, and skipping them silently was letting real regressions
    // through. Opt OUT with `NORNIR_SKIP_HEAVY=1` (the CLI `--skip-heavy` sets
    // it) for a fast local smoke. Back-compat: an explicit `NORNIR_HEAVY_TESTS=0`
    // is still honored as a skip.
    let skip_heavy = std::env::var("NORNIR_SKIP_HEAVY").map(|v| !v.is_empty() && v != "0").unwrap_or(false)
        || matches!(std::env::var("NORNIR_HEAVY_TESTS").as_deref(), Ok("0") | Ok(""));
    let heavy = !skip_heavy;
    let verbose = std::env::var("NORNIR_VERBOSE")
        .map(|v| !v.is_empty() && v != "0")
        .unwrap_or(false);
    // NOTE: deliberately *no* top-level `--quiet`. We want cargo's
    // `Running tests/foo.rs` lines on stderr and libtest's
    // `test result:` rollup on stdout. Warnings are filtered out
    // below so the operator sees signal, not noise.
    // Always serial across the workspace:
    //   `-j 1`            → one test *binary* at a time (otherwise cargo
    //                       runs different test binaries in parallel and
    //                       e.g. two Sonatype Nexus testcontainers fight
    //                       for RAM + ports).
    //   `--test-threads=1` → one test *function* at a time inside each
    //                       binary (kills the classic libtest race on
    //                       shared `/tmp` caches).
    // Extra cargo features for the test gate, e.g. `NORNIR_TEST_FEATURES=viz,server`
    // so nornir's OWN release compiles + runs the feature-gated viz matrix tests
    // (the workspace×function sweep). Cargo args go BEFORE `--`; harness args after.
    let feat = std::env::var("NORNIR_TEST_FEATURES").ok().filter(|s| !s.is_empty());
    let mut args: Vec<&str> = vec!["test", "--workspace", "--no-fail-fast", "-j", "1"];
    if let Some(f) = &feat {
        args.push("--features");
        args.push(f);
    }
    args.push("--");
    args.push("--test-threads=1");
    if heavy {
        args.push("--include-ignored");
    }
    println!("🧪 cargo {}", args.join(" "));

    let mut child = Command::new("cargo")
        .args(&args)
        .current_dir(root)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .context("spawn cargo test")?;

    let stdout = child.stdout.take().expect("piped stdout");
    let stderr = child.stderr.take().expect("piped stderr");

    // Shared queue: stderr pushes binary names in cargo's emission
    // order; stdout pops one each time libtest announces a new
    // `running N tests` block. This is the only correct way to
    // pair the two streams without parsing cargo JSON output.
    let pending_bins: Arc<Mutex<VecDeque<String>>> = Arc::new(Mutex::new(VecDeque::new()));
    let bins_for_stderr = Arc::clone(&pending_bins);
    let progress_for_stderr = progress.cloned();
    let repo_for_stderr = repo_label.clone();

    let stderr_handle = std::thread::spawn(move || {
        for line in BufReader::new(stderr).lines().map_while(Result::ok) {
            let trimmed = line.trim_start();
            if let Some(what) = trimmed.strip_prefix("Running ") {
                let what = what.trim().to_string();
                println!("   📂 {what}");
                if let Some(p) = &progress_for_stderr {
                    p.emit(&ReleaseEvent::BinaryStart {
                        ts: now(),
                        repo: repo_for_stderr.clone(),
                        binary: what.clone(),
                    });
                }
                bins_for_stderr.lock().unwrap().push_back(what);
            } else if verbose && trimmed.starts_with("Compiling ") {
                println!("   🔨 {}", trimmed.trim_start_matches("Compiling "));
            } else if trimmed.starts_with("error") {
                // Compiler / cargo errors — always surface.
                println!("   {line}");
            }
            // Warnings are intentionally dropped.
        }
    });

    let mut passed = 0u32;
    let mut failed = 0u32;
    let mut failure_names: Vec<String> = Vec::new();
    let mut current_bin: String = String::new();
    // Counters scoped to the currently-running binary, so we can
    // emit BinaryDone with the right rollup before the next block.
    let mut bin_passed: u32 = 0;
    let mut bin_failed: u32 = 0;

    for line in BufReader::new(stdout).lines().map_while(Result::ok) {
        let trimmed = line.trim_start();
        if let Some(rest) = trimmed.strip_prefix("running ") {
            // `running N tests` → start of a new binary block.
            if let Some((n, _)) = rest.split_once(' ') {
                if let Ok(n) = n.parse::<u32>() {
                    current_bin = pending_bins
                        .lock()
                        .unwrap()
                        .pop_front()
                        .unwrap_or_default();
                    let bin_short = current_bin
                        .split_whitespace()
                        .next()
                        .unwrap_or(&current_bin);
                    println!("      \u{2192} {n} test(s)  {bin_short}");
                    bin_passed = 0;
                    bin_failed = 0;
                }
            }
        } else if let Some(name) = trimmed.strip_prefix("test ") {
            if let Some(nm) = name.strip_suffix(" ... FAILED") {
                println!("      \u{2717} {nm}");
                failure_names.push(format!(
                    "{}::{nm}",
                    current_bin
                        .split_whitespace()
                        .next()
                        .unwrap_or("?")
                ));
                bin_failed += 1;
                emit(ReleaseEvent::TestFail {
                    ts: now(),
                    repo: repo_label.clone(),
                    binary: current_bin.clone(),
                    name: nm.to_string(),
                });
            } else if let Some(nm) = name.strip_suffix(" ... ok") {
                if verbose {
                    println!("      \u{2713} {nm}");
                }
                bin_passed += 1;
                emit(ReleaseEvent::TestPass {
                    ts: now(),
                    repo: repo_label.clone(),
                    binary: current_bin.clone(),
                    name: nm.to_string(),
                });
            }
        } else if trimmed.starts_with("test result:") {
            for chunk in trimmed.split(';') {
                let chunk = chunk.trim();
                if let Some(n) = chunk.strip_suffix(" passed") {
                    if let Some(n) = n.split_whitespace().last() {
                        passed += n.parse::<u32>().unwrap_or(0);
                    }
                } else if let Some(n) = chunk.strip_suffix(" failed") {
                    if let Some(n) = n.split_whitespace().last() {
                        failed += n.parse::<u32>().unwrap_or(0);
                    }
                }
            }
            println!("      \u{2261} {trimmed}");
            emit(ReleaseEvent::BinaryDone {
                ts: now(),
                repo: repo_label.clone(),
                binary: current_bin.clone(),
                passed: bin_passed,
                failed: bin_failed,
            });
        }
    }

    let status = child.wait().context("wait cargo test")?;
    let _ = stderr_handle.join();

    // If cargo aborted before printing the per-binary `test result:`
    // rollup, the libtest counters above can read 0/0 while we still
    // captured concrete failure names from `... FAILED` lines. Promote
    // the failure-name list to the failed counter in that case so the
    // operator-facing summary doesn't lie.
    if failure_names.len() as u32 > failed {
        failed = failure_names.len() as u32;
    }

    if !failure_names.is_empty() {
        println!("   ✗ {} failure(s):", failure_names.len());
        for fn_ in &failure_names {
            println!("      • {fn_}");
        }
    }

    Ok((passed, failed, status.success() && failed == 0))
}

/// Append one `release_lineage` row per record, pinning the Urðr snapshot ids
/// each record carries. Used by both the library orchestrator (`run_pipeline`)
/// and the rich CLI path (`bin/nornir.rs::run_release_run`) so the table is
/// written the same way regardless of which drives the release.
pub async fn persist_lineage(
    wh: &IcebergWarehouse,
    release_id: Uuid,
    workspace_name: &str,
    dep_graph_snapshot_id: &Uuid,
    records: &[RepoReleaseRecord],
    dry_run: bool,
) -> Result<()> {
    if records.is_empty() {
        return Ok(());
    }
    let table = wh
        .catalog()
        .load_table(&wh.table_ident(wh_iceberg::TABLE_RELEASE_LINEAGE))
        .await?;
    let arrow_schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);

    let ts = Utc::now().timestamp_micros();
    let rid = release_id.to_string();
    let dsid = dep_graph_snapshot_id.to_string();

    let n = records.len();
    let mut release_ids = Vec::with_capacity(n);
    let mut ws_names = Vec::with_capacity(n);
    let mut ts_vals = Vec::with_capacity(n);
    let mut dgsids = Vec::with_capacity(n);
    let mut order_idx = Vec::with_capacity(n);
    let mut repos = Vec::with_capacity(n);
    let mut shas = Vec::with_capacity(n);
    let mut branches = Vec::with_capacity(n);
    let mut dirties = Vec::with_capacity(n);
    let mut statuses = Vec::with_capacity(n);
    let mut tpass = Vec::with_capacity(n);
    let mut tfail = Vec::with_capacity(n);
    let mut dryruns = Vec::with_capacity(n);
    let mut pubs = Vec::with_capacity(n);

    for r in records {
        release_ids.push(rid.clone());
        ws_names.push(workspace_name.to_string());
        ts_vals.push(ts);
        dgsids.push(dsid.clone());
        order_idx.push(r.build_order_idx as i32);
        repos.push(r.repo.clone());
        shas.push(r.git.sha.clone());
        branches.push(r.git.branch.clone());
        dirties.push(r.git.dirty);
        statuses.push(r.gate_status.clone());
        tpass.push(r.tests_passed as i32);
        tfail.push(r.tests_failed as i32);
        dryruns.push(dry_run);
        pubs.push(
            r.published_versions
                .iter()
                .map(|(c, v)| format!("{c}@{v}"))
                .collect::<Vec<_>>()
                .join(","),
        );
    }

    let cols: Vec<Arc<dyn Array>> = vec![
        Arc::new(StringArray::from(release_ids)),
        Arc::new(StringArray::from(ws_names)),
        Arc::new(TimestampMicrosecondArray::from(ts_vals).with_timezone("+00:00")),
        Arc::new(StringArray::from(dgsids)),
        Arc::new(Int32Array::from(order_idx)),
        Arc::new(StringArray::from(repos)),
        Arc::new(StringArray::from(shas)),
        Arc::new(StringArray::from(branches)),
        Arc::new(BooleanArray::from(dirties)),
        Arc::new(StringArray::from(statuses)),
        Arc::new(Int32Array::from(tpass)),
        Arc::new(Int32Array::from(tfail)),
        Arc::new(BooleanArray::from(dryruns)),
        Arc::new(StringArray::from(pubs)),
        // Time-machine pin columns (15-17). `tantivy_snapshot_id` +
        // `dwarf_snapshot_id` come from each record's capture phase
        // (`run_release_run`); the legacy `run_pipeline` leaves them None.
        // `gimli_snapshot_id` (17) is not yet produced.
        Arc::new(StringArray::from(
            records.iter().map(|r| r.tantivy_snapshot_id.clone()).collect::<Vec<_>>(),
        )),
        Arc::new(StringArray::from(
            records.iter().map(|r| r.dwarf_snapshot_id.clone()).collect::<Vec<_>>(),
        )),
        Arc::new(StringArray::from(vec![None as Option<String>; n])),
        // Cargo-pipeline pins (fields 18-21) — also null from the
        // legacy orchestrator; populated by gates/wirings as they land.
        Arc::new(StringArray::from(vec![None as Option<String>; n])),  // git_tag
        Arc::new(BooleanArray::from(vec![None as Option<bool>; n])),   // git_tag_matches_published
        Arc::new(StringArray::from(vec![None as Option<String>; n])),  // rustdoc_json_snapshot_id
        Arc::new(Int32Array::from(vec![None as Option<i32>; n])),      // patch_lines_stripped
    ];
    let batch = RecordBatch::try_new(arrow_schema, cols)?;
    wh_iceberg::append_batch(wh.catalog(), table, batch).await?;
    Ok(())
}

/// Read every recorded release for `workspace_name`, grouped by
/// release_id, sorted oldest-first. Optional `limit` keeps only the
/// most recent N. An **empty** `workspace_name` matches every workspace
/// (used by the cross-workspace `regression::trace`).
pub async fn query_release_history(
    wh: &IcebergWarehouse,
    workspace_name: &str,
    limit: Option<usize>,
) -> Result<Vec<ReleaseReport>> {
    let table = wh
        .catalog()
        .load_table(&wh.table_ident(wh_iceberg::TABLE_RELEASE_LINEAGE))
        .await?;
    let scan = table.scan().build()?;
    let stream = scan.to_arrow().await?;
    let batches: Vec<RecordBatch> = stream.try_collect().await?;

    let mut by_release: BTreeMap<(Uuid, i64), (String, Uuid, Vec<RepoReleaseRecord>)> = BTreeMap::new();

    for batch in &batches {
        let rid = down::<StringArray>(batch, 0)?;
        let wss = down::<StringArray>(batch, 1)?;
        let tss = down::<TimestampMicrosecondArray>(batch, 2)?;
        let dgs = down::<StringArray>(batch, 3)?;
        let oid = down::<Int32Array>(batch, 4)?;
        let rep = down::<StringArray>(batch, 5)?;
        let sha = down::<StringArray>(batch, 6)?;
        let br = down::<StringArray>(batch, 7)?;
        let dty = down::<BooleanArray>(batch, 8)?;
        let st = down::<StringArray>(batch, 9)?;
        let tp = down::<Int32Array>(batch, 10)?;
        let tf = down::<Int32Array>(batch, 11)?;
        let _dr = down::<BooleanArray>(batch, 12)?;
        let pub_csv = down::<StringArray>(batch, 13)?;
        // Time-machine pins (arrow cols 14/15 = field ids 15/16).
        let tantivy = down::<StringArray>(batch, 14)?;
        let dwarf = down::<StringArray>(batch, 15)?;

        for i in 0..batch.num_rows() {
            if !workspace_name.is_empty() && wss.value(i) != workspace_name {
                continue;
            }
            let release_uuid = Uuid::parse_str(rid.value(i))?;
            let snap_uuid = Uuid::parse_str(dgs.value(i)).unwrap_or(Uuid::nil());
            let key = (release_uuid, tss.value(i));
            let entry = by_release
                .entry(key)
                .or_insert_with(|| (wss.value(i).to_string(), snap_uuid, Vec::new()));
            let published = if pub_csv.value(i).is_empty() {
                Vec::new()
            } else {
                pub_csv
                    .value(i)
                    .split(',')
                    .filter_map(|s| {
                        let (c, v) = s.split_once('@')?;
                        Some((c.to_string(), v.to_string()))
                    })
                    .collect()
            };
            entry.2.push(RepoReleaseRecord {
                repo: rep.value(i).to_string(),
                build_order_idx: oid.value(i) as usize,
                git: RepoGitState {
                    sha: sha.value(i).to_string(),
                    branch: br.value(i).to_string(),
                    dirty: dty.value(i),
                },
                gate_status: st.value(i).to_string(),
                tests_passed: tp.value(i) as u32,
                tests_failed: tf.value(i) as u32,
                published_versions: published,
                tantivy_snapshot_id: (!tantivy.is_null(i)).then(|| tantivy.value(i).to_string()),
                dwarf_snapshot_id: (!dwarf.is_null(i)).then(|| dwarf.value(i).to_string()),
            });
        }
    }

    let mut out: Vec<ReleaseReport> = by_release
        .into_iter()
        .map(|((release_id, _ts), (ws, snap, mut repos))| {
            repos.sort_by_key(|r| r.build_order_idx);
            ReleaseReport {
                release_id,
                workspace_name: ws,
                dep_graph_snapshot_id: snap,
                repos,
            }
        })
        .collect();
    if let Some(n) = limit {
        let drop_n = out.len().saturating_sub(n);
        out.drain(..drop_n);
    }
    Ok(out)
}

fn down<T: 'static>(batch: &RecordBatch, idx: usize) -> Result<&T> {
    batch
        .column(idx)
        .as_any()
        .downcast_ref::<T>()
        .ok_or_else(|| anyhow::anyhow!("column {idx} downcast failed"))
}

/// Invokes the repo's bench runner if present and parses the JSON
/// line it prints into a [`crate::bench::BenchRun`].
///
/// Two supported layouts (checked in order):
///
/// 1. **Single-package repo** — `examples/nornir-bench.rs` at the
///    repo root. Run with `cargo run --release --example nornir-bench`.
/// 2. **Virtual workspace** (znippy, holger, …) — convention is
///    `xtask/examples/nornir-bench.rs`. Run with
///    `cargo run -p xtask --release --example nornir-bench`.
///
/// Returns:
/// - `Ok(None)` if no runner is wired (repo hasn't opted in).
/// - `Ok(Some(run))` on a successful run.
/// - `Err(_)` if a runner exists but failed or its stdout didn't
///   parse — real bugs we want surfaced loudly.
/// Spawn the repo's `examples/nornir-bench.rs` (or `xtask/examples/nornir-bench.rs`)
/// as a subprocess and parse the last non-empty stdout line as a
/// [`crate::bench::BenchRun`]. Returns `Ok(None)` when the repo has
/// no nornir-bench example wired up.
pub fn run_bench_example(root: &Path) -> Result<Option<crate::bench::BenchRun>> {
    let s = |v: Vec<&str>| -> Vec<String> { v.into_iter().map(String::from).collect() };
    let (mut args, run_dir): (Vec<String>, PathBuf) =
        if root.join("examples/nornir-bench.rs").is_file() {
            (s(vec!["run", "--release", "--example", "nornir-bench", "--quiet"]), root.to_path_buf())
        } else if root.join("xtask/examples/nornir-bench.rs").is_file() {
            (
                s(vec!["run", "-p", "xtask", "--release", "--example", "nornir-bench", "--quiet"]),
                root.to_path_buf(),
            )
        } else if root.join("bench/examples/nornir-bench.rs").is_file() {
            // Detached bench crate (its own [workspace]) under <repo>/bench. Keeps
            // the example's dev-deps (nornir, bench scenarios) out of the published
            // crate without needing an xtask workspace member — run cargo *in* bench/.
            (s(vec!["run", "--release", "--example", "nornir-bench", "--quiet"]), root.join("bench"))
        } else {
            return Ok(None);
        };
    // Optional cargo features for the bench example (e.g. `s3` to enable the
    // RustFS/Nessie/Polaris container targets): `NORNIR_BENCH_FEATURES=s3`.
    if let Ok(feat) = std::env::var("NORNIR_BENCH_FEATURES") {
        if !feat.trim().is_empty() {
            args.push("--features".into());
            args.push(feat);
        }
    }
    // Stream the bench's stderr LIVE to our own stderr instead of buffering it.
    // The bench binary emits its `bench: <name> …` / `[BENCH]` progress lines on
    // stderr (see `bench::progress`); `.output()` would swallow them until the
    // child exits, so a multi-minute bench looked hung. Inheriting stderr lets
    // the operator watch which bench is running (and its `cores-busy/N`) live,
    // while stdout stays piped because we must parse its final JSON line.
    let mut child = Command::new("cargo")
        .args(&args)
        .current_dir(&run_dir)
        .stdout(Stdio::piped())
        .stderr(Stdio::inherit())
        .spawn()
        .context("spawn cargo run --example nornir-bench")?;
    let mut stdout_buf = String::new();
    {
        use std::io::Read;
        let mut so = child.stdout.take().expect("piped stdout");
        so.read_to_string(&mut stdout_buf).context("read nornir-bench stdout")?;
    }
    let status = child.wait().context("wait for nornir-bench")?;
    if !status.success() {
        anyhow::bail!(
            "cargo {} exited {} for {} (stderr was streamed above)",
            args.join(" "),
            status,
            root.display(),
        );
    }
    let stdout = stdout_buf;
    let line = stdout
        .lines()
        .filter(|l| !l.trim().is_empty())
        .last()
        .ok_or_else(|| anyhow::anyhow!("nornir-bench produced no stdout"))?;
    let run: crate::bench::BenchRun = serde_json::from_str(line)
        .with_context(|| format!("parse nornir-bench stdout as BenchRun:\n{line}"))?;
    Ok(Some(run))
}

#[cfg(test)]
mod gate_status_tests {
    use super::gate_succeeded;

    #[test]
    fn only_success_statuses_count_as_succeeded() {
        assert!(gate_succeeded("succeeded"));
        assert!(gate_succeeded("succeeded_dry_run"));
        // Every fatal status must NOT count as success — this is the
        // single predicate that drives both loop-abort and final_ok, so
        // a regression here would resurrect BUG#1 (silent bench ship).
        for bad in [
            "failed_test",
            "failed_bench",
            "failed_bench_exec",
            "failed_regression",
            "failed_guard_intact",
            "skipped",
            "",
        ] {
            assert!(!gate_succeeded(bad), "`{bad}` must not be a success status");
        }
    }
}