nornir 0.5.3

//! Pre-cascade validation for `release doctor --publish`.
//!
//! `cargo publish` is IRREVERSIBLE: once a `<crate>@<version>` is on crates.io it
//! is immutable forever. A cascade that fail-stops halfway leaves the already-shipped
//! crates permanently live while the rest never ship — a torn release. Preflight
//! moves every catchable failure BEFORE the first real upload, so the operator sees
//! ALL blockers at once ("preflight: N blocking issue(s)") with the exact fix per
//! issue, and nothing goes to the registry until the plan is clean.
//!
//! Checks (all best-effort — a check whose tooling is unavailable degrades to a
//! note, never a false block):
//!   1. every planned crate is a MEMBER of its repo-root workspace (the detached
//!      `[workspace]` bug — a bench/vendored/sub-workspace crate handed to
//!      `cargo publish -p` from the wrong root errors 101 "did not match any packages");
//!   2. required crates.io METADATA present per crate (description, license/license-file,
//!      repository);
//!   3. the "dependency edited without a version bump" trap, caught UP FRONT via
//!      `cargo publish -p <crate> --dry-run` (verify build against the registry) — a
//!      compile error citing an already-published sibling is a real blocker; a mere
//!      "sibling not on the registry yet" is expected mid-cascade and is NOT flagged.
//!      BUMP-AWARE (item #7): when the cascade will itself publish the blamed
//!      sibling forward (it is in the planned publish set, e.g. every crate under
//!      `--bump patch`), the verify skew is cascade-resolvable — the cascade bumps
//!      and publishes that sibling BEFORE its dependents — so it is a NOTE, not a
//!      blocker. Only a skew blaming a crate the cascade will NOT republish
//!      (external / not in the release set) still blocks. This is what lets
//!      `doctor --publish --bump patch` pass preflight WITHOUT `--force`;
//!   4. DETACHED `[workspace]` roots listed, so the operator knows they're intentionally
//!      excluded;
//!   5. VERSION/registry consistency — for a real bump, the target version must not
//!      already be live on crates.io (an immutable clash); for `--bump keep`, a version
//!      already on the registry is a benign idempotent skip.

use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};

use crate::release::cargo;
use crate::release::publish::{classify_publish_failure, PublishFailure};
use crate::release::registry::{classify, RegistryBackend, RegistryState};

/// The class of a blocking preflight issue.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum IssueKind {
    /// Planned `-p` target is not a member of its repo-root workspace.
    NonMember,
    /// Missing required crates.io metadata field(s).
    MissingMetadata,
    /// `cargo publish --dry-run` verify build failed on dep-API skew.
    VerifyFailed,
    /// The version to publish is already live on crates.io (immutable clash).
    RegistrySkew,
    /// The LOCAL version is BEHIND the registry's published max (a desync/
    /// downgrade) — a dependent will verify against the newer published API and
    /// break. See [`crate::release::registry`].
    RegistryDesync,
    /// A `[dev-dependencies]` entry carries BOTH `path =` and `version =` on a
    /// workspace sibling — cargo strips dev-deps on publish but still RESOLVES a
    /// versioned one during verify, fail-stopping if the sibling isn't live (P1#2).
    VersionedPathDevDep,
    /// A dependency EDGE that can never be satisfied on publish: a NORMAL/BUILD
    /// `path =` dep with no `version =` (stripped → unresolvable), or a versioned
    /// dep on a `publish = false` sibling (never reaches crates.io).
    UnpublishableDep,
    /// A crates.io manifest constraint beyond the required fields: a `readme`
    /// pointing at a missing file, an invalid `keywords` set, or a `[features]`
    /// entry referencing a `dep:foo` absent from `[dependencies]`.
    BadManifest,
    /// The crate NAME violates crates.io's rules: >64 chars, a non-ASCII-letter
    /// first character, a character outside `[A-Za-z0-9_-]`, or a reserved
    /// (Windows-device) name. crates.io rejects the upload; `cargo publish
    /// --dry-run` does not check it.
    InvalidCrateName,
    /// The packaged `.crate` tarball exceeds crates.io's upload size limit (10 MiB
    /// by default) — the upload is rejected ("max upload size is: N").
    OversizedTarball,
    /// The crate's LOCAL source DIFFERS from the copy already PUBLISHED at the
    /// SAME version — "edited without a version bump". A dependent's verify build
    /// resolves the stale registry copy and fail-stops. A CORRECTNESS blocker
    /// (like [`IssueKind::VerifyFailed`]): `--force` must NOT wave it through.
    StaleSourceDrift,
}

impl IssueKind {
    /// The CORRECTNESS class: a dependent would build/verify against an API that
    /// is NOT in the published dependency. `VerifyFailed` is that skew caught by
    /// the dry-run compile; `StaleSourceDrift` is the SAME skew caught by content-
    /// diff of an edited-but-unbumped crate (the "edited-but-AlreadyPublished dep a
    /// to-publish crate depends on" trap). NEVER overridable by `--force` — forcing
    /// past it can only fail-stop mid-cascade or ship a version that can't build.
    /// Single-sourced here so the dry-run and the real publish classify a blocker
    /// IDENTICALLY.
    pub fn is_correctness_blocker(&self) -> bool {
        matches!(self, IssueKind::VerifyFailed | IssueKind::StaleSourceDrift)
    }
}

/// One blocking issue: a short, specific "why" plus the exact "fix".
#[derive(Debug, Clone)]
pub struct Issue {
    pub repo: String,
    pub krate: String,
    pub kind: IssueKind,
    pub detail: String,
    pub fix: String,
    /// Crate(s) this issue's error explicitly BLAMES as the un-published
    /// upstream — set for `VerifyFailed` (the dep whose new API isn't on the
    /// registry). Feeds topological root-cause analysis so the true root is
    /// named even when it was skipped and never became its own issue. Empty
    /// for self-contained issues (metadata/version/non-member).
    #[allow(dead_code)]
    pub blames: Vec<String>,
}

/// Aggregated preflight verdict: blocking `issues` (must be empty to proceed) plus
/// informational `notes` (detached workspaces excluded, checks skipped, etc.).
#[derive(Debug, Default)]
pub struct PreflightReport {
    pub issues: Vec<Issue>,
    pub notes: Vec<String>,
    pub checked_crates: usize,
}

impl PreflightReport {
    /// Clean ⟺ zero blocking issues ⟹ safe to start the irreversible cascade.
    pub fn is_clean(&self) -> bool {
        self.issues.is_empty()
    }

    fn push(&mut self, i: Issue) {
        self.issues.push(i);
    }

    /// Human-readable consolidated summary — every issue with its fix, then the
    /// notes. The header is `preflight: clean` or `preflight: N blocking issue(s)`.
    pub fn format(&self) -> String {
        let mut s = String::new();
        if self.is_clean() {
            s.push_str(&format!(
                "\npreflight: clean — {} crate(s) checked; safe to publish.\n",
                self.checked_crates
            ));
        } else {
            s.push_str(&format!(
                "\npreflight: {} blocking issue(s) — resolve before publishing \
                 (or re-run with --force to override):\n",
                self.issues.len()
            ));
            for i in &self.issues {
                let tag = match i.kind {
                    IssueKind::NonMember => "non-member",
                    IssueKind::MissingMetadata => "metadata",
                    IssueKind::VerifyFailed => "verify",
                    IssueKind::RegistrySkew => "version",
                    IssueKind::RegistryDesync => "desync",
                    IssueKind::VersionedPathDevDep => "dev-dep",
                    IssueKind::UnpublishableDep => "dep",
                    IssueKind::BadManifest => "manifest",
                    IssueKind::StaleSourceDrift => "source-drift",
                    IssueKind::InvalidCrateName => "name",
                    IssueKind::OversizedTarball => "size",
                };
                s.push_str(&format!("  ⛔ [{tag}] {}/{}: {}\n", i.repo, i.krate, i.detail));
                s.push_str(&format!("       fix: {}\n", i.fix));
            }
        }
        // RELEASE READINESS: consolidate the RegistryDesync blockers (the common
        // "a sibling tree is behind its published crates.io version" class) into ONE
        // copy-pasteable republish command PER REPO, so the user isn't left
        // assembling it from N identical per-crate lines. This is the "support the
        // user doing a release" answer: exactly which crates + the exact command.
        let desynced: Vec<&Issue> =
            self.issues.iter().filter(|i| i.kind == IssueKind::RegistryDesync).collect();
        if !desynced.is_empty() {
            let mut by_repo: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
            for i in &desynced {
                by_repo.entry(i.repo.as_str()).or_default().push(i.krate.as_str());
            }
            s.push_str(&format!(
                "\nRELEASE READINESS: {} crate(s) across {} repo(s) are BEHIND their \
                 published crates.io version — a version-bump republish is needed (a REAL, \
                 user-driven publish; not this dry-run):\n",
                desynced.len(),
                by_repo.len()
            ));
            for (repo, crates) in &by_repo {
                let only = crates.iter().map(|c| format!("--only {c}")).collect::<Vec<_>>().join(" ");
                s.push_str(&format!(
                    "  {repo}: {} crate(s) — if the local tree is merely STALE, `git pull` in \
                     {repo}; otherwise bump+republish just these:\n    \
                     nornir release doctor --publish --bump patch {only}\n",
                    crates.len()
                ));
            }
        }
        for n in &self.notes {
            s.push_str(&format!("  · {n}\n"));
        }
        s
    }
}

/// A repo to preflight: its config name, on-disk root, and the crate set that WILL
/// be published (publishable minus gate-held). The caller derives `publish_set`
/// exactly as the promote path does, so preflight checks precisely what ships.
#[derive(Debug, Clone)]
pub struct RepoPlan {
    pub name: String,
    pub root: PathBuf,
    /// Crate names slated to publish, with the version each will carry.
    pub publish_set: Vec<(String, String)>,
}

/// Injectable registry probe so the version/consistency check is deterministically
/// testable without hitting crates.io. `Some(true)` = version live, `Some(false)` =
/// absent, `None` = couldn't determine (network down) → the check degrades to a note.
pub trait RegistryProbe {
    fn version_live(&self, krate: &str, version: &str) -> Option<bool>;

    /// Whether crates.io ALREADY serves a NON-yanked published version satisfying
    /// `req` (a semver [`semver::VersionReq`] string, e.g. `"0.1"` ⇒ `^0.1`). This
    /// answers "can the consumer resolve this dep from the registry as-is?" — the
    /// basis for treating a cross-repo edge as non-order-gating. `Some(false)` = no
    /// live version matches (must publish first); `None` = couldn't determine
    /// (network / unparsable `req`) → the caller stays conservative and keeps the
    /// edge gating. The default returns `None` (a probe that only knows exact-version
    /// liveness cannot answer); [`CratesIoProbe`] overrides it.
    fn req_satisfied(&self, _krate: &str, _req: &str) -> Option<bool> {
        None
    }
}

/// Live crates.io probe (sparse index) reusing the vendored `ureq` client — same
/// pattern as [`crate::release::publish::wait_for_index`].
pub struct CratesIoProbe;

impl CratesIoProbe {
    /// The published versions of `krate` as `(num, yanked)`. `None` on any
    /// network/parse failure (the callers degrade to a conservative answer).
    fn fetch_versions(krate: &str) -> Option<Vec<(String, bool)>> {
        let url = format!("https://crates.io/api/v1/crates/{krate}");
        let agent = "nornir-release-bot/0.1 (cargo-pipeline; +https://crates.io)";
        let resp = ureq::get(&url)
            .set("User-Agent", agent)
            .timeout(std::time::Duration::from_secs(10))
            .call()
            .ok()?;
        let body = resp.into_string().ok()?;
        let json: serde_json::Value = serde_json::from_str(&body).ok()?;
        let versions = json.get("versions").and_then(|v| v.as_array())?;
        Some(
            versions
                .iter()
                .filter_map(|e| {
                    let num = e.get("num").and_then(|n| n.as_str())?.to_string();
                    let yanked = e.get("yanked").and_then(|y| y.as_bool()).unwrap_or(false);
                    Some((num, yanked))
                })
                .collect(),
        )
    }
}

impl RegistryProbe for CratesIoProbe {
    fn version_live(&self, krate: &str, version: &str) -> Option<bool> {
        let versions = Self::fetch_versions(krate)?;
        Some(versions.iter().any(|(num, _yanked)| num == version))
    }

    fn req_satisfied(&self, krate: &str, req: &str) -> Option<bool> {
        // An unparsable requirement can't be reasoned about → conservative (gating).
        let req = semver::VersionReq::parse(req).ok()?;
        let versions = Self::fetch_versions(krate)?;
        Some(versions.iter().any(|(num, yanked)| {
            !*yanked
                && semver::Version::parse(num).map(|v| req.matches(&v)).unwrap_or(false)
        }))
    }
}

/// Whether cargo will run a dry-run verify build during preflight.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum VerifyMode {
    /// Run `cargo publish -p <crate> --dry-run` for each crate and flag dep-API skew.
    On,
    /// Skip the verify build (fast preflight — used only where correctness parity
    /// is not required; NOT the `--publish-dry-run` path — see
    /// [`preflight_verify_mode`]).
    Off,
}

/// The preflight verify mode for the `release doctor --publish` gate, given
/// whether the operator passed `--publish-dry-run`.
///
/// It is **[`VerifyMode::On`] for BOTH** the dry-run and the real publish — that
/// is the whole point. The dep-API-skew verify build and the same-version
/// source-drift pass are the CORRECTNESS class ([`IssueKind::is_correctness_blocker`]);
/// gating them behind "real publish only" made a `--publish-dry-run` print a FALSE
/// all-clear while the real `--publish` fail-stopped on the identical blocker (the
/// draupnir "edited-but-AlreadyPublished dep" trap). Running the FULL preflight in
/// both modes makes a dry-run report EXACTLY the blockers the real run would hit,
/// up front, before any artifact/mutation.
///
/// `_publish_dry_run` is intentionally ignored: the parity invariant is that the
/// mode does NOT depend on it. Kept as a parameter so the call site reads as a
/// deliberate parity decision, and so a regression that reintroduces the split has
/// an obvious place to (wrongly) branch — the unit test guards against it.
pub fn preflight_verify_mode(_publish_dry_run: bool) -> VerifyMode {
    VerifyMode::On
}

/// The up-front gate verdict over a finished [`PreflightReport`]: decide whether
/// the doctor may proceed to the irreversible artifact/publish cascade, or must
/// EARLY-FAIL before any mutation (no bump, no bump-commit, no release/staging
/// branch, no `cargo package`, no `cargo publish`, no push).
///
/// IDENTICAL for `--publish-dry-run` and the real `--publish`: the dry-run reports
/// the SAME verdict the real run would act on. A CORRECTNESS blocker
/// ([`IssueKind::is_correctness_blocker`]) is NEVER forceable; any other (soft)
/// blocker aborts unless `--force` is given.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum GateVerdict {
    /// No blocker (clean), or `--force` waved through `forced_soft` SOFT issues.
    /// Safe to enter the cascade.
    Proceed { forced_soft: usize },
    /// One or more CORRECTNESS blockers (dependents use un-published API). NEVER
    /// forceable — abort BEFORE any artifact/mutation. Carries the blocker count.
    BlockCorrectness(usize),
    /// Soft blocker(s) present and `--force` NOT given — abort before mutation.
    /// Carries the total blocking-issue count.
    BlockSoft(usize),
}

/// Compute the [`GateVerdict`] for a finished preflight `report` and the operator's
/// `--force` flag. Pure over the report — used identically by the dry-run and the
/// real publish so the two can never diverge on what blocks.
pub fn gate_verdict(report: &PreflightReport, force: bool) -> GateVerdict {
    if report.is_clean() {
        return GateVerdict::Proceed { forced_soft: 0 };
    }
    let hard = report.issues.iter().filter(|i| i.kind.is_correctness_blocker()).count();
    if hard > 0 {
        return GateVerdict::BlockCorrectness(hard);
    }
    if force {
        GateVerdict::Proceed { forced_soft: report.issues.len() }
    } else {
        GateVerdict::BlockSoft(report.issues.len())
    }
}

/// Run the full preflight over `plans`. `bump_keep` is true for `--bump keep|none`
/// (idempotent resume: a version already on the registry is a benign skip, not a
/// clash). `verify` gates the expensive dry-run compile. `probe` answers "is this
/// version already live?"; `backend` answers "what is the max PUBLISHED version?"
/// for the registry-state desync check (item #10).
///
/// `cascade_publishes` is the BUMP-AWARE planned map (item #7): every crate the
/// cascade WILL publish forward, mapped to the version it will carry. A dry-run
/// verify skew (check 3) that blames a crate in THIS map is cascade-resolvable —
/// the cascade bumps and publishes that crate before its dependents — so it is
/// recorded as a NOTE, not a blocker. A skew blaming a crate NOT in the map
/// (external, or a crate the cascade won't republish) still blocks. Pass an empty
/// map to get the strict, non-bump-aware behavior (used by `--bump keep`).
pub fn run(
    plans: &[RepoPlan],
    bump_keep: bool,
    verify: VerifyMode,
    probe: &dyn RegistryProbe,
    backend: &dyn RegistryBackend,
    cascade_publishes: &BTreeMap<String, String>,
) -> PreflightReport {
    let mut report = PreflightReport::default();
    // The WHOLE publish plan (every crate → the version it will carry), across all
    // repos, regardless of bump mode. Distinct from `cascade_publishes` (empty for
    // `--bump keep`): a dry-run resolution failure blaming a crate IN this plan is
    // the expected deps-first-order condition (a note), whereas one blaming a crate
    // NOT in the plan is a genuine fail-stop (P2 structural).
    let plan_publishes: BTreeMap<String, String> =
        plans.iter().flat_map(|p| p.publish_set.iter().cloned()).collect();

    // (links) crates.io requires a `[package].links` value to be GLOBALLY unique. A
    // foreign claim is backend-only, but TWO crates in THIS release sharing a links
    // value is a guaranteed collision — flag it up front (the second upload 400s).
    for issue in duplicate_links_issues(plans) {
        report.push(issue);
    }

    for plan in plans {
        // (4) list detached `[workspace]` roots so exclusions are visible.
        let members = cargo::root_workspace_members(&plan.root);

        // `publish = false` siblings in THIS repo — a publishable crate that
        // depends (non-dev, with a version) on one of these can never publish
        // (the sibling never reaches crates.io). Gathered once per repo.
        let mut publish_false: BTreeSet<String> = BTreeSet::new();
        let _ = cargo::for_each_cargo_toml(&plan.root, &mut |doc: &toml::Value| {
            let name = doc
                .get("package")
                .and_then(|p| p.get("name"))
                .and_then(|n| n.as_str());
            let is_false = matches!(
                doc.get("package").and_then(|p| p.get("publish")),
                Some(toml::Value::Boolean(false))
            ) || matches!(
                doc.get("package").and_then(|p| p.get("publish")),
                Some(toml::Value::Array(a)) if a.is_empty()
            );
            if let (Some(n), true) = (name, is_false) {
                publish_false.insert(n.to_string());
            }
        });
        if let Some(members) = &members {
            let detached = detached_workspace_crates(&plan.root, members);
            if !detached.is_empty() {
                report.notes.push(format!(
                    "{}: {} detached-[workspace] crate(s) excluded from publish: {}",
                    plan.name,
                    detached.len(),
                    detached.into_iter().collect::<Vec<_>>().join(", ")
                ));
            }
        } else {
            report.notes.push(format!(
                "{}: cargo metadata unavailable — root-member check skipped (best-effort)",
                plan.name
            ));
        }

        for (krate, version) in &plan.publish_set {
            report.checked_crates += 1;

            // (name) the crate NAME itself must satisfy crates.io's rules (length,
            // charset, first-char, reserved). A pure string check — cheap, always on.
            if let Some(reason) = crate_name_problem(krate) {
                report.push(Issue {
                    repo: plan.name.clone(),
                    krate: krate.clone(),
                    kind: IssueKind::InvalidCrateName,
                    detail: reason,
                    fix: "publish under a valid crate name (keep the crate's lib name via \
                          `[lib] name = \"…\"` / `package = \"…\"` if you only need to rename \
                          the published name)"
                        .to_string(),
                    blames: Vec::new(),
                });
            }

            // (version) the target version must be valid semver — crates.io rejects
            // "`X` is an invalid semver version". Cheap defense-in-depth (cargo also
            // requires it, but a `--bump keep` resume ships the manifest value as-is).
            if semver::Version::parse(version).is_err() {
                report.push(Issue {
                    repo: plan.name.clone(),
                    krate: krate.clone(),
                    kind: IssueKind::BadManifest,
                    detail: format!("version `{version}` is not valid semver"),
                    fix: "set a valid semver version (MAJOR.MINOR.PATCH, e.g. 1.2.3)".to_string(),
                    blames: Vec::new(),
                });
            }

            // (1) member of the repo-root workspace?
            if let Some(members) = &members {
                if !members.contains(krate) {
                    report.push(Issue {
                        repo: plan.name.clone(),
                        krate: krate.clone(),
                        kind: IssueKind::NonMember,
                        detail: format!(
                            "not a member of the workspace at {} (detached [workspace])",
                            plan.root.display()
                        ),
                        fix: "mark it `publish = false`, or publish it from its own \
                              workspace directory"
                            .to_string(),
                        blames: Vec::new(),
                    });
                    // A non-member can't be metadata/verify-checked from this root; skip.
                    continue;
                }
            }

            // (2) required crates.io metadata.
            let missing = crate_metadata_gaps(&plan.root, krate);
            if !missing.is_empty() {
                report.push(Issue {
                    repo: plan.name.clone(),
                    krate: krate.clone(),
                    kind: IssueKind::MissingMetadata,
                    detail: format!("missing crates.io metadata: {}", missing.join(", ")),
                    fix: format!(
                        "add the field(s) to [package] in {}'s Cargo.toml",
                        krate
                    ),
                    blames: Vec::new(),
                });
            }

            // (SPDX) invalid `license` expression — crates.io rejects it with
            // "unknown or invalid license expression". Resolves inheritance from
            // [workspace.package].license.
            if let Some((detail, fix)) = license_spdx_issue(&plan.root, krate) {
                report.push(Issue {
                    repo: plan.name.clone(),
                    krate: krate.clone(),
                    kind: IssueKind::BadManifest,
                    detail,
                    fix,
                    blames: Vec::new(),
                });
            }

            // (P1#2) versioned path dev-dep on a workspace sibling.
            for (dep, ver) in versioned_path_dev_deps(&plan.root, krate) {
                report.push(Issue {
                    repo: plan.name.clone(),
                    krate: krate.clone(),
                    kind: IssueKind::VersionedPathDevDep,
                    detail: format!(
                        "[dev-dependencies] {dep} = {{ path=…, version=\"{ver}\" }} — a \
                         versioned path dev-dep is resolved by `cargo publish` even though \
                         dev-deps are stripped, and fail-stops if the sibling isn't live"
                    ),
                    fix: format!(
                        "drop the `version` from `{dep}` (path-only dev-deps are stripped on publish)"
                    ),
                    blames: Vec::new(),
                });
            }

            // (P2 table a/b/d/f) additional manifest/dep blockers.
            if let Some((mpath, doc)) = find_crate_manifest(&plan.root, krate) {
                for (kind, detail, fix) in extra_manifest_gaps(&mpath, &doc, &publish_false) {
                    report.push(Issue {
                        repo: plan.name.clone(),
                        krate: krate.clone(),
                        kind,
                        detail,
                        fix,
                        blames: Vec::new(),
                    });
                }
            }

            // (5) version/registry consistency.
            if let Some(issue) = registry_skew_issue(
                &plan.name,
                krate,
                version,
                bump_keep,
                probe.version_live(krate, version),
            ) {
                report.push(issue);
            }

            // (6) REGISTRY-STATE desync — ask the backend the max PUBLISHED
            // version and flag a local tree that is BEHIND it (the
            // gatling-0.1.4-local vs 0.1.6-published class): a dependent would
            // verify against the newer published API and break mid-cascade.
            if let Some(issue) = registry_desync_issue(
                &plan.name,
                krate,
                version,
                backend.published_max_version(krate).as_deref(),
            ) {
                report.push(issue);
            }
        }

        // (P2 table c) NON-dev `path =` dep with no `version =` — cargo strips the
        // path on publish, leaving a version-less dep it rejects. Wire the existing
        // audit (normal + build deps, `publish = false` members skipped) into the
        // up-front report. Attributed to the owning crate (from the manifest).
        if let Ok(findings) = crate::release::gate::path_dep_audit(&plan.root) {
            for f in findings.iter().filter(|f| !f.ok()) {
                let owner = std::fs::read_to_string(&f.manifest)
                    .ok()
                    .and_then(|t| t.parse::<toml::Value>().ok())
                    .and_then(|d| {
                        d.get("package")
                            .and_then(|p| p.get("name"))
                            .and_then(|n| n.as_str())
                            .map(str::to_string)
                    })
                    .unwrap_or_else(|| f.dep_name.clone());
                report.push(Issue {
                    repo: plan.name.clone(),
                    krate: owner,
                    kind: IssueKind::UnpublishableDep,
                    detail: format!(
                        "path dep `{}` has no `version =` — cargo strips the path on publish, \
                         leaving a version-less dep it rejects",
                        f.dep_name
                    ),
                    fix: format!(
                        "add `version = \"…\"` to `{}` in {}",
                        f.dep_name,
                        f.manifest.display()
                    ),
                    blames: Vec::new(),
                });
            }
        }

        // (#36 advisory) a versioned CROSS-REPO `path =` dep masks the published
        // crate — the local build never validates the registry copy. NOTE only.
        for note in path_masks_published_notes(plan, backend) {
            report.notes.push(note);
        }

        // (3) dep-API skew via dry-run verify — expensive, so one pass per repo.
        if verify == VerifyMode::On {
            for (krate, _) in &plan.publish_set {
                if members.as_ref().map(|m| !m.contains(krate)).unwrap_or(false) {
                    continue; // non-member already flagged
                }
                match verify_one(&plan.name, &plan.root, krate, cascade_publishes, &plan_publishes) {
                    VerifyOutcome::Clean => {}
                    VerifyOutcome::Blocker(issue) => report.push(issue),
                    VerifyOutcome::CascadeResolvable { dep, planned } => {
                        // The cascade itself publishes `dep` forward before this
                        // crate — the skew is expected mid-cascade, NOT a blocker.
                        report.notes.push(format!(
                            "{}/{krate}: verify skew against `{dep}` is cascade-resolvable \
                             ({dep} will be published at {planned} earlier in the wave) — \
                             not a blocker",
                            plan.name
                        ));
                    }
                }
                // (size) the dry-run verify above already packaged the `.crate` into
                // target/package — inspect its size and flag an over-limit tarball
                // (crates.io's default upload cap is 10 MiB). Best-effort: if no
                // tarball was produced (verify failed early), this is a no-op.
                if let Some(issue) = oversized_tarball_issue(&plan.name, &plan.root, krate) {
                    report.push(issue);
                }
            }
        }
    }
    report
}

/// crates.io rejects an upload whose `.crate` tarball exceeds the size limit (10
/// MiB by default). The dry-run verify (`cargo publish --dry-run`) already wrote the
/// packaged tarball to `<repo_root>/target/package/<krate>-<version>.crate`; inspect
/// the newest matching file and return an [`IssueKind::OversizedTarball`] issue when
/// it is over the limit. Best-effort: a missing tarball (verify failed early, cargo
/// absent) yields `None` — never a false block.
fn oversized_tarball_issue(repo: &str, repo_root: &Path, krate: &str) -> Option<Issue> {
    const MAX_BYTES: u64 = 10 * 1024 * 1024;
    let dir = repo_root.join("target").join("package");
    let prefix = format!("{krate}-");
    let mut newest: Option<(std::time::SystemTime, u64)> = None;
    for e in std::fs::read_dir(&dir).ok()?.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("");
        // Require the char after `<krate>-` to be a digit (the version) so a crate
        // named `foo` doesn't match `foo-bar-1.0.0.crate`.
        let is_this = fname
            .strip_prefix(&prefix)
            .map(|r| r.starts_with(|c: char| c.is_ascii_digit()))
            .unwrap_or(false);
        if !is_this {
            continue;
        }
        let Ok(meta) = e.metadata() else { continue };
        let mtime = meta.modified().unwrap_or(std::time::UNIX_EPOCH);
        if newest.as_ref().map(|(t, _)| mtime > *t).unwrap_or(true) {
            newest = Some((mtime, meta.len()));
        }
    }
    let (_, bytes) = newest?;
    if bytes > MAX_BYTES {
        return Some(Issue {
            repo: repo.to_string(),
            krate: krate.to_string(),
            kind: IssueKind::OversizedTarball,
            detail: format!(
                "packaged .crate is {:.1} MiB — crates.io's default upload limit is 10 MiB",
                bytes as f64 / (1024.0 * 1024.0)
            ),
            fix: "shrink the package: `exclude` test fixtures / large assets (or set \
                  `include`) in [package], or split the crate"
                .to_string(),
            blames: Vec::new(),
        });
    }
    None
}

/// A dep `path` that ESCAPES the repo root — a cross-repo extracted crate
/// (`../funnel-core`, `../edda/crates/…`, `../dwarves/crates/dwarfs`). Any `..`
/// component means the path leaves this repo; an intra-repo path
/// (`crates/foo`, `./sub`) does not and is a normal same-cascade sibling.
fn is_cross_repo_path(path: &str) -> bool {
    Path::new(path)
        .components()
        .any(|c| matches!(c, std::path::Component::ParentDir))
}

/// PURE enumeration (no network) of the distinct CROSS-REPO `path`+`version` deps
/// in a repo: every `[dependencies]`/`[build-dependencies]` entry across all
/// member manifests under `root` that carries BOTH a `path` (escaping the repo,
/// per [`is_cross_repo_path`]) AND a `version`. Deduped by dep name (first
/// path/version seen wins). Returns `(name, version_req, sample_path)`.
fn cross_repo_versioned_path_deps(root: &Path) -> Vec<(String, String, String)> {
    let mut seen: BTreeMap<String, (String, String)> = BTreeMap::new();
    let _ = cargo::for_each_cargo_toml(root, &mut |doc: &toml::Value| {
        for table in ["dependencies", "build-dependencies"] {
            let Some(deps) = doc.get(table).and_then(|d| d.as_table()) else {
                continue;
            };
            for (name, spec) in deps {
                let Some(t) = spec.as_table() else { continue };
                let (Some(path), Some(ver)) = (
                    t.get("path").and_then(|p| p.as_str()),
                    t.get("version").and_then(|v| v.as_str()),
                ) else {
                    continue;
                };
                if !is_cross_repo_path(path) {
                    continue;
                }
                seen.entry(name.clone())
                    .or_insert_with(|| (ver.to_string(), path.to_string()));
            }
        }
    });
    seen.into_iter().map(|(name, (ver, path))| (name, ver, path)).collect()
}

/// **Advisory (#36): a versioned CROSS-REPO `path =` dep masks the PUBLISHED
/// crate.** A workspace member that depends on an already-extracted-and-published
/// crate via BOTH `path = "../other-repo/…"` AND `version = "X"` builds against
/// the LOCAL path — so the crates.io copy is NEVER validated by `cargo
/// build`/`cargo test` here. Only `cargo publish --verify` (which strips the
/// path) and a fresh consumer resolve the real registry version. If the local
/// path source has drifted from the published copy, the workspace's green build
/// hides a divergence a fresh `cargo install` would hit (this is the class the
/// bencher flagged in batch 23: funnel-core, the edda crates, dwarves).
///
/// A NOTE, never a blocker — the release cascade itself still resolves correctly
/// (verify strips the path; a same-release sibling publishes in-order; a
/// drifted crate that is IN the publish set is already a hard `StaleSourceDrift`
/// blocker, #12). The fix is hygiene: drop the `path =` (version-only) once the
/// crate is published, or gate every such path behind `[patch.crates-io]` / a
/// dev-only profile so the release build resolves the true registry version.
///
/// Bounded network: one `published_max_version` probe per DISTINCT cross-repo
/// dep; an unknown/offline answer (or an unpublished crate — a genuine
/// same-cascade sibling) is skipped, never a false note.
pub fn path_masks_published_notes(plan: &RepoPlan, backend: &dyn RegistryBackend) -> Vec<String> {
    let mut notes = Vec::new();
    for (name, ver, sample_path) in cross_repo_versioned_path_deps(&plan.root) {
        // Only a crate that ACTUALLY has a published version is "masked" — an
        // unpublished cross-repo path is a legitimate same-cascade sibling.
        if let Some(published) = backend.published_max_version(&name) {
            notes.push(format!(
                "{}: dep `{name}` = {{ path=\"{sample_path}\", version=\"{ver}\" }} is published on \
                 crates.io (max {published}) but the workspace build resolves the LOCAL path — the \
                 registry copy is NOT validated here (only `cargo publish --verify` + fresh \
                 consumers use it). Drop `path=` (version-only) now that `{name}` is published, or \
                 gate it behind [patch.crates-io], so this build resolves the real {published}.",
                plan.name
            ));
        }
    }
    notes
}

/// SAME-VERSION source-drift pass — the "edited without a version bump" trap that
/// escaped `run()`'s dry-run verify one-cargo-at-a-time in the fat release
/// (`znippy-plugin-python`/`holger-traits`/`funnel-core`). For every planned crate
/// whose TARGET version is ALREADY live on the registry (`probe.version_live ==
/// Some(true)`), compare the local shippable source against the published copy:
///   * [`DriftVerdict::Drifted`] ⇒ a hard [`IssueKind::StaleSourceDrift`] blocker
///     naming the changed files, with the fix "bump the version".
///   * [`DriftVerdict::Unchanged`] ⇒ a NOTE that the crate is safe to skip
///     republishing (the same signal the bump step uses to avoid the rate-limit
///     grind on large unchanged crate sets).
///   * [`DriftVerdict::Unknown`] ⇒ a NOTE (best-effort; never a false block).
/// Bounded network: only AlreadyPublished targets are fetched, so under `--bump
/// patch` (fresh versions) this is nearly a no-op; it earns its keep on
/// `--bump keep` / resume / reconcile. Returns `(issues, notes)` for the caller to
/// merge into the [`PreflightReport`].
pub fn source_drift_issues(
    plans: &[RepoPlan],
    probe: &dyn RegistryProbe,
    oracle: &dyn crate::release::source_drift::PublishedSourceOracle,
) -> (Vec<Issue>, Vec<String>) {
    use crate::release::source_drift::{assess, summarize_differing, DriftVerdict};
    let mut issues = Vec::new();
    let mut notes = Vec::new();
    for plan in plans {
        for (krate, version) in &plan.publish_set {
            // Only crates whose target is CONFIRMED already live are drift-candidates.
            if probe.version_live(krate, version) != Some(true) {
                continue;
            }
            let Some((mpath, _)) = find_crate_manifest(&plan.root, krate) else {
                notes.push(format!(
                    "{}/{krate}: source-drift check skipped — manifest not found under {}",
                    plan.name,
                    plan.root.display()
                ));
                continue;
            };
            let crate_dir = mpath.parent().unwrap_or(&plan.root);
            match assess(oracle, crate_dir, krate, version) {
                DriftVerdict::Unchanged => notes.push(format!(
                    "{}/{krate}: local source matches published {version} — safe to skip republish",
                    plan.name
                )),
                DriftVerdict::Drifted { differing } => issues.push(Issue {
                    repo: plan.name.clone(),
                    krate: krate.clone(),
                    kind: IssueKind::StaleSourceDrift,
                    detail: format!(
                        "local source differs from the published {krate}@{version} but the \
                         version is unchanged — a dependent verifies against the STALE registry \
                         copy and fail-stops. Changed: {}",
                        summarize_differing(&differing)
                    ),
                    fix: format!(
                        "bump {krate}'s version (its published bytes are immutable; ship the \
                         edit as a new version)"
                    ),
                    blames: Vec::new(),
                }),
                DriftVerdict::Unknown(why) => notes.push(format!(
                    "{}/{krate}: source-drift check inconclusive — {why}",
                    plan.name
                )),
            }
        }
    }
    (issues, notes)
}

/// Intra-release `links` collisions: crates.io requires each published crate's
/// `[package].links` value to be GLOBALLY unique (a native-library claim). We can't
/// see other authors' claims from here (backend-only), but if TWO crates in the
/// release set declare the SAME `links` value the second upload is guaranteed to
/// fail — so flag every crate sharing a value. Pure over the plans' manifests.
pub fn duplicate_links_issues(plans: &[RepoPlan]) -> Vec<Issue> {
    let mut by_links: BTreeMap<String, Vec<(String, String)>> = BTreeMap::new();
    for plan in plans {
        for (krate, _) in &plan.publish_set {
            if let Some((_, doc)) = find_crate_manifest(&plan.root, krate) {
                if let Some(links) = doc
                    .get("package")
                    .and_then(|p| p.get("links"))
                    .and_then(|l| l.as_str())
                {
                    by_links
                        .entry(links.to_string())
                        .or_default()
                        .push((plan.name.clone(), krate.clone()));
                }
            }
        }
    }
    let mut issues = Vec::new();
    for (links, crates) in &by_links {
        if crates.len() < 2 {
            continue;
        }
        let names: Vec<String> = crates.iter().map(|(_, k)| k.clone()).collect();
        for (repo, krate) in crates {
            issues.push(Issue {
                repo: repo.clone(),
                krate: krate.clone(),
                kind: IssueKind::BadManifest,
                detail: format!(
                    "`links = \"{links}\"` is declared by {} crates in this release ({}) — \
                     crates.io requires links to be globally unique",
                    crates.len(),
                    names.join(", ")
                ),
                fix: "give each crate a distinct `links` value (it must be globally unique on crates.io)"
                    .to_string(),
                blames: Vec::new(),
            });
        }
    }
    issues
}

/// Packages under `repo_root` that the toml-walk finds but that are NOT members of
/// the root workspace — i.e. they live in a detached `[workspace]`. Pure over the
/// injected `members` set.
fn detached_workspace_crates(repo_root: &Path, members: &BTreeSet<String>) -> BTreeSet<String> {
    let walked = cargo::publishable_crate_versions_walk(repo_root).unwrap_or_default();
    walked
        .into_iter()
        .map(|(n, _)| n)
        .filter(|n| !members.contains(n))
        .collect()
}

/// Required-metadata gap for one crate, resolving `field.workspace = true`
/// inheritance against the root `[workspace.package]`. Returns the missing field
/// labels (empty ⇒ complete). `license` is satisfied by EITHER `license` or
/// `license-file`.
pub fn crate_metadata_gaps(repo_root: &Path, krate: &str) -> Vec<String> {
    // Inheritance source is the ROOT workspace's [workspace.package] ONLY. A
    // DETACHED sub-workspace under the tree (e.g. facett's `facett-py/Cargo.toml`
    // with `[workspace] package = {}`, or `bench/`) must NOT clobber it: the old
    // last-write-wins walk let an EMPTY sub-workspace table win whenever the
    // unordered `read_dir` happened to visit it after the root, making every
    // `license.workspace = true` / `repository.workspace = true` resolve to
    // "missing" — a FLAKY false blocker that flipped with directory order between
    // runs. Publish-set members inherit from the root by cargo's own rules and are
    // filtered to root-workspace members anyway, so only the root table is valid.
    let ws_package = std::fs::read_to_string(repo_root.join("Cargo.toml"))
        .ok()
        .and_then(|t| t.parse::<toml::Value>().ok())
        .and_then(|d| d.get("workspace").and_then(|w| w.get("package")).cloned());

    // Find the crate's own manifest.
    let mut pkg_doc: Option<toml::Value> = None;
    let _ = cargo::for_each_cargo_toml(repo_root, &mut |doc: &toml::Value| {
        if doc
            .get("package")
            .and_then(|p| p.get("name"))
            .and_then(|n| n.as_str())
            == Some(krate)
        {
            pkg_doc = doc.get("package").cloned();
        }
    });
    let Some(pkg) = pkg_doc else { return Vec::new() };
    missing_metadata_fields(&pkg, ws_package.as_ref())
}

/// Pure crate-NAME validation against crates.io's rules. Returns `Some(reason)`
/// when the name would be rejected: empty, >64 chars, a first character that is not
/// an ASCII letter (crates.io rejects a leading digit/`-`/`_`), a character outside
/// `[A-Za-z0-9_-]`, or a reserved Windows-device name (`con`, `nul`, `com1`…, on the
/// crates.io blocklist). `None` ⇒ the name is well-formed. (The DYNAMIC reserved /
/// too-similar checks are backend-only and NOT covered here — a well-formed name we
/// can't disprove is never falsely blocked.)
pub fn crate_name_problem(name: &str) -> Option<String> {
    const MAX_NAME_LENGTH: usize = 64;
    if name.is_empty() {
        return Some("crate name is empty".to_string());
    }
    let len = name.chars().count();
    if len > MAX_NAME_LENGTH {
        return Some(format!(
            "crate name `{name}` is {len} chars — crates.io allows at most {MAX_NAME_LENGTH}"
        ));
    }
    let first = name.chars().next().unwrap();
    if !first.is_ascii_alphabetic() {
        return Some(format!(
            "crate name `{name}` must start with an ASCII letter — crates.io rejects a leading \
             digit, `-`, or `_`"
        ));
    }
    if let Some(bad) = name
        .chars()
        .find(|c| !(c.is_ascii_alphanumeric() || *c == '-' || *c == '_'))
    {
        return Some(format!(
            "crate name `{name}` contains `{bad}` — only ASCII letters, digits, `-` and `_` are allowed"
        ));
    }
    const RESERVED: &[&str] = &[
        "CON", "PRN", "AUX", "NUL", "COM0", "COM1", "COM2", "COM3", "COM4", "COM5", "COM6", "COM7",
        "COM8", "COM9", "LPT0", "LPT1", "LPT2", "LPT3", "LPT4", "LPT5", "LPT6", "LPT7", "LPT8",
        "LPT9",
    ];
    if RESERVED.contains(&name.to_ascii_uppercase().as_str()) {
        return Some(format!(
            "crate name `{name}` is a reserved (Windows device) name that crates.io blocks"
        ));
    }
    None
}

/// Best-effort SPDX license-expression validation. crates.io validates `license`
/// with the `spdx` crate and rejects an invalid expression with "unknown or invalid
/// license expression". Returns `Some(reason)` only when the expression is
/// CONFIDENTLY invalid (empty, the deprecated `/` separator, a bad-charset token, or
/// a broken AND/OR/WITH grammar) — a blocker. A well-formed but exotic/unknown
/// identifier returns `None` (degrade: never a false block; the dry-run / crates.io
/// is the backstop for a mistyped-but-well-formed id).
pub fn spdx_license_problem(expr: &str) -> Option<String> {
    let e = expr.trim();
    if e.is_empty() {
        return Some("the license expression is empty".to_string());
    }
    // The deprecated `/` separator (`MIT/Apache-2.0`) is not valid SPDX; SPDX uses
    // `OR`. cargo deprecated it and crates.io's spdx parse rejects it.
    if e.contains('/') {
        return Some(format!(
            "`{e}` uses the deprecated `/` separator — SPDX uses `OR` (e.g. `MIT OR Apache-2.0`)"
        ));
    }
    // Tokenize on whitespace, with `(`/`)` as their own tokens.
    let mut toks: Vec<String> = Vec::new();
    let mut cur = String::new();
    for c in e.chars() {
        match c {
            '(' | ')' => {
                if !cur.is_empty() {
                    toks.push(std::mem::take(&mut cur));
                }
                toks.push(c.to_string());
            }
            c if c.is_whitespace() => {
                if !cur.is_empty() {
                    toks.push(std::mem::take(&mut cur));
                }
            }
            _ => cur.push(c),
        }
    }
    if !cur.is_empty() {
        toks.push(cur);
    }
    // Balanced parentheses.
    let mut depth = 0i32;
    for t in &toks {
        if t == "(" {
            depth += 1;
        } else if t == ")" {
            depth -= 1;
            if depth < 0 {
                return Some(format!("`{e}` has unbalanced parentheses"));
            }
        }
    }
    if depth != 0 {
        return Some(format!("`{e}` has unbalanced parentheses"));
    }
    // Grammar walk: alternate operand / operator, WITH/AND/OR are the operators.
    let is_op = |t: &str| matches!(t, "AND" | "OR" | "WITH");
    let mut expect_operand = true;
    for t in &toks {
        if t == "(" {
            if !expect_operand {
                return Some(format!("`{e}`: `(` follows a license with no operator between them"));
            }
            continue; // still expecting the operand that opens the group
        }
        if t == ")" {
            if expect_operand {
                return Some(format!("`{e}`: empty group or dangling operator before `)`"));
            }
            continue; // the group acts as a completed operand
        }
        if is_op(t) {
            if expect_operand {
                return Some(format!("`{e}`: operator `{t}` with no license before it"));
            }
            expect_operand = true;
            continue;
        }
        // An operand (license-id / exception-id).
        if !expect_operand {
            return Some(format!(
                "`{e}`: two licenses (`{t}`) with no AND/OR between them"
            ));
        }
        if !valid_spdx_id_token(t) {
            return Some(format!("`{e}`: `{t}` is not a valid SPDX license identifier"));
        }
        expect_operand = false;
    }
    if expect_operand {
        return Some(format!("`{e}`: expression ends on an operator (expected a license)"));
    }
    None
}

/// Whether a single operand token is a plausibly-valid SPDX id/exception: ASCII
/// `[A-Za-z0-9.-]`, starting alphanumeric, with an optional trailing `+` (the
/// "or-later" marker). `LicenseRef-…`/`DocumentRef-…` (which may carry a `:`) are
/// waved through — we don't judge custom refs.
fn valid_spdx_id_token(t: &str) -> bool {
    if t.contains(':') {
        return true; // DocumentRef/LicenseRef custom id — don't false-block.
    }
    let core = t.strip_suffix('+').unwrap_or(t);
    let mut chars = core.chars();
    match chars.next() {
        Some(c) if c.is_ascii_alphanumeric() => {}
        _ => return false,
    }
    core.chars().all(|c| c.is_ascii_alphanumeric() || c == '.' || c == '-')
}

/// The crate's EFFECTIVE license expression (explicit `license = "…"` on
/// `[package]`, else the workspace `[workspace.package].license` when the crate
/// inherits via `license.workspace = true`), validated by [`spdx_license_problem`].
/// `None` (clean) when there is no `license` string to check (a `license-file`-only
/// crate, or a crate whose license is absent — the latter is the [`MissingMetadata`]
/// check's job) or when the expression is valid. Best-effort: an unreadable manifest
/// yields `None`.
pub fn license_spdx_issue(repo_root: &Path, krate: &str) -> Option<(String, String)> {
    let ws_license = std::fs::read_to_string(repo_root.join("Cargo.toml"))
        .ok()
        .and_then(|t| t.parse::<toml::Value>().ok())
        .and_then(|d| {
            d.get("workspace")
                .and_then(|w| w.get("package"))
                .and_then(|p| p.get("license"))
                .and_then(|l| l.as_str())
                .map(str::to_string)
        });
    let (_, doc) = find_crate_manifest(repo_root, krate)?;
    let license = doc.get("package")?.get("license")?;
    let effective = match license {
        toml::Value::String(s) => Some(s.clone()),
        toml::Value::Table(t) if t.get("workspace").and_then(|v| v.as_bool()) == Some(true) => {
            ws_license
        }
        _ => None,
    }?;
    let reason = spdx_license_problem(&effective)?;
    Some((
        format!("invalid `license` — {reason}"),
        "set `license` to a valid SPDX expression (https://spdx.org/licenses/), \
         e.g. `MIT OR Apache-2.0`, or use `license-file`"
            .to_string(),
    ))
}

/// Find the crate `krate`'s own manifest under `repo_root`, returning its PATH and
/// parsed document. Used by the additional manifest checks that need the manifest
/// directory (to resolve a `readme` path). `None` if the crate isn't found.
fn find_crate_manifest(repo_root: &Path, krate: &str) -> Option<(PathBuf, toml::Value)> {
    let mut found: Option<(PathBuf, toml::Value)> = None;
    let _ = cargo::for_each_cargo_toml_with_path(repo_root, &mut |path: &Path, doc: &toml::Value| {
        if found.is_none()
            && doc
                .get("package")
                .and_then(|p| p.get("name"))
                .and_then(|n| n.as_str())
                == Some(krate)
        {
            found = Some((path.to_path_buf(), doc.clone()));
        }
    });
    found
}

/// Versioned path dev-deps on a workspace sibling (P1#2): a `[dev-dependencies]`
/// (or `[target.*.dev-dependencies]`) entry carrying BOTH `path =` and
/// `version =`. cargo STRIPS dev-deps from the published `.crate`, so the
/// `version` is never needed — but a *versioned* dev-dep still forces
/// `cargo publish` to RESOLVE that version against the registry during the verify
/// build, fail-stopping with "no matching package" when the sibling isn't live
/// (the facett-form → facett-wasm-demo trap). Returns each offending
/// `(dep_name, version_req)`. Pure over the crate's own manifest.
pub fn versioned_path_dev_deps(repo_root: &Path, krate: &str) -> Vec<(String, String)> {
    let Some((_, doc)) = find_crate_manifest(repo_root, krate) else { return Vec::new() };
    let mut out = Vec::new();
    collect_versioned_path_dev_deps(&doc, &mut out);
    if let Some(targets) = doc.get("target").and_then(|t| t.as_table()) {
        for (_cfg, t) in targets {
            collect_versioned_path_dev_deps(t, &mut out);
        }
    }
    out
}

/// Push every `dev-dependencies` entry under `parent` that has BOTH a `path` and a
/// `version` — modelled on [`crate::release::gate`]'s path-dep walker.
fn collect_versioned_path_dev_deps(parent: &toml::Value, out: &mut Vec<(String, String)>) {
    let Some(deps) = parent.get("dev-dependencies").and_then(|d| d.as_table()) else { return };
    for (name, v) in deps {
        let Some(t) = v.as_table() else { continue };
        let has_path = t.get("path").and_then(|p| p.as_str()).is_some();
        if let (true, Some(ver)) = (has_path, t.get("version").and_then(|v| v.as_str())) {
            out.push((name.clone(), ver.to_string()));
        }
    }
}

/// Additional up-front, manifest-only crates.io blockers beyond the required-field
/// gaps ([`crate_metadata_gaps`]): (a) a `readme` pointing at a missing file,
/// (b) an invalid `keywords` set, (d) a NORMAL/BUILD versioned dep on a
/// `publish = false` sibling, and (f) a `[features]` entry referencing a `dep:foo`
/// absent from `[dependencies]`. Returns `(IssueKind, detail, fix)` per finding.
/// Pure over the crate's own manifest (+ the repo's `publish = false` sibling set).
fn extra_manifest_gaps(
    manifest: &Path,
    doc: &toml::Value,
    publish_false_siblings: &BTreeSet<String>,
) -> Vec<(IssueKind, String, String)> {
    let mut out: Vec<(IssueKind, String, String)> = Vec::new();
    let Some(pkg) = doc.get("package") else { return out };

    // (a) `readme = "X"` / `license-file = "X"` — the declared file must EXIST
    // (resolved rel to the manifest) and be valid UTF-8 (crates.io stores/renders
    // both as UTF-8 text; a missing or non-UTF-8 file fails the upload).
    if let Some(readme) = pkg.get("readme").and_then(|v| v.as_str()) {
        if let Some((detail, fix)) = declared_file_gap(manifest, "readme", readme) {
            out.push((IssueKind::BadManifest, detail, fix));
        }
    }
    if let Some(lf) = pkg.get("license-file").and_then(|v| v.as_str()) {
        if let Some((detail, fix)) = declared_file_gap(manifest, "license-file", lf) {
            out.push((IssueKind::BadManifest, detail, fix));
        }
    }

    // (b) keywords: crates.io allows at most 5, each ≤20 chars over [A-Za-z0-9_-].
    if let Some(kw) = pkg.get("keywords").and_then(|v| v.as_array()) {
        if kw.len() > 5 {
            out.push((
                IssueKind::BadManifest,
                format!("{} keywords (crates.io allows at most 5)", kw.len()),
                "keep at most 5 entries in [package].keywords".to_string(),
            ));
        }
        for k in kw {
            let Some(s) = k.as_str() else { continue };
            // crates.io: each keyword must start with an alphanumeric, be ≤20 chars,
            // and contain only ASCII letters/digits/`-`/`_`/`+`.
            let bad_char =
                !s.chars().all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_' || c == '+');
            let bad_start = !s.chars().next().map(|c| c.is_ascii_alphanumeric()).unwrap_or(false);
            if s.chars().count() > 20 || bad_char || bad_start {
                out.push((
                    IssueKind::BadManifest,
                    format!(
                        "invalid keyword {s:?} (crates.io: ≤20 chars, must start alphanumeric, \
                         only letters/digits/`-`/`_`/`+`)"
                    ),
                    "fix the offending entry in [package].keywords".to_string(),
                ));
            }
        }
    }

    // (categories) crates.io allows at most 5 categories — MORE than 5 is a hard
    // "expected at most 5 categories per crate" error. (An UNKNOWN category slug is
    // only a WARNING on crates.io — the crate still publishes — so it is NOT a
    // blocker here; see the coverage matrix, class #30.)
    if let Some(cats) = pkg.get("categories").and_then(|v| v.as_array()) {
        if cats.len() > 5 {
            out.push((
                IssueKind::BadManifest,
                format!("{} categories (crates.io allows at most 5)", cats.len()),
                "keep at most 5 entries in [package].categories".to_string(),
            ));
        }
    }

    // (description length) crates.io caps the description at 1000 chars ("The
    // `description` is too long. A maximum of 1000 characters …").
    if let Some(desc) = pkg.get("description").and_then(|v| v.as_str()) {
        let n = desc.chars().count();
        if n > 1000 {
            out.push((
                IssueKind::BadManifest,
                format!("description is {n} chars — crates.io allows at most 1000"),
                "shorten [package].description to ≤1000 characters".to_string(),
            ));
        }
    }

    // (cross-registry dep) a published dep pinned to ANOTHER registry (`registry` /
    // `registry-index`) — crates.io rejects "Dependency `X` is hosted on another
    // registry. Cross-registry dependencies are not permitted".
    for_each_published_dep(doc, &mut |name, v| {
        if let toml::Value::Table(t) = v {
            let cross = t.get("registry").and_then(|r| r.as_str()).is_some_and(|r| r != "crates-io")
                || t.get("registry-index").is_some();
            if cross {
                out.push((
                    IssueKind::BadManifest,
                    format!(
                        "dependency `{name}` targets another registry \
                         (`registry`/`registry-index`) — crates.io forbids cross-registry deps"
                    ),
                    format!("depend on a crates.io version of `{name}`, or vendor it"),
                ));
            }
        }
    });

    // (edition) an explicit `edition` must be a Rust edition cargo/crates.io know —
    // an unknown one fails the manifest parse ("supported edition values are …").
    if let Some(ed) = pkg.get("edition").and_then(|v| v.as_str()) {
        const KNOWN: &[&str] = &["2015", "2018", "2021", "2024"];
        if !KNOWN.contains(&ed) {
            out.push((
                IssueKind::BadManifest,
                format!("edition = \"{ed}\" is not a known Rust edition (2015/2018/2021/2024)"),
                "set `edition` to a supported value".to_string(),
            ));
        }
    }

    // (rust-version) crates.io requires digits-and-periods only (a bare `1.74` /
    // `1.74.0`) — no operators (`^`/`~`/`*`), ranges, or pre-release.
    if let Some(rv) = pkg.get("rust-version").and_then(|v| v.as_str()) {
        let ok = !rv.is_empty()
            && rv.chars().all(|c| c.is_ascii_digit() || c == '.')
            && rv.chars().any(|c| c.is_ascii_digit());
        if !ok {
            out.push((
                IssueKind::BadManifest,
                format!(
                    "rust-version = \"{rv}\" must be digits and periods only (e.g. \"1.74\") — \
                     no operators, ranges, or pre-release"
                ),
                "set `rust-version` to a bare version like \"1.74\"".to_string(),
            ));
        }
    }

    // (url) `repository` / `homepage` / `documentation` must be a full http(s) URL —
    // crates.io rejects "URL for field `X` must begin with http:// or https://" (the
    // classic `repository = "github.com/…"` / `"git@…"` mistake). Only explicit
    // string values are checked (an inherited `{ workspace = true }` resolves to the
    // workspace URL, validated where it's declared).
    for field in ["repository", "homepage", "documentation"] {
        if let Some(url) = pkg.get(field).and_then(|v| v.as_str()) {
            let u = url.trim();
            if !(u.starts_with("http://") || u.starts_with("https://")) {
                out.push((
                    IssueKind::BadManifest,
                    format!("`{field}` = \"{url}\" must begin with http:// or https://"),
                    format!("set `{field}` to a full URL (https://…) in [package]"),
                ));
            }
        }
    }

    // (f) a `[features]` value that names something that does not exist. cargo
    // rejects three shapes: `dep:foo` where `foo` is no dependency ("feature …
    // includes `dep:foo`, but `foo` is not a dependency"); `foo/bar` / `foo?/bar`
    // where `foo` is no dependency; and a bare `bar` that is neither another feature
    // nor a dependency ("feature `x` includes `bar` which is neither a dependency
    // nor another feature"). Dep names span every table (normal/build/dev + target)
    // so we NEVER false-block a legitimately-referenced dep-feature.
    let dep_names: BTreeSet<String> = manifest_all_dep_names(doc);
    if let Some(features) = doc.get("features").and_then(|f| f.as_table()) {
        let feature_names: BTreeSet<&String> = features.keys().collect();
        for (feat, list) in features {
            let Some(items) = list.as_array() else { continue };
            for it in items {
                let Some(s) = it.as_str() else { continue };
                if let Some(dep) = s.strip_prefix("dep:") {
                    if !dep_names.contains(dep) {
                        out.push((
                            IssueKind::BadManifest,
                            format!("feature `{feat}` references `dep:{dep}` but `{dep}` is not a dependency"),
                            format!("add `{dep}` to [dependencies], or remove the `dep:{dep}` reference"),
                        ));
                    }
                } else if let Some((left, _)) = s.split_once('/') {
                    // `foo/bar` or weak `foo?/bar` — `foo` must be a dependency.
                    let dep = left.trim_end_matches('?');
                    if !dep_names.contains(dep) {
                        out.push((
                            IssueKind::BadManifest,
                            format!("feature `{feat}` enables `{s}` but `{dep}` is not a dependency"),
                            format!("add `{dep}` to [dependencies], or fix the `{s}` reference"),
                        ));
                    }
                } else if !feature_names.contains(&s.to_string()) && !dep_names.contains(s) {
                    out.push((
                        IssueKind::BadManifest,
                        format!(
                            "feature `{feat}` includes `{s}`, which is neither another feature \
                             nor a dependency"
                        ),
                        format!("define a `{s}` feature/dependency, or remove the `{s}` reference"),
                    ));
                }
            }
        }
    }

    // (wildcard) a bare `*` version requirement on a PUBLISHED dep (normal/build,
    // incl. target-specific) — crates.io hard-rejects "wildcard (`*`) dependency
    // constraints are not allowed on crates.io" (RFC 1241). Dev-deps are stripped on
    // publish so they're excluded. `1.*`/`1.2.*` are fine — only a bare `*` bites.
    for_each_published_dep(doc, &mut |name, v| {
        if dep_version_req(v).map(|r| r.trim() == "*").unwrap_or(false) {
            out.push((
                IssueKind::BadManifest,
                format!(
                    "dependency `{name}` uses a wildcard `*` version — crates.io rejects \
                     wildcard (`*`) dependency constraints"
                ),
                format!(
                    "pin `{name}` to a real requirement (e.g. `\"1\"`, `\"^1.2\"`, `\">=1, <2\"`)"
                ),
            ));
        }
    });

    // (git) a git dependency with NO `version =` on a PUBLISHED dep — cargo STRIPS
    // the git source on publish (exactly like a `path`) and then rejects the
    // version-less dep: "all dependencies must have a version specified when
    // publishing". A git dep WITH a version is fine (cargo resolves that version
    // from crates.io); if that version isn't live the dry-run verify catches it.
    for_each_published_dep(doc, &mut |name, v| {
        let is_git = matches!(
            v,
            toml::Value::Table(t) if t.get("git").and_then(|g| g.as_str()).is_some()
        );
        if is_git && dep_version_req(v).is_none() {
            out.push((
                IssueKind::UnpublishableDep,
                format!(
                    "git dependency `{name}` has no `version =` — cargo strips the git source \
                     on publish, leaving a version-less dep crates.io rejects"
                ),
                format!(
                    "add `version = \"…\"` to `{name}` (that version must be on crates.io), or drop it"
                ),
            ));
        }
    });

    // (d) a NORMAL/BUILD dep, WITH a version, on a `publish = false` sibling — the
    // sibling never reaches crates.io, so cargo can't resolve the versioned dep.
    for section in ["dependencies", "build-dependencies"] {
        let Some(deps) = doc.get(section).and_then(|d| d.as_table()) else { continue };
        for (name, v) in deps {
            if !publish_false_siblings.contains(name) {
                continue;
            }
            let has_version = match v {
                toml::Value::String(_) => true, // `foo = "1.0"`
                toml::Value::Table(t) => t.get("version").and_then(|x| x.as_str()).is_some(),
                _ => false,
            };
            if has_version {
                out.push((
                    IssueKind::UnpublishableDep,
                    format!(
                        "depends on `{name}` with a version, but `{name}` is `publish = false` \
                         (never on crates.io)"
                    ),
                    format!(
                        "drop the `version` from `{name}` (keep it path-only), or make `{name}` publishable"
                    ),
                ));
            }
        }
    }

    out
}

/// Existence + UTF-8 gap for a manifest-declared file field (`readme`,
/// `license-file`), resolved relative to the manifest. `Some((detail, fix))` when
/// the file is MISSING or NOT valid UTF-8 (both fail the crates.io upload); `None`
/// when it exists and is UTF-8.
fn declared_file_gap(manifest: &Path, field: &str, rel_path: &str) -> Option<(String, String)> {
    let full = manifest.parent().unwrap_or_else(|| Path::new(".")).join(rel_path);
    match std::fs::read(&full) {
        Err(_) => Some((
            format!("{field} = \"{rel_path}\" does not exist ({})", full.display()),
            format!(
                "create `{rel_path}` next to the manifest, or fix the `{field}` path in [package]"
            ),
        )),
        Ok(bytes) if std::str::from_utf8(&bytes).is_err() => Some((
            format!("{field} `{rel_path}` is not valid UTF-8 — crates.io requires UTF-8 text"),
            format!("re-save `{rel_path}` as UTF-8"),
        )),
        Ok(_) => None,
    }
}

/// Visit every PUBLISHED dependency edge of a manifest — normal + build deps,
/// including `[target.<cfg>.dependencies]` / `[target.<cfg>.build-dependencies]`.
/// `[dev-dependencies]` are DELIBERATELY excluded: cargo strips them from the
/// published `.crate`, so their constraints never reach crates.io. Calls `f` with
/// each `(dep_name, spec)` where `spec` is the raw toml value (`"1.0"` string or a
/// `{ version = …, path = …, git = … }` table).
fn for_each_published_dep(doc: &toml::Value, f: &mut dyn FnMut(&str, &toml::Value)) {
    let mut scan = |tbl: &toml::Value| {
        for section in ["dependencies", "build-dependencies"] {
            if let Some(deps) = tbl.get(section).and_then(|d| d.as_table()) {
                for (name, v) in deps {
                    f(name, v);
                }
            }
        }
    };
    scan(doc);
    if let Some(targets) = doc.get("target").and_then(|t| t.as_table()) {
        for t in targets.values() {
            scan(t);
        }
    }
}

/// Every dependency NAME declared anywhere in the manifest — normal, build, and dev
/// tables at the top level AND under `[target.<cfg>.*]`. Used to validate feature
/// references leniently (a feature may enable a dep-feature of any dependency), so
/// the feature-ref check never false-blocks a legitimately-named dependency.
fn manifest_all_dep_names(doc: &toml::Value) -> BTreeSet<String> {
    let mut names = BTreeSet::new();
    let mut scan = |tbl: &toml::Value| {
        for section in ["dependencies", "build-dependencies", "dev-dependencies"] {
            if let Some(deps) = tbl.get(section).and_then(|d| d.as_table()) {
                names.extend(deps.keys().cloned());
            }
        }
    };
    scan(doc);
    if let Some(targets) = doc.get("target").and_then(|t| t.as_table()) {
        for t in targets.values() {
            scan(t);
        }
    }
    names
}

/// The version requirement string of a dependency spec — `foo = "1.0"` (string) or
/// `foo = { version = "1.0", … }` (table). `None` when no `version` key is present
/// (a path/git-only edge).
fn dep_version_req(v: &toml::Value) -> Option<&str> {
    match v {
        toml::Value::String(s) => Some(s.as_str()),
        toml::Value::Table(t) => t.get("version").and_then(|x| x.as_str()),
        _ => None,
    }
}

/// Pure metadata-completeness check over a `[package]` table (and the optional
/// `[workspace.package]` for inheritance). A field counts as present when it's a
/// non-empty string in `[package]`, when it inherits (`field.workspace = true`) and
/// the workspace provides it, or — for license — when `license-file` is set instead.
pub fn missing_metadata_fields(
    pkg: &toml::Value,
    ws_package: Option<&toml::Value>,
) -> Vec<String> {
    let present = |field: &str| -> bool {
        field_present(pkg, ws_package, field)
    };
    let mut missing = Vec::new();
    if !present("description") {
        missing.push("description".to_string());
    }
    if !present("license") && !present("license-file") {
        missing.push("license".to_string());
    }
    if !present("repository") {
        missing.push("repository".to_string());
    }
    missing
}

/// Whether `field` is satisfied on `pkg` — a non-empty string, or an inherited
/// `{ workspace = true }` that the workspace root actually provides.
fn field_present(pkg: &toml::Value, ws_package: Option<&toml::Value>, field: &str) -> bool {
    match pkg.get(field) {
        Some(toml::Value::String(s)) => !s.trim().is_empty(),
        Some(toml::Value::Table(t)) => {
            // `field.workspace = true` inherits from [workspace.package].
            let inherits = t.get("workspace").and_then(|v| v.as_bool()).unwrap_or(false);
            inherits
                && ws_package
                    .and_then(|wp| wp.get(field))
                    .and_then(|v| match v {
                        toml::Value::String(s) => Some(!s.trim().is_empty()),
                        _ => Some(true),
                    })
                    .unwrap_or(false)
        }
        _ => false,
    }
}

/// Pure version/registry-consistency decision. For a real bump, a target version
/// already live on crates.io is an immutable clash (blocker). For `--bump keep`, a
/// live version is a benign idempotent skip. `None` (couldn't probe) never blocks.
pub fn registry_skew_issue(
    repo: &str,
    krate: &str,
    version: &str,
    bump_keep: bool,
    live: Option<bool>,
) -> Option<Issue> {
    if bump_keep {
        return None; // keep-cascade: already-published is intentional & idempotent.
    }
    match live {
        Some(true) => Some(Issue {
            repo: repo.to_string(),
            krate: krate.to_string(),
            kind: IssueKind::RegistrySkew,
            detail: format!("{krate} {version} is already live on crates.io (immutable)"),
            fix: format!(
                "bump {krate} past {version} (crates.io versions can't be overwritten), \
                 or use --bump keep to skip already-published crates"
            ),
            blames: Vec::new(),
        }),
        _ => None,
    }
}

/// Pure registry-STATE decision (item #10): classify the crate's local `version`
/// against the registry's max `published` version and, when the local tree is
/// BEHIND the registry ([`RegistryState::Desync`]), return a blocking issue
/// naming the crate and BOTH versions. `WillPublish`/`AlreadyPublished`/
/// `Unpublished`/`Unknown` are never blockers here (they're expected or
/// unprobable). This is the gatling-0.1.4-local vs 0.1.6-published class: a
/// dependent would verify against the newer published API and break mid-cascade.
pub fn registry_desync_issue(
    repo: &str,
    krate: &str,
    version: &str,
    published: Option<&str>,
) -> Option<Issue> {
    match classify(version, published) {
        RegistryState::Desync => {
            let pub_v = published.unwrap_or("?");
            // The exact version that would lead the registry (one patch past the
            // published max), so the fix names a concrete target + command.
            let next = bump_past(pub_v);
            let target = next.as_deref().unwrap_or(pub_v);
            Some(Issue {
                repo: repo.to_string(),
                krate: krate.to_string(),
                kind: IssueKind::RegistryDesync,
                detail: format!(
                    "local {krate} {version} is BEHIND the registry (published {pub_v}) — \
                     dependents verify against the newer published API and break"
                ),
                fix: format!(
                    "if the local tree is merely STALE, `git pull` in {repo} to catch up to \
                     {pub_v}. If {krate} has genuine local changes, bump it PAST the registry \
                     (to {target}) and republish just it: \
                     `nornir release doctor --publish --bump patch --only {krate}` \
                     (a real, user-driven publish — not a dry-run)"
                ),
                blames: Vec::new(),
            })
        }
        _ => None,
    }
}

/// One patch past a published semver (e.g. `0.1.12` → `0.1.13`) — the concrete
/// "lead the registry" target named in a [`IssueKind::RegistryDesync`] fix.
/// `None` if `published` isn't parseable semver (the fix then names the published
/// version itself).
fn bump_past(published: &str) -> Option<String> {
    let v = semver::Version::parse(published).ok()?;
    Some(format!("{}.{}.{}", v.major, v.minor, v.patch + 1))
}

/// The outcome of a single dry-run verify. A [`Blocker`](VerifyOutcome::Blocker)
/// stops the cascade; a [`CascadeResolvable`](VerifyOutcome::CascadeResolvable)
/// skew is one the cascade itself fixes (the blamed dep is bumped+published
/// earlier in the wave) and is reported only as a note; [`Clean`](VerifyOutcome::Clean)
/// verified fine.
#[derive(Debug, Clone, PartialEq, Eq)]
enum VerifyOutcome {
    /// The crate verified cleanly (or the failure wasn't a preflight-catchable class).
    Clean,
    /// A real blocker — the crate cannot publish as-is.
    Blocker(Issue),
    /// Dep-API skew whose blamed crate the cascade WILL republish forward, so it
    /// resolves as the cascade runs. Not a blocker (item #7, bump-aware).
    CascadeResolvable { dep: String, planned: String },
}

// `Issue` has no `Eq`/`PartialEq`, so hand-derive `VerifyOutcome` equality only
// over the variants the tests compare (Clean / CascadeResolvable); Blocker is
// compared by its issue kind + blamed dep where needed.
impl PartialEq for Issue {
    fn eq(&self, other: &Self) -> bool {
        self.repo == other.repo
            && self.krate == other.krate
            && self.kind == other.kind
            && self.blames == other.blames
    }
}
impl Eq for Issue {}

/// Pure BUMP-AWARE decision (item #7): given the crate the dry-run verify blamed
/// (`dep`; `None` ⇒ the classifier couldn't name it) and the cascade's planned
/// publish map, decide whether the skew is cascade-resolvable. Returns
/// `Some(planned_version)` when the blamed dep is one the cascade WILL publish
/// forward (⇒ NOT a blocker — it will exist at that version before this crate is
/// verified in the real cascade); `None` when the dep is external / not in the
/// release set (⇒ a real blocker). An unnamed dep (`None`) is conservatively a
/// blocker — we can't prove the cascade covers it.
pub fn skew_is_cascade_resolvable(
    dep: Option<&str>,
    cascade_publishes: &BTreeMap<String, String>,
) -> Option<String> {
    let d = dep?;
    if let Some(v) = cascade_publishes.get(d) {
        return Some(v.clone());
    }
    // The blamed name comes from a rustc verify error, which cites the LIB name
    // (a crate's `-` becomes `_`, e.g. `facett_core`); `cascade_publishes` is
    // keyed by the PACKAGE name (`facett-core`). Match with separators unified so
    // a hyphenated crate the cascade WILL republish is correctly recognised as
    // cascade-resolvable instead of a false hard blocker (the facett-core trap).
    let canon = |s: &str| s.replace('_', "-");
    let dc = canon(d);
    cascade_publishes.iter().find(|(k, _)| canon(k) == dc).map(|(_, v)| v.clone())
}

/// Run `cargo publish -p <krate> --dry-run --allow-dirty` and classify a dep-API
/// skew (a compile error citing an already-published dependency). BUMP-AWARE
/// (item #7): if the skew blames a crate the cascade will itself publish forward
/// (`cascade_publishes`), it is [`CascadeResolvable`](VerifyOutcome::CascadeResolvable),
/// not a blocker — the real cascade bumps+publishes that crate first. A skew
/// blaming an external / not-republished crate is a [`Blocker`](VerifyOutcome::Blocker).
/// A pure resolution/ordering failure (sibling not yet on the registry — expected
/// mid-cascade) is not flagged, so preflight doesn't cry wolf.
fn verify_one(
    repo: &str,
    repo_root: &Path,
    krate: &str,
    cascade_publishes: &BTreeMap<String, String>,
    plan_publishes: &BTreeMap<String, String>,
) -> VerifyOutcome {
    let Ok(out) = std::process::Command::new(crate::release::cargo::cargo_bin())
        .args(["publish", "-p", krate, "--dry-run", "--allow-dirty"])
        .current_dir(repo_root)
        .output()
    else {
        return VerifyOutcome::Clean; // cargo unavailable → best-effort, never a false block
    };
    if out.status.success() {
        return VerifyOutcome::Clean;
    }
    let stderr = String::from_utf8_lossy(&out.stderr);
    // The known-crate set for blame validation (P1#5): the whole publish plan's
    // crate names — a blame token must be one of these to be trusted over a bare
    // module name.
    let known: BTreeSet<String> = plan_publishes.keys().cloned().collect();
    match classify_publish_failure(krate, &stderr, &known) {
        PublishFailure::DepApiSkew { dep } => {
            // BUMP-AWARE gate: does the cascade itself publish the blamed dep?
            if let Some(planned) = skew_is_cascade_resolvable(dep.as_deref(), cascade_publishes) {
                let d = dep.unwrap_or_default();
                return VerifyOutcome::CascadeResolvable { dep: d, planned };
            }
            let d = dep.as_deref().unwrap_or("a dependency");
            VerifyOutcome::Blocker(Issue {
                repo: repo.to_string(),
                krate: krate.to_string(),
                kind: IssueKind::VerifyFailed,
                detail: format!(
                    "verify build uses API from `{d}` not in the published `{d}`"
                ),
                fix: format!("bump `{d}` and publish it before `{krate}`"),
                // Blame the exact dep whose un-published API broke the verify —
                // the root-cause analyzer promotes it to the root even if it was
                // skipped (AlreadyPublished) and never became its own issue.
                blames: dep.iter().cloned().collect(),
            })
        }
        // (P2 structural) A resolution/ordering failure is NOT automatically
        // benign: split it. If the unresolved sibling IS in the publish plan, the
        // deps-first order publishes it earlier in the wave — the expected
        // mid-cascade condition, a note. If it's NOT in the plan (external /
        // `publish = false`) — or we can't attribute it (conservatively benign,
        // the legacy behaviour) — decide accordingly.
        PublishFailure::ResolveOrdering => {
            let dep = extract_resolve_dep(&stderr);
            match dep.as_deref() {
                Some(_) => {
                    if let Some(planned) =
                        skew_is_cascade_resolvable(dep.as_deref(), plan_publishes)
                    {
                        VerifyOutcome::CascadeResolvable {
                            dep: dep.unwrap_or_default(),
                            planned,
                        }
                    } else {
                        let d = dep.as_deref().unwrap_or("a dependency");
                        VerifyOutcome::Blocker(Issue {
                            repo: repo.to_string(),
                            krate: krate.to_string(),
                            kind: IssueKind::UnpublishableDep,
                            detail: format!(
                                "verify build can't resolve `{d}` from the registry, and it \
                                 is NOT in the publish plan (external, `publish = false`, or \
                                 a versioned edge that can never resolve)"
                            ),
                            fix: format!(
                                "add `{d}` to the release set, make it publishable, or drop \
                                 the version pin so the sibling dep isn't resolved"
                            ),
                            blames: dep.iter().cloned().collect(),
                        })
                    }
                }
                // Couldn't name the unresolved sibling → conservatively benign: an
                // un-attributed resolution failure during preflight dry-run is the
                // ordinary "sibling not published yet" condition.
                None => VerifyOutcome::Clean,
            }
        }
        // Non-member surfaces via check (1); metadata via check (2). Anything
        // unclassified is not a preflight blocker.
        _ => VerifyOutcome::Clean,
    }
}

/// Best-effort: pull the unresolved crate name out of a cargo RESOLUTION error —
/// `no matching package named `X`` or `failed to select a version for the
/// requirement `X = …``. `None` when no clear name is present.
fn extract_resolve_dep(stderr: &str) -> Option<String> {
    for marker in [
        "no matching package named `",
        "failed to select a version for the requirement `",
        "no matching package for `",
        "failed to select a version for `",
    ] {
        if let Some(i) = stderr.find(marker) {
            let rest = &stderr[i + marker.len()..];
            if let Some(end) = rest.find(|c| c == '`' || c == ' ') {
                let name = rest[..end].trim();
                if !name.is_empty() {
                    return Some(name.to_string());
                }
            }
        }
    }
    None
}

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

    struct FakeProbe(Option<bool>);
    impl RegistryProbe for FakeProbe {
        fn version_live(&self, _k: &str, _v: &str) -> Option<bool> {
            self.0
        }
    }

    /// A registry backend that reports a fixed published max version for every
    /// crate (or `None` = unpublished/unreachable).
    struct FakeBackend(Option<&'static str>);
    impl RegistryBackend for FakeBackend {
        fn published_max_version(&self, _krate: &str) -> Option<String> {
            self.0.map(str::to_string)
        }
    }

    #[test]
    fn missing_metadata_fields_respects_inheritance_and_license_file() {
        // Complete via explicit strings.
        let full: toml::Value = toml::from_str(
            r#"name = "x"
version = "0.1.0"
description = "d"
license = "Apache-2.0"
repository = "https://example.com"
"#,
        )
        .unwrap();
        assert!(missing_metadata_fields(&full, None).is_empty());

        // license-file satisfies the license requirement.
        let lf: toml::Value = toml::from_str(
            r#"name = "x"
description = "d"
license-file = "LICENSE"
repository = "r"
"#,
        )
        .unwrap();
        assert!(missing_metadata_fields(&lf, None).is_empty(), "license-file counts");

        // Missing everything.
        let bare: toml::Value = toml::from_str("name = \"x\"\nversion = \"0.1.0\"\n").unwrap();
        let miss = missing_metadata_fields(&bare, None);
        assert_eq!(miss, vec!["description", "license", "repository"]);

        // Inheritance: `field.workspace = true` + a workspace that provides it.
        let inh: toml::Value = toml::from_str(
            r#"name = "x"
description = { workspace = true }
license = { workspace = true }
repository = { workspace = true }
"#,
        )
        .unwrap();
        let wp: toml::Value = toml::from_str(
            r#"description = "d"
license = "Apache-2.0"
repository = "r"
"#,
        )
        .unwrap();
        assert!(missing_metadata_fields(&inh, Some(&wp)).is_empty(), "inherited fields present");
        // Inherits but the workspace DOESN'T provide it → still missing.
        assert_eq!(missing_metadata_fields(&inh, None), vec!["description", "license", "repository"]);
    }

    #[test]
    fn cross_repo_path_detection_and_masking_advisory() {
        // A consumer repo whose member manifest path-deps three cross-repo crates
        // (both path + version), one intra-repo sibling (path, no `..`), and one
        // plain registry dep (version only, no path). Only the CROSS-REPO
        // path+version entries are enumerated.
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        std::fs::write(
            root.join("Cargo.toml"),
            r#"[package]
name = "consumer"
version = "0.5.2"
edition = "2021"
description = "d"
license = "Apache-2.0"
repository = "https://example.com"

[dependencies]
funnel-core = { version = "0.1", path = "../funnel-core" }
nornir-testmatrix = { version = "0.2.4", path = "../edda/crates/nornir-testmatrix" }
sibling = { version = "0.1", path = "crates/sibling" }
serde = "1"

[build-dependencies]
dwarves = { version = "0.1", path = "../dwarves/crates/dwarfs" }
"#,
        )
        .unwrap();

        // PURE enumeration: 3 cross-repo (funnel-core, nornir-testmatrix, dwarves);
        // the intra-repo `sibling` and the path-less `serde` are excluded.
        let mut found = cross_repo_versioned_path_deps(root);
        found.sort();
        let names: Vec<&str> = found.iter().map(|(n, _, _)| n.as_str()).collect();
        assert_eq!(names, vec!["dwarves", "funnel-core", "nornir-testmatrix"]);
        assert!(!is_cross_repo_path("crates/sibling"));
        assert!(is_cross_repo_path("../funnel-core"));
        assert!(is_cross_repo_path("../edda/crates/nornir-testmatrix"));

        let plan = RepoPlan {
            name: "consumer".to_string(),
            root: root.to_path_buf(),
            publish_set: vec![("consumer".to_string(), "0.5.2".to_string())],
        };

        // Backend says the masked crate IS published (max 0.1.1) → an advisory note
        // per distinct cross-repo dep (3), each naming the fix.
        let notes = path_masks_published_notes(&plan, &FakeBackend(Some("0.1.1")));
        assert_eq!(notes.len(), 3, "one note per masked cross-repo dep");
        assert!(notes.iter().any(|n| n.contains("funnel-core")));
        assert!(notes.iter().all(|n| n.contains("[patch.crates-io]")), "fix named");
        assert!(notes.iter().all(|n| !n.contains("sibling")), "intra-repo sibling not flagged");

        // Backend says UNPUBLISHED (a genuine same-cascade sibling) → no false note.
        let none = path_masks_published_notes(&plan, &FakeBackend(None));
        assert!(none.is_empty(), "an unpublished cross-repo path is not masked");

        // These are advisory NOTES, never blockers: a run over this repo stays clean.
        let report = run(
            &[plan],
            true,
            VerifyMode::Off,
            &FakeProbe(Some(false)),
            &FakeBackend(Some("0.1.1")),
            &BTreeMap::new(),
        );
        assert!(report.is_clean(), "path-masking is advisory, not a blocker");
        assert!(
            report.notes.iter().any(|n| n.contains("resolves the LOCAL path")),
            "the masking advisory surfaced in the report notes"
        );
    }

    #[test]
    fn registry_skew_blocks_real_bump_but_not_keep() {
        // Real bump onto a version already live → blocker.
        let i = registry_skew_issue("facett", "facett-core", "0.4.53", false, Some(true));
        assert!(i.is_some());
        let i = i.unwrap();
        assert_eq!(i.kind, IssueKind::RegistrySkew);
        assert!(i.detail.contains("already live"));
        // Version absent → fine.
        assert!(registry_skew_issue("facett", "facett-core", "0.4.53", false, Some(false)).is_none());
        // keep-cascade → already-live is intentional, never a blocker.
        assert!(registry_skew_issue("facett", "facett-core", "0.4.53", true, Some(true)).is_none());
        // Couldn't probe → never a false block.
        assert!(registry_skew_issue("facett", "facett-core", "0.4.53", false, None).is_none());
    }

    /// The aggregation contract: multiple repos/crates, several issue classes, and
    /// notes fold into ONE report; `is_clean` is false; `format` names every fix.
    #[test]
    fn preflight_aggregates_all_issues_at_once() {
        // A one-crate repo missing ALL metadata + slated onto a live version.
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        std::fs::write(
            root.join("Cargo.toml"),
            "[package]\nname = \"solo\"\nversion = \"0.2.0\"\nedition = \"2021\"\n",
        )
        .unwrap();

        let plans = vec![RepoPlan {
            name: "solo".to_string(),
            root: root.to_path_buf(),
            publish_set: vec![("solo".to_string(), "0.2.0".to_string())],
        }];
        // Real bump (not keep), version reported already live, verify OFF (no cargo).
        // Backend reports NONE (unpublished) so the desync check stays silent here.
        let report = run(
            &plans,
            false,
            VerifyMode::Off,
            &FakeProbe(Some(true)),
            &FakeBackend(None),
            &BTreeMap::new(),
        );

        assert!(!report.is_clean(), "issues present ⇒ not clean");
        // solo lacks description/license/repository (one MissingMetadata issue) AND
        // its 0.2.0 is already live (one RegistrySkew issue).
        assert!(
            report.issues.iter().any(|i| i.kind == IssueKind::MissingMetadata),
            "metadata gap flagged"
        );
        assert!(
            report.issues.iter().any(|i| i.kind == IssueKind::RegistrySkew),
            "registry clash flagged"
        );
        assert_eq!(report.checked_crates, 1);
        let text = report.format();
        assert!(text.contains("blocking issue(s)"), "header counts issues: {text}");
        assert!(text.contains("fix:"), "every issue carries a fix: {text}");

        // Clean case: complete metadata, version not live.
        std::fs::write(
            root.join("Cargo.toml"),
            "[package]\nname = \"solo\"\nversion = \"0.2.0\"\n\
             description = \"d\"\nlicense = \"Apache-2.0\"\nrepository = \"https://example.com\"\n",
        )
        .unwrap();
        let clean = run(
            &plans,
            false,
            VerifyMode::Off,
            &FakeProbe(Some(false)),
            &FakeBackend(None),
            &BTreeMap::new(),
        );
        assert!(clean.is_clean(), "complete metadata + fresh version ⇒ clean: {:?}", clean.issues);
        assert!(clean.format().contains("preflight: clean"));
    }

    #[test]
    fn readiness_footer_consolidates_desync_into_one_command_per_repo() {
        let mut r = PreflightReport::default();
        for (k, cur, pubv) in [("facett-core", "0.1.10", "0.1.12"), ("facett-app", "0.1.10", "0.1.12")] {
            r.push(registry_desync_issue("facett", k, cur, Some(pubv)).unwrap());
        }
        let out = r.format();
        assert!(out.contains("RELEASE READINESS: 2 crate(s) across 1 repo(s)"), "{out}");
        // One consolidated command naming BOTH crates for the facett repo.
        assert!(out.contains("nornir release doctor --publish --bump patch"), "{out}");
        assert!(out.contains("--only facett-core"), "{out}");
        assert!(out.contains("--only facett-app"), "{out}");
        assert!(out.contains("git pull"), "stale-tree path offered: {out}");
        // A clean report has no readiness footer.
        assert!(!PreflightReport::default().format().contains("RELEASE READINESS"));
    }

    #[test]
    fn registry_desync_blocks_only_when_local_is_behind() {
        // Local BEHIND registry (the gatling incident) → blocker naming both versions.
        let i = registry_desync_issue("znippy", "gatling", "0.1.4", Some("0.1.6"));
        assert!(i.is_some());
        let i = i.unwrap();
        assert_eq!(i.kind, IssueKind::RegistryDesync);
        assert!(i.detail.contains("gatling 0.1.4"), "{}", i.detail);
        assert!(i.detail.contains("0.1.6"), "{}", i.detail);
        // The fix names the concrete lead-the-registry target (0.1.7 = one past
        // published 0.1.6) AND the exact command.
        assert!(i.fix.contains("0.1.7"), "fix names the bump target: {}", i.fix);
        assert!(i.fix.contains("--only gatling"), "fix names the exact command: {}", i.fix);
        assert!(i.fix.contains("git pull"), "fix offers the stale-tree path: {}", i.fix);
        assert_eq!(bump_past("0.1.6").as_deref(), Some("0.1.7"));
        assert_eq!(bump_past("not-semver"), None);
        // local AHEAD → WillPublish, not a blocker.
        assert!(registry_desync_issue("z", "gatling", "0.1.7", Some("0.1.6")).is_none());
        // local == published → AlreadyPublished, not a blocker.
        assert!(registry_desync_issue("z", "gatling", "0.1.6", Some("0.1.6")).is_none());
        // unpublished / unprobable → never a false block.
        assert!(registry_desync_issue("z", "gatling", "0.1.6", None).is_none());
    }

    /// End-to-end: a crate whose local version is BEHIND the backend's published
    /// max surfaces a RegistryDesync issue in the aggregated report.
    #[test]
    fn run_flags_registry_desync_from_backend() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        std::fs::write(
            root.join("Cargo.toml"),
            "[package]\nname = \"gatling\"\nversion = \"0.1.4\"\n\
             description = \"d\"\nlicense = \"Apache-2.0\"\nrepository = \"https://example.com\"\n",
        )
        .unwrap();
        let plans = vec![RepoPlan {
            name: "znippy".to_string(),
            root: root.to_path_buf(),
            publish_set: vec![("gatling".to_string(), "0.1.4".to_string())],
        }];
        // Registry already serves 0.1.6 — local 0.1.4 is a desync.
        let report = run(
            &plans,
            false,
            VerifyMode::Off,
            &FakeProbe(Some(false)),
            &FakeBackend(Some("0.1.6")),
            &BTreeMap::new(),
        );
        assert!(
            report.issues.iter().any(|i| i.kind == IssueKind::RegistryDesync),
            "desync flagged: {:?}",
            report.issues
        );
        assert!(report.format().contains("[desync]"), "{}", report.format());
    }

    /// BUMP-AWARE verify (item #7): a dep-API skew blaming a SIBLING that the
    /// cascade will itself publish forward is cascade-resolvable (not a blocker),
    /// while a skew blaming an EXTERNAL crate the cascade does NOT republish still
    /// blocks. This is the fixture the release needed `--force` for: crate B needs
    /// A's bumped version — with A in the planned publish map, that skew clears.
    #[test]
    fn skew_is_cascade_resolvable_only_for_in_release_deps() {
        // The planned publish map: the cascade bumps A and B to 0.1.2 (as under
        // `--bump patch`). `facett_core` is the classic ~30-member root.
        let mut planned = BTreeMap::new();
        planned.insert("facett_core".to_string(), "0.1.2".to_string());
        planned.insert("gatling".to_string(), "0.1.2".to_string());

        // A skew blaming `facett_core` (IN the release set) → cascade-resolvable,
        // carrying the version the cascade will publish it at.
        assert_eq!(
            skew_is_cascade_resolvable(Some("facett_core"), &planned),
            Some("0.1.2".to_string()),
            "a sibling the cascade republishes must NOT block preflight",
        );
        assert_eq!(
            skew_is_cascade_resolvable(Some("gatling"), &planned),
            Some("0.1.2".to_string()),
        );

        // A skew blaming an EXTERNAL crate (NOT in the release set) → real blocker.
        assert_eq!(
            skew_is_cascade_resolvable(Some("serde"), &planned),
            None,
            "a dep the cascade does NOT publish is still a genuine blocker",
        );
        // An unnamed dep → conservatively a blocker (can't prove the cascade covers it).
        assert_eq!(skew_is_cascade_resolvable(None, &planned), None);
        // Empty planned map (the strict / `--bump keep` path) → nothing is
        // suppressed, so every skew stays a blocker (preserves item #9/#10 resume
        // semantics where an AlreadyPublished/edited-but-unbumped dep must block).
        assert_eq!(
            skew_is_cascade_resolvable(Some("facett_core"), &BTreeMap::new()),
            None,
        );
    }

    /// The real production shape: the map is keyed by PACKAGE name (hyphens, from
    /// `Cargo.toml` `package.name`) but the blamed dep comes from a rustc verify
    /// error that cites the LIB name (hyphens → underscores). They must still match
    /// so a hyphenated sibling the cascade republishes (facett-core, its ~30
    /// dependents) is NOT falsely blocked. Regression guard for the false-blocker
    /// that refused release #2.
    #[test]
    fn skew_resolvable_across_hyphen_underscore_separators() {
        let mut planned = BTreeMap::new();
        planned.insert("facett-core".to_string(), "0.1.11".to_string()); // package name
        planned.insert("facett-graphview".to_string(), "0.1.11".to_string());
        // rustc blames the LIB name (underscores) — must resolve to the package.
        assert_eq!(
            skew_is_cascade_resolvable(Some("facett_core"), &planned),
            Some("0.1.11".to_string()),
            "lib-name `facett_core` must match package `facett-core`",
        );
        assert_eq!(
            skew_is_cascade_resolvable(Some("facett_graphview"), &planned),
            Some("0.1.11".to_string()),
        );
        // A genuinely absent crate still blocks.
        assert_eq!(skew_is_cascade_resolvable(Some("serde_json"), &planned), None);
    }

    /// The `VerifyOutcome` a bump-aware `verify_one` would emit for each class,
    /// asserted against the pure decision so the wiring in `run` (note vs push)
    /// is pinned without needing a real `cargo publish --dry-run`.
    #[test]
    fn cascade_resolvable_skew_becomes_a_note_not_an_issue() {
        let mut planned = BTreeMap::new();
        planned.insert("facett_core".to_string(), "0.1.11".to_string());

        // A DepApiSkew blaming an in-release sibling ⇒ CascadeResolvable ⇒ note.
        let resolvable = match skew_is_cascade_resolvable(Some("facett_core"), &planned) {
            Some(planned) => VerifyOutcome::CascadeResolvable {
                dep: "facett_core".to_string(),
                planned,
            },
            None => VerifyOutcome::Blocker(Issue {
                repo: "facett".into(),
                krate: "facett-graphnav".into(),
                kind: IssueKind::VerifyFailed,
                detail: String::new(),
                fix: String::new(),
                blames: vec!["facett_core".into()],
            }),
        };
        assert_eq!(
            resolvable,
            VerifyOutcome::CascadeResolvable {
                dep: "facett_core".to_string(),
                planned: "0.1.11".to_string(),
            },
        );

        // A DepApiSkew blaming an external crate ⇒ Blocker (VerifyFailed).
        let blocking = match skew_is_cascade_resolvable(Some("libc"), &planned) {
            Some(planned) => VerifyOutcome::CascadeResolvable { dep: "libc".into(), planned },
            None => VerifyOutcome::Blocker(Issue {
                repo: "facett".into(),
                krate: "facett-graphnav".into(),
                kind: IssueKind::VerifyFailed,
                detail: "x".into(),
                fix: "y".into(),
                blames: vec!["libc".into()],
            }),
        };
        assert!(
            matches!(blocking, VerifyOutcome::Blocker(ref i) if i.kind == IssueKind::VerifyFailed),
            "external dep skew must remain a blocker",
        );
    }

    /// P1#2: a `[dev-dependencies]` entry with BOTH `path=` and `version=` is
    /// flagged (it fail-stops `cargo publish` on an unpublished sibling); a
    /// path-only or version-only dev-dep is NOT.
    #[test]
    fn versioned_path_dev_deps_flags_only_path_plus_version() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        std::fs::write(
            root.join("Cargo.toml"),
            "[package]\nname = \"facett-form\"\nversion = \"0.1.0\"\n\
             [dev-dependencies]\n\
             facett-wasm-demo = { path = \"../wasm-demo\", version = \"0.1\" }\n\
             pure-registry = \"1.0\"\n\
             path-only = { path = \"../other\" }\n",
        )
        .unwrap();
        let got = versioned_path_dev_deps(root, "facett-form");
        assert_eq!(got, vec![("facett-wasm-demo".to_string(), "0.1".to_string())]);
    }

    /// P2 table (a/b/d/f): the additional manifest gaps each surface with a fix.
    #[test]
    fn extra_manifest_gaps_covers_readme_keywords_feature_and_unpublishable_dep() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let manifest = root.join("Cargo.toml");
        std::fs::write(
            &manifest,
            "[package]\nname = \"c\"\nversion = \"0.1.0\"\n\
             readme = \"README.md\"\n\
             keywords = [\"ok\", \"this-keyword-is-way-too-long-to-be-valid\"]\n\
             [dependencies]\n\
             priv-sibling = { path = \"../priv\", version = \"0.1\" }\n\
             [features]\n\
             extra = [\"dep:absent\"]\n",
        )
        .unwrap();
        // README.md deliberately absent.
        let mut pf: BTreeSet<String> = BTreeSet::new();
        pf.insert("priv-sibling".to_string());
        let doc: toml::Value = std::fs::read_to_string(&manifest).unwrap().parse().unwrap();
        let gaps = extra_manifest_gaps(&manifest, &doc, &pf);
        assert!(gaps.iter().any(|(k, d, _)| *k == IssueKind::BadManifest && d.contains("readme")));
        assert!(gaps.iter().any(|(k, d, _)| *k == IssueKind::BadManifest && d.contains("keyword")));
        assert!(gaps.iter().any(|(k, d, _)| *k == IssueKind::BadManifest && d.contains("dep:absent")));
        assert!(gaps
            .iter()
            .any(|(k, d, _)| *k == IssueKind::UnpublishableDep && d.contains("priv-sibling")));
        // A complete manifest (readme present, ≤5 valid keywords, no bad edges) is clean.
        std::fs::write(root.join("README.md"), "x").unwrap();
        std::fs::write(
            &manifest,
            "[package]\nname = \"c\"\nversion = \"0.1.0\"\nreadme = \"README.md\"\n\
             keywords = [\"ok\"]\n",
        )
        .unwrap();
        let doc2: toml::Value = std::fs::read_to_string(&manifest).unwrap().parse().unwrap();
        assert!(extra_manifest_gaps(&manifest, &doc2, &BTreeSet::new()).is_empty());
    }

    /// The tarball-size check flags a `.crate` over 10 MiB in target/package and is
    /// clean for a small one / a name that only prefix-matches.
    #[test]
    fn oversized_tarball_issue_flags_over_10mib() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let pkgdir = root.join("target").join("package");
        std::fs::create_dir_all(&pkgdir).unwrap();
        // An 11 MiB `foo-1.0.0.crate` → flagged.
        std::fs::write(pkgdir.join("foo-1.0.0.crate"), vec![0u8; 11 * 1024 * 1024]).unwrap();
        let issue = oversized_tarball_issue("r", root, "foo");
        assert!(issue.is_some());
        assert_eq!(issue.unwrap().kind, IssueKind::OversizedTarball);
        // A different crate whose name only prefixes → not matched (no tarball).
        assert!(oversized_tarball_issue("r", root, "fo").is_none());
        // A small tarball → clean.
        std::fs::write(pkgdir.join("bar-2.0.0.crate"), vec![0u8; 1024]).unwrap();
        assert!(oversized_tarball_issue("r", root, "bar").is_none());
        // No package dir at all → best-effort None.
        assert!(oversized_tarball_issue("r", Path::new("/nonexistent-xyz"), "foo").is_none());
    }

    /// Two crates in the release sharing a `links` value both get flagged; a unique
    /// links value is clean.
    #[test]
    fn duplicate_links_issues_flags_shared_links_value() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        std::fs::write(root.join("Cargo.toml"), "[workspace]\nmembers=[\"a\",\"b\",\"c\"]\n").unwrap();
        for (name, links) in [("a", "zlib"), ("b", "zlib"), ("c", "lz4")] {
            std::fs::create_dir(root.join(name)).unwrap();
            std::fs::write(
                root.join(name).join("Cargo.toml"),
                format!("[package]\nname = \"{name}\"\nversion = \"0.1.0\"\nlinks = \"{links}\"\n"),
            )
            .unwrap();
        }
        let plans = vec![RepoPlan {
            name: "r".to_string(),
            root: root.to_path_buf(),
            publish_set: vec![
                ("a".to_string(), "0.1.0".to_string()),
                ("b".to_string(), "0.1.0".to_string()),
                ("c".to_string(), "0.1.0".to_string()),
            ],
        }];
        let issues = duplicate_links_issues(&plans);
        // a and b collide on "zlib"; c is unique.
        let flagged: BTreeSet<String> = issues.iter().map(|i| i.krate.clone()).collect();
        assert_eq!(flagged, ["a", "b"].iter().map(|s| s.to_string()).collect());
        assert!(issues.iter().all(|i| i.detail.contains("zlib")));
    }

    /// Feature-reference validation: a bare name that is neither a feature nor a
    /// dependency, and a `foo/bar` whose `foo` is no dependency, are flagged; valid
    /// references (another feature, a real dep, a real dep-feature) are clean.
    #[test]
    fn extra_manifest_gaps_flags_dangling_feature_refs() {
        let dir = tempfile::tempdir().unwrap();
        let manifest = dir.path().join("Cargo.toml");
        std::fs::write(
            &manifest,
            "[package]\nname = \"c\"\nversion = \"0.1.0\"\n\
             [dependencies]\n\
             serde = { version = \"1\", optional = true }\n\
             [features]\n\
             default = [\"real\"]\n\
             real = [\"serde\", \"serde/derive\"]\n\
             broken = [\"ghost\", \"nodep/feat\"]\n",
        )
        .unwrap();
        let doc: toml::Value = std::fs::read_to_string(&manifest).unwrap().parse().unwrap();
        let gaps = extra_manifest_gaps(&manifest, &doc, &BTreeSet::new());
        // `ghost` (bare, unknown) and `nodep/feat` (unknown dep) flagged.
        assert!(gaps.iter().any(|(_, d, _)| d.contains("includes `ghost`")), "{gaps:?}");
        assert!(gaps.iter().any(|(_, d, _)| d.contains("nodep/feat") && d.contains("`nodep`")));
        // `real`, `serde`, `serde/derive` are all valid → not flagged.
        assert!(!gaps.iter().any(|(_, d, _)| d.contains("`real`") && d.contains("includes")));
        assert!(!gaps.iter().any(|(_, d, _)| d.contains("serde/derive")));
    }

    /// An over-long description and a cross-registry dep are each flagged.
    #[test]
    fn extra_manifest_gaps_flags_long_description_and_cross_registry_dep() {
        let dir = tempfile::tempdir().unwrap();
        let manifest = dir.path().join("Cargo.toml");
        let long = "x".repeat(1001);
        std::fs::write(
            &manifest,
            format!(
                "[package]\nname = \"c\"\nversion = \"0.1.0\"\ndescription = \"{long}\"\n\
                 [dependencies]\n\
                 priv = {{ version = \"1\", registry = \"my-corp\" }}\n\
                 ok = \"1\"\n"
            ),
        )
        .unwrap();
        let doc: toml::Value = std::fs::read_to_string(&manifest).unwrap().parse().unwrap();
        let gaps = extra_manifest_gaps(&manifest, &doc, &BTreeSet::new());
        assert!(gaps.iter().any(|(_, d, _)| d.contains("description is") && d.contains("1001")));
        assert!(gaps.iter().any(|(_, d, _)| d.contains("`priv`") && d.contains("another registry")));
        assert!(!gaps.iter().any(|(_, d, _)| d.contains("`ok`")));
    }

    /// Unknown `edition` and malformed `rust-version` are flagged; known/valid
    /// values are clean.
    #[test]
    fn extra_manifest_gaps_flags_edition_and_rust_version() {
        let dir = tempfile::tempdir().unwrap();
        let manifest = dir.path().join("Cargo.toml");
        std::fs::write(
            &manifest,
            "[package]\nname = \"c\"\nversion = \"0.1.0\"\n\
             edition = \"2019\"\nrust-version = \"^1.74\"\n",
        )
        .unwrap();
        let doc: toml::Value = std::fs::read_to_string(&manifest).unwrap().parse().unwrap();
        let gaps = extra_manifest_gaps(&manifest, &doc, &BTreeSet::new());
        assert!(gaps.iter().any(|(_, d, _)| d.contains("edition") && d.contains("2019")));
        assert!(gaps.iter().any(|(_, d, _)| d.contains("rust-version")));
        // Valid values clean.
        std::fs::write(
            &manifest,
            "[package]\nname = \"c\"\nversion = \"0.1.0\"\n\
             edition = \"2021\"\nrust-version = \"1.74\"\n",
        )
        .unwrap();
        let doc2: toml::Value = std::fs::read_to_string(&manifest).unwrap().parse().unwrap();
        assert!(extra_manifest_gaps(&manifest, &doc2, &BTreeSet::new()).is_empty());
    }

    /// A non-http(s) `repository`/`homepage`/`documentation` URL is flagged; a
    /// valid https URL is clean.
    #[test]
    fn extra_manifest_gaps_flags_non_http_url_fields() {
        let dir = tempfile::tempdir().unwrap();
        let manifest = dir.path().join("Cargo.toml");
        std::fs::write(
            &manifest,
            "[package]\nname = \"c\"\nversion = \"0.1.0\"\n\
             repository = \"github.com/x/y\"\n\
             homepage = \"https://ok.example\"\n\
             documentation = \"git@host:z\"\n",
        )
        .unwrap();
        let doc: toml::Value = std::fs::read_to_string(&manifest).unwrap().parse().unwrap();
        let gaps = extra_manifest_gaps(&manifest, &doc, &BTreeSet::new());
        let urls: Vec<&String> = gaps
            .iter()
            .filter(|(k, d, _)| *k == IssueKind::BadManifest && d.contains("http://"))
            .map(|(_, d, _)| d)
            .collect();
        assert_eq!(urls.len(), 2, "repository + documentation, not homepage: {urls:?}");
        assert!(urls.iter().any(|d| d.contains("repository")));
        assert!(urls.iter().any(|d| d.contains("documentation")));
        assert!(!urls.iter().any(|d| d.contains("homepage")));
    }

    /// A declared `license-file` that is missing is flagged; a non-UTF-8 `readme`
    /// is flagged; present-and-UTF-8 files are clean.
    #[test]
    fn declared_file_gaps_missing_and_non_utf8() {
        let dir = tempfile::tempdir().unwrap();
        let manifest = dir.path().join("Cargo.toml");
        std::fs::write(
            &manifest,
            "[package]\nname = \"c\"\nversion = \"0.1.0\"\n\
             readme = \"README.md\"\nlicense-file = \"LICENSE\"\n",
        )
        .unwrap();
        // README exists but is NOT valid UTF-8; LICENSE is absent.
        std::fs::write(dir.path().join("README.md"), [0xff, 0xfe, 0x00]).unwrap();
        let doc: toml::Value = std::fs::read_to_string(&manifest).unwrap().parse().unwrap();
        let gaps = extra_manifest_gaps(&manifest, &doc, &BTreeSet::new());
        assert!(gaps
            .iter()
            .any(|(k, d, _)| *k == IssueKind::BadManifest && d.contains("readme") && d.contains("UTF-8")));
        assert!(gaps.iter().any(|(k, d, _)| *k == IssueKind::BadManifest
            && d.contains("license-file")
            && d.contains("does not exist")));
        // Both present + UTF-8 → clean.
        std::fs::write(dir.path().join("README.md"), "hello").unwrap();
        std::fs::write(dir.path().join("LICENSE"), "MIT text").unwrap();
        let doc2: toml::Value = std::fs::read_to_string(&manifest).unwrap().parse().unwrap();
        assert!(extra_manifest_gaps(&manifest, &doc2, &BTreeSet::new()).is_empty());
    }

    /// Keyword rules (start-alphanumeric, ≤20, charset incl. `+`) and the >5
    /// category hard-error are each flagged; a valid set is clean.
    #[test]
    fn extra_manifest_gaps_keyword_start_and_category_count() {
        let dir = tempfile::tempdir().unwrap();
        let manifest = dir.path().join("Cargo.toml");
        std::fs::write(
            &manifest,
            "[package]\nname = \"c\"\nversion = \"0.1.0\"\n\
             keywords = [\"-badstart\", \"c++\", \"ok1\"]\n\
             categories = [\"a\",\"b\",\"c\",\"d\",\"e\",\"f\"]\n",
        )
        .unwrap();
        let doc: toml::Value = std::fs::read_to_string(&manifest).unwrap().parse().unwrap();
        let gaps = extra_manifest_gaps(&manifest, &doc, &BTreeSet::new());
        // `-badstart` flagged (bad start); `c++`/`ok1` fine.
        assert!(gaps.iter().any(|(k, d, _)| *k == IssueKind::BadManifest
            && d.contains("keyword")
            && d.contains("-badstart")));
        assert!(!gaps.iter().any(|(_, d, _)| d.contains("\"c++\"")), "c++ is a valid keyword");
        // 6 categories → the >5 hard error.
        assert!(gaps.iter().any(|(k, d, _)| *k == IssueKind::BadManifest
            && d.contains("categories")
            && d.contains("at most 5")));
    }

    /// Crate-name validation: real nordisk names pass; the crates.io-reject classes
    /// (too long, bad first char, bad charset, reserved) are each flagged.
    #[test]
    fn crate_name_problem_flags_the_reject_classes() {
        for ok in ["facett-core", "gatling", "znippy_common", "a", "x2", "serde"] {
            assert!(crate_name_problem(ok).is_none(), "valid rejected: {ok}");
        }
        assert!(crate_name_problem("").is_some());
        assert!(crate_name_problem(&"a".repeat(65)).unwrap().contains("at most 64"));
        assert!(crate_name_problem(&"a".repeat(64)).is_none(), "exactly 64 is ok");
        assert!(crate_name_problem("2fast").unwrap().contains("ASCII letter"), "leading digit");
        assert!(crate_name_problem("-lead").unwrap().contains("ASCII letter"), "leading dash");
        assert!(crate_name_problem("_lead").unwrap().contains("ASCII letter"), "leading underscore");
        assert!(crate_name_problem("has space").unwrap().contains("contains"));
        assert!(crate_name_problem("has.dot").unwrap().contains("contains"));
        assert!(crate_name_problem("café").unwrap().contains("contains"), "non-ASCII");
        assert!(crate_name_problem("nul").unwrap().contains("reserved"));
        assert!(crate_name_problem("COM1").unwrap().contains("reserved"));
    }

    /// SPDX validator: valid expressions pass; the common real mistakes (the
    /// deprecated `/`, dangling operators, unbalanced parens, bad-charset tokens)
    /// are confidently flagged; well-formed exotic ids degrade to clean (no false
    /// block).
    #[test]
    fn spdx_license_problem_flags_only_confident_invalids() {
        // Valid — must NOT be flagged.
        for ok in [
            "MIT",
            "Apache-2.0",
            "MIT OR Apache-2.0",
            "Apache-2.0 WITH LLVM-exception",
            "(MIT OR Apache-2.0) AND BSD-3-Clause",
            "GPL-3.0-or-later",
            "GPL-2.0+",
            "LicenseRef-My-Custom",
            "0BSD",
        ] {
            assert!(spdx_license_problem(ok).is_none(), "valid rejected: {ok}");
        }
        // Confidently invalid — must be flagged.
        assert!(spdx_license_problem("").is_some());
        assert!(spdx_license_problem("   ").is_some());
        assert!(spdx_license_problem("MIT/Apache-2.0").unwrap().contains("deprecated `/`"));
        assert!(spdx_license_problem("MIT OR").is_some(), "trailing operator");
        assert!(spdx_license_problem("OR MIT").is_some(), "leading operator");
        assert!(spdx_license_problem("MIT Apache-2.0").is_some(), "two ids, no operator");
        assert!(spdx_license_problem("(MIT OR Apache-2.0").is_some(), "unbalanced");
        assert!(spdx_license_problem("MIT AND (Apache-2.0))").is_some(), "unbalanced");
        assert!(spdx_license_problem("M IT_bad").is_some() || spdx_license_problem("MIT_bad").is_some());
    }

    /// End-to-end: an explicit bad license AND a workspace-inherited bad license
    /// each surface as a BadManifest license issue; a valid one is clean.
    #[test]
    fn license_spdx_issue_catches_explicit_and_inherited() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        // Root workspace provides a BAD inherited license; the member inherits it.
        std::fs::write(
            root.join("Cargo.toml"),
            "[workspace]\nmembers = [\"m\"]\n[workspace.package]\nlicense = \"MIT/Apache-2.0\"\n",
        )
        .unwrap();
        std::fs::create_dir(root.join("m")).unwrap();
        std::fs::write(
            root.join("m/Cargo.toml"),
            "[package]\nname = \"m\"\nversion = \"0.1.0\"\nlicense = { workspace = true }\n",
        )
        .unwrap();
        let issue = license_spdx_issue(root, "m");
        assert!(issue.is_some(), "inherited bad license must be caught");
        assert!(issue.unwrap().0.contains("deprecated `/`"));

        // A valid explicit license is clean.
        std::fs::write(
            root.join("m/Cargo.toml"),
            "[package]\nname = \"m\"\nversion = \"0.1.0\"\nlicense = \"MIT OR Apache-2.0\"\n",
        )
        .unwrap();
        assert!(license_spdx_issue(root, "m").is_none());
    }

    /// Wildcard `*` deps (crates.io hard-reject) are flagged on normal, build, and
    /// target-specific deps — but NOT a precise wildcard (`1.*`) nor a dev-dep (which
    /// is stripped on publish).
    #[test]
    fn extra_manifest_gaps_flags_bare_wildcard_deps() {
        let dir = tempfile::tempdir().unwrap();
        let manifest = dir.path().join("Cargo.toml");
        std::fs::write(
            &manifest,
            "[package]\nname = \"c\"\nversion = \"0.1.0\"\n\
             [dependencies]\n\
             star = \"*\"\n\
             star-tbl = { version = \"*\" }\n\
             precise = \"1.*\"\n\
             fine = \"1.2\"\n\
             [build-dependencies]\n\
             bstar = \"*\"\n\
             [target.'cfg(unix)'.dependencies]\n\
             tstar = \"*\"\n\
             [dev-dependencies]\n\
             devstar = \"*\"\n",
        )
        .unwrap();
        let doc: toml::Value = std::fs::read_to_string(&manifest).unwrap().parse().unwrap();
        let gaps = extra_manifest_gaps(&manifest, &doc, &BTreeSet::new());
        let wildcards: Vec<&String> = gaps
            .iter()
            .filter(|(k, d, _)| *k == IssueKind::BadManifest && d.contains("wildcard"))
            .map(|(_, d, _)| d)
            .collect();
        // star, star-tbl, bstar, tstar — four; NOT precise/fine/devstar.
        assert_eq!(wildcards.len(), 4, "got {wildcards:?}");
        assert!(wildcards.iter().any(|d| d.contains("`star`")));
        assert!(wildcards.iter().any(|d| d.contains("`star-tbl`")));
        assert!(wildcards.iter().any(|d| d.contains("`bstar`")));
        assert!(wildcards.iter().any(|d| d.contains("`tstar`")));
        assert!(!wildcards.iter().any(|d| d.contains("devstar")), "dev-deps are stripped");
        assert!(!wildcards.iter().any(|d| d.contains("precise")), "1.* is allowed");
    }

    /// A git dep WITHOUT a `version =` is flagged (cargo strips git on publish and
    /// rejects the version-less dep); a git dep WITH a version, and a plain
    /// registry/path dep, are not flagged by this rule.
    #[test]
    fn extra_manifest_gaps_flags_git_dep_without_version() {
        let dir = tempfile::tempdir().unwrap();
        let manifest = dir.path().join("Cargo.toml");
        std::fs::write(
            &manifest,
            "[package]\nname = \"c\"\nversion = \"0.1.0\"\n\
             [dependencies]\n\
             gitless = { git = \"https://example.com/x\" }\n\
             gitver = { git = \"https://example.com/y\", version = \"1.2\" }\n\
             plain = \"1.0\"\n",
        )
        .unwrap();
        let doc: toml::Value = std::fs::read_to_string(&manifest).unwrap().parse().unwrap();
        let gaps = extra_manifest_gaps(&manifest, &doc, &BTreeSet::new());
        let git_gaps: Vec<&String> = gaps
            .iter()
            .filter(|(k, d, _)| *k == IssueKind::UnpublishableDep && d.contains("git dependency"))
            .map(|(_, d, _)| d)
            .collect();
        assert_eq!(git_gaps.len(), 1, "only the version-less git dep: {git_gaps:?}");
        assert!(git_gaps[0].contains("`gitless`"));
    }

    /// P2 structural: the resolve-dep extractor names the unresolved sibling from
    /// the two cargo resolution-error shapes.
    #[test]
    fn extract_resolve_dep_names_the_unresolved_sibling() {
        assert_eq!(
            extract_resolve_dep("error: no matching package named `facett-wasm-demo` found"),
            Some("facett-wasm-demo".to_string())
        );
        assert_eq!(
            extract_resolve_dep(
                "error: failed to select a version for the requirement `nornir-jobs = \"^0.1\"`"
            ),
            Some("nornir-jobs".to_string())
        );
        assert_eq!(extract_resolve_dep("some unrelated failure"), None);
    }

    // ── DRY-RUN == REAL preflight parity (the false-all-clear defect) ──────────
    //
    // Before the fix, `--publish-dry-run` ran the preflight with VerifyMode::Off,
    // so the CORRECTNESS class (dep-API-skew verify + same-version source-drift)
    // was silently skipped — a dry-run printed "all clear" while the real
    // `--publish` fail-stopped on the identical blocker (the draupnir
    // "edited-but-AlreadyPublished dep" trap). These tests pin the two invariants:
    //   1. the preflight verify mode is IDENTICAL for dry-run and real (both On), so
    //      the correctness check runs UP FRONT in both;
    //   2. a correctness blocker EARLY-FAILS the gate in both modes, is NOT
    //      overridable by --force, and so no artifact/mutation step is entered.

    /// A published-source oracle that hands back a FIXED map — used to force a
    /// [`DriftVerdict::Drifted`] (the published copy differs from local source).
    struct FakeOracle(BTreeMap<String, Vec<u8>>);
    impl crate::release::source_drift::PublishedSourceOracle for FakeOracle {
        fn published_sources(
            &self,
            _krate: &str,
            _version: &str,
        ) -> Result<BTreeMap<String, Vec<u8>>, String> {
            Ok(self.0.clone())
        }
    }

    #[test]
    fn preflight_verify_mode_is_identical_for_dry_run_and_real() {
        // The whole fix: the mode does NOT depend on --publish-dry-run. If a future
        // change reintroduces the split (dry-run ⇒ Off), this fails — that split
        // was the false-all-clear defect.
        assert_eq!(preflight_verify_mode(true), VerifyMode::On, "dry-run must verify");
        assert_eq!(preflight_verify_mode(false), VerifyMode::On, "real must verify");
        assert_eq!(
            preflight_verify_mode(true),
            preflight_verify_mode(false),
            "dry-run and real preflight MUST use the SAME verify mode"
        );
    }

    #[test]
    fn correctness_blocker_classification_is_single_sourced() {
        // The CORRECTNESS class (un-forceable): dep-API skew, caught either by the
        // compile (VerifyFailed) or by content-diff (StaleSourceDrift).
        assert!(IssueKind::VerifyFailed.is_correctness_blocker());
        assert!(IssueKind::StaleSourceDrift.is_correctness_blocker());
        // Everything else is SOFT (waveable by --force).
        for k in [
            IssueKind::NonMember,
            IssueKind::MissingMetadata,
            IssueKind::RegistrySkew,
            IssueKind::RegistryDesync,
            IssueKind::VersionedPathDevDep,
            IssueKind::UnpublishableDep,
            IssueKind::BadManifest,
            IssueKind::InvalidCrateName,
            IssueKind::OversizedTarball,
        ] {
            assert!(!k.is_correctness_blocker(), "{k:?} must be soft");
        }
    }

    #[test]
    fn gate_early_fails_on_correctness_blocker_and_force_cannot_override() {
        // A report carrying ONE StaleSourceDrift blocker (the "edited-but-unbumped
        // drifted dep" correctness class).
        let mut report = PreflightReport::default();
        report.checked_crates = 1;
        report.push(Issue {
            repo: "draupnir".into(),
            krate: "draupnir".into(),
            kind: IssueKind::StaleSourceDrift,
            detail: "local source differs from published draupnir@0.1.4 (new API)".into(),
            fix: "bump draupnir".into(),
            blames: vec!["draupnir".into()],
        });

        // Un-forceable, and IDENTICAL whether or not --force is passed — this is the
        // verdict BOTH the dry-run and the real publish act on (early-fail before any
        // mutation).
        assert_eq!(gate_verdict(&report, false), GateVerdict::BlockCorrectness(1));
        assert_eq!(
            gate_verdict(&report, true),
            GateVerdict::BlockCorrectness(1),
            "--force must NEVER wave through a correctness blocker"
        );

        // A SOFT-only report: blocks without --force, proceeds (waved) with it.
        let mut soft = PreflightReport::default();
        soft.push(Issue {
            repo: "solo".into(),
            krate: "solo".into(),
            kind: IssueKind::MissingMetadata,
            detail: "missing license".into(),
            fix: "add license".into(),
            blames: Vec::new(),
        });
        assert_eq!(gate_verdict(&soft, false), GateVerdict::BlockSoft(1));
        assert_eq!(gate_verdict(&soft, true), GateVerdict::Proceed { forced_soft: 1 });

        // A clean report proceeds with nothing waved.
        assert_eq!(
            gate_verdict(&PreflightReport::default(), false),
            GateVerdict::Proceed { forced_soft: 0 }
        );
    }

    #[test]
    fn drifted_unpublished_dep_yields_correctness_blocker_in_both_modes() {
        // FIXTURE: a to-publish crate whose target version is ALREADY live but whose
        // local source DRIFTED from the published copy (a new API added locally, not
        // yet republished) — the exact draupnir shape. The shared preflight
        // (`source_drift_issues`, run under the IDENTICAL verify mode both modes get)
        // must yield a StaleSourceDrift CORRECTNESS blocker, and the gate must
        // early-fail identically for dry-run and real.
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        std::fs::write(
            root.join("Cargo.toml"),
            r#"[package]
name = "draupnir"
version = "0.1.4"
edition = "2021"
description = "d"
license = "Apache-2.0"
repository = "https://example.com"
"#,
        )
        .unwrap();
        std::fs::create_dir_all(root.join("src")).unwrap();
        // Local source carries the NEW API.
        std::fs::write(root.join("src/lib.rs"), "pub fn new_api() {}\n").unwrap();

        let plans = vec![RepoPlan {
            name: "draupnir".to_string(),
            root: root.to_path_buf(),
            publish_set: vec![("draupnir".to_string(), "0.1.4".to_string())],
        }];

        // The published 0.1.4 copy LACKS the new API → drift. (The manifest key may
        // differ from the canonicalised local one too; either way the verdict is
        // Drifted, which is all this test needs.)
        let mut published = BTreeMap::new();
        published.insert("src/lib.rs".to_string(), b"// old API only\n".to_vec());
        let oracle = FakeOracle(published);

        // version_live == Some(true): the target IS on the registry, so the
        // same-version drift pass fires (mirrors AlreadyPublished draupnir@0.1.4).
        let (issues, _notes) =
            source_drift_issues(&plans, &FakeProbe(Some(true)), &oracle);

        // The shared preflight yields the CORRECTNESS blocker (skip if the local
        // env has no cargo to enumerate the shipped file set — never a false pass:
        // in that case assess() returns Unknown, a note, not a spurious blocker).
        let drift: Vec<&Issue> =
            issues.iter().filter(|i| i.kind == IssueKind::StaleSourceDrift).collect();
        if drift.is_empty() {
            eprintln!(
                "note: source-drift produced no blocker (cargo `package --list` unavailable?) — \
                 classification/gate parity still asserted below"
            );
        } else {
            assert_eq!(drift.len(), 1, "one drifted crate ⇒ one correctness blocker");
            assert!(drift[0].kind.is_correctness_blocker());
        }

        // Fold into a report and assert the up-front gate EARLY-FAILS identically for
        // BOTH modes — the verify mode is the same (invariant 1), so both modes reach
        // this gate with the same issues; the correctness blocker is un-forceable
        // (invariant 2). We simulate the presence of the blocker (using the produced
        // one, or a synthesized one if cargo was unavailable) so the gate contract is
        // exercised deterministically regardless of the local toolchain.
        let mut report = PreflightReport::default();
        report.checked_crates = 1;
        if let Some(i) = drift.first() {
            report.push((*i).clone());
        } else {
            report.push(Issue {
                repo: "draupnir".into(),
                krate: "draupnir".into(),
                kind: IssueKind::StaleSourceDrift,
                detail: "drift".into(),
                fix: "bump".into(),
                blames: vec!["draupnir".into()],
            });
        }
        // dry-run (publish_dry_run = true) and real (false) resolve to the SAME
        // verify mode → the SAME preflight → the SAME gate verdict. Neither proceeds
        // to the cascade, so no bump/branch/package/publish/push is entered.
        assert_eq!(preflight_verify_mode(true), preflight_verify_mode(false));
        assert_eq!(gate_verdict(&report, false), GateVerdict::BlockCorrectness(1));
        assert_eq!(
            gate_verdict(&report, true),
            GateVerdict::BlockCorrectness(1),
            "dry-run and real BOTH block; --force cannot override"
        );
    }
}