alkblobs 0.1.0

//! The pin-token layer + the ops context spine (`ops/mod.rs`'s
//! placement note): the token is the pin's ops-layer handle.

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

use parking_lot::Mutex;

use super::fetch::FanoutRegistry;
use crate::facade::Pin;
use crate::keys::Key;
use crate::store::Store;
use crate::Error;

/// The pin token (ADR-010 §4; ADR-012 §2): the ops layer's opaque,
/// unguessable handle over a landed put's pin. 32 bytes of OS-entropy
/// CSPRNG output — the minting-form design point's decided shape (see
/// the module doc); no key, owner, or namespace bytes ride it. The
/// validity domain is the minting node: a peer's registry does not
/// know the token and answers as unknown, present: false (not an
/// error). Wire form: hex (ADR-012 §2's `{token, digest}` shapes).
///
/// `Debug` is hand-written and masks the bytes: the token is a
/// capability — it must not land in logs, panics, or error payloads.
#[derive(Clone, Copy, PartialEq, Eq, Hash)]
pub struct PinToken {
    bytes: [u8; 32],
}

impl PinToken {
    /// Mint a fresh token: 32 bytes from the OS entropy source (the
    /// CSPRNG design point). Fallible only on the entropy source's
    /// failure — an operational condition, the `Io` family.
    pub fn generate() -> Result<PinToken, Error> {
        let mut bytes = [0u8; 32];
        getrandom::fill(&mut bytes)
            .map_err(|e| Error::Io(format!("pin-token minting: the entropy source failed: {e}")))?;
        Ok(PinToken { bytes })
    }

    /// The token's raw bytes (the registry's map key; the renewal
    /// lookup's identity).
    pub fn as_bytes(&self) -> &[u8; 32] {
        &self.bytes
    }

    /// The wire form (lowercase hex, 64 chars).
    pub fn to_hex(&self) -> String {
        hex::encode(self.bytes)
    }

    /// The wire form's ingress: lowercase canonical, decode
    /// case-insensitive; `None` on wrong length or non-hex bytes — a
    /// malformed token is an unknown token (present: false, not an
    /// error), which is the have handler's mapping if it wants it.
    ///
    /// The length check runs **before** the decode (review-001 S-5):
    /// a wrong-length input answers `None` with no proportional
    /// allocation, never an arbitrarily large hex-decode ahead of the
    /// `try_into`'s rejection. The wire schema's bound
    /// (`super::wire::token_wire_schema`, the crate-internal
    /// `maxLength` 64 + the case-insensitive hex pattern) gates the
    /// have renewal form's field to the same length/form ahead of
    /// this decode; this guard is the ingress's own for direct
    /// embedder use.
    pub fn from_hex(token_hex: &str) -> Option<PinToken> {
        if token_hex.len() != 64 {
            return None;
        }
        let bytes = hex::decode(token_hex).ok()?;
        bytes.try_into().ok().map(|bytes| PinToken { bytes })
    }
}

impl std::fmt::Debug for PinToken {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("PinToken(#redacted)")
    }
}

/// The renewal probe's answer vocabulary (the have handler's renewal
/// form, ADR-012 §2): every variant but [`RenewalOutcome::Renewed`]
/// answers `present: false` on the wire, and only `Renewed` carries a
/// renewal. The putter's backstop for every non-renewed answer is the
/// re-put (delete-then-recover, ADR-005) — CAS makes it idempotent.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RenewalOutcome {
    /// The token mapped, the row was live, and the renewal landed —
    /// `present: true`.
    Renewed,
    /// The registry does not know the token: the validity-domain
    /// refusal — a foreign/unminted token, a mapping already retired
    /// at conversion, or a presented namespace that differs from the
    /// token's mint record (ADR-017 §3's namespace parity — the same
    /// "not this token's domain" answer confirm's refusal is).
    /// `present: false`; no renewal occurred; the mapping is intact.
    UnknownToken,
    /// The token mapped, but to a different digest than the one
    /// presented: one token binds one pin row. `present: false`; no
    /// renewal occurred.
    DigestMismatch,
    /// The token mapped, but the pin row is not live (expired, raced,
    /// unrecoverable — the engine's never-resurrect discipline,
    /// ADR-012 §2). `present: false`; no renewal occurred.
    Expired,
}

impl RenewalOutcome {
    /// The have wire answer (`present[]`'s element): `true` only for
    /// [`RenewalOutcome::Renewed`].
    pub fn present(&self) -> bool {
        matches!(self, RenewalOutcome::Renewed)
    }
}

/// The embedder's per-namespace converter — the binding the embedder
/// plugs its registry into (ops-surface.md's hand-over contract step
/// 3): the conversion's registration act, invoked by the ops layer's
/// [`OpsContext::confirm`] verb.
///
/// The act is sync and must be quick, indexed, non-blocking: in the
/// guarded (non-fleet) form it runs **under the store's arbitration
/// lock** (the ordered-step conversion's
/// one-atomic-section discipline — the pin's release and this
/// registration never split). Registering a slow or async act here
/// would block every arbitration in the store; that shape belongs
/// above the facade, not in a conversion.
pub trait PinConverter: Send + Sync + 'static {
    /// The registration act for `key`: the embedder registers the
    /// digest into its own registry (whatever shape that is), making
    /// the entry live per the source ADR-005's registered-liveness
    /// seam consults. The token's mapping retires at this act's
    /// confirmation — conversion-confirmed is the retention
    /// discipline's first retirement arm.
    fn convert(&self, key: &Key);
}

/// The token registry's entry: the minting metadata (the namespace
/// the put arrived under — the confirm verb's converter-routing
/// key) and the custody branch.
struct PinTicket {
    namespace: String,
    key: Key,
    custody: PinCustody,
}

/// The custody branch — the token layer's core split (ADR-010 §4,
/// the module doc's section): the fleet form holds only the mapping
/// (the pin row's expiry is the protection; the guard dropped at
/// response completion), the non-fleet form holds the guard.
enum PinCustody {
    /// Fleet: the mapping only — `owner` is the serving store
    /// instance's id, validated against the bound store at minting.
    Fleet { owner: Vec<u8> },
    /// Non-fleet: the registry holds the put's guard — nothing
    /// expires — until the conversion consumes it.
    Held(Pin),
}

/// A renewal's custody snapshot — the mapping peek (the guards are not
/// cloneable; the branch is carried as its shape).
enum RenewalCustody {
    Held,
    Fleet { owner: Vec<u8> },
}

struct RenewalTarget {
    namespace: String,
    key: Key,
    custody: RenewalCustody,
}

/// The token registry — the token↔(key, owner) in-memory mapping with
/// process lifetime (ADR-010 §4; ADR-012 §2). Per-process state: the
/// validity domain is the minting node; a peer registry's lookup
/// misses (`present: false`, not an error). Entries retire at
/// conversion-confirmed or process end — no expiry machinery of its
/// own (the pin row's expiry is the fleet liveness signal; nothing
/// expires non-fleet).
///
/// **The admission posture (ADR-019, resolving OQ-16 — review-002
/// C-2): the registry admits up to its cap and refuses mints beyond
/// it — never evicting.** The cap measures live entries (both custody
/// forms); the knob is [`OpsConfig::token_registry_max_entries`]
/// (`usize::MAX` = uncapped, for an embedder that has bounded the
/// loop's reach above the store; `with_cap`/`new` for the standalone
/// registry). A mint at cap answers [`crate::Error::RegistryFull`] —
/// nothing inserted, a custodial guard dropped, no custody taken.
/// The attack shape the cap answers (the non-confirming upload loop:
/// a hostile client streaming conforming content and abandoning the
/// confirm step each iteration — unbounded registry growth plus
/// never-expiring pin custody on a non-fleet store) is on record in
/// ADR-019 and the module doc; the mint policy at the deployer's door
/// remains the embedder's responsibility, and the fleet form (the pin
/// row's TTL) is the safe posture for a door that cannot prove
/// confirm.
///
/// Constructed standalone it is the state holder; the
/// [`OpsContext`] verbs are the custody-validated shapes (the
/// context enforces the mint's pairing against the bound store's
/// fleet declaration — the registry's own verbs do not, which is why
/// the handlers mint only through the context).
pub struct TokenRegistry {
    tickets: Mutex<BTreeMap<[u8; 32], PinTicket>>,
    max_entries: usize,
}

impl Default for TokenRegistry {
    fn default() -> Self {
        Self::new()
    }
}

impl TokenRegistry {
    /// The default admission cap (ADR-019): generous above honest
    /// concurrently-unconfirmed traffic (tokens retire at confirm)
    /// while bounding the unconfirming loop's registry growth; tune
    /// with [`with_cap`](Self::with_cap) or
    /// [`OpsConfig::token_registry_max_entries`].
    pub const DEFAULT_CAP: usize = 65_536;

    /// A fresh, empty registry — process-local bookkeeping, admitting
    /// up to [`Self::DEFAULT_CAP`] entries (ADR-019). The explicit
    /// cap is [`Self::with_cap`].
    pub fn new() -> TokenRegistry {
        TokenRegistry::with_cap(Self::DEFAULT_CAP)
    }

    /// A fresh, empty registry with an explicit admission cap (ADR-019):
    /// mints beyond `max_entries` live entries are refused
    /// ([`crate::Error::RegistryFull`], never evicted);
    /// `usize::MAX` is the documented uncapped form — an embedder that
    /// has bounded the unconfirming loop's reach above the store (its
    /// own admission/rate policy at the door) may want it.
    pub fn with_cap(max_entries: usize) -> TokenRegistry {
        TokenRegistry {
            tickets: Mutex::new(BTreeMap::new()),
            max_entries,
        }
    }

    /// The live-entry count — the admission cap's measure (both
    /// custody forms; entries retire at conversion-confirmed or
    /// process end). The embedder's monitoring window on proximity to
    /// the cap ([`OpsConfig::token_registry_max_entries`]).
    pub fn len(&self) -> usize {
        self.tickets.lock().len()
    }

    /// Whether the registry holds no live entries.
    pub fn is_empty(&self) -> bool {
        self.tickets.lock().is_empty()
    }

    /// The insert-side mint: the admission cap's check and the insert
    /// share one lock section (a refusal inserts nothing and leaves
    /// the count exact); a refused custodial custody's guard drops at
    /// this function's return — no custody taken.
    fn insert_ticket(
        &self,
        namespace: String,
        key: Key,
        custody: PinCustody,
    ) -> Result<PinToken, Error> {
        let mut tickets = self.tickets.lock();
        if tickets.len() >= self.max_entries {
            return Err(Error::RegistryFull);
        }
        let token = PinToken::generate()?;
        let previous = tickets.insert(
            *token.as_bytes(),
            PinTicket {
                namespace,
                key,
                custody,
            },
        );
        drop(tickets);
        debug_assert!(previous.is_none(), "the CSPRNG minted a colliding token");
        Ok(token)
    }

    /// Mint a token over the (key, owner) mapping — the fleet form's
    /// registry act: the entry is mapping-only, no guard is held. The
    /// custody-validated shape is [`OpsContext::mint_fleet`] (the
    /// fleet declaration and the owner's instance check ride there).
    /// Refused ([`crate::Error::RegistryFull`]) at the registry's
    /// admission cap (ADR-019): no token, no mapping.
    pub fn mint_fleet(&self, namespace: &str, key: Key, owner: Vec<u8>) -> Result<PinToken, Error> {
        self.insert_ticket(namespace.to_owned(), key, PinCustody::Fleet { owner })
    }

    /// Mint a token and take custody of the put's guard — the
    /// non-fleet form: the token is the in-process guard's handle,
    /// nothing expires, until the conversion consumes the guard. The
    /// custody-validated shape is [`OpsContext::mint_custodial`] (the
    /// fleet-store refusal rides there). Refused
    /// ([`crate::Error::RegistryFull`]) at the registry's admission
    /// cap (ADR-019): no token, no mapping, the guard dropped — no
    /// custody taken, the entry stands as unreferenced content a
    /// sweep may reclaim.
    pub fn mint_custodial(&self, namespace: &str, key: Key, guard: Pin) -> Result<PinToken, Error> {
        self.insert_ticket(namespace.to_owned(), key, PinCustody::Held(guard))
    }

    /// Whether the token maps right now (the validity-domain
    /// registry's lookup — a peer's token misses).
    pub fn holds(&self, token: &PinToken) -> bool {
        self.tickets.lock().contains_key(token.as_bytes())
    }
}

/// The module's configuration — what an embedder tunes; the handlers
/// capture the context, not the knobs.
///
/// The fetch ranged form's bounds ([`OpsConfig::fetch_max_ranges`],
/// [`OpsConfig::fetch_max_total_bytes`]) are this config's knobs for
/// the same reason the fanout's bounds are: the schemas the context's
/// handlers build ride the per-context values — for the ranged form,
/// the `maxItems` bound lands in the wire schema itself (an over-count
/// request is the registry's `INVALID_INPUT`) while the total-bytes
/// bound is the establisher's ingress check (JSON Schema cannot sum).
#[derive(Debug, Clone)]
pub struct OpsConfig {
    /// The fanout's per-digest subscriber-arm bound
    /// ([`FanoutRegistry::new`]).
    pub fanout_max_subscribers: usize,
    /// The fanout's per-arm channel bound — the backpressure
    /// discipline's knob (the POC's 8).
    pub fanout_channel_bound: usize,
    /// The `blobs/fetch` ranged form's range-count bound: the ranges
    /// schema's `maxItems` — an over-count open is rejected at the
    /// registry's schema gate (`INVALID_INPUT`), never the handler.
    pub fetch_max_ranges: usize,
    /// The `blobs/fetch` ranged form's total-bytes bound: the sum of
    /// the **requested** lens (the ingress sees the wire values; the
    /// entry's length is not known there — clamping stays per-range at
    /// the store seam). Enforced at the establisher's ranges ingress;
    /// an over-bytes request is the establishment
    /// `HandlerError` refusal (stable domain text).
    pub fetch_max_total_bytes: u64,
    /// The token registry's admission cap (ADR-019): the live-entry
    /// bound (both custody forms) mints are refused past — a mint at
    /// cap answers [`crate::Error::RegistryFull`] (no token, no
    /// custody, never an eviction), and the `blobs/put` handler maps
    /// the refusal to the declared `MINT_REFUSED` wire error. The
    /// attack shape it bounds and the embedder duty it complements
    /// (the door's mint policy; the fleet TTL as the safe posture for
    /// a door that cannot prove confirm) are the module doc's
    /// retention-posture record and ADR-019.
    pub token_registry_max_entries: usize,
}

impl Default for OpsConfig {
    fn default() -> OpsConfig {
        OpsConfig {
            fanout_max_subscribers: 64,
            fanout_channel_bound: 8,
            fetch_max_ranges: 256,
            fetch_max_total_bytes: 64 * 1024 * 1024,
            token_registry_max_entries: TokenRegistry::DEFAULT_CAP,
        }
    }
}

/// The ops-layer context — the module's state bound to one store
/// (ADR-001 §3's placement; the substrate this crate's ops surface
/// rides): the handler pairs (later tasks) capture this, the module
/// config, the token registry, the namespace converters, and the
/// fanout registry.
///
/// The store access is the public facade plus the crate-internal
/// renewal seam (`Store::ops_pin_renew`/`Store::ops_pin_check`)
/// — no backend side-door, no engine types, no public-API widening.
pub struct OpsContext {
    store: Arc<Store>,
    config: OpsConfig,
    tokens: TokenRegistry,
    converters: Mutex<BTreeMap<String, Arc<dyn PinConverter>>>,
    fanout: FanoutRegistry,
}

impl OpsContext {
    /// Bind the substrate to a store instance. The store rides as
    /// `Arc<Store>` (the facade is `&self`-shaped over async ops); this
    /// context is what the handler pairs capture. The token registry's
    /// admission cap is the config's
    /// [`OpsConfig::token_registry_max_entries`].
    pub fn new(store: Arc<Store>, config: OpsConfig) -> OpsContext {
        let fanout =
            FanoutRegistry::new(config.fanout_max_subscribers, config.fanout_channel_bound);
        let tokens = TokenRegistry::with_cap(config.token_registry_max_entries);
        OpsContext {
            store,
            config,
            tokens,
            converters: Mutex::new(BTreeMap::new()),
            fanout,
        }
    }

    /// The module config the context was built with — the knobs the
    /// ops schemas and handlers ride (the fanout's bounds, the fetch
    /// ranged form's bounds).
    pub fn config(&self) -> &OpsConfig {
        &self.config
    }

    /// The bound store — the facade access (get/put/stat/read_range/
    /// has/list + the GC protocol verbs; the ADR-001 §4 consumer
    /// posture).
    pub fn store(&self) -> &Arc<Store> {
        &self.store
    }

    /// The token registry ([`OpsContext::renew`]/[`OpsContext::confirm`]
    /// act on it; the handlers mint through the context verbs).
    pub fn tokens(&self) -> &TokenRegistry {
        &self.tokens
    }

    /// The fanout registry (the producer/consumer wiring of
    /// ops-put/ops-fetch rides it; the mechanics are
    /// registry-level).
    pub fn fanout(&self) -> &FanoutRegistry {
        &self.fanout
    }

    /// The bound store's fleet declaration's instance id — the
    /// `blobs/put` handler's `owner` half for [`OpsContext::mint_fleet`];
    /// `None` on every non-fleet store (where minting is custodial).
    pub fn fleet_owner(&self) -> Option<Vec<u8>> {
        self.store.fleet().map(|fleet| fleet.owner().to_vec())
    }

    /// Bind or re-bind the namespace's converter — the embedder's
    /// registration of the act its registry performs at conversion.
    /// `true` when a previous binding was replaced. The binding is
    /// checked at conversion time: a confirm whose namespace is
    /// unbound is refused (`false`) and the token's mapping survives.
    pub fn bind_converter(&self, namespace: &str, converter: Arc<dyn PinConverter>) -> bool {
        self.converters
            .lock()
            .insert(namespace.to_owned(), converter)
            .is_some()
    }

    /// **Fleet minting**: the custody-validated fleet form — the
    /// mapping only (`owner` must be the bound store's instance id —
    /// a foreign owner is refused at the mint, never trusted).
    /// Refused on a non-fleet store: the pin row's expiry is the
    /// fleet protection, and a guard-free entry over an in-process
    /// pin would be protection-less. Refused with
    /// [`crate::Error::RegistryFull`] at the registry's admission
    /// cap (ADR-019): no token, no mapping — the committed row stands
    /// and expires by its TTL.
    ///
    /// The put handler's guard is not passed here on purpose: its
    /// **drop at response completion** is the handler's own act
    /// (ADR-010 §4 — the TTL machinery stays alive on the ops path;
    /// the token's whole-life guard custody is the non-fleet form,
    /// and holding it under fleet would dead-code the TTL machinery).
    pub fn mint_fleet(&self, namespace: &str, key: Key, owner: &[u8]) -> Result<PinToken, Error> {
        let Some(fleet) = self.store.fleet() else {
            return Err(Error::Io(
                "fleet-token minting refused: the bound store is not fleet-declared \
                 (ADR-012 §2) — a non-fleet serving node's put mints the custodial form, \
                 whose guard the registry holds"
                    .to_owned(),
            ));
        };
        if fleet.owner() != owner {
            return Err(Error::Io(
                "fleet-token minting refused: the owner is not the bound store's \
                 instance id (ADR-011's vocabulary: the pin row's identity half is \
                 the minting instance)"
                    .to_owned(),
            ));
        }
        self.tokens.mint_fleet(namespace, key, owner.to_vec())
    }

    /// **Custodial minting**: the non-fleet form (any engine;
    /// ADR-012 §2's generalization) — the registry **takes custody**
    /// of the put's guard: the token is the in-process guard's
    /// handle, nothing expires, until the conversion consumes it
    /// (conversion-confirmed) or the process ends. Refused on a
    /// declared fleet store, whose put path keeps the guard for
    /// response completion only and whose protection is the row's
    /// expiry — a custodially-held guard under fleet would hold the
    /// TTL machinery dead exactly as ADR-010 §4 rejected. Refused
    /// with [`crate::Error::RegistryFull`] at the registry's
    /// admission cap (ADR-019): no token, no custody, the guard
    /// dropped — the entry stands as unreferenced content a sweep
    /// may reclaim.
    pub fn mint_custodial(&self, namespace: &str, key: Key, guard: Pin) -> Result<PinToken, Error> {
        if self.store.fleet().is_some() {
            return Err(Error::Io(
                "custodial-token minting refused: the bound store is fleet-declared \
                 (ADR-012 §2) — the fleet put path's guard drops at response completion, \
                 the pin row's expiry is the protection, and the mint is the fleet form"
                    .to_owned(),
            ));
        }
        self.tokens.mint_custodial(namespace, key, guard)
    }

    /// **The renewal probe's substrate** (the have handler's renewal
    /// form, ADR-012 §2's verify-the-mapping-then-renew duty — the
    /// module doc's decided composition): token lookup (the validity
    /// domain), the namespace parity check (ADR-017 §3 — the mint
    /// records the namespace the put arrived under, and confirm's
    /// refusal discipline is the parity's contrast: the presented
    /// namespace must match the mint record), the one-token-one-row
    /// digest check, and — fleet — the row's classification then the
    /// renewal through the store core; non-fleet — the answer from
    /// custody.
    ///
    /// Every non-renewed outcome answers the wire as `present: false`
    /// with no renewal occurred; the putter's backstop is the re-put.
    pub async fn renew(&self, namespace: &str, token: &PinToken, digest: &Key) -> RenewalOutcome {
        let Some(target) = self.renewal_target(token) else {
            return RenewalOutcome::UnknownToken;
        };
        if target.namespace != namespace {
            return RenewalOutcome::UnknownToken;
        }
        if &target.key != digest {
            return RenewalOutcome::DigestMismatch;
        }
        match target.custody {
            RenewalCustody::Held => RenewalOutcome::Renewed,
            RenewalCustody::Fleet { owner } => self.renew_fleet(&target.key, &owner).await,
        }
    }

    /// The fleet arm's composition: the store core's
    /// `ops_pin_check` first (the row-classification verify — no
    /// renewal write on a row not live), then `ops_pin_renew` whose
    /// never-resurrect bool is the final answer (a check/renew race
    /// lands on its `false` the same way). The ops renewal access is
    /// the store core's; here the outcome only folds.
    #[cfg(feature = "postgres")]
    async fn renew_fleet(&self, key: &Key, owner: &[u8]) -> RenewalOutcome {
        use crate::facade::PinState;

        match self.store.ops_pin_check(&key.key_bytes()).await {
            Ok(PinState::Protected) => {}
            Ok(PinState::Unprotected) => return RenewalOutcome::Expired,
            Err(_) => return RenewalOutcome::Expired,
        }
        match self.store.ops_pin_renew(&key.key_bytes(), owner).await {
            Ok(true) => RenewalOutcome::Renewed,
            Ok(false) | Err(_) => RenewalOutcome::Expired,
        }
    }

    #[cfg(not(feature = "postgres"))]
    async fn renew_fleet(&self, key: &Key, owner: &[u8]) -> RenewalOutcome {
        let _ = (key, owner);
        RenewalOutcome::Expired
    }

    /// The registry peek the renewal rides: the mapping's key and the
    /// custody branch, cloned out under the lock (never held across
    /// the fleet arm's engine awaits; the guards themselves are not
    /// cloneable — the custody is carried as the branch's shape).
    fn renewal_target(&self, token: &PinToken) -> Option<RenewalTarget> {
        self.tokens
            .tickets
            .lock()
            .get(token.as_bytes())
            .map(|ticket| RenewalTarget {
                namespace: ticket.namespace.clone(),
                key: ticket.key.clone(),
                custody: match &ticket.custody {
                    PinCustody::Held(_) => RenewalCustody::Held,
                    PinCustody::Fleet { owner } => RenewalCustody::Fleet {
                        owner: owner.clone(),
                    },
                },
            })
    }

    /// **The conversion trigger's verb** (the embedder-called confirm;
    /// the module doc's decided design point): given the namespace and
    /// token, invoke the namespace's converter with the mapping's key —
    /// the guarded form through the ordered-step conversion (the
    /// guard's release and the registration are one atomic step), the
    /// fleet form as the registration act itself — and retire the
    /// mapping (conversion-confirmed's retirement arm).
    ///
    /// Refusals (`false`), mapping intact: an unknown/retired token, a
    /// namespace that does not match the minting namespace (the mint
    /// records it), or an unbound converter. The mapping retires only
    /// on a successful conversion.
    ///
    /// The fleet conversion retires the token *at* the confirmation:
    /// from then on the entry's protection is the registered liveness
    /// source, a renewal probe for the token answers `present: false`
    /// (the putter's signal to stop renewing; a continued probe's
    /// backstop — the re-put — is CAS-idempotent and harmless).
    pub fn confirm(&self, namespace: &str, token: &PinToken) -> bool {
        let ticket = {
            let mut tickets = self.tokens.tickets.lock();
            match tickets.remove(token.as_bytes()) {
                Some(ticket) => ticket,
                None => return false,
            }
        };
        if ticket.namespace != namespace {
            self.tokens.tickets.lock().insert(*token.as_bytes(), ticket);
            return false;
        }
        let converter = {
            let converters = self.converters.lock();
            converters.get(&ticket.namespace).cloned()
        };
        let Some(converter) = converter else {
            self.tokens.tickets.lock().insert(*token.as_bytes(), ticket);
            return false;
        };
        match ticket.custody {
            PinCustody::Held(pin) => self
                .store
                .gc()
                .convert(pin, || converter.convert(&ticket.key)),
            PinCustody::Fleet { .. } => converter.convert(&ticket.key),
        }
        true
    }
}
#[cfg(test)]
mod tests {
    use super::*;
    use crate::ops::tests::{key_of, test_store, RecordingConverter};

    /// The CSPRNG minting-form's shape (the design point): 32 opaque
    /// bytes, unique per mint, wire form hex, malformed hex ingress
    /// rejected as not-a-token (never an error), and the Debug mask
    /// keeps the capability out of rendered output.
    #[test]
    fn token_mints_unguessable_and_the_hex_wire_form_round_trips() {
        let first = PinToken::generate().unwrap();
        let second = PinToken::generate().unwrap();
        assert_ne!(
            first, second,
            "CSPRNG minting never collides in the same process"
        );
        assert_eq!(first.as_bytes().len(), 32);

        let token_hex = first.to_hex();
        assert_eq!(token_hex.len(), 64);
        assert!(token_hex
            .chars()
            .all(|c| c.is_ascii_digit() || ('a'..='f').contains(&c)));
        assert_eq!(PinToken::from_hex(&token_hex), Some(first));
        assert_eq!(PinToken::from_hex(&token_hex.to_uppercase()), Some(first));

        let wrong_length = format!("{token_hex}ab");
        assert_eq!(PinToken::from_hex("0"), None);
        assert_eq!(PinToken::from_hex("non-hexadecimal-material!!"), None);
        assert_eq!(PinToken::from_hex(&wrong_length), None);

        assert_eq!(format!("{first:?}"), "PinToken(#redacted)");
        assert!(!format!("{first:?}").contains(&token_hex));
    }

    /// S-5's length-first ingress: a wrong-length token answers
    /// `None` before any decode — a huge input never allocates
    /// proportionally (the order itself is inspection-verified; the
    /// test asserts the rejection on the huge form and the near-miss
    /// lengths).
    #[test]
    fn from_hex_rejects_wrong_length_input_before_decoding() {
        assert_eq!(
            PinToken::from_hex(&"a".repeat(1 << 20)),
            None,
            "a huge string is rejected at the length guard"
        );
        assert_eq!(PinToken::from_hex(&"a".repeat(65)), None);
        assert_eq!(PinToken::from_hex(&"a".repeat(63)), None);
        assert_eq!(PinToken::from_hex(""), None);
    }

    /// The custodial custody split end to end (ADR-012 §2's non-fleet
    /// form): the registry holds the guard, the renewal answers from
    /// custody (nothing expires), the one-token-one-row digest check,
    /// the refusals keep the mapping intact, and the confirm verb
    /// converts through the ordered-step primitive — the guard's
    /// release and the converter's registration — and retires the
    /// mapping.
    #[cfg(any(feature = "mem", feature = "sqlite"))]
    #[tokio::test]
    async fn custodial_custody_holds_the_guard_and_converts_at_the_confirmation() {
        let store = test_store().await;
        let ctx = OpsContext::new(Arc::clone(&store), OpsConfig::default());
        let key = key_of(b"custodial");
        let other = key_of(b"other-digest");
        let unknown = PinToken::from_hex(&"0".repeat(64)).unwrap();

        let pin = store.gc().acquire(&key).await;
        let token = ctx.mint_custodial("ns", key.clone(), pin).unwrap();

        assert_eq!(
            store.gc().pin_count(&key),
            1,
            "the registry holds the guard"
        );
        assert!(ctx.tokens().holds(&token));

        assert_eq!(ctx.renew("ns", &token, &key).await, RenewalOutcome::Renewed);
        assert!(ctx.renew("ns", &token, &key).await.present());
        assert_eq!(
            ctx.renew("ns", &token, &other).await,
            RenewalOutcome::DigestMismatch
        );
        assert_eq!(
            ctx.renew("ns", &unknown, &key).await,
            RenewalOutcome::UnknownToken
        );
        assert_eq!(
            store.gc().pin_count(&key),
            1,
            "the refusals left the custody intact"
        );

        assert_eq!(
            ctx.renew("other-namespace", &token, &key).await,
            RenewalOutcome::UnknownToken,
            "the namespace parity (ADR-017 §3): a namespace-mismatched renewal is the \
             validity-domain refusal — the same 'not this token's domain' answer confirm's \
             refusal is"
        );
        assert!(
            ctx.tokens().holds(&token),
            "the parity refusal kept the mapping intact"
        );
        assert!(
            ctx.renew("ns", &token, &key).await.present(),
            "a subsequent correct-namespace renewal succeeds (the parity check retires nothing)"
        );
        assert_eq!(
            store.gc().pin_count(&key),
            1,
            "the parity refusals left the custody intact"
        );

        assert!(
            !ctx.confirm("other-namespace", &token),
            "the minting namespace mismatches"
        );
        assert!(
            ctx.tokens().holds(&token),
            "the refused confirm kept the mapping"
        );

        assert!(
            !ctx.confirm("ns", &token),
            "an unbound converter refuses, mapping intact"
        );
        assert!(ctx.tokens().holds(&token));

        let converted = Arc::new(RecordingConverter {
            converted: Mutex::new(Vec::new()),
        });
        assert!(!ctx.bind_converter("ns", Arc::clone(&converted) as Arc<dyn PinConverter>));
        assert!(ctx.confirm("ns", &token));
        assert_eq!(
            converted.captured(),
            vec![key.clone()],
            "the converter received the mapping's key"
        );
        assert_eq!(
            store.gc().pin_count(&key),
            0,
            "the ordered-step conversion released the held guard"
        );
        assert!(
            !ctx.tokens().holds(&token),
            "conversion-confirmed retires the mapping"
        );
        assert_eq!(
            ctx.renew("ns", &token, &key).await,
            RenewalOutcome::UnknownToken,
            "a retired token's renewal probe is the validity-domain refusal"
        );
        assert!(
            !ctx.confirm("ns", &token),
            "a retired token confirms nowhere twice"
        );
    }

    /// The fleet custody's registry-level form (ADR-010 §4's mapping
    /// only): no guard is held (the put handler's guard drops at
    /// response completion — its act, outside the mint), the renewal
    /// probe on this shape degrades closed (the pin-row access is the
    /// fleet store core's; on a non-fleet bound store the answer is
    /// present: false, never an error), and conversion retires the
    /// mapping through the converter act alone.
    #[cfg(any(feature = "mem", feature = "sqlite"))]
    #[tokio::test]
    async fn fleet_custody_holds_no_guard_and_the_renewal_degrades_closed() {
        let store = test_store().await;
        let ctx = OpsContext::new(Arc::clone(&store), OpsConfig::default());
        let key = key_of(b"fleet-form");
        let other = key_of(b"other-fleet");

        let token = ctx
            .tokens()
            .mint_fleet("ns", key.clone(), b"instance-owner".to_vec())
            .unwrap();
        assert_eq!(
            store.gc().pin_count(&key),
            0,
            "the fleet form records the mapping; no guard rides it"
        );

        assert_eq!(
            ctx.renew("ns", &token, &other).await,
            RenewalOutcome::DigestMismatch
        );
        assert_eq!(
            ctx.renew("ns", &token, &key).await,
            RenewalOutcome::Expired,
            "a fleet mapping's renewal reaches the store core's fleet host and, on a \
             non-fleet-bound context shape, folds closed"
        );

        let converted = Arc::new(RecordingConverter {
            converted: Mutex::new(Vec::new()),
        });
        ctx.bind_converter("ns", Arc::clone(&converted) as Arc<dyn PinConverter>);
        assert!(ctx.confirm("ns", &token));
        assert_eq!(
            converted.captured(),
            vec![key.clone()],
            "the fleet conversion is the converter act itself (no guard to release)"
        );
        assert!(
            !ctx.tokens().holds(&token),
            "conversion-confirmed retires the mapping"
        );
    }

    /// The custody-mint validations (ADR-010 §4's total split, the
    /// constructible half): the fleet form is refused over a
    /// non-fleet-declared store — the pin row's expiry cannot be the
    /// protection for a mapping-only token where no row exists.
    #[cfg(any(feature = "mem", feature = "sqlite"))]
    #[tokio::test]
    async fn mint_fleet_is_refused_over_a_non_fleet_store() {
        let store = test_store().await;
        let ctx = OpsContext::new(Arc::clone(&store), OpsConfig::default());
        assert!(ctx
            .mint_fleet("ns", key_of(b"attempted"), b"owner")
            .is_err());
    }

    /// The fleet-mint owner validation sits at the mint: a foreign
    /// owner is refused before any mapping lands (the pin row's
    /// identity half is the minting instance's id).
    #[test]
    fn fleet_owner_lookup_follows_the_bound_store_and_the_registry_mints_standalone() {
        let tokens = TokenRegistry::new();
        let without = PinToken::from_hex(&"7".repeat(64)).unwrap();
        assert!(!tokens.holds(&without));
        let minted = tokens
            .mint_fleet("ns", key_of(b"standalone"), b"instance".to_vec())
            .unwrap();
        assert!(tokens.holds(&minted));
        assert_ne!(minted, without);
    }

    /// The converter binding is per-namespace, and a re-bind replaces:
    /// the latest bound converter is the one the conversion invokes.
    #[cfg(any(feature = "mem", feature = "sqlite"))]
    #[tokio::test]
    async fn converter_bindings_are_per_namespace_and_rebinding_replaces() {
        let store = test_store().await;
        let ctx = OpsContext::new(Arc::clone(&store), OpsConfig::default());
        let first = Arc::new(RecordingConverter {
            converted: Mutex::new(Vec::new()),
        });
        let second = Arc::new(RecordingConverter {
            converted: Mutex::new(Vec::new()),
        });
        assert!(!ctx.bind_converter("ns", Arc::clone(&first) as Arc<dyn PinConverter>));
        assert!(ctx.bind_converter("ns", Arc::clone(&second) as Arc<dyn PinConverter>));

        let pin = store.gc().acquire(&key_of(b"rebind")).await;
        let token = ctx.mint_custodial("ns", key_of(b"rebind"), pin).unwrap();
        assert!(ctx.confirm("ns", &token));
        assert!(
            first.captured().is_empty(),
            "the replaced binding never converted"
        );
        assert_eq!(second.captured(), vec![key_of(b"rebind")]);
    }

    /// ADR-019's admission cap, reproduced from POC-L's recorded
    /// attack shape: the non-confirming upload loop mints a held pin
    /// and a registry entry per iteration; at the cap the mint
    /// refuses — never evicts — and a refused custodial mint drops
    /// its guard (no custody taken, nothing registered). Both mint
    /// forms sit under the one cap, and the earlier tokens' custody
    /// is untouched.
    #[cfg(any(feature = "mem", feature = "sqlite"))]
    #[tokio::test]
    async fn the_non_confirming_loop_hits_the_admission_cap_and_refusal_takes_no_custody() {
        let store = test_store().await;
        let config = OpsConfig {
            token_registry_max_entries: 3,
            ..OpsConfig::default()
        };
        let ctx = OpsContext::new(Arc::clone(&store), config);
        ctx.bind_converter(
            "ns",
            Arc::new(RecordingConverter {
                converted: Mutex::new(Vec::new()),
            }),
        );

        let mut minted = Vec::new();
        for round in 0..3 {
            let key = key_of(format!("loop-{round}").as_bytes());
            let pin = store.gc().acquire(&key).await;
            let token = ctx
                .mint_custodial("ns", key.clone(), pin)
                .unwrap_or_else(|e| panic!("the pre-cap mint (round {round}): {e}"));
            assert_eq!(
                store.gc().pin_count(&key),
                1,
                "each iteration's registry hold (round {round})"
            );
            minted.push((key, token));
        }
        assert_eq!(ctx.tokens().len(), 3, "the loop's entries registered");

        // The iteration at cap: the conforming put landed its pin —
        // the mint refuses and the refused mint's guard drops.
        let refused_key = key_of(b"loop-at-cap");
        let pin = store.gc().acquire(&refused_key).await;
        let refused = ctx.mint_custodial("ns", refused_key.clone(), pin);
        assert_eq!(refused.unwrap_err(), Error::RegistryFull);
        assert_eq!(
            store.gc().pin_count(&refused_key),
            0,
            "no custody taken for the refused mint"
        );
        assert_eq!(
            ctx.tokens().len(),
            3,
            "the refusal inserted nothing (never evicted)"
        );

        // Nothing evicted: every earlier token still holds and renews.
        for (key, token) in &minted {
            assert!(ctx.tokens().holds(token));
            assert!(ctx.renew("ns", token, key).await.present());
        }

        // The fleet arm's mapping-only mint sits under the same
        // registry-level cap.
        assert_eq!(
            ctx.tokens()
                .mint_fleet("ns", key_of(b"fleet-at-cap"), b"owner".to_vec())
                .unwrap_err(),
            Error::RegistryFull
        );
        assert_eq!(ctx.tokens().len(), 3);

        // A confirm frees headroom — admission resumes (the
        // embedder's mint policy is the loop's real bound).
        assert!(ctx.confirm("ns", &minted[0].1));
        let resume_key = key_of(b"loop-resumes");
        let pin = store.gc().acquire(&resume_key).await;
        assert!(
            ctx.mint_custodial("ns", resume_key, pin).is_ok(),
            "a confirm's retirement frees the headroom"
        );
    }
}