liminal-server 0.21.1

//! Lys grant revocation applied to admission and live rights (ACCESS-005).
//!
//! Liminal never decides who may revoke: a revocation reaches this module
//! only as an authoritative change on the grant log Lys's own `judge_revoke`
//! admitted (the issuer, a holder of an ancestor, or the root authority).
//! This module holds what liminal does with it:
//!
//! - **The binding (R1).** A connection admitted under a revocation policy
//!   carries a [`GrantDecisionBinding`]: the grant log it was decided from,
//!   the receipt revision its pass requires, and every bound right with each
//!   complete grant chain that allows it. It is built only by
//!   [`verify_binding`], from an [`AuthenticatedBinding`] the Lys producer
//!   seam ([`GrantBindingSource`]) authenticated, after its issuer, audience,
//!   holder and exact canonical pass digest were compared with the presented
//!   credential. No producer is published yet (Lys DIRECTORY-089 R2), so the
//!   installed seam is [`UnpublishedGrantBinding`], which refuses
//!   `revocation_binding_missing` by name.
//! - **The consumer (R2).** One [`RevocationAuthority`] per configured grant
//!   log applies the ordered change stream liminal reads from Lys
//!   ([`GrantFeedEvent`]): a bounded baseline, then every tail revision in
//!   order, then `Ready`. A missing revision is a `Gap`, a lost transport is
//!   `Unavailable`, a replaced log is `Reset`, an unknown event is terminal.
//!   Nothing here polls, sleeps or retries: the reader that owns the stream
//!   hands each event to [`RevocationAuthority::apply`].
//! - **The fence (R3).** [`RevocationAuthority::stage`] verifies and, outside
//!   the guard, acquires any cold decisions; [`RevocationAuthority::commit`]
//!   is the one ordered transition that checks readiness, log identity, the
//!   required revision and the admitting right, and binds the connection into
//!   the reverse index. A reply older than a deny never overwrites it.
//! - **Live rights (R4).** Applying a revocation visits only the connections
//!   the reverse index binds to the revoked grant. A lost right that did not
//!   admit the connection is withdrawn ([`RevocationEffect::Withdraw`]); the
//!   loss of the admitting right closes it ([`RevocationEffect::Close`]). Hot
//!   checks read the connection's own [`ConnectionGrants`] without the
//!   authority's guard and never ask Lys.
//! - **Cleanup (R6).** [`RevocationAuthority::release`] is called once from the
//!   connection runtime's single removal funnel; it retires the connection's
//!   index entries and prunes decisions no binding and no in-flight ask needs.
//! - **Diagnostics (R7).** [`RevocationAuthority::diagnostics`] reports the
//!   readiness, cursor, producer high water, gap length (or that it is
//!   unknown), transitions and per-phase counts.
//!
//! The cache and indexes are process memory: nothing here is stored, flushed
//! or synced.

use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::fmt;
use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, MutexGuard};

use liminal::protocol::DisconnectReason;

use crate::auth_lys::HotAction;

/// The longest grant-log identity accepted, in bytes.
pub const MAX_LOG_IDENTITY_LEN: usize = 256;
/// The longest grant id accepted, in bytes.
pub const MAX_GRANT_ID_LEN: usize = 128;
/// The most rights one binding may carry.
pub const MAX_BOUND_RIGHTS: usize = 64;
/// The most independent grant chains one right may carry.
pub const MAX_CHAINS_PER_RIGHT: usize = 8;
/// The longest grant chain (leaf to root) one right may carry.
pub const MAX_CHAIN_LEN: usize = 16;
/// The longest subject prefix one right may carry, in bytes.
pub const MAX_RIGHT_PREFIX_LEN: usize = 256;
/// The most decisions one baseline or one cold reply may carry.
pub const MAX_DECISIONS_PER_EVENT: usize = 65_536;
/// How many applied revisions keep their evidence digest, so a duplicate can
/// be told from a conflicting one.
pub const EVIDENCE_RETAINED: usize = 1_024;
/// How many readiness transitions the diagnostic ring keeps; an overflow is
/// counted and announced, never silent.
pub const MAX_DIAGNOSTIC_EVENTS: usize = 64;

/// A revocation decision failure, by its stable public name (ACCESS-005 R5).
///
/// A refused `Connect` carries the name alone as its `ConnectError` message
/// under reason `0x0006`; a refused hot operation carries it as its error
/// message. No variant carries credential, holder, live-prefix or grant-chain
/// bytes.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, thiserror::Error)]
pub enum RevocationRefusal {
    /// The right the operation or connection needs was withdrawn by a Lys
    /// grant revocation.
    #[error("grant_revoked")]
    GrantRevoked,
    /// No authenticated grant-decision binding matches the presented
    /// credential, or the binding covers no right for the operation.
    #[error("revocation_binding_missing")]
    BindingMissing,
    /// The local cache has not applied the revision the binding requires, or
    /// a needed decision could not be acquired at a current revision.
    #[error("revocation_revision_unavailable")]
    RevisionUnavailable,
    /// The grant-change stream is not ready: never ready yet, its transport
    /// is lost, or it ended on a terminal error.
    #[error("revocation_stream_unavailable")]
    StreamUnavailable,
    /// The grant-change stream skipped a revision no trusted evidence covers.
    #[error("revocation_stream_gap")]
    StreamGap,
    /// The grant log was replaced; its revisions cannot be compared with the
    /// binding's until the new log's baseline settles.
    #[error("revocation_log_reset")]
    LogReset,
    /// The peer cannot negotiate the reason-bearing close this policy
    /// requires before admission.
    #[error("revocation_protocol_unsupported")]
    ProtocolUnsupported,
}

impl RevocationRefusal {
    /// Every refusal, in the order R5 lists them.
    pub const ALL: [Self; 7] = [
        Self::GrantRevoked,
        Self::BindingMissing,
        Self::RevisionUnavailable,
        Self::StreamUnavailable,
        Self::StreamGap,
        Self::LogReset,
        Self::ProtocolUnsupported,
    ];

    /// The stable public name.
    #[must_use]
    pub const fn name(self) -> &'static str {
        match self {
            Self::GrantRevoked => "grant_revoked",
            Self::BindingMissing => "revocation_binding_missing",
            Self::RevisionUnavailable => "revocation_revision_unavailable",
            Self::StreamUnavailable => "revocation_stream_unavailable",
            Self::StreamGap => "revocation_stream_gap",
            Self::LogReset => "revocation_log_reset",
            Self::ProtocolUnsupported => "revocation_protocol_unsupported",
        }
    }

    /// The reason-bearing close this refusal ends a live connection with,
    /// when it is one a live connection can be closed for.
    #[must_use]
    pub const fn close_reason(self) -> Option<DisconnectReason> {
        match self {
            Self::GrantRevoked => Some(DisconnectReason::GrantRevoked),
            Self::LogReset => Some(DisconnectReason::RevocationLogReset),
            Self::BindingMissing
            | Self::RevisionUnavailable
            | Self::StreamUnavailable
            | Self::StreamGap
            | Self::ProtocolUnsupported => None,
        }
    }
}

/// A value from the grant stream or the producer that could not be accepted
/// as an identity: empty, or over its bound.
#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
#[error("{what} is empty or longer than its bound")]
pub struct InvalidGrantInput {
    /// Which value was refused.
    pub what: &'static str,
}

fn bounded(value: &str, max: usize, what: &'static str) -> Result<String, InvalidGrantInput> {
    if value.is_empty() || value.len() > max {
        Err(InvalidGrantInput { what })
    } else {
        Ok(value.to_owned())
    }
}

/// A grant log's identity and its reset epoch: the revision domain a receipt
/// revision belongs to. Revisions from different identities or epochs never
/// compare.
#[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct GrantLogId {
    identity: String,
    epoch: u64,
}

impl GrantLogId {
    /// The log `identity` in reset `epoch`.
    ///
    /// # Errors
    /// [`InvalidGrantInput`] when the identity is empty or longer than
    /// [`MAX_LOG_IDENTITY_LEN`].
    pub fn new(identity: &str, epoch: u64) -> Result<Self, InvalidGrantInput> {
        Ok(Self {
            identity: bounded(identity, MAX_LOG_IDENTITY_LEN, "grant log identity")?,
            epoch,
        })
    }

    /// The log's identity.
    #[must_use]
    pub fn identity(&self) -> &str {
        &self.identity
    }

    /// The log's reset epoch.
    #[must_use]
    pub const fn epoch(&self) -> u64 {
        self.epoch
    }
}

/// A Lys grant id, opaque here. Its `Debug` names its length only, so no
/// grant-chain bytes reach a log.
#[derive(Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct GrantId(String);

impl GrantId {
    /// The grant `id`.
    ///
    /// # Errors
    /// [`InvalidGrantInput`] when the id is empty or longer than
    /// [`MAX_GRANT_ID_LEN`].
    pub fn new(id: &str) -> Result<Self, InvalidGrantInput> {
        bounded(id, MAX_GRANT_ID_LEN, "grant id").map(Self)
    }

    /// The id's text.
    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl fmt::Debug for GrantId {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(formatter, "GrantId({} bytes)", self.0.len())
    }
}

/// A receipt revision: `GrantReceipt::revision()`, the receipt coordinate's
/// index plus one. Never an event-format version or a timestamp.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct Revision(u64);

impl Revision {
    /// The revision a decision must stand at to reflect the change logged at
    /// receipt coordinate `index`: `index + 1`. `None` when that overflows.
    #[must_use]
    pub const fn from_receipt_index(index: u64) -> Option<Self> {
        match index.checked_add(1) {
            Some(revision) => Some(Self(revision)),
            None => None,
        }
    }

    /// A revision already expressed as `GrantReceipt::revision()`.
    #[must_use]
    pub const fn new(revision: u64) -> Self {
        Self(revision)
    }

    /// The revision's number.
    #[must_use]
    pub const fn get(self) -> u64 {
        self.0
    }

    const fn next(self) -> Option<Self> {
        match self.0.checked_add(1) {
            Some(next) => Some(Self(next)),
            None => None,
        }
    }
}

/// The revisions strictly between `from` and `to`: how many are missing when
/// `to` arrives after `from`.
const fn missing_between(from: Revision, to: Revision) -> Option<u64> {
    match to.0.checked_sub(from.0) {
        Some(distance) => distance.checked_sub(1),
        None => None,
    }
}

/// The action a bound right allows: admission itself, or one hot action.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum RightAction {
    /// The right that admits `Connect`. Its loss closes the connection.
    Connect,
    /// One hot action on subjects under the right's prefix.
    Hot(HotAction),
}

/// One right as the producer bound it.
///
/// Its action, the subject prefix it covers (empty for
/// [`RightAction::Connect`]) and every complete grant chain (leaf first) that
/// allows it. The right stays effective while any one chain has no revoked
/// grant.
#[derive(Clone, PartialEq, Eq)]
pub struct BoundRightClaim {
    /// The action allowed.
    pub action: RightAction,
    /// The subject prefix covered.
    pub prefix: String,
    /// Every independent chain allowing the right.
    pub chains: Vec<Vec<GrantId>>,
}

impl fmt::Debug for BoundRightClaim {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("BoundRightClaim")
            .field("action", &self.action)
            .field("prefix", &format_args!("{} bytes", self.prefix.len()))
            .field("chains", &self.chains.len())
            .finish()
    }
}

/// A 32-byte digest or key id, compared exactly, shown by length only.
#[derive(Clone, Copy, PartialEq, Eq)]
pub struct Digest32(pub [u8; 32]);

impl fmt::Debug for Digest32 {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("Digest32(32 bytes)")
    }
}

/// A grant-decision binding the Lys producer seam authenticated
/// (DIRECTORY-089 R2).
///
/// Who issued it with which key, for which audience and
/// holder, over which exact canonical pass, from which log, at which
/// required revision, with which rights. It is the seam's output, not yet
/// trusted for this connection: [`verify_binding`] compares it with the
/// presented credential first.
#[derive(Clone, PartialEq, Eq)]
pub struct AuthenticatedBinding {
    /// The issuing authority's key identity.
    pub issuer: Digest32,
    /// The audience the binding was produced for.
    pub audience: String,
    /// The holder the binding names, as the pass names it.
    pub holder: Vec<u8>,
    /// The exact canonical digest of the pass the binding was produced for.
    pub pass_digest: Digest32,
    /// The grant log every dependency comes from.
    pub log: GrantLogId,
    /// The receipt revision the binding was decided at.
    pub required: Revision,
    /// Every bound right.
    pub rights: Vec<BoundRightClaim>,
}

impl fmt::Debug for AuthenticatedBinding {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("AuthenticatedBinding")
            .field("audience_bytes", &self.audience.len())
            .field("holder_bytes", &self.holder.len())
            .field("log", &self.log)
            .field("required", &self.required)
            .field("rights", &self.rights.len())
            .finish_non_exhaustive()
    }
}

/// The Lys producer seam (DIRECTORY-089 R2).
///
/// It authenticates the separate
/// versioned binding for a presented credential and computes that
/// contract's exact canonical pass digest. Its implementation is the trust
/// anchor; liminal compares what it returns with what it verified itself.
pub trait GrantBindingSource: Send + Sync + fmt::Debug {
    /// The contract's canonical digest of `credential`'s exact bytes.
    ///
    /// # Errors
    /// [`RevocationRefusal::BindingMissing`] when no digest contract is
    /// available.
    fn pass_digest(&self, credential: &[u8]) -> Result<Digest32, RevocationRefusal>;

    /// The authenticated binding produced for `credential`.
    ///
    /// # Errors
    /// [`RevocationRefusal::BindingMissing`] when no binding exists, its
    /// signature does not verify, or it is only partly published.
    fn binding_for(&self, credential: &[u8]) -> Result<AuthenticatedBinding, RevocationRefusal>;
}

/// The producer seam while Lys publishes none: every credential is refused
/// `revocation_binding_missing`, never admitted on a guessed binding.
#[derive(Clone, Copy, Debug, Default)]
pub struct UnpublishedGrantBinding;

impl GrantBindingSource for UnpublishedGrantBinding {
    fn pass_digest(&self, _credential: &[u8]) -> Result<Digest32, RevocationRefusal> {
        Err(RevocationRefusal::BindingMissing)
    }

    fn binding_for(&self, _credential: &[u8]) -> Result<AuthenticatedBinding, RevocationRefusal> {
        Err(RevocationRefusal::BindingMissing)
    }
}

/// One cold decision reply: whether each asked grant is revoked, decided at
/// `revision` on `log`.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct DecisionReply {
    /// The log the decisions come from.
    pub log: GrantLogId,
    /// The revision every decision was made at.
    pub revision: Revision,
    /// Each grant and whether it is revoked.
    pub decisions: Vec<(GrantId, bool)>,
}

/// The cold-decision seam: Lys answers whether grants are revoked on an
/// identified log at no earlier than `at_least`. Asked only outside the
/// authority's guard, during staging, never on a hot path.
pub trait GrantDecisionSource: Send + Sync + fmt::Debug {
    /// The decisions on `grants`.
    ///
    /// # Errors
    /// The refusal naming why no current decision is available.
    fn decide(
        &self,
        log: &GrantLogId,
        grants: &[GrantId],
        at_least: Revision,
    ) -> Result<DecisionReply, RevocationRefusal>;
}

/// The decision seam while Lys's batch check carries no log identity: every
/// cold ask is refused `revocation_revision_unavailable`.
#[derive(Clone, Copy, Debug, Default)]
pub struct UnpublishedGrantDecisions;

impl GrantDecisionSource for UnpublishedGrantDecisions {
    fn decide(
        &self,
        _log: &GrantLogId,
        _grants: &[GrantId],
        _at_least: Revision,
    ) -> Result<DecisionReply, RevocationRefusal> {
        Err(RevocationRefusal::RevisionUnavailable)
    }
}

/// What a binding must match to be used here: the configured log, this
/// server's audience and the issuer key it trusts.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RevocationPolicy {
    /// The grant log this authority consumes.
    pub log: GrantLogId,
    /// This server's audience.
    pub audience: String,
    /// The issuer key identity a binding must name.
    pub issuer: Digest32,
}

#[derive(Clone, PartialEq, Eq)]
struct BoundRight {
    action: RightAction,
    prefix: String,
    chains: Vec<Vec<GrantId>>,
}

/// The binding one `Connect` verification proved, carried privately into
/// the connection. Only [`verify_binding`] builds one.
#[derive(Clone, PartialEq, Eq)]
pub struct GrantDecisionBinding {
    log: GrantLogId,
    required: Revision,
    rights: Vec<BoundRight>,
    admitting: usize,
}

impl GrantDecisionBinding {
    /// The grant log the binding was decided from.
    #[must_use]
    pub const fn log(&self) -> &GrantLogId {
        &self.log
    }

    /// The receipt revision the binding requires the cache to have applied.
    #[must_use]
    pub const fn required(&self) -> Revision {
        self.required
    }

    /// How many rights the binding carries.
    #[must_use]
    pub fn right_count(&self) -> usize {
        self.rights.len()
    }

    fn grants(&self) -> impl Iterator<Item = &GrantId> {
        self.rights
            .iter()
            .flat_map(|right| right.chains.iter().flatten())
    }
}

impl fmt::Debug for GrantDecisionBinding {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("GrantDecisionBinding")
            .field("log", &self.log)
            .field("required", &self.required)
            .field("rights", &self.rights.len())
            .finish_non_exhaustive()
    }
}

/// Verifies the binding for `credential`, presented by `holder`, against
/// `policy` (ACCESS-005 R1).
///
/// The seam authenticates it, then its issuer,
/// audience, holder and exact canonical pass digest must each equal what was
/// presented, and every bound must hold, before any dependency is used.
///
/// # Errors
/// [`RevocationRefusal::BindingMissing`] when the seam has no binding, or
/// any comparison or bound fails: a substituted, partial or over-bound
/// binding is refused, never repaired.
pub fn verify_binding(
    policy: &RevocationPolicy,
    source: &dyn GrantBindingSource,
    credential: &[u8],
    holder: &[u8],
) -> Result<GrantDecisionBinding, RevocationRefusal> {
    let claimed = source.binding_for(credential)?;
    let digest = source.pass_digest(credential)?;
    let mismatch = if claimed.issuer != policy.issuer {
        Some("issuer")
    } else if claimed.audience != policy.audience {
        Some("audience")
    } else if claimed.holder.as_slice() != holder {
        Some("holder")
    } else if claimed.pass_digest != digest {
        Some("pass_digest")
    } else {
        None
    };
    if let Some(field) = mismatch {
        tracing::warn!(
            field,
            "grant-decision binding does not match the presented credential; refused"
        );
        return Err(RevocationRefusal::BindingMissing);
    }
    bound_rights(claimed)
}

fn bound_rights(claimed: AuthenticatedBinding) -> Result<GrantDecisionBinding, RevocationRefusal> {
    let refuse = |what: &'static str| {
        tracing::warn!(
            what,
            "grant-decision binding is over its bound or incomplete"
        );
        RevocationRefusal::BindingMissing
    };
    if claimed.rights.is_empty() || claimed.rights.len() > MAX_BOUND_RIGHTS {
        return Err(refuse("rights"));
    }
    let mut rights = Vec::with_capacity(claimed.rights.len());
    for right in claimed.rights {
        if right.prefix.len() > MAX_RIGHT_PREFIX_LEN {
            return Err(refuse("prefix"));
        }
        if right.chains.is_empty() || right.chains.len() > MAX_CHAINS_PER_RIGHT {
            return Err(refuse("chains"));
        }
        if right
            .chains
            .iter()
            .any(|chain| chain.is_empty() || chain.len() > MAX_CHAIN_LEN)
        {
            return Err(refuse("chain"));
        }
        if matches!(right.action, RightAction::Hot(_)) && right.prefix.is_empty() {
            return Err(refuse("hot right without a prefix"));
        }
        rights.push(BoundRight {
            action: right.action,
            prefix: right.prefix,
            chains: right.chains,
        });
    }
    let admitting = rights
        .iter()
        .position(|right| right.action == RightAction::Connect)
        .ok_or_else(|| refuse("admitting right"))?;
    Ok(GrantDecisionBinding {
        log: claimed.log,
        required: claimed.required,
        rights,
        admitting,
    })
}

const CLOSE_NONE: u8 = 0;
const CLOSE_GRANT_REVOKED: u8 = 1;
const CLOSE_LOG_RESET: u8 = 2;

const fn close_code(reason: RevocationRefusal) -> u8 {
    match reason {
        RevocationRefusal::LogReset => CLOSE_LOG_RESET,
        RevocationRefusal::GrantRevoked
        | RevocationRefusal::BindingMissing
        | RevocationRefusal::RevisionUnavailable
        | RevocationRefusal::StreamUnavailable
        | RevocationRefusal::StreamGap
        | RevocationRefusal::ProtocolUnsupported => CLOSE_GRANT_REVOKED,
    }
}

/// The rights one admitted connection holds now, shared between the
/// authority (which withdraws) and the connection (which reads). Reads take
/// no lock.
#[derive(Debug)]
struct LiveRights {
    lost: Box<[AtomicBool]>,
    close: AtomicU8,
    changes: AtomicU64,
}

#[derive(Debug, Default)]
struct HotCounts {
    allowed: AtomicU64,
    refused: AtomicU64,
}

/// One admitted connection's binding and its current rights (ACCESS-005 R3,
/// R4).
///
/// Every hot operation of the connection is decided here, locally, at
/// the authority's ordered generation, without the authority's guard and
/// without asking Lys.
#[derive(Clone)]
pub struct ConnectionGrants {
    pid: u64,
    binding: Arc<GrantDecisionBinding>,
    live: Arc<LiveRights>,
    hot: Arc<HotCounts>,
}

impl fmt::Debug for ConnectionGrants {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("ConnectionGrants")
            .field("pid", &self.pid)
            .field("binding", &self.binding)
            .finish_non_exhaustive()
    }
}

impl ConnectionGrants {
    /// The connection this binding admitted.
    #[must_use]
    pub const fn pid(&self) -> u64 {
        self.pid
    }

    /// The binding the connection was admitted on; never replaced.
    #[must_use]
    pub fn binding(&self) -> &GrantDecisionBinding {
        &self.binding
    }

    /// Whether `action` on `subject` is allowed now.
    ///
    /// # Errors
    /// [`RevocationRefusal::GrantRevoked`] when every bound right covering
    /// it was withdrawn or the connection is closing for a revocation;
    /// [`RevocationRefusal::LogReset`] when it is closing for a log reset;
    /// [`RevocationRefusal::BindingMissing`] when no bound right covers it.
    pub fn allows(&self, action: HotAction, subject: &str) -> Result<(), RevocationRefusal> {
        let decided = self.decide(action, subject);
        let counter = if decided.is_ok() {
            &self.hot.allowed
        } else {
            &self.hot.refused
        };
        counter.fetch_add(1, Ordering::Relaxed);
        decided
    }

    /// Whether `action` on `subject` is allowed now, without counting it as
    /// an operation: the connection's own sweep reads this to retire the
    /// work of withdrawn rights.
    #[must_use]
    pub fn permits(&self, action: HotAction, subject: &str) -> bool {
        self.decide(action, subject).is_ok()
    }

    fn decide(&self, action: HotAction, subject: &str) -> Result<(), RevocationRefusal> {
        if let Some(reason) = self.close_pending() {
            return Err(reason);
        }
        let mut covered = false;
        for (right, lost) in self.binding.rights.iter().zip(self.live.lost.iter()) {
            if right.action == RightAction::Hot(action) && subject.starts_with(&right.prefix) {
                covered = true;
                if !lost.load(Ordering::Acquire) {
                    return Ok(());
                }
            }
        }
        Err(if covered {
            RevocationRefusal::GrantRevoked
        } else {
            RevocationRefusal::BindingMissing
        })
    }

    /// The reason the connection must close for, once its admitting right
    /// is lost or its log was reset.
    #[must_use]
    pub fn close_pending(&self) -> Option<RevocationRefusal> {
        match self.live.close.load(Ordering::Acquire) {
            CLOSE_NONE => None,
            CLOSE_LOG_RESET => Some(RevocationRefusal::LogReset),
            _ => Some(RevocationRefusal::GrantRevoked),
        }
    }

    /// How many withdrawals and closes this connection's rights have seen.
    /// A connection compares it with the last value it acted on, so a
    /// sweep runs only after a change.
    #[must_use]
    pub fn changes(&self) -> u64 {
        self.live.changes.load(Ordering::Acquire)
    }
}

/// The consumer's readiness (ACCESS-005 R2).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Readiness {
    /// No baseline and cursor have agreed yet.
    NotReady,
    /// Baseline, cursor and rights indexes agree; admissions may commit.
    Ready,
    /// A revision arrived with no trusted evidence for the interval before it.
    Gap,
    /// The stream's transport was lost.
    Unavailable,
    /// The log was replaced; the new log's baseline has not settled.
    Reset,
    /// The stream ended on an unknown or invalid event, or conflicting
    /// evidence. Sticky: a new consumer must be started.
    Terminal,
}

impl Readiness {
    const fn refusal(self) -> Option<RevocationRefusal> {
        match self {
            Self::Ready => None,
            Self::NotReady | Self::Unavailable | Self::Terminal => {
                Some(RevocationRefusal::StreamUnavailable)
            }
            Self::Gap => Some(RevocationRefusal::StreamGap),
            Self::Reset => Some(RevocationRefusal::LogReset),
        }
    }
}

/// One grant change on the stream.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum GrantFeedChange {
    /// A grant was issued.
    Issue(GrantId),
    /// A grant was revoked, and with it everything derived from it.
    Revoke(GrantId),
}

/// One event of the consumer-opened Lys grant-change stream (Lys
/// DIRECTORY-089 R1), as the reader that owns the stream decoded it.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum GrantFeedEvent {
    /// An authenticated bounded baseline: whether each needed grant is
    /// revoked, through `through`.
    Baseline {
        /// The log.
        log: GrantLogId,
        /// The revision the baseline stands at.
        through: Revision,
        /// Each grant and whether it is revoked.
        decisions: Vec<(GrantId, bool)>,
    },
    /// One committed change.
    Change {
        /// The log.
        log: GrantLogId,
        /// The change's receipt revision.
        revision: Revision,
        /// The change.
        change: GrantFeedChange,
        /// The digest of its signed evidence, telling a duplicate from a
        /// conflict.
        evidence: Digest32,
    },
    /// Trusted evidence that nothing visible to this consumer changed
    /// through `through`.
    Watermark {
        /// The log.
        log: GrantLogId,
        /// The revision covered through.
        through: Revision,
    },
    /// The stream states the baseline and replay are installed through
    /// `through`.
    Ready {
        /// The log.
        log: GrantLogId,
        /// The revision installed through.
        through: Revision,
    },
    /// The stream's transport was lost.
    Lost,
    /// The log was replaced.
    Reset {
        /// The replaced log.
        from: GrantLogId,
        /// The new log.
        to: GrantLogId,
    },
    /// An event this consumer cannot read: unknown version or invalid body.
    Invalid {
        /// The event's stable error name.
        reason: &'static str,
    },
}

/// What applying an event asks the connection runtime to do.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum RevocationEffect {
    /// Retire the work of one lost right that did not admit the connection.
    Withdraw {
        /// The connection.
        pid: u64,
        /// The right's index in the binding.
        right: usize,
    },
    /// Close the connection with a reason.
    Close {
        /// The connection.
        pid: u64,
        /// The reason.
        reason: RevocationRefusal,
    },
}

impl RevocationEffect {
    /// The connection the effect is for.
    #[must_use]
    pub const fn pid(self) -> u64 {
        match self {
            Self::Withdraw { pid, .. } | Self::Close { pid, .. } => pid,
        }
    }
}

/// How an event was taken.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Applied {
    /// A baseline was installed.
    Baseline,
    /// A contiguous change was applied.
    Change,
    /// A revision at or below the cursor, with matching evidence.
    Duplicate,
    /// A watermark advanced the cursor.
    Watermark,
    /// A revision arrived past a missing interval; nothing was applied.
    Gap,
    /// `Ready` was published, after rechecking every open connection when a
    /// resume required it.
    Ready,
    /// The transport was lost.
    Lost,
    /// The log was replaced.
    Reset,
    /// The event was refused, by name.
    Refused(&'static str),
    /// The stream ended terminally, by name.
    Terminal(&'static str),
}

/// The result of applying one event.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ApplyOutcome {
    /// How the event was taken.
    pub applied: Applied,
    /// What the connection runtime must do, in order.
    pub effects: Vec<RevocationEffect>,
}

/// How a reason-bearing close's write ended, as the connection observed it.
/// None of these is a claim the peer received it.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum CloseWrite {
    /// The frame was written to the socket in full.
    Flushed,
    /// Part of the outbound queue was still unwritten when the close ended.
    Residue,
    /// The frame could not be queued (outbound full or encode refused).
    NotQueued,
    /// The socket write failed.
    TransportFailed,
    /// The connection had no stream left to write to.
    NoStream,
}

/// Per-phase counts (ACCESS-005 R7).
///
/// Every count is incremented at the
/// phase it names, never inferred from time or a successful exit. Nothing
/// here touches a store: store calls, flushes and syncs are zero by
/// construction and have no counter.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct RevocationCounts {
    /// Revocation changes observed on the stream.
    pub observed_revocations: u64,
    /// Unique contiguous revisions applied.
    pub applied_revisions: u64,
    /// Revisions at or below the cursor, ignored.
    pub duplicate_revisions: u64,
    /// Watermarks that advanced the cursor.
    pub watermark_advances: u64,
    /// Events refused by name.
    pub refused_events: u64,
    /// Connections visited through the reverse index for a change.
    pub affected_visits: u64,
    /// Connections rechecked by a settled resume.
    pub resume_rechecks: u64,
    /// Admissions committed through the fence.
    pub admissions_committed: u64,
    /// Admissions refused, at staging or at the fence.
    pub admissions_refused: u64,
    /// Attempts to bind a second binding to a connection, refused.
    pub binding_replacements_refused: u64,
    /// Rights withdrawn without closing.
    pub withdrawals: u64,
    /// Closes decided.
    pub closes_decided: u64,
    /// Closes whose control could not be published; the denied state still
    /// gates the connection and its own wake delivers the close.
    pub close_publication_refused: u64,
    /// Reason-bearing closes the connection attempted to write.
    pub closes_attempted: u64,
    /// Of those, the ones written to the socket in full.
    pub closes_flushed: u64,
    /// Of those, the ones whose write failed or had no stream.
    pub close_transport_failures: u64,
    /// Connections released through the removal funnel.
    pub cleanups: u64,
    /// Decisions pruned as unreferenced.
    pub decisions_pruned: u64,
    /// Transitions into `Gap`.
    pub gap_transitions: u64,
    /// Transitions into `Unavailable`.
    pub loss_transitions: u64,
    /// Transitions into `Reset`.
    pub reset_transitions: u64,
    /// Terminal stream errors.
    pub terminal_errors: u64,
    /// Cold decision asks made to Lys (network calls).
    pub decision_asks: u64,
    /// Cold replies older than the revision they were asked at, refused.
    pub stale_replies_refused: u64,
    /// Reply decisions that would have overwritten a deny, refused.
    pub stale_overwrites_prevented: u64,
    /// Decision rows read from baselines and replies.
    pub rows_read: u64,
}

/// Whether the consumer knows how many revisions it is missing.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum GapLength {
    /// No interval is missing.
    None,
    /// Exactly this many revisions are missing.
    Known(u64),
    /// Revisions may be missing, and how many is not known: the producer's
    /// high water is unknown. Never reported as zero.
    Unknown,
}

/// One announced readiness transition.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Transition {
    /// The readiness entered.
    pub readiness: Readiness,
    /// Its stable reason, when it is a loss.
    pub reason: Option<RevocationRefusal>,
    /// The cursor at the transition.
    pub applied: Option<Revision>,
    /// The missing interval at the transition.
    pub missing: GapLength,
    /// Cumulative admissions refused at the transition.
    pub refused_admissions: u64,
    /// Cumulative resume rechecks at the transition.
    pub rechecks: u64,
}

/// The revocation authority's lock was poisoned by a panicking holder; its
/// state is untrusted and is not read.
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
#[error("revocation_authority_poisoned")]
pub struct AuthorityPoisoned;

/// A bounded snapshot of the consumer (ACCESS-005 R7). It carries no
/// credential, holder, prefix or grant bytes.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RevocationDiagnostics {
    /// The log consumed now.
    pub log: GrantLogId,
    /// The readiness.
    pub readiness: Readiness,
    /// The loss reason new admissions are refused with, if any.
    pub reason: Option<RevocationRefusal>,
    /// The last applied receipt revision.
    pub applied: Option<Revision>,
    /// The last trusted producer high-water revision; `None` is unknown.
    pub high_water: Option<Revision>,
    /// The missing interval.
    pub missing: GapLength,
    /// Resume generations begun.
    pub resume_generation: u64,
    /// The resume generation whose rechecks settled.
    pub settled_generation: u64,
    /// Connections bound now.
    pub bound_connections: usize,
    /// Decisions retained now.
    pub retained_decisions: usize,
    /// Reverse-index entries now.
    pub index_entries: usize,
    /// Cold asks in flight now.
    pub inflight_asks: u64,
    /// The per-phase counts.
    pub counts: RevocationCounts,
    /// Hot operations allowed.
    pub hot_allowed: u64,
    /// Hot operations refused.
    pub hot_refused: u64,
    /// The retained transitions, oldest first.
    pub transitions: Vec<Transition>,
    /// Transitions dropped from the bounded ring.
    pub transitions_dropped: u64,
}

#[derive(Clone, Copy, Debug)]
struct Decision {
    revoked: bool,
    at: Revision,
    baseline: bool,
}

#[derive(Debug)]
struct Bound {
    binding: Arc<GrantDecisionBinding>,
    live: Arc<LiveRights>,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum RightState {
    Allowed,
    Lost,
    Unknown,
}

#[derive(Debug)]
struct Inner {
    log: GrantLogId,
    readiness: Readiness,
    applied: Option<Revision>,
    high_water: Option<Revision>,
    pending_resume: bool,
    resume_generation: u64,
    settled_generation: u64,
    decisions: BTreeMap<GrantId, Decision>,
    evidence: VecDeque<(Revision, Digest32)>,
    connections: BTreeMap<u64, Bound>,
    index: BTreeMap<GrantId, BTreeSet<u64>>,
    orphans: BTreeSet<GrantId>,
    inflight_asks: u64,
    counts: RevocationCounts,
    transitions: VecDeque<Transition>,
    transitions_dropped: u64,
    announce: Option<Transition>,
}

const fn bump(counter: &mut u64) {
    *counter = counter.saturating_add(1);
}

impl Inner {
    fn new(log: GrantLogId) -> Self {
        Self {
            log,
            readiness: Readiness::NotReady,
            applied: None,
            high_water: None,
            pending_resume: false,
            resume_generation: 0,
            settled_generation: 0,
            decisions: BTreeMap::new(),
            evidence: VecDeque::new(),
            connections: BTreeMap::new(),
            index: BTreeMap::new(),
            orphans: BTreeSet::new(),
            inflight_asks: 0,
            counts: RevocationCounts::default(),
            transitions: VecDeque::new(),
            transitions_dropped: 0,
            announce: None,
        }
    }

    fn missing(&self) -> GapLength {
        if self.readiness == Readiness::Ready {
            return GapLength::None;
        }
        match (self.applied, self.high_water) {
            (Some(applied), Some(high)) => match missing_between(applied, high) {
                Some(missing) if missing > 0 => GapLength::Known(missing),
                _ => GapLength::None,
            },
            // Not ready and the producer's high water (or the cursor) is not
            // known: how much is missing is unknown, never zero.
            _ => GapLength::Unknown,
        }
    }

    /// Enters `readiness`, recording one transition when it changes; the
    /// transition is announced after the guard is released.
    fn transition(&mut self, readiness: Readiness) {
        if self.readiness == readiness {
            return;
        }
        self.readiness = readiness;
        match readiness {
            Readiness::Gap => bump(&mut self.counts.gap_transitions),
            Readiness::Unavailable => bump(&mut self.counts.loss_transitions),
            Readiness::Reset => bump(&mut self.counts.reset_transitions),
            Readiness::Terminal => bump(&mut self.counts.terminal_errors),
            Readiness::NotReady | Readiness::Ready => {}
        }
        let transition = Transition {
            readiness,
            reason: readiness.refusal(),
            applied: self.applied,
            missing: self.missing(),
            refused_admissions: self.counts.admissions_refused,
            rechecks: self.counts.resume_rechecks,
        };
        if self.transitions.len() >= MAX_DIAGNOSTIC_EVENTS {
            self.transitions.pop_front();
            bump(&mut self.transitions_dropped);
        }
        self.transitions.push_back(transition);
        self.announce = Some(transition);
    }

    fn precheck(&self, binding: &GrantDecisionBinding) -> Result<(), RevocationRefusal> {
        if let Some(refusal) = self.readiness.refusal() {
            return Err(refusal);
        }
        if binding.log != self.log {
            return Err(RevocationRefusal::LogReset);
        }
        if self
            .applied
            .is_none_or(|applied| applied < binding.required)
        {
            return Err(RevocationRefusal::RevisionUnavailable);
        }
        Ok(())
    }

    fn missing_grants(&self, binding: &GrantDecisionBinding) -> Vec<GrantId> {
        let unknown: BTreeSet<&GrantId> = binding
            .grants()
            .filter(|grant| !self.decisions.contains_key(*grant))
            .collect();
        unknown.into_iter().cloned().collect()
    }

    fn right_state(&self, right: &BoundRight) -> RightState {
        let mut unknown = false;
        for chain in &right.chains {
            let mut chain_state = RightState::Allowed;
            for grant in chain {
                match self.decisions.get(grant) {
                    Some(decision) if decision.revoked => {
                        chain_state = RightState::Lost;
                        break;
                    }
                    Some(_) => {}
                    None => chain_state = RightState::Unknown,
                }
            }
            match chain_state {
                RightState::Allowed => return RightState::Allowed,
                RightState::Unknown => unknown = true,
                RightState::Lost => {}
            }
        }
        if unknown {
            RightState::Unknown
        } else {
            RightState::Lost
        }
    }

    /// Merges one decision, never letting an older or allowing answer
    /// overwrite a deny.
    fn merge_decision(&mut self, grant: &GrantId, revoked: bool, at: Revision, baseline: bool) {
        match self.decisions.get_mut(grant) {
            Some(existing) if existing.revoked => {
                if !revoked {
                    bump(&mut self.counts.stale_overwrites_prevented);
                }
                existing.baseline |= baseline;
            }
            Some(existing) if existing.at >= at && !revoked => {
                existing.baseline |= baseline;
            }
            Some(existing) => {
                existing.revoked = revoked;
                existing.at = at;
                existing.baseline |= baseline;
            }
            None => {
                self.decisions.insert(
                    grant.clone(),
                    Decision {
                        revoked,
                        at,
                        baseline,
                    },
                );
                if !baseline && !self.index.contains_key(grant) {
                    self.orphans.insert(grant.clone());
                }
            }
        }
    }

    fn install_reply(
        &mut self,
        reply: &DecisionReply,
        at_least: Revision,
    ) -> Result<(), RevocationRefusal> {
        if reply.log != self.log {
            return Err(RevocationRefusal::LogReset);
        }
        if reply.revision < at_least || reply.decisions.len() > MAX_DECISIONS_PER_EVENT {
            bump(&mut self.counts.stale_replies_refused);
            return Err(RevocationRefusal::RevisionUnavailable);
        }
        for (grant, revoked) in &reply.decisions {
            bump(&mut self.counts.rows_read);
            self.merge_decision(grant, *revoked, reply.revision, false);
        }
        Ok(())
    }

    /// The final fence and the binding, in one ordered transition.
    fn bind(
        &mut self,
        pid: u64,
        binding: &Arc<GrantDecisionBinding>,
        hot: &Arc<HotCounts>,
    ) -> Result<ConnectionGrants, RevocationRefusal> {
        self.precheck(binding)?;
        if self.connections.contains_key(&pid) {
            bump(&mut self.counts.binding_replacements_refused);
            tracing::warn!(
                connection_pid = pid,
                "a second grant-decision binding was offered; the first is kept"
            );
            return Err(RevocationRefusal::BindingMissing);
        }
        let mut lost = Vec::with_capacity(binding.rights.len());
        for (index, right) in binding.rights.iter().enumerate() {
            match (self.right_state(right), index == binding.admitting) {
                (RightState::Allowed, _) => lost.push(AtomicBool::new(false)),
                (RightState::Lost, true) => return Err(RevocationRefusal::GrantRevoked),
                (RightState::Lost, false) => lost.push(AtomicBool::new(true)),
                (RightState::Unknown, _) => return Err(RevocationRefusal::RevisionUnavailable),
            }
        }
        let live = Arc::new(LiveRights {
            lost: lost.into_boxed_slice(),
            close: AtomicU8::new(CLOSE_NONE),
            changes: AtomicU64::new(0),
        });
        for grant in binding.grants() {
            self.index.entry(grant.clone()).or_default().insert(pid);
            self.orphans.remove(grant);
        }
        self.connections.insert(
            pid,
            Bound {
                binding: Arc::clone(binding),
                live: Arc::clone(&live),
            },
        );
        bump(&mut self.counts.admissions_committed);
        Ok(ConnectionGrants {
            pid,
            binding: Arc::clone(binding),
            live,
            hot: Arc::clone(hot),
        })
    }

    /// Re-evaluates one bound connection, withdrawing newly lost rights and
    /// closing it when its admitting right is lost.
    fn evaluate(&mut self, pid: u64, effects: &mut Vec<RevocationEffect>) {
        let Some(bound) = self.connections.get(&pid) else {
            return;
        };
        let binding = Arc::clone(&bound.binding);
        let live = Arc::clone(&bound.live);
        if live.close.load(Ordering::Acquire) != CLOSE_NONE {
            return;
        }
        if binding.log != self.log {
            self.close(pid, &live, RevocationRefusal::LogReset, effects);
            return;
        }
        for (index, (right, lost)) in binding.rights.iter().zip(live.lost.iter()).enumerate() {
            if lost.load(Ordering::Acquire) || self.right_state(right) == RightState::Allowed {
                continue;
            }
            // A miss is never a grant: an undecided chain counts as lost.
            lost.store(true, Ordering::Release);
            live.changes.fetch_add(1, Ordering::AcqRel);
            if index == binding.admitting {
                self.close(pid, &live, RevocationRefusal::GrantRevoked, effects);
                return;
            }
            bump(&mut self.counts.withdrawals);
            effects.push(RevocationEffect::Withdraw { pid, right: index });
        }
    }

    fn close(
        &mut self,
        pid: u64,
        live: &LiveRights,
        reason: RevocationRefusal,
        effects: &mut Vec<RevocationEffect>,
    ) {
        live.close.store(close_code(reason), Ordering::Release);
        live.changes.fetch_add(1, Ordering::AcqRel);
        bump(&mut self.counts.closes_decided);
        effects.push(RevocationEffect::Close { pid, reason });
    }

    fn remember_evidence(&mut self, revision: Revision, evidence: Digest32) {
        if self.evidence.len() >= EVIDENCE_RETAINED {
            self.evidence.pop_front();
        }
        self.evidence.push_back((revision, evidence));
    }

    fn raise_high_water(&mut self, revision: Revision) {
        if self.high_water.is_none_or(|high| high < revision) {
            self.high_water = Some(revision);
        }
    }

    fn apply_change(
        &mut self,
        revision: Revision,
        change: &GrantFeedChange,
        evidence: Digest32,
        effects: &mut Vec<RevocationEffect>,
    ) -> Applied {
        self.raise_high_water(revision);
        let Some(applied) = self.applied else {
            return Applied::Refused("change before baseline");
        };
        if revision <= applied {
            let conflicting = self
                .evidence
                .iter()
                .any(|(seen, digest)| *seen == revision && *digest != evidence);
            if conflicting {
                self.transition(Readiness::Terminal);
                return Applied::Terminal("conflicting_evidence");
            }
            bump(&mut self.counts.duplicate_revisions);
            return Applied::Duplicate;
        }
        if applied.next() != Some(revision) {
            self.pending_resume = true;
            self.transition(Readiness::Gap);
            return Applied::Gap;
        }
        self.applied = Some(revision);
        self.remember_evidence(revision, evidence);
        bump(&mut self.counts.applied_revisions);
        match change {
            GrantFeedChange::Issue(grant) => {
                if self.decisions.contains_key(grant) || self.index.contains_key(grant) {
                    self.merge_decision(grant, false, revision, false);
                }
            }
            GrantFeedChange::Revoke(grant) => {
                bump(&mut self.counts.observed_revocations);
                let referenced = self.index.contains_key(grant);
                if referenced || self.inflight_asks > 0 || self.decisions.contains_key(grant) {
                    // A tombstone, kept while a binding or an in-flight ask
                    // could still read this grant.
                    self.merge_decision(grant, true, revision, false);
                }
                let affected: Vec<u64> = self
                    .index
                    .get(grant)
                    .map(|pids| pids.iter().copied().collect())
                    .unwrap_or_default();
                for pid in affected {
                    bump(&mut self.counts.affected_visits);
                    self.evaluate(pid, effects);
                }
            }
        }
        Applied::Change
    }

    fn apply_ready(&mut self, through: Revision, effects: &mut Vec<RevocationEffect>) -> Applied {
        if self.applied != Some(through) {
            return Applied::Refused("ready does not match the applied cursor");
        }
        self.raise_high_water(through);
        if self.readiness == Readiness::Ready && !self.pending_resume {
            return Applied::Duplicate;
        }
        self.resume_generation = self.resume_generation.saturating_add(1);
        let open: Vec<u64> = self.connections.keys().copied().collect();
        for pid in open {
            bump(&mut self.counts.resume_rechecks);
            self.evaluate(pid, effects);
        }
        self.pending_resume = false;
        self.settled_generation = self.resume_generation;
        self.transition(Readiness::Ready);
        Applied::Ready
    }

    fn apply(&mut self, event: &GrantFeedEvent, effects: &mut Vec<RevocationEffect>) -> Applied {
        if self.readiness == Readiness::Terminal {
            return Applied::Refused("stream is terminal");
        }
        match event {
            GrantFeedEvent::Baseline {
                log,
                through,
                decisions,
            } => self.apply_baseline(log, *through, decisions),
            GrantFeedEvent::Change {
                log,
                revision,
                change,
                evidence,
            } => {
                if *log != self.log {
                    return Applied::Refused("change from another log");
                }
                self.apply_change(*revision, change, *evidence, effects)
            }
            GrantFeedEvent::Watermark { log, through } => {
                if *log != self.log {
                    return Applied::Refused("watermark from another log");
                }
                self.raise_high_water(*through);
                match self.applied {
                    None => Applied::Refused("watermark before baseline"),
                    Some(applied) if *through <= applied => {
                        bump(&mut self.counts.duplicate_revisions);
                        Applied::Duplicate
                    }
                    Some(_) => {
                        self.applied = Some(*through);
                        bump(&mut self.counts.watermark_advances);
                        Applied::Watermark
                    }
                }
            }
            GrantFeedEvent::Ready { log, through } => {
                if *log != self.log {
                    return Applied::Refused("ready from another log");
                }
                self.apply_ready(*through, effects)
            }
            GrantFeedEvent::Lost => {
                self.high_water = None;
                self.pending_resume = true;
                self.transition(Readiness::Unavailable);
                Applied::Lost
            }
            GrantFeedEvent::Reset { from, to } => self.apply_reset(from, to),
            GrantFeedEvent::Invalid { reason } => {
                self.transition(Readiness::Terminal);
                Applied::Terminal(reason)
            }
        }
    }

    fn apply_baseline(
        &mut self,
        log: &GrantLogId,
        through: Revision,
        decisions: &[(GrantId, bool)],
    ) -> Applied {
        if *log != self.log {
            return Applied::Refused("baseline from another log");
        }
        if decisions.len() > MAX_DECISIONS_PER_EVENT {
            return Applied::Refused("baseline over its bound");
        }
        if self.applied.is_some_and(|applied| through < applied) {
            return Applied::Refused("baseline older than the applied cursor");
        }
        for (grant, revoked) in decisions {
            bump(&mut self.counts.rows_read);
            self.merge_decision(grant, *revoked, through, true);
        }
        self.applied = Some(through);
        self.raise_high_water(through);
        Applied::Baseline
    }

    fn apply_reset(&mut self, from: &GrantLogId, to: &GrantLogId) -> Applied {
        if *from != self.log {
            return Applied::Refused("reset from a log not consumed here");
        }
        self.log = to.clone();
        // Revisions of the old log never compare with the new one's. Applied
        // denials stay on each connection's own rights and never revert.
        self.decisions.clear();
        self.evidence.clear();
        self.orphans.clear();
        self.applied = None;
        self.high_water = None;
        self.pending_resume = true;
        self.transition(Readiness::Reset);
        Applied::Reset
    }

    fn prune_orphans(&mut self) {
        if self.inflight_asks > 0 {
            return;
        }
        for grant in std::mem::take(&mut self.orphans) {
            if self.index.contains_key(&grant) {
                continue;
            }
            if self
                .decisions
                .get(&grant)
                .is_some_and(|decision| !decision.baseline)
            {
                self.decisions.remove(&grant);
                bump(&mut self.counts.decisions_pruned);
            }
        }
    }

    fn finish_ask(&mut self) {
        self.inflight_asks = self.inflight_asks.saturating_sub(1);
        self.prune_orphans();
    }

    fn release(&mut self, pid: u64) -> bool {
        let Some(bound) = self.connections.remove(&pid) else {
            return false;
        };
        for grant in bound.binding.grants() {
            if let Some(pids) = self.index.get_mut(grant) {
                pids.remove(&pid);
                if pids.is_empty() {
                    self.index.remove(grant);
                    self.orphans.insert(grant.clone());
                }
            }
        }
        bump(&mut self.counts.cleanups);
        self.prune_orphans();
        true
    }
}

/// A verified binding whose cold decisions are acquired, waiting for the
/// fence. While it lives it holds an in-flight ask, so no deny it could read
/// is pruned; dropping it uncommitted releases that.
pub struct StagedAdmission {
    binding: Arc<GrantDecisionBinding>,
    inner: Arc<Mutex<Inner>>,
    registered: bool,
}

impl fmt::Debug for StagedAdmission {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("StagedAdmission")
            .field("binding", &self.binding)
            .finish_non_exhaustive()
    }
}

impl Drop for StagedAdmission {
    fn drop(&mut self) {
        if self.registered
            && let Ok(mut inner) = self.inner.lock()
        {
            inner.finish_ask();
        }
    }
}

/// One grant log's revocation consumer, decision cache and admission fence
/// (ACCESS-005 R1-R4, R6, R7).
pub struct RevocationAuthority {
    policy: RevocationPolicy,
    bindings: Arc<dyn GrantBindingSource>,
    decisions: Arc<dyn GrantDecisionSource>,
    inner: Arc<Mutex<Inner>>,
    hot: Arc<HotCounts>,
}

impl fmt::Debug for RevocationAuthority {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("RevocationAuthority")
            .field("log", &self.policy.log)
            .finish_non_exhaustive()
    }
}

impl RevocationAuthority {
    /// A consumer for `policy`'s log, `NotReady` until its stream's baseline
    /// and `Ready` arrive, with the producer and decision seams given.
    #[must_use]
    pub fn new(
        policy: RevocationPolicy,
        bindings: Arc<dyn GrantBindingSource>,
        decisions: Arc<dyn GrantDecisionSource>,
    ) -> Self {
        let inner = Arc::new(Mutex::new(Inner::new(policy.log.clone())));
        Self {
            policy,
            bindings,
            decisions,
            inner,
            hot: Arc::new(HotCounts::default()),
        }
    }

    /// The authority with the seams Lys publishes today: neither, so every
    /// admission is refused by name.
    #[must_use]
    pub fn unpublished(policy: RevocationPolicy) -> Self {
        Self::new(
            policy,
            Arc::new(UnpublishedGrantBinding),
            Arc::new(UnpublishedGrantDecisions),
        )
    }

    fn lock(&self) -> Result<MutexGuard<'_, Inner>, RevocationRefusal> {
        self.inner
            .lock()
            .map_err(|_| RevocationRefusal::StreamUnavailable)
    }

    fn refused(&self, refusal: RevocationRefusal) -> RevocationRefusal {
        if let Ok(mut inner) = self.inner.lock() {
            bump(&mut inner.counts.admissions_refused);
        }
        refusal
    }

    /// Verifies `credential`'s binding for `holder` and acquires any decision
    /// the cache lacks, outside the guard. Nothing is admitted yet.
    ///
    /// # Errors
    /// The refusal by name: the binding is missing or does not match, the
    /// stream is not ready, the log was reset, or the required revision or a
    /// needed decision is unavailable.
    pub fn stage(
        &self,
        credential: &[u8],
        holder: &[u8],
    ) -> Result<StagedAdmission, RevocationRefusal> {
        let binding = verify_binding(&self.policy, self.bindings.as_ref(), credential, holder)
            .map_err(|refusal| self.refused(refusal))?;
        let binding = Arc::new(binding);
        let (needed, at_least) = {
            let mut inner = self.lock()?;
            if let Err(refusal) = inner.precheck(&binding) {
                bump(&mut inner.counts.admissions_refused);
                return Err(refusal);
            }
            inner.inflight_asks = inner.inflight_asks.saturating_add(1);
            let needed = inner.missing_grants(&binding);
            let at_least = inner.applied;
            drop(inner);
            (needed, at_least)
        };
        let staged = StagedAdmission {
            binding,
            inner: Arc::clone(&self.inner),
            registered: true,
        };
        if needed.is_empty() {
            return Ok(staged);
        }
        let at_least =
            at_least.ok_or_else(|| self.refused(RevocationRefusal::RevisionUnavailable))?;
        let reply = self
            .decisions
            .decide(&staged.binding.log, &needed, at_least);
        let installed = {
            let mut inner = self.lock()?;
            bump(&mut inner.counts.decision_asks);
            let installed = reply.and_then(|reply| inner.install_reply(&reply, at_least));
            if installed.is_err() {
                bump(&mut inner.counts.admissions_refused);
            }
            drop(inner);
            installed
        };
        installed.map(|()| staged)
    }

    /// The fence (ACCESS-005 R3): readiness, log identity, the required
    /// revision and the admitting right, checked and bound into the reverse
    /// index in one ordered transition with change application. A revocation
    /// applied between [`Self::stage`] and this call wins.
    ///
    /// # Errors
    /// The refusal by name; nothing is bound on any refusal.
    pub fn commit(
        &self,
        pid: u64,
        mut staged: StagedAdmission,
    ) -> Result<ConnectionGrants, RevocationRefusal> {
        if !Arc::ptr_eq(&staged.inner, &self.inner) {
            return Err(self.refused(RevocationRefusal::BindingMissing));
        }
        staged.registered = false;
        let mut inner = self.lock()?;
        let bound = inner.bind(pid, &staged.binding, &self.hot);
        if bound.is_err() {
            bump(&mut inner.counts.admissions_refused);
        }
        inner.finish_ask();
        drop(inner);
        bound
    }

    /// [`Self::stage`] then [`Self::commit`].
    ///
    /// # Errors
    /// As either step.
    pub fn admit(
        &self,
        pid: u64,
        credential: &[u8],
        holder: &[u8],
    ) -> Result<ConnectionGrants, RevocationRefusal> {
        let staged = self.stage(credential, holder)?;
        self.commit(pid, staged)
    }

    /// Applies one stream event in order (ACCESS-005 R2, R4). Announcements
    /// are logged after the guard is released.
    #[must_use]
    pub fn apply(&self, event: &GrantFeedEvent) -> ApplyOutcome {
        let mut effects = Vec::new();
        let Ok(mut inner) = self.inner.lock() else {
            return ApplyOutcome {
                applied: Applied::Terminal("revocation authority lock poisoned"),
                effects,
            };
        };
        let before = inner.transitions_dropped;
        inner.announce = None;
        let applied = inner.apply(event, &mut effects);
        if matches!(applied, Applied::Refused(_)) {
            bump(&mut inner.counts.refused_events);
        }
        let announce = inner.announce.take();
        let dropped = inner.transitions_dropped > before;
        drop(inner);
        announce_outcome(applied, announce, dropped);
        ApplyOutcome { applied, effects }
    }

    /// Releases connection `pid` from the reverse index, once, from the
    /// connection runtime's removal funnel (ACCESS-005 R6). Returns whether a
    /// binding was released; a second call releases nothing.
    #[must_use]
    pub fn release(&self, pid: u64) -> bool {
        let Ok(mut inner) = self.inner.lock() else {
            tracing::error!(
                pid,
                "revocation authority lock poisoned; the binding is not released"
            );
            return false;
        };
        inner.release(pid)
    }

    /// Records that the close control for `pid` could not be published. The
    /// denied state already gates the connection; its own wake delivers the
    /// close.
    pub fn record_close_publication_refused(&self, pid: u64) {
        if let Ok(mut inner) = self.inner.lock() {
            bump(&mut inner.counts.close_publication_refused);
        }
        tracing::warn!(
            connection_pid = pid,
            "revocation close control could not be published; the connection's wake delivers it"
        );
    }

    /// Records how connection `pid`'s reason-bearing close write ended.
    /// Attempted is not delivered.
    pub fn record_close_write(&self, pid: u64, write: CloseWrite) {
        if let Ok(mut inner) = self.inner.lock() {
            bump(&mut inner.counts.closes_attempted);
            match write {
                CloseWrite::Flushed => bump(&mut inner.counts.closes_flushed),
                CloseWrite::TransportFailed | CloseWrite::NoStream => {
                    bump(&mut inner.counts.close_transport_failures);
                }
                CloseWrite::Residue | CloseWrite::NotQueued => {}
            }
        }
        if matches!(write, CloseWrite::TransportFailed | CloseWrite::NoStream) {
            tracing::warn!(
                connection_pid = pid,
                write = ?write,
                "revocation close could not be written; the denied state is kept and cleanup finishes"
            );
        }
    }

    /// Drops every decision no binding and no in-flight ask needs. A later
    /// need is a miss, answered by a cold ask, never an implicit allow.
    pub fn prune_unreferenced(&self) {
        let Ok(mut inner) = self.inner.lock() else {
            return;
        };
        if inner.inflight_asks > 0 {
            return;
        }
        let unreferenced: Vec<GrantId> = inner
            .decisions
            .keys()
            .filter(|grant| !inner.index.contains_key(*grant))
            .cloned()
            .collect();
        for grant in unreferenced {
            inner.decisions.remove(&grant);
            bump(&mut inner.counts.decisions_pruned);
        }
        inner.orphans.clear();
    }

    /// A bounded snapshot for diagnostics (ACCESS-005 R7).
    ///
    /// # Errors
    /// [`AuthorityPoisoned`] when a holder of the authority's lock panicked:
    /// the state is untrusted and no snapshot of it is answered.
    pub fn diagnostics(&self) -> Result<RevocationDiagnostics, AuthorityPoisoned> {
        let inner = self.inner.lock().map_err(|_| AuthorityPoisoned)?;
        let snapshot = RevocationDiagnostics {
            log: inner.log.clone(),
            readiness: inner.readiness,
            reason: inner.readiness.refusal(),
            applied: inner.applied,
            high_water: inner.high_water,
            missing: inner.missing(),
            resume_generation: inner.resume_generation,
            settled_generation: inner.settled_generation,
            bound_connections: inner.connections.len(),
            retained_decisions: inner.decisions.len(),
            index_entries: inner.index.values().map(BTreeSet::len).sum(),
            inflight_asks: inner.inflight_asks,
            counts: inner.counts,
            hot_allowed: self.hot.allowed.load(Ordering::Relaxed),
            hot_refused: self.hot.refused.load(Ordering::Relaxed),
            transitions: inner.transitions.iter().copied().collect(),
            transitions_dropped: inner.transitions_dropped,
        };
        drop(inner);
        Ok(snapshot)
    }
}

/// Announces a readiness transition loudly, with its stable reason and the
/// cumulative refused-admission and recheck counts, and names a dropped
/// diagnostic transition.
fn announce_outcome(applied: Applied, transition: Option<Transition>, dropped: bool) {
    if let Some(transition) = transition {
        match transition.readiness {
            Readiness::Ready => tracing::info!(
                applied = ?transition.applied.map(Revision::get),
                rechecks = transition.rechecks,
                "revocation stream ready"
            ),
            Readiness::NotReady
            | Readiness::Gap
            | Readiness::Unavailable
            | Readiness::Reset
            | Readiness::Terminal => tracing::warn!(
                readiness = ?transition.readiness,
                reason = ?transition.reason.map(RevocationRefusal::name),
                applied = ?transition.applied.map(Revision::get),
                missing = ?transition.missing,
                refused_admissions = transition.refused_admissions,
                rechecks = transition.rechecks,
                "revocation stream lost readiness; new admissions are refused by name"
            ),
        }
    }
    if dropped {
        tracing::warn!("revocation diagnostic transition ring overflowed; the oldest was dropped");
    }
    match applied {
        Applied::Refused(reason) => {
            tracing::warn!(reason, "revocation stream event refused");
        }
        Applied::Terminal(reason) => {
            tracing::error!(reason, "revocation stream ended on a terminal error");
        }
        Applied::Baseline
        | Applied::Change
        | Applied::Duplicate
        | Applied::Watermark
        | Applied::Gap
        | Applied::Ready
        | Applied::Lost
        | Applied::Reset => {}
    }
}