gunnar-server 3.0.0

//! **The metered appliance, red first.** `Entitlement.Set` verifies, logs,
//! stores and swaps; `authorise_service` refuses a lapsed tenant's push before
//! the bytes and never their clone. Every refusal below has its allow twin in
//! the same test (measure, never assert: a gate that always allows is invisible to a test that
//! only pushes valid work), and every fact is signed with a real Ed25519 key
//! over `monetize_embed::signing::fact_message` — the ONE canonical form — so
//! nothing here is verified against a stand-in.
//!
//! Built only with `--features metered` (`required-features` in Cargo.toml):
//! without the feature the types this file names do not exist, which is the
//! doctrine and not a gap.

use std::collections::BTreeMap;
use std::sync::Arc;

use ed25519_dalek::{Signer, SigningKey};
use gunnar_common::monetize_key::MonetizePubkey;
use gunnar_core::{
    EntitlementState, Level, NewRepo, Principal, PrincipalKind, RepoRef, Visibility,
};
use gunnar_grpc::{ControlError, ControlPlane as _};
use gunnar_proto::{EntitlementFact, EntitlementState as PbState};
use gunnar_server::attest::AttestationLog;
use gunnar_server::control_plane::CoreControlPlane;
use gunnar_server::entitlement::{EntitlementSlot, MeteredConfig};
use gunnar_server::stores::ServerStores;
use gunnar_ssh::command::{GitService, Operation};
use gunnar_ssh::server::{Dispatch, DispatchError};
use gunnar_ssh::PlanGate;
use monetize_embed::{State, TenantId};

const OCT_1_2026: u64 = 1_790_812_800_000;
const BILLING: &str = "https://gunnar.rs/billing/{tenant}";

/// The issue time [`Rig::signed`] stamps unless a test says otherwise. Fixed,
/// so a test that wants two facts in a known order names the offsets itself
/// rather than racing the wall clock.
const ISSUED_AT: u64 = 1_789_000_000_000;

/// The explode budget of a namespace the door refuses — `entitlement.rs`'s
/// `NO_CACHE_BUDGET_BYTES`, spelled again here on purpose. It is private to
/// the server crate, and a test that imported it could not catch the one
/// mistake that matters: `0`, which every layer of this appliance reads as
/// UNLIMITED. This copy fails if the server's ever becomes that.
const NO_CACHE: u64 = 1;

/// **The read cache's disk line**, znippy `exploded_cache::DISK_LINE_PERCENT`:
/// a store on a filesystem this full refuses every write-back into its lazy
/// pool (the engine's own tests hold that; `git_ops`
/// `a_push_past_the_disk_line_evicts_the_oldest_cache_without_a_read`).
const DISK_LINE_PERCENT: u64 = 80;

/// How full the filesystem holding `dir` is, read the way the cache reads it:
/// `statvfs` used / (used + available to an unprivileged writer), as `df`.
fn disk_used_percent(dir: &std::path::Path) -> Option<u64> {
    let s = rustix::fs::statvfs(dir).ok()?;
    let used = s.f_blocks.saturating_sub(s.f_bfree);
    let seen = used + s.f_bavail;
    (seen > 0).then(|| used * 100 / seen)
}

struct Rig {
    _dir: tempfile::TempDir,
    stores: Arc<ServerStores>,
    key: SigningKey,
    forger: SigningKey,
    config: MeteredConfig,
    /// How many facts this rig has minted — monetize's clock, stood in for.
    minted: std::sync::atomic::AtomicU64,
}

impl Rig {
    fn new() -> Rig {
        Rig::in_dir(tempfile::tempdir().expect("tempdir"))
    }

    /// **A rig whose store the read cache may fill**: in TMPDIR when its
    /// filesystem is under the disk line, else in `/dev/shm` (Linux tmpfs) when
    /// that is; the host's own fullness is not what a cache test measures. A
    /// host with neither is said by the test's own assertion, by name.
    fn under_the_disk_line() -> Rig {
        let tmp = std::env::temp_dir();
        let shm = std::path::Path::new("/dev/shm");
        let under = |dir: &std::path::Path| {
            disk_used_percent(dir).is_some_and(|used| used < DISK_LINE_PERCENT)
        };
        let dir = if !under(&tmp) && under(shm) {
            tempfile::tempdir_in(shm).expect("tempdir in /dev/shm")
        } else {
            tempfile::tempdir().expect("tempdir")
        };
        Rig::in_dir(dir)
    }

    fn in_dir(dir: tempfile::TempDir) -> Rig {
        let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
        let key = SigningKey::from_bytes(&[7u8; 32]);
        let forger = SigningKey::from_bytes(&[9u8; 32]);
        let pubkey = dir.path().join("monetize.pub");
        std::fs::write(&pubkey, key.verifying_key().to_bytes()).expect("write key");
        let config =
            MeteredConfig::load(&MonetizePubkey::Path(pubkey), Some(BILLING)).expect("config");
        Rig {
            _dir: dir,
            stores,
            key,
            forger,
            config,
            minted: std::sync::atomic::AtomicU64::new(0),
        }
    }

    fn slot(&self) -> Arc<EntitlementSlot> {
        EntitlementSlot::open(Arc::clone(&self.stores), &self.config).expect("slot")
    }

    fn plane(&self, slot: &Arc<EntitlementSlot>) -> CoreControlPlane {
        CoreControlPlane::new(Arc::clone(&self.stores)).with_entitlement(Arc::clone(slot))
    }

    fn dispatch(&self, slot: &Arc<EntitlementSlot>) -> Dispatch {
        Dispatch::new(
            self.stores
                .ssh_control()
                .with_plan_gate(Arc::clone(slot) as Arc<dyn PlanGate>),
        )
    }

    /// The owner of `ns`: a principal whose id IS the namespace (rule 2 of
    /// `AuthTable::may`), with the namespace and one PUBLIC repository.
    fn tenant(&self, ns: &str) -> Principal {
        let owner = Principal {
            id: ns.to_owned(),
            name: ns.to_owned(),
            kind: PrincipalKind::User,
            created: 0,
        };
        self.stores
            .auth()
            .add_principal(owner.clone())
            .expect("principal");
        self.stores
            .catalog()
            .create_namespace(ns)
            .expect("namespace");
        let path = RepoRef::new(ns, "proj").expect("legal");
        self.stores
            .create_repo(NewRepo::new(path.clone()), false)
            .expect("repo");
        self.stores
            .catalog()
            .set_visibility(&path, Visibility::Public)
            .expect("public");
        owner
    }

    /// A fact stamped with the moment monetize issued it — what
    /// `monetize::Signer::sign_fact` mints now.
    ///
    /// **Every call stamps LATER than the last**, because that is what a clock
    /// does and what every test here already assumes when it pushes a tenant up
    /// and down the ladder. A rig that stamped them all the same would make the
    /// second push of any test a replay, which is true of the wire and false of
    /// monetize.
    fn signed(
        &self,
        key: &SigningKey,
        ns: &str,
        state: State,
        caps: &[(&str, u64)],
    ) -> EntitlementFact {
        let issued = ISSUED_AT
            + self
                .minted
                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        self.signed_at(key, ns, state, caps, Some(issued))
    }

    /// [`Rig::signed`] with the issue time chosen. `None` is the OLD form — a
    /// fact signed before the stamp existed, V1 signature only — which gunnar
    /// refuses.
    fn signed_at(
        &self,
        key: &SigningKey,
        ns: &str,
        state: State,
        caps: &[(&str, u64)],
        issued: Option<u64>,
    ) -> EntitlementFact {
        let mut fact = monetize_embed::EntitlementFact {
            tenant: TenantId(ns.to_owned()),
            plan: "10gb".into(),
            state,
            paid_until_unix_ms: Some(OCT_1_2026),
            caps: caps
                .iter()
                .map(|(k, v)| (k.to_string(), *v))
                .collect::<BTreeMap<_, _>>(),
            source: "payment:invoice:ocr-42".into(),
            signature: vec![],
            issued_unix_ms: issued,
            issued_signature: vec![],
        };
        fact.signature = key
            .sign(&monetize_embed::signing::fact_message(&fact))
            .to_bytes()
            .to_vec();
        if issued.is_some() {
            fact.issued_signature = key
                .sign(&monetize_embed::signing::fact_message_issued(&fact))
                .to_bytes()
                .to_vec();
        }
        EntitlementFact {
            namespace: fact.tenant.0,
            plan: fact.plan,
            state: match state {
                State::Free => PbState::Free,
                State::Paid => PbState::Paid,
                State::Grace => PbState::Grace,
                State::Suspended => PbState::Suspended,
                State::Retention => PbState::Retention,
            } as i32,
            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,
        }
    }

    fn log_lines(&self) -> Vec<String> {
        std::fs::read_to_string(self.stores.root().join(gunnar_server::ATTESTATION_LOG))
            .map(|t| t.lines().map(str::to_owned).collect())
            .unwrap_or_default()
    }
}

fn repo(ns: &str) -> RepoRef {
    RepoRef::new(ns, "proj").expect("legal")
}

// ───────────────────────────────────────────────────────────────────────────
// Step 2: Set verifies, logs, stores, swaps
// ───────────────────────────────────────────────────────────────────────────

#[test]
fn a_forged_fact_is_refused_by_name_and_nothing_moves() {
    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    rig.tenant("team");

    let forged = rig.signed(&rig.forger, "team", State::Paid, &[]);
    let err = plane
        .entitlement_set(forged)
        .expect_err("a signature the monetize key did not make must be refused");
    assert!(
        matches!(err, ControlError::PermissionDenied(_)),
        "a forged fact is PERMISSION_DENIED, not a stored fact: {err:?}"
    );
    assert!(
        err.to_string().contains("does not verify") && err.to_string().contains("team"),
        "the refusal names the reason and the tenant: {err}"
    );
    // NOTHING moved: not the log, not the document, not the snapshot.
    assert!(
        rig.log_lines().is_empty(),
        "a refused fact must not be logged as accepted: {:?}",
        rig.log_lines()
    );
    assert_eq!(plane.entitlement_get("team").unwrap(), None);
    assert!(plane.entitlement_list().unwrap().is_empty());
    assert_eq!(slot.counts().paid, 0);

    // A fact edited AFTER signing is a forgery too, even under the right key.
    let mut edited = rig.signed(&rig.key, "team", State::Paid, &[]);
    edited.state = PbState::Suspended as i32;
    let err = plane
        .entitlement_set(edited)
        .expect_err("edited after signing");
    assert!(matches!(err, ControlError::PermissionDenied(_)), "{err:?}");
    assert!(rig.log_lines().is_empty());

    // The allow twin: the same fact under the real key is accepted, logged
    // ONCE, and enforced.
    let (state, seq) = plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Paid, &[]))
        .expect("a genuine fact is accepted");
    assert_eq!(state, PbState::Paid);
    assert_eq!(seq, 0, "the first line of an empty log is sequence 0");
    let lines = rig.log_lines();
    assert_eq!(lines.len(), 1, "exactly one attestation line: {lines:?}");
    assert!(
        lines[0].contains("\"entitlement\"")
            && lines[0].contains("\"tenant\":\"team\"")
            && lines[0].contains("\"source\":\"payment:invoice:ocr-42\"")
            && lines[0].contains("\"state\":\"Paid\""),
        "the line carries the fact as monetize signed it: {}",
        lines[0]
    );
    assert_eq!(
        plane.entitlement_get("team").unwrap().map(|f| f.state),
        Some(PbState::Paid as i32)
    );
    assert_eq!(slot.counts().paid, 1);
    // …and a push line beside it still reads as a push log: the entitlement
    // line is skipped, not choked on.
    let log = AttestationLog::under(rig.stores.root());
    assert!(log.read().expect("the push log still parses").is_empty());
}

#[test]
fn an_accepted_fact_survives_a_reload_and_a_tampered_document_does_not() {
    let rig = Rig::new();
    rig.tenant("team");
    {
        let slot = rig.slot();
        let plane = rig.plane(&slot);
        plane
            .entitlement_set(rig.signed(
                &rig.key,
                "team",
                State::Suspended,
                &[("cache_bytes", 4096)],
            ))
            .expect("accepted");
        assert_eq!(
            rig.stores
                .catalog()
                .namespace_explode_budget("team")
                .unwrap(),
            Some(NO_CACHE),
            "a SUSPENDED fact does not get its plan's cache cap: the door refuses \
             this tenant, so the budget is the floor and the 4096 in the fact is \
             not applied (see cache_reclaim tests below)"
        );
    }
    // A fresh slot over the same data root: what a reboot does.
    let reborn = rig.slot();
    let plane = rig.plane(&reborn);
    let facts = plane.entitlement_list().unwrap();
    assert_eq!(facts.len(), 1, "{facts:?}");
    assert_eq!(facts[0].state, PbState::Suspended as i32);
    assert_eq!(reborn.counts().suspended, 1);
    // Enforced after the reload, not merely listed.
    let dispatch = rig.dispatch(&reborn);
    let owner = Principal {
        id: "team".into(),
        name: "team".into(),
        kind: PrincipalKind::User,
        created: 0,
    };
    let err = dispatch
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect_err("suspended after reload");
    assert!(matches!(err, DispatchError::Plan { .. }), "{err:?}");

    // The document tampered with (the state flipped to Paid behind the
    // signature): refused at boot, logged, and the namespace is Free — never
    // Paid, because the signature no longer covers it.
    let store = gunnar_core::EntitlementStore::new(Arc::clone(rig.stores.control_store())).unwrap();
    let mut stored = store.get("team").unwrap().unwrap();
    stored.state = EntitlementState::Paid;
    store.set(stored).unwrap();
    let tampered = rig.slot();
    assert_eq!(
        tampered.verdict("team", gunnar_ssh::PlanAction::Write),
        gunnar_ssh::PlanVerdict::Allow,
        "an unverifiable fact is Free (allowed), not the Paid it claims"
    );
    assert_eq!(
        tampered.counts().paid,
        0,
        "the counts are what is ENFORCED, and nothing is"
    );
    assert!(
        tampered.enforced_state("team").is_none(),
        "the cache holds NO fact for it: the tampered fact did not load"
    );
}

#[test]
fn list_shows_the_latest_fact_per_namespace() {
    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    rig.tenant("a");
    rig.tenant("b");
    plane
        .entitlement_set(rig.signed(&rig.key, "a", State::Paid, &[]))
        .unwrap();
    plane
        .entitlement_set(rig.signed(&rig.key, "b", State::Grace, &[]))
        .unwrap();
    let (_, seq) = plane
        .entitlement_set(rig.signed(&rig.key, "a", State::Suspended, &[]))
        .unwrap();
    assert_eq!(seq, 2, "three accepted facts are three lines, 0..=2");
    let facts = plane.entitlement_list().unwrap();
    assert_eq!(facts.len(), 2, "one row per namespace: {facts:?}");
    assert_eq!(
        (facts[0].namespace.as_str(), facts[0].state),
        ("a", PbState::Suspended as i32)
    );
    assert_eq!(
        (facts[1].namespace.as_str(), facts[1].state),
        ("b", PbState::Grace as i32)
    );
    // Four lines, not three: the third fact is Suspended, and a suspension
    // journals what its derived cache gave back on the line after the fact
    // that caused it.
    let lines = rig.log_lines();
    assert_eq!(
        lines.len(),
        4,
        "the log keeps the history; List keeps the latest: {lines:?}"
    );
    assert_eq!(
        lines
            .iter()
            .filter(|l| l.contains("\"cache_reclaim\""))
            .count(),
        1,
        "one reclaim line, for the one namespace that fell to Suspended: {lines:?}"
    );
    let counts = slot.counts();
    assert_eq!((counts.paid, counts.grace, counts.suspended), (0, 1, 1));
}

#[test]
fn a_plane_without_a_monetize_key_refuses_every_set_by_name() {
    let rig = Rig::new();
    let plane = CoreControlPlane::new(Arc::clone(&rig.stores));
    let err = plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Paid, &[]))
        .expect_err("no key, no verification, no acceptance");
    assert!(err.to_string().contains("--monetize-pubkey"), "{err}");
}

// ───────────────────────────────────────────────────────────────────────────
// Step 3: the check, at the one choke point, each refuse with its allow twin
// ───────────────────────────────────────────────────────────────────────────

#[test]
fn a_suspended_push_is_refused_by_name_and_the_same_push_after_a_paid_fact_succeeds() {
    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    let owner = rig.tenant("team");
    let dispatch = rig.dispatch(&slot);

    // Unknown namespace = Free = allowed. The gate's silence for a tenant
    // monetize never mentioned is the product's default, not a refusal.
    dispatch
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect("no fact at all: Free, push allowed");

    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Suspended, &[]))
        .unwrap();
    let err = dispatch
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect_err("a Suspended tenant's push is refused");
    let DispatchError::Plan { namespace, line } = &err else {
        panic!("refused for the wrong reason: {err:?}");
    };
    assert_eq!(namespace, "team");
    // The line names the state, the date and the URL — all three.
    assert!(line.contains("expired 2026-10-01"), "the date: {line}");
    assert!(
        line.contains("writes closed"),
        "the state, as what it means: {line}"
    );
    assert!(
        line.ends_with("renew at https://gunnar.rs/billing/team"),
        "the URL, naming the tenant: {line}"
    );
    assert_eq!(
        err.client_message().as_deref(),
        Some(line.as_str()),
        "and the client is told exactly that"
    );
    // LFS upload is a write too.
    let err = dispatch
        .authorise_service(
            &owner,
            None,
            GitService::LfsTransfer(Operation::Upload),
            repo("team"),
        )
        .expect_err("LFS upload is a write");
    assert!(matches!(err, DispatchError::Plan { .. }), "{err:?}");
    // Retention: the same refusal, and it says the data is kept.
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Retention, &[]))
        .unwrap();
    let err = dispatch
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect_err("Retention refuses writes");
    let DispatchError::Plan { line, .. } = &err else {
        panic!("{err:?}")
    };
    assert!(line.contains("data kept for export"), "{line}");

    // The allow twin: a Paid fact, and the identical push goes through.
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Paid, &[]))
        .unwrap();
    let request = dispatch
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect("Paid: the same push succeeds");
    assert_eq!(request.access, Level::Admin);
    assert_eq!(request.plan_notice, None, "Paid carries no warning");
    // Grace: allowed, with the warning line on the request.
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Grace, &[]))
        .unwrap();
    let request = dispatch
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect("Grace: allowed");
    let notice = request.plan_notice.expect("Grace carries the warning line");
    assert!(
        notice.contains("expired 2026-10-01")
            && notice.ends_with("renew at https://gunnar.rs/billing/team"),
        "{notice}"
    );
}

#[test]
fn the_owner_always_clones_and_anonymous_is_closed_when_suspended() {
    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    let owner = rig.tenant("team");
    let dispatch = rig.dispatch(&slot);
    let anonymous = Principal::anonymous();

    // Before any fact: the public repo is readable by a stranger.
    dispatch
        .authorise_service(&anonymous, None, GitService::UploadPack, repo("team"))
        .expect("Free: anonymous ls-refs/clone of a Public repo");

    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Suspended, &[]))
        .unwrap();
    // Never hold data hostage: the owner's clone succeeds in every state.
    let request = dispatch
        .authorise_service(&owner, None, GitService::UploadPack, repo("team"))
        .expect("the owner clones from a Suspended namespace");
    assert_eq!(request.plan_notice, None);
    dispatch
        .authorise_service(
            &owner,
            None,
            GitService::LfsTransfer(Operation::Download),
            repo("team"),
        )
        .expect("…and downloads their LFS objects");
    // Anonymous read of a Public repo in a Suspended namespace: refused.
    let err = dispatch
        .authorise_service(&anonymous, None, GitService::UploadPack, repo("team"))
        .expect_err("anonymous ls-refs on a Suspended public repo is refused");
    let DispatchError::Plan { line, .. } = &err else {
        panic!("{err:?}")
    };
    assert!(line.contains("anonymous access closed"), "{line}");
    // The allow twin: Paid again, and the stranger is back.
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Paid, &[]))
        .unwrap();
    dispatch
        .authorise_service(&anonymous, None, GitService::UploadPack, repo("team"))
        .expect("Paid: anonymous read allowed again");
    // A Machine principal (replication) is never asked: a standby must keep
    // receiving a lapsed tenant's data or it diverges from the primary.
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Suspended, &[]))
        .unwrap();
    let machine = Principal {
        id: "replicator".into(),
        name: "replicator".into(),
        kind: PrincipalKind::Machine,
        created: 0,
    };
    rig.stores.auth().add_principal(machine.clone()).unwrap();
    rig.stores
        .auth()
        .grant(gunnar_core::Grant {
            principal: "replicator".into(),
            scope: gunnar_core::Scope::Server,
            level: Level::Write,
        })
        .unwrap();
    dispatch
        .authorise_service(&machine, None, GitService::ReceivePack, repo("team"))
        .expect("replication into a Suspended namespace is not a tenant write");
}

#[test]
fn a_push_at_the_pack_quota_is_refused_before_the_bytes_and_under_it_is_not() {
    use gunnar_store::conformance::pack::PackBuilder;
    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    let owner = rig.tenant("team");
    let dispatch = rig.dispatch(&slot);
    // Put ~200 kB of pack in the namespace.
    let repo_row = rig.stores.catalog().get(&repo("team")).unwrap().unwrap();
    let handle = rig.stores.repo_store(&repo_row).unwrap();
    let mut pb = PackBuilder::new(gix_hash::Kind::Sha256);
    pb.object(gix_object::Kind::Blob, vec![b'q'; 200_000]);
    handle.serve().put_pack(&pb.build().unwrap().bytes).unwrap();
    drop(handle);
    rig.stores.close_repo(&repo("team"));
    let held = plane.usage(Some("team")).unwrap()[0].pack_bytes;
    assert!(held > 200_000, "the fixture is on disk: {held}");

    // A cap BELOW what is held: refused, naming the numbers and the URL.
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Paid, &[("pack_bytes", held - 1)]))
        .unwrap();
    let err = dispatch
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect_err("at the quota: refused before any byte");
    let DispatchError::Plan { line, .. } = &err else {
        panic!("{err:?}")
    };
    assert!(
        line.contains("pack quota") && line.contains(&format!("{held} of {} bytes", held - 1)),
        "{line}"
    );
    assert!(line.ends_with("https://gunnar.rs/billing/team"), "{line}");
    // Reads are untouched by a quota.
    dispatch
        .authorise_service(&owner, None, GitService::UploadPack, repo("team"))
        .expect("a full tenant still clones");
    // The allow twins: room under the cap, and no cap at all (0 = unlimited).
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Paid, &[("pack_bytes", held + 1)]))
        .unwrap();
    dispatch
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect("under the cap: allowed");
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Paid, &[("pack_bytes", 0)]))
        .unwrap();
    dispatch
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect("0 is unlimited");
    // A Suspended tenant at the quota is told they are suspended, not full.
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Suspended, &[("pack_bytes", 1)]))
        .unwrap();
    let err = dispatch
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .unwrap_err();
    let DispatchError::Plan { line, .. } = &err else {
        panic!("{err:?}")
    };
    assert!(
        line.contains("writes closed") && !line.contains("quota"),
        "{line}"
    );
}

/// ★ **At an estate that SELLS room, a person's account with no fact may push nothing**
/// until its fact lands (the free-room leak's last window, 2026-10-08); its clone is never
/// touched. The seller's own namespace, a fact that lands, and an estate that does not sell
/// (nothing fed: a sold private estate) keep "no fact is uncapped".
#[test]
fn at_a_selling_estate_a_person_with_no_fact_pushes_nothing_until_its_fact_lands() {
    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    let owner = rig.tenant("team");
    let dispatch = rig.dispatch(&slot);
    dispatch
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect("nothing fed: no fact is uncapped, as always");
    slot.sells_room_flag()
        .store(true, std::sync::atomic::Ordering::Relaxed);
    let err = dispatch
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect_err("a selling estate: no room yet");
    let DispatchError::Plan { line, .. } = &err else {
        panic!("{err:?}")
    };
    assert!(
        line.starts_with(gunnar_grpc::selling::NO_ROOM_YET)
            && line.ends_with("https://gunnar.rs/billing/team"),
        "{line}"
    );
    dispatch
        .authorise_service(&owner, None, GitService::UploadPack, repo("team"))
        .expect("a clone is never refused");
    // The fact lands (the shop's account fact, or a bought room): the plan rules again.
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Paid, &[("pack_bytes", 10 << 30)]))
        .unwrap();
    dispatch
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect("its fact rules");
    // The seller's own: today's rule.
    let op = rig.tenant("nordisk");
    rig.stores
        .auth()
        .grant(gunnar_core::Grant {
            principal: "nordisk".into(),
            scope: gunnar_core::Scope::Server,
            level: Level::Admin,
        })
        .unwrap();
    dispatch
        .authorise_service(&op, None, GitService::ReceivePack, repo("nordisk"))
        .expect("the seller's own namespace");
}

/// **The mid-upload ceiling the front door hands to the receive loop.**
///
/// `a_push_at_the_pack_quota_is_refused_before_the_bytes_and_under_it_is_not`
/// above covers the refusal for a tenant who is ALREADY full. This covers the
/// one that was missing: a tenant with room, whose single push crosses the cap
/// **in flight**. A pack header carries an object count and not a byte count,
/// so nothing before the transfer can know the size — the number asserted here
/// is what `gunnar-wire` counts the pack against, and
/// `gunnar-wire/tests/pack_ceiling.rs` is where the cutting itself is proved.
///
/// Red first — MEASURED 2026-09-05, before `EntitlementSlot::pack_ceiling`
/// existed: `authorise_service` answered a `ServiceRequest` with
/// `pack_ceiling: None` for a capped, paying tenant, so the receive loop had no
/// number to hold the push to and read it to its end.
///
/// The arithmetic is asserted from the constants rather than as a literal
/// total, so the slack can be re-argued without rewriting a magic number — and
/// the two "not this" assertions underneath are what stop it from being a
/// restatement of the code: a ceiling equal to the remaining room, or one that
/// ignored what is held, would both satisfy an assertion written as
/// `ceiling > 0`.
#[test]
fn a_capped_tenant_carries_a_byte_ceiling_into_the_transfer() {
    use gunnar_server::entitlement::{PACK_CEILING_SLACK_DIVISOR, PACK_CEILING_SLACK_FLOOR_BYTES};
    use gunnar_store::conformance::pack::PackBuilder;
    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    let owner = rig.tenant("team");
    let dispatch = rig.dispatch(&slot);

    // ~200 kB held, so `held` is a real number off the disk and not zero.
    let repo_row = rig.stores.catalog().get(&repo("team")).unwrap().unwrap();
    let handle = rig.stores.repo_store(&repo_row).unwrap();
    let mut pb = PackBuilder::new(gix_hash::Kind::Sha256);
    pb.object(gix_object::Kind::Blob, vec![b'q'; 200_000]);
    handle.serve().put_pack(&pb.build().unwrap().bytes).unwrap();
    drop(handle);
    rig.stores.close_repo(&repo("team"));
    let held = plane.usage(Some("team")).unwrap()[0].pack_bytes;
    assert!(held > 200_000, "the fixture is on disk: {held}");

    // ── a big cap: the slack is the proportional term ─────────────────────
    let cap = 10 * 1024 * 1024 * 1024u64; // 10 GiB, so cap/16 = 640 MiB > 64 MiB
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Paid, &[("pack_bytes", cap)]))
        .unwrap();
    let request = dispatch
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect("under the cap: allowed");
    let ceiling = request
        .pack_ceiling
        .expect("a capped tenant's push must carry a ceiling into the transfer");
    let slack = cap / PACK_CEILING_SLACK_DIVISOR;
    assert!(
        slack > PACK_CEILING_SLACK_FLOOR_BYTES,
        "this half of the test is about the proportional term winning"
    );
    assert_eq!(
        ceiling.bytes,
        (cap - held) + slack,
        "the ceiling is the room left under the cap plus its slack"
    );
    // The two shapes this could have been and is not.
    assert_ne!(
        ceiling.bytes,
        cap - held,
        "a ceiling with no slack cuts the push that would have landed exactly at the cap"
    );
    assert_ne!(
        ceiling.bytes,
        cap + slack,
        "the ceiling ignored what the namespace already holds"
    );
    // The refusal is a sentence a tenant can act on, not an indexer's opinion.
    assert!(
        ceiling.refusal.contains("pack quota")
            && ceiling.refusal.contains(&cap.to_string())
            && ceiling.refusal.contains(&held.to_string()),
        "{}",
        ceiling.refusal
    );
    assert!(
        ceiling.refusal.ends_with("https://gunnar.rs/billing/team"),
        "a refusal that does not say where to pay: {}",
        ceiling.refusal
    );

    // ── a small cap: the absolute floor wins ──────────────────────────────
    //
    // 256 MiB / 16 is 16 MiB, smaller than one ordinary push, which is the
    // whole reason the floor exists. Without it a tenant with 200 kB held would
    // be cut mid-clone by arithmetic rather than by their plan.
    let small = 256 * 1024 * 1024u64;
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Paid, &[("pack_bytes", small)]))
        .unwrap();
    let request = dispatch
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect("allowed");
    assert_eq!(
        request.pack_ceiling.expect("a ceiling").bytes,
        (small - held) + PACK_CEILING_SLACK_FLOOR_BYTES,
        "under a small cap the absolute floor is the slack, not cap/{PACK_CEILING_SLACK_DIVISOR}"
    );

    // ── the twins: no ceiling where there is nothing to hold ──────────────
    //
    // A gate that answered `Some` for everything would satisfy every assertion
    // above and would cut a clone (measured, not asserted).
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Paid, &[("pack_bytes", 0)]))
        .unwrap();
    assert_eq!(
        dispatch
            .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
            .expect("0 is unlimited")
            .pack_ceiling,
        None,
        "an uncapped plan must not carry a ceiling"
    );
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Paid, &[("pack_bytes", cap)]))
        .unwrap();
    assert_eq!(
        dispatch
            .authorise_service(&owner, None, GitService::UploadPack, repo("team"))
            .expect("a clone")
            .pack_ceiling,
        None,
        "a read carries no byte ceiling: nothing it does grows the tenant's data"
    );
}

#[test]
fn admin_stats_counts_the_tenants_by_enforced_state() {
    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    for ns in ["a", "b", "c", "d"] {
        rig.tenant(ns);
    }
    plane
        .entitlement_set(rig.signed(&rig.key, "a", State::Paid, &[]))
        .unwrap();
    plane
        .entitlement_set(rig.signed(&rig.key, "b", State::Grace, &[]))
        .unwrap();
    plane
        .entitlement_set(rig.signed(&rig.key, "c", State::Suspended, &[]))
        .unwrap();
    plane
        .entitlement_set(rig.signed(&rig.key, "d", State::Retention, &[]))
        .unwrap();
    let extras = plane.stats_extras().unwrap();
    assert_eq!(
        (
            extras.tenants_paid,
            extras.tenants_grace,
            extras.tenants_suspended
        ),
        (1, 1, 2),
        "paid 1, grace 1, suspended 2 (Suspended + Retention)"
    );
    // The band reads the same slot and prints the same three numbers.
    let view = gunnar_server::dashboard::TenantView::new();
    let for_band = Arc::clone(&slot);
    view.fill(move || for_band.counts());
    assert_eq!(
        view.counts(),
        Some(gunnar_server::dashboard::TenantCounts {
            paid: 1,
            grace: 1,
            suspended: 2
        })
    );
    // And a plane with no slot counts nothing — zeros, not a guess.
    let bare = CoreControlPlane::new(Arc::clone(&rig.stores))
        .stats_extras()
        .unwrap();
    assert_eq!(
        (
            bare.tenants_paid,
            bare.tenants_grace,
            bare.tenants_suspended
        ),
        (0, 0, 0)
    );
}

/// **Both gates, in the server's order** (`metered` build): the disk-headroom
/// gate is the first link and the entitlement gate the second, wired by the
/// SAME `serve::front_control` the binary runs. A full disk outranks a
/// Suspended tenant's line; a roomy disk lets the plan speak.
#[test]
fn the_metered_front_door_carries_the_disk_gate_first_and_the_plan_gate_second() {
    use gunnar_ssh::{DiskFloor, DiskHeadroomGate};

    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    let owner = rig.tenant("team");
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Suspended, &[]))
        .unwrap();
    let gate = |bytes: u64| {
        Arc::new(DiskHeadroomGate::new(
            rig.stores.root(),
            DiskFloor { percent: 0, bytes },
        ))
    };

    let control = gunnar_server::serve::front_control(
        &rig.stores,
        Default::default(),
        gate(u64::MAX),
        Some(&slot),
    );
    assert_eq!(
        control
            .plan_gates()
            .iter()
            .map(|g| g.name())
            .collect::<Vec<_>>(),
        ["disk-headroom", "entitlement"],
        "two links, disk first: {control:?}"
    );
    let err = Dispatch::new(control)
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect_err("refused");
    let DispatchError::Plan { line, .. } = &err else {
        panic!("{err:?}")
    };
    assert!(
        line.starts_with(gunnar_ssh::AT_FLOOR) && !line.contains("renew at"),
        "a full disk is told about the disk, not the plan: {line}"
    );

    let control =
        gunnar_server::serve::front_control(&rig.stores, Default::default(), gate(1), Some(&slot));
    let err = Dispatch::new(control)
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect_err("still refused — by the plan");
    let DispatchError::Plan { line, .. } = &err else {
        panic!("{err:?}")
    };
    assert!(
        line.ends_with("renew at https://gunnar.rs/billing/team"),
        "with room, the plan's line: {line}"
    );

    // No monetize key at all: the metered binary still carries the disk gate.
    let control =
        gunnar_server::serve::front_control(&rig.stores, Default::default(), gate(u64::MAX), None);
    assert_eq!(
        control
            .plan_gates()
            .iter()
            .map(|g| g.name())
            .collect::<Vec<_>>(),
        ["disk-headroom"]
    );
}

/// **The metered appliance's key, given INLINE, verifies a real signed fact.**
///
/// The path arm every other test here uses is unreachable on an appliance: the
/// installer only stages files onto `/data` under `korp.reinstall=wipe` and the
/// ESP is read-only at runtime, so a keep-upgrade can only reach a booting
/// gunnar through the kernel command line. This is that route, measured all the
/// way through — the key is spelled as the 64 hex characters a `korp.args=`
/// would carry, and the fact it accepts is signed with the matching private key
/// over `monetize_embed::signing::fact_message`.
///
/// The assertion is not `is_ok()`. It is that a fact signed by the RIGHT key is
/// accepted and one signed by a FORGER is refused — a config carrying the wrong
/// 32 bytes fails the first arm, and one that verifies nothing at all fails the
/// second.
#[test]
fn a_key_given_inline_accepts_the_signer_and_refuses_a_forger() {
    let dir = tempfile::tempdir().expect("tempdir");
    let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
    let key = SigningKey::from_bytes(&[7u8; 32]);
    let forger = SigningKey::from_bytes(&[9u8; 32]);

    // Exactly what a kernel command line carries: the key as hex text.
    let hex: String = key
        .verifying_key()
        .to_bytes()
        .iter()
        .map(|b| format!("{b:02x}"))
        .collect();
    assert_eq!(hex.len(), 64, "the cmdline spelling is 64 hex characters");
    let inline: MonetizePubkey = hex.parse().expect("the hex key parses");

    let config = MeteredConfig::load(&inline, Some(BILLING)).expect("an inline key needs no file");
    assert_eq!(
        config.monetize_pubkey,
        Some(key.verifying_key().to_bytes()),
        "the inline key must be the signer's 32 verifying-key bytes"
    );

    let rig = Rig {
        _dir: dir,
        stores,
        key,
        forger,
        config,
        minted: std::sync::atomic::AtomicU64::new(0),
    };
    let slot = rig.slot();

    // The real signer is honoured...
    let fact = rig.signed(&rig.key, "acme", State::Paid, &[("cache_bytes", 1 << 30)]);
    slot.set(fact)
        .expect("a fact signed by the inline key is accepted");
    assert_eq!(
        slot.enforced_state("acme"),
        Some(EntitlementState::Paid),
        "the plan must have moved — an accepted fact that changes nothing is not acceptance"
    );

    // ...and a forger is not. Without this arm the test would pass on a config
    // that verified nothing at all.
    let forged = rig.signed(&rig.forger, "acme", State::Suspended, &[]);
    slot.set(forged)
        .expect_err("a fact signed by another key must be refused");
    assert_eq!(
        slot.enforced_state("acme"),
        Some(EntitlementState::Paid),
        "the forged fact must not have moved the plan"
    );
}

// ───────────────────────────────────────────────────────────────────────────
// The allowance on `Usage.Read`: a fact's caps, on the tenant's own grant
// ───────────────────────────────────────────────────────────────────────────

/// `Usage.Read` as `principal`, through the real service over `plane`.
fn usage_read_as(
    plane: CoreControlPlane,
    principal: &str,
    namespace: &str,
) -> Result<gunnar_proto::ReadUsageResponse, tonic::Status> {
    use gunnar_proto::usage_server::Usage as _;
    let control: Arc<dyn gunnar_grpc::ControlPlane> = Arc::new(plane);
    let service = gunnar_grpc::UsageService::new(gunnar_grpc::ServiceContext::new(
        control,
        Arc::new(gunnar_grpc::EventBus::new(16)),
        gunnar_grpc::ServerConfig::default(),
    ));
    let mut request = tonic::Request::new(gunnar_proto::ReadUsageRequest {
        namespace: namespace.to_owned(),
    });
    request.extensions_mut().insert(gunnar_grpc::Caller {
        principal_id: principal.to_owned(),
        fingerprint: "SHA256:test".to_owned(),
        machine: false,
        authenticated: true,
    });
    tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .expect("runtime")
        .block_on(service.read(request))
        .map(tonic::Response::into_inner)
}

/// **alice's cap is on her own `Usage.Read`; bob's is absent; and neither
/// needed `Entitlement.Get`'s server-wide grant.**
///
/// Two DIFFERENT caps on two different meters, so a cap copied into the
/// wrong field is visible; the five meters the fact does not name are absent,
/// not 0; and bob — a tenant with no fact — sees every cap absent on his own
/// read while being refused alice's. Seen red with `allowance_of` reading
/// `caps["pack_bytes"]` into `cache_bytes` (*"cache_bytes cap: left
/// Some(7340032) right Some(1572864)"*), and with the handler returning
/// `Allowance::default()` on the metered build (*"pack_bytes cap: left None
/// right Some(7340032)"*).
#[test]
fn a_tenants_cap_is_on_their_own_usage_read_and_a_tenant_without_a_fact_is_uncapped() {
    let rig = Rig::new();
    let slot = rig.slot();
    rig.tenant("alice");
    rig.tenant("bob");
    rig.plane(&slot)
        .entitlement_set(rig.signed(
            &rig.key,
            "alice",
            State::Paid,
            &[("pack_bytes", 7_340_032), ("cache_bytes", 1_572_864)],
        ))
        .expect("alice's fact");

    let hers = usage_read_as(rig.plane(&slot), "alice", "alice").expect("alice reads alice");
    let allowance = hers.allowance.expect("the allowance is always present");
    assert_eq!(allowance.pack_bytes, Some(7_340_032), "pack_bytes cap");
    assert_eq!(allowance.cache_bytes, Some(1_572_864), "cache_bytes cap");
    for (meter, cap) in [
        ("cold_bytes", allowance.cold_bytes),
        ("lfs_bytes", allowance.lfs_bytes),
        ("tombstoned_bytes", allowance.tombstoned_bytes),
        ("repos", allowance.repos),
        ("objects", allowance.objects),
    ] {
        assert!(
            cap.is_none(),
            "{meter}: the fact names no cap, so none is reported — not {cap:?}"
        );
    }
    // The meters came with it: this is `Usage.Read`, not `Entitlement.Get`.
    assert_eq!(hers.usage.expect("meters").namespace, "alice");

    let his = usage_read_as(rig.plane(&slot), "bob", "bob").expect("bob reads bob");
    // Every cap absent — and the seven enforcement words still there. A fact is
    // what puts a NUMBER on a meter; what this box DOES at a cap is a property
    // of the build and is reported to a tenant who has no fact exactly as it is
    // to one who has (`service::meters`). `Allowance::default()`, which this
    // read for, would assert the opposite: that a tenant on the free tier is
    // told nothing about which meters are doors.
    assert_eq!(
        his.allowance.expect("present"),
        gunnar_proto::Allowance {
            enforcement: gunnar_grpc::service::meters::enforcement(),
            ..gunnar_proto::Allowance::default()
        },
        "bob has no fact: every cap absent"
    );
    let refused =
        usage_read_as(rig.plane(&slot), "bob", "alice").expect_err("bob may not read alice");
    assert_eq!(refused.code(), tonic::Code::PermissionDenied, "{refused:?}");
}

/// ★ **What is left is gunnar's answer, on the tenant's own read**
/// (2026-10-04, the thin layer): cap − held for each capped meter, absent for
/// every uncapped one — and a tenant with no fact has nothing remaining
/// because nothing is capped. alice holds nothing yet, so what is left is
/// the whole cap.
#[test]
fn what_is_left_under_each_cap_is_on_the_tenants_own_usage_read() {
    let rig = Rig::new();
    let slot = rig.slot();
    rig.tenant("alice");
    rig.tenant("bob");
    rig.plane(&slot)
        .entitlement_set(rig.signed(
            &rig.key,
            "alice",
            State::Paid,
            &[("pack_bytes", 7_340_032), ("cache_bytes", 1_572_864)],
        ))
        .expect("alice's fact");

    let hers = usage_read_as(rig.plane(&slot), "alice", "alice").expect("alice reads alice");
    let held = hers.usage.clone().expect("meters");
    let left = hers.remaining.expect("remaining is always present");
    assert_eq!(left.pack_bytes, Some(7_340_032 - held.pack_bytes));
    assert_eq!(left.cache_bytes, Some(1_572_864 - held.cache_bytes));
    for (meter, value) in [
        ("cold_bytes", left.cold_bytes),
        ("lfs_bytes", left.lfs_bytes),
        ("tombstoned_bytes", left.tombstoned_bytes),
        ("repos", left.repos),
        ("objects", left.objects),
    ] {
        assert!(
            value.is_none(),
            "{meter} is uncapped, so nothing is 'remaining': {value:?}"
        );
    }
    let his = usage_read_as(rig.plane(&slot), "bob", "bob").expect("bob reads bob");
    assert_eq!(his.remaining, Some(gunnar_proto::Remaining::default()));
}

/// **A cap of 0 is ABSENT, because 0 is what the push door reads as
/// unlimited** (`a_push_at_the_pack_quota_…`: "0 is unlimited"). A reader of
/// the allowance must never be shown `Some(0)` for a tenant the door lets
/// push without limit. Seen red with the `> 0` filter removed: *"a 0 cap is
/// the door's 'unlimited' and must be ABSENT here, not Some(0)"*.
#[test]
fn a_cap_of_zero_is_reported_as_absent_because_the_door_reads_it_as_unlimited() {
    let rig = Rig::new();
    let slot = rig.slot();
    rig.tenant("team");
    rig.plane(&slot)
        .entitlement_set(rig.signed(
            &rig.key,
            "team",
            State::Paid,
            &[("pack_bytes", 0), ("cache_bytes", 4096)],
        ))
        .expect("the fact");
    let allowance = usage_read_as(rig.plane(&slot), "team", "team")
        .expect("read")
        .allowance
        .expect("present");
    assert_eq!(
        allowance.pack_bytes, None,
        "a 0 cap is the door's 'unlimited' and must be ABSENT here, not Some(0)"
    );
    // The twin: the non-zero cap beside it IS reported, so the rule is "0 is
    // absent" and not "everything is absent".
    assert_eq!(allowance.cache_bytes, Some(4096));
}

/// **A metered server started without `--monetize-pubkey` still answers
/// `Usage.Read`**, with every cap absent: it holds no facts (every namespace
/// is Free, `CoreControlPlane::entitlement` says so by name) and a tenant's
/// meters must not become unreadable because of it. Seen red with the
/// `FailedPrecondition` arm removed from `UsageService::allowance`: the read
/// fails with *"this metered server was started without `--monetize-pubkey`"*.
#[test]
fn a_metered_server_without_a_monetize_key_still_answers_usage_read_uncapped() {
    let rig = Rig::new();
    rig.tenant("team");
    let keyless = CoreControlPlane::new(Arc::clone(&rig.stores));
    let answer = usage_read_as(keyless, "team", "team")
        .expect("a keyless metered server reads meters; it simply holds no caps");
    // Uncapped, and labelled: a server with no key holds no facts, which is not
    // the same as a server that cannot say what its doors are.
    assert_eq!(
        answer.allowance.expect("present"),
        gunnar_proto::Allowance {
            enforcement: gunnar_grpc::service::meters::enforcement(),
            ..gunnar_proto::Allowance::default()
        }
    );
    assert_eq!(answer.usage.expect("meters").namespace, "team");
}

/// **The LFS wall, from the signed fact to the front door's answer.**
///
/// The twin of [`a_capped_tenant_carries_a_byte_ceiling_into_the_transfer`] one
/// wire over, and it exists because the pack ceiling did not cover LFS and
/// nothing else did either. `Dispatch::ask_plan` has always asked a write gate
/// for a ceiling on `LfsTransfer(Upload)` — and the LFS session never read it,
/// because a pack ceiling is measured in `pack_bytes` and an LFS object is not a
/// pack. So a tenant one byte under a 10 GiB `pack_bytes` cap could open an
/// upload session and put 8 GiB in it, under every limit the box had, and then
/// open another. `Usage.Read` billed every byte of it.
///
/// Red first — MEASURED 2026-09-07, before `EntitlementSlot::lfs_allowance`
/// existed: `gunnar_ssh::PlanGate` had no such method and `ServiceRequest` had
/// no such field, so this file did not compile:
///
/// ```text
/// error[E0599]: no method named `lfs_allowance` found for reference
///               `&Arc<EntitlementSlot>` in the current scope
/// ```
///
/// `gunnar-lfs/tests/namespace_allowance.rs` is where the refusal itself is
/// proved on the wire; this is the arithmetic and the wiring.
#[test]
fn a_capped_tenants_lfs_upload_carries_the_rooms_left_into_the_transfer() {
    use std::io::Write as _;
    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    let owner = rig.tenant("team");
    let dispatch = rig.dispatch(&slot);

    // ── real LFS bytes on disk, through the shipping store ────────────────
    //
    // Not a number injected into a fixture: `repo_bytes(..).lfs` walks
    // `<git_dir>/lfs/`, which is where `FsLfsStore` puts objects, and the
    // allowance has to be computed off the same walk `Usage.Read` bills from
    // (one implementation) or the number a tenant is refused at is not the number on their
    // bill.
    const PAYLOAD: &[u8] = b"an lfs object that is really on the disk\n";
    let repo_row = rig.stores.catalog().get(&repo("team")).unwrap().unwrap();
    let handle = rig.stores.repo_store(&repo_row).unwrap();
    let lfs = gunnar_lfs::FsLfsStore::open_or_init(handle.git_dir()).unwrap();
    let mut hasher = gix_hash::hasher(gix_hash::Kind::Sha256);
    hasher.update(PAYLOAD);
    let oid = match hasher.try_finalize().unwrap() {
        gix_hash::ObjectId::Sha256(d) => gunnar_lfs::Oid::from_digest(d),
        other => unreachable!("{other:?}"),
    };
    use gunnar_lfs::LfsStore as _;
    let mut pending = lfs.begin_write(&oid, PAYLOAD.len() as u64).unwrap();
    pending.write_all(PAYLOAD).unwrap();
    pending.commit().unwrap();
    drop(lfs);
    drop(handle);
    rig.stores.close_repo(&repo("team"));

    let held = plane.usage(Some("team")).unwrap()[0].lfs_bytes;
    assert!(
        held >= PAYLOAD.len() as u64,
        "the fixture must be on disk and in the meter: {held}"
    );

    // ── the wall ──────────────────────────────────────────────────────────
    let cap = 10 * 1024 * 1024 * 1024u64;
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Paid, &[("lfs_bytes", cap)]))
        .unwrap();
    let request = dispatch
        .authorise_service(
            &owner,
            None,
            GitService::LfsTransfer(Operation::Upload),
            repo("team"),
        )
        .expect("under the cap: allowed");
    // The gate hands back a `NamespaceAllowance` — the measured number bound to
    // the ledger standing that counts sibling uploads. What is asserted here is
    // the measurement, so read it through `allowance()`; the charging half has
    // its own tests in `gunnar-lfs`.
    let namespace_allowance = request
        .lfs_allowance
        .expect("a capped tenant's lfs upload must carry its room into the transfer");
    let allowance = namespace_allowance.allowance();
    assert_eq!(
        allowance.bytes,
        cap - held,
        "the room is the cap minus what is held, EXACTLY — an LFS object is \
         stored raw and its size is declared before the payload, so there is \
         none of the wire-versus-stored uncertainty the pack ceiling adds slack \
         for"
    );
    // The two shapes this could have been and is not. Either would satisfy an
    // assertion written as `allowance.bytes > 0`.
    assert_ne!(allowance.bytes, cap, "the allowance ignored what is held");
    assert!(
        !allowance.refusal.is_empty()
            && allowance.refusal.contains("lfs room left")
            && allowance.refusal.contains(&cap.to_string())
            && allowance.refusal.contains(&held.to_string()),
        "the sentence a tenant reads must name the cap and what is held: {}",
        allowance.refusal
    );
    assert!(
        allowance
            .refusal
            .ends_with("https://gunnar.rs/billing/team"),
        "a refusal that does not say where to pay: {}",
        allowance.refusal
    );

    // ── the twins: no allowance where there is nothing to hold ────────────
    //
    // A gate that answered `Some` for everything would satisfy every assertion
    // above and would refuse a `git lfs pull` (measured, not asserted).
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Paid, &[("lfs_bytes", 0)]))
        .unwrap();
    assert!(
        dispatch
            .authorise_service(
                &owner,
                None,
                GitService::LfsTransfer(Operation::Upload),
                repo("team")
            )
            .expect("0 is unlimited")
            .lfs_allowance
            .is_none(),
        "an uncapped plan must not carry an lfs allowance"
    );
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Paid, &[("lfs_bytes", cap)]))
        .unwrap();
    assert!(
        dispatch
            .authorise_service(
                &owner,
                None,
                GitService::LfsTransfer(Operation::Download),
                repo("team")
            )
            .expect("a download")
            .lfs_allowance
            .is_none(),
        "a download carries no allowance: nothing it does grows the tenant's data"
    );

    // ── and the meters stay apart ─────────────────────────────────────────
    //
    // The load-bearing negative. A `pack_bytes` cap must NOT silently bound the
    // LFS tier: `Usage.Read` reports `allowance.lfs_bytes` ABSENT for this fact,
    // absent has one documented meaning (uncapped), and a door that refused at a
    // number no surface shows is the support incident this work exists to
    // prevent. What this asserts is that gunnar's wall is honest, not that the
    // tier is safe — a plan that does not buy `lfs_bytes` still has an unbounded
    // one, and that is a fact about what monetize sells.
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Paid, &[("pack_bytes", cap)]))
        .unwrap();
    assert!(
        dispatch
            .authorise_service(
                &owner,
                None,
                GitService::LfsTransfer(Operation::Upload),
                repo("team")
            )
            .expect("allowed")
            .lfs_allowance
            .is_none(),
        "a pack_bytes cap must not become an lfs_bytes wall behind the tenant's back"
    );
}

/// **A metered appliance with no `--billing-url` REFUSES TO BOOT.**
///
/// This is the metered half of the open product's "no money here" rule. The
/// open half is `tests/open_build_speaks_no_money.rs`, which asserts the
/// default build's command line never says the word; this asserts that the
/// build which *may* say it is never allowed to say Vetra's URL on somebody
/// else's behalf.
///
/// What was here before was `unwrap_or("https://gunnar.rs/billing")` in
/// `gunnar_server::dispatch`. A silent default is the worst shape available
/// for this particular string: the box boots, the plans enforce correctly, and
/// the only symptom is the last clause of every refusal a tenant reads —
/// pointing at the page of whoever wrote the fallback rather than whoever runs
/// the appliance. Nothing fails; somebody else's customers just arrive at your
/// checkout.
///
/// The alternative was to tolerate the absence and let refusals carry no link.
/// Rejected: this URL is the ONLY way out of a refusal — it is in the Grace
/// warning, in the `PlanGate` verdict and in the boot log — so an appliance
/// without one refuses tenants into a dead end, and nobody finds out until the
/// support tickets. One line at boot, read by the one person who can fix it in
/// one flag, is the cheap failure.
#[test]
fn a_metered_build_without_a_lapse_url_refuses_at_boot() {
    let dir = tempfile::tempdir().expect("tempdir");
    let key = SigningKey::from_bytes(&[7u8; 32]);
    let pubkey = dir.path().join("monetize.pub");
    std::fs::write(&pubkey, key.verifying_key().to_bytes()).expect("write key");

    let refusal = MeteredConfig::load(&MonetizePubkey::Path(pubkey.clone()), None)
        .expect_err("a metered build with no lapse URL must not start")
        .to_string();

    assert!(
        refusal.contains("--billing-url"),
        "the refusal must name the flag the operator has to add: {refusal}"
    );
    assert!(
        refusal.contains("--monetize-pubkey"),
        "and the flag whose presence made it required: {refusal}"
    );
    assert!(
        !refusal.contains("gunnar.rs"),
        "and it must NOT quietly suggest the URL it just refused to invent: {refusal}"
    );

    // The same key WITH a URL is accepted, so the assertion above cannot be
    // passing because `load` refuses this key for some other reason.
    let ok = MeteredConfig::load(&MonetizePubkey::Path(pubkey), Some(BILLING))
        .expect("the same key with a URL loads");
    assert_eq!(ok.billing_url, BILLING);
}

/// ★ UW-6: **the SITE's pay page is fed, and every refusal ends in it.** The
/// medium is constant (one signed medium for every estate), so the URL it bakes
/// is a placeholder; the installer's door feeds `billing_url` per site
/// (`https://<site>/billing/{tenant}`, a page that site's console serves), and
/// the slot takes it LIVE: the next refusal and Grace line name the site's own
/// page. The same URL again changes nothing; a URL that is not `https://` is
/// refused by name and the page in force stays.
#[test]
fn the_sites_fed_pay_page_replaces_the_mediums_in_every_refusal() {
    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    let owner = rig.tenant("team");
    let dispatch = rig.dispatch(&slot);
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Suspended, &[]))
        .unwrap();
    let refused = |dispatch: &Dispatch| match dispatch.authorise_service(
        &owner,
        None,
        GitService::ReceivePack,
        repo("team"),
    ) {
        Err(DispatchError::Plan { line, .. }) => line,
        other => panic!("a Suspended push is refused by the plan: {other:?}"),
    };
    assert!(
        refused(&dispatch).ends_with("renew at https://gunnar.rs/billing/team"),
        "the medium's page before any feed"
    );

    assert_eq!(
        slot.set_billing_url("https://test2.gunnar.rs/billing/{tenant}"),
        Ok(true),
        "a new page is taken"
    );
    let line = refused(&dispatch);
    assert!(
        line.ends_with("renew at https://test2.gunnar.rs/billing/team"),
        "the site's own page, naming the tenant: {line}"
    );
    assert_eq!(
        slot.set_billing_url("https://test2.gunnar.rs/billing/{tenant}"),
        Ok(false),
        "the same page: nothing moves"
    );
    assert_eq!(
        slot.set_billing_url("https://test2.gunnar.rs/billing"),
        Ok(false),
        "`/{{tenant}}` appended, as the flag's: the same page"
    );

    let err = slot
        .set_billing_url("http://test2.gunnar.rs/billing/{tenant}")
        .expect_err("not https");
    assert!(
        err.contains(gunnar_server::feed::FEED_BILLING_URL) && err.contains("https"),
        "refused by name: {err}"
    );
    assert!(
        refused(&dispatch).ends_with("https://test2.gunnar.rs/billing/team"),
        "a refused page leaves the one in force"
    );

    // Grace carries the same page.
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Grace, &[]))
        .unwrap();
    let request = dispatch
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect("Grace: allowed");
    assert!(request
        .plan_notice
        .expect("Grace warns")
        .ends_with("renew at https://test2.gunnar.rs/billing/team"));
}

// ───────────────────────────────────────────────────────────────────────────
// The cache a tenant the door refuses is no longer allowed to hold
//
// A plan lapsing is NOT a deletion: not one pack, not one ref, not one byte
// of anybody's history moves in any test below, and none of this code can
// move one. What it reclaims is the *derived* explode cache — regenerable
// from the packs it does not touch, never mirrored to a twin
// (`twin::under_root` refuses a cache path outright) — because a tenant who
// has stopped paying is holding space nobody can sell.
// ───────────────────────────────────────────────────────────────────────────

/// The reclaim record for `ns`, parsed out of the attestation log.
fn reclaim_lines(rig: &Rig, ns: &str) -> Vec<serde_json::Value> {
    rig.log_lines()
        .iter()
        .filter_map(|line| serde_json::from_str::<serde_json::Value>(line).ok())
        .filter_map(|value| value.get("cache_reclaim").cloned())
        .filter(|record| record.get("namespace").and_then(|n| n.as_str()) == Some(ns))
        .collect()
}

/// **The trap, asserted rather than trusted: `0` is unlimited.**
///
/// `Namespace::explode_budget_bytes` is `Option<u64>` where `None` is the
/// server default and `Some(0)` is "unlimited — a deliberate answer, not an
/// absent one", and the engine agrees (`over_budget` is `budget != 0 && …`).
/// The obvious spelling of "this tenant gets no cache" is therefore the exact
/// inverse of the intent, and nothing but a test says so out loud: a
/// suspended namespace on `Some(0)` would grow without limit on a full disk
/// and every dashboard would report it as capped.
#[test]
fn a_suspended_namespaces_budget_is_the_floor_and_never_the_zero_that_means_unlimited() {
    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    rig.tenant("lapsed");

    // Paid first, with a real cap, so the assertion below cannot pass on a
    // namespace that never had a budget at all.
    plane
        .entitlement_set(rig.signed(&rig.key, "lapsed", State::Paid, &[("cache_bytes", 8 << 30)]))
        .expect("paid");
    assert_eq!(
        rig.stores
            .catalog()
            .namespace_explode_budget("lapsed")
            .unwrap(),
        Some(8 << 30),
        "a paying tenant is on their plan's number"
    );

    // …and then suspended, with the SAME cap still in the fact. The plan's
    // number is not what a refused tenant gets.
    plane
        .entitlement_set(rig.signed(
            &rig.key,
            "lapsed",
            State::Suspended,
            &[("cache_bytes", 8 << 30)],
        ))
        .expect("suspended");
    let budget = rig
        .stores
        .catalog()
        .namespace_explode_budget("lapsed")
        .unwrap();
    assert_ne!(
        budget,
        Some(0),
        "0 is UNLIMITED at every layer; a suspended tenant on 0 would hold an \
         unbounded cache, which is the opposite of a reclaim"
    );
    assert_eq!(
        budget,
        Some(NO_CACHE),
        "the floor the type can express, and no more"
    );
}

/// Retention is the other refused state and reclaims the same way. Without
/// this arm the whole feature could be keyed on `Suspended` alone and a box
/// that went straight to Retention would keep paying for the cache.
#[test]
fn retention_reclaims_exactly_as_suspended_does() {
    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    rig.tenant("gone");

    plane
        .entitlement_set(rig.signed(
            &rig.key,
            "gone",
            State::Retention,
            &[("cache_bytes", 1 << 30)],
        ))
        .expect("retention");
    assert_eq!(
        rig.stores
            .catalog()
            .namespace_explode_budget("gone")
            .unwrap(),
        Some(NO_CACHE)
    );
    let records = reclaim_lines(&rig, "gone");
    assert_eq!(records.len(), 1, "one reclaim line: {records:?}");
    assert_eq!(records[0]["state"], "Retention");
}

/// **Grace must not reclaim.** A tenant inside the 14-day window is expected
/// to pay and come back; dropping their exploded objects makes the first push
/// after they pay re-inflate every delta chain it touches, for nothing —
/// Grace is short, and the space comes back by itself if they do not return.
#[test]
fn grace_keeps_its_cache_and_journals_no_reclaim() {
    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    rig.tenant("late");

    plane
        .entitlement_set(rig.signed(&rig.key, "late", State::Grace, &[("cache_bytes", 2 << 30)]))
        .expect("grace");

    assert_eq!(
        rig.stores
            .catalog()
            .namespace_explode_budget("late")
            .unwrap(),
        Some(2 << 30),
        "a tenant in Grace keeps the plan's cache: they are expected back"
    );
    assert!(
        reclaim_lines(&rig, "late").is_empty(),
        "Grace reclaims nothing, so there is nothing to journal: {:?}",
        rig.log_lines()
    );

    // The refuse twin, in the same test: the SAME tenant one rung further
    // down does reclaim, so the assertion above cannot be passing because the
    // reclaim never runs for anybody.
    plane
        .entitlement_set(rig.signed(
            &rig.key,
            "late",
            State::Suspended,
            &[("cache_bytes", 2 << 30)],
        ))
        .expect("suspended");
    assert_eq!(
        rig.stores
            .catalog()
            .namespace_explode_budget("late")
            .unwrap(),
        Some(NO_CACHE)
    );
    assert_eq!(reclaim_lines(&rig, "late").len(), 1);
}

/// **The way back up, from the fact's own caps and never from a default
/// invented here** — and the case that has no caps at all, which is the one
/// that used to strand a returning customer on the floor budget for ever.
#[test]
fn paying_again_restores_the_budget_from_the_facts_own_caps() {
    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    rig.tenant("back");
    rig.tenant("plain");

    // A plan that names a cap: suspended, then paid, lands on the fact's
    // number — not on 8 GiB, not on the server default.
    plane
        .entitlement_set(rig.signed(
            &rig.key,
            "back",
            State::Suspended,
            &[("cache_bytes", 4 << 30)],
        ))
        .expect("suspended");
    assert_eq!(
        rig.stores
            .catalog()
            .namespace_explode_budget("back")
            .unwrap(),
        Some(NO_CACHE)
    );
    plane
        .entitlement_set(rig.signed(&rig.key, "back", State::Paid, &[("cache_bytes", 3 << 30)]))
        .expect("paid again");
    assert_eq!(
        rig.stores
            .catalog()
            .namespace_explode_budget("back")
            .unwrap(),
        Some(3 << 30),
        "restored from THIS fact's caps, not from the one before it and not from a constant"
    );

    // A plan that names none: the floor budget was ours, so it is ours to
    // undo, back to `None` = the server's default. Left as `Some(1)` — which
    // is what the old "only write the row when the fact has a cap" rule did —
    // a customer who had just paid would read at uncached speed with the
    // catalog saying they were on the default.
    plane
        .entitlement_set(rig.signed(&rig.key, "plain", State::Suspended, &[]))
        .expect("suspended");
    assert_eq!(
        rig.stores
            .catalog()
            .namespace_explode_budget("plain")
            .unwrap(),
        Some(NO_CACHE)
    );
    plane
        .entitlement_set(rig.signed(&rig.key, "plain", State::Paid, &[]))
        .expect("paid again");
    assert_eq!(
        rig.stores
            .catalog()
            .namespace_explode_budget("plain")
            .unwrap(),
        None,
        "a plan with no cache cap goes back to the server's default, not to our floor"
    );
}

/// An operator's hand-set quota is not something a plan change may clear. The
/// restore path undoes OUR floor budget and nothing else that happens to be
/// on the row.
#[test]
fn the_restore_does_not_clear_a_quota_the_operator_set_by_hand() {
    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    rig.tenant("hand");

    // `gunnar namespace set-explode-budget hand 512M`, as admin.rs does it.
    rig.stores
        .catalog()
        .set_namespace_explode_budget("hand", Some(512 << 20))
        .expect("operator quota");

    // A capless Paid fact arrives. Nothing of the operator's is touched.
    plane
        .entitlement_set(rig.signed(&rig.key, "hand", State::Paid, &[]))
        .expect("paid");
    assert_eq!(
        rig.stores
            .catalog()
            .namespace_explode_budget("hand")
            .unwrap(),
        Some(512 << 20),
        "a plan change with no cache cap is not entitled to clear an operator's quota"
    );
}

/// **The reclaim is journalled per namespace, including when it is zero.**
///
/// The owner's question is "what came back", and a line that only appears
/// when the answer flatters the feature is not an answer: in this rig no
/// store is open, so the honest number is 0 and it is on the record as 0.
///
/// The second half is the trap that would have taken the whole log down.
/// `AttestationLog::read_indexed` hard-errors on any line it cannot parse as
/// a `PushAttestation`, skipping only the shapes it knows. A new record kind
/// whose key is not in that list makes every `gunnar verify` and every
/// transparency proof fail from the first suspension onward.
#[test]
fn the_reclaim_is_journalled_by_namespace_and_the_push_log_still_parses() {
    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    rig.tenant("one");
    rig.tenant("two");

    plane
        .entitlement_set(rig.signed(&rig.key, "one", State::Suspended, &[]))
        .expect("suspended");
    plane
        .entitlement_set(rig.signed(&rig.key, "two", State::Suspended, &[]))
        .expect("suspended");

    for ns in ["one", "two"] {
        let records = reclaim_lines(&rig, ns);
        assert_eq!(
            records.len(),
            1,
            "one line per namespace, not one for the box: {records:?}"
        );
        let record = &records[0];
        assert_eq!(record["state"], "Suspended");
        assert_eq!(record["budget_bytes"], NO_CACHE);
        assert_eq!(
            record["repos"], 1,
            "the namespace's one repository was looked at: {record}"
        );
        assert_eq!(
            record["bytes_freed"], 0,
            "no store is open in this rig, so 0 is the honest number and it is written \
             rather than omitted: {record}"
        );
    }

    // And the log a verifier reads is still a log: the reclaim lines are
    // skipped like the entitlement lines beside them, never choked on.
    let log = AttestationLog::under(rig.stores.root());
    assert!(
        log.read()
            .expect("the push log still parses past the reclaim lines")
            .is_empty(),
        "there were no pushes; the four non-push lines must not become four errors"
    );
    assert_eq!(rig.log_lines().len(), 4, "two facts and two reclaims");
}

/// **The measured arm: a store that IS open gives real bytes back, and the
/// journal carries the number.**
///
/// Everything above this point would pass on an implementation that wrote a
/// budget and reclaimed nothing — every rig in this file has its stores
/// closed, where `Engine::drop_explode_cache` correctly answers 0. So this
/// one pushes a real pack, reads every blob back through `GitServe::read` to
/// make the lazily-exploded pool exist, and then suspends the tenant with the
/// handle still held. The assertion is on bytes, not on an intention.
///
/// Nothing here deletes anything of the customer's: the pack is still there
/// afterwards and the same objects still read, which the tail of the test
/// asserts, because the whole claim of this feature is that only DERIVED
/// bytes move.
#[test]
fn an_open_store_gives_its_derived_bytes_back_and_the_journal_carries_the_number() {
    use gunnar_store::conformance::pack::PackBuilder;

    let rig = Rig::under_the_disk_line();
    // **The read cache refuses every refill on a disk at or past its line**
    // (znippy `exploded_cache::DISK_LINE_PERCENT`, owner 2026-09-28: past 80 %
    // the lazy pool gives its oldest segments back and writes nothing new).
    // That is the product working; a store on such a disk caches nothing, so
    // the reclaim below would measure nothing. Said by name, before anything
    // is read — not as "the reads must have produced a cache" afterwards.
    let full = disk_used_percent(rig._dir.path());
    assert!(
        full.is_some_and(|used| used < DISK_LINE_PERCENT),
        "this rig's store sits on a filesystem {full:?} % full, at or past the read cache's \
         {DISK_LINE_PERCENT} % disk line: the cache rightly refuses every write-back there"
    );
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    rig.tenant("busy");

    // **The pack has to be one the cache actually WANTS.** `worth_caching` is
    // `inflated <= 1 MiB && depth > 2`: a whole 256 KiB blob is declined
    // before a byte is copied, on purpose ("the cheapest thing a cache can do
    // is not cache"), so a fixture of fat base objects would leave the pool
    // empty and this test would be measuring nothing while looking green.
    // Deep chains of small objects are what the pool is for, so that is what
    // is pushed: 40 chains of depth 6, 64 KiB an object — ~10 MiB of
    // admissible content, comfortably past the 4 MiB write-back buffer that
    // makes a segment reach the disk.
    let path = repo("busy");
    let repo_row = rig
        .stores
        .catalog()
        .get(&path)
        .expect("row")
        .expect("exists");
    let handle = rig.stores.repo_store(&repo_row).expect("open");
    let mut pb = PackBuilder::new(gix_hash::Kind::Sha256);
    for chain in 0..40u32 {
        let mut body = vec![b'a' + (chain % 26) as u8; 64 * 1024];
        body[..4].copy_from_slice(&chain.to_le_bytes());
        let mut at = pb.object(gix_object::Kind::Blob, body.clone());
        for link in 0..6u32 {
            body[4..8].copy_from_slice(&link.to_le_bytes());
            at = pb.delta(at, body.clone());
        }
    }
    let fixture = pb.build().expect("pack");
    let oids = fixture.ids();
    handle.serve().put_pack(&fixture.bytes).expect("put_pack");
    handle.serve().index_now().expect("index");

    // The read that creates the cache. Twice: the first read inflates out of
    // the pack and writes back, the second is the one the cache is FOR, and
    // doing both is what makes this a reader rather than a validation pass.
    for _ in 0..2 {
        for oid in &oids {
            let got = handle
                .serve()
                .read(oid.as_bytes())
                .expect("read")
                .expect("present");
            assert_eq!(got.1.len(), 64 * 1024);
        }
    }
    let cached = rig
        .stores
        .engine()
        .explode_use()
        .into_iter()
        .find(|(dir, _)| dir == handle.git_dir())
        .map(|(_, band)| band.used())
        .unwrap_or(0);
    assert!(
        cached > 0,
        "the reads must have produced a cache, or the reclaim below measures nothing"
    );

    // Suspend, with the handle still open — the case the reclaim is for.
    plane
        .entitlement_set(rig.signed(&rig.key, "busy", State::Suspended, &[]))
        .expect("suspended");

    let records = reclaim_lines(&rig, "busy");
    assert_eq!(records.len(), 1, "{records:?}");
    let freed = records[0]["bytes_freed"].as_u64().expect("a number");
    assert!(
        freed > 0,
        "an OPEN store's derived cache must come back as bytes, not as an intention: {}",
        records[0]
    );
    assert_eq!(records[0]["repos_reclaimed"], 1);
    // The WHOLE band, not a slice of it. This fixture pushes blobs only, so
    // the eager commit-and-tree pool — which a reclaim deliberately keeps,
    // because it is the graph and not a cache — is empty here and the band is
    // entirely the lazy pool. A fixture with commits in it would make these
    // two differ by exactly the graph, which is the intended behaviour and
    // would need a different assertion, not a looser one.
    assert_eq!(
        freed, cached,
        "every derived byte the store was holding came back: {}",
        records[0]
    );
    let after = rig
        .stores
        .engine()
        .explode_use()
        .into_iter()
        .find(|(dir, _)| dir == handle.git_dir())
        .map(|(_, band)| band.used())
        .unwrap_or(0);
    assert_eq!(
        after, 0,
        "and the store is no longer holding it — the reclaim was applied, not reported"
    );

    // And the customer's data is untouched: the same objects still read, out
    // of the packs the reclaim did not go near.
    for oid in &oids {
        let got = handle
            .serve()
            .read(oid.as_bytes())
            .expect("read")
            .expect("still present");
        assert_eq!(
            got.1.len(),
            64 * 1024,
            "a reclaim that lost an object is a data loss bug"
        );
    }
}

// ───────────────────────────────────────────────────────────────────────────
// The replay guard: a fact is good once, and never again
// ───────────────────────────────────────────────────────────────────────────

/// **A captured `Paid` fact must not restore a tenant who has since lapsed.**
///
/// Nothing here is forged. Every fact is monetize's own, signed by monetize's
/// own key, and step 3 pushes the EXACT BYTES of step 1 back — which is all an
/// attacker who can reach `Entitlement.Set` (server-admin) has to keep. The
/// signature verifies, because it is a real signature; the question this test
/// asks is whether verifying is enough, and the answer must be no.
///
/// `paid_until` cannot answer it: the lapsed fact in step 2 carries the SAME
/// `paid_until` as the paid one in step 1 — that is what a lapse IS, the date
/// stayed put and the ladder moved past it — so there is no field in the fact
/// a box could compare to tell the replay from the original. Only something
/// monotonic in the SIGNED form can, which is why this test's fix is not local
/// to gunnar.
#[test]
fn a_captured_paid_fact_is_refused_after_the_tenant_lapsed() {
    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    let owner = rig.tenant("team");
    let dispatch = rig.dispatch(&slot);

    // 1. The tenant pays, and the wire bytes are CAPTURED.
    let captured = rig.signed(&rig.key, "team", State::Paid, &[]);
    plane
        .entitlement_set(captured.clone())
        .expect("Paid is accepted");
    dispatch
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect("Paid: the push goes through");

    // 2. The tenant lapses. Same `paid_until`; the ladder moved, not the date.
    plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Suspended, &[]))
        .expect("Suspended is accepted");
    dispatch
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect_err("Suspended: the push is refused");

    // 3. THE REPLAY. The captured bytes, verbatim, months later.
    let refused = plane
        .entitlement_set(captured)
        .expect_err("a fact this box has already moved past is not accepted a second time");
    let message = format!("{refused:?}");
    assert!(
        message.contains("team"),
        "the refusal names the namespace: {message}"
    );

    // 4. And the service stayed withdrawn.
    dispatch
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect_err("the replay did not restore the write door");
    assert_eq!(
        slot.enforced_state("team"),
        Some(EntitlementState::Suspended),
        "the box still enforces the lapse"
    );
}

/// **The poller's retry path must stay green.** `retry_pushes` re-sends a fact
/// the product never acknowledged, verbatim — same bytes, same stamp — and that
/// is how a lost push is repaired at all. The guard must not read it as a
/// replay, because the two look identical from the outside except for one
/// thing: a retry is the SAME fact, and a replay is an older one.
///
/// Written alongside the replay test on purpose. Get this wrong and the guard
/// trades a rare attack for a common outage.
#[test]
fn the_same_fact_pushed_again_is_idempotent_and_is_not_a_replay() {
    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    let owner = rig.tenant("team");
    let dispatch = rig.dispatch(&slot);

    // One fact, minted once. Both pushes carry these exact bytes — which is what
    // `retry_pushes` does after an unacknowledged push.
    let fact = rig.signed_at(
        &rig.key,
        "team",
        State::Paid,
        &[("pack_bytes", 1 << 30)],
        Some(ISSUED_AT),
    );
    let (first, seq_one) = plane.entitlement_set(fact.clone()).expect("the push");
    let (again, seq_two) = plane
        .entitlement_set(fact.clone())
        .expect("the RETRY of the same push");
    assert_eq!(first, again, "the same fact enforces the same state");
    assert_ne!(
        seq_one, seq_two,
        "and the retry is its own attestation line, honestly"
    );
    assert_eq!(
        slot.enforced_state("team"),
        Some(EntitlementState::Paid),
        "nothing moved, because nothing was different"
    );
    dispatch
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect("the tenant is still served");

    // A third push, still the same bytes, after a reboot: the latch is in the
    // document and the retry survives it too.
    let reopened = rig.slot();
    rig.plane(&reopened)
        .entitlement_set(fact)
        .expect("a retry that arrives after a restart is still not a replay");

    // …but a DIFFERENT fact at the same instant is not a retry, and is refused.
    let twin = rig.signed_at(&rig.key, "team", State::Suspended, &[], Some(ISSUED_AT));
    let err = rig
        .plane(&reopened)
        .entitlement_set(twin)
        .expect_err("same instant, different fact: not newer, so not accepted");
    assert!(matches!(err, ControlError::PermissionDenied(_)), "{err:?}");
}

/// ★ **An unstamped fact is refused, on a new namespace too** (stinky row 12).
/// The grandfather clause accepted facts with no issue time until the
/// namespace's first stamped one, so a captured unstamped `Paid` fact could be
/// pushed for any namespace that had never been stamped — and over and over for
/// one that never would be. monetize stamps every fact it signs and the
/// monetize this talks to is a test system, so there is no legacy to keep: no
/// issue time, no fact.
#[test]
fn an_unstamped_fact_is_refused_even_for_a_namespace_that_has_none() {
    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);

    // A namespace with no fact at all: the unstamped form is refused.
    let old = rig.signed_at(&rig.key, "team", State::Paid, &[], None);
    let err = plane
        .entitlement_set(old.clone())
        .expect_err("an unstamped fact for a fresh namespace");
    assert!(matches!(err, ControlError::PermissionDenied(_)), "{err:?}");
    assert_eq!(
        slot.enforced_state("team"),
        None,
        "and the refusal moved nothing"
    );

    // A stamped one is accepted…
    let stamped = rig.signed_at(&rig.key, "team", State::Suspended, &[], Some(ISSUED_AT));
    plane
        .entitlement_set(stamped)
        .expect("a stamped fact is accepted");

    // …and the unstamped form stays refused: the captured `Paid` from the top.
    let err = plane
        .entitlement_set(old)
        .expect_err("an unstamped fact for a namespace that has one");
    assert!(matches!(err, ControlError::PermissionDenied(_)), "{err:?}");
    assert_eq!(
        slot.enforced_state("team"),
        Some(EntitlementState::Suspended),
        "and the refusal moved nothing"
    );
}

/// **The latch is in the control document, so a reboot does not forget it.** A
/// guard held only in memory would be defeated by waiting for a restart, which
/// on an appliance is a thing that happens by itself.
///
/// The refusal is also read here for what it does NOT say: `Entitlement.Set` is
/// server-admin and `Get` is server-read today, but the words must not become
/// the way to read one tenant's billing state through a push at another.
#[test]
fn the_high_water_mark_survives_a_restart_and_the_refusal_says_nothing_extra() {
    let rig = Rig::new();
    let owner = rig.tenant("team");

    let captured = {
        let slot = rig.slot();
        let plane = rig.plane(&slot);
        let captured = rig.signed(&rig.key, "team", State::Paid, &[]);
        plane.entitlement_set(captured.clone()).unwrap();
        plane
            .entitlement_set(rig.signed(&rig.key, "team", State::Suspended, &[]))
            .unwrap();
        captured
    };

    // A new slot over the same stores: what the appliance comes back as.
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    assert_eq!(
        slot.enforced_state("team"),
        Some(EntitlementState::Suspended),
        "the lapse survived the restart"
    );
    let err = plane
        .entitlement_set(captured)
        .expect_err("and so did the high-water mark");
    let ControlError::PermissionDenied(words) = &err else {
        panic!("{err:?}")
    };
    assert!(
        words.contains("team"),
        "the refusal names the namespace: {words}"
    );
    assert!(
        words.contains("replay"),
        "and says what it refused it as: {words}"
    );
    for leak in [
        "Suspended",
        "Paid",
        &ISSUED_AT.to_string(),
        &OCT_1_2026.to_string(),
    ] {
        assert!(
            !words.contains(leak),
            "the refusal must not carry {leak}: a caller who cannot read this namespace's fact \
             must not learn it by pushing at it — {words}"
        );
    }
    rig.dispatch(&slot)
        .authorise_service(&owner, None, GitService::ReceivePack, repo("team"))
        .expect_err("the replay did not restore the write door");
}

/// **An issue time that was added or bumped by hand is refused at the
/// signature**, not at the comparison — otherwise the guard would be its own
/// weapon: an unsigned stamp of `u64::MAX` would pin the namespace past every
/// fact monetize will ever mint for it.
#[test]
fn refuse_twin_a_hand_written_issue_time_is_refused_as_forged_and_pins_nothing() {
    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);

    plane
        .entitlement_set(rig.signed_at(&rig.key, "team", State::Paid, &[], Some(ISSUED_AT)))
        .expect("a genuine stamped fact");

    // The same genuine fact with the stamp moved to the end of time.
    let mut bumped = rig.signed_at(&rig.key, "team", State::Paid, &[], Some(ISSUED_AT + 1));
    bumped.issued_unix_ms = Some(u64::MAX);
    let err = plane
        .entitlement_set(bumped)
        .expect_err("a bumped stamp is not monetize's word");
    let ControlError::PermissionDenied(words) = &err else {
        panic!("{err:?}")
    };
    assert!(
        words.contains("does not verify"),
        "refused as forged, not as stale: {words}"
    );

    // And nothing was pinned: the next genuine fact still lands.
    plane
        .entitlement_set(rig.signed_at(
            &rig.key,
            "team",
            State::Suspended,
            &[],
            Some(ISSUED_AT + 2),
        ))
        .expect("the namespace was not wedged by the attempt");
    assert_eq!(
        slot.enforced_state("team"),
        Some(EntitlementState::Suspended)
    );
}

// ───────────────────────────────────────────────────────────────────────────
// A restore under the WRONG monetize key
// ───────────────────────────────────────────────────────────────────────────

/// **A box restored under a monetize key that did not sign its facts keeps
/// every fact on disk, enforces none of them, and must SAY so — and must not
/// report a plan it does not enforce.**
///
/// This is the shape a twin restore takes when the promoted box is given the
/// wrong `--monetize-pubkey`: the control document came across byte for byte,
/// so every fact is still there; not one verifies, so every paying namespace is
/// served as Free. MEASURED 2026-09-21 before the fix, on this very rig:
/// `refused_at_boot` did not exist (the boot line said `tenants = 0`, the line a
/// healthy empty box logs), and `Entitlement.Get` answered `PAID` — out of the
/// store — for a namespace the cache and therefore the front door held Free.
///
/// Then the way back, because it is the operator's whole remedy: the RIGHT key,
/// and nothing else. No fact is re-pushed; re-opening the same document under
/// the key that signed it restores every plan. A re-push under the wrong key is
/// refused as forged, which is why the boot line says "never a re-push".
#[test]
fn a_box_restored_under_the_wrong_key_reports_no_plan_it_does_not_enforce() {
    let rig = Rig::new();
    rig.tenant("team");
    rig.tenant("acme");
    {
        let slot = rig.slot();
        let plane = rig.plane(&slot);
        plane
            .entitlement_set(rig.signed(&rig.key, "team", State::Paid, &[]))
            .expect("genuine");
        plane
            .entitlement_set(rig.signed(&rig.key, "acme", State::Grace, &[]))
            .expect("genuine");
        // THE CONTROL: under the key that signed them, both are enforced and
        // both are reported, and nothing was refused.
        assert_eq!(slot.refused_at_boot(), 0);
        assert_eq!(plane.entitlement_refused_at_boot().unwrap(), None);
        assert!(plane.health_alarms().is_empty());
        assert_eq!(plane.entitlement_list().unwrap().len(), 2);
    }

    // The restore: the same stores, a different monetize key.
    let wrong = rig._dir.path().join("wrong.pub");
    std::fs::write(&wrong, rig.forger.verifying_key().to_bytes()).expect("write key");
    let wrong = MeteredConfig::load(&MonetizePubkey::Path(wrong), Some(BILLING)).expect("config");
    let slot =
        EntitlementSlot::open(Arc::clone(&rig.stores), &wrong).expect("a wrong key still boots");
    let plane = rig.plane(&slot);

    assert_eq!(
        slot.refused_at_boot(),
        2,
        "both stored facts were refused, and the box COUNTS them"
    );
    // ★ ON THE WIRE, not only in a log line: on the UpCloud machine there is no
    // serial console and the boot WARN reaches nobody (MEASURED 2026-09-21).
    // `Entitlement.List` carries the refusal and `Admin.Health` an alarm.
    let refused = plane
        .entitlement_refused_at_boot()
        .unwrap()
        .expect("Entitlement.List must carry the refusal");
    assert_eq!(refused.count, 2);
    assert_eq!(
        refused.namespaces,
        vec!["acme".to_string(), "team".to_string()]
    );
    assert!(
        refused.at_unix_ms > 0 && refused.reason.contains("monetize public key"),
        "{refused:?}"
    );
    let alarms = plane.health_alarms();
    assert_eq!(alarms.len(), 1, "{alarms:?}");
    assert_eq!(
        alarms[0].event, "entitlement.key_mismatch",
        "--monetize-vital is the default"
    );
    assert!(
        alarms[0].detail.contains("acme, team"),
        "the alarm names the namespaces: {}",
        alarms[0].detail
    );
    assert_eq!(alarms[0].count, 2);
    assert_eq!(slot.enforced_state("team"), None, "Free at the front door");
    assert_eq!(slot.counts().paid, 0);
    // ★ The reads agree with the door. Before the fix both of these answered
    // the stored fact: `PAID` with a renewal date nothing enforced.
    assert_eq!(
        plane.entitlement_get("team").unwrap(),
        None,
        "Entitlement.Get reported a plan the box does not enforce"
    );
    assert!(
        plane.entitlement_list().unwrap().is_empty(),
        "Entitlement.List listed facts the box refused"
    );

    // A re-push signed by the key that is NOT configured is refused, by name.
    let err = plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Paid, &[]))
        .expect_err("a fact the configured key did not sign must be refused");
    assert!(matches!(err, ControlError::PermissionDenied(_)), "{err:?}");

    // The remedy is the right key and nothing else: the facts never left.
    drop(plane);
    drop(slot);
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    assert_eq!(slot.refused_at_boot(), 0);
    assert_eq!(slot.enforced_state("team"), Some(EntitlementState::Paid));
    assert_eq!(
        plane.entitlement_refused_at_boot().unwrap(),
        None,
        "the right key refuses nothing"
    );
    assert!(plane.health_alarms().is_empty());
    assert_eq!(slot.enforced_state("acme"), Some(EntitlementState::Grace));
    assert_eq!(
        plane.entitlement_get("team").unwrap().map(|f| f.state),
        Some(PbState::Paid as i32)
    );
}

// ── `monetize_vital` (owner ruling 2026-09-22) ─────────────────────────────

/// The restore under another key, set up once: two facts on disk signed by `rig.key`, and a
/// slot opened under `rig.forger`'s key with `vital`.
fn restored_under_another_key(rig: &Rig, vital: bool) -> Arc<EntitlementSlot> {
    rig.tenant("team");
    rig.tenant("acme");
    {
        let slot = rig.slot();
        let plane = rig.plane(&slot);
        plane
            .entitlement_set(rig.signed(&rig.key, "team", State::Paid, &[]))
            .expect("genuine");
        plane
            .entitlement_set(rig.signed(&rig.key, "acme", State::Grace, &[]))
            .expect("genuine");
        assert!(
            plane.maintenance_reasons().is_empty(),
            "the control: the right key is not maintenance"
        );
    }
    let other = rig._dir.path().join("other.pub");
    std::fs::write(&other, rig.forger.verifying_key().to_bytes()).expect("write key");
    let mut config =
        MeteredConfig::load(&MonetizePubkey::Path(other), Some(BILLING)).expect("config");
    config.vital = vital;
    EntitlementSlot::open(Arc::clone(&rig.stores), &config)
        .expect("a mismatched key still boots and serves")
}

/// A key mismatch at boot raises `entitlement.key_mismatch`, naming the namespaces, and the box
/// reports maintenance — and it still started.
#[test]
fn a_key_mismatch_raises_the_alarm_and_maintenance() {
    let rig = Rig::new();
    let slot = restored_under_another_key(&rig, true);
    let plane = rig.plane(&slot);
    assert!(slot.maintenance());
    assert_eq!(
        plane.maintenance_reasons(),
        vec!["entitlement.key_mismatch".to_string()]
    );
    let alarms = plane.health_alarms();
    assert_eq!(alarms.len(), 1, "{alarms:?}");
    assert_eq!(alarms[0].event, "entitlement.key_mismatch");
    assert_eq!(alarms[0].count, 2);
    assert!(
        alarms[0].detail.contains("acme, team") && alarms[0].detail.contains("MAINTENANCE"),
        "{}",
        alarms[0].detail
    );
}

/// The first fact that VERIFIES under the box's key clears both: monetize and the box agree again.
#[test]
fn a_new_verifying_fact_clears_the_alarm_and_maintenance() {
    let rig = Rig::new();
    let slot = restored_under_another_key(&rig, true);
    let plane = rig.plane(&slot);
    // A fact the box's key did NOT sign clears nothing.
    assert!(plane
        .entitlement_set(rig.signed(&rig.key, "team", State::Paid, &[]))
        .is_err());
    assert!(slot.maintenance(), "a refused fact is not a verifying one");
    // monetize, now holding the key this box trusts, pushes a fact: it verifies.
    plane
        .entitlement_set(rig.signed(&rig.forger, "team", State::Paid, &[]))
        .expect("verifies under the box's key");
    assert!(!slot.maintenance());
    assert!(plane.maintenance_reasons().is_empty());
    assert!(
        plane.health_alarms().is_empty(),
        "{:?}",
        plane.health_alarms()
    );
    assert_eq!(
        slot.enforced_state("team"),
        Some(gunnar_core::EntitlementState::Paid)
    );
}

/// `--monetize-vital false`: the same refusal is only logged — no alarm, no maintenance.
#[test]
fn with_monetize_vital_false_a_mismatch_only_logs() {
    let rig = Rig::new();
    let slot = restored_under_another_key(&rig, false);
    let plane = rig.plane(&slot);
    assert_eq!(
        slot.refused_at_boot(),
        2,
        "still counted and still on Entitlement.List"
    );
    assert!(plane.entitlement_refused_at_boot().unwrap().is_some());
    assert!(!slot.maintenance());
    assert!(plane.maintenance_reasons().is_empty());
    assert!(
        plane.health_alarms().is_empty(),
        "{:?}",
        plane.health_alarms()
    );
}

/// The server flag: `true` unless it is set to false, by flag or by `GUNNAR_MONETIZE_VITAL`.
#[test]
fn monetize_vital_defaults_to_true_and_parses_false() {
    use clap::Parser;
    let parse = |extra: &[&str]| {
        let mut argv = vec![
            "gunnar",
            "serve",
            "--monetize-pubkey",
            "1c3663248d60900e2996389fba4efa1d7179d6dbd08627d0426f16f94e44c4e6",
            "--billing-url",
            BILLING,
        ];
        argv.extend_from_slice(extra);
        match gunnar_common::cli::Cli::try_parse_from(argv)
            .expect("parses")
            .command
        {
            gunnar_common::cli::Command::Serve(a) => a.monetize_vital,
            _ => unreachable!(),
        }
    };
    assert!(parse(&[]));
    assert!(!parse(&["--monetize-vital", "false"]));
    assert!(parse(&["--monetize-vital", "true"]));
}

/// **The fed monetize key applies LIVE** (owner 2026-09-25: avoid restarts).
/// A metered box on a `fed` medium opens its slot KEYLESS; a fact signed by
/// monetize is refused with nothing moved (no key to check it with), and every
/// namespace is Free. The key arrives — through the SAME door the pairing loop
/// and the installer's feed use (`feed::monetize_arrived` → the registered
/// takers) — and the very same fact is then ACCEPTED and ENFORCED by the same
/// slot object, with no restart. A second, different key is refused by name.
#[test]
fn a_fed_monetize_key_applies_in_place_and_a_fact_signed_with_it_is_accepted_without_a_restart() {
    let rig = Rig::new();
    let keyless =
        MeteredConfig::awaiting_feed(Some(BILLING)).expect("a fed medium with a pay page");
    assert_eq!(keyless.monetize_pubkey, None);
    let slot =
        EntitlementSlot::open(Arc::clone(&rig.stores), &keyless).expect("a keyless slot opens");
    rig.tenant("acme");
    let paid = rig.signed(&rig.key, "acme", State::Suspended, &[]);
    let refused = slot
        .set(paid.clone())
        .expect_err("no key yet: nothing can be verified");
    assert!(
        refused.to_string().contains("no monetize key yet"),
        "{refused}"
    );
    assert_eq!(
        slot.verdict("acme", gunnar_ssh::PlanAction::Write),
        gunnar_ssh::PlanVerdict::Allow,
        "keyless: Free, as a box with no slot was"
    );

    // The key arrives while running, through the feed's door.
    gunnar_server::feed::record_monetize_in_force(None);
    let taker = Arc::clone(&slot);
    gunnar_server::feed::on_monetize_key(
        "entitlement slot (test)",
        Box::new(move |k| taker.set_key(k).map(|_| ())),
    );
    let hex: String = rig
        .key
        .verifying_key()
        .to_bytes()
        .iter()
        .map(|b| format!("{b:02x}"))
        .collect();
    let arrived = gunnar_server::feed::monetize_arrived(Some(&hex));
    assert!(
        matches!(
            arrived,
            Some(gunnar_server::feed::MonetizeArrival::Applied(_))
        ),
        "{arrived:?}"
    );
    assert_eq!(
        gunnar_server::feed::monetize_in_force(),
        Some(rig.key.verifying_key().to_bytes())
    );
    assert_eq!(
        gunnar_server::feed::monetize_arrived(Some(&hex)),
        None,
        "the same key again: nothing to do"
    );

    // The SAME slot, no restart: the fact is accepted and enforced.
    let (state, _) = slot
        .set(paid)
        .expect("a fact signed with the fed key is accepted without a restart");
    assert_eq!(state, EntitlementState::Suspended);
    assert!(
        matches!(
            slot.verdict("acme", gunnar_ssh::PlanAction::Write),
            gunnar_ssh::PlanVerdict::Refuse(_)
        ),
        "and enforced"
    );
    // A forged fact is still refused under the fed key.
    assert!(slot
        .set(rig.signed(&rig.forger, "acme", State::Paid, &[]))
        .is_err());
    // Another key while one is in force is refused by name, and nothing moves.
    let other = slot
        .set_key(&rig.forger.verifying_key().to_bytes())
        .expect_err("a second key is not taken silently");
    assert!(other.contains("[pairing.feed_mismatch]"), "{other}");
    assert_eq!(
        slot.set_key(&rig.key.verifying_key().to_bytes()),
        Ok(false),
        "the key in force: nothing to do"
    );
}

/// A keyless slot applies a key that the STORE already holds facts for: the
/// facts on disk are re-verified under it at once, as a boot under that key
/// would have — so a box restored before its key arrived is enforcing the
/// moment it does.
#[test]
fn a_key_applied_in_place_reverifies_the_facts_already_on_disk() {
    let rig = Rig::new();
    rig.tenant("acme");
    // A keyed slot stores a Suspended fact; the box "restarts" keyless.
    rig.slot()
        .set(rig.signed(&rig.key, "acme", State::Suspended, &[]))
        .expect("stored");
    let slot = EntitlementSlot::open(
        Arc::clone(&rig.stores),
        &MeteredConfig::awaiting_feed(Some(BILLING)).unwrap(),
    )
    .unwrap();
    assert_eq!(
        slot.verdict("acme", gunnar_ssh::PlanAction::Write),
        gunnar_ssh::PlanVerdict::Allow,
        "keyless: Free"
    );
    assert_eq!(slot.set_key(&rig.key.verifying_key().to_bytes()), Ok(true));
    assert!(
        matches!(
            slot.verdict("acme", gunnar_ssh::PlanAction::Write),
            gunnar_ssh::PlanVerdict::Refuse(_)
        ),
        "the stored fact is enforced the moment the key arrives"
    );
    assert_eq!(slot.refused_at_boot(), 0);
}

/// A fact with its paid time and its payment chosen (c-stale-suspend).
fn fact_paid(
    key: &SigningKey,
    ns: &str,
    state: State,
    paid_until: u64,
    source: &str,
    issued: u64,
) -> EntitlementFact {
    let mut fact = monetize_embed::EntitlementFact {
        tenant: TenantId(ns.to_owned()),
        plan: "10gb".into(),
        state,
        paid_until_unix_ms: Some(paid_until),
        caps: BTreeMap::new(),
        source: source.into(),
        signature: vec![],
        issued_unix_ms: Some(issued),
        issued_signature: vec![],
    };
    fact.signature = key
        .sign(&monetize_embed::signing::fact_message(&fact))
        .to_bytes()
        .to_vec();
    fact.issued_signature = key
        .sign(&monetize_embed::signing::fact_message_issued(&fact))
        .to_bytes()
        .to_vec();
    EntitlementFact {
        namespace: fact.tenant.0,
        plan: fact.plan,
        state: match state {
            State::Free => PbState::Free,
            State::Paid => PbState::Paid,
            State::Grace => PbState::Grace,
            State::Suspended => PbState::Suspended,
            State::Retention => PbState::Retention,
        } as i32,
        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,
    }
}

/// **c-stale-suspend.** monetize restored from an OLDER backup (its front laid
/// again) lost a renewal the box holds: its poller re-signs the older ledger's
/// verdict at the current time — Suspended, paid only until the old date, from
/// an older payment — and the replay guard, which compares only issue times,
/// took it: a customer who PAID was suspended. The box refuses a newer-stamped
/// fact that takes back paid time it holds, from a payment other than the one it
/// holds (the refusal reads as "already holds", so monetize's re-push files it
/// as held, not as a refused cap). A downgrade about the SAME payment (a
/// refund) and a lapse (the held paid time is over) still go through.
#[test]
fn a_restored_older_ledger_cannot_suspend_a_tenant_whose_renewal_the_box_holds() {
    let rig = Rig::new();
    let slot = rig.slot();
    let plane = rig.plane(&slot);
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_millis() as u64;
    let day = 24 * 3600 * 1000;
    plane
        .entitlement_set(fact_paid(
            &rig.key,
            "team",
            State::Paid,
            now + 60 * day,
            "payment:stripe:renewal-2",
            ISSUED_AT + 10,
        ))
        .expect("the renewal is taken");

    let stale = fact_paid(
        &rig.key,
        "team",
        State::Suspended,
        now - day,
        "payment:stripe:first",
        ISSUED_AT + 20,
    );
    let refused = plane
        .entitlement_set(stale)
        .expect_err("a restored ledger that lost the renewal must not suspend a tenant who paid");
    let words = format!("{refused}");
    assert!(
        words.contains("already holds") && words.contains("team"),
        "{words}"
    );
    assert_eq!(
        slot.enforced_state("team"),
        Some(gunnar_core::EntitlementState::Paid),
        "still paid"
    );

    // A refund of the payment the box holds is a real downgrade.
    plane
        .entitlement_set(fact_paid(
            &rig.key,
            "team",
            State::Suspended,
            now - day,
            "payment:stripe:renewal-2",
            ISSUED_AT + 30,
        ))
        .expect("a downgrade about the held payment goes through");
    assert_eq!(
        slot.enforced_state("team"),
        Some(gunnar_core::EntitlementState::Suspended)
    );

    // A lapse: the held paid time is over, any later verdict goes through.
    let other = rig.slot();
    let p2 = rig.plane(&other);
    p2.entitlement_set(fact_paid(
        &rig.key,
        "lapsed",
        State::Paid,
        now - 2 * day,
        "payment:stripe:a",
        ISSUED_AT + 40,
    ))
    .unwrap();
    p2.entitlement_set(fact_paid(
        &rig.key,
        "lapsed",
        State::Suspended,
        now - 2 * day,
        "payment:stripe:older",
        ISSUED_AT + 50,
    ))
    .expect("the held paid time is over: a lapse is no regression");
}