gunnar-server 3.0.0

//! **The metered appliance: verify, log, store, swap — and answer the front
//! door in nanoseconds.** Compiled only with the `metered` feature.
//!
//! ```text
//!   monetize ──Entitlement.Set(fact, sig)──▶ EntitlementSlot::set
//!                                              │ 1. verify_fact (monetize-embed's canonical form, monetize's public key)
//!                                              │ 1b. check_not_replayed                     (newer than the fact this box holds)
//!                                              │ 2. append one line to attestations.jsonl   (the durable, signed-tree-head'd record)
//!                                              │ 3. EntitlementStore::set                   (the control document, so a reboot loads it)
//!                                              │ 4. EntitlementCache::push                  (the arc-swap snapshot the front door reads)
//!                                              │ 5. caps["cache_bytes"] → Namespace::explode_budget_bytes
//!                                              ▼
//!   authorise_service ──PlanGate::verdict──▶ EntitlementCache::allows   (one atomic load, one map lookup)
//! ```
//!
//! The order in `set` is the correctness argument: a fact is logged before it
//! is enforced, so there is no window in which the box acts on a plan its own
//! log does not show; and nothing at all happens to a fact that does not
//! verify — not a log line, not a store write, not a swap. Step 1 refusing is
//! the whole security property, and it is `monetize_embed::signing::verify_fact`
//! over `monetize_embed::signing::fact_message`: gunnar writes no canonical
//! form of its own, so there is no second encoding for a forger to aim at.
//!
//! **Step 1b is there because step 1 is not enough.** A signature says the fact
//! is monetize's; it does not say WHEN. Until 2026-09-17 an `EntitlementFact`
//! was replayable for ever — capture a `Paid` fact, wait for the tenant to
//! lapse, push the captured bytes back, and every check above passed, because
//! they were all real. `paid_until` cannot tell a replayed `Paid` from a
//! legitimate `Suspended` at the same date; that is what a lapse IS. So the
//! fact now carries `issued_unix_ms`, signed, and
//! [`EntitlementSlot::check_not_replayed`] refuses one that is not newer than
//! the fact this box holds. It runs under the same lock as the write, and
//! before the log, so a refused replay leaves no trace of having been tried
//! anywhere but the refusal line.
//!
//! # Boot
//!
//! [`EntitlementSlot::open`] loads every stored fact and pushes it into the
//! cache **through the same verification**. A fact that was accepted under a
//! key this box no longer trusts, or a control document somebody edited, is
//! refused at boot and logged by namespace; that namespace is then Free until
//! monetize pushes a fact that verifies. Refusing to boot would be the outage
//! that ends the relationship (the metered plan); serving a tenant for free until the
//! next signed fact arrives is the cheap failure.
//!
//! # What is NOT here
//!
//! No clock arithmetic: the state is monetize's verdict, stored, and the
//! ladder moves when monetize pushes the next fact. No network: monetize
//! reaches this box, never the reverse, and `git push` never waits on anyone's
//! uptime. No payment code, ever (the metered plan).

use std::sync::{Arc, Mutex};

use gunnar_common::monetize_key::MonetizePubkey;
use gunnar_core::{Entitlement, EntitlementState, EntitlementStore};
use gunnar_proto::EntitlementFact as ProtoFact;
use monetize_embed::{Action, EntitlementCache, State, TenantId, Verdict};

use crate::attest::{AttestationLog, ENTITLEMENT_KEY};
use crate::stores::ServerStores;

/// **The explode budget of a namespace the front door is refusing: one byte,
/// and not zero.**
///
/// # `0` is UNLIMITED, at every layer, and that is the trap this names
///
/// `Namespace::explode_budget_bytes` is `Option<u64>` where `None` is "the
/// server's default" and `Some(0)` is "unlimited — a deliberate answer, not an
/// absent one" (its own doc). The engine agrees: `ExplodedCache::over_budget`
/// is `self.budget != 0 && ...` and `evict_to_budget` returns early on
/// `budget == 0`. So writing `Some(0)` for a suspended tenant would hand them
/// an UNBOUNDED cache — the exact inverse of the intent, silently, and the
/// symptom would be a suspended namespace growing on a full disk.
///
/// # The type cannot express "no cache", and this is said rather than faked
///
/// There is no value of `Option<u64>` that means zero bytes of cache, and
/// `1` does not become one by being small. What `1` actually buys, read out of
/// `exploded_cache.rs` rather than assumed:
///
/// * `segment_for(1)` = `(1/16).clamp(min(16 MiB, 0).max(64 KiB), 1 GiB)` =
///   **64 KiB**, so the lazy pool rotates every 64 KiB;
/// * `evict_to_budget` drops the cheapest segment while more than one exists,
///   so the pool converges on its single live segment — ~64 KiB per store, not
///   0;
/// * the **eager** pool is never evicted by either path, by design: it is the
///   commit-and-tree graph that `graph`, `reach` and a tag peel read, not a
///   cache. A suspended namespace keeps it.
///
/// So: a suspended tenant is held to ~64 KiB of lazily-exploded objects per
/// store plus their graph pool. That is the floor this appliance can express,
/// it is two to five orders of magnitude under a real budget, and calling it
/// "no cache" in a changelog would be the lie. Making a true zero expressible
/// means a new variant on the field — a schema change to a persisted control
/// document, which is not something a cache-reclaim change gets to do on its
/// way past.
const NO_CACHE_BUDGET_BYTES: u64 = 1;

/// What the metered server is started with: the key that signs facts and
/// where a tenant goes to pay.
#[derive(Clone, Debug)]
pub struct MeteredConfig {
    /// monetize-server's Ed25519 verifying key, 32 bytes. Either spelled
    /// inline on `--monetize-pubkey` as its 64 hex characters, or read from
    /// the file that flag names — which holds the raw 32 bytes, or their 64 hex
    /// characters, or `ed25519:<base64>`. See [`MeteredConfig::load`].
    ///
    /// `None` while the key is AWAITED: the medium says `fed`
    /// ([`MeteredConfig::awaiting_feed`]) and the installer has not fed it yet.
    /// The key then arrives LIVE ([`EntitlementSlot::set_key`]), never by a
    /// restart (owner 2026-09-25: avoid restarts).
    pub monetize_pubkey: Option<[u8; 32]>,
    /// The pay page, with `{tenant}` where the namespace goes:
    /// `https://gunnar.rs/billing/{tenant}`. A template without the
    /// placeholder gets `/{tenant}` appended, so the URL always names the
    /// tenant.
    pub billing_url: String,
    /// `--monetize-vital` (default `true`): a boot that refuses stored facts raises
    /// `entitlement.key_mismatch` and reports `maintenance`; `false` only logs.
    pub vital: bool,
}

impl MeteredConfig {
    /// Resolve `key` — inline, or the file it names — and pair it with
    /// `billing_url`.
    ///
    /// An inline key was already decoded when the flag parsed, so there is
    /// nothing to read and nothing that can fail here: that is the whole point
    /// of it on an appliance whose only writable channel is the cmdline.
    ///
    /// # `billing_url` is `Option`, and `None` is a REFUSAL
    ///
    /// It used to be `&str`, and the caller in `gunnar_server::dispatch`
    /// filled a missing `--billing-url` in with `https://gunnar.rs/billing` —
    /// Vetra's own page, hardcoded as the OPEN product's default. Anybody else
    /// building a metered gunnar and forgetting the flag sent their lapsed
    /// tenants to somebody else's payment page, and nothing anywhere said so:
    /// the box booted, the plans enforced correctly, and the only symptom was
    /// a URL at the end of a refusal that a tenant would read as their own
    /// supplier's.
    ///
    /// The alternative considered was tolerating the absence — refusals would
    /// simply carry no link. It was rejected because of what this URL is FOR.
    /// [`EntitlementSlot::open`] logs it at boot, every Grace warning ends in
    /// it, and `PlanGate`'s refusal is a sentence whose last clause is the way
    /// out; a metered appliance whose refusals dead-end is a support ticket
    /// per tenant, silently, forever. An operator who reads one line at boot
    /// and adds one flag is the cheap failure. So a `metered` build handed
    /// `--monetize-pubkey` and no `--billing-url` stops, in the same voice as
    /// the open build's refusal of `--monetize-pubkey` itself.
    pub fn load(key: &MonetizePubkey, billing_url: Option<&str>) -> anyhow::Result<Self> {
        let billing_url = billing_url.ok_or_else(|| {
            anyhow::anyhow!(
                "--monetize-pubkey was given, but --billing-url was not, and this build has no \
                 default for it: it would have to invent the page your lapsed tenants are sent \
                 to. Every refusal and every Grace warning ends in that link, so a metered \
                 appliance without one refuses tenants with no way out. Pass \
                 --billing-url <URL> (or GUNNAR_BILLING_URL), with `{{tenant}}` where the \
                 namespace goes."
            )
        })?;
        let monetize_pubkey = read_pubkey(key)?;
        Ok(MeteredConfig {
            monetize_pubkey: Some(monetize_pubkey),
            billing_url: billing_template(billing_url),
            vital: true,
        })
    }

    /// **A metered box whose key the installer feeds** (`--monetize-pubkey fed`
    /// with no fed key yet — the first boot, before the join). The slot opens
    /// KEYLESS: every namespace is Free and `Entitlement.Set` refuses every fact
    /// until the key arrives, and then it applies in place
    /// ([`EntitlementSlot::set_key`]). `--billing-url` is refused missing here
    /// exactly as in [`MeteredConfig::load`]: a key that arrives later would
    /// otherwise have no pay page to enforce with.
    pub fn awaiting_feed(billing_url: Option<&str>) -> anyhow::Result<Self> {
        let billing_url = billing_url.ok_or_else(|| {
            anyhow::anyhow!(
                "--monetize-pubkey {} (the installer feeds the key) was given, but --billing-url was not, and this \
                 build has no default for it: the fed key could never be enforced with a pay page. Pass \
                 --billing-url <URL> (or GUNNAR_BILLING_URL), with `{{tenant}}` where the namespace goes.",
                gunnar_common::monetize_key::FED
            )
        })?;
        Ok(MeteredConfig {
            monetize_pubkey: None,
            billing_url: billing_template(billing_url),
            vital: true,
        })
    }
}

/// The 32 key bytes `key` names: inline, or read out of the file it names.
/// Split out of [`MeteredConfig::load`] so the start-up can compare the
/// medium's key with the installer's fed one (`crate::feed`) before loading.
pub fn read_pubkey(key: &MonetizePubkey) -> anyhow::Result<[u8; 32]> {
    Ok(match key {
            MonetizePubkey::Fed => anyhow::bail!(
                "--monetize-pubkey {} names no key: the installer's door feeds it (gunnar-server resolves \
                 `fed` against <data>/pairing/joined.toml before it ever reads a key)",
                gunnar_common::monetize_key::FED
            ),
            MonetizePubkey::Literal(k) => *k,
            MonetizePubkey::Path(path) => {
                let bytes = std::fs::read(path).map_err(|err| {
                    anyhow::anyhow!("reading the monetize public key {}: {err}", path.display())
                })?;
                parse_pubkey(&bytes).ok_or_else(|| {
                    anyhow::anyhow!(
                        "{} is not an Ed25519 public key: expected 32 raw bytes, 64 hex \
                         characters, or `ed25519:<base64>` ({} byte(s) read)",
                        path.display(),
                        bytes.len()
                    )
                })?
            }
    })
}

/// `{tenant}` in the template, or appended: the refusal line must always end
/// in a URL that names the tenant, because that is the link a lapsed
/// customer clicks.
fn billing_template(url: &str) -> String {
    if url.contains("{tenant}") {
        url.to_owned()
    } else {
        format!("{}/{{tenant}}", url.trim_end_matches('/'))
    }
}

/// The three spellings a key file may use. Raw bytes first, because a file
/// that is exactly 32 bytes long is a key and not text.
fn parse_pubkey(bytes: &[u8]) -> Option<[u8; 32]> {
    if bytes.len() == 32 {
        return bytes.try_into().ok();
    }
    let text = std::str::from_utf8(bytes).ok()?.trim();
    if let Some(b64) = text.strip_prefix("ed25519:") {
        return decode_base64(b64.trim())?.try_into().ok();
    }
    // The hex spelling is ONE decoder, shared with the inline `--monetize-pubkey`
    // arm in `gunnar_common::monetize_key` — a second copy here is a second
    // place for a byte order to drift.
    gunnar_common::monetize_key::parse_hex32(text).ok()
}

/// Standard base64 with or without padding. Written out rather than linked,
/// because a decoder for one key file is not worth a crate in PID 1.
fn decode_base64(text: &str) -> Option<Vec<u8>> {
    fn value(c: u8) -> Option<u32> {
        Some(match c {
            b'A'..=b'Z' => (c - b'A') as u32,
            b'a'..=b'z' => (c - b'a') as u32 + 26,
            b'0'..=b'9' => (c - b'0') as u32 + 52,
            b'+' | b'-' => 62,
            b'/' | b'_' => 63,
            _ => return None,
        })
    }
    let mut out = Vec::with_capacity(text.len() * 3 / 4);
    let mut acc = 0u32;
    let mut bits = 0u32;
    for c in text.bytes().filter(|c| *c != b'=') {
        acc = (acc << 6) | value(c)?;
        bits += 6;
        if bits >= 8 {
            bits -= 8;
            out.push((acc >> bits) as u8);
            acc &= (1 << bits) - 1;
        }
    }
    Some(out)
}

/// What [`EntitlementSlot::restate`] did: facts applied, and facts this box
/// already held as new or newer.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct Restated {
    pub applied: usize,
    pub held: usize,
}

/// Why a fact was not accepted. Every variant names its reason, and the
/// gRPC layer maps the first to `PERMISSION_DENIED` and the rest to
/// `INVALID_ARGUMENT` / `FAILED_PRECONDITION` — a forged fact is told apart
/// from a malformed one and from a box that cannot store one.
#[derive(Debug, thiserror::Error)]
pub enum SetRefused {
    /// The signature does not verify under the configured monetize key.
    /// **Nothing was logged, stored or swapped.**
    #[error("the fact for {namespace} does not verify under the configured monetize key ({why}); refused, nothing moved")]
    Forged { namespace: String, why: String },
    /// The fact's shape is wrong (a state gunnar cannot name).
    #[error("invalid fact: {0}")]
    Malformed(String),
    /// The fact verified but could not be logged or stored. It is NOT
    /// enforced either: a plan the log does not show is not a plan.
    #[error("the fact for {namespace} verified but could not be recorded: {why}")]
    NotRecorded { namespace: String, why: String },
    /// **A fact this box has already moved past.** The signature is monetize's
    /// own and it verifies; the fact is simply not newer than the one stored
    /// for this namespace, so accepting it would let a captured `Paid` fact
    /// restore a tenant who has since lapsed. **Nothing was logged, stored or
    /// swapped.**
    ///
    /// THE WORDS SAY NOTHING THE CALLER DID NOT BRING. Not the issue time this
    /// box holds, not the state, not whether the namespace is stamped at all —
    /// the numbers go to gunnar's own log and no further. `Entitlement.Set` is
    /// server-admin and `Entitlement.Get` is only server-read, so a caller who
    /// can reach this can usually read that anyway; the refusal is kept blind
    /// so that the day those grants are split, this does not quietly become the
    /// way to read one tenant's billing state through another's push.
    #[error("the fact for {namespace} is not newer than the one this server already holds for it; refused as a replay, nothing moved")]
    Replayed { namespace: String },
    /// **The fact carries no issue time** (stinky row 12): nothing tells it
    /// from a replay of any fact ever signed for the namespace, so it is
    /// refused whatever the box holds.
    #[error("the fact for {namespace} carries no issue time (issued_unix_ms); an unstamped fact cannot be told from a replay and is refused, nothing moved")]
    Unstamped { namespace: String },
    /// **No monetize key yet**: the medium said `fed` and the installer has
    /// not fed it. Nothing can be verified, so nothing is taken — once the key
    /// arrives ([`EntitlementSlot::set_key`]) the same push is accepted with no
    /// restart.
    #[error("the fact for {namespace} cannot be checked: this box has no monetize key yet (the installer feeds it); refused, nothing moved")]
    NoKey { namespace: String },
    /// **c-stale-suspend**: a newer-stamped fact that takes back paid time the
    /// box holds, from another payment — a monetize restored from an OLDER
    /// backup lost the renewal this box holds. Worded "already holds" so
    /// monetize's re-push files it as held, not as a refused cap.
    #[error("the fact for {namespace} is behind this box: it already holds a verdict paid until {held_until} by `{held_source}`, and this one pays only until {offered_until} by `{offered_source}` — a monetize restored from an older backup; refused, nothing moved")]
    Behind {
        namespace: String,
        held_until: u64,
        held_source: String,
        offered_until: u64,
        offered_source: String,
    },
}

/// The counts the console band and `Admin.Stats` show — the band's own type,
/// so the two surfaces cannot disagree about what is counted.
pub use crate::dashboard::TenantCounts;

/// What `EntitlementSlot::open` refused, kept for the wire.
#[derive(Debug, Clone, Default)]
struct Refusal {
    namespaces: Vec<String>,
    at_unix_ms: i64,
    reason: String,
}

fn now_unix_ms() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as i64)
        .unwrap_or(0)
}

/// The `Admin.Health` alarm a vital box raises while its stored facts do not verify.
pub const KEY_MISMATCH_ALARM: &str = "entitlement.key_mismatch";

/// **The verified cache, swappable** — `None` while the monetize key is awaited
/// (the medium said `fed`, the installer has not fed it yet). The key arrives
/// LIVE ([`EntitlementSlot::set_key`]): a new cache is built under it, loaded
/// with every stored fact, and swapped in, so no restart is needed. A read is
/// an `Arc` clone under an uncontended read lock; the cache's own snapshot
/// swap underneath is unchanged.
///
/// Keyless, it answers what a box with no slot answered: every namespace Free
/// (`allows` is Allow, no notice, no fact).
struct Cache(std::sync::RwLock<Option<Arc<EntitlementCache>>>);

impl Cache {
    fn current(&self) -> Option<Arc<EntitlementCache>> {
        self.0.read().unwrap_or_else(|p| p.into_inner()).clone()
    }
    fn swap_in(&self, cache: Arc<EntitlementCache>) {
        *self.0.write().unwrap_or_else(|p| p.into_inner()) = Some(cache);
    }
    fn len(&self) -> usize {
        self.current().map_or(0, |c| c.len())
    }
    fn fact(&self, tenant: &TenantId) -> Option<monetize_embed::EntitlementFact> {
        self.current()?.fact(tenant)
    }
    fn allows(&self, tenant: &TenantId, action: Action) -> Verdict {
        match self.current() {
            Some(c) => c.allows(tenant, action),
            None => Verdict::Allow,
        }
    }
    fn notice(&self, tenant: &TenantId) -> Option<String> {
        self.current()?.notice(tenant)
    }
}

/// The live entitlement state of this server, shared by the control plane
/// (which writes it) and the front doors (which read it).
pub struct EntitlementSlot {
    cache: Cache,
    store: EntitlementStore,
    log: AttestationLog,
    stores: Arc<ServerStores>,
    /// The pay page template: the medium's `--billing-url` until the installer's door feeds
    /// the SITE's own ([`EntitlementSlot::set_billing_url`], UW-6), then that one.
    billing_url: std::sync::RwLock<String>,
    /// The stored facts that did NOT verify when this slot was opened: which
    /// namespaces, when, and why. Set once, at boot, and never moved: a fact
    /// that verifies later arrives through `set`. See
    /// [`EntitlementSlot::refused_at_boot`] and [`EntitlementSlot::refusal`].
    refused: std::sync::RwLock<Refusal>,
    /// Serialises `set`: count-and-append on the log must not interleave with
    /// another `set`, and neither must store-then-swap. Never held across a
    /// read — the front door reads the cache, which is lock-free.
    write: Mutex<()>,
    /// `--monetize-vital`: a boot refusal is an alarm and `maintenance` (true) or a log line (false).
    vital: bool,
    /// A fact has VERIFIED since this boot (`set` accepted one). While the boot refused facts and
    /// this is false, a vital box is in maintenance; the first verifying fact clears it, because it
    /// proves monetize and this box now agree on the key.
    verified_since_boot: std::sync::atomic::AtomicBool,
    /// **This estate SELLS room** (the feed's `sells_room`; gunnar-grpc `selling`): a person's
    /// own account with no fact may push nothing here until its fact lands. Shared with the
    /// gRPC context ([`EntitlementSlot::sells_room_flag`]).
    sells_room: Arc<std::sync::atomic::AtomicBool>,
}

impl std::fmt::Debug for EntitlementSlot {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("EntitlementSlot")
            .field("tenants", &self.cache.len())
            .field("log", &self.log.path())
            .finish()
    }
}

impl EntitlementSlot {
    /// Open the slot over the server's control store and attestation log,
    /// loading — and re-verifying — every stored fact. See the module docs
    /// for what a fact that no longer verifies does at boot.
    pub fn open(stores: Arc<ServerStores>, config: &MeteredConfig) -> anyhow::Result<Arc<Self>> {
        let store = EntitlementStore::new(Arc::clone(stores.control_store()))?;
        let log = AttestationLog::under(stores.root());
        let slot = EntitlementSlot {
            cache: Cache(std::sync::RwLock::new(None)),
            store,
            log,
            stores,
            billing_url: std::sync::RwLock::new(config.billing_url.clone()),
            refused: std::sync::RwLock::new(Refusal::default()),
            write: Mutex::new(()),
            vital: config.vital,
            verified_since_boot: std::sync::atomic::AtomicBool::new(false),
            sells_room: Arc::new(std::sync::atomic::AtomicBool::new(false)),
        };
        match &config.monetize_pubkey {
            Some(key) => slot.load_under(key)?,
            None => tracing::warn!(
                target: "gunnar::entitlement",
                event = "entitlement.awaiting_key",
                billing = %config.billing_url,
                "metered: the monetize key is FED by the installer and has not arrived — every namespace is Free and \
                 Entitlement.Set refuses every fact until it does; it then applies in place, with no restart"
            ),
        }
        Ok(Arc::new(slot))
    }

    /// **Build the cache under `key`, load — and re-verify — every stored fact,
    /// and swap it in.** What `open` does at boot with a key, and what
    /// [`EntitlementSlot::set_key`] does when the key arrives while running.
    fn load_under(&self, key: &[u8; 32]) -> anyhow::Result<()> {
        let cache = Arc::new(
            EntitlementCache::new(key, &self.billing_url()).map_err(|_| {
                anyhow::anyhow!("the monetize public key is not a valid Ed25519 point")
            })?,
        );
        let mut loaded = 0usize;
        // ★ **THE REFUSED ONES ARE COUNTED, and the summary says so.**
        //
        // MEASURED 2026-09-21 (lane T3): this loop counted only the facts that
        // verified, and the summary below reported `tenants = loaded`. A box
        // that refused EVERY fact it held therefore logged `tenants = 0` and
        // booted normally — the same line a box with no tenants at all logs, and
        // the same line a healthy box logs on the day nobody bought anything.
        //
        // That is this estate's recurring disease in its most expensive place: a
        // health line that cannot fail. The shape it hides is the one that costs
        // money — a box RESTORED FROM ITS TWIN under a monetize key that is not
        // the one the facts were signed with keeps every fact on disk, verifies
        // none of them, and serves every paying tenant as Free. Nothing on the
        // box is wrong; the only evidence was one `error!` line per namespace in
        // a journal nobody reads on the day of a rescue.
        //
        // The count does not change what happens to a refused fact — it is still
        // kept, and the namespace is still Free until a fact that verifies
        // arrives. **Whether a box that refused facts should REFUSE TO START is
        // a ruling for the owner, not a change to smuggle in under a log line**,
        // and the line below is what makes that ruling possible to make: until
        // now there was no number to rule about.
        let mut refused = 0usize;
        let mut refusal = Refusal {
            at_unix_ms: now_unix_ms(),
            ..Refusal::default()
        };
        for stored in self.store.all()? {
            let namespace = stored.namespace.clone();
            match cache.push(to_monetize(&stored)) {
                Ok(()) => loaded += 1,
                Err(err) => {
                    refused += 1;
                    refusal.namespaces.push(namespace.clone());
                    if refusal.reason.is_empty() {
                        refusal.reason = format!(
                            "does not verify under the configured monetize public key ({err}); the likeliest cause after a restore is the wrong --monetize-pubkey"
                        );
                    }
                    tracing::error!(
                        target: "gunnar::entitlement",
                        event = "entitlement.boot.refused",
                        %namespace,
                        state = %stored.state,
                        error = %err,
                        "a stored entitlement fact does not verify under the configured monetize key; \
                         the namespace is Free until a fact that verifies arrives"
                    )
                }
            }
        }
        if refused > 0 {
            // A second line, at WARN, that names the sum rather than the parts:
            // one line per namespace is a thing to scroll past, and `refused=40`
            // beside `tenants=0` is a thing to stop at.
            tracing::warn!(
                target: "gunnar::entitlement",
                event = "entitlement.boot.mismatch",
                tenants = loaded,
                refused,
                billing = %self.billing_url(),
                "THIS BOX IS SERVING {refused} NAMESPACE(S) WITH A SIGNED FACT AS FREE: their facts are on disk and \
                 not one of them verifies under the monetize public key this box was given. Nothing on \
                 the box is broken — the KEY is the wrong one. After a restore from a twin that is the \
                 likeliest cause, and the fix is the right --monetize-pubkey, never a re-push"
            );
        }
        tracing::info!(
            target: "gunnar::entitlement",
            event = "entitlement.boot",
            tenants = loaded,
            refused,
            billing = %self.billing_url(),
            "metered: entitlement facts loaded"
        );
        *self.refused.write().unwrap_or_else(|p| p.into_inner()) = refusal;
        self.cache.swap_in(cache);
        Ok(())
    }

    /// **The monetize key, arriving while this box runs** (owner 2026-09-25:
    /// avoid restarts). The installer feeds `monetize_pubkey` (or vm1 offers
    /// it) after this slot opened keyless; the key is applied IN PLACE — a
    /// cache under it, every stored fact re-verified, swapped in — and from
    /// the next request on the front doors and `Entitlement.Set` enforce it.
    ///
    /// `Ok(true)` applied, `Ok(false)` already in force. A DIFFERENT key while
    /// one is in force is refused by name ([`crate::feed::FEED_MISMATCH`]):
    /// which monetize this box bills for is not changed silently mid-life.
    pub fn set_key(&self, key: &[u8; 32]) -> Result<bool, String> {
        let _serial = self.write.lock().unwrap_or_else(|e| e.into_inner());
        if let Some(current) = self.cache.current() {
            return if current.public_key().as_bytes() == key {
                Ok(false)
            } else {
                Err(format!(
                    "REFUSED [{}] this box already enforces monetize key {} and was handed {}; it will not switch which monetize it bills for while running",
                    crate::feed::FEED_MISMATCH,
                    hex32(current.public_key().as_bytes()),
                    hex32(key)
                ))
            };
        }
        self.load_under(key).map_err(|e| format!("{e:#}"))?;
        tracing::info!(target: "gunnar::entitlement", event = "entitlement.key_applied", key = %hex32(key), "metered: the fed monetize key is IN FORCE — applied in place, no restart");
        Ok(true)
    }

    /// **`restate_entitlements`** (5.4): the ONE monetize's snapshot, after a
    /// recharge brought this box back without its facts. Verified WHOLE first —
    /// the envelope and every fact under this box's monetize key — so a forgery
    /// anywhere moves nothing; then each fact through [`EntitlementSlot::set`]:
    /// the same fact or one this box has moved past (`Replayed`, `Behind`) is
    /// counted as held, never re-logged. Nothing is deleted. Without a key only an
    /// empty snapshot is answered (there is nothing in it to verify).
    pub fn restate(&self, snap: &monetize_embed::signing::Snapshot) -> Result<Restated, String> {
        let Some(cache) = self.cache.current() else {
            return if snap.facts.is_empty() {
                Ok(Restated::default())
            } else {
                Err(format!("this box enforces no monetize key yet, so the snapshot's {} entitlement facts cannot be verified; nothing restated", snap.facts.len()))
            };
        };
        monetize_embed::signing::verify_snapshot(snap, cache.public_key())
            .map_err(|e| format!("the snapshot does not verify under this box's monetize key ({e}); refused whole, nothing restated"))?;
        let mut done = Restated::default();
        for fact in &snap.facts {
            let held = self.store.get(&fact.tenant.0).ok().flatten();
            if held.is_some_and(|h| {
                h.signature == fact.signature && h.issued_signature == fact.issued_signature
            }) {
                done.held += 1;
                continue;
            }
            match self.set(monetize_to_proto(fact)) {
                Ok(_) => done.applied += 1,
                Err(SetRefused::Replayed { .. } | SetRefused::Behind { .. }) => done.held += 1,
                Err(e) => {
                    return Err(format!(
                        "restated {} entitlement facts, then stopped: {e}",
                        done.applied
                    ))
                }
            }
        }
        tracing::info!(target: "gunnar::entitlement", event = "entitlement.restated", applied = done.applied, held = done.held, snapshot_issued_unix_ms = snap.issued_unix_ms, "monetize's snapshot restated");
        Ok(done)
    }

    /// Whether a monetize key is in force (a fed box may not have one yet).
    pub fn keyed(&self) -> bool {
        self.cache.current().is_some()
    }

    /// [`EntitlementSlot::set`] for monetize's own fact (the pulled `entitlement` order).
    pub fn push(
        &self,
        fact: &monetize_embed::EntitlementFact,
    ) -> Result<(EntitlementState, u64), SetRefused> {
        self.set(monetize_to_proto(fact))
    }

    /// **Verify, log, store, swap.** See the module docs for the order and
    /// why. Returns the state now enforced and the attestation sequence
    /// number of the line the fact was written as.
    pub fn set(&self, fact: ProtoFact) -> Result<(EntitlementState, u64), SetRefused> {
        let namespace = fact.namespace.clone();
        let state = state_from_proto(fact.state()).ok_or_else(|| {
            SetRefused::Malformed(format!(
                "state {} is not one of Free, Paid, Grace, Suspended, Retention",
                fact.state
            ))
        })?;
        let monetize = monetize_embed::EntitlementFact {
            tenant: TenantId(fact.namespace.clone()),
            plan: fact.plan.clone(),
            state: to_monetize_state(state),
            paid_until_unix_ms: fact.paid_until_unix_ms,
            caps: fact.caps.clone(),
            source: fact.source.clone(),
            signature: fact.signature.clone(),
            issued_unix_ms: fact.issued_unix_ms,
            issued_signature: fact.issued_signature.clone(),
        };
        // 1. Verify — with monetize-embed's verifier over monetize-embed's
        //    canonical form. Refused here means NOTHING below runs.
        let cache = self.cache.current().ok_or_else(|| SetRefused::NoKey {
            namespace: namespace.clone(),
        })?;
        monetize_embed::signing::verify_fact(&monetize, cache.public_key()).map_err(|err| {
            SetRefused::Forged {
                namespace: namespace.clone(),
                why: err.to_string(),
            }
        })?;
        let accepted_at_unix_ms = gunnar_core::now_ms();
        let stored = Entitlement {
            namespace: namespace.clone(),
            plan: fact.plan,
            state,
            paid_until_unix_ms: fact.paid_until_unix_ms,
            caps: fact.caps,
            source: fact.source,
            signature: fact.signature,
            issued_unix_ms: fact.issued_unix_ms,
            issued_signature: fact.issued_signature,
            accepted_at_unix_ms,
        };
        let _serial = self.write.lock().unwrap_or_else(|e| e.into_inner());
        // 1b. **THE REPLAY GUARD**, under the same lock as the write it
        //     guards — read the held fact and decide before anything moves, or
        //     two concurrent `Set`s could both read the old high-water mark.
        //     Read from the STORE and not the cache: the store is what survives
        //     a reboot, and it still holds a fact the cache refused at boot.
        let held = self.store.get(&namespace).ok().flatten();
        self.check_not_replayed(&namespace, held.as_ref(), &stored)?;
        // 2. Log. The line carries monetize's own JSON of the fact (its serde
        //    form, `tenant` and all), so a reader with only monetize's
        //    vocabulary can re-verify the signature from the log alone.
        let line = serde_json::json!({
            ENTITLEMENT_KEY: monetize,
            "accepted_at_unix_ms": accepted_at_unix_ms,
        });
        let seq =
            self.log
                .append_line(&line.to_string())
                .map_err(|err| SetRefused::NotRecorded {
                    namespace: namespace.clone(),
                    why: format!("appending to {}: {err:#}", self.log.path().display()),
                })?;
        // 3. Store.
        self.store
            .set(stored.clone())
            .map_err(|err| SetRefused::NotRecorded {
                namespace: namespace.clone(),
                why: format!("writing the control document: {err}"),
            })?;
        // 4. Swap. `push` re-verifies; it cannot fail here because step 1
        //    passed the same bytes through the same function, and a failure
        //    would be a bug worth a loud line rather than a silent Free.
        if let Err(err) = cache.push(to_monetize(&stored)) {
            tracing::error!(
                target: "gunnar::entitlement",
                %namespace,
                error = %err,
                "a fact that verified in step 1 was refused by the cache in step 4"
            );
        }
        // 5. The one number a plan is allowed to move — and, at the bottom of
        //    the ladder, the derived bytes that number was buying. Still under
        //    `_serial`: the reclaim appends its own line, and `append_line`
        //    counts the file to number it.
        self.enforce_cache_for_state(&namespace, state, stored.cache_cap());
        tracing::info!(
            target: "gunnar::entitlement",
            event = "entitlement.set",
            %namespace,
            state = %state,
            plan = %stored.plan,
            source = %stored.source,
            seq,
            "entitlement fact accepted"
        );
        if !self
            .verified_since_boot
            .swap(true, std::sync::atomic::Ordering::SeqCst)
            && self.vital
            && !self.refused().namespaces.is_empty()
        {
            tracing::warn!(
                target: "gunnar::entitlement",
                event = "entitlement.key_mismatch.cleared",
                %namespace,
                "a fact VERIFIES under this box's monetize key: the key mismatch at boot is over, the alarm and the maintenance status clear"
            );
        }
        Ok((state, seq))
    }

    /// **Is this fact newer than the one this box already holds?**
    ///
    /// Everything in a fact except the issue time can be replayed: the six
    /// signed fields of a `Paid` fact from April are still a genuine, genuinely
    /// signed `Paid` fact in September, and `paid_until` cannot tell the two
    /// apart because a lapse keeps the date and moves the ladder. So the one
    /// comparison that can is against `issued_unix_ms`, and it is made HERE,
    /// on the box being lied to, before the fact is logged, stored or swapped.
    ///
    /// # The rules, and the one that is not obvious
    ///
    /// * **The offered fact is unstamped** — refused, ALWAYS (stinky row 12,
    ///   2026-10-10). It used to be accepted for a namespace with no stamped
    ///   fact yet — a grandfather clause for facts signed before the field
    ///   existed — which meant a captured old `Paid` fact could be replayed onto
    ///   every namespace that had not had a stamped one yet, and stripping
    ///   fields 8 and 9 off any captured fact made it the old form again.
    ///   monetize stamps every fact it signs, and no box holds facts from
    ///   before the stamp (monetize is a test system), so nothing legitimate
    ///   is unstamped.
    /// * **The namespace has no stamped fact** — nothing to compare a stamped
    ///   fact to, so it is accepted; the guard ARMS on it and never disarms.
    /// * **Older** — refused. That is the replay.
    /// * **Newer** — accepted.
    /// * **THE SAME MILLISECOND, and the same fact** — accepted, and this is the
    ///   rule that has to be right. monetize's poller re-pushes a fact the
    ///   product never acknowledged (`retry_pushes`), verbatim, which is how a
    ///   lost push is repaired at all. Those bytes carry the stamp they were
    ///   signed with, so a re-push is never "newer" — and refusing it would
    ///   break the one path whose job is to fix a push that went missing. It is
    ///   the SAME fact, so re-applying it changes nothing: idempotent by being
    ///   identical, not by being excused.
    /// * **The same millisecond, a DIFFERENT fact** — refused. Two verdicts for
    ///   one tenant inside one millisecond is not something monetize's poller
    ///   produces (one ladder move per tenant per tick), and allowing it would
    ///   leave exactly the hole this guard exists to close: an attacker holding
    ///   a `Paid` fact minted in the same millisecond as the `Suspended` one
    ///   could still push it. The cost is a refusal an operator can see and
    ///   answer, against a replay nobody would ever see.
    ///
    /// # What this does NOT do
    ///
    /// It does not refuse a fact stamped in the future. It cannot: monetize
    /// holds the key, so only monetize can stamp one, and refusing it would
    /// withdraw service from a paying customer over a clock. It is logged
    /// loudly instead — see the `entitlement.issued.ahead` line — because a
    /// stamp far in the future DOES wedge the namespace: no later fact can beat
    /// it, and the remedy is monetize's clock, not this box.
    fn check_not_replayed(
        &self,
        namespace: &str,
        held: Option<&Entitlement>,
        offered: &Entitlement,
    ) -> Result<(), SetRefused> {
        let refuse = || {
            Err(SetRefused::Replayed {
                namespace: namespace.to_owned(),
            })
        };
        // c-stale-suspend: paid time the box holds, taken back by a verdict
        // about ANOTHER payment, while that paid time is still running.
        if let (Some(h), Some(offered_until)) = (held, offered.paid_until_unix_ms) {
            let now = gunnar_core::now_ms().max(0) as u64;
            if let Some(held_until) = h
                .paid_until_unix_ms
                .filter(|u| *u > now && offered_until < *u)
            {
                if h.state == EntitlementState::Paid && offered.source != h.source {
                    tracing::warn!(target: "gunnar::entitlement", event = "entitlement.behind", %namespace, held_until, offered_until, held_source = %h.source, offered_source = %offered.source, "a fact behind this box (a restored monetize lost a payment it holds); refused");
                    return Err(SetRefused::Behind {
                        namespace: namespace.to_owned(),
                        held_until,
                        held_source: h.source.clone(),
                        offered_until,
                        offered_source: offered.source.clone(),
                    });
                }
            }
        }
        let Some(issued) = offered.issued_unix_ms else {
            tracing::warn!(
                target: "gunnar::entitlement",
                event = "entitlement.replay_guard.refused",
                %namespace,
                why = "unstamped",
                held_issued_unix_ms = held.and_then(|h| h.issued_unix_ms),
                "a fact with no issue time; refused — nothing tells it from a replay"
            );
            return Err(SetRefused::Unstamped {
                namespace: namespace.to_owned(),
            });
        };
        let Some(high_water) = held.and_then(|h| h.issued_unix_ms) else {
            // Unlatched: the first stamped fact this namespace has had.
            tracing::info!(
                target: "gunnar::entitlement",
                event = "entitlement.replay_guard.armed",
                %namespace,
                issued_unix_ms = issued,
                "the first stamped fact for this namespace: replays of anything older are \
                 refused from here on"
            );
            return Ok(());
        };
        if issued < high_water || (issued == high_water && !same_fact(held, offered)) {
            tracing::warn!(
                target: "gunnar::entitlement",
                event = "entitlement.replay_guard.refused",
                %namespace,
                why = if issued < high_water { "older" } else { "same instant, different fact" },
                offered_issued_unix_ms = issued,
                held_issued_unix_ms = high_water,
                state = %offered.state,
                source = %offered.source,
                "a fact that is not newer than the one this server holds; refused as a replay"
            );
            return refuse();
        }
        let now = gunnar_core::now_ms();
        if now > 0 && issued > (now as u64).saturating_add(MAX_ISSUED_AHEAD_MS) {
            tracing::warn!(
                target: "gunnar::entitlement",
                event = "entitlement.issued.ahead",
                %namespace,
                issued_unix_ms = issued,
                now_unix_ms = now,
                "ACCEPTED a fact issued far ahead of this server's clock — service is never \
                 withdrawn over a clock — but no later fact can beat this stamp, so the \
                 namespace is now pinned to it. Check monetize's clock."
            );
        }
        Ok(())
    }

    /// **The cache half of a plan change: what the new state is allowed to
    /// hold, and — at the bottom of the ladder — giving back what it holds
    /// now.**
    ///
    /// Suspended and Retention are the two states [`PlanGate::verdict`] already
    /// refuses at the door. A tenant the door refuses is holding space nobody
    /// can sell: the packs stay (a plan lapsing is NOT a deletion, and nothing
    /// here touches one), but the *derived* read cache is regenerable from
    /// those packs and is therefore the one thing that can be given back at no
    /// risk at all. So it is dropped, and the budget is put where a new one
    /// cannot grow.
    ///
    /// **Grace does not reclaim, and that is the point of Grace.** A tenant in
    /// the 14-day window is expected to pay and come back; pulling their
    /// exploded objects would make the first push after they paid re-inflate
    /// every delta chain it touches (~0.5–2 ms an object) — a punishment for a
    /// customer who just paid, bought with nothing, because Grace is short and
    /// the space comes back anyway if they do not.
    ///
    /// Every other state applies `caps["cache_bytes"]` exactly as before.
    fn enforce_cache_for_state(
        &self,
        namespace: &str,
        state: EntitlementState,
        cache_cap: Option<u64>,
    ) {
        match state {
            EntitlementState::Suspended | EntitlementState::Retention => {
                self.write_explode_budget(namespace, Some(NO_CACHE_BUDGET_BYTES));
                self.reclaim_explode_caches(namespace, state);
            }
            // Free, Paid and Grace: the plan's own number, never a default
            // invented here.
            EntitlementState::Free | EntitlementState::Paid | EntitlementState::Grace => {
                match cache_cap {
                    Some(cache_bytes) => self.apply_cache_cap(namespace, cache_bytes),
                    // **The way back up, for a plan that names no cache cap.**
                    //
                    // Without this arm a tenant who was suspended and then paid
                    // stayed on `NO_CACHE_BUDGET_BYTES` for ever: the old code
                    // only ever wrote the row when the fact carried a
                    // `cache_bytes` cap, so a plan without one left our own
                    // sentinel in place and the returning customer read at
                    // uncached speed with no line anywhere saying why.
                    //
                    // Only OUR sentinel is undone, and only back to `None` =
                    // the server's default. An operator's hand-set quota
                    // (`gunnar namespace set-explode-budget`) is not a thing a
                    // plan change is entitled to clear, so any other value is
                    // left exactly as found.
                    None => {
                        if matches!(
                            self.stores.catalog().namespace_explode_budget(namespace),
                            Ok(Some(NO_CACHE_BUDGET_BYTES))
                        ) {
                            self.write_explode_budget(namespace, None);
                            self.reopen_on_server_default(namespace);
                        }
                    }
                }
            }
        }
    }

    /// **Let the namespace's open stores fall back to the server's default
    /// budget, by forgetting their handles.**
    ///
    /// Needed only on the way back up out of [`NO_CACHE_BUDGET_BYTES`], and
    /// only because `None` on the catalog row is an ABSENCE rather than a
    /// number: `ServerStores::apply_explode_quota` skips a `None` row
    /// (`Ok(None) => {}`), which is right for its usual caller and wrong here —
    /// a store that was open through the suspension would otherwise keep the
    /// 1-byte budget in its own atomic for as long as the handle cache held it,
    /// and a tenant who had just paid would read at uncached speed with the
    /// catalog row saying they were on the default.
    ///
    /// The server's default is computed from the filesystem inside the engine
    /// (`exploded_cache::budget_from_env`) and there is no "put it back"
    /// number to write, so the honest move is to let the store be opened
    /// again: `close_repo` releases the handle, the next request re-opens on
    /// the default. The cost is one refold per repository, once, at the moment
    /// a customer returns — paid by the side that just started paying again,
    /// and only for the repositories whose handles we ourselves had capped.
    fn reopen_on_server_default(&self, namespace: &str) {
        for repo in self
            .stores
            .catalog()
            .list(Some(namespace))
            .unwrap_or_default()
        {
            self.stores.close_repo(&repo.path());
        }
    }

    /// **Drop the derived read cache of every repository in `namespace`, and
    /// journal what came back.**
    ///
    /// Safe by construction, and each leg of that was checked rather than
    /// assumed:
    ///
    /// * the exploded cache is derived — every row in it is re-inflatable from
    ///   the packs, which this does not touch;
    /// * `drop_cache` unlinks the *lazy* pool only. The eager pool stays: it is
    ///   the commit-and-tree graph that `graph`, `reach` and a tag peel read,
    ///   not a cache;
    /// * a twin never held it. `twin::under_root` refuses any framed path
    ///   `txlog::never_framed` names — "a cache or per-box path — never applied
    ///   on a twin" — so there is no mirror anywhere that this can desynchronise.
    ///
    /// # Only the stores that are OPEN, and why that is the right line
    ///
    /// `Engine::drop_explode_cache` reaches a store through the live handle
    /// cache and answers 0 for a closed one; it will not open a store to empty
    /// it. Taken deliberately, not worked around:
    ///
    /// * an open store is where the *process's* share of this is — the
    ///   descriptors, the redb page caches, the segment being written into —
    ///   and it is the only store that can still be GROWING a cache right now.
    ///   That growth is what a suspension has to stop, and this stops it;
    /// * a closed store's lazy segments stay on disk, and the sentinel budget
    ///   means they cannot survive the next open: `evict_to_budget` runs on the
    ///   first maintenance tick after the quota is applied at open
    ///   (`ServerStores::apply_explode_quota`, from the catalog row this just
    ///   wrote) and takes the pool down to its one live segment;
    /// * opening every repository of a suspended namespace to empty it is a
    ///   full store open (the refold cost) per repository, on the control-plane
    ///   thread, under `EntitlementSlot`'s write mutex, for a tenant the front
    ///   door is refusing. The reclaim would cost more at the moment of
    ///   suspension than the space is worth, and a monetize push that fans out
    ///   into hundreds of store opens is a stall an operator would read as an
    ///   outage.
    ///
    /// **So the honest limit is stated rather than hidden**: a suspended
    /// namespace whose repositories are all closed reclaims 0 bytes here, and
    /// the journal line says 0 rather than a number nobody gave back. That
    /// disk comes back on `gunnar repo gc` (`admin::gc`), which opens the store
    /// anyway and already drops the cache as part of its own work.
    fn reclaim_explode_caches(&self, namespace: &str, state: EntitlementState) {
        let engine = self.stores.engine();
        let mut freed = 0u64;
        let mut repos = 0usize;
        let mut reclaimed_repos = 0usize;
        for repo in self
            .stores
            .catalog()
            .list(Some(namespace))
            .unwrap_or_default()
        {
            let Ok(dir) = self.stores.repo_dir(&repo.path()) else {
                continue;
            };
            repos += 1;
            let bytes = engine.drop_explode_cache(&dir);
            if bytes > 0 {
                reclaimed_repos += 1;
                freed += bytes;
            }
        }
        // **Journalled even when it is 0.** The owner's question is "what did
        // suspending this tenant give back", and a line that only ever appears
        // when the answer is flattering is not an answer. A zero here is the
        // closed-store case above, on the record, by namespace.
        let line = serde_json::json!({
            crate::attest::RECLAIM_KEY: {
                "namespace": namespace,
                "state": state.to_string(),
                "bytes_freed": freed,
                "repos": repos,
                "repos_reclaimed": reclaimed_repos,
                "budget_bytes": NO_CACHE_BUDGET_BYTES,
            },
            "reclaimed_at_unix_ms": gunnar_core::now_ms(),
        });
        // The caller holds `self.write`; this must not take it again.
        if let Err(err) = self.log.append_line(&line.to_string()) {
            // Same rule as a failed push attestation (`attest`): the plan has
            // already moved and the cache is already gone, so returning an
            // error here could only report a refusal that did not happen.
            tracing::error!(
                target: "gunnar::entitlement",
                %namespace,
                error = %err,
                "the cache reclaim could not be journalled; the bytes are gone but the line is not there"
            );
        }
        tracing::info!(
            target: "gunnar::entitlement",
            event = "entitlement.cache_reclaimed",
            %namespace,
            state = %state,
            bytes = freed,
            repos,
            repos_reclaimed = reclaimed_repos,
            "the derived read cache of a namespace the door refuses was given back"
        );
    }

    /// `caps["cache_bytes"]` → `Namespace::explode_budget_bytes`, on the
    /// catalog row and on every store of the namespace that is open right now.
    /// `0` is "unlimited", which is `Some(0)` on the row — a deliberate answer,
    /// as the field's own doc says.
    fn apply_cache_cap(&self, namespace: &str, cache_bytes: u64) {
        self.write_explode_budget(namespace, Some(cache_bytes));
    }

    /// The one writer of `Namespace::explode_budget_bytes` on this path: the
    /// catalog row, then every store of the namespace that is open right now.
    /// `None` is "back to the server's default", which is what the field's
    /// absence means.
    fn write_explode_budget(&self, namespace: &str, budget: Option<u64>) {
        match self
            .stores
            .catalog()
            .set_namespace_explode_budget(namespace, budget)
        {
            // Applied to the stores open right now as well, exactly as
            // `gunnar namespace set-explode-budget` does (admin.rs).
            Ok(_) => {
                for repo in self
                    .stores
                    .catalog()
                    .list(Some(namespace))
                    .unwrap_or_default()
                {
                    if let Ok(dir) = self.stores.repo_dir(&repo.path()) {
                        self.stores.apply_explode_quota(namespace, &dir);
                    }
                }
            }
            // A namespace monetize knows and the catalog does not yet (a plan
            // bought before the first push): the cap is in the stored fact
            // and is applied when the namespace is created. Not an error.
            Err(gunnar_core::Error::NoSuchNamespace(_)) => {}
            Err(err) => tracing::warn!(
                target: "gunnar::entitlement",
                %namespace,
                error = %err,
                "caps[cache_bytes] could not be applied to the namespace"
            ),
        }
    }

    /// **The fact being ENFORCED for `namespace`**, or `None` = Free.
    ///
    /// ★ Off the CACHE, the snapshot `PlanGate` reads — not off the document.
    ///
    /// MEASURED 2026-09-21 (lane T3): this read the STORE, which keeps a fact the
    /// cache refused at boot. On a box restored under the wrong
    /// `--monetize-pubkey`, `Entitlement.Get` therefore answered `PAID` with a
    /// renewal date for a tenant the front door was serving as Free, and
    /// `Usage.Read` — which reports the plan from here — told the tenant the
    /// same. The seam said so in its own doc (`ControlPlane::entitlement_get`:
    /// "a rejected fact must not be reported as an allowance … must answer out
    /// of the same snapshot `PlanGate` reads") and nothing did it. Now every
    /// reader of a plan answers what the box DOES; the refused facts are still
    /// on disk, and [`EntitlementSlot::refused_at_boot`] is how they are seen.
    pub fn get(&self, namespace: &str) -> anyhow::Result<Option<ProtoFact>> {
        Ok(self
            .cache
            .fact(&TenantId(namespace.to_owned()))
            .as_ref()
            .map(monetize_to_proto))
    }

    /// Every ENFORCED fact, one per namespace, in namespace order. A namespace
    /// whose stored fact was refused at boot is absent — it is Free, which is
    /// what "no fact" means on this wire. See [`EntitlementSlot::get`].
    pub fn list(&self) -> anyhow::Result<Vec<ProtoFact>> {
        Ok(self
            .store
            .all()?
            .iter()
            .filter_map(|stored| self.cache.fact(&TenantId(stored.namespace.clone())))
            .map(|fact| monetize_to_proto(&fact))
            .collect())
    }

    /// How many stored facts did not verify under this box's monetize key when
    /// it booted. Non-zero is a box serving signed tenants as Free; after a
    /// restore the likeliest cause is the wrong `--monetize-pubkey`.
    fn refused(&self) -> Refusal {
        self.refused
            .read()
            .unwrap_or_else(|p| p.into_inner())
            .clone()
    }

    pub fn refused_at_boot(&self) -> usize {
        self.refused().namespaces.len()
    }

    /// The boot's refusal as the wire carries it (`Entitlement.List`), or `None`
    /// when nothing was refused.
    pub fn refusal(&self) -> Option<gunnar_proto::RefusedAtBoot> {
        (!self.refused().namespaces.is_empty()).then(|| gunnar_proto::RefusedAtBoot {
            count: self.refused().namespaces.len() as u64,
            at_unix_ms: self.refused().at_unix_ms,
            reason: self.refused().reason.clone(),
            namespaces: self.refused().namespaces.clone(),
        })
    }

    /// **Is this box in maintenance for its monetize key?** Owner ruling 2026-09-22: with
    /// `--monetize-vital` (the default), a boot that refused stored facts is `maintenance` until
    /// monetize pushes a fact that VERIFIES; the box starts and serves meanwhile. With
    /// `--monetize-vital false` it never is — the refusal is only logged.
    pub fn maintenance(&self) -> bool {
        self.vital
            && !self.refused().namespaces.is_empty()
            && !self
                .verified_since_boot
                .load(std::sync::atomic::Ordering::SeqCst)
    }

    /// The same, as an `Admin.Health` alarm, `entitlement.key_mismatch`, naming the namespaces:
    /// the WARN `entitlement.boot.mismatch` on the wire, for a machine whose console nobody reads.
    /// Only while [`EntitlementSlot::maintenance`] holds; `None` on a box that is not vital.
    pub fn alarm(&self) -> Option<gunnar_proto::HealthAlarm> {
        if !self.maintenance() {
            return None;
        }
        self.refusal().map(|r| gunnar_proto::HealthAlarm {
            event: KEY_MISMATCH_ALARM.into(),
            count: r.count,
            since_unix_ms: r.at_unix_ms,
            detail: format!(
                "MAINTENANCE: {} namespace(s) hold a signed entitlement this box does not verify, and are served as Free: {}: {}. \
                 It clears when monetize pushes a fact that verifies. The fix is the right --monetize-pubkey, never a re-push",
                r.count,
                r.namespaces.join(", "),
                r.reason
            ),
        })
    }

    /// **The state being ENFORCED for `namespace`**, off the cache: `None` is
    /// Free — including a stored fact that did not verify at boot. The band
    /// and `Admin.Stats` count from this and not from the document, so what
    /// the operator sees is what the front door does.
    pub fn enforced_state(&self, namespace: &str) -> Option<EntitlementState> {
        self.cache
            .fact(&TenantId(namespace.to_owned()))
            .map(|fact| from_monetize_state(fact.state))
    }

    /// The counts for the band and `Admin.Stats`: the namespaces the document
    /// names, each at the state the cache enforces.
    pub fn counts(&self) -> TenantCounts {
        let mut counts = TenantCounts::default();
        for fact in self.store.all().unwrap_or_default() {
            match self.enforced_state(&fact.namespace) {
                Some(EntitlementState::Paid) => counts.paid += 1,
                Some(EntitlementState::Grace) => counts.grace += 1,
                Some(EntitlementState::Suspended | EntitlementState::Retention) => {
                    counts.suspended += 1
                }
                Some(EntitlementState::Free) | None => {}
            }
        }
        counts
    }

    /// The pay page for `namespace`: the same template the cache renders
    /// into every refusal line, so the quota line and the plan line name one
    /// URL.
    fn url_for(&self, namespace: &str) -> String {
        self.billing_url().replace("{tenant}", namespace)
    }

    /// The pay page template in force.
    pub fn billing_url(&self) -> String {
        self.billing_url
            .read()
            .unwrap_or_else(|p| p.into_inner())
            .clone()
    }

    /// ★ UW-6: **the SITE's pay page, fed by the installer's door** (`refusal_link`), taken
    /// LIVE: the medium is one for every estate, so the page it bakes is a placeholder until
    /// the site says its own. `Ok(true)` when it changed (the cache is rebuilt under the key
    /// in force, so the next refusal and Grace line name it), `Ok(false)` for the page
    /// already in force. Only an `https://` URL is taken; any other is refused by name and
    /// the page in force stays.
    pub fn set_billing_url(&self, url: &str) -> Result<bool, String> {
        let url = url.trim();
        let host = url
            .strip_prefix("https://")
            .map(|r| r.split('/').next().unwrap_or_default())
            .unwrap_or_default();
        if host.is_empty() || url.chars().any(char::is_whitespace) {
            return Err(format!("REFUSED [pairing-feed-billing-url] the installer fed `{}` = `{url}`, which is not an https:// page; the pay page in force stays", crate::feed::FEED_BILLING_URL));
        }
        let url = billing_template(url);
        let _serial = self.write.lock().unwrap_or_else(|e| e.into_inner());
        if self.billing_url() == url {
            return Ok(false);
        }
        *self.billing_url.write().unwrap_or_else(|p| p.into_inner()) = url.clone();
        if let Some(current) = self.cache.current() {
            let key = *current.public_key().as_bytes();
            self.load_under(&key).map_err(|e| format!("the pay page {url} is set, but the cache did not rebuild under the key in force: {e:#}"))?;
        }
        tracing::info!(target: "gunnar::entitlement", event = "entitlement.billing_url", billing = %url, "the site's pay page is fed: every refusal and Grace line ends in it");
        Ok(true)
    }
}

/// The metered server's [`gunnar_ssh::PlanGate`]: the cache's verdict,
/// rendered as the front door's.
impl EntitlementSlot {
    /// **THE counting point for a `cap` refusal**: the plan said no, before a
    /// byte moved.
    ///
    /// Both doors this gate has come through here — the ladder
    /// (`PlanGate::verdict`, a Suspended or Retention tenant) and the pack
    /// quota (`PlanGate::pack_quota`, a tenant already at `caps["pack_bytes"]`)
    /// — because to a tenant they are one sentence: *your plan will not let you
    /// do this*. Which of the two it was is already in the refusal line the
    /// client reads and in the attestation log; the flow ledger counts how
    /// OFTEN, which is the number that says whether a customer is stuck.
    ///
    /// **The two MID-transfer cuts are deliberately not counted here**, and
    /// they are not missing. `gunnar_wire::receive_pack`'s pack ceiling and
    /// `gunnar_lfs`'s allowance both cut a transfer that is already running,
    /// and a cut transfer reaches the flow ledger as what it is: bytes in, and
    /// no push. A tenant whose pushes arrive and never complete is visible in
    /// exactly that shape, and counting the cut as a refusal as well would
    /// report one event twice under two names.
    fn charge_refusal(&self, namespace: &str) {
        self.stores
            .flow()
            .refuse(namespace, gunnar_core::now_ms(), gunnar_core::Refusal::Cap);
    }
}

impl gunnar_ssh::PlanGate for EntitlementSlot {
    fn verdict(&self, namespace: &str, action: gunnar_ssh::PlanAction) -> gunnar_ssh::PlanVerdict {
        let tenant = TenantId(namespace.to_owned());
        let action = match action {
            gunnar_ssh::PlanAction::Write => Action::Write,
            gunnar_ssh::PlanAction::AnonymousRead => Action::AnonymousRead,
        };
        match self.cache.allows(&tenant, action) {
            Verdict::Refuse { reason_line, .. } => {
                self.charge_refusal(namespace);
                gunnar_ssh::PlanVerdict::Refuse(reason_line)
            }
            Verdict::Allow => match self.cache.notice(&tenant) {
                Some(line) => gunnar_ssh::PlanVerdict::Warn(line),
                None => gunnar_ssh::PlanVerdict::Allow,
            },
        }
    }

    fn name(&self) -> &'static str {
        "entitlement"
    }

    fn pack_quota(&self, namespace: &str) -> gunnar_ssh::PlanVerdict {
        let Some(quota) = self.pack_quota_state(namespace) else {
            return gunnar_ssh::PlanVerdict::Allow;
        };
        if quota.plan == gunnar_grpc::selling::NO_ROOM_YET {
            self.charge_refusal(namespace);
            return gunnar_ssh::PlanVerdict::Refuse(format!(
                "{}: {}",
                gunnar_grpc::selling::NO_ROOM_YET,
                self.url_for(namespace)
            ));
        }
        if quota.held >= quota.cap {
            self.charge_refusal(namespace);
            gunnar_ssh::PlanVerdict::Refuse(format!(
                "plan {} is at its pack quota ({} of {} bytes held): writes closed; \
                 delete or buy more at {}",
                quota.plan,
                quota.held,
                quota.cap,
                self.url_for(namespace)
            ))
        } else {
            gunnar_ssh::PlanVerdict::Allow
        }
    }

    /// **The mid-upload ceiling for `caps["pack_bytes"]`.**
    ///
    /// [`Self::pack_quota`] above refuses a tenant who is ALREADY at the cap,
    /// before a byte moves. It cannot refuse the push that crosses the cap
    /// *during* the transfer, because a pack header carries an object count and
    /// not a byte count — so until this existed, a tenant one byte under a
    /// 10 GiB cap could send 400 GiB and the appliance would take all of it,
    /// spool it, and only then notice. This is the number the receive loop
    /// counts against; `gunnar_wire::receive_pack` cuts the stream past it.
    ///
    /// # The slack, and why it is not zero
    ///
    /// The ceiling is `what is left under the cap` **plus slack**, and the
    /// slack is deliberate rather than defensive rounding. Three reasons, each
    /// of which alone would make a zero-slack ceiling wrong:
    ///
    /// 1. **Wire bytes are not stored bytes.** What is counted here is what the
    ///    client sent; what the cap is measured in is
    ///    `control_plane::repo_bytes().pack`, which is what the engine wrote.
    ///    A thin push is completed with bases the client never sent (larger),
    ///    the engine stores the pack verbatim and compressed (smaller), and
    ///    neither direction is known before the push. A ceiling exactly equal
    ///    to the remaining room would therefore refuse pushes that would have
    ///    fitted, and it would do so *mid-transfer*, which is the worst moment
    ///    to be wrong.
    /// 2. **A push that lands exactly at the cap must land.** `pack_quota`
    ///    refuses at `held >= cap`, so `held + n == cap` is a legal push; a
    ///    hard ceiling with no slack cuts it at its last byte.
    /// 3. **This is a backstop, not a meter.** Billing is `Usage.Read`'s job
    ///    and the polite refusal is `pack_quota`'s, before the bytes. What is
    ///    left for this to prevent is the *unbounded* push — 400 GiB, not
    ///    400 MiB — so it is set generously on purpose. A tenant who overshoots
    ///    by a little is billed for it and refused on their NEXT push, with a
    ///    sentence they can read, rather than cut off mid-transfer.
    ///
    /// The number is the larger of a proportion and an absolute floor, which is
    /// the shape `gunnar_ssh::headroom::DiskFloor` already argues for one
    /// module away and for the same reason: a percentage alone is nothing on a
    /// small plan, and an absolute alone is nothing on a large one.
    ///
    /// * **`cap / 16`** (6.25 %) so a 1 TiB plan gets 64 GiB of room and a
    ///   10 GiB plan gets 640 MiB — proportionate to how wrong the wire-versus-
    ///   stored estimate can be, since that error scales with the push.
    /// * **`64 MiB`** as the floor, because on a small plan 6.25 % is smaller
    ///   than one ordinary push: at a 256 MiB cap it is 16 MiB, so a tenant
    ///   holding 250 MiB would be held to a 22 MiB ceiling and a 60 MiB
    ///   repository would be cut in flight by the arithmetic. With the floor
    ///   the ceiling is 70 MiB and that push lands — and is then refused,
    ///   politely and before its first byte, on the NEXT one.
    ///
    /// Neither number is measured — they cannot be, the quantity they cover is
    /// the difference between two things gunnar does not know before the push —
    /// so they are stated as a judgement with its reasons, and the test that
    /// holds them is `entitlement.rs`'s, which asserts the arithmetic rather
    /// than a magic total.
    fn pack_ceiling(&self, namespace: &str) -> Option<gunnar_ssh::PackCeiling> {
        let quota = self.pack_quota_state(namespace)?;
        let remaining = quota.cap.saturating_sub(quota.held);
        let slack = (quota.cap / PACK_CEILING_SLACK_DIVISOR).max(PACK_CEILING_SLACK_FLOOR_BYTES);
        let bytes = remaining.saturating_add(slack);
        Some(gunnar_ssh::PackCeiling {
            bytes,
            // The same three facts `pack_quota`'s refusal carries — the plan,
            // the cap, where to pay — because a tenant who meets one of these
            // sentences has usually just met the other, and two different
            // spellings of "you are out of room" is how a support ticket
            // becomes two.
            refusal: format!(
                "plan {} would pass its pack quota with this push (cap {} bytes, {} already \
                 held, this push sent more than {bytes} bytes): the transfer was stopped and \
                 nothing was stored; delete or buy more at {}",
                quota.plan,
                quota.cap,
                quota.held,
                self.url_for(namespace)
            ),
        })
    }

    /// The LFS twin of [`gunnar_ssh::PlanGate::pack_ceiling`]. See
    /// [`Self::lfs_allowance_state`] for the arithmetic and for why it is
    /// `caps["lfs_bytes"]` and not the room left under `caps["pack_bytes"]`.
    fn lfs_allowance(&self, namespace: &str) -> Option<gunnar_lfs::Allowance> {
        self.lfs_allowance_state(namespace)
    }
}

impl EntitlementSlot {
    /// **The tenant's remaining LFS room**, for `gunnar_lfs::Session` and the
    /// HTTP object endpoint.
    ///
    /// Off `caps["lfs_bytes"]` and the namespace's measured `lfs` bytes — the
    /// same two numbers `Usage.Read` reports as `allowance.lfs_bytes` and
    /// `usage.lfs_bytes`, so what the console shows, what the door enforces and
    /// what monetize bills are one arithmetic and not three (one implementation). That
    /// identity is the whole reason this reads `pack_quota_state`'s twin rather
    /// than re-walking the disk with its own rules.
    ///
    /// # Why this is `lfs_bytes` and NOT the room left under `pack_bytes`
    ///
    /// Tried the other way first, and it is wrong for a reason worth writing
    /// down, because it looks generous: LFS bytes land on the same filesystem
    /// the tenant bought, so bounding them by the pack allowance would close the
    /// hole with no new cap and no monetize change at all.
    ///
    /// It would also make `Usage.Read` a liar. That surface reports
    /// `allowance.lfs_bytes` ABSENT for a fact that names no `lfs_bytes` cap,
    /// and absent has one documented meaning — uncapped (`usage.proto`). A door
    /// that silently held LFS to the pack allowance would refuse a tenant at a
    /// number no surface anywhere shows them, which is precisely the support
    /// incident this work exists to prevent. The number a user is shown, the
    /// number the gate enforces and the number monetize bills have to be the
    /// same number or be labelled as different things.
    ///
    /// So the wall is honest and narrow: it binds exactly when a signed fact
    /// says `lfs_bytes`, and it is `None` — take what arrives — when nothing
    /// does. **A plan that does not set `lfs_bytes` still has an unbounded LFS
    /// tier**, and that is now a fact about what monetize sells rather than a
    /// hole in what gunnar enforces: `products/gunnar`'s `METERS` says
    /// `lfs_bytes` is `Enforced` and the console prices it, so an order that
    /// caps it gets a real wall on this box.
    ///
    /// # No slack, unlike [`Self::pack_ceiling`]
    ///
    /// That one is generous because a pack's wire bytes and its stored bytes are
    /// different numbers nobody can know before the transfer. An LFS object is
    /// stored raw and byte-identical, and its size is declared before the
    /// payload and then verified byte for byte — so the two numbers are the same
    /// number and the answer here is exact. `saturating_sub`, so a tenant already
    /// over their cap gets `0` and is refused every non-empty object rather than
    /// wrapping into 18 exabytes of room.
    fn lfs_allowance_state(&self, namespace: &str) -> Option<gunnar_lfs::Allowance> {
        // A CLAN's LFS objects come out of its one room (round 2, owner
        // 2026-10-05): what is left is the room in force less what it stores
        // in packs and in LFS together.
        if let Ok(Some(clan)) = self.stores.auth().clan(namespace) {
            let (pack, lfs) = self.namespace_bytes(namespace);
            let room = clan.room_in_force(&self.lapsed_members(&clan));
            let held = pack.saturating_add(lfs);
            return Some(gunnar_lfs::Allowance {
                bytes: room.saturating_sub(held),
                refusal: format!(
                    "clan {namespace} has {} bytes of room left ({room} bytes given by its members, {held} \
                     already held in packs and LFS objects): this upload does not fit; its members give it \
                     more room",
                    room.saturating_sub(held)
                ),
            });
        }
        let tenant = TenantId(namespace.to_owned());
        let fact = self.cache.fact(&tenant)?;
        // The same "0 is unlimited, not zero" rule `pack_quota_state` documents:
        // a plan document that omitted the number renders as zero, and reading
        // that as a total prohibition would close every LFS upload on the box
        // the day a plan grew a typo.
        let cap = fact.caps.get("lfs_bytes").copied().filter(|cap| *cap > 0)?;
        let held: u64 = self
            .stores
            .catalog()
            .list(Some(namespace))
            .unwrap_or_default()
            .iter()
            .filter_map(|repo| gunnar_core::RepoRef::new(&repo.namespace, &repo.name).ok())
            .filter_map(|path| self.stores.repo_dir(&path).ok())
            .map(|dir| crate::control_plane::repo_bytes(&dir).lfs)
            .sum();
        Some(gunnar_lfs::Allowance {
            bytes: cap.saturating_sub(held),
            // The same four facts the pack refusals carry — the plan, the cap,
            // what is held, where to pay — because a tenant who meets one of
            // these sentences has usually just met the other, and two different
            // spellings of "you are out of room" is how one support ticket
            // becomes two.
            refusal: format!(
                "plan {} has {} bytes of lfs room left (cap {} bytes, {} already held): \
                 this upload does not fit; delete or buy more at {}",
                fact.plan,
                cap.saturating_sub(held),
                cap,
                held,
                self.url_for(namespace)
            ),
        })
    }
}

/// **The slack in a mid-upload ceiling, as a divisor of the cap** — `16`, i.e.
/// 6.25 %. See [`EntitlementSlot::pack_ceiling`] for why there is slack at all.
pub const PACK_CEILING_SLACK_DIVISOR: u64 = 16;

/// **…and the floor under it**, 64 MiB, for the small plan where a percentage
/// is smaller than one ordinary push. See [`EntitlementSlot::pack_ceiling`].
pub const PACK_CEILING_SLACK_FLOOR_BYTES: u64 = 64 * 1024 * 1024;

/// What a namespace holds against what its plan allows.
///
/// One walk and one lookup, computed once and read by both
/// [`gunnar_ssh::PlanGate::pack_quota`] and
/// [`gunnar_ssh::PlanGate::pack_ceiling`]: the refusal before the bytes and the
/// ceiling during them are two answers about ONE fact, and two copies of this
/// arithmetic could disagree about what a tenant holds (one implementation).
struct PackQuota {
    /// The plan's name, for the refusal line.
    plan: String,
    /// `caps["pack_bytes"]`.
    cap: u64,
    /// What the namespace's repositories hold, off the disk.
    held: u64,
}

impl EntitlementSlot {
    /// [`PackQuota`] for `namespace`, or `None` when there is nothing to hold
    /// it to: no signed fact, or a fact with no `pack_bytes` cap.
    ///
    /// A cap of `0` is "no cap" and not "no bytes": a plan document that
    /// omitted the number renders as zero, and reading that as a total
    /// prohibition would close every write on the box the day a plan grew a
    /// typo.
    fn pack_quota_state(&self, namespace: &str) -> Option<PackQuota> {
        // **A CLAN's room is its members' contributions** (owner, 2026-10-05:
        // "each member contributes storage to the clan") — the sum, and a sum
        // of 0 is 0 bytes, not "no cap": a clan nobody gave room to stores
        // nothing. No fact is signed for a clan; nothing is bought for one.
        //
        // Round 2 (owner, 2026-10-05): the sum of the shares IN FORCE — a
        // member whose paid period has ended gives the clan nothing the door
        // counts ([`Self::lapsed_members`]) — held against what the clan
        // stores in packs AND in LFS objects: one room for both.
        if let Ok(Some(clan)) = self.stores.auth().clan(namespace) {
            let (pack, lfs) = self.namespace_bytes(namespace);
            return Some(PackQuota {
                plan: format!("clan {namespace}"),
                cap: clan.room_in_force(&self.lapsed_members(&clan)),
                held: pack.saturating_add(lfs),
            });
        }
        let tenant = TenantId(namespace.to_owned());
        let Some(fact) = self.cache.fact(&tenant) else {
            return self.no_fact_quota(namespace);
        };
        let cap = fact
            .caps
            .get("pack_bytes")
            .copied()
            .filter(|cap| *cap > 0)?;
        // …and what a member gives its clans is out of its OWN room: the
        // account's cap, less that. An unreadable table gives nothing away.
        let cap = cap.saturating_sub(self.stores.auth().contributed_by(namespace).unwrap_or(0));
        // The namespace's pack bytes, off the disk — the same walk
        // `Usage.Read` bills from (one implementation), so the number a tenant is refused
        // at is the number on their bill. Only reached for a write on a
        // namespace WITH a cap.
        let held = self.namespace_pack_bytes(namespace);
        Some(PackQuota {
            plan: fact.plan.clone(),
            cap,
            held,
        })
    }

    /// **The flag "this estate sells room"** — the one `serve` sets from the feed and hands
    /// the gRPC context too.
    pub fn sells_room_flag(&self) -> Arc<std::sync::atomic::AtomicBool> {
        Arc::clone(&self.sells_room)
    }

    /// [`PackQuota`] for a namespace with NO fact: none (uncapped, as always) — except at an
    /// estate that SELLS room, where a person's own account holds no room until its fact
    /// lands ([`gunnar_grpc::selling::no_room_yet`]). An unreadable mark says seller.
    fn no_fact_quota(&self, namespace: &str) -> Option<PackQuota> {
        let sells = self.sells_room.load(std::sync::atomic::Ordering::Relaxed);
        if !sells {
            return None;
        }
        let auth = self.stores.auth();
        let person = matches!(auth.principal(namespace), Ok(Some(p)) if p.kind == gunnar_core::PrincipalKind::User);
        let seller = auth.holds_server_grant(namespace).unwrap_or(true);
        gunnar_grpc::selling::no_room_yet(sells, false, person, seller).then(|| PackQuota {
            plan: gunnar_grpc::selling::NO_ROOM_YET.to_string(),
            cap: 0,
            held: self.namespace_pack_bytes(namespace),
        })
    }

    /// What a namespace's repositories hold, off the disk — the same walk
    /// `Usage.Read` bills from (one implementation).
    fn namespace_pack_bytes(&self, namespace: &str) -> u64 {
        self.namespace_bytes(namespace).0
    }

    /// `(pack, lfs)` bytes a namespace's repositories hold, off the disk.
    fn namespace_bytes(&self, namespace: &str) -> (u64, u64) {
        self.stores
            .catalog()
            .list(Some(namespace))
            .unwrap_or_default()
            .iter()
            .filter_map(|repo| gunnar_core::RepoRef::new(&repo.namespace, &repo.name).ok())
            .filter_map(|path| self.stores.repo_dir(&path).ok())
            .map(|dir| crate::control_plane::repo_bytes(&dir))
            .fold((0u64, 0u64), |(pack, lfs), bytes| {
                (
                    pack.saturating_add(bytes.pack),
                    lfs.saturating_add(bytes.lfs),
                )
            })
    }

    /// **Is `member`'s share of its clans in force?** (owner, 2026-10-05: "the
    /// clan KEEPS that member's room share until the member's paid period
    /// ends, then the share drops".) Out of force once its fact says the paid
    /// period is over — `paid_until` passed (Grace included: grace is the
    /// ACCOUNT's, after the period), or Suspended, or Retention. A member with
    /// no fact, or a Free one with no period, has nothing that ends.
    fn share_in_force(&self, member: &str, now: u64) -> bool {
        let Some(fact) = self.cache.fact(&TenantId(member.to_owned())) else {
            return true;
        };
        match fact.state {
            State::Suspended | State::Retention => false,
            _ => fact.paid_until_unix_ms.map_or(true, |until| until > now),
        }
    }

    /// **The members of `clan` whose shares are out of force now** — the one
    /// reading the door and the `Clans` answers both make.
    pub fn lapsed_members(&self, clan: &gunnar_core::Clan) -> Vec<String> {
        let now = u64::try_from(now_unix_ms()).unwrap_or(0);
        clan.members
            .iter()
            .filter(|m| !self.share_in_force(&m.principal, now))
            .map(|m| m.principal.clone())
            .collect()
    }
}

// ── the three vocabularies, one mapping each way ────────────────────────────

fn state_from_proto(state: gunnar_proto::EntitlementState) -> Option<EntitlementState> {
    use gunnar_proto::EntitlementState as P;
    Some(match state {
        P::Unspecified => return None,
        P::Free => EntitlementState::Free,
        P::Paid => EntitlementState::Paid,
        P::Grace => EntitlementState::Grace,
        P::Suspended => EntitlementState::Suspended,
        P::Retention => EntitlementState::Retention,
    })
}

pub(crate) fn state_to_proto(state: EntitlementState) -> gunnar_proto::EntitlementState {
    use gunnar_proto::EntitlementState as P;
    match state {
        EntitlementState::Free => P::Free,
        EntitlementState::Paid => P::Paid,
        EntitlementState::Grace => P::Grace,
        EntitlementState::Suspended => P::Suspended,
        EntitlementState::Retention => P::Retention,
    }
}

fn from_monetize_state(state: State) -> EntitlementState {
    match state {
        State::Free => EntitlementState::Free,
        State::Paid => EntitlementState::Paid,
        State::Grace => EntitlementState::Grace,
        State::Suspended => EntitlementState::Suspended,
        State::Retention => EntitlementState::Retention,
    }
}

fn to_monetize_state(state: EntitlementState) -> State {
    match state {
        EntitlementState::Free => State::Free,
        EntitlementState::Paid => State::Paid,
        EntitlementState::Grace => State::Grace,
        EntitlementState::Suspended => State::Suspended,
        EntitlementState::Retention => State::Retention,
    }
}

/// **How far ahead of this server's clock a fact may be stamped before the log
/// says so.** Not a refusal — see [`EntitlementSlot::check_not_replayed`] — and
/// generous, because the appliance syncs its clock from the front and a few
/// minutes of drift is ordinary. An hour is not.
const MAX_ISSUED_AHEAD_MS: u64 = 60 * 60 * 1000;

/// **Are these the same signed statement?** The V2 signature covers every field
/// including the issue time, so equal signatures mean equal facts — that is what
/// Ed25519 is for, and comparing the bytes monetize signed is stronger than
/// comparing fields gunnar copied. The V1 half is compared too: a fact whose two
/// signatures disagree with the held one's is not the held one however its
/// fields read.
fn same_fact(held: Option<&Entitlement>, offered: &Entitlement) -> bool {
    held.is_some_and(|h| {
        h.signature == offered.signature && h.issued_signature == offered.issued_signature
    })
}

/// The stored fact as monetize signed it. Field for field, nothing derived:
/// this is what `verify_fact` hashes, so it must be the bytes monetize saw.
fn to_monetize(fact: &Entitlement) -> monetize_embed::EntitlementFact {
    monetize_embed::EntitlementFact {
        tenant: TenantId(fact.namespace.clone()),
        plan: fact.plan.clone(),
        state: to_monetize_state(fact.state),
        paid_until_unix_ms: fact.paid_until_unix_ms,
        caps: fact.caps.clone(),
        source: fact.source.clone(),
        signature: fact.signature.clone(),
        issued_unix_ms: fact.issued_unix_ms,
        issued_signature: fact.issued_signature.clone(),
    }
}

/// The enforced fact, as the wire carries it. Field for field: the signatures travel so a reader can re-verify what it is told.
fn monetize_to_proto(fact: &monetize_embed::EntitlementFact) -> ProtoFact {
    ProtoFact {
        namespace: fact.tenant.0.clone(),
        plan: fact.plan.clone(),
        state: state_to_proto(from_monetize_state(fact.state)) as i32,
        paid_until_unix_ms: fact.paid_until_unix_ms,
        caps: fact.caps.clone(),
        source: fact.source.clone(),
        signature: fact.signature.clone(),
        issued_unix_ms: fact.issued_unix_ms,
        issued_signature: fact.issued_signature.clone(),
    }
}

fn hex32(k: &[u8; 32]) -> String {
    k.iter().map(|b| format!("{b:02x}")).collect()
}