nornir 0.5.3

//! Publish-order walker. Per-repo TOML defines `publish_order` as
//! `Vec<Vec<String>>`: outer entries are sequential phases, inner
//! entries are independent and may be published in parallel.
//!
//! Tag creation is done in-process via the pure-Rust `gix` crate (no
//! `git` shellout). Tag *pushing* over the network is deferred to a
//! follow-up that wires `gix` network features — for now we print the
//! exact `git push` invocation so a human or CI can complete the step.
//!
//! `cargo publish` itself remains a `cargo` subprocess because cargo
//! owns crate packaging (sources → `.crate` tarball → upload). That is
//! cargo's domain, not a script we can replace with a few lines of
//! HTTP. Isolated to [`run_cargo_publish`] and clearly annotated.

use std::path::Path;
use std::process::Command;

use anyhow::{anyhow, Context, Result};

/// Outcome reported per crate so the orchestrator (and progress
/// writer) can distinguish "really uploaded" from "already on the
/// registry" — re-running a release of a tree that's partially-shipped
/// is a routine case, not an error.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PublishOutcome {
    Published,
    AlreadyPublished,
    DryRun,
}

/// Walk the publish order, invoking cargo to publish each crate.
/// `dry_run` adds `--dry-run`. Returns the per-crate outcomes in the
/// same order they were published (outer phase order, inner declared
/// order). Errors only on real cargo failures — "crate already on the
/// registry" downgrades to [`PublishOutcome::AlreadyPublished`].
pub fn publish_all(
    repo_root: &Path,
    order: &[Vec<String>],
    dry_run: bool,
) -> Result<Vec<(String, PublishOutcome)>> {
    // No `publish_order` configured → DERIVE it from the workspace dependency
    // graph (topological, deps-first) instead of silently publishing nothing
    // (the bug that made `nornir release publish` exit 0 having uploaded nothing).
    // The manual `publish_order` is now an OVERRIDE, not a requirement.
    let derived;
    let order: &[Vec<String>] = if order.iter().all(|phase| phase.is_empty()) {
        derived = derive_publish_order(repo_root).with_context(|| {
            "no `publish_order` configured and could not derive one from `cargo metadata`"
        })?;
        if derived.iter().all(|p| p.is_empty()) {
            anyhow::bail!(
                "nothing to publish: no publishable crates in the workspace at {} \
                 (all `publish = false`?). Set `publish_order` in nornir.toml to override.",
                repo_root.display()
            );
        }
        &derived
    } else {
        order
    };
    let mut out: Vec<(String, PublishOutcome)> = Vec::new();
    for phase in order {
        for krate in phase {
            let outcome = run_cargo_publish(repo_root, krate, dry_run)
                .with_context(|| format!("cargo publish -p {krate}"))?;
            out.push((krate.clone(), outcome));
        }
    }
    Ok(out)
}

/// Deterministically derive the publish order from the workspace dependency
/// graph via `cargo metadata`: a topological sort (deps first) of the
/// **publishable** workspace members, grouped into phases (crates with no
/// remaining intra-workspace dependency publish together). The publish graph is a
/// DAG over the deps that actually gate publishing — `Normal` AND `Build`
/// dependencies (a build-dep must be on crates.io before its dependent), but NOT
/// `Development` deps: a dev-dependency is stripped by cargo at publish time, so
/// counting it would invent false cycles and publish a dependent before its real
/// dependency. `publish = false` crates are dropped. Fully deterministic: members
/// and phase contents are alphabetically ordered.
pub fn derive_publish_order(repo_root: &Path) -> Result<Vec<Vec<String>>> {
    let meta = cargo_metadata::MetadataCommand::new()
        .current_dir(repo_root)
        .exec()
        .with_context(|| format!("cargo metadata in {}", repo_root.display()))?;
    let ws: std::collections::BTreeSet<String> =
        meta.workspace_packages().iter().map(|p| p.name.to_string()).collect();
    let pkgs: Vec<(String, bool, Vec<(String, cargo_metadata::DependencyKind)>)> = meta
        .workspace_packages()
        .iter()
        .map(|p| {
            // `publish = false` serializes as `Some([])` → unpublishable.
            let publishable = !matches!(&p.publish, Some(v) if v.is_empty());
            let deps = p.dependencies.iter().map(|d| (d.name.clone(), d.kind)).collect();
            (p.name.to_string(), publishable, deps)
        })
        .collect();
    Ok(topo_phases(members_from_deps(&ws, &pkgs)))
}

/// Pure publish-graph builder (bug #9, testable without `cargo metadata`): drop
/// `publish = false` members, and for each remaining member keep only its
/// intra-workspace deps that gate publishing — `Normal` + `Build`, never
/// `Development`. A dev-dependency back-edge (e.g. `core` dev-depends on `app`
/// while `app` normal-depends on `core`) would otherwise produce a false cycle
/// and mis-order the publish; cargo strips dev-deps at publish time, so they must
/// not count here.
pub(crate) fn members_from_deps(
    ws: &std::collections::BTreeSet<String>,
    pkgs: &[(String, bool, Vec<(String, cargo_metadata::DependencyKind)>)],
) -> Vec<(String, Vec<String>)> {
    let mut members: Vec<(String, Vec<String>)> = Vec::new();
    for (name, publishable, deps) in pkgs {
        if !*publishable {
            continue;
        }
        let deps: Vec<String> = deps
            .iter()
            .filter(|(_, kind)| *kind != cargo_metadata::DependencyKind::Development)
            .map(|(n, _)| n.clone())
            .filter(|n| ws.contains(n))
            .collect();
        members.push((name.clone(), deps));
    }
    members
}

/// Pure, deterministic Kahn topological sort into phases (deps-first). Each
/// `(name, deps)` lists the crate's intra-workspace dependencies. A crate is
/// "ready" when none of its workspace deps remain; ready crates form one phase
/// (alphabetical). Robust to a cycle (shouldn't occur for a publishable
/// workspace): the unresolved remainder is appended as a final phase so nothing
/// is silently dropped.
pub(crate) fn topo_phases(members: Vec<(String, Vec<String>)>) -> Vec<Vec<String>> {
    use std::collections::{BTreeMap, BTreeSet};
    let names: BTreeSet<String> = members.iter().map(|(n, _)| n.clone()).collect();
    let mut indeg: BTreeMap<String, usize> = BTreeMap::new();
    let mut dependents: BTreeMap<String, Vec<String>> = BTreeMap::new(); // dep → its dependents
    for (n, deps) in &members {
        let real: Vec<&String> =
            deps.iter().filter(|d| names.contains(*d) && d.as_str() != n).collect();
        indeg.insert(n.clone(), real.len());
        for d in real {
            dependents.entry(d.clone()).or_default().push(n.clone());
        }
    }
    let mut phases: Vec<Vec<String>> = Vec::new();
    let mut remaining: BTreeSet<String> = names;
    while !remaining.is_empty() {
        let ready: Vec<String> = remaining
            .iter()
            .filter(|n| indeg.get(*n).copied().unwrap_or(0) == 0)
            .cloned()
            .collect();
        if ready.is_empty() {
            phases.push(remaining.iter().cloned().collect()); // cycle guard — drop nothing
            break;
        }
        for n in &ready {
            remaining.remove(n);
            if let Some(deps) = dependents.get(n) {
                for dep in deps {
                    if let Some(e) = indeg.get_mut(dep) {
                        *e = e.saturating_sub(1);
                    }
                }
            }
        }
        phases.push(ready); // already alphabetical (from the BTreeSet iteration)
    }
    phases
}

/// Default cap on crates.io 429 rate-limit retries for one crate. crates.io's
/// publish limit replenishes one slot per ~60 s after a burst. A FAT cascade
/// publishing dozens of BRAND-NEW crate names (e.g. facett's ~60-crate repo) trips
/// crates.io's stricter *new-crate* limiter, which hands back ~10-min "try again"
/// waits — so a large burst needs many honored retries, not a handful, or it bails
/// mid-repo. Each retry waits the server-suggested duration (capped by
/// [`MAX_RATE_LIMIT_WAIT`]); 20 covers a full big-repo burst without an unbounded loop.
pub const DEFAULT_RATE_LIMIT_RETRIES: u32 = 20;

/// Hard ceiling on a single parsed 429 wait so a server-suggested
/// multi-hour delay can't hang the pipeline. crates.io's NEW-crate limiter
/// hands back an absolute "try again after <t>" ~10 min out once the burst
/// bucket is spent (see [`wait_until_http_date`]); 30 min leaves margin above
/// that window without risking an unbounded hang.
pub const MAX_RATE_LIMIT_WAIT: std::time::Duration = std::time::Duration::from_secs(30 * 60);

/// Parse the wait duration out of crates.io's publish rate-limit (HTTP 429)
/// diagnostic. cargo surfaces the body verbatim, e.g.:
///   "You have published too many crates in a short period of time,
///    please try again after Wed, 18 Jun 2026 12:34:56 GMT, or email ..."
/// or the shorter relative form:
///   "... please try again after 56 seconds ..."
///   "... please try again in 4 minutes ..."
///
/// Returns `Some(duration)` if the message is recognisably a publish rate
/// limit, parsing a relative "<n> seconds/minutes" if present and otherwise
/// falling back to `default` (we can't reliably parse an absolute RFC-2822
/// timestamp without a date lib, and a fixed fallback is safer than guessing).
/// Returns `None` when the message is not a rate-limit error at all.
pub fn parse_rate_limit_wait(
    stderr: &str,
    default: std::time::Duration,
) -> Option<std::time::Duration> {
    let lower = stderr.to_ascii_lowercase();
    let is_rate_limit = lower.contains("too many crates")
        || (lower.contains("rate limit") && lower.contains("publish"))
        || (lower.contains("429") && lower.contains("publish"));
    if !is_rate_limit {
        return None;
    }
    // Look for a relative "<n> second(s)" / "<n> minute(s)" hint anywhere in
    // the message (handles both "after N seconds" and "in N minutes").
    let parse_after = |unit: &str, scale: u64| -> Option<std::time::Duration> {
        let idx = lower.find(unit)?;
        // Walk back over the unit's leading whitespace to the trailing digits.
        let prefix = &lower[..idx];
        let digits: String = prefix
            .trim_end()
            .chars()
            .rev()
            .take_while(|c| c.is_ascii_digit())
            .collect::<String>()
            .chars()
            .rev()
            .collect();
        let n: u64 = digits.parse().ok()?;
        Some(std::time::Duration::from_secs(n.saturating_mul(scale)))
    };
    // crates.io's NEW-crate limiter does NOT emit a relative hint — once the
    // burst bucket is spent it hands back an ABSOLUTE deadline ("...try again
    // after Fri, 03 Jul 2026 05:36:18 GMT"). Wait until then (not a blind 60s,
    // which burns every retry before the ~10 min window opens). Relative hints
    // still win if present; absolute is the fallback before `default`.
    let now_unix = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    let parsed = parse_after(" second", 1)
        .or_else(|| parse_after(" minute", 60))
        .or_else(|| wait_until_http_date(stderr, now_unix))
        .unwrap_or(default);
    Some(parsed.min(MAX_RATE_LIMIT_WAIT))
}

/// Parse a crates.io ABSOLUTE rate-limit deadline — "...try again after Fri,
/// 03 Jul 2026 05:36:18 GMT" — into the wait from `now_unix` (seconds since the
/// epoch). `None` if the message carries no absolute deadline. A small buffer is
/// added so we don't wake a hair before the window opens; a deadline already in
/// the past yields the buffer (retry almost immediately). `now_unix` is injected
/// so the parse is deterministically testable.
pub fn wait_until_http_date(stderr: &str, now_unix: u64) -> Option<std::time::Duration> {
    const BUFFER_SECS: u64 = 15;
    let lower = stderr.to_ascii_lowercase();
    let key = "try again after ";
    let pos = lower.find(key)? + key.len();
    // to_ascii_lowercase preserves byte length (ASCII), so `pos` indexes the
    // original too — slice it to keep the month name's case ("Jul").
    let rest = &stderr[pos..];
    let end = rest
        .to_ascii_lowercase()
        .find("gmt")
        .map(|i| i + 3)
        .unwrap_or(rest.len());
    let date_str = rest[..end].trim().trim_end_matches('.').trim();
    let deadline = http_date_to_unix(date_str)?;
    Some(std::time::Duration::from_secs(
        deadline.saturating_sub(now_unix).saturating_add(BUFFER_SECS),
    ))
}

/// Convert an RFC-1123 / HTTP-date "Fri, 03 Jul 2026 05:36:18 GMT" (the optional
/// leading weekday is tolerated) to seconds since the Unix epoch — pure calendar
/// arithmetic (Howard Hinnant's `days_from_civil`), no date crate. `None` on any
/// malformed field.
fn http_date_to_unix(s: &str) -> Option<u64> {
    // Drop the optional "Fri, " weekday prefix.
    let s = s.trim();
    let s = s.splitn(2, ", ").last().unwrap_or(s); // "03 Jul 2026 05:36:18 GMT"
    let mut it = s.split_whitespace();
    let day: i64 = it.next()?.parse().ok()?;
    let mon: i64 = match it.next()? {
        "Jan" => 1, "Feb" => 2, "Mar" => 3, "Apr" => 4,
        "May" => 5, "Jun" => 6, "Jul" => 7, "Aug" => 8,
        "Sep" => 9, "Oct" => 10, "Nov" => 11, "Dec" => 12,
        _ => return None,
    };
    let year: i64 = it.next()?.parse().ok()?;
    let mut hms = it.next()?.split(':');
    let h: i64 = hms.next()?.parse().ok()?;
    let mi: i64 = hms.next()?.parse().ok()?;
    let se: i64 = hms.next()?.parse().ok()?;
    if !(1..=31).contains(&day) || !(0..=23).contains(&h) || mi > 59 || se > 60 {
        return None;
    }
    // days_from_civil: days since 1970-01-01 for a proleptic Gregorian date.
    let y = if mon <= 2 { year - 1 } else { year };
    let era = if y >= 0 { y } else { y - 399 } / 400;
    let yoe = y - era * 400; // [0, 399]
    let mp = if mon > 2 { mon - 3 } else { mon + 9 }; // Mar=0..Feb=11
    let doy = (153 * mp + 2) / 5 + day - 1; // [0, 365]
    let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy; // [0, 146096]
    let days = era * 146097 + doe - 719468;
    let secs = days * 86400 + h * 3600 + mi * 60 + se;
    u64::try_from(secs).ok()
}

/// A classified `cargo publish` failure — the machine-readable core behind
/// [`explain_publish_failure`]. Lets the preflight distinguish a real blocker
/// (a non-member, a dep-API skew, missing metadata) from an EXPECTED
/// mid-cascade condition (a sibling that just isn't on the registry yet), so a
/// dry-run verify doesn't cry wolf on ordinary publish ordering.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PublishFailure {
    /// cargo error 101 "package ID specification `<crate>` did not match any
    /// packages" — the crate lives in a detached `[workspace]`.
    NonMember,
    /// `cargo publish --verify` compiled the crate against the REGISTRY copy of
    /// a dependency and failed on API that only exists in the local (un-bumped)
    /// source. `dep` is the offending dependency crate when we could name it.
    DepApiSkew { dep: Option<String> },
    /// crates.io rejected the manifest for missing required metadata.
    MissingMetadata { fields: Vec<String> },
    /// A resolution/ordering condition, NOT a code defect: a dependency version
    /// isn't on the registry yet (normal mid-cascade — the deps-first order
    /// fixes it). Never a preflight blocker.
    ResolveOrdering,
    /// crates.io publish rate-limit (HTTP 429).
    RateLimited,
    /// Unclassified — surface cargo's raw stderr.
    Other,
}

/// Classify a failed `cargo publish` from its captured stderr. Pure and
/// order-sensitive: the most specific signatures are tested first. `krate` is
/// only used to disambiguate an "unresolved import `<krate>`" self-reference.
/// `known_crates` is the set of workspace/release crate names used to VALIDATE a
/// dep-API-skew blame token (so a rustc `could not find X in `kind`` blames the
/// CRATE, not the module `kind`); pass an empty set to skip validation and keep
/// the legacy extraction order.
pub fn classify_publish_failure(
    _krate: &str,
    stderr: &str,
    known_crates: &std::collections::BTreeSet<String>,
) -> PublishFailure {
    let lower = stderr.to_ascii_lowercase();

    // (1) Not a workspace member — detached `[workspace]` (the skade bug).
    if lower.contains("did not match any packages") {
        return PublishFailure::NonMember;
    }

    // (rate limit) — recognised the same way the retry loop recognises it.
    if lower.contains("too many crates")
        || (lower.contains("rate limit") && lower.contains("publish"))
        || (lower.contains("429") && lower.contains("publish"))
    {
        return PublishFailure::RateLimited;
    }

    // (3) Missing crates.io metadata — cargo lists the offending fields.
    if lower.contains("missing") && (lower.contains("metadata") || lower.contains("field")) {
        let mut fields = Vec::new();
        for f in ["description", "license", "license-file", "repository"] {
            if lower.contains(f) {
                fields.push(f.to_string());
            }
        }
        // `license-file` contains `license`; dedup the redundant `license`.
        if fields.iter().any(|f| f == "license-file") {
            fields.retain(|f| f != "license");
        }
        if !fields.is_empty() {
            return PublishFailure::MissingMetadata { fields };
        }
    }

    // (2) A COMPILE failure inside `cargo publish --verify` citing a dependency
    // whose local source moved ahead of its published version. Distinguish this
    // (a real code/version defect) from a pure RESOLUTION failure (the sibling
    // simply isn't published yet — expected mid-cascade). Compile errors carry a
    // rustc `error[E04xx]` / "cannot find" / "could not find `X` in `<dep>`";
    // resolution errors carry cargo's "failed to select a version" / "no matching
    // package" / "which is neither ...".
    let is_resolution = lower.contains("failed to select a version")
        || lower.contains("no matching package")
        || lower.contains("failed to get")
        || lower.contains("which is neither");
    let is_compile = lower.contains("error[e0432]")   // unresolved import
        || lower.contains("error[e0433]")             // failed to resolve
        || lower.contains("error[e0405]")             // cannot find trait
        || lower.contains("error[e0412]")             // cannot find type
        || lower.contains("error[e0425]")             // cannot find value
        || lower.contains("cannot find")
        || lower.contains("could not find `");
    if is_compile && !is_resolution {
        return PublishFailure::DepApiSkew { dep: extract_skew_dep(stderr, known_crates) };
    }
    if is_resolution {
        return PublishFailure::ResolveOrdering;
    }

    PublishFailure::Other
}

/// Best-effort: pull the dependency crate name out of a verify compile error.
///
/// Two shapes carry the culprit: a crate-qualified import path
/// (`unresolved import `dep::…``) whose FIRST segment IS the crate, and the
/// `could not find `Sym` in `X`` form whose `X` is only the innermost path
/// segment — which is a MODULE (e.g. `kind`), not the crate, when the symbol is
/// re-exported through a module (the dwarves↔nornir-jobs class: blame was `kind`,
/// not `nornir-jobs`, so `skew_is_cascade_resolvable` never matched it).
///
/// So PREFER the crate-qualified import-path segment and validate the extracted
/// token against `known_crates`: accept the import-path crate when it's a known
/// crate; else accept the `in `X`` token only when IT is a known crate (rejecting
/// a bare module); else fall back to the import-path crate segment. With an EMPTY
/// `known_crates` (no set to validate against) keep the legacy order (`in `X``
/// first). `None` when no candidate is present.
fn extract_skew_dep(stderr: &str, known_crates: &std::collections::BTreeSet<String>) -> Option<String> {
    // "could not find `Sym` in `X`" → the SECOND backtick pair (X).
    let in_token = stderr.find(" in `").and_then(|i| {
        let rest = &stderr[i + " in `".len()..];
        rest.find('`').and_then(|end| {
            let dep = rest[..end].trim();
            (!dep.is_empty() && !dep.contains(char::is_whitespace))
                .then(|| dep.split("::").next().unwrap_or(dep).to_string())
        })
    });
    // "unresolved import `dep::path`" → first backtick pair, crate before `::`.
    let import_crate = ["unresolved import `", "use of undeclared crate or module `"]
        .iter()
        .find_map(|marker| {
            let i = stderr.to_ascii_lowercase().find(marker)?;
            let rest = &stderr[i + marker.len()..];
            let end = rest.find('`')?;
            let dep = rest[..end].trim().split("::").next().unwrap_or("").trim();
            (!dep.is_empty()).then(|| dep.to_string())
        });

    if known_crates.is_empty() {
        // No set to validate against — legacy order: the `in `X`` form first.
        return in_token.or(import_crate);
    }
    let canon = |s: &str| s.replace('-', "_");
    let is_known = |t: &str| {
        let tc = canon(t);
        known_crates.iter().any(|k| canon(k) == tc)
    };
    // Prefer the crate-qualified import-path segment when it names a known crate.
    if let Some(c) = &import_crate {
        if is_known(c) {
            return Some(c.clone());
        }
    }
    // Else accept the `in `X`` token only when IT is a known crate (reject a
    // bare module like `kind`).
    if let Some(t) = &in_token {
        if is_known(t) {
            return Some(t.clone());
        }
    }
    // Neither validated — fall back to the import-path crate segment (a
    // crate-qualified name beats the module-only `in `X`` token).
    import_crate.or(in_token)
}

/// Turn a failed `cargo publish` into a concise plain-language "why this failed
/// + how to fix" note, or `None` when nothing specific is recognised (the caller
/// then shows cargo's raw stderr alone). `repo_root` is named in the fix advice.
/// Covers the five classes the release cascade actually hits; see
/// [`classify_publish_failure`].
pub fn explain_publish_failure(krate: &str, repo_root: &Path, stderr: &str) -> Option<String> {
    // Messaging path — no workspace-crate set handy here; the blame extraction
    // degrades to its legacy (validation-free) order.
    match classify_publish_failure(krate, stderr, &std::collections::BTreeSet::new()) {
        PublishFailure::NonMember => Some(format!(
            "  ↳ why: `{krate}` is NOT a member of the workspace at {} — it likely lives \
             in a detached `[workspace]` (a bench/vendored/sub-workspace crate).\n\
             \x20    fix: mark it `publish = false`, or publish it from its own workspace \
             directory. (nornir now filters these out of the publish set automatically.)",
            repo_root.display()
        )),
        PublishFailure::DepApiSkew { dep } => {
            let d = dep.as_deref().unwrap_or("a dependency");
            Some(format!(
                "  ↳ why: `{krate}` uses API from `{d}` that isn't in the PUBLISHED `{d}` — \
                 `{d}` was edited locally without a version bump, so `cargo publish --verify` \
                 compiled against the older registry copy.\n\
                 \x20    fix: bump `{d}` and publish it FIRST, then publish `{krate}` \
                 (deps-first order). Re-run the cascade once `{d}` is on crates.io."
            ))
        }
        PublishFailure::MissingMetadata { fields } => Some(format!(
            "  ↳ why: `{krate}` is missing required crates.io metadata: {}.\n\
             \x20    fix: add the field(s) to `{krate}`'s [package] in its Cargo.toml under {} \
             (e.g. `description = \"…\"`, `license = \"Apache-2.0\"`, `repository = \"…\"`).",
            fields.join(", "),
            repo_root.display()
        )),
        PublishFailure::ResolveOrdering => Some(format!(
            "  ↳ why: `{krate}` couldn't resolve a dependency version from the registry — \
             a sibling it depends on isn't published yet.\n\
             \x20    fix: publish that dependency first (the deps-first cascade normally \
             handles this — check the publish order)."
        )),
        // Rate-limit has its own retry/backoff messaging; don't double-explain.
        PublishFailure::RateLimited => None,
        PublishFailure::Other => None,
    }
}

/// The single intentional `cargo` subprocess. Cargo owns crate
/// packaging+upload; replacing it would require reimplementing
/// `cargo package` (thousands of lines covering manifest validation,
/// `.cargo_vcs_info.json`, exclude/include globs, tarball
/// determinism, license-file handling, etc.). Out of scope; not
/// every wheel needs reinventing.
///
/// On a crates.io publish rate-limit (HTTP 429 "You have published too many
/// crates in a short period of time, please try again after <t>") the wait is
/// parsed, slept off, and the same crate retried — up to
/// [`DEFAULT_RATE_LIMIT_RETRIES`] times — mirroring [`wait_for_index`]'s
/// bounded poll. A dry-run is never rate-limited (no upload), so the retry
/// loop is a no-op there.
pub fn run_cargo_publish(repo_root: &Path, krate: &str, dry_run: bool) -> Result<PublishOutcome> {
    run_cargo_publish_with_retry(
        repo_root,
        krate,
        dry_run,
        DEFAULT_RATE_LIMIT_RETRIES,
        std::time::Duration::from_secs(60),
        |d| std::thread::sleep(d),
    )
}

/// Testable core of [`run_cargo_publish`]: the `sleep` is injected so the
/// inject-assert test can drive the retry loop without real time or network.
/// `default_wait` is used when the 429 body carries no parseable relative wait.
pub fn run_cargo_publish_with_retry(
    repo_root: &Path,
    krate: &str,
    dry_run: bool,
    max_retries: u32,
    default_wait: std::time::Duration,
    mut sleep: impl FnMut(std::time::Duration),
) -> Result<PublishOutcome> {
    // PROGRESS (#6): `cargo publish` output is CAPTURED below (we need stderr for
    // failure classification), so without this banner the multi-minute per-crate
    // verify build is indistinguishable from a hang — the "is it hung?" grind. Frame
    // each crate up front and time it, so the cascade is never silent.
    let started = std::time::Instant::now();
    if dry_run {
        eprintln!("  ▸ {krate}: cargo publish --dry-run — packaging + verify-build…");
    } else {
        eprintln!("  ▸ {krate}: cargo publish — packaging → verify-build → upload → index (can take minutes)…");
    }
    let mut attempt = 0u32;
    loop {
        let mut cmd = Command::new(crate::release::cargo::cargo_bin());
        cmd.arg("publish").arg("-p").arg(krate);
        // Publishing a workspace member re-resolves Cargo.lock (recording the
        // just-published sibling versions), which dirties the tree and makes cargo
        // refuse the NEXT crate. The promote runs on a controlled, freshly-cut
        // ephemeral branch whose only uncommitted change is that lock churn, so
        // allow it — the real source changes were already captured by the bump commit.
        cmd.arg("--allow-dirty");
        if dry_run {
            cmd.arg("--dry-run");
        }
        cmd.current_dir(repo_root);
        let output = cmd.output().context("spawn cargo publish")?;
        if output.status.success() {
            let secs = started.elapsed().as_secs_f64();
            eprintln!(
                "    ✓ {krate} {} in {secs:.1}s",
                if dry_run { "verified (dry-run)" } else { "published" }
            );
            return Ok(if dry_run {
                PublishOutcome::DryRun
            } else {
                PublishOutcome::Published
            });
        }
        let stderr = String::from_utf8_lossy(&output.stderr);
        // cargo's "already uploaded" diagnostic. Treat as success so a
        // re-run of a partially-shipped tree is idempotent.
        if stderr.contains("already uploaded")
            || stderr.contains("already exists on crates.io")
            || stderr.contains("crate version") && stderr.contains("is already uploaded")
        {
            eprintln!("    (skip: {krate} already on registry)");
            return Ok(PublishOutcome::AlreadyPublished);
        }
        // crates.io publish rate-limit (HTTP 429): parse the wait, sleep, retry.
        if let Some(wait) = parse_rate_limit_wait(&stderr, default_wait) {
            if attempt < max_retries {
                attempt += 1;
                eprintln!(
                    "    (rate-limited publishing {krate}: waiting {}s then retrying \
                     [attempt {attempt}/{max_retries}])",
                    wait.as_secs()
                );
                sleep(wait);
                continue;
            }
            eprintln!("{stderr}");
            return Err(anyhow!(
                "cargo publish -p {krate}: crates.io rate limit not cleared after \
                 {max_retries} retries. Resume once the limit clears with: \
                 `nornir release doctor --publish --publish-from <repo>` (skips already-shipped repos)."
            ));
        }
        // Re-emit cargo's stderr so the operator sees the real reason, then add
        // nornir's plain-language "why + how to fix" note (Task 2 diagnostics),
        // and fold that note into the returned error so it also lands in the
        // FAIL-STOP summary.
        eprintln!("{stderr}");
        let why = explain_publish_failure(krate, repo_root, &stderr);
        if let Some(w) = &why {
            eprintln!("{w}");
        }
        return Err(match why {
            Some(w) => anyhow!("cargo publish -p {krate} exited {}\n{w}", output.status),
            None => anyhow!("cargo publish -p {krate} exited {}", output.status),
        });
    }
}

/// Stage every worktree change (modifications, new files, deletions)
/// and create a commit on `HEAD`, returning the new commit SHA — or
/// `Ok(None)` when the working tree had nothing to commit (the common
/// case after a release where the README/CHANGELOG rendered
/// byte-identical to what's on disk).
///
/// 100% pure-Rust via `gix` — no `git` subprocess. The flow mirrors
/// `git add -A && git commit`:
///   1. enumerate `HEAD`→index and index→worktree status in one pass;
///   2. refuse to run if the index carries *staged* changes that differ
///      from `HEAD`. The release pipeline always regenerates docs on a
///      clean checkout (index == HEAD), so a dirty index signals an
///      unexpected state; committing anyway could silently diverge from
///      `git add -A` semantics, so we bail loudly instead;
///   3. apply every worktree delta onto a tree editor seeded from the
///      `HEAD` tree, write the tree, and commit it with `HEAD` as parent
///      (updates the `HEAD` ref + reflog);
///   4. refresh the on-disk index from the new tree so `git status`
///      reads clean afterwards.
///
/// Note: blobs are written from raw worktree bytes. nornir only commits
/// its own generated docs (UTF-8 markdown/TOML) in repos without
/// CRLF/clean `.gitattributes` filters, so no smudge/clean normalisation
/// is performed; repos relying on content filters are out of scope.
pub fn commit_release(repo_root: &Path, message: &str) -> Result<Option<String>> {
    use gix::bstr::BString;
    use gix::dir::entry::{Kind as DiskKind, Status as DiskStatus};
    use gix::status::index_worktree::Item as IwItem;
    use gix::status::plumbing::index_as_worktree::{Change, EntryStatus};

    let repo =
        gix::open(repo_root).with_context(|| format!("gix::open {}", repo_root.display()))?;
    let head_commit = repo
        .head_commit()
        .context("resolve HEAD commit (unborn branch is unsupported here)")?;
    let head_id = head_commit.id;
    let head_tree_id = head_commit.tree().context("resolve HEAD tree")?.id;

    let iter = repo
        .status(gix::progress::Discard)
        .context("init status")?
        .head_tree(head_tree_id)
        .untracked_files(gix::status::UntrackedFiles::Files)
        .into_iter(Vec::<BString>::new())
        .context("start status walk")?;

    let mut editor = repo
        .edit_tree(head_tree_id)
        .context("seed tree editor from HEAD")?;
    let mut changed = false;

    for item in iter {
        match item.context("status item")? {
            // A tree→index difference means the index is dirty (staged
            // content differs from HEAD). Out of contract for releases.
            gix::status::Item::TreeIndex(change) => {
                return Err(anyhow!(
                    "refusing to commit: index has a staged change at `{}`; \
                     commit or reset it before releasing",
                    change.location()
                ));
            }
            gix::status::Item::IndexWorktree(iw) => match iw {
                IwItem::Modification {
                    rela_path, status, ..
                } => match status {
                    EntryStatus::Change(Change::Removed) => {
                        let rp: &gix::bstr::BStr = rela_path.as_ref();
                        editor
                            .remove(rp)
                            .with_context(|| format!("tree remove {rela_path}"))?;
                        changed = true;
                    }
                    EntryStatus::Conflict { .. } => {
                        return Err(anyhow!(
                            "refusing to commit: unresolved merge conflict at `{rela_path}`"
                        ));
                    }
                    // Modification / Type change / NeedsUpdate / IntentToAdd /
                    // SubmoduleModification: re-stage current worktree content.
                    _ => {
                        upsert_from_worktree(&repo, &mut editor, repo_root, rela_path.as_ref())?;
                        changed = true;
                    }
                },
                IwItem::DirectoryContents { entry, .. } => {
                    if matches!(entry.status, DiskStatus::Untracked)
                        && matches!(
                            entry.disk_kind,
                            Some(DiskKind::File) | Some(DiskKind::Symlink)
                        )
                    {
                        upsert_from_worktree(
                            &repo,
                            &mut editor,
                            repo_root,
                            entry.rela_path.as_ref(),
                        )?;
                        changed = true;
                    }
                }
                IwItem::Rewrite { .. } => {}
            },
        }
    }

    if !changed {
        return Ok(None);
    }
    let new_tree = editor.write().context("write release tree")?.detach();
    if new_tree == head_tree_id {
        return Ok(None);
    }
    let commit = repo
        .commit("HEAD", message, new_tree, Some(head_id))
        .context("create release commit")?;

    // Refresh the on-disk index from the new tree so the working tree
    // reads clean after the commit (as `git commit` would leave it).
    let mut index = repo
        .index_from_tree(&new_tree)
        .context("rebuild index from release tree")?;
    index
        .write(gix::index::write::Options::default())
        .context("persist refreshed index")?;

    Ok(Some(commit.to_string()))
}

/// Read `rela_path` from the working tree and upsert it into `editor`
/// with the correct blob kind (regular / executable / symlink).
fn upsert_from_worktree(
    repo: &gix::Repository,
    editor: &mut gix::object::tree::Editor<'_>,
    repo_root: &Path,
    rela_path: &gix::bstr::BStr,
) -> Result<()> {
    use gix::object::tree::EntryKind;

    let disk_path = repo_root.join(gix::path::from_bstr(rela_path).as_ref());
    let meta = std::fs::symlink_metadata(&disk_path)
        .with_context(|| format!("stat {}", disk_path.display()))?;

    let (bytes, kind): (Vec<u8>, EntryKind) = if meta.file_type().is_symlink() {
        let target = std::fs::read_link(&disk_path)
            .with_context(|| format!("readlink {}", disk_path.display()))?;
        let target = gix::path::into_bstr(target).into_owned();
        (target.into(), EntryKind::Link)
    } else {
        let bytes = std::fs::read(&disk_path)
            .with_context(|| format!("read {}", disk_path.display()))?;
        #[cfg(unix)]
        let kind = {
            use std::os::unix::fs::PermissionsExt;
            if meta.permissions().mode() & 0o111 != 0 {
                EntryKind::BlobExecutable
            } else {
                EntryKind::Blob
            }
        };
        #[cfg(not(unix))]
        let kind = EntryKind::Blob;
        (bytes, kind)
    };

    let blob = repo.write_blob(&bytes).context("write blob")?;
    editor
        .upsert(rela_path, kind, blob.detach())
        .with_context(|| format!("tree upsert {rela_path}"))?;
    Ok(())
}

/// Release commits are created locally; pushing is intentionally left to
/// the operator/CI. gix (0.84) implements fetch but not push/send-pack,
/// and nornir forbids both `git` subprocesses and C dependencies
/// (libgit2), so there is no in-process push path. This mirrors
/// [`tag()`], which likewise creates the ref locally and prints the push
/// command. Returns `false` to signal "not pushed" so callers don't
/// report a push that didn't happen.
pub fn push(repo_root: &Path, push_tags: bool) -> Result<bool> {
    let branch = crate::gitio::head_branch(repo_root).unwrap_or_else(|_| "HEAD".to_string());
    eprintln!("  ⏭  push skipped (pure-Rust build has no in-process push).");
    eprintln!("     run: git -C {} push origin {branch}", repo_root.display());
    if push_tags {
        eprintln!("     run: git -C {} push origin --tags", repo_root.display());
    }
    Ok(false)
}

/// Create an annotated `vX.Y.Z` tag pointing at HEAD using the
/// pure-Rust `gix` crate (no `git` shellout). Does *not* push — prints
/// the exact push command the operator/CI should run.
pub fn tag(repo_root: &Path, version: &str) -> Result<()> {
    let tag = format!("v{version}");
    let repo = gix::open(repo_root)
        .with_context(|| format!("gix::open {}", repo_root.display()))?;
    let head_commit = repo.head_commit().context("resolve HEAD commit")?;

    let message = format!("Release {tag}\n");
    let signature = repo
        .author()
        .ok_or_else(|| anyhow!("git author not configured (user.name / user.email)"))?
        .map_err(|e| anyhow!("read git author: {e}"))?;

    repo.tag(
        &tag,
        head_commit.id,
        gix::objs::Kind::Commit,
        Some(signature),
        &message,
        gix::refs::transaction::PreviousValue::MustNotExist,
    )
    .with_context(|| format!("create tag {tag}"))?;

    eprintln!(
        "created local tag {tag} at {}. Push with: git push origin {tag}",
        head_commit.id
    );
    Ok(())
}

// ─── Feature 3: crates.io index propagation poll ────────────────────

/// Whether the crates.io `/api/v1/crates/<name>` payload advertises `version` —
/// either as `crate.max_version` OR as an exact entry in `versions[].num`. Bug
/// #20: keying ONLY on `max_version` means publishing a BACKPORT on an older line
/// (e.g. `1.2.7` while `2.x` is current) never matches and the poll times out.
/// Pure, so it's unit-testable against a recorded payload.
fn index_advertises_version(json: &serde_json::Value, version: &str) -> bool {
    if json.get("crate").and_then(|c| c.get("max_version")).and_then(|v| v.as_str())
        == Some(version)
    {
        return true;
    }
    json.get("versions")
        .and_then(|v| v.as_array())
        .map(|arr| arr.iter().any(|e| e.get("num").and_then(|n| n.as_str()) == Some(version)))
        .unwrap_or(false)
}

/// Block until `crates.io/api/v1/crates/<name>` advertises `version`
/// (as `max_version` OR an exact `versions[].num`), or `timeout` elapses.
/// Returns the wait time in milliseconds (0 if the version was already there) so
/// callers can log it / persist it into `publish_attempts.wait_ms`.
///
/// Replaces "just `sleep 20` between phases" — crates.io's index
/// propagates in 5-90 s depending on load, so a fixed sleep is either
/// too long (slowing the release) or too short (next phase tries to
/// resolve a dep that's not yet visible and fails).
///
/// Uses the already-vendored `ureq` client (no reqwest/tokio in the
/// library footprint) for the poll loop — matches the existing
/// crates.io-probe pattern in the publish flow.
/// crates.io returns HTTP 403 without a non-default User-Agent.
pub fn wait_for_index(
    crate_name: &str,
    version: &str,
    timeout: std::time::Duration,
) -> Result<u64> {
    use std::time::{Duration, Instant};
    let url = format!("https://crates.io/api/v1/crates/{crate_name}");
    let agent = "nornir-release-bot/0.1 (cargo-pipeline; +https://crates.io)";
    let started = Instant::now();
    loop {
        if let Ok(resp) = ureq::get(&url)
            .set("User-Agent", agent)
            .timeout(Duration::from_secs(10))
            .call()
        {
            if let Ok(body) = resp.into_string() {
                if let Ok(json) = serde_json::from_str::<serde_json::Value>(&body) {
                    if index_advertises_version(&json, version) {
                        return Ok(started.elapsed().as_millis() as u64);
                    }
                }
            }
        }
        if started.elapsed() >= timeout {
            return Err(anyhow!(
                "timed out waiting for crates.io index: {crate_name}@{version} after {:?}",
                timeout
            ));
        }
        std::thread::sleep(Duration::from_secs(3));
    }
}

// ─── Feature 5: .crate tarball stats ────────────────────────────────

/// Stats from a `cargo package --list` run. Used by the bloat gate
/// and persisted into the `crate_tarball_stats` iceberg table.
#[derive(Debug, Clone)]
pub struct TarballStats {
    pub crate_name: String,
    pub version: String,
    pub tarball_bytes: u64,
    pub file_count: usize,
    pub largest_file: Option<String>,
    pub largest_file_bytes: Option<u64>,
}

/// Run `cargo package -p <krate> --list --allow-dirty` and inspect
/// the resulting `.crate` tarball in `target/package/`. Does *not*
/// upload anything. Reports file count + size of largest entry so the
/// caller can warn on bloat (typical libraries are <100 kB; >5 MB is
/// almost always test fixtures or planet-scale data leaking in via a
/// missing `exclude = [...]`).
pub fn tarball_stats(repo_root: &Path, krate: &str) -> Result<TarballStats> {
    use std::process::Command;
    let pkg = Command::new(crate::release::cargo::cargo_bin())
        .args(["package", "-p", krate, "--allow-dirty", "--no-verify"])
        .current_dir(repo_root)
        .output()
        .context("spawn cargo package")?;
    if !pkg.status.success() {
        return Err(anyhow!(
            "cargo package -p {krate} exited {}: {}",
            pkg.status,
            String::from_utf8_lossy(&pkg.stderr)
        ));
    }
    let target = repo_root.join("target").join("package");
    // Find the .crate file we just produced; pick the newest matching.
    let mut newest: Option<(std::path::PathBuf, std::time::SystemTime)> = None;
    if let Ok(rd) = std::fs::read_dir(&target) {
        for e in rd.flatten() {
            let p = e.path();
            if p.extension().and_then(|s| s.to_str()) != Some("crate") { continue }
            let fname = p.file_name().and_then(|s| s.to_str()).unwrap_or("");
            if !fname.starts_with(&format!("{krate}-")) { continue }
            let mtime = e.metadata().and_then(|m| m.modified()).unwrap_or(std::time::UNIX_EPOCH);
            if newest.as_ref().map(|(_, t)| mtime > *t).unwrap_or(true) {
                newest = Some((p, mtime));
            }
        }
    }
    let (path, _) = newest.ok_or_else(|| anyhow!("no .crate tarball found in {}", target.display()))?;
    let tarball_bytes = std::fs::metadata(&path)?.len();
    let fname = path.file_name().and_then(|s| s.to_str()).unwrap_or("");
    let version = fname
        .strip_prefix(&format!("{krate}-"))
        .and_then(|s| s.strip_suffix(".crate"))
        .unwrap_or("?")
        .to_string();
    // Walk the tarball to count files + find the largest.
    use flate2::read::GzDecoder;
    use tar::Archive;
    let f = std::fs::File::open(&path)?;
    let dec = GzDecoder::new(f);
    let mut ar = Archive::new(dec);
    let mut file_count = 0usize;
    let mut largest: Option<(String, u64)> = None;
    for entry in ar.entries()? {
        let entry = entry?;
        let size = entry.header().size().unwrap_or(0);
        let p = entry.path()?.into_owned();
        file_count += 1;
        if largest.as_ref().map(|(_, s)| size > *s).unwrap_or(true) {
            largest = Some((p.display().to_string(), size));
        }
    }
    Ok(TarballStats {
        crate_name: krate.to_string(),
        version,
        tarball_bytes,
        file_count,
        largest_file: largest.as_ref().map(|(n, _)| n.clone()),
        largest_file_bytes: largest.map(|(_, s)| s),
    })
}

/// Default bloat threshold: 5 MB. Tweakable per-call.
pub const DEFAULT_TARBALL_BYTES_THRESHOLD: u64 = 5 * 1024 * 1024;

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

    fn m(n: &str, deps: &[&str]) -> (String, Vec<String>) {
        (n.to_string(), deps.iter().map(|s| s.to_string()).collect())
    }

    /// Bug #9: derive_publish_order must EXCLUDE dev-dependency edges. Model a
    /// dev-dep back-edge (`app` normal-depends on `core`; `core` dev-depends on
    /// `app`). Counting the dev edge invents a cycle and could publish `app`
    /// before `core`; filtering it keeps the publish graph a DAG with core first.
    #[test]
    fn members_from_deps_excludes_dev_dependencies() {
        use cargo_metadata::DependencyKind::{Build, Development, Normal};
        let ws: std::collections::BTreeSet<String> =
            ["app", "core"].iter().map(|s| s.to_string()).collect();
        let pkgs = vec![
            ("app".to_string(), true, vec![("core".to_string(), Normal)]),
            // core dev-depends on app (a test-only back-edge) AND build-deps core-macros…
            ("core".to_string(), true, vec![("app".to_string(), Development)]),
        ];
        let members = members_from_deps(&ws, &pkgs);
        // The dev edge core→app is dropped, so core has no deps.
        let core = members.iter().find(|(n, _)| n == "core").unwrap();
        assert!(core.1.is_empty(), "dev-dep back-edge must be excluded: {core:?}");
        let phases = topo_phases(members);
        let flat: Vec<&str> = phases.iter().flatten().map(String::as_str).collect();
        let pos = |c: &str| flat.iter().position(|x| *x == c).unwrap();
        assert!(pos("core") < pos("app"), "core (dep) must publish before app: {phases:?}");
        // No cycle phase: every member resolved in its own ready phase.
        assert_eq!(flat.len(), 2);

        // Build deps are KEPT (must publish first).
        let pkgs2 = vec![
            ("app".to_string(), true, vec![("buildtool".to_string(), Build)]),
            ("buildtool".to_string(), true, vec![]),
        ];
        let ws2: std::collections::BTreeSet<String> =
            ["app", "buildtool"].iter().map(|s| s.to_string()).collect();
        let m2 = members_from_deps(&ws2, &pkgs2);
        let app = m2.iter().find(|(n, _)| n == "app").unwrap();
        assert_eq!(app.1, vec!["buildtool".to_string()], "Build deps must be kept");

        // publish=false members are dropped entirely.
        let pkgs3 = vec![("priv".to_string(), false, vec![])];
        assert!(members_from_deps(&ws, &pkgs3).is_empty());
    }

    /// Bug #20: wait_for_index must match a backport published on an older line
    /// (present in versions[] though max_version points at a newer release).
    #[test]
    fn index_advertises_version_matches_max_or_versions_list() {
        let payload: serde_json::Value = serde_json::json!({
            "crate": { "max_version": "2.0.0" },
            "versions": [ { "num": "2.0.0" }, { "num": "1.2.7" }, { "num": "1.2.6" } ]
        });
        // max_version match.
        assert!(index_advertises_version(&payload, "2.0.0"));
        // backport on the older line — present in versions[] though not max.
        assert!(index_advertises_version(&payload, "1.2.7"));
        // genuinely-absent version.
        assert!(!index_advertises_version(&payload, "1.2.8"));
    }

    #[test]
    fn topo_phases_orders_deps_before_dependents_deterministically() {
        // testmatrix ← robotui ← nornir ; airgap ← nornir ; testmatrix ← nornir.
        let members = vec![
            m("nornir", &["nornir-testmatrix", "nornir-robotui", "skidbladnir"]),
            m("nornir-robotui", &["nornir-testmatrix"]),
            m("nornir-testmatrix", &[]),
            m("skidbladnir", &[]),
        ];
        let phases = topo_phases(members);
        let flat: Vec<&str> = phases.iter().flatten().map(String::as_str).collect();
        let pos = |c: &str| flat.iter().position(|x| *x == c).unwrap();
        // deps strictly before dependents
        assert!(pos("nornir-testmatrix") < pos("nornir-robotui"));
        assert!(pos("nornir-robotui") < pos("nornir"));
        assert!(pos("skidbladnir") < pos("nornir"));
        assert!(pos("nornir-testmatrix") < pos("nornir"));
        // deterministic: the two leaves share phase 0 in the BTreeSet's
        // alphabetical iteration order ("nornir-testmatrix" < "skidbladnir").
        assert_eq!(phases[0], vec!["nornir-testmatrix", "skidbladnir"]);
        // nornir is last (its own phase)
        assert_eq!(phases.last().unwrap(), &vec!["nornir".to_string()]);
    }

    #[test]
    fn topo_phases_cycle_drops_nothing() {
        let members = vec![m("a", &["b"]), m("b", &["a"])]; // pathological cycle
        let phases = topo_phases(members);
        let flat: std::collections::BTreeSet<&str> =
            phases.iter().flatten().map(String::as_str).collect();
        assert!(flat.contains("a") && flat.contains("b"), "cycle members not dropped");
    }

    #[test]
    fn parse_rate_limit_wait_parses_relative_seconds() {
        // crates.io's real 429 body, relative form.
        let msg = "error: failed to publish to registry at https://crates.io\n\
                   Caused by: the remote server responded with an error (status 429 Too \
                   Many Requests): You have published too many crates in a short period \
                   of time, please try again after 56 seconds, or email help@crates.io";
        let got = parse_rate_limit_wait(msg, std::time::Duration::from_secs(60))
            .expect("recognised as rate limit");
        assert_eq!(got, std::time::Duration::from_secs(56), "parsed the 56s wait");
    }

    #[test]
    fn parse_rate_limit_wait_parses_relative_minutes_and_falls_back() {
        let mins = "status 429 ... too many crates ... please try again in 4 minutes";
        assert_eq!(
            parse_rate_limit_wait(mins, std::time::Duration::from_secs(60)),
            Some(std::time::Duration::from_secs(240)),
            "4 minutes → 240s"
        );
        // Absolute HTTP-date form (crates.io's new-crate limiter) is now PARSED,
        // not defaulted. This deadline is in the past (before today), so the wait
        // collapses to just the buffer — NOT the 42s default.
        let abs = "too many crates in a short period of time, please try again after \
                   Wed, 18 Jun 2026 12:34:56 GMT";
        let got = parse_rate_limit_wait(abs, std::time::Duration::from_secs(42))
            .expect("recognised as rate limit");
        assert!(
            got <= std::time::Duration::from_secs(20),
            "a past absolute deadline waits only the ~15s buffer, got {got:?}"
        );
    }

    #[test]
    fn http_date_to_unix_matches_known_epoch() {
        // Cross-checked against `date -u -d '...' +%s`.
        assert_eq!(super::http_date_to_unix("Fri, 03 Jul 2026 05:36:18 GMT"), Some(1_783_056_978));
        assert_eq!(super::http_date_to_unix("03 Jul 2026 05:36:18 GMT"), Some(1_783_056_978)); // no weekday
        assert_eq!(super::http_date_to_unix("Wed, 01 Jan 2020 00:00:00 GMT"), Some(1_577_836_800));
        assert_eq!(super::http_date_to_unix("garbage"), None);
        assert_eq!(super::http_date_to_unix("Fri, 32 Jul 2026 05:36:18 GMT"), None); // bad day
    }

    #[test]
    fn wait_until_http_date_waits_until_the_deadline() {
        let msg = "status 429 Too Many Requests: You have published too many new crates \
                   in a short period of time. Please try again after Fri, 03 Jul 2026 \
                   05:36:18 GMT and see https://crates.io/docs/rate-limits";
        let deadline = 1_783_056_978u64;
        // 10 minutes before the window → wait ≈ 600s + 15s buffer.
        let got = super::wait_until_http_date(msg, deadline - 600).expect("absolute deadline");
        assert_eq!(got, std::time::Duration::from_secs(615));
        // Already past the deadline → just the buffer, never negative.
        let past = super::wait_until_http_date(msg, deadline + 5_000).expect("absolute deadline");
        assert_eq!(past, std::time::Duration::from_secs(15));
        // No absolute deadline present → None (lets the relative/default path win).
        assert_eq!(super::wait_until_http_date("please try again in 4 minutes", 0), None);
    }

    /// Task 2: the publish-failure classifier must map each real cargo stderr to
    /// the right class + actionable explanation. Uses the ACTUAL strings this run
    /// hit (skade's "did not match any packages" and the facett-graphnav →
    /// facett_graphview::analysis / facett_core::Elm verify skew).
    #[test]
    fn classify_publish_failure_covers_each_class() {
        let root = std::path::Path::new("/home/rickard/git/skade");
        let no_crates = std::collections::BTreeSet::new();

        // (1) detached-workspace non-member — skade's exact error.
        let non_member = "error: package ID specification `skade` did not match any packages";
        assert_eq!(classify_publish_failure("skade", non_member, &no_crates), PublishFailure::NonMember);
        let e = explain_publish_failure("skade", root, non_member).unwrap();
        assert!(e.contains("member of the workspace"), "{e}");
        assert!(e.contains("publish = false"), "names the fix: {e}");

        // (2) dep-API skew — the real facett-graphnav verify failure. With NO
        // known-crate set the legacy extraction order applies: the `in `X`` form
        // wins, naming `facett_core`.
        let skew = "   Compiling facett-graphnav v0.4.53\n\
                     error[E0432]: unresolved import `facett_graphview::analysis`\n\
                     error[E0433]: failed to resolve: could not find `Elm` in `facett_core`\n\
                     error: could not compile `facett-graphnav` due to 2 previous errors\n\
                     error: failed to verify package tarball";
        match classify_publish_failure("facett-graphnav", skew, &no_crates) {
            PublishFailure::DepApiSkew { dep } => {
                // legacy order (empty set): the culprit is named from
                // "could not find `Elm` in `facett_core`".
                assert_eq!(dep.as_deref(), Some("facett_core"));
            }
            other => panic!("expected DepApiSkew, got {other:?}"),
        }
        let e = explain_publish_failure("facett-graphnav", root, skew).unwrap();
        assert!(e.contains("facett_core"), "names the dep: {e}");
        assert!(e.contains("bump"), "advises bumping the dep first: {e}");

        // (3) missing metadata — name the field(s).
        let meta = "error: failed to publish\n\
                    Caused by: missing or empty metadata fields: description, license. \
                    Please see https://doc.rust-lang.org/cargo/reference/manifest.html";
        match classify_publish_failure("solo", meta, &no_crates) {
            PublishFailure::MissingMetadata { fields } => {
                assert!(fields.contains(&"description".to_string()));
                assert!(fields.contains(&"license".to_string()));
            }
            other => panic!("expected MissingMetadata, got {other:?}"),
        }
        assert!(explain_publish_failure("solo", root, meta).unwrap().contains("description"));

        // (4) rate-limit is its own class (retry loop owns the messaging).
        let rl = "status 429 Too Many Requests: You have published too many crates in a \
                  short period of time, please try again after 30 seconds";
        assert_eq!(classify_publish_failure("x", rl, &no_crates), PublishFailure::RateLimited);
        assert!(explain_publish_failure("x", root, rl).is_none(), "no double-explain for 429");

        // (5) pure resolution/ordering — a sibling not published yet — is NOT a
        // code defect and must NOT be mistaken for a dep-API skew (no false preflight block).
        let ordering = "error: failed to select a version for the requirement `facett-core = \"^0.4.53\"`\n\
                        candidate versions found which didn't match: 0.4.52\n\
                        location searched: crates.io index";
        assert_eq!(
            classify_publish_failure("facett-graphview", ordering, &no_crates),
            PublishFailure::ResolveOrdering
        );

        // unclassified → Other, no explanation.
        assert_eq!(classify_publish_failure("x", "some unknown cargo hiccup", &no_crates), PublishFailure::Other);
        assert!(explain_publish_failure("x", root, "some unknown cargo hiccup").is_none());
    }

    /// P1#5: a `could not find X in `kind`` blames the MODULE `kind`, not the
    /// crate. With the workspace-crate set threaded in, the crate-qualified
    /// import-path segment (`nornir_jobs`) is preferred and validated, and the
    /// bare module `kind` is rejected — so the blame names the crate the cascade
    /// can actually republish (the dwarves↔nornir-jobs class).
    #[test]
    fn extract_skew_dep_prefers_crate_over_module_with_known_set() {
        let known: std::collections::BTreeSet<String> =
            ["nornir-jobs", "dwarves"].iter().map(|s| s.to_string()).collect();
        // import path names the CRATE (lib-name underscores); `in `kind`` names a module.
        let skew = "error[E0432]: unresolved import `nornir_jobs::runner`\n\
                    error[E0433]: failed to resolve: could not find `Spawn` in `kind`\n\
                    error: could not compile `dwarves`";
        match classify_publish_failure("dwarves", skew, &known) {
            PublishFailure::DepApiSkew { dep } => {
                assert_eq!(dep.as_deref(), Some("nornir_jobs"), "crate wins over module `kind`");
            }
            other => panic!("expected DepApiSkew, got {other:?}"),
        }
        // Even when ONLY the module form is present, the unknown `kind` is not
        // silently trusted; with no import path to fall back to it degrades but
        // never mis-blames a known crate.
        let only_mod = "error[E0433]: failed to resolve: could not find `Spawn` in `kind`";
        match classify_publish_failure("dwarves", only_mod, &known) {
            PublishFailure::DepApiSkew { dep } => {
                assert_ne!(dep.as_deref(), Some("nornir-jobs"), "must not invent a crate");
            }
            other => panic!("expected DepApiSkew, got {other:?}"),
        }
    }

    #[test]
    fn parse_rate_limit_wait_ignores_non_rate_limit_errors() {
        assert_eq!(
            parse_rate_limit_wait(
                "error: failed to verify package tarball\nexpected `foo`",
                std::time::Duration::from_secs(60)
            ),
            None,
            "a normal cargo failure is NOT a rate limit"
        );
    }

    #[test]
    fn parse_rate_limit_wait_caps_at_max() {
        let huge = "429 too many crates, please try again in 600 minutes"; // 10h
        let got = parse_rate_limit_wait(huge, std::time::Duration::from_secs(60)).unwrap();
        assert_eq!(got, MAX_RATE_LIMIT_WAIT, "absurd wait is clamped to the ceiling");
    }

    #[test]
    fn run_cargo_publish_retries_after_rate_limit_then_succeeds() {
        let _g = crate::serial_guard(); // mutates global PATH (fake cargo) — serialize
        // Inject-assert: feed the real 429 string the FIRST two cargo invocations,
        // then a success, through a fake `cargo` on PATH. Assert the loop slept the
        // PARSED waits (3s, then 3s) and that the final outcome is Published.
        use std::io::Write;
        let dir = std::env::temp_dir().join(format!("nornir-rl-test-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let counter = dir.join("count");
        std::fs::write(&counter, "0").unwrap();
        // Fake cargo: 1st+2nd call print the 429 body + exit 1; 3rd call exits 0.
        let fake = dir.join("cargo");
        let script = format!(
            "#!/bin/sh\n\
             c=$(cat '{cnt}')\n\
             c=$((c+1)); echo $c > '{cnt}'\n\
             if [ \"$c\" -lt 3 ]; then\n\
               echo 'status 429 Too Many Requests: You have published too many crates in a short period of time, please try again after 3 seconds' 1>&2\n\
               exit 1\n\
             fi\n\
             exit 0\n",
            cnt = counter.display()
        );
        {
            let mut f = std::fs::File::create(&fake).unwrap();
            f.write_all(script.as_bytes()).unwrap();
        }
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            std::fs::set_permissions(&fake, std::fs::Permissions::from_mode(0o755)).unwrap();
        }
        // `cargo_bin()` resolves `$CARGO` FIRST (PATH is not consulted when it's set,
        // and `cargo test` always sets it), so inject the fake cargo via `$CARGO`.
        let old_cargo = std::env::var_os("CARGO");
        // SAFETY: single-threaded (serial_guard); CARGO restored before return.
        unsafe {
            std::env::set_var("CARGO", &fake);
        }

        let mut slept: Vec<std::time::Duration> = Vec::new();
        let outcome = run_cargo_publish_with_retry(
            &dir,
            "demo-crate",
            false,
            5,
            std::time::Duration::from_secs(60),
            |d| slept.push(d),
        );

        match &old_cargo {
            Some(c) => unsafe { std::env::set_var("CARGO", c) },
            None => unsafe { std::env::remove_var("CARGO") },
        }
        let calls: u32 = std::fs::read_to_string(&counter)
            .ok()
            .and_then(|s| s.trim().parse().ok())
            .unwrap_or(0);
        let _ = std::fs::remove_dir_all(&dir);

        let outcome = outcome.expect("publish should succeed after retries");
        assert_eq!(outcome, PublishOutcome::Published, "succeeded on the 3rd attempt");
        assert_eq!(
            slept,
            vec![
                std::time::Duration::from_secs(3),
                std::time::Duration::from_secs(3)
            ],
            "slept the PARSED 3s wait before each of the 2 retries"
        );
        assert_eq!(calls, 3, "cargo was invoked 3× (2 rate-limited + 1 success)");
    }

    // LIVE integration smoke against nornir's own workspace. Shells out to
    // `cargo metadata`, which flakes under the FULL parallel suite + concurrent cargo
    // (returns an incomplete member set under sustained load) though it passes 100%
    // alone. The publish-order LOGIC is covered by the pure `topo_phases_*` tests
    // above, so this is `#[ignore]`d out of the default `-j` run — invoke explicitly:
    // `cargo test --features server derive_publish_order_from_this_real_workspace -- --ignored`.
    #[test]
    #[ignore = "live cargo-metadata smoke; flakes under the parallel suite — run explicitly"]
    fn derive_publish_order_from_this_real_workspace() {
        let _g = crate::serial_guard(); // serialize the cargo-metadata-spawning tests
        // Deterministic derivation against nornir's OWN workspace: testmatrix must
        // come before nornir (nornir depends on it), and nornir is the last crate.
        // Use the COMPILE-TIME crate dir, not `"."` (cwd-race-immune). `derive_publish_
        // order` shells out to `cargo metadata`, which can transiently race the OTHER
        // cargo-spawning tests in the suite (Cargo.lock writes under `-j`) and return an
        // incomplete member set — so retry a few times (isolation passes first try).
        let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR"));
        let mut flat: Vec<String> = Vec::new();
        for _ in 0..8 {
            if let Ok(order) = derive_publish_order(root) {
                flat = order.into_iter().flatten().collect();
                let has = |c: &str| flat.iter().any(|x| x == c);
                if has("nornir") && has("nornir-testmatrix") {
                    break;
                }
            }
            std::thread::sleep(std::time::Duration::from_millis(250));
        }
        let pos = |c: &str| flat.iter().position(|x| x == c);
        assert!(flat.contains(&"nornir".to_string()), "nornir is publishable: {flat:?}");
        assert!(
            flat.contains(&"nornir-testmatrix".to_string()),
            "testmatrix is publishable: {flat:?}"
        );
        assert!(
            pos("nornir-testmatrix") < pos("nornir"),
            "testmatrix publishes before nornir: {flat:?}"
        );
    }
}