autumn-web 0.9.0

//! Build-time cache-coherence proof (issue #1716).
//!
//! Autumn owns both ends of the staleness dependency: cached reads flow through
//! [`#[cached]`](autumn_macros::cached), the fragment cache and the read-through
//! cache; writes flow through [`#[repository]`](autumn_macros_repository::repository).
//! Until now nothing linked the two, so a cached value derived from `Post` rows
//! kept being served after a `PostRepository::save` inserted a new one and the
//! build said nothing.
//!
//! This module is the link. The macros publish what they know as `inventory`
//! descriptors — which models a cached read is derived from, which model each
//! repository write mutates, and which reads a write declares it invalidates —
//! and [`check`] proves, over the whole binary, that no mutation can strand a
//! cached value.
//!
//! # The rule
//!
//! A `(read, mutation)` pair is a **violation** when all four hold:
//!
//! 1. the read's dependency set is known (its provenance is not
//!    [`DependencyProvenance::Undetermined`]),
//! 2. neither side is acknowledged-stale,
//! 3. the read's dependency set intersects the mutation's model, and
//! 4. the mutation does not declare an invalidation naming that read.
//!
//! Reads whose dependency set could **not** be established are never failed by
//! the default gate — a checker that cries wolf gets deleted from CI. They are
//! reported as [`undetermined_reads`] instead, and `--strict` turns them into a
//! failure for an app that wants the stronger posture.
//!
//! # Identity
//!
//! A cached read is identified by its **cache-key namespace** —
//! `concat!(module_path!(), "::", <fn name>)`, the exact prefix
//! [`make_cache_key`](super::make_cache_key) already stamps on every entry. That
//! keeps the manifest's identity and the runtime's key space the same string.
//!
//! Models are matched on their **last path segment**, because the two sides
//! learn the name differently: a `#[repository]` always has the model *type* in
//! scope and publishes `core::any::type_name` (`blog::models::Post`), while a
//! dependency recovered from a `#[cached]` body may only be the bare ident
//! (`Post`). Two same-named models in different modules therefore
//! over-approximate — a false *failure*, never a false pass — which is the safe
//! direction for a coherence gate and is dischargeable with
//! `acknowledge_stale`.
//!
//! See `docs/guide/cache-coherence.md`.

use std::fmt::Write as _;

use serde::{Deserialize, Serialize};

/// Schema version of the emitted cache-coherence manifest. Bumped only on
/// breaking changes to the document shape.
pub const MANIFEST_SCHEMA_VERSION: u32 = 1;

/// Machine-readable stdout marker preceding the manifest JSON emitted by the
/// `AUTUMN_DUMP_CACHE_COHERENCE=1` dump mode.
///
/// A process-boundary protocol: `autumn cache audit` runs the built binary as a
/// child and scans its stdout for this marker, so an app that prints anything
/// else during startup cannot corrupt the parse.
pub const COHERENCE_MANIFEST_MARKER: &str = "[autumn:cache-coherence] ";

// ── Descriptors published by the macros ──────────────────────────────

/// Which cache surface a read is served from.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ReadKind {
    /// A `#[cached]` memoized function.
    Cached,
    /// A `cache_fragment(...)` template fragment.
    Fragment,
    /// A `get_or_compute(...)` read-through entry.
    ReadThrough,
}

impl ReadKind {
    /// The stable manifest spelling.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Cached => "cached",
            Self::Fragment => "fragment",
            Self::ReadThrough => "read_through",
        }
    }
}

/// How a cached read's dependency set was established.
///
/// The classes are strictly decreasing in strength, and an entry may only claim
/// the class it can defend — the same discipline
/// `docs/guide/security-posture-manifest.md` applies to the security manifest.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DependencyProvenance {
    /// The developer wrote `#[cached(reads(Post, Comment))]`. The strongest
    /// claim: the set is exactly what was declared.
    Declared,
    /// Recovered by the macro from the function body (repository types, model
    /// finder calls). Sound for what it found; it can miss a dependency reached
    /// through a helper it cannot read.
    Derived,
    /// Nothing could be established. The read is reported, never gated.
    Undetermined,
}

impl DependencyProvenance {
    /// The stable manifest spelling.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Declared => "declared",
            Self::Derived => "derived",
            Self::Undetermined => "undetermined",
        }
    }
}

/// A cached read, as published by the macro that created it.
///
/// `'static` and const-constructible so it can be `inventory::submit!`ed from
/// macro-generated code with no runtime cost.
pub struct CachedReadDescriptor {
    /// The cache-key namespace: `concat!(module_path!(), "::", <fn name>)`.
    pub id: &'static str,
    /// Which cache surface serves this read.
    pub kind: ReadKind,
    /// The models this read is derived from.
    ///
    /// Function pointers rather than plain strings because a model's identity is
    /// its *type*: where the type is nameable at the declaration site the entry
    /// is `|| core::any::type_name::<Post>()`, and where derivation only
    /// recovered an ident it is `|| "Post"`.
    pub reads: &'static [fn() -> &'static str],
    /// How [`Self::reads`] was established.
    pub provenance: DependencyProvenance,
    /// `Some(reason)` when the developer opted this read out of the gate.
    pub acknowledged_stale: Option<&'static str>,
    /// `file:line` of the declaration.
    pub location: &'static str,
}

inventory::collect!(CachedReadDescriptor);

/// A repository write method, as published by `#[repository]`.
pub struct MutationDescriptor {
    /// The repository trait name (e.g. `PostRepository`).
    pub repository: &'static str,
    /// The write method (e.g. `save`).
    pub method: &'static str,
    /// `core::any::type_name` of the mutated model.
    pub model: fn() -> &'static str,
    /// The mutated table.
    pub table: &'static str,
    /// Cache-read ids this write declares it invalidates.
    ///
    /// Populated from `#[repository(..., invalidates(path::to::cached_fn))]`,
    /// which resolves to the `#[cached]` function's own generated id constant —
    /// so rustc, not a string table, proves the target exists.
    pub invalidates: &'static [&'static str],
    /// `Some(reason)` when the developer opted this write out of the gate.
    pub acknowledged_stale: Option<&'static str>,
    /// `file:line` of the declaration.
    pub location: &'static str,
    /// For a cascade write: the child repository trait name, and whether the
    /// cascade runs the child's own cascade.
    ///
    /// The flag is `true` only for `destroy`. `nullify` and `delete_all` write
    /// one table. [`close_cascades`] uses both to find grandchild writes.
    pub cascades_into: Option<(&'static str, bool)>,
}

inventory::collect!(MutationDescriptor);

// ── Owned views (what the checker works over) ────────────────────────

/// An owned, comparable view of a [`CachedReadDescriptor`].
///
/// The checker is deliberately a pure function over owned values rather than
/// over `inventory` iterators, so the rule can be tested — and a seeded corpus
/// of staleness bugs exercised — without linking a whole app.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CachedRead {
    /// The cache-key namespace identifying this read.
    pub id: String,
    /// Which cache surface serves it.
    pub kind: ReadKind,
    /// The models it is derived from.
    pub reads: Vec<String>,
    /// How [`Self::reads`] was established.
    pub provenance: DependencyProvenance,
    /// `Some(reason)` when opted out of the gate.
    pub acknowledged_stale: Option<String>,
    /// `file:line` of the declaration.
    pub location: String,
}

/// An owned, comparable view of a [`MutationDescriptor`].
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Mutation {
    /// The repository trait name.
    pub repository: String,
    /// The write method.
    pub method: String,
    /// The mutated model's type name.
    pub model: String,
    /// The mutated table.
    pub table: String,
    /// Cache-read ids this write declares it invalidates.
    pub invalidates: Vec<String>,
    /// `Some(reason)` when opted out of the gate.
    pub acknowledged_stale: Option<String>,
    /// `file:line` of the declaration.
    pub location: String,
    /// For a cascade write: the child repository it writes to.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cascades_into: Option<String>,
    /// Whether the cascade runs the child's own cascade.
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub cascade_recurses: bool,
}

impl Mutation {
    /// `Repository::method` — how a mutation is named in every diagnostic.
    #[must_use]
    pub fn qualified_name(&self) -> String {
        format!("{}::{}", self.repository, self.method)
    }
}

/// One proven staleness bug: a mutation that can strand a cached value.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct StalenessFinding {
    /// The cached read left stale.
    pub read: String,
    /// Where that read is declared.
    pub read_location: String,
    /// `Repository::method` of the mutation that dirties it.
    pub mutation: String,
    /// Where that mutation is declared.
    pub mutation_location: String,
    /// The model both sides touch — the reason they are linked.
    pub model: String,
}

// ── Identity ─────────────────────────────────────────────────────────

/// Normalize a model name to its bare type name: the last path segment, with
/// references and generic arguments stripped.
///
/// `&blog::models::Post` and `Post` both key on `Post`; `Wrapper<Post>` keys on
/// `Wrapper`. Use [`models_match`] to compare two names — this is the fallback
/// half of that comparison, not the whole of it.
#[must_use]
pub fn model_key(name: &str) -> &str {
    let name = normalize_model_path(name);
    name.rsplit("::").next().unwrap_or(name).trim()
}

/// Strip the decoration around a model's path: references, pointers, a `mut`
/// binding mode, and any generic argument list.
fn normalize_model_path(name: &str) -> &str {
    let name = name.trim().trim_start_matches(['&', '*']).trim();
    let name = name.strip_prefix("mut ").unwrap_or(name).trim();
    name.split_once('<')
        .map_or(name, |(head, _)| head)
        .trim_end()
}

/// Whether two model names refer to the same model.
///
/// The two sides of a dependency learn the name differently. A `#[repository]`
/// always has the model *type* in scope and publishes `core::any::type_name`
/// (`blog::models::Post`); so does a `#[cached(reads(...))]` declaration. But a
/// dependency the macro *derived* from a function body may only be the bare
/// ident (`Post`), because the model type is often not in scope at the cached
/// function at all — a `PgPostRepository` parameter does not bring `Post` with
/// it.
///
/// So: when both sides are fully qualified, compare the full paths. That keeps
/// `plugin::models::User` and `crate::models::User` apart — two same-named
/// models in different modules are an ordinary shape, and collapsing them would
/// fail a correct app. Fall back to the bare type name only when at least one
/// side is all the analysis could recover, where over-approximating is the safe
/// direction.
#[must_use]
pub fn models_match(a: &str, b: &str) -> bool {
    let (a, b) = (normalize_model_path(a), normalize_model_path(b));
    if a.contains("::") && b.contains("::") {
        return a == b;
    }
    model_key(a) == model_key(b)
}

// ── Cascade closure ──────────────────────────────────────────────────

/// Add the writes a `destroy` cascade makes below its direct child.
///
/// A repository names only its own `dependent(...)` children. Only the full
/// set of descriptors names the grandchildren. This function follows
/// `cascades_into` through that set. It adds one write per model it reaches.
///
/// An added write keeps the root's method, edges, acknowledgement and
/// location. So the root must declare an edge for each model it reaches.
/// The walk visits each repository once, so a cycle ends. A second call adds
/// nothing.
#[must_use]
pub fn close_cascades(mutations: &[Mutation]) -> Vec<Mutation> {
    let mut closed = mutations.to_vec();
    // The location keeps two same-named roots apart: each owes its own edge.
    let key = |m: &Mutation, model: &str| {
        (
            m.repository.clone(),
            m.method.clone(),
            model.to_string(),
            m.location.clone(),
        )
    };
    let mut seen: std::collections::HashSet<(String, String, String, String)> =
        mutations.iter().map(|m| key(m, &m.model)).collect();
    let mut by_repo: std::collections::HashMap<&str, Vec<&Mutation>> =
        std::collections::HashMap::new();
    for m in mutations.iter().filter(|m| m.cascades_into.is_some()) {
        by_repo.entry(m.repository.as_str()).or_default().push(m);
    }
    for root in mutations.iter().filter(|m| m.cascades_into.is_some()) {
        // Not seeded with the root: a child may share its parent's trait name.
        let mut visited = std::collections::HashSet::new();
        let mut queue: Vec<&str> = root
            .cascades_into
            .as_deref()
            .filter(|_| root.cascade_recurses)
            .into_iter()
            .collect();
        while let Some(repo) = queue.pop() {
            if !visited.insert(repo) {
                continue;
            }
            for hop in by_repo.get(repo).into_iter().flatten() {
                if hop.cascade_recurses {
                    queue.extend(hop.cascades_into.as_deref());
                }
                if seen.insert(key(root, &hop.model)) {
                    closed.push(Mutation {
                        model: hop.model.clone(),
                        table: String::new(),
                        cascades_into: None,
                        ..root.clone()
                    });
                }
            }
        }
    }
    closed
}

// ── The rule ─────────────────────────────────────────────────────────

/// Prove that no mutation can strand a cached value.
///
/// Returns one [`StalenessFinding`] per uncovered `(read, mutation)` pair,
/// deterministically ordered by read, then mutation, then model, so a manifest
/// diffed across builds only changes when the app does.
///
/// See the module docs for the exact rule, and why reads with an
/// [`Undetermined`](DependencyProvenance::Undetermined) dependency set are
/// reported by [`undetermined_reads`] rather than failed here.
#[must_use]
pub fn check(reads: &[CachedRead], mutations: &[Mutation]) -> Vec<StalenessFinding> {
    let mut findings = Vec::new();

    for read in reads {
        if read.provenance == DependencyProvenance::Undetermined
            || read.acknowledged_stale.is_some()
        {
            continue;
        }
        for mutation in mutations {
            if mutation.acknowledged_stale.is_some() {
                continue;
            }
            // `any` — not a count — so a read that names one model twice (a
            // `declare_cached_read!` listing it under two paths, say) still
            // yields a single finding per mutation.
            if !read
                .reads
                .iter()
                .any(|dep| models_match(dep, &mutation.model))
            {
                continue;
            }
            if mutation.invalidates.iter().any(|id| id == &read.id) {
                continue;
            }
            findings.push(StalenessFinding {
                read: read.id.clone(),
                read_location: read.location.clone(),
                mutation: mutation.qualified_name(),
                mutation_location: mutation.location.clone(),
                model: mutation.model.clone(),
            });
        }
    }

    findings
        .sort_by(|a, b| (&a.read, &a.mutation, &a.model).cmp(&(&b.read, &b.mutation, &b.model)));
    findings
}

/// Cache-read identities claimed by more than one registration.
///
/// An identity is `module_path!()::<fn name>`, which two `#[cached]` associated
/// functions with the same name in two `impl` blocks in one module both
/// produce — as do a `declare_cached_read!` id chosen to collide with one. The
/// consequences are real: `invalidates(...)` cannot say which read it means,
/// and the two share a namespace at runtime. Invalidation clears every store
/// registered under the identity, so nothing is silently left stale, but the
/// ambiguity still deserves a name.
///
/// Returns the offending ids, sorted and deduplicated.
#[must_use]
pub fn duplicate_read_ids(reads: &[CachedRead]) -> Vec<String> {
    let mut ids: Vec<&str> = reads.iter().map(|r| r.id.as_str()).collect();
    ids.sort_unstable();
    let mut duplicates: Vec<String> = ids
        .windows(2)
        .filter(|pair| pair[0] == pair[1])
        .map(|pair| pair[0].to_string())
        .collect();
    duplicates.dedup();
    duplicates
}

/// Every cached read whose dependency set could not be established.
///
/// These gate nothing by default — see the module docs — but a green audit that
/// is mostly `undetermined` proves very little, so they are always reported and
/// `--strict` turns them into a failure.
///
/// A read carrying `acknowledge_stale` is excluded: that attribute is the
/// documented opt-out from the coherence gate, and a read the author has
/// already signed off as knowingly-stale must not fail `--strict` through the
/// undetermined door. Those reads are still reported, under their own
/// acknowledged heading, so the opt-out stays visible rather than silent.
#[must_use]
pub fn undetermined_reads(reads: &[CachedRead]) -> Vec<&CachedRead> {
    reads
        .iter()
        .filter(|r| {
            r.provenance == DependencyProvenance::Undetermined && r.acknowledged_stale.is_none()
        })
        .collect()
}

/// Whether the gate fails for this manifest.
///
/// The one place the rule lives, so `autumn cache audit`, an app's own test,
/// and the tests that pin the behavior are all asking the same function rather
/// than three copies of one condition.
///
/// Fails on a proven violation always, and on an undetermined dependency set
/// only under `strict` — the default never fails on what the analysis merely
/// could not read.
#[must_use]
pub const fn gate_failed(manifest: &CoherenceManifest, strict: bool) -> bool {
    !manifest.violations.is_empty() || (strict && !manifest.undetermined_reads.is_empty())
}

/// Exit code for the gate: `1` when [`gate_failed`], `0` otherwise.
#[must_use]
pub fn audit_exit_code(manifest: &CoherenceManifest, strict: bool) -> i32 {
    i32::from(gate_failed(manifest, strict))
}

// ── Diagnostics ──────────────────────────────────────────────────────

/// Render the precise diagnostic for a set of findings: for each, the cached
/// read, the mutations that can strand it, the model they share, and the two
/// ways to discharge it.
///
/// Grouped by `(read, model)` rather than printed one block per pair. A single
/// missing `invalidates(...)` on a `soft_delete` repository produces a dozen
/// findings — one per generated write method — and a dozen identical six-line
/// blocks with the same one-line fix is the output shape that gets a gate
/// switched off. The manifest keeps every pair; the human report names the one
/// edit.
#[must_use]
pub fn format_diagnostic(findings: &[StalenessFinding]) -> String {
    if findings.is_empty() {
        return String::new();
    }

    // Preserves `check`'s ordering: findings arrive sorted, so first-seen order
    // of each (read, model) group is already deterministic.
    let mut groups: Vec<(&StalenessFinding, Vec<&StalenessFinding>)> = Vec::new();
    for finding in findings {
        match groups
            .iter_mut()
            .find(|(head, _)| head.read == finding.read && head.model == finding.model)
        {
            Some((_, members)) => members.push(finding),
            None => groups.push((finding, vec![finding])),
        }
    }

    let stale_reads: usize = {
        let mut reads: Vec<&str> = groups.iter().map(|(head, _)| head.read.as_str()).collect();
        reads.sort_unstable();
        reads.dedup();
        reads.len()
    };

    let mut out = format!(
        "error: {stale_reads} cached read{} can be left stale by a repository write\n\n",
        if stale_reads == 1 { "" } else { "s" }
    );
    for (head, members) in &groups {
        let writes: Vec<&str> = members.iter().map(|f| f.mutation.as_str()).collect();
        // Every write method generated by one `#[repository]` carries that
        // attribute's own location, so this is normally a single site — but a
        // model with two repositories has two, and naming both is the point.
        let mut sites: Vec<&str> = members
            .iter()
            .map(|f| f.mutation_location.as_str())
            .collect();
        sites.dedup();
        let _ = writeln!(
            out,
            "  the cached read {read} is derived from {model},\n    \
             which {count} repository write{plural} mutate{verb} without invalidating it:\n      \
             {writes}\n    \
             read at    {read_loc}\n    \
             written at {sites}\n    \
             fix: add invalidates({read}) to the repository, or\n         \
             acknowledge the staleness with #[acknowledge_stale(reason = \"…\")].\n",
            read = head.read,
            model = head.model,
            count = writes.len(),
            plural = if writes.len() == 1 { "" } else { "s" },
            verb = if writes.len() == 1 { "s" } else { "" },
            writes = writes.join(", "),
            read_loc = head.read_location,
            sites = sites.join(", "),
        );
    }
    out
}

/// Render the human summary line printed by `autumn cache audit`.
///
/// A convenience over [`CoherenceManifest::summary`] for callers that hold the
/// values rather than the assembled manifest; both share one implementation of
/// the counting.
#[must_use]
pub fn format_summary(reads: &[CachedRead], mutations: &[Mutation]) -> String {
    CoherenceManifest::build(reads, mutations).summary()
}

// ── Manifest ─────────────────────────────────────────────────────────

/// A provenance-tagged manifest dimension.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Dimension<T> {
    /// `provable` / `declared` / `runtime-only`.
    pub provenance: String,
    /// Where the fact comes from (e.g. `macro:#[cached]`).
    pub source: String,
    /// The one part of the story a build cannot prove. Empty when there is none.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub runtime_caveat: String,
    /// The dimension's entries, stable-ordered.
    pub entries: Vec<T>,
}

/// One cached read in the manifest.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ManifestRead {
    /// The cache-key namespace.
    pub id: String,
    /// `cached` / `fragment` / `read_through`.
    pub kind: String,
    /// Models this read is derived from, stable-ordered.
    pub reads: Vec<String>,
    /// `declared` / `derived` / `undetermined`.
    pub provenance: String,
    /// The acknowledged-stale reason, when opted out.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub acknowledged_stale: Option<String>,
    /// `file:line`.
    pub location: String,
}

/// One repository write in the manifest.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ManifestMutation {
    /// `Repository::method`.
    pub name: String,
    /// The mutated model's type name.
    pub model: String,
    /// The mutated table.
    pub table: String,
    /// The acknowledged-stale reason, when opted out.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub acknowledged_stale: Option<String>,
    /// `file:line`.
    pub location: String,
}

/// One declared invalidation edge in the manifest.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ManifestInvalidation {
    /// `Repository::method` of the write that declares the edge.
    pub mutation: String,
    /// The cached read it covers.
    pub read: String,
}

/// A cached read the analysis could not link to any model.
///
/// Carries its source location for the same reason every other diagnostic in
/// this feature does: an id alone makes the reader grep.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct UndeterminedRead {
    /// The read's cache-key namespace.
    pub id: String,
    /// `file:line` of the declaration.
    pub location: String,
}

/// A dimension deliberately left out of the manifest, and why.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExcludedDimension {
    /// The dimension's name.
    pub dimension: String,
    /// The provenance class it could eventually claim.
    pub eventual_provenance: String,
    /// Why it is not emitted today.
    pub reason: String,
}

/// The manifest's emitted dimensions.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Dimensions {
    /// Every cached read the binary registers.
    pub cached_reads: Dimension<ManifestRead>,
    /// Every repository write the binary registers.
    pub mutations: Dimension<ManifestMutation>,
    /// Every declared invalidation edge.
    pub invalidations: Dimension<ManifestInvalidation>,
}

/// The cache-coherence manifest: what the build can prove about which cached
/// reads a write can strand, and how it knows.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CoherenceManifest {
    /// Document shape version.
    pub schema_version: u32,
    /// The emitted dimensions.
    pub dimensions: Dimensions,
    /// Proven staleness bugs, stable-ordered.
    pub violations: Vec<StalenessFinding>,
    /// Reads whose dependency set could not be established.
    pub undetermined_reads: Vec<UndeterminedRead>,
    /// Identities claimed by more than one cached read — see
    /// [`duplicate_read_ids`]. An `invalidates(...)` edge cannot distinguish
    /// them.
    #[serde(default)]
    pub duplicate_read_ids: Vec<String>,
    /// Dimensions deliberately not emitted, and why.
    pub excluded: Vec<ExcludedDimension>,
}

impl CoherenceManifest {
    /// Assemble the manifest from the app's registered reads and mutations.
    #[must_use]
    pub fn build(reads: &[CachedRead], mutations: &[Mutation]) -> Self {
        let mutations = &close_cascades(mutations);
        Self {
            schema_version: MANIFEST_SCHEMA_VERSION,
            dimensions: Dimensions {
                cached_reads: cached_reads_dimension(reads),
                mutations: mutations_dimension(mutations),
                invalidations: invalidations_dimension(mutations),
            },
            violations: check(reads, mutations),
            undetermined_reads: undetermined_dimension(reads),
            duplicate_read_ids: duplicate_read_ids(reads),
            excluded: excluded_dimensions(),
        }
    }

    /// The one-line human summary printed by `autumn cache audit`.
    ///
    /// Computed from the manifest's own entries rather than from the values it
    /// was built from, so the CLI and the framework can never disagree about
    /// the counts — and so a manifest read back from a file summarizes exactly
    /// the same way as one just built.
    #[must_use]
    pub fn summary(&self) -> String {
        let reads = &self.dimensions.cached_reads.entries;
        let count = |class: &str| reads.iter().filter(|e| e.provenance == class).count();
        let declared = count("declared");
        let derived = count("derived");
        let undetermined = reads.len() - declared - derived;
        let acknowledged = reads
            .iter()
            .filter(|e| e.acknowledged_stale.is_some())
            .count()
            + self
                .dimensions
                .mutations
                .entries
                .iter()
                .filter(|e| e.acknowledged_stale.is_some())
                .count();
        format!(
            "{} cached reads ({declared} declared, {derived} derived, {undetermined} \
             undetermined), {} repository mutations, {acknowledged} acknowledged-stale",
            reads.len(),
            self.dimensions.mutations.entries.len(),
        )
    }

    /// Serialize the manifest as pretty JSON.
    ///
    /// # Panics
    ///
    /// Never in practice: every field is a plain owned value with an infallible
    /// `Serialize` impl.
    #[must_use]
    pub fn to_json(&self) -> String {
        serde_json::to_string_pretty(self).expect("cache-coherence manifest is always serializable")
    }
}

/// The `cached_reads` dimension: every read the binary registers, stable-ordered.
fn cached_reads_dimension(reads: &[CachedRead]) -> Dimension<ManifestRead> {
    let mut entries: Vec<ManifestRead> = reads
        .iter()
        .map(|r| {
            let mut models: Vec<String> = r.reads.clone();
            models.sort();
            models.dedup();
            ManifestRead {
                id: r.id.clone(),
                kind: r.kind.as_str().to_string(),
                reads: models,
                provenance: r.provenance.as_str().to_string(),
                acknowledged_stale: r.acknowledged_stale.clone(),
                location: r.location.clone(),
            }
        })
        .collect();
    entries.sort_by(|a, b| (&a.id, &a.location).cmp(&(&b.id, &b.location)));
    Dimension {
        provenance: "provable".to_string(),
        source: "macro:#[cached] / declare_cached_read!".to_string(),
        runtime_caveat: "each entry's EXISTENCE is proven — a `#[cached]` function registers \
                         itself — but its dependency set is only as strong as the entry's own \
                         `provenance` field says: `declared` is trusted verbatim (a `reads(...)` \
                         naming the wrong model audits clean), `derived` is what a syntactic \
                         analysis of the function could see, and `undetermined` is nothing at \
                         all. A `declare_cached_read!` entry is hand-written throughout, \
                         including the fact that the call site exists; an undeclared \
                         `cache_fragment` / `get_or_compute` call is invisible here — see the \
                         undeclared_cache_call_sites entry in `excluded`."
            .to_string(),
        entries,
    }
}

/// The `mutations` dimension: every repository write the binary registers.
fn mutations_dimension(mutations: &[Mutation]) -> Dimension<ManifestMutation> {
    let mut entries: Vec<ManifestMutation> = mutations
        .iter()
        .map(|m| ManifestMutation {
            name: m.qualified_name(),
            model: m.model.clone(),
            table: m.table.clone(),
            acknowledged_stale: m.acknowledged_stale.clone(),
            location: m.location.clone(),
        })
        .collect();
    // `location` and `table` join the key so a tie cannot fall back to
    // `inventory`'s link order, which is unspecified: two same-named repository
    // traits in different modules would otherwise swap rows between
    // byte-identical builds.
    entries.sort_by(|a, b| {
        (&a.name, &a.model, &a.table, &a.location).cmp(&(&b.name, &b.model, &b.table, &b.location))
    });
    Dimension {
        provenance: "provable".to_string(),
        source: "macro:#[repository]".to_string(),
        runtime_caveat: "every entry is proven, but the SET is not exhaustive: only \
                         `#[repository]` write methods are here. Nothing proves an app's writes \
                         all go through one — see the writes_outside_repository entry in \
                         `excluded` — and a write that reaches another model's table through a \
                         `counter_cache` is registered under its own model, not the counter's."
            .to_string(),
        entries,
    }
}

/// The `invalidations` dimension: every declared edge from a write to a read.
///
/// `provable`, not `declared`: the edge is recovered from macro-expanded code
/// with no config read and no process started, which is question 1 of the
/// provenance rubric in `docs/guide/security-posture-manifest.md`. The weak step
/// is an ADJACENT one — whether the invalidator is called — and the rubric's
/// tie-breaker is a `runtime_caveat`, not a demotion that would understate what
/// the build proves.
fn invalidations_dimension(mutations: &[Mutation]) -> Dimension<ManifestInvalidation> {
    let mut entries: Vec<ManifestInvalidation> = mutations
        .iter()
        .flat_map(|m| {
            m.invalidates.iter().map(move |read| ManifestInvalidation {
                mutation: m.qualified_name(),
                read: read.clone(),
            })
        })
        .collect();
    entries.sort_by(|a, b| (&a.mutation, &a.read).cmp(&(&b.mutation, &b.read)));
    Dimension {
        provenance: "provable".to_string(),
        source: "macro:#[repository(..., invalidates(...))]".to_string(),
        runtime_caveat: "the edge's target is proven — `invalidates(path)` resolves to the \
                         `#[cached]` function's own generated id constant, so rustc rejects a \
                         path that names anything else. Each generated write method calls the \
                         invalidator after its own transaction commits (#3056), and a durable \
                         commit-hook row retries it when `commit_hooks` is on. Not proven: that \
                         the backend sweep succeeds (a failure is logged and counted in \
                         `autumn_cache_invalidation_failures_total`). Also not proven on a \
                         backend with no shared fill epoch: that no fill on another replica \
                         writes an old value back. `RedisCache` has a shared epoch. See \
                         docs/guide/cache-coherence.md."
            .to_string(),
        entries,
    }
}

/// Reads whose dependency set could not be established, stable-ordered and
/// carrying their locations.
fn undetermined_dimension(reads: &[CachedRead]) -> Vec<UndeterminedRead> {
    let mut entries: Vec<UndeterminedRead> = undetermined_reads(reads)
        .into_iter()
        .map(|r| UndeterminedRead {
            id: r.id.clone(),
            location: r.location.clone(),
        })
        .collect();
    entries.sort_by(|a, b| (&a.id, &a.location).cmp(&(&b.id, &b.location)));
    entries
}

/// Dimensions this slice deliberately does not emit, and why — so a reader can
/// tell "we checked and it was fine" from "we never looked".
fn excluded_dimensions() -> Vec<ExcludedDimension> {
    vec![
        ExcludedDimension {
            dimension: "row_and_column_scoped_dependencies".to_string(),
            eventual_provenance: "provable".to_string(),
            reason: "model/table granularity is what this slice proves; a read derived from one \
                     column of one row is currently treated as depending on the whole model, \
                     which over-approximates (a false failure, never a false pass)"
                .to_string(),
        },
        ExcludedDimension {
            dimension: "invalidation_call_sites".to_string(),
            eventual_provenance: "provable".to_string(),
            reason: "generated write methods call the declared invalidator after they commit \
                     (#3056); what that does not prove is carried as the invalidations \
                     dimension's runtime_caveat"
                .to_string(),
        },
        ExcludedDimension {
            dimension: "writes_outside_repository".to_string(),
            eventual_provenance: "provable".to_string(),
            reason: "only `#[repository]` write methods are in the mutated set. A hand-rolled \
                     repository, a raw diesel `update`/`insert`/`delete` against the handle, a \
                     migration, or a job that writes directly is invisible here, so a cached \
                     read those dirty audits clean. Declare such a write's effect by routing it \
                     through a `#[repository]`, or acknowledge the read"
                .to_string(),
        },
        ExcludedDimension {
            dimension: "undeclared_cache_call_sites".to_string(),
            eventual_provenance: "provable".to_string(),
            reason: "`cache_fragment` and `get_or_compute` are plain function calls with a \
                     runtime key — there is no annotated item for a macro to find, so a call \
                     site that does not opt in with `declare_cached_read!` is not in this \
                     manifest at all. A clean audit says nothing about it. Opting in is also \
                     not enough on its own: the call site must key its entries under the \
                     declared id, or `invalidate_namespace` matches nothing and reports \
                     success. `cache_fragment_in` and `make_cache_key` are the namespaced \
                     forms; the bare `cache_fragment` keys under `fragment:` and is not \
                     reachable by a namespace sweep"
                .to_string(),
        },
        ExcludedDimension {
            dimension: "counter_cache_and_cascade_writes".to_string(),
            eventual_provenance: "provable".to_string(),
            reason: "both are declared on the MODEL, which `#[repository]` cannot see. A \
                     `#[belongs_to(Parent, counter_cache)]` column means a child's write \
                     updates the parent's counter, registered under the child's model only; a \
                     `#[has_many(Child, dependent = destroy)]` means the parent's delete writes \
                     the child's table with no descriptor for it. A cascade declared on the \
                     REPOSITORY — `#[repository(..., dependent(...))]` — IS registered against \
                     the child model it deletes or nullifies. So is each model further \
                     down a chain of `destroy` cascades. The audit links the chain by \
                     repository trait name. `on_delete = restrict` is excluded because it \
                     only probes and never writes"
                .to_string(),
        },
        ExcludedDimension {
            dimension: "http_response_cache".to_string(),
            eventual_provenance: "provable".to_string(),
            reason: "`CacheResponseLayer` caches whole HTTP responses keyed by URI, with no \
                     annotated item naming what the response was derived from. It has no \
                     ReadKind and no declaration form in this slice; a response cache over \
                     mutable data is not covered by this gate"
                .to_string(),
        },
        ExcludedDimension {
            dimension: "cross_service_coherence".to_string(),
            eventual_provenance: "runtime-only".to_string(),
            reason: "a second service's writes are not in this binary's dependency graph; \
                     single-app scope by design"
                .to_string(),
        },
        ExcludedDimension {
            dimension: "ttl_expiry".to_string(),
            eventual_provenance: "declared".to_string(),
            reason: "time-based expiry is orthogonal to derived-data coherence: a TTL bounds how \
                     long a value stays stale, it does not stop it becoming stale"
                .to_string(),
        },
    ]
}

// ── Manual registration for non-macro cache surfaces ─────────────────

/// Register a cached read the proc macros cannot see.
///
/// `#[cached]` registers itself, but the fragment cache
/// ([`cache_fragment`](super::cache_fragment)) and the read-through cache
/// ([`get_or_compute`](super::get_or_compute)) are plain function calls with a
/// runtime key — there is no annotated item for a macro to analyse. Declaring
/// the read here puts it in the same manifest and under the same gate.
///
/// The `id` must be the same key namespace the call site passes to the cache,
/// so an invalidation edge and the runtime key agree.
///
/// ```rust,ignore
/// autumn_web::declare_cached_read! {
///     id = "blog::sidebar_fragment",
///     kind = Fragment,
///     reads = [crate::models::Post, crate::models::Tag],
/// }
/// ```
///
/// # Keying the call site under the declared `id`
///
/// A declaration alone does not make the read *invalidatable* — the entry has
/// to live under `id` in the cache's key space, because that is the only thing
/// [`invalidate_namespace`] can match on. The plain helpers do not do this:
///
/// * [`cache_fragment`](super::cache_fragment) keys entries as
///   `fragment:{len}:{identity}:{version}`. That prefix carries no per-read
///   identity, so a sweep for `blog::sidebar_fragment` matches none of them.
///   Use [`cache_fragment_in`](super::cache_fragment_in) or
///   [`cache_fragment_global_in`](super::cache_fragment_global_in), which take
///   the namespace and put it where the sweep looks.
/// * [`get_or_compute`](super::get_or_compute) keys by whatever you pass it, so
///   build that key with [`make_cache_key`](super::make_cache_key) and this
///   `id` as its namespace.
///
/// A declared read whose call site keys under something else is still audited
/// — the gate will still demand an edge for it — but the edge's
/// `invalidate_namespace` will clear nothing and report success. Either key it
/// under the `id`, or declare it with `acknowledge_stale` so the gap is stated
/// rather than implied.
///
/// An acknowledged-stale opt-out takes a trailing `acknowledge_stale = "…"`.
#[macro_export]
macro_rules! declare_cached_read {
    // At least one model is required, for the same reason `#[cached(reads())]`
    // is a compile error: an EMPTY declared set is trivially coherent and would
    // silence the gate while inflating the `declared` count — a hatch with none
    // of the visibility the real hatches carry.
    (
        id = $id:expr,
        kind = $kind:ident,
        reads = [$first:path $(, $rest:path)* $(,)?] $(,)?
    ) => {
        $crate::declare_cached_read!(
            id = $id,
            kind = $kind,
            reads = [$first $(, $rest)*],
            acknowledged_stale = ::core::option::Option::None,
        );
    };
    (
        id = $id:expr,
        kind = $kind:ident,
        reads = [$first:path $(, $rest:path)* $(,)?],
        acknowledge_stale = $reason:literal $(,)?
    ) => {
        // Same rule the attribute macros enforce, and the same BLANK test:
        // an escape hatch whose justification is three spaces is the one nobody
        // can review.
        const _: () = assert!(
            !$crate::cache::coherence::reason_is_blank($reason),
            "acknowledge_stale requires a non-blank reason",
        );
        $crate::declare_cached_read!(
            id = $id,
            kind = $kind,
            reads = [$first $(, $rest)*],
            acknowledged_stale = ::core::option::Option::Some($reason),
        );
    };
    (
        id = $id:expr,
        kind = $kind:ident,
        reads = [$($model:path),+ $(,)?],
        acknowledged_stale = $ack:expr $(,)?
    ) => {
        $crate::reexports::inventory::submit! {
            $crate::cache::coherence::CachedReadDescriptor {
                id: $id,
                kind: $crate::cache::coherence::ReadKind::$kind,
                reads: &[$(|| ::core::any::type_name::<$model>()),*],
                // A hand-declared set is exactly as strong a claim as
                // `#[cached(reads(...))]`: a human wrote it down.
                provenance: $crate::cache::coherence::DependencyProvenance::Declared,
                acknowledged_stale: $ack,
                location: concat!(file!(), ":", line!()),
            }
        }
    };
}

/// Decode the UTF-8 code point starting at `i`, returning it and its width.
///
/// A hand-rolled decoder because `str::chars` is not `const`. The input is a
/// `&str`, so it is valid UTF-8 by construction and every continuation byte is
/// in range.
const fn decode_utf8(bytes: &[u8], i: usize) -> (u32, usize) {
    let first = bytes[i];
    if first < 0x80 {
        (first as u32, 1)
    } else if first < 0xE0 {
        (
            (((first as u32) & 0x1F) << 6) | ((bytes[i + 1] as u32) & 0x3F),
            2,
        )
    } else if first < 0xF0 {
        (
            (((first as u32) & 0x0F) << 12)
                | (((bytes[i + 1] as u32) & 0x3F) << 6)
                | ((bytes[i + 2] as u32) & 0x3F),
            3,
        )
    } else {
        (
            (((first as u32) & 0x07) << 18)
                | (((bytes[i + 1] as u32) & 0x3F) << 12)
                | (((bytes[i + 2] as u32) & 0x3F) << 6)
                | ((bytes[i + 3] as u32) & 0x3F),
            4,
        )
    }
}

/// Whether an acknowledged-stale reason is blank.
///
/// Blank means what `str::trim().is_empty()` means — **Unicode** whitespace, not
/// just ASCII. An EM SPACE (`U+2003`) is a blank reason, and a check that only
/// knew about ASCII would leave `declare_cached_read!` as a way to silence every
/// coherence finding with a justification nobody can see.
///
/// `str::trim` is not `const` and neither is `str::chars`, so this decodes the
/// bytes itself and defers to `char::is_whitespace`, which is. The rule has to
/// be the same one `#[cached]` and `#[repository]` enforce at parse time, or the
/// weakest of the three declaration forms becomes the free pass.
#[must_use]
pub const fn reason_is_blank(reason: &str) -> bool {
    let bytes = reason.as_bytes();
    let mut i = 0;
    while i < bytes.len() {
        let (code_point, width) = decode_utf8(bytes, i);
        match char::from_u32(code_point) {
            Some(c) if c.is_whitespace() => {}
            _ => return false,
        }
        i += width;
    }
    true
}

// ── Namespace invalidation ───────────────────────────────────────────

/// What one cache-key namespace owns: the dedicated stores registered under it,
/// and the epoch that fences an in-flight fill.
#[derive(Default)]
struct Namespace {
    /// Every store registered under this namespace.
    ///
    /// A `Vec`, not a single store: two `#[cached]` associated functions with
    /// the same name in two `impl` blocks in one module produce the same
    /// `module_path!()`-derived identity, and overwriting would leave one of
    /// them stale while reporting success. Clearing all of them
    /// over-invalidates, which is safe; the audit reports the shared identity
    /// so the ambiguity gets a name (see [`duplicate_read_ids`]).
    stores: Vec<std::sync::Arc<dyn super::Cache>>,
    /// Bumped by every [`invalidate_namespace`]. A fill that started before the
    /// bump and finishes after it must not write its now-stale value back.
    epoch: std::sync::Arc<std::sync::atomic::AtomicU64>,
}

/// Per-namespace cache state, keyed by the read's cache-key namespace.
///
/// A `#[cached]` function's dedicated Moka store holds **only** that function's
/// entries, so clearing it *is* namespace invalidation — no key enumeration
/// required. The store registers itself here the first time the function runs,
/// which is what lets the generated `__autumn_cache_invalidate__<fn>()` reach a
/// store that lives inside the function body.
static NAMESPACES: std::sync::RwLock<Option<std::collections::HashMap<String, Namespace>>> =
    std::sync::RwLock::new(None);

/// Register a cached read's dedicated store so [`invalidate_namespace`] can
/// reach it, and take the namespace's fill fence.
///
/// Returns the epoch counter the caller must sample **before** computing a value
/// and re-check **before** inserting it. Without that fence a fill already in
/// flight when the invalidation lands writes its stale value back afterwards,
/// and the invalidator has already reported success. `#[cached]`-generated code
/// does this automatically; a hand-written cache that registers here should do
/// the same, or accept that window.
///
/// # Panics
///
/// Panics if the internal `RwLock` is poisoned.
pub fn register_namespace_store(
    namespace: &'static str,
    store: std::sync::Arc<dyn super::Cache>,
) -> std::sync::Arc<std::sync::atomic::AtomicU64> {
    let mut guard = NAMESPACES
        .write()
        .expect("cache namespace store lock poisoned");
    let entry = guard
        .get_or_insert_with(std::collections::HashMap::new)
        .entry(namespace.to_string())
        .or_default();
    entry.stores.push(store);
    let epoch = std::sync::Arc::clone(&entry.epoch);
    // Explicit, so the process-wide write lock is never held across the return.
    drop(guard);
    epoch
}

/// The fill epoch for `namespace`, without registering a store against it.
///
/// [`register_namespace_store`] is for a cache that *owns* the namespace and
/// wants [`invalidate_namespace`] to clear it. This is for the other case: a
/// call site writing into a store somebody else already registered — or into
/// the process-global backend — that still needs the fence, because its insert
/// races the same invalidation. `cache_fragment_in` uses it.
///
/// Sample the returned counter **before** the lookup, and insert through
/// [`with_fill_fence`].
///
/// # Panics
///
/// Panics if the internal `RwLock` is poisoned.
#[must_use]
pub fn namespace_epoch(namespace: &str) -> std::sync::Arc<std::sync::atomic::AtomicU64> {
    // Read first: after the namespace's first use this is the only path taken,
    // and a fragment render must not queue behind a process-wide write lock.
    {
        let guard = NAMESPACES
            .read()
            .expect("cache namespace store lock poisoned");
        if let Some(epoch) = guard
            .as_ref()
            .and_then(|map| map.get(namespace))
            .map(|entry| std::sync::Arc::clone(&entry.epoch))
        {
            return epoch;
        }
    }

    let mut guard = NAMESPACES
        .write()
        .expect("cache namespace store lock poisoned");
    let entry = guard
        .get_or_insert_with(std::collections::HashMap::new)
        .entry(namespace.to_string())
        .or_default();
    let epoch = std::sync::Arc::clone(&entry.epoch);
    drop(guard);
    epoch
}

/// Whether a value computed while the namespace was at `sampled` is still safe
/// to insert.
///
/// `false` means an invalidation landed mid-fill, so the value in hand is
/// already stale and inserting it would resurrect exactly what was just
/// dropped.
///
/// This is the bare comparison, and on its own it is **not** enough: between
/// the `true` it returns and the insert it guards, an invalidation can bump the
/// epoch and clear the namespace, and the insert then lands after the clear.
/// Use [`with_fill_fence`], which performs the same comparison and the insert
/// as one step that cannot interleave with an invalidation. This stays public
/// for callers that only want to *observe* the fence — a test, a metric, a
/// decision not to bother inserting — and for reading the epoch outside any
/// critical section.
#[must_use]
pub fn fill_is_still_current(epoch: &std::sync::atomic::AtomicU64, sampled: u64) -> bool {
    epoch.load(std::sync::atomic::Ordering::Acquire) == sampled
}

/// Serializes cache fills against namespace invalidation.
///
/// Held shared by every in-flight fill's check-and-insert (fills never block
/// each other) and exclusively by [`invalidate_namespace`] for the epoch bump
/// alone. The data is `()`: the lock carries no state, only the mutual
/// exclusion that makes the epoch check and the insert it guards indivisible.
static FILL_FENCE: std::sync::RwLock<()> = std::sync::RwLock::new(());

/// Insert a freshly computed cache value **iff** no invalidation has landed
/// since `sampled`, with the check and the insert indivisible.
///
/// [`fill_is_still_current`] alone leaves a window: it can return `true`, an
/// invalidation can then bump the epoch and clear the namespace, and the
/// caller's insert lands afterwards — writing a pre-invalidation value into a
/// namespace that was just emptied, where it can sit until its TTL, or forever
/// when there is none. This closes that window. `insert` runs while the fence
/// is held shared and [`invalidate_namespace`] bumps the epoch while holding it
/// exclusively, so only two orderings exist:
///
/// * this call completes first — the invalidation's clear, which follows its
///   bump, removes what was just inserted; or
/// * the bump completes first — this call observes it and skips the insert.
///
/// Returns `Some(insert(...))` when the value was written, `None` when the fill
/// was fenced out.
///
/// # Cost
///
/// `insert` runs under a shared lock, so a slow backend write delays
/// *invalidators*, not other fills. Keep it to the insert itself, and never call
/// [`invalidate_namespace`] from inside it — `std`'s `RwLock` is not reentrant
/// and that self-deadlocks.
pub fn with_fill_fence<R>(
    epoch: &std::sync::atomic::AtomicU64,
    sampled: u64,
    insert: impl FnOnce() -> R,
) -> Option<R> {
    // A poisoned fence means some other fill's insert panicked. The lock guards
    // no data — `()` cannot be left inconsistent — so recovering is right, and
    // propagating would turn one unrelated panic into a permanently broken
    // cache path.
    let guard = FILL_FENCE
        .read()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    let result = fill_is_still_current(epoch, sampled).then(insert);
    drop(guard);
    result
}

/// Drop every entry belonging to one cached read.
///
/// Clears the read's own dedicated in-process store **and**, when a
/// process-level shared backend is registered via
/// [`set_global_cache`](super::set_global_cache), asks that backend to drop the
/// namespace too — which is where the value actually lives once one is
/// configured.
///
/// Returns whether the invalidation was **complete**, which means exactly this:
/// every entry the namespace held is gone, and no fill already in flight can
/// put a stale one back. The second part holds for fills in this process. It
/// holds for fills on other replicas only if the backend has a shared fill
/// epoch (see below).
///
/// It is `false` when a registered process-level backend could not drop the
/// namespace — a `RedisCache` whose `SCAN`/`DEL` errored, or a custom
/// [`Cache`](super::Cache) that cannot pattern-match its key space and so
/// returns `false` from
/// [`Cache::invalidate_namespace`](super::Cache::invalidate_namespace). That
/// `false` is reported here verbatim: the honest signal that stale entries may
/// still be served.
///
/// # What it reaches, and what it does not
///
/// It clears every store registered for `namespace` via
/// [`register_namespace_store`] — which `#[cached]` does for you on first call —
/// and asks the process-level backend
/// ([`set_global_cache`](super::set_global_cache)) to drop the namespace.
///
/// It cannot reach a cache nobody registered. A `declare_cached_read!` entry
/// over a caller-owned store — `cache_fragment(Some(&my_cache), …)`,
/// `get_or_compute(&my_cache, …)` — is invisible here, and this returns `true`
/// having cleared nothing, because "no store registered" is indistinguishable
/// from "the function has not run yet". Register such a store to close that
/// gap.
///
/// # The fill fence
///
/// The local fence is **per process**. A backend can extend it to all replicas
/// with [`Cache::fill_epoch`](super::Cache::fill_epoch) and
/// [`Cache::insert_raw_bytes_if_epoch`](super::Cache::insert_raw_bytes_if_epoch).
/// `RedisCache` does. It raises a per-namespace epoch in Redis **before** it
/// sweeps. A fill stores its value only if that epoch did not move. With such a
/// backend, a `true` is valid for all replicas. If the epoch bump fails, this
/// returns `false`.
///
/// The shared fence covers `#[cached]` and `cache_fragment_in` fills. It does
/// not cover `get_or_compute`, `Cache::clear`, or a key-level `invalidate`.
///
/// With a backend that has no shared epoch, a fill on another replica that
/// started before this call can still write a stale value after it. A `true`
/// is then valid for this process only.
///
/// # Panics
///
/// Panics if the internal `RwLock` is poisoned.
#[must_use = "a `false` means stale entries can still be served"]
pub fn invalidate_namespace(namespace: &str) -> bool {
    // A replay answers from the tape and reaches no store (#2351).
    if let Some(answer) = super::replayed_namespace_removal(namespace) {
        return answer;
    }
    let removal = super::NamespaceRemoval::reserve();
    let mut complete = true;
    for store in fence_and_collect_stores(namespace) {
        // A shared store raises its epoch here. `clear` alone would not.
        let swept = store.invalidate_namespace(namespace);
        store.clear();
        complete &= registered_store_complete(&*store, swept, namespace);
    }

    // The dedicated stores are only half the story: once a process-level backend
    // is registered, that is where the value actually lives, and the per-
    // function store holds nothing. Ask the backend — `MokaCache` and
    // `RedisCache` both drop the namespace; a backend that cannot, or one whose
    // sweep failed, says so, and that `false` is what the caller reports.
    let global_failure = super::global_cache().and_then(|global| {
        if super::NamespaceRemoval::call_installed(|| global.invalidate_namespace(namespace)) {
            return None;
        }
        let reason = "the backend cannot drop a namespace, or its sweep failed";
        report_incomplete_invalidation(namespace, reason);
        Some(reason.to_owned())
    });
    finish_removal(removal, namespace, complete, global_failure)
}

/// Record a namespace removal's combined answer on the capsule seam, and
/// return it.
fn finish_removal(
    removal: super::NamespaceRemoval,
    namespace: &str,
    stores_complete: bool,
    global_failure: Option<String>,
) -> bool {
    let failure = global_failure.or_else(|| {
        (!stores_complete).then(|| "a registered store could not drop the namespace".to_owned())
    });
    let complete = failure.is_none();
    removal.record(namespace, failure.as_deref());
    complete
}

/// Whether a registered store counts as invalidated.
///
/// A store with no shared epoch is cleared by `clear`, so its answer does not
/// matter. A store with a shared epoch must have raised it.
fn registered_store_complete(store: &dyn super::Cache, swept: bool, namespace: &str) -> bool {
    if swept || !store.shares_fill_epoch() {
        return true;
    }
    report_incomplete_invalidation(
        namespace,
        "a registered shared store could not raise its fill epoch",
    );
    false
}

/// Async form of [`invalidate_namespace`], with the same contract.
///
/// The epoch bump and the local store clears are the same. The wait for the
/// fill fence runs on the blocking pool. The backend call is
/// [`Cache::invalidate_namespace_async`](super::Cache::invalidate_namespace_async),
/// so a network backend (Redis) does not block a runtime worker and works on a
/// current-thread runtime.
///
/// A `false` result is also logged with `warn!` and counted in
/// `autumn_cache_invalidation_failures_total`.
#[must_use = "a `false` means stale entries can still be served"]
pub async fn invalidate_namespace_async(namespace: &str) -> bool {
    // A replay answers from the tape and reaches no store (#2351).
    if let Some(answer) = super::replayed_namespace_removal(namespace) {
        return answer;
    }
    // The fence is a std `RwLock`. A fill holds it shared during its insert,
    // which can be a Redis round trip. Take it at once when it is free. Else
    // wait for it on the blocking pool, so a writer never parks a runtime
    // worker (and so never starves the task that the fill waits on).
    let removal = super::NamespaceRemoval::reserve();
    let stores = if let Some(stores) = try_fence_and_collect_stores(namespace) {
        stores
    } else {
        let owned = namespace.to_owned();
        match crate::time::spawn_blocking(move || fence_and_collect_stores(&owned)).await {
            Ok(stores) => stores,
            Err(join) => match join.try_into_panic() {
                Ok(panic) => std::panic::resume_unwind(panic),
                Err(_cancelled) => {
                    let reason = "the runtime is shutting down";
                    report_incomplete_invalidation(namespace, reason);
                    return finish_removal(removal, namespace, false, Some(reason.to_owned()));
                }
            },
        }
    };
    let mut complete = true;
    for store in stores {
        // A shared store raises its epoch here. `clear` alone would not.
        let swept = store.invalidate_namespace_async(namespace).await.is_ok();
        store.clear();
        complete &= registered_store_complete(&*store, swept, namespace);
    }
    let mut global_failure = None;
    if let Some(global) = super::global_cache() {
        let sweep = super::NamespaceRemoval::call_installed(|| {
            global.invalidate_namespace_async(namespace)
        });
        if let Err(error) = sweep.await {
            report_incomplete_invalidation(namespace, error.reason());
            global_failure = Some(error.reason().to_owned());
        }
    }
    finish_removal(removal, namespace, complete, global_failure)
}

/// Runs a repository's declared invalidation after each commit of a write
/// (#3056).
///
/// Generated write methods make one before the write body. The body calls
/// [`committed`](Self::committed) after each transaction, and
/// [`flush`](Self::flush) before each `after_create` or `after_update` hook.
/// Thus a hook, and every other reader, sees the committed row while the hook
/// runs. The method calls [`run`](Self::run) at its end, on `Ok` and on `Err`,
/// for a commit that no hook followed. The guard starts pending, so a write
/// with no tracked transaction still invalidates at its end.
///
/// Each invalidation runs on a spawned task, so a caller that is dropped while
/// it waits does not stop it. If the guard drops while pending (a panic, or a
/// cancelled request), it spawns the invalidation on the current runtime.
/// Over-invalidation is safe: the cost is one cache miss. A request cancelled
/// during its `COMMIT` round trip can invalidate before the server applies the
/// commit. A durable commit-hook row (`commit_hooks`) covers that case.
#[doc(hidden)]
#[must_use = "call `run().await` after the write body"]
pub struct InvalidateAfterWrite {
    make: fn() -> super::CacheFuture<'static, bool>,
    pending: bool,
}

impl InvalidateAfterWrite {
    /// Arm the guard. `make` builds the repository's invalidation future.
    pub const fn new(make: fn() -> super::CacheFuture<'static, bool>) -> Self {
        Self {
            make,
            pending: true,
        }
    }

    /// Record that a transaction ended. The next flush invalidates.
    pub const fn committed(&mut self) {
        self.pending = true;
    }

    /// Invalidate now if a commit is not yet invalidated, and wait for it.
    ///
    /// Returns whether it was complete. A `false` is already logged and
    /// counted.
    pub async fn flush(&mut self) -> bool {
        if !self.pending {
            return true;
        }
        self.pending = false;
        let Ok(handle) = tokio::runtime::Handle::try_current() else {
            report_skipped_invalidation("the write ran outside a Tokio runtime");
            return false;
        };
        // Awaited at once, so the capsule scope and the replay tape go along:
        // the removal is recorded and replayed like inline work (#2351).
        #[cfg(feature = "reporting")]
        let invalidation = crate::capsule::boundary::carry_scopes((self.make)());
        #[cfg(not(feature = "reporting"))]
        let invalidation = (self.make)();
        // A caller dropped while it waits leaves the task detached.
        #[cfg(feature = "reporting")]
        let detached = crate::capsule::boundary::DetachGuard::arm();
        let complete = handle.spawn(invalidation).await.unwrap_or_else(|_| {
            report_skipped_invalidation("the invalidation task panicked or was cancelled");
            false
        });
        #[cfg(feature = "reporting")]
        detached.disarm();
        complete
    }

    /// Flush at the end of the write.
    pub async fn run(mut self) -> bool {
        self.flush().await
    }

    /// Skip the invalidation: the write wrote nothing (for example a
    /// `find_or_create_by_*` that found the row).
    pub fn disarm(mut self) {
        self.pending = false;
    }
}

impl Drop for InvalidateAfterWrite {
    fn drop(&mut self) {
        if !self.pending {
            return;
        }
        self.pending = false;
        if let Ok(handle) = tokio::runtime::Handle::try_current() {
            // Not awaited: the capsule says so, and a replay gives the task
            // the tape (#2351).
            #[cfg(feature = "reporting")]
            let invalidation = {
                crate::capsule::boundary::note_detached_work();
                crate::capsule::boundary::carry_detached((self.make)())
            };
            #[cfg(not(feature = "reporting"))]
            let invalidation = (self.make)();
            handle.spawn(invalidation);
        } else {
            report_skipped_invalidation("the write ended outside a Tokio runtime");
        }
    }
}

/// Log and count a repository invalidation that did not run to its end.
fn report_skipped_invalidation(reason: &str) {
    tracing::warn!(
        reason,
        "a repository write's declared cache invalidation did not complete; stale entries \
         can be served until their TTL"
    );
    super::record_invalidation_failure();
}

/// Log and count a namespace invalidation that did not complete.
fn report_incomplete_invalidation(namespace: &str, reason: &str) {
    tracing::warn!(
        namespace,
        reason,
        "cache namespace invalidation incomplete; stale entries can be served until their TTL"
    );
    super::record_invalidation_failure();
}

/// Bump the namespace epoch under the exclusive fill fence, and return the
/// stores registered for it.
///
/// The epoch bump comes FIRST. Then a fill that is already computing is
/// fenced out before the clear. In the other order, a fill can pass the fence
/// and insert after the clear.
///
/// The bump holds the fill fence exclusively. Thus a concurrent
/// `with_fill_fence` either inserts before the bump, and the clear removes its
/// value, or it sees the bump and skips its insert.
///
/// Only the bump is exclusive. Both guards drop before the caller calls a
/// backend: `clear()` can be a network round trip, and a `clear()` that calls
/// a `#[cached]` function would re-enter these `RwLock`s and deadlock.
fn fence_and_collect_stores(namespace: &str) -> Vec<std::sync::Arc<dyn super::Cache>> {
    let fence = FILL_FENCE
        .write()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    bump_and_collect_stores(namespace, fence)
}

/// [`fence_and_collect_stores`] if the fence is free now, else `None`.
fn try_fence_and_collect_stores(namespace: &str) -> Option<Vec<std::sync::Arc<dyn super::Cache>>> {
    let fence = match FILL_FENCE.try_write() {
        Ok(fence) => fence,
        Err(std::sync::TryLockError::Poisoned(poisoned)) => poisoned.into_inner(),
        Err(std::sync::TryLockError::WouldBlock) => return None,
    };
    Some(bump_and_collect_stores(namespace, fence))
}

/// The epoch bump and store lookup. `fence` is the exclusive fill fence.
fn bump_and_collect_stores(
    namespace: &str,
    fence: std::sync::RwLockWriteGuard<'_, ()>,
) -> Vec<std::sync::Arc<dyn super::Cache>> {
    let guard = NAMESPACES
        .read()
        .expect("cache namespace store lock poisoned");
    let stores = guard
        .as_ref()
        .and_then(|map| map.get(namespace))
        .map_or_else(Vec::new, |entry| {
            entry
                .epoch
                .fetch_add(1, std::sync::atomic::Ordering::AcqRel);
            entry.stores.clone()
        });
    drop(guard);
    drop(fence);
    stores
}

// ── Reading the binary's own registrations ───────────────────────────

/// Every cached read registered in this binary, as owned values.
#[must_use]
pub fn registered_reads() -> Vec<CachedRead> {
    inventory::iter::<CachedReadDescriptor>()
        .map(|d| CachedRead {
            id: d.id.to_string(),
            kind: d.kind,
            reads: d.reads.iter().map(|f| f().to_string()).collect(),
            provenance: d.provenance,
            acknowledged_stale: d.acknowledged_stale.map(ToString::to_string),
            location: d.location.to_string(),
        })
        .collect()
}

/// Every repository mutation registered in this binary, as owned values.
#[must_use]
pub fn registered_mutations() -> Vec<Mutation> {
    let direct: Vec<Mutation> = inventory::iter::<MutationDescriptor>()
        .map(|d| Mutation {
            repository: d.repository.to_string(),
            method: d.method.to_string(),
            model: (d.model)().to_string(),
            table: d.table.to_string(),
            invalidates: d.invalidates.iter().map(|s| (*s).to_string()).collect(),
            acknowledged_stale: d.acknowledged_stale.map(ToString::to_string),
            location: d.location.to_string(),
            cascades_into: d.cascades_into.map(|(name, _)| name.to_string()),
            cascade_recurses: d.cascades_into.is_some_and(|(_, recurses)| recurses),
        })
        .collect();
    close_cascades(&direct)
}

/// Prove cache coherence over everything this binary registers.
///
/// This is the library form of the `autumn cache audit` gate: an app can assert
/// on it from its own test suite (`assert!(autumn_web::cache::coherence::audit()
/// .violations.is_empty())`) and get a red `cargo test` the moment a write can
/// strand a cached read.
#[must_use]
pub fn audit() -> CoherenceManifest {
    CoherenceManifest::build(&registered_reads(), &registered_mutations())
}

/// Whether the app was started to dump its cache-coherence manifest.
#[must_use]
pub fn is_dump_mode() -> bool {
    std::env::var("AUTUMN_DUMP_CACHE_COHERENCE").as_deref() == Ok("1")
}

/// Print the manifest on the marker line `autumn cache audit` parses.
///
/// # Panics
///
/// Never in practice: every field is a plain owned value with an infallible
/// `Serialize` impl.
pub fn print_manifest_dump(manifest: &CoherenceManifest) {
    println!(
        "{COHERENCE_MANIFEST_MARKER}{}",
        serde_json::to_string(manifest).expect("cache-coherence manifest is always serializable")
    );
}

/// Extract the manifest JSON from a child process's stdout.
///
/// Returns `None` when the marker line is absent — an app built before this
/// feature, or one that failed before the dump.
#[must_use]
pub fn parse_manifest_dump(stdout: &str) -> Option<CoherenceManifest> {
    let line = stdout
        .lines()
        .find_map(|l| l.strip_prefix(COHERENCE_MANIFEST_MARKER))?;
    serde_json::from_str(line).ok()
}

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

    fn read(id: &str, models: &[&str]) -> CachedRead {
        CachedRead {
            id: id.to_string(),
            kind: ReadKind::Cached,
            reads: models.iter().map(|m| (*m).to_string()).collect(),
            provenance: DependencyProvenance::Declared,
            acknowledged_stale: None,
            location: "src/views.rs:10".to_string(),
        }
    }

    fn mutation(repo: &str, method: &str, model: &str) -> Mutation {
        Mutation {
            repository: repo.to_string(),
            method: method.to_string(),
            model: model.to_string(),
            table: "posts".to_string(),
            invalidates: Vec::new(),
            acknowledged_stale: None,
            location: "src/repositories.rs:20".to_string(),
            cascades_into: None,
            cascade_recurses: false,
        }
    }

    // ── Cascade closure (#2364) ──────────────────────────────────────

    fn cascade(repo: &str, method: &str, model: &str, into: &str) -> Mutation {
        Mutation {
            cascades_into: Some(into.to_string()),
            cascade_recurses: true,
            ..mutation(repo, method, model)
        }
    }

    /// Post -> Comment -> Like, `destroy` at each step.
    fn post_comment_like() -> Vec<Mutation> {
        vec![
            mutation("PostRepository", "delete_by_id", "app::Post"),
            cascade(
                "PostRepository",
                "delete_by_id",
                "app::Comment",
                "CommentRepository",
            ),
            mutation("CommentRepository", "delete_by_id", "app::Comment"),
            cascade(
                "CommentRepository",
                "delete_by_id",
                "app::Like",
                "LikeRepository",
            ),
            mutation("LikeRepository", "delete_by_id", "app::Like"),
        ]
    }

    fn models_of(mutations: &[Mutation], repo: &str, method: &str) -> Vec<String> {
        let mut models: Vec<String> = mutations
            .iter()
            .filter(|m| m.repository == repo && m.method == method)
            .map(|m| m.model.clone())
            .collect();
        models.sort();
        models
    }

    #[test]
    fn a_nested_cascade_mutates_the_grandchild_model() {
        let closed = close_cascades(&post_comment_like());
        assert_eq!(
            models_of(&closed, "PostRepository", "delete_by_id"),
            ["app::Comment", "app::Like", "app::Post"]
        );
    }

    #[test]
    fn a_grandchild_read_without_a_root_edge_is_flagged() {
        let reads = vec![read("app::likes", &["app::Like"])];
        let mut muts = post_comment_like();
        // The edge sits on the child, not on the root.
        for m in muts
            .iter_mut()
            .filter(|m| m.repository == "CommentRepository")
        {
            m.invalidates = vec!["app::likes".to_string()];
        }
        let findings = check(&reads, &close_cascades(&muts));
        assert!(
            findings
                .iter()
                .any(|f| f.mutation == "PostRepository::delete_by_id"),
            "{findings:?}"
        );
    }

    #[test]
    fn a_root_edge_covers_the_grandchild_read() {
        let reads = vec![read("app::likes", &["app::Like"])];
        let mut muts = post_comment_like();
        for m in muts.iter_mut().filter(|m| m.repository == "PostRepository") {
            m.invalidates = vec!["app::likes".to_string()];
        }
        let closed = close_cascades(&muts);
        let like_writes = closed
            .iter()
            .filter(|m| m.repository == "PostRepository" && m.model == "app::Like")
            .count();
        assert_eq!(like_writes, 1, "{closed:?}");
        assert!(
            check(&reads, &closed)
                .iter()
                .all(|f| f.mutation != "PostRepository::delete_by_id"),
            "{closed:?}"
        );
    }

    #[test]
    fn the_root_acknowledgement_carries_to_the_grandchild() {
        let mut muts = post_comment_like();
        muts[1].acknowledged_stale = Some("reason".to_string());
        let closed = close_cascades(&muts);
        assert!(
            closed
                .iter()
                .filter(|m| m.repository == "PostRepository" && m.model == "app::Like")
                .all(|m| m.acknowledged_stale.is_some())
        );
    }

    #[test]
    fn a_restrict_step_adds_no_mutation() {
        // `restrict` registers no cascade descriptor, so the chain stops.
        let mut muts = post_comment_like();
        muts.retain(|m| m.cascades_into.as_deref() != Some("LikeRepository"));
        let closed = close_cascades(&muts);
        assert_eq!(
            models_of(&closed, "PostRepository", "delete_by_id"),
            ["app::Comment", "app::Post"]
        );
    }

    #[test]
    fn a_nullify_step_is_written_but_not_followed() {
        // Post -destroy-> Comment -nullify-> Like -destroy-> Reaction.
        // The `nullify` leg writes `Like` only. It never reaches `Reaction`.
        let mut muts = post_comment_like();
        muts[3].cascade_recurses = false;
        muts.push(cascade(
            "LikeRepository",
            "delete_by_id",
            "app::Reaction",
            "ReactionRepository",
        ));
        let closed = close_cascades(&muts);
        assert_eq!(
            models_of(&closed, "PostRepository", "delete_by_id"),
            ["app::Comment", "app::Like", "app::Post"]
        );
    }

    #[test]
    fn a_cascade_link_round_trips_through_json() {
        let m = cascade(
            "PostRepository",
            "delete_by_id",
            "app::Comment",
            "CommentRepository",
        );
        let back: Mutation = serde_json::from_str(&serde_json::to_string(&m).unwrap()).unwrap();
        assert_eq!(back, m);
        let plain = serde_json::to_string(&mutation("A", "save", "app::A")).unwrap();
        assert!(!plain.contains("cascade"), "{plain}");
    }

    #[test]
    fn a_cascade_cycle_terminates() {
        let muts = vec![
            cascade("ARepository", "delete_by_id", "app::B", "BRepository"),
            cascade("BRepository", "delete_by_id", "app::A", "ARepository"),
            cascade("CRepository", "delete_by_id", "app::C", "CRepository"),
        ];
        let closed = close_cascades(&muts);
        assert_eq!(
            models_of(&closed, "ARepository", "delete_by_id"),
            ["app::A", "app::B"]
        );
        assert_eq!(
            models_of(&closed, "CRepository", "delete_by_id"),
            ["app::C"]
        );
    }

    #[test]
    fn two_same_named_roots_keep_their_own_edges() {
        let mut a = post_comment_like();
        let mut b = post_comment_like();
        for m in &mut b {
            m.location = "src/other.rs:1".to_string();
        }
        a.extend(b);
        let like_rows = close_cascades(&a)
            .iter()
            .filter(|m| m.repository == "PostRepository" && m.model == "app::Like")
            .count();
        assert_eq!(like_rows, 2);
    }

    #[test]
    fn a_child_may_share_its_parents_trait_name() {
        // `admin::CommentRepository` destroys into `blog::CommentRepository`.
        let muts = vec![
            cascade(
                "CommentRepository",
                "delete_by_id",
                "app::Comment",
                "CommentRepository",
            ),
            Mutation {
                location: "src/blog.rs:1".to_string(),
                ..cascade(
                    "CommentRepository",
                    "delete_by_id",
                    "app::Like",
                    "LikeRepository",
                )
            },
        ];
        let closed = close_cascades(&muts);
        let root_like = closed
            .iter()
            .filter(|m| m.model == "app::Like" && m.location == "src/repositories.rs:20")
            .count();
        assert_eq!(root_like, 1, "{closed:?}");
    }

    #[test]
    fn closing_twice_changes_nothing() {
        let once = close_cascades(&post_comment_like());
        assert_eq!(close_cascades(&once), once);
    }

    #[test]
    fn the_manifest_lists_the_closed_mutations() {
        let manifest = CoherenceManifest::build(&[], &post_comment_like());
        assert!(
            manifest
                .dimensions
                .mutations
                .entries
                .iter()
                .any(|m| m.name == "PostRepository::delete_by_id" && m.model == "app::Like")
        );
    }

    // ── The rule ─────────────────────────────────────────────────────

    #[test]
    fn intersecting_model_without_invalidation_is_a_violation() {
        let reads = vec![read("blog::views::recent_posts", &["blog::models::Post"])];
        let muts = vec![mutation("PostRepository", "save", "blog::models::Post")];
        let findings = check(&reads, &muts);
        assert_eq!(findings.len(), 1);
        assert_eq!(findings[0].read, "blog::views::recent_posts");
        assert_eq!(findings[0].mutation, "PostRepository::save");
        assert_eq!(findings[0].model, "blog::models::Post");
    }

    #[test]
    fn disjoint_models_are_coherent() {
        let reads = vec![read("blog::views::recent_posts", &["blog::models::Post"])];
        let muts = vec![mutation("TagRepository", "save", "blog::models::Tag")];
        assert!(check(&reads, &muts).is_empty());
    }

    #[test]
    fn declared_invalidation_discharges_the_obligation() {
        let reads = vec![read("blog::views::recent_posts", &["blog::models::Post"])];
        let mut m = mutation("PostRepository", "save", "blog::models::Post");
        m.invalidates = vec!["blog::views::recent_posts".to_string()];
        assert!(check(&reads, &[m]).is_empty());
    }

    #[test]
    fn invalidation_of_a_different_read_does_not_discharge() {
        let reads = vec![read("blog::views::recent_posts", &["blog::models::Post"])];
        let mut m = mutation("PostRepository", "save", "blog::models::Post");
        m.invalidates = vec!["blog::views::post_count".to_string()];
        assert_eq!(check(&reads, &[m]).len(), 1);
    }

    #[test]
    fn acknowledged_stale_read_is_exempt() {
        let mut r = read("blog::views::recent_posts", &["blog::models::Post"]);
        r.acknowledged_stale = Some("ttl of 5s is short enough".to_string());
        let muts = vec![mutation("PostRepository", "save", "blog::models::Post")];
        assert!(check(&[r], &muts).is_empty());
    }

    #[test]
    fn acknowledged_stale_mutation_is_exempt() {
        let reads = vec![read("blog::views::recent_posts", &["blog::models::Post"])];
        let mut m = mutation("PostRepository", "save", "blog::models::Post");
        m.acknowledged_stale = Some("import-only path".to_string());
        assert!(check(&reads, &[m]).is_empty());
    }

    #[test]
    fn undetermined_reads_never_fail_the_default_gate() {
        let mut r = read("blog::views::mystery", &[]);
        r.provenance = DependencyProvenance::Undetermined;
        let muts = vec![mutation("PostRepository", "save", "blog::models::Post")];
        assert!(check(&[r], &muts).is_empty());
    }

    #[test]
    fn undetermined_reads_are_reported_separately() {
        let mut r = read("blog::views::mystery", &[]);
        r.provenance = DependencyProvenance::Undetermined;
        let ok = read("blog::views::recent_posts", &["blog::models::Post"]);
        let all = [r, ok];
        let undetermined = undetermined_reads(&all);
        assert_eq!(undetermined.len(), 1);
        assert_eq!(undetermined[0].id, "blog::views::mystery");
    }

    /// `acknowledge_stale` is documented as the opt-out from the gate, so it
    /// has to hold on the undetermined path too — otherwise a read the author
    /// signed off on fails `--strict` through a second door, and the only way
    /// to get green is to delete the acknowledgement that made the opt-out
    /// explicit.
    #[test]
    fn an_acknowledged_read_is_not_a_strict_undetermined_failure() {
        let mut r = read("blog::views::mystery", &[]);
        r.provenance = DependencyProvenance::Undetermined;
        r.acknowledged_stale = Some("ttl of 5s is short enough".to_string());
        assert!(
            undetermined_reads(std::slice::from_ref(&r)).is_empty(),
            "an acknowledged read must not be reported as undetermined"
        );
        let manifest = CoherenceManifest::build(std::slice::from_ref(&r), &[]);
        assert!(!gate_failed(&manifest, true), "--strict must stay green");
        // Still visible: opting out is loud, not silent.
        assert_eq!(
            manifest.dimensions.cached_reads.entries[0]
                .acknowledged_stale
                .as_deref(),
            Some("ttl of 5s is short enough")
        );
        assert!(manifest.summary().contains("1 acknowledged-stale"));
    }

    #[test]
    fn one_read_over_two_models_reports_one_finding_per_mutation() {
        let reads = vec![read("blog::views::feed", &["Post", "Comment"])];
        let muts = vec![
            mutation("PostRepository", "save", "blog::models::Post"),
            mutation("CommentRepository", "save", "blog::models::Comment"),
        ];
        assert_eq!(check(&reads, &muts).len(), 2);
    }

    #[test]
    fn a_model_named_twice_by_one_read_still_yields_one_finding() {
        // A read may carry both a declared and a derived spelling of the same
        // dependency; that must not double-report.
        let reads = vec![read("blog::views::feed", &["Post", "Post"])];
        let muts = vec![mutation("PostRepository", "save", "blog::models::Post")];
        assert_eq!(check(&reads, &muts).len(), 1);
    }

    #[test]
    fn findings_are_stable_ordered() {
        let reads = vec![read("z::read", &["Post"]), read("a::read", &["Post"])];
        let muts = vec![
            mutation("PostRepository", "update", "Post"),
            mutation("PostRepository", "delete_by_id", "Post"),
        ];
        let findings = check(&reads, &muts);
        let keys: Vec<_> = findings
            .iter()
            .map(|f| format!("{}|{}", f.read, f.mutation))
            .collect();
        let mut sorted = keys.clone();
        sorted.sort();
        assert_eq!(
            keys, sorted,
            "findings must come back deterministically ordered"
        );
    }

    // ── Model identity ───────────────────────────────────────────────

    #[test]
    fn model_key_ignores_generic_arguments_and_references() {
        assert_eq!(model_key("&blog::models::Post"), "Post");
        assert_eq!(
            model_key("blog::models::Wrapper<blog::models::Post>"),
            "Wrapper"
        );
        assert_eq!(model_key("  Post  "), "Post");
        assert_eq!(model_key("&mut blog::Post"), "Post");
        assert_eq!(model_key("*const blog::Post"), "Post");
    }

    #[test]
    fn a_derived_bare_ident_matches_a_qualified_type_name() {
        // The read names the bare ident derivation recovered; the mutation
        // names the fully-qualified `type_name`. Same model.
        let reads = vec![read("blog::views::recent_posts", &["Post"])];
        let muts = vec![mutation("PostRepository", "save", "blog::models::Post")];
        assert_eq!(check(&reads, &muts).len(), 1);
    }

    #[test]
    fn two_qualified_same_named_models_are_kept_apart() {
        // `plugin::models::User` and `crate::models::User` are different models.
        // Collapsing them would fail a perfectly correct app.
        assert!(!models_match("plugin::models::User", "app::models::User"));
        assert!(models_match("app::models::User", "app::models::User"));

        let reads = vec![read("app::views::roster", &["app::models::User"])];
        let muts = vec![mutation("UserRepository", "save", "plugin::models::User")];
        assert!(
            check(&reads, &muts).is_empty(),
            "a different module's same-named model must not be flagged"
        );
    }

    #[test]
    fn a_bare_ident_still_over_approximates_across_modules() {
        // When one side is all the analysis could recover, matching falls back
        // to the bare name — a false failure, never a false pass.
        assert!(models_match("User", "plugin::models::User"));
    }

    // ── Diagnostics ──────────────────────────────────────────────────

    #[test]
    fn diagnostic_names_read_mutation_and_shared_model() {
        let reads = vec![read("blog::views::recent_posts", &["blog::models::Post"])];
        let muts = vec![mutation("PostRepository", "save", "blog::models::Post")];
        let text = format_diagnostic(&check(&reads, &muts));
        assert!(text.contains("blog::views::recent_posts"), "{text}");
        assert!(text.contains("PostRepository::save"), "{text}");
        assert!(text.contains("blog::models::Post"), "{text}");
        assert!(text.contains("src/views.rs:10"), "{text}");
        assert!(text.contains("src/repositories.rs:20"), "{text}");
        assert!(text.contains("invalidates("), "must name the fix: {text}");
    }

    #[test]
    fn diagnostic_groups_one_missing_edge_into_one_block() {
        // A missing `invalidates(...)` on a repository produces one finding per
        // generated write method. Twelve identical blocks with the same one-line
        // fix is the output shape that gets a gate switched off.
        let reads = vec![read("blog::views::recent_posts", &["Post"])];
        let muts: Vec<Mutation> = [
            "save",
            "update",
            "delete_by_id",
            "save_many",
            "update_many",
            "delete_many",
        ]
        .iter()
        .map(|m| mutation("PostRepository", m, "Post"))
        .collect();

        let findings = check(&reads, &muts);
        assert_eq!(findings.len(), 6, "the manifest still carries every pair");

        let text = format_diagnostic(&findings);
        assert_eq!(
            text.matches("fix: add invalidates(").count(),
            1,
            "one edit, one fix line: {text}"
        );
        assert!(text.contains("1 cached read can be left stale"), "{text}");
        for method in ["save", "update", "delete_by_id", "delete_many"] {
            assert!(
                text.contains(&format!("PostRepository::{method}")),
                "every uncovered write is still named: {text}"
            );
        }
    }

    #[test]
    fn diagnostic_separates_two_reads_over_the_same_model() {
        let reads = vec![read("a::read", &["Post"]), read("b::read", &["Post"])];
        let muts = vec![mutation("PostRepository", "save", "Post")];
        let text = format_diagnostic(&check(&reads, &muts));
        assert_eq!(text.matches("fix: add invalidates(").count(), 2, "{text}");
        assert!(text.contains("2 cached reads can be left stale"), "{text}");
    }

    #[test]
    fn diagnostic_is_empty_when_nothing_is_wrong() {
        assert!(format_diagnostic(&[]).is_empty());
    }

    #[test]
    fn summary_counts_each_provenance_class_and_every_acknowledgement() {
        let mut undetermined = read("u", &[]);
        undetermined.provenance = DependencyProvenance::Undetermined;
        let mut derived = read("d", &["Post"]);
        derived.provenance = DependencyProvenance::Derived;
        let mut declared = read("c", &["Post"]);
        declared.acknowledged_stale = Some("deliberate".to_string());
        let mut m = mutation("PostRepository", "save", "Post");
        m.acknowledged_stale = Some("seed only".to_string());

        let text = format_summary(&[undetermined, derived, declared], &[m]);
        assert!(text.contains("3 cached reads"), "{text}");
        assert!(text.contains("1 declared"), "{text}");
        assert!(text.contains("1 derived"), "{text}");
        assert!(text.contains("1 undetermined"), "{text}");
        assert!(text.contains("1 repository mutations"), "{text}");
        assert!(text.contains("2 acknowledged-stale"), "{text}");
    }

    // ── Manifest ─────────────────────────────────────────────────────

    #[test]
    fn manifest_is_provenance_tagged_and_stable_ordered() {
        let reads = vec![read("z::read", &["Post"]), read("a::read", &["Post"])];
        let muts = vec![mutation("PostRepository", "save", "Post")];
        let manifest = CoherenceManifest::build(&reads, &muts);
        assert_eq!(manifest.schema_version, MANIFEST_SCHEMA_VERSION);
        assert_eq!(manifest.dimensions.cached_reads.provenance, "provable");
        assert_eq!(manifest.dimensions.mutations.provenance, "provable");
        assert_eq!(manifest.dimensions.invalidations.provenance, "provable");
        assert!(
            !manifest.dimensions.invalidations.runtime_caveat.is_empty(),
            "a `declared` dimension must carry its caveat"
        );
        let ids: Vec<_> = manifest
            .dimensions
            .cached_reads
            .entries
            .iter()
            .map(|e| e.id.as_str())
            .collect();
        assert_eq!(ids, vec!["a::read", "z::read"]);
        assert_eq!(manifest.violations.len(), 2);
    }

    #[test]
    fn manifest_carries_every_entry_through_json() {
        let mut r = read("blog::views::recent_posts", &["blog::models::Post"]);
        r.acknowledged_stale = Some("deliberate".to_string());
        let mut m = mutation("PostRepository", "save", "blog::models::Post");
        m.invalidates = vec!["blog::views::recent_posts".to_string()];

        let manifest = CoherenceManifest::build(&[r], &[m]);
        let json: serde_json::Value = serde_json::from_str(&manifest.to_json()).unwrap();

        assert_eq!(json["schema_version"], MANIFEST_SCHEMA_VERSION);
        let entry = &json["dimensions"]["cached_reads"]["entries"][0];
        assert_eq!(entry["id"], "blog::views::recent_posts");
        assert_eq!(entry["kind"], "cached");
        assert_eq!(entry["provenance"], "declared");
        assert_eq!(entry["reads"][0], "blog::models::Post");
        assert_eq!(entry["acknowledged_stale"], "deliberate");
        assert_eq!(entry["location"], "src/views.rs:10");

        let write = &json["dimensions"]["mutations"]["entries"][0];
        assert_eq!(write["name"], "PostRepository::save");
        assert_eq!(write["model"], "blog::models::Post");
        assert_eq!(write["table"], "posts");

        let edge = &json["dimensions"]["invalidations"]["entries"][0];
        assert_eq!(edge["mutation"], "PostRepository::save");
        assert_eq!(edge["read"], "blog::views::recent_posts");
    }

    #[test]
    fn manifest_omits_an_absent_runtime_caveat_rather_than_emitting_an_empty_one() {
        let json: serde_json::Value =
            serde_json::from_str(&CoherenceManifest::build(&[], &[]).to_json()).unwrap();
        // Every dimension here carries a caveat, so the `skip_serializing_if`
        // is exercised by constructing one directly rather than via `build`.
        let bare: Dimension<ManifestRead> = Dimension {
            provenance: "provable".to_string(),
            source: "test".to_string(),
            runtime_caveat: String::new(),
            entries: Vec::new(),
        };
        let bare_json: serde_json::Value = serde_json::to_value(&bare).unwrap();
        assert!(bare_json["runtime_caveat"].is_null());
        assert!(json["dimensions"]["invalidations"]["runtime_caveat"].is_string());
    }

    #[test]
    fn manifest_undetermined_entries_carry_their_location() {
        let mut r = read("blog::views::mystery", &[]);
        r.provenance = DependencyProvenance::Undetermined;
        let manifest = CoherenceManifest::build(&[r], &[]);
        assert_eq!(manifest.undetermined_reads.len(), 1);
        assert_eq!(manifest.undetermined_reads[0].id, "blog::views::mystery");
        assert_eq!(manifest.undetermined_reads[0].location, "src/views.rs:10");
    }

    #[test]
    fn excluded_dimensions_name_the_holes_this_slice_leaves() {
        // The manifest's whole discipline is "say what we did not look at". A
        // reader must be able to tell "checked and fine" from "never looked".
        let named: Vec<&str> = excluded_dimensions()
            .iter()
            .map(|d| d.dimension.as_str())
            .collect::<Vec<_>>()
            .iter()
            .map(|s| Box::leak(s.to_string().into_boxed_str()) as &str)
            .collect();
        for expected in [
            "row_and_column_scoped_dependencies",
            "invalidation_call_sites",
            "writes_outside_repository",
            "undeclared_cache_call_sites",
            "counter_cache_and_cascade_writes",
            "http_response_cache",
            "cross_service_coherence",
            "ttl_expiry",
        ] {
            assert!(named.contains(&expected), "missing {expected} in {named:?}");
        }
        assert!(
            excluded_dimensions().iter().all(|d| !d.reason.is_empty()),
            "an exclusion without a reason is not an exclusion"
        );
    }

    #[test]
    fn round_trips_through_the_dump_protocol() {
        let reads = vec![read("blog::views::recent_posts", &["Post"])];
        let muts = vec![mutation("PostRepository", "save", "Post")];
        let manifest = CoherenceManifest::build(&reads, &muts);
        let dump = format!(
            "some unrelated startup logging\n{COHERENCE_MANIFEST_MARKER}{}\n",
            serde_json::to_string(&manifest).unwrap()
        );
        let parsed = parse_manifest_dump(&dump).expect("marker line must be found");
        assert_eq!(parsed.violations.len(), 1);
        assert_eq!(parsed.summary(), manifest.summary());
    }

    #[test]
    fn a_dump_without_the_marker_is_not_mistaken_for_an_empty_manifest() {
        assert!(parse_manifest_dump("no marker here\n").is_none());
    }

    #[test]
    fn a_corrupt_manifest_is_not_mistaken_for_an_empty_one() {
        let dump = format!("{COHERENCE_MANIFEST_MARKER}{{\"schema_version\": \n");
        assert!(parse_manifest_dump(&dump).is_none());
    }

    // ── The gate ─────────────────────────────────────────────────────

    #[test]
    fn exit_code_is_nonzero_iff_a_violation_exists() {
        let reads = vec![read("r", &["Post"])];
        let muts = vec![mutation("PostRepository", "save", "Post")];
        assert_eq!(
            audit_exit_code(&CoherenceManifest::build(&reads, &muts), false),
            1
        );
        assert_eq!(
            audit_exit_code(&CoherenceManifest::build(&[], &[]), false),
            0
        );
    }

    #[test]
    fn strict_mode_fails_on_undetermined_reads() {
        let mut r = read("blog::views::mystery", &[]);
        r.provenance = DependencyProvenance::Undetermined;
        let manifest = CoherenceManifest::build(std::slice::from_ref(&r), &[]);
        assert!(!gate_failed(&manifest, false));
        assert!(gate_failed(&manifest, true));
    }

    // ── Namespace invalidation ───────────────────────────────────────

    #[cfg(all(feature = "cache-moka", feature = "reporting"))]
    #[test]
    fn a_local_namespace_invalidation_is_on_the_capsule_seam() {
        // Codex review on #3222: with no installed cache, an invalidation
        // reaches only process-local stores, off the seam. It is recorded
        // here, and a replay answers it from the tape without the stores.
        let _guard = super::super::GLOBAL_CACHE_TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        super::super::clear_global_cache();
        let store = std::sync::Arc::new(super::super::MokaCache::new(16, None));
        register_namespace_store("tests::seam_ns", store.clone());

        let scope = std::sync::Arc::new(crate::capsule::CaptureScope::new(
            "coherence".to_owned(),
            std::sync::Arc::new(crate::capsule::CaptureSettings::default()),
            std::sync::Arc::new(crate::log::filter::ParameterFilter::default()),
        ));
        let rt = tokio::runtime::Builder::new_current_thread()
            .build()
            .unwrap();
        assert!(rt.block_on(crate::capsule::capture::with_capture_scope(
            std::sync::Arc::clone(&scope),
            invalidate_namespace_async("tests::seam_ns"),
        )));
        let recorded = scope.effects_snapshot();
        assert_eq!(recorded.cache.len(), 1, "{:?}", recorded.cache);

        // Replayed: the recorded removal is consumed, and the store is left.
        super::super::insert(&*store, "tests::seam_ns:1", 1_i32);
        let tape = std::sync::Arc::new(crate::capsule::ReplayEffects::new(recorded));
        assert!(rt.block_on(crate::capsule::with_effect_tape(
            std::sync::Arc::clone(&tape),
            invalidate_namespace_async("tests::seam_ns"),
        )));
        assert!(tape.finish().is_empty(), "{:?}", tape.finish());
        assert_eq!(
            super::super::get::<i32>(&*store, "tests::seam_ns:1"),
            Some(1)
        );

        // A replay with no recorded removal diverges.
        let empty = std::sync::Arc::new(crate::capsule::ReplayEffects::new(
            crate::capsule::CapsuleEffects::default(),
        ));
        let _ = rt.block_on(crate::capsule::with_effect_tape(
            std::sync::Arc::clone(&empty),
            invalidate_namespace_async("tests::seam_ns"),
        ));
        assert!(!empty.divergences().is_empty());
    }

    /// Codex review on #3222: a shared registered store that fails while
    /// the installed cache succeeds makes the call return `false`. The
    /// capsule records that combined answer, once, so a replay returns it.
    #[cfg(all(feature = "cache-moka", feature = "reporting"))]
    #[test]
    fn a_failed_registered_store_is_in_the_recorded_namespace_removal() {
        struct SharedFailingStore;
        impl super::super::Cache for SharedFailingStore {
            fn get_value(
                &self,
                _key: &str,
            ) -> Option<std::sync::Arc<dyn std::any::Any + Send + Sync>> {
                None
            }
            fn insert_value(
                &self,
                _key: &str,
                _value: std::sync::Arc<dyn std::any::Any + Send + Sync>,
            ) {
            }
            fn invalidate(&self, _key: &str) {}
            fn clear(&self) {}
            fn invalidate_namespace(&self, _namespace: &str) -> bool {
                false
            }
            fn shares_fill_epoch(&self) -> bool {
                true
            }
        }

        let _guard = super::super::GLOBAL_CACHE_TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        super::super::set_global_cache(std::sync::Arc::new(super::super::MokaCache::new(16, None)));
        register_namespace_store(
            "tests::seam_mixed_ns",
            std::sync::Arc::new(SharedFailingStore),
        );
        let rt = tokio::runtime::Builder::new_current_thread()
            .build()
            .unwrap();
        for sync in [true, false] {
            let scope = std::sync::Arc::new(crate::capsule::CaptureScope::new(
                "coherence".to_owned(),
                std::sync::Arc::new(crate::capsule::CaptureSettings::default()),
                std::sync::Arc::new(crate::log::filter::ParameterFilter::default()),
            ));
            let run = || async move {
                if sync {
                    invalidate_namespace("tests::seam_mixed_ns")
                } else {
                    invalidate_namespace_async("tests::seam_mixed_ns").await
                }
            };
            assert!(!rt.block_on(crate::capsule::capture::with_capture_scope(
                std::sync::Arc::clone(&scope),
                run(),
            )));
            let recorded = scope.effects_snapshot();
            assert_eq!(recorded.cache.len(), 1, "sync={sync}: {:?}", recorded.cache);

            let tape = std::sync::Arc::new(crate::capsule::ReplayEffects::new(recorded));
            assert!(
                !rt.block_on(crate::capsule::with_effect_tape(
                    std::sync::Arc::clone(&tape),
                    run(),
                )),
                "sync={sync}: the replay returns the recorded `false`"
            );
            assert!(tape.finish().is_empty(), "{:?}", tape.finish());
        }
        super::super::clear_global_cache();
    }

    #[cfg(feature = "cache-moka")]
    #[test]
    fn invalidating_a_namespace_clears_only_that_reads_store() {
        // Serialized against every other test that mutates the process-wide
        // global cache: `invalidate_namespace` reads it.
        let _guard = super::super::GLOBAL_CACHE_TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        super::super::clear_global_cache();

        let a = std::sync::Arc::new(super::super::MokaCache::new(16, None));
        let b = std::sync::Arc::new(super::super::MokaCache::new(16, None));
        register_namespace_store("tests::ns_a", a.clone());
        register_namespace_store("tests::ns_b", b.clone());
        super::super::insert(&*a, "tests::ns_a:1", 1_i32);
        super::super::insert(&*b, "tests::ns_b:1", 2_i32);

        assert!(invalidate_namespace("tests::ns_a"));

        assert_eq!(super::super::get::<i32>(&*a, "tests::ns_a:1"), None);
        assert_eq!(super::super::get::<i32>(&*b, "tests::ns_b:1"), Some(2));
    }

    #[cfg(feature = "cache-moka")]
    #[test]
    fn invalidating_an_unregistered_namespace_is_complete_and_harmless() {
        let _guard = super::super::GLOBAL_CACHE_TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        super::super::clear_global_cache();
        assert!(invalidate_namespace("tests::never_registered"));
    }

    #[cfg(feature = "cache-moka")]
    #[test]
    fn a_shared_backend_is_asked_to_drop_the_namespace_too() {
        // Once a process-level backend is registered THAT is where the value
        // lives; clearing only the per-function store would be a no-op dressed
        // up as an invalidation.
        let _guard = super::super::GLOBAL_CACHE_TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        let global = std::sync::Arc::new(super::super::MokaCache::new(64, None));
        super::super::set_global_cache(global.clone());
        super::super::insert(&*global, "tests::shared_ns:aaa", 1_i32);
        super::super::insert(&*global, "tests::shared_ns:bbb", 2_i32);
        super::super::insert(&*global, "tests::other_ns:aaa", 3_i32);

        assert!(
            invalidate_namespace("tests::shared_ns"),
            "a backend that can drop a namespace reports a complete invalidation"
        );
        assert_eq!(
            super::super::get::<i32>(&*global, "tests::shared_ns:aaa"),
            None
        );
        assert_eq!(
            super::super::get::<i32>(&*global, "tests::shared_ns:bbb"),
            None
        );
        assert_eq!(
            super::super::get::<i32>(&*global, "tests::other_ns:aaa"),
            Some(3),
            "a neighbouring namespace must survive"
        );

        super::super::clear_global_cache();
    }

    #[test]
    fn a_backend_that_cannot_enumerate_reports_an_incomplete_invalidation() {
        // The honest signal: a custom backend with no way to pattern-match its
        // key space must not let a caller believe the value is gone.
        struct OpaqueBackend;
        impl super::super::Cache for OpaqueBackend {
            fn get_value(
                &self,
                _key: &str,
            ) -> Option<std::sync::Arc<dyn std::any::Any + Send + Sync>> {
                None
            }
            fn insert_value(
                &self,
                _key: &str,
                _value: std::sync::Arc<dyn std::any::Any + Send + Sync>,
            ) {
            }
            fn invalidate(&self, _key: &str) {}
            fn clear(&self) {}
        }

        let _guard = super::super::GLOBAL_CACHE_TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        super::super::set_global_cache(std::sync::Arc::new(OpaqueBackend));
        assert!(!invalidate_namespace("tests::opaque_ns"));
        super::super::clear_global_cache();
    }

    // ── Codex round 1 ────────────────────────────────────────────────

    #[test]
    fn a_blank_acknowledgement_reason_is_blank_by_the_same_rule_everywhere() {
        // `declare_cached_read!` asserts on this in a const context, and it has
        // to agree with the `trim().is_empty()` the attribute macros apply — or
        // the weakest declaration form becomes the way to get a free pass.
        assert!(reason_is_blank(""));
        assert!(reason_is_blank("   "));
        assert!(reason_is_blank("\t\n "));
        assert!(!reason_is_blank("5s lag is fine"));
        assert!(!reason_is_blank("  x  "));
    }

    /// "Blank" has to mean the same thing on both sides, and `str::trim` — what
    /// the attribute macros apply — is Unicode-aware. An ASCII-only const check
    /// would let `acknowledge_stale = "\u{2003}"` through `declare_cached_read!`
    /// while `#[cached]` rejected it: the same reason accepted or refused
    /// depending on which form you wrote it in, with the weaker form the way to
    /// buy a free pass.
    #[test]
    fn unicode_whitespace_is_a_blank_reason_too() {
        for blank in [
            "\u{2003}",
            "\u{00a0}",
            "\u{3000}",
            "\u{2028}",
            " \u{2003}\t",
        ] {
            assert!(
                reason_is_blank(blank),
                "{blank:?} is whitespace to `trim`, so it must be blank here too"
            );
            assert!(blank.trim().is_empty(), "the premise: {blank:?}");
        }
        // A multi-byte non-space is still a reason: the decoder must not
        // mistake every wide character for whitespace.
        for reason in ["\u{2003}ok", "日本語", "→ stale is fine"] {
            assert!(!reason_is_blank(reason), "{reason:?}");
            assert!(!reason.trim().is_empty(), "the premise: {reason:?}");
        }
    }

    #[test]
    fn an_identity_claimed_twice_is_reported() {
        let reads = vec![
            read("app::Stats::count", &["Post"]),
            read("app::Stats::count", &["Post"]),
            read("app::views::feed", &["Post"]),
        ];
        assert_eq!(duplicate_read_ids(&reads), vec!["app::Stats::count"]);
        assert!(
            CoherenceManifest::build(&reads, &[])
                .duplicate_read_ids
                .contains(&"app::Stats::count".to_string())
        );
    }

    #[test]
    fn distinct_identities_are_not_reported_as_duplicates() {
        let reads = vec![read("a", &["Post"]), read("b", &["Post"])];
        assert!(duplicate_read_ids(&reads).is_empty());
        assert!(
            CoherenceManifest::build(&reads, &[])
                .duplicate_read_ids
                .is_empty()
        );
    }

    #[cfg(feature = "cache-moka")]
    #[test]
    fn every_store_sharing_a_namespace_is_cleared() {
        // Two `#[cached]` associated functions with the same name in one module
        // register under the same identity. Overwriting the registration would
        // leave one of them stale while reporting success.
        let _guard = super::super::GLOBAL_CACHE_TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        super::super::clear_global_cache();

        let first = std::sync::Arc::new(super::super::MokaCache::new(16, None));
        let second = std::sync::Arc::new(super::super::MokaCache::new(16, None));
        register_namespace_store("tests::shared_identity", first.clone());
        register_namespace_store("tests::shared_identity", second.clone());
        super::super::insert(&*first, "tests::shared_identity:1", 1_i32);
        super::super::insert(&*second, "tests::shared_identity:2", 2_i32);

        assert!(invalidate_namespace("tests::shared_identity"));

        assert_eq!(
            super::super::get::<i32>(&*first, "tests::shared_identity:1"),
            None
        );
        assert_eq!(
            super::super::get::<i32>(&*second, "tests::shared_identity:2"),
            None,
            "the second registration must not be left stale"
        );
    }

    #[cfg(feature = "cache-moka")]
    #[test]
    fn an_invalidation_fences_a_fill_that_was_already_in_flight() {
        // The race: a miss starts computing, a write commits and invalidates,
        // then the in-flight fill lands and resurrects the value that was just
        // dropped. The epoch is what makes the invalidator's `true` true.
        let _guard = super::super::GLOBAL_CACHE_TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        super::super::clear_global_cache();

        let store = std::sync::Arc::new(super::super::MokaCache::new(16, None));
        let epoch = register_namespace_store("tests::fenced_ns", store);

        // A fill samples the epoch before computing...
        let sampled = epoch.load(std::sync::atomic::Ordering::Acquire);
        assert!(fill_is_still_current(&epoch, sampled));

        // ...a write invalidates while it is still computing...
        assert!(invalidate_namespace("tests::fenced_ns"));

        // ...so the fill must not write its now-stale value back.
        assert!(
            !fill_is_still_current(&epoch, sampled),
            "an in-flight fill must be fenced out by the invalidation"
        );

        // A fill that starts after the invalidation is fine again.
        let after = epoch.load(std::sync::atomic::Ordering::Acquire);
        assert!(fill_is_still_current(&epoch, after));
    }

    #[test]
    fn the_fill_fence_inserts_when_current_and_skips_when_invalidated() {
        let epoch = std::sync::atomic::AtomicU64::new(0);
        let sampled = epoch.load(std::sync::atomic::Ordering::Acquire);
        assert_eq!(
            with_fill_fence(&epoch, sampled, || "inserted"),
            Some("inserted")
        );

        epoch.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
        assert_eq!(
            with_fill_fence(&epoch, sampled, || "inserted"),
            None,
            "a fill whose epoch moved must not run its insert at all"
        );
    }

    #[cfg(feature = "cache-moka")]
    #[test]
    fn an_invalidation_cannot_land_between_the_fence_check_and_the_insert() {
        // The window the epoch comparison alone leaves open: the check returns
        // `true`, an invalidation bumps and clears, and the insert lands after
        // the clear — writing a pre-invalidation value into a namespace that
        // was just emptied, where it sits until its TTL, or forever without
        // one. `with_fill_fence` closes it by making the check and the insert
        // indivisible, which is observable here as an invalidation being unable
        // to bump the epoch while a fill's insert is running.
        let _guard = super::super::GLOBAL_CACHE_TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        super::super::clear_global_cache();

        let store = std::sync::Arc::new(super::super::MokaCache::new(16, None));
        let epoch = register_namespace_store("tests::atomic_fill_ns", store);
        let sampled = epoch.load(std::sync::atomic::Ordering::Acquire);

        let inserted = with_fill_fence(&epoch, sampled, || {
            // An invalidation racing this insert must block until it finishes,
            // not slip in between the check above and the write below.
            let racer = std::thread::spawn(|| invalidate_namespace("tests::atomic_fill_ns"));
            std::thread::sleep(std::time::Duration::from_millis(100));
            let observed = epoch.load(std::sync::atomic::Ordering::Acquire);
            (racer, observed)
        });

        let (racer, observed) = inserted.expect("the fill was current, so it must have run");
        assert_eq!(
            observed, sampled,
            "the epoch moved while the insert was running: the check and the \
             insert are not atomic, so this value would land after the clear"
        );
        assert!(racer.join().expect("invalidator thread panicked"));
        assert!(
            !fill_is_still_current(&epoch, sampled),
            "the invalidation must land once the fill releases the fence"
        );
    }

    #[test]
    fn a_backend_whose_sweep_failed_reports_an_incomplete_invalidation() {
        // A `RedisCache` whose SCAN or DEL errored returns `false` rather than
        // a silent "done"; the caller is using this bool to decide whether
        // stale data may still be served.
        struct FailingSweep;
        impl super::super::Cache for FailingSweep {
            fn get_value(
                &self,
                _key: &str,
            ) -> Option<std::sync::Arc<dyn std::any::Any + Send + Sync>> {
                None
            }
            fn insert_value(
                &self,
                _key: &str,
                _value: std::sync::Arc<dyn std::any::Any + Send + Sync>,
            ) {
            }
            fn invalidate(&self, _key: &str) {}
            fn clear(&self) {}
            fn invalidate_namespace(&self, _namespace: &str) -> bool {
                false
            }
        }

        let _guard = super::super::GLOBAL_CACHE_TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        super::super::set_global_cache(std::sync::Arc::new(FailingSweep));
        assert!(!invalidate_namespace("tests::failed_sweep_ns"));
        super::super::clear_global_cache();
    }

    // ── #3056: async invalidation, surfaced failures ──────────────────

    /// A backend whose namespace sweep always fails.
    struct RefusingBackend;
    impl super::super::Cache for RefusingBackend {
        fn get_value(&self, _key: &str) -> Option<std::sync::Arc<dyn std::any::Any + Send + Sync>> {
            None
        }
        fn insert_value(
            &self,
            _key: &str,
            _value: std::sync::Arc<dyn std::any::Any + Send + Sync>,
        ) {
        }
        fn invalidate(&self, _key: &str) {}
        fn clear(&self) {}
        fn invalidate_namespace(&self, _namespace: &str) -> bool {
            false
        }
    }

    fn block_on<F: std::future::Future>(future: F) -> F::Output {
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .expect("runtime")
            .block_on(future)
    }

    #[test]
    fn async_invalidation_with_no_backend_is_complete() {
        let _guard = super::super::GLOBAL_CACHE_TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        super::super::clear_global_cache();
        assert!(block_on(invalidate_namespace_async(
            "tests::async_no_backend"
        )));
    }

    #[test]
    fn async_invalidation_bumps_the_epoch_first() {
        let epoch = namespace_epoch("tests::async_epoch");
        let before = epoch.load(std::sync::atomic::Ordering::Acquire);
        let _guard = super::super::GLOBAL_CACHE_TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        super::super::clear_global_cache();
        assert!(block_on(invalidate_namespace_async("tests::async_epoch")));
        assert!(
            epoch.load(std::sync::atomic::Ordering::Acquire) > before,
            "a fill already in flight must be fenced out"
        );
    }

    #[test]
    fn a_failed_async_invalidation_is_reported_and_counted() {
        let _guard = super::super::GLOBAL_CACHE_TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        super::super::set_global_cache(std::sync::Arc::new(RefusingBackend));
        let before = super::super::invalidation_failures_total();
        let complete = block_on(invalidate_namespace_async("tests::async_refused"));
        let after = super::super::invalidation_failures_total();
        super::super::clear_global_cache();
        assert!(!complete, "a refused sweep must not report success");
        assert!(after > before, "the failure must be counted");
    }

    #[test]
    fn a_failed_sync_invalidation_is_counted_too() {
        let _guard = super::super::GLOBAL_CACHE_TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        super::super::set_global_cache(std::sync::Arc::new(RefusingBackend));
        let before = super::super::invalidation_failures_total();
        let complete = invalidate_namespace("tests::sync_refused");
        let after = super::super::invalidation_failures_total();
        super::super::clear_global_cache();
        assert!(!complete);
        assert!(after > before, "the failure must be counted");
    }

    // ── declare_cached_read! ─────────────────────────────────────────

    struct DeclaredModelA;
    struct DeclaredModelB;

    crate::declare_cached_read! {
        id = "autumn_web::cache::coherence::tests::declared_fragment",
        kind = Fragment,
        reads = [DeclaredModelA, DeclaredModelB],
    }

    crate::declare_cached_read! {
        id = "autumn_web::cache::coherence::tests::declared_read_through",
        kind = ReadThrough,
        reads = [DeclaredModelA],
        acknowledge_stale = "the sidebar tolerates a stale count",
    }

    #[test]
    fn a_manually_declared_read_reaches_the_manifest() {
        let declared = registered_reads()
            .into_iter()
            .find(|r| r.id.ends_with("tests::declared_fragment"))
            .expect("declare_cached_read! must register the read");
        assert_eq!(declared.kind, ReadKind::Fragment);
        assert_eq!(declared.provenance, DependencyProvenance::Declared);
        assert_eq!(declared.reads.len(), 2);
        assert!(
            declared
                .reads
                .iter()
                .any(|m| model_key(m) == "DeclaredModelA"),
            "{:?}",
            declared.reads
        );
        assert!(declared.location.contains("coherence.rs"));
    }

    #[test]
    fn a_manually_declared_read_can_acknowledge_staleness() {
        let declared = registered_reads()
            .into_iter()
            .find(|r| r.id.ends_with("tests::declared_read_through"))
            .expect("declare_cached_read! must register the read");
        assert_eq!(declared.kind, ReadKind::ReadThrough);
        assert_eq!(
            declared.acknowledged_stale.as_deref(),
            Some("the sidebar tolerates a stale count")
        );
    }

    #[test]
    fn read_kind_spellings_are_stable() {
        // These strings are the manifest's schema; renaming one silently would
        // break every consumer.
        assert_eq!(ReadKind::Cached.as_str(), "cached");
        assert_eq!(ReadKind::Fragment.as_str(), "fragment");
        assert_eq!(ReadKind::ReadThrough.as_str(), "read_through");
        assert_eq!(DependencyProvenance::Declared.as_str(), "declared");
        assert_eq!(DependencyProvenance::Derived.as_str(), "derived");
        assert_eq!(DependencyProvenance::Undetermined.as_str(), "undetermined");
    }

    // ── InvalidateAfterWrite (#3056) ─────────────────────────────────

    static GUARD_RUNS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);

    fn count_guard_run() -> crate::cache::CacheFuture<'static, bool> {
        Box::pin(async {
            GUARD_RUNS.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            true
        })
    }

    fn guard_runs() -> usize {
        GUARD_RUNS.load(std::sync::atomic::Ordering::SeqCst)
    }

    static SAW_SCOPE: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);

    #[cfg(feature = "reporting")]
    fn note_scope() -> crate::cache::CacheFuture<'static, bool> {
        Box::pin(async {
            SAW_SCOPE.store(
                crate::capsule::current_scope().is_some(),
                std::sync::atomic::Ordering::SeqCst,
            );
            true
        })
    }

    /// Review fix (#2351): a flushed invalidation runs on a spawned task, and
    /// still records into the request's capsule.
    #[cfg(feature = "reporting")]
    #[tokio::test]
    async fn a_flushed_invalidation_keeps_the_capture_scope() {
        let scope = std::sync::Arc::new(crate::capsule::CaptureScope::new(
            "coherence".to_owned(),
            std::sync::Arc::new(crate::capsule::CaptureSettings::default()),
            std::sync::Arc::new(crate::log::filter::ParameterFilter::default()),
        ));
        crate::capsule::capture::with_capture_scope(std::sync::Arc::clone(&scope), async {
            assert!(InvalidateAfterWrite::new(note_scope).run().await);
        })
        .await;
        assert!(SAW_SCOPE.load(std::sync::atomic::Ordering::SeqCst));
        assert!(
            !scope.is_truncated(),
            "an awaited invalidation is not detached"
        );
    }

    #[cfg(feature = "reporting")]
    fn never_ends() -> crate::cache::CacheFuture<'static, bool> {
        Box::pin(std::future::pending())
    }

    /// Codex review on #3222: a caller dropped while it waits for a flushed
    /// invalidation leaves the task detached, so the capsule is incomplete.
    #[cfg(feature = "reporting")]
    #[tokio::test]
    async fn a_flush_dropped_while_waiting_marks_the_capsule_incomplete() {
        let scope = std::sync::Arc::new(crate::capsule::CaptureScope::new(
            "coherence".to_owned(),
            std::sync::Arc::new(crate::capsule::CaptureSettings::default()),
            std::sync::Arc::new(crate::log::filter::ParameterFilter::default()),
        ));
        crate::capsule::capture::with_capture_scope(std::sync::Arc::clone(&scope), async {
            let mut guard = InvalidateAfterWrite::new(never_ends);
            let waited =
                tokio::time::timeout(std::time::Duration::from_millis(10), guard.flush()).await;
            assert!(waited.is_err(), "the invalidation never ends");
        })
        .await;
        assert!(scope.is_truncated(), "the invalidation runs detached");
    }

    /// Codex review on #3222: a replay that drops the flush while it waits
    /// logs the detached task now, before the verdict.
    #[cfg(feature = "reporting")]
    #[tokio::test]
    async fn a_replay_flush_dropped_while_waiting_is_a_divergence() {
        let tape = std::sync::Arc::new(crate::capsule::ReplayEffects::new(
            crate::capsule::CapsuleEffects::default(),
        ));
        crate::capsule::with_effect_tape(std::sync::Arc::clone(&tape), async {
            let mut guard = InvalidateAfterWrite::new(never_ends);
            let waited =
                tokio::time::timeout(std::time::Duration::from_millis(10), guard.flush()).await;
            assert!(waited.is_err(), "the invalidation never ends");
        })
        .await;
        let divergences = tape.divergences();
        assert_eq!(divergences.len(), 1, "{divergences:?}");
        assert_eq!(
            divergences[0].seam,
            crate::capsule::effects::EffectSeam::Detached
        );
    }

    #[tokio::test]
    async fn the_after_write_guard_invalidates_once_per_commit() {
        // One test owns GUARD_RUNS, so the counts are exact.
        let before = guard_runs();

        // A hook flush after the commit, then the end: one invalidation.
        let mut guard = InvalidateAfterWrite::new(count_guard_run);
        guard.committed();
        assert!(guard.flush().await);
        assert!(guard.flush().await, "a second flush has nothing to do");
        assert!(guard.run().await);
        assert_eq!(guard_runs(), before + 1);

        // A commit after the flush: the end invalidates again.
        let mut guard = InvalidateAfterWrite::new(count_guard_run);
        assert!(guard.flush().await);
        guard.committed();
        assert!(guard.run().await);
        assert_eq!(guard_runs(), before + 3);

        // No commit seen and no flush: the end still invalidates.
        assert!(InvalidateAfterWrite::new(count_guard_run).run().await);
        assert_eq!(guard_runs(), before + 4);

        // Disarmed: nothing runs, also on drop.
        InvalidateAfterWrite::new(count_guard_run).disarm();
        tokio::task::yield_now().await;
        assert_eq!(guard_runs(), before + 4);

        // Dropped while pending (a panic or a cancelled request): spawned.
        drop(InvalidateAfterWrite::new(count_guard_run));
        for _ in 0..100 {
            if guard_runs() == before + 5 {
                break;
            }
            tokio::task::yield_now().await;
        }
        assert_eq!(guard_runs(), before + 5);

        // Dropped after the final flush: nothing more runs.
        let mut guard = InvalidateAfterWrite::new(count_guard_run);
        assert!(guard.flush().await);
        drop(guard);
        tokio::task::yield_now().await;
        assert_eq!(guard_runs(), before + 6);
    }
}