nornir 0.5.2

//! `nornir release stage` — the brave-but-safe tier: rehearse the whole release
//! cascade against the embedded holger `/sparring` registry before crates.io.
//!
//! This module owns the **plan** (pure, testable): given the cross-repo graph, the
//! topo publish order and, for each repo, the direct dependents to build-verify
//! from `/sparring` once it's published. Execution (cut a release branch →
//! `cargo publish --registry sparring` in order → `cargo build` each dependent
//! against `/sparring` → `seal`) drives this plan; `--dry-run` just prints it.
//!
//! Promote (`release promote`) replays the same order against crates.io once the
//! sparring run is green.

use serde::Serialize;

use crate::release::doctor::{crate_publish_order, RepoGraph};

#[derive(Debug, Clone, Serialize)]
pub struct StageStep {
    /// Repo to package + publish into `/sparring` at this position.
    pub repo: String,
    /// In-constellation repos this repo directly depends on. The cascade GATES on
    /// these: `repo` is only attempted once every dependency that precedes it in the
    /// order has published (or was already staged). A dependency that failed DEFERS
    /// `repo` and reports it blocked, rather than attempting-and-failing it.
    #[serde(default)]
    pub deps: Vec<String>,
    /// Direct dependents to `cargo build` against `/sparring` after publishing
    /// `repo` — proves the just-published crates actually resolve from a registry.
    pub verify: Vec<String>,
}

#[derive(Debug, Clone, Serialize)]
pub struct StagePlan {
    /// Sparse registry the cascade publishes into, e.g.
    /// `sparse+http://127.0.0.1:18464/sparring/index/`.
    pub registry: String,
    /// Ephemeral release branch the version bumps + cascade live on.
    pub branch: String,
    /// Publish steps in dependency order (dependencies first).
    pub order: Vec<StageStep>,
    /// Repos left unordered by a dependency cycle (must be empty to execute).
    pub cycle: Vec<String>,
}

impl StagePlan {
    /// A plan is executable only when the graph is a clean DAG.
    pub fn is_executable(&self) -> bool {
        self.cycle.is_empty() && !self.order.is_empty()
    }
}

/// Repos that directly depend on `repo` (one hop): X depends on `repo` when X
/// declares a dependency on a crate `repo` produces.
fn direct_dependents(graphs: &[RepoGraph], repo: &str) -> Vec<String> {
    let Some(target) = graphs.iter().find(|g| g.repo == repo) else {
        return vec![];
    };
    let mut out: Vec<String> = graphs
        .iter()
        .filter(|g| g.repo != repo && g.deps.iter().any(|d| target.produces.contains(d)))
        .map(|g| g.repo.clone())
        .collect();
    out.sort();
    out
}

/// Repos `repo` directly depends on (one hop) — the inverse of [`direct_dependents`]:
/// `repo` depends on X when it declares a dependency on a crate X produces. These are
/// the in-constellation deps the cascade must see published before it may attempt `repo`.
fn direct_dependencies(graphs: &[RepoGraph], repo: &str) -> Vec<String> {
    let Some(me) = graphs.iter().find(|g| g.repo == repo) else {
        return vec![];
    };
    let mut out: Vec<String> = graphs
        .iter()
        .filter(|g| g.repo != repo && me.deps.iter().any(|d| g.produces.contains(d)))
        .map(|g| g.repo.clone())
        .collect();
    out.sort();
    out
}

/// Build the stage plan: topo publish order + per-repo verify (direct dependents).
pub fn plan_stage(graphs: &[RepoGraph], registry: &str, branch: &str) -> StagePlan {
    // Order at CRATE granularity, not repo: cargo publishes crates, and the
    // constellation's repo-level graph has SCCs that are clean crate-level DAGs
    // (mutual repo deps carried by DIFFERENT, feature-gated crates). The naive
    // repo-level `publish_order` flagged those as a false cycle; `crate_publish_order`
    // (the same order the doctor trusts) orders the crate DAG and only reports a
    // REAL crate-level cycle as blocking.
    let topo = crate_publish_order(graphs);
    let order = topo
        .order
        .iter()
        .map(|repo| StageStep {
            repo: repo.clone(),
            deps: direct_dependencies(graphs, repo),
            verify: direct_dependents(graphs, repo),
        })
        .collect();
    StagePlan { registry: registry.to_string(), branch: branch.to_string(), order, cycle: topo.cycle }
}

/// Human-readable plan (the dry-run output).
pub fn format_plan(p: &StagePlan) -> String {
    let mut s = String::new();
    s.push_str("nornir release stage — plan (dry-run)\n\n");
    s.push_str(&format!("  registry: {}\n", p.registry));
    s.push_str(&format!("  branch:   {}\n\n", p.branch));
    if !p.cycle.is_empty() {
        s.push_str(&format!(
            "  ⚠ dependency cycle — NOT executable until resolved: {}\n\n",
            p.cycle.join(", ")
        ));
    }
    s.push_str("  Publish order (dependencies first):\n");
    for (i, step) in p.order.iter().enumerate() {
        s.push_str(&format!("    {}. cargo publish {} → /sparring\n", i + 1, step.repo));
        if !step.verify.is_empty() {
            s.push_str(&format!("       verify (build from /sparring): {}\n", step.verify.join(", ")));
        }
    }
    if p.is_executable() {
        s.push_str("\n  → green plan; run with --execute to publish into /sparring, then promote.\n");
    }
    s
}

// ─────────────────────── execution (steps 1–3) ───────────────────────

use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::process::Command;
#[cfg(not(feature = "embed-holger"))]
use std::process::{Child, Stdio};

use anyhow::{Context, Result};

/// Cargo config the cascade runs under: defines the `sparring` registry to publish
/// INTO, and source-replaces crates.io with it so a dependent's build resolves the
/// just-published workspace crates from `/sparring` (everything else falls through
/// `/sparring`'s crates.io upstream).
pub fn cargo_config_toml(http_addr: &str) -> String {
    format!(
        r#"[registries.sparring]
index = "sparse+http://{http_addr}/sparring/index/"

[source.crates-io]
replace-with = "sparring-mirror"

[source.sparring-mirror]
registry = "sparse+http://{http_addr}/sparring/index/"

[net]
retry = 2
"#
    )
}

/// The embedded holger subprocess — killed on drop. Only the lean
/// `--no-default-features` build uses this; the default build embeds holger
/// in-process (`crate::holger_embed`), so it's compiled out when `embed-holger`
/// is on.
#[cfg(not(feature = "embed-holger"))]
struct HolgerProc(Child);
#[cfg(not(feature = "embed-holger"))]
impl Drop for HolgerProc {
    fn drop(&mut self) {
        let _ = self.0.kill();
        let _ = self.0.wait();
    }
}

#[cfg(not(feature = "embed-holger"))]
fn spawn_holger(holger_bin: &Path, data_dir: &Path, grpc: &str, http: &str) -> Result<HolgerProc> {
    let child = Command::new(holger_bin)
        .arg("dev-pair")
        .arg("--data")
        .arg(data_dir)
        .args(["--grpc", grpc, "--http", http])
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .spawn()
        .with_context(|| format!("spawn holger dev-pair via {}", holger_bin.display()))?;
    Ok(HolgerProc(child))
}

/// Poll the HTTP gateway until it serves (or give up after ~15s).
#[cfg(not(feature = "embed-holger"))]
fn wait_ready(http_addr: &str) -> Result<()> {
    let url = format!("http://{http_addr}/cache/index/config.json");
    for _ in 0..60 {
        if ureq::get(&url)
            .timeout(std::time::Duration::from_millis(500))
            .call()
            .is_ok()
        {
            return Ok(());
        }
        std::thread::sleep(std::time::Duration::from_millis(250));
    }
    anyhow::bail!("holger dev-pair never became ready at {url}")
}

#[derive(Debug, Default)]
pub struct StageOutcome {
    /// Repos whose crates published into /sparring.
    pub published: Vec<String>,
    /// Repos already present in /sparring at this version — a benign skip (the crate
    /// is staged and still build-verifies). NOT an error: re-running a rehearsal
    /// without a version bump lands here, exactly as crates.io would reject a re-publish.
    pub already_staged: Vec<String>,
    /// Verify targets that built against /sparring.
    pub verified: Vec<String>,
    /// Non-fatal per-step failures — a repo that itself failed to publish (repo: reason).
    pub errors: Vec<String>,
    /// Repos DEFERRED and never attempted because an in-constellation dependency
    /// failed (or is itself blocked). Reported distinctly from `errors`: these did not
    /// fail on their own merits — an upstream did (repo: blocked by failed dependency X).
    pub blocked: Vec<String>,
}

/// RAII cleanup for the ephemeral release branches the rehearsal cuts. Records each
/// repo's branch BEFORE `checkout -B <release>`, and on Drop — success, `?`-early-return,
/// OR panic — switches back to the original branch and deletes the staging branch, so a
/// rehearsal leaves NO trace.
///
/// CRITICAL: it must NEVER blanket-discard the working tree (`git checkout -- .`) — that
/// would eat the user's *uncommitted* edits (it once ate this very file's). The staging
/// branch is cut from the original with no commits, so switching back carries any
/// uncommitted changes cleanly; we then restore only the generated `Cargo.lock` (the one
/// file `cargo publish` churns), leaving every other change untouched.
pub(crate) struct BranchGuard {
    /// `(repo_path, original_branch)` captured pre-cut, restored newest-first on Drop.
    restores: Vec<(PathBuf, String)>,
    /// The ephemeral branch to delete after restoring (e.g. `release/staging`).
    branch: String,
}

impl BranchGuard {
    pub(crate) fn new(branch: &str) -> Self {
        BranchGuard { restores: Vec::new(), branch: branch.to_string() }
    }

    /// Capture `path`'s current branch so it can be restored later. Skips capture when
    /// the repo is already on the staging branch or HEAD can't be read (best-effort).
    pub(crate) fn capture(&mut self, path: &Path) {
        if let Ok(o) = Command::new("git")
            .arg("-C")
            .arg(path)
            .args(["rev-parse", "--abbrev-ref", "HEAD"])
            .output()
        {
            let prev = String::from_utf8_lossy(&o.stdout).trim().to_string();
            if !prev.is_empty() && prev != self.branch {
                self.restores.push((path.to_path_buf(), prev));
            }
        }
    }
}

impl Drop for BranchGuard {
    fn drop(&mut self) {
        for (path, prev) in self.restores.iter().rev() {
            // Switch back — uncommitted changes (the user's work + the publish's
            // Cargo.lock churn) carry across cleanly since the staging branch was cut
            // from here with no commits. Then drop the ephemeral branch and restore ONLY
            // Cargo.lock (the publish artifact). Never `checkout -- .`.
            let _ = Command::new("git").arg("-C").arg(path).args(["checkout", prev]).status();
            let _ = Command::new("git")
                .arg("-C")
                .arg(path)
                .args(["branch", "-D", &self.branch])
                .status();
            // `.output()` swallows the harmless "pathspec 'Cargo.lock' did not match"
            // stderr when the repo tracks no lockfile (e.g. a library workspace).
            let _ = Command::new("git")
                .arg("-C")
                .arg(path)
                .args(["checkout", "--", "Cargo.lock"])
                .output();
            eprintln!("stage: restored {} → {prev} (release branch cleaned)", path.display());
        }
    }
}

/// The result of handing one repo to the publish step — the seam the cascade drives.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum PublishAttempt {
    /// The repo's crates went into the registry on this attempt.
    Published,
    /// The version is already present — a benign idempotent skip. Counts as Done: it
    /// still satisfies dependents' gate (exactly as crates.io rejects a re-publish).
    AlreadyStaged,
    /// This attempt failed; the payload is the reason. Retryable until `MAX_ATTEMPTS`.
    Failed(String),
}

/// Bound on per-repo publish attempts: a transiently-failing repo is retried on a
/// later pass, but one that keeps failing converges to `errors` instead of spinning.
const MAX_ATTEMPTS: u32 = 2;

/// Per-repo state during the wave cascade.
#[derive(Clone, Copy, PartialEq, Eq)]
enum CascadeState {
    /// Unresolved — still waiting on a dep, or has retries left.
    Pending,
    /// Published or already-staged: its dependents' gate on it is satisfied.
    Done,
    /// Attempted `MAX_ATTEMPTS` times and still failing (its OWN failure).
    Failed,
    /// A dependency terminally failed (or is itself blocked): never attempted.
    Blocked,
}

/// DEPENDENCY-AWARE PUBLISH CASCADE — the wave sequencer that replaces the old linear
/// fail-forward walk. A repo is handed to `publish` only once every in-constellation
/// dependency that PRECEDES it in `order` has reached [`CascadeState::Done`] (published
/// or already-staged). If a dependency terminally fails, its dependents are DEFERRED
/// and reported blocked — never attempted-and-failed. A repo that fails transiently is
/// retried on a later pass, up to `MAX_ATTEMPTS`; one that keeps failing converges to
/// `out.errors`. Idempotency (already-staged ⇒ Done) is preserved.
///
/// Gating on deps that appear EARLIER in `order` (not on all deps) is deliberate: the
/// order already resolves repo-level SCCs — mutual repo deps carried by *different*,
/// feature-gated crates — into a valid crate-level sequence, so a back-edge to a
/// later repo is not a blocker. Honouring it would deadlock the very SCCs that
/// `crate_publish_order` untangled, re-reporting a clean crate DAG as blocked.
///
/// Termination is bounded: every pass either transitions a repo toward a terminal
/// state or consumes a (capped) publish attempt, so the fixpoint loop always halts.
pub(crate) fn run_cascade<F>(order: &[StageStep], mut publish: F) -> StageOutcome
where
    F: FnMut(&str) -> PublishAttempt,
{
    let n = order.len();
    let pos: BTreeMap<&str, usize> =
        order.iter().enumerate().map(|(i, s)| (s.repo.as_str(), i)).collect();
    // Blocking deps of repo i: its in-constellation deps that come BEFORE it in order.
    let blocking: Vec<Vec<usize>> = order
        .iter()
        .enumerate()
        .map(|(i, s)| {
            s.deps
                .iter()
                .filter_map(|d| pos.get(d.as_str()).copied())
                .filter(|&j| j < i)
                .collect()
        })
        .collect();

    let mut state = vec![CascadeState::Pending; n];
    let mut attempts = vec![0u32; n];
    let mut last_err: Vec<Option<String>> = vec![None; n];
    let mut out = StageOutcome::default();

    // Fixpoint over publish waves. A single ascending pass already drains a clean run
    // (deps are earlier indices, so they resolve before we reach a dependent); extra
    // passes exist only to retry transient failures.
    loop {
        let mut progressed = false;
        for i in 0..n {
            if state[i] != CascadeState::Pending {
                continue;
            }
            // A terminally-failed / blocked dependency blocks this repo outright.
            let dep_dead = blocking[i]
                .iter()
                .any(|&d| matches!(state[d], CascadeState::Failed | CascadeState::Blocked));
            if dep_dead {
                state[i] = CascadeState::Blocked;
                progressed = true;
                continue;
            }
            // Otherwise wait until EVERY blocking dep has published / already-staged.
            if !blocking[i].iter().all(|&d| state[d] == CascadeState::Done) {
                continue;
            }
            if attempts[i] >= MAX_ATTEMPTS {
                state[i] = CascadeState::Failed;
                progressed = true;
                continue;
            }
            attempts[i] += 1;
            progressed = true;
            match publish(&order[i].repo) {
                PublishAttempt::Published => {
                    out.published.push(order[i].repo.clone());
                    state[i] = CascadeState::Done;
                }
                PublishAttempt::AlreadyStaged => {
                    out.already_staged.push(order[i].repo.clone());
                    state[i] = CascadeState::Done;
                }
                PublishAttempt::Failed(e) => {
                    last_err[i] = Some(e);
                    if attempts[i] >= MAX_ATTEMPTS {
                        state[i] = CascadeState::Failed; // retries exhausted → own failure
                    }
                    // else: stays Pending, retried on a later pass.
                }
            }
        }
        if !progressed {
            break;
        }
    }

    // Emit terminal outcomes in publish order. A repo that failed on its own merits →
    // `errors`; one deferred by a dead dependency → `blocked` (distinct). Any residual
    // Pending can only be a stuck sub-graph (a dep that never resolved) → also blocked.
    for i in 0..n {
        match state[i] {
            CascadeState::Failed => {
                let why =
                    last_err[i].clone().unwrap_or_else(|| "cargo publish failed".to_string());
                out.errors.push(format!("{}: {}", order[i].repo, why));
            }
            CascadeState::Blocked | CascadeState::Pending => {
                let culprit = blocking[i]
                    .iter()
                    .find(|&&d| state[d] != CascadeState::Done)
                    .map(|&d| order[d].repo.clone())
                    .unwrap_or_else(|| "upstream".to_string());
                out.blocked
                    .push(format!("{}: blocked by failed dependency {}", order[i].repo, culprit));
            }
            CascadeState::Done => {}
        }
    }
    out
}

/// Steps 1–3: spawn the embedded holger, cut the release branch per repo, then in
/// plan order `cargo publish --workspace` each repo into `/sparring`, then
/// `cargo build` each verify target against `/sparring`. Stops before seal/promote.
/// holger is torn down on return. Per-repo failures are collected, not fatal, so
/// the outcome shows how far the rehearsal got.
pub fn execute(
    plan: &StagePlan,
    repo_paths: &BTreeMap<String, PathBuf>,
    _holger_bin: &Path,
    data_dir: &Path,
    grpc: &str,
    http: &str,
) -> Result<StageOutcome> {
    if !plan.is_executable() {
        anyhow::bail!("stage plan not executable (dependency cycle: {})", plan.cycle.join(", "));
    }
    std::fs::create_dir_all(data_dir)?;
    let cfg_path = data_dir.join("cargo-config.toml");
    std::fs::write(&cfg_path, cargo_config_toml(http))?;

    // Bring up the /cache + /sparring dev pair. Default build: in-process (the
    // nornir⇆holger marriage). Lean `--no-default-features` build: subprocess.
    // `_holger` stays alive until this fn returns, then the rehearsal registry is
    // torn down (runtime dropped / child killed).
    #[cfg(feature = "embed-holger")]
    let _holger = {
        let h = crate::holger_embed::EmbeddedHolger::start(&data_dir.join("registry"), grpc, http)
            .context("start embedded holger")?;
        h.wait_ready(std::time::Duration::from_secs(15))
            .context("embedded holger did not become ready")?;
        h
    };
    #[cfg(not(feature = "embed-holger"))]
    let _holger = {
        let p = spawn_holger(_holger_bin, &data_dir.join("registry"), grpc, http)?;
        wait_ready(http)?;
        p
    };
    eprintln!("stage: holger /cache + /sparring ready on http://{http}");

    // Auto-restore the working trees when the rehearsal ends (Drop runs on every exit
    // path, so no repo is left stranded on `release/staging`).
    let mut guard = BranchGuard::new(&plan.branch);

    // 1 + 2 — cut branch, then publish repos in DEPENDENCY WAVES (not a single linear
    // fail-forward walk): each repo is attempted only once its in-constellation deps
    // have published, a dependent of a FAILED dep is deferred/blocked (never
    // attempted-and-failed), and a transient failure is retried on a later pass.
    let mut out = run_cascade(&plan.order, |repo| {
        let Some(path) = repo_paths.get(repo) else {
            // No working tree here (a verify-only / external repo): nothing to publish,
            // and it must not block its dependents — treat as already satisfied.
            return PublishAttempt::AlreadyStaged;
        };
        guard.capture(path);
        let _ = Command::new("git")
            .arg("-C")
            .arg(path)
            .args(["checkout", "-B", &plan.branch])
            .status(); // best-effort branch isolation
        let res = Command::new("cargo")
            .current_dir(path)
            .args(["publish", "--workspace", "--no-verify", "--allow-dirty", "--registry", "sparring"])
            .arg("--config")
            .arg(&cfg_path)
            .env("CARGO_REGISTRIES_SPARRING_TOKEN", "stage-rehearsal")
            .output();
        let res = match res {
            Ok(r) => r,
            Err(e) => return PublishAttempt::Failed(format!("cargo publish spawn failed: {e}")),
        };
        if res.status.success() {
            eprintln!("stage: published {repo} → /sparring");
            PublishAttempt::Published
        } else {
            let err = String::from_utf8_lossy(&res.stderr);
            // crates.io / sparring both reject re-publishing an existing version. For a
            // rehearsal that's a benign "already staged" — surface it, then let the
            // verify step still prove it builds. A real publish failure (compile, auth,
            // missing metadata) is a genuine error.
            if err.contains("already exists") || err.contains("already uploaded") {
                eprintln!("stage: {repo} already staged in /sparring (skip; bump to re-stage)");
                PublishAttempt::AlreadyStaged
            } else {
                eprint!("{err}");
                PublishAttempt::Failed(format!("cargo publish failed ({})", res.status))
            }
        }
    });

    // 3 — verify each dependent builds against /sparring (unique targets). Only verify
    // dependents of repos whose crates actually reached /sparring — a repo that failed
    // or was blocked never published, so building against it would fail for that reason.
    let done: std::collections::BTreeSet<String> =
        out.published.iter().chain(out.already_staged.iter()).cloned().collect();
    let mut seen = std::collections::BTreeSet::new();
    for step in &plan.order {
        if !done.contains(step.repo.as_str()) {
            continue;
        }
        for dep in &step.verify {
            if !seen.insert(dep.clone()) {
                continue;
            }
            let Some(path) = repo_paths.get(dep) else { continue };
            let status = Command::new("cargo")
                .current_dir(path)
                .args(["build", "--quiet"])
                .arg("--config")
                .arg(&cfg_path)
                .status()
                .with_context(|| format!("cargo build {dep}"))?;
            if status.success() {
                out.verified.push(dep.clone());
                eprintln!("stage: verified {} builds from /sparring", dep);
            } else {
                out.errors.push(format!("{dep}: verify build failed ({status})"));
            }
        }
    }

    Ok(out)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::BTreeSet;

    fn graph(repo: &str, produces: &[&str], deps: &[&str]) -> RepoGraph {
        let dep_set = deps.iter().map(|s| s.to_string()).collect::<BTreeSet<_>>();
        RepoGraph {
            repo: repo.to_string(),
            produces: produces.iter().map(|s| s.to_string()).collect::<BTreeSet<_>>(),
            deps: dep_set.clone(),
            // Each produced crate carries the repo's deps as its crate-level deps, so
            // crate_publish_order / detect_cycles see the real per-crate graph: a
            // repo-level SCC that is a clean crate DAG is orderable, while a true
            // crate cycle (a-crate↔b-crate) still blocks.
            crate_deps: produces
                .iter()
                .map(|p| (p.to_string(), dep_set.clone()))
                .collect::<std::collections::BTreeMap<_, _>>(),
            ..Default::default()
        }
    }

    /// The plan publishes in dependency order, and each base repo lists the
    /// dependents to verify: znippy & skade before nornir; both list nornir.
    #[test]
    fn plan_orders_topologically_with_verify_targets() {
        let graphs = [
            graph("znippy", &["znippy-common"], &["serde"]),
            graph("skade", &["skade-katalog"], &["arrow"]),
            graph("nornir", &["nornir"], &["znippy-common", "skade-katalog"]),
        ];
        let plan = plan_stage(&graphs, "sparse+http://h/sparring/index/", "release/arrow-58");
        assert!(plan.is_executable(), "clean DAG must be executable");
        assert_eq!(plan.branch, "release/arrow-58");

        let order: Vec<&str> = plan.order.iter().map(|s| s.repo.as_str()).collect();
        let pos = |r: &str| order.iter().position(|x| *x == r).unwrap();
        assert!(pos("znippy") < pos("nornir") && pos("skade") < pos("nornir"));

        let znippy = plan.order.iter().find(|s| s.repo == "znippy").unwrap();
        assert_eq!(znippy.verify, vec!["nornir"], "publishing znippy must verify nornir");
        let nornir = plan.order.iter().find(|s| s.repo == "nornir").unwrap();
        assert!(nornir.verify.is_empty(), "nothing depends on nornir here");
    }

    #[test]
    fn cargo_config_targets_sparring_and_replaces_crates_io() {
        let c = cargo_config_toml("127.0.0.1:18464");
        assert!(c.contains("[registries.sparring]"), "defines the publish registry");
        assert!(c.contains("sparse+http://127.0.0.1:18464/sparring/index/"));
        assert!(c.contains(r#"replace-with = "sparring-mirror""#), "resolves deps via /sparring");
    }

    #[test]
    fn plan_with_cycle_is_not_executable() {
        let graphs = [
            graph("a", &["a-crate"], &["b-crate"]),
            graph("b", &["b-crate"], &["a-crate"]),
        ];
        let plan = plan_stage(&graphs, "reg", "rel");
        assert!(!plan.is_executable(), "a cycle must block execution");
        assert_eq!(plan.cycle.len(), 2);
    }

    /// plan_stage now records each repo's in-constellation dependencies so the cascade
    /// can gate on them: nornir depends on znippy + skade; the bases depend on nothing.
    #[test]
    fn plan_records_in_constellation_deps() {
        let graphs = [
            graph("znippy", &["znippy-common"], &["serde"]),
            graph("skade", &["skade-katalog"], &["arrow"]),
            graph("nornir", &["nornir"], &["znippy-common", "skade-katalog"]),
        ];
        let plan = plan_stage(&graphs, "reg", "rel");
        let nornir = plan.order.iter().find(|s| s.repo == "nornir").unwrap();
        assert_eq!(nornir.deps, vec!["skade", "znippy"], "nornir gates on both bases");
        let znippy = plan.order.iter().find(|s| s.repo == "znippy").unwrap();
        assert!(znippy.deps.is_empty(), "a base repo gates on nothing");
    }

    fn step(repo: &str, deps: &[&str], verify: &[&str]) -> StageStep {
        StageStep {
            repo: repo.to_string(),
            deps: deps.iter().map(|s| s.to_string()).collect(),
            verify: verify.iter().map(|s| s.to_string()).collect(),
        }
    }

    /// B depends on A: A is attempted first, and B only AFTER A has published.
    #[test]
    fn cascade_defers_dependent_until_dep_published() {
        let order = [step("a", &[], &["b"]), step("b", &["a"], &[])];
        let mut seen: Vec<String> = Vec::new();
        let out = run_cascade(&order, |r| {
            seen.push(r.to_string());
            PublishAttempt::Published
        });
        assert_eq!(seen, ["a", "b"], "A must be attempted before B");
        assert_eq!(out.published, ["a", "b"]);
        assert!(out.errors.is_empty() && out.blocked.is_empty());
    }

    /// When A fails, B is NEVER attempted — it is reported blocked-by-dep, distinct
    /// from A which is reported as its own failure in `errors`.
    #[test]
    fn cascade_blocks_dependent_when_dep_fails() {
        let order = [step("a", &[], &["b"]), step("b", &["a"], &[])];
        let mut attempted: Vec<String> = Vec::new();
        let out = run_cascade(&order, |r| {
            attempted.push(r.to_string());
            if r == "a" {
                PublishAttempt::Failed("boom".to_string())
            } else {
                PublishAttempt::Published
            }
        });
        assert!(!attempted.iter().any(|r| r == "b"), "B must not be attempted once A fails");
        assert!(out.published.is_empty());
        assert!(out.errors.iter().any(|e| e.starts_with("a:")), "A is its own failure");
        assert!(
            out.blocked.iter().any(|e| e.starts_with("b:") && e.contains("a")),
            "B is reported blocked-by-failed-dependency a, not as a failure: {:?}",
            out.blocked
        );
    }

    /// A transient failure is RETRIED on a later pass and can then succeed.
    #[test]
    fn cascade_retries_transient_failure_then_succeeds() {
        let order = [step("a", &[], &[])];
        let mut calls = 0u32;
        let out = run_cascade(&order, |_| {
            calls += 1;
            if calls == 1 {
                PublishAttempt::Failed("transient".to_string())
            } else {
                PublishAttempt::Published
            }
        });
        assert!(calls >= 2, "a transient failure must get another attempt (got {calls})");
        assert_eq!(out.published, ["a"]);
        assert!(out.errors.is_empty() && out.blocked.is_empty());
    }

    /// A permanently-failing repo converges to `errors` after MAX_ATTEMPTS — it does
    /// NOT spin forever, and its attempts are bounded.
    #[test]
    fn cascade_bounds_retries_of_permanent_failure() {
        let order = [step("a", &[], &[])];
        let mut calls = 0u32;
        let out = run_cascade(&order, |_| {
            calls += 1;
            PublishAttempt::Failed("always".to_string())
        });
        assert_eq!(calls, MAX_ATTEMPTS, "retries are bounded to MAX_ATTEMPTS");
        assert!(out.errors.iter().any(|e| e.starts_with("a:")));
        assert!(out.published.is_empty() && out.blocked.is_empty());
    }

    /// Idempotency: an already-staged dependency still SATISFIES its dependents' gate,
    /// so the dependent proceeds (exactly as a re-run without a version bump would).
    #[test]
    fn cascade_already_staged_satisfies_dependents() {
        let order = [step("a", &[], &["b"]), step("b", &["a"], &[])];
        let out = run_cascade(&order, |r| {
            if r == "a" {
                PublishAttempt::AlreadyStaged
            } else {
                PublishAttempt::Published
            }
        });
        assert_eq!(out.already_staged, ["a"]);
        assert_eq!(out.published, ["b"], "an already-staged dep unblocks its dependent");
        assert!(out.blocked.is_empty() && out.errors.is_empty());
    }

    /// A back-edge to a LATER repo in the order (a repo-level SCC that crate ordering
    /// already untangled) is NOT a blocker — both repos still publish.
    #[test]
    fn cascade_ignores_back_edges_within_ordered_scc() {
        // Order is [a, b]; b depends on a (forward, gating) AND a "depends on" b
        // (back-edge — the SCC's other direction, already resolved by the order).
        let order = [step("a", &["b"], &["b"]), step("b", &["a"], &[])];
        let out = run_cascade(&order, |_| PublishAttempt::Published);
        assert_eq!(out.published, ["a", "b"], "the back-edge must not deadlock the SCC");
        assert!(out.blocked.is_empty() && out.errors.is_empty());
    }
}