liminal-server 0.21.1

//! Channel membership bound to the verified holder and the placed resource
//! (ACCESS-006).
//!
//! A membership is Lys's answer to one question asked through the one shared
//! contract, `lys_pass::membership`: may this verified subject read, or post
//! on, this placed channel resource within this workspace, at least at this
//! grant revision. It is never a liminal roster, a guess from a subject
//! string, a readable parent taken as membership of a child, or a consequence
//! of signing in.
//!
//! - **The mapping is explicit** ([`crate::config::types::MembershipConfig`]):
//!   a channel is membership-protected only when the operator's product
//!   mapping lists it with its workspace and placed resource; a participant
//!   conversation is placed under a channel only when the mapping lists it.
//!   Every other channel keeps its prior contract, byte for byte.
//! - **The subject is the verified holder** ([`MembershipSubject`]): built only
//!   from the admitted Lys session, with the kind the signed pass states, so an
//!   AI is asked about as its own agent identity and never as its responsible
//!   person. A connection that proved no Lys holder has no membership at all.
//! - **The provider is a named seam** ([`MembershipProvider`]); the live one is
//!   [`HttpMembershipProvider`] over Lys's `POST /grants/membership` routes.
//!   Nothing here answers a membership question itself.
//! - **The authority is the ACCESS-005 grant owner** ([`MembershipAuthority`]):
//!   a protected act proceeds only while it states `Ready` on the configured
//!   grant log. `NotReady`, `Unavailable`, `Gap` and `Reset` refuse every
//!   protected act and hold every protected delivery with that cause; no last
//!   applied allow is a fallback (stricter than ACCESS-005 R2's legacy
//!   continuation, and only for protected channels). The owner drives ordered
//!   changes and resumes into the gate through [`MembershipListener`].
//! - **Hot checks are local** ([`MembershipGate::check`]): a decision is asked
//!   once per (holder, channel, action) and retained in a bounded index at the
//!   authority's generation; later acts and delivery releases read it under a
//!   short lock and never ask Lys. A change re-decides only the bindings its
//!   resource reaches (a reverse index), a resume re-decides every retained
//!   binding once, and a binding is released when the last connection holding
//!   it closes.
//! - **Pages are bounded** ([`MembershipGate::resource_page`],
//!   [`MembershipGate::recipient_page`]): both bounds are the operator's,
//!   every row is checked against the question, and a continuation is bound
//!   to its exact question and generation.

use std::collections::{BTreeMap, BTreeSet};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, MutexGuard, OnceLock};

use lys_pass::membership::{
    CONTRACT_UNSUPPORTED, CONTRACT_VERSION, GrantLog, LOG_MISMATCH, MembershipDecision,
    MembershipRequest, Subject, Verdict,
};
use lys_pass::membership_admission::{AdmissionDecision, AdmissionRequest};
use lys_pass::membership_pages::{
    CURSOR_FOREIGN, CURSOR_RESET, MembershipPage, PageBounds, PageOutcome, PageRow,
    RecipientPageRequest, ResourcePageRequest,
};
use lys_pass::rights::{Mode, Resource};

use crate::ServerError;
use crate::auth_lys::HotAction;
use crate::auth_lys::session::LysSession;
use crate::config::types::MembershipConfig;

/// The Lys route answering one membership decision.
pub const DECISION_ROUTE: &str = "grants/membership";
/// The Lys route answering scoped guest admission.
pub const ADMISSION_ROUTE: &str = "grants/membership/admission";
/// The Lys route answering a page of a subject's permitted resources.
pub const RESOURCES_ROUTE: &str = "grants/membership/resources";
/// The Lys route answering a page of a channel's permitted recipients.
pub const RECIPIENTS_ROUTE: &str = "grants/membership/recipients";

/// The two membership actions. Read and post are separate: an answer about
/// one says nothing of the other (ACCESS-006 R1).
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum MembershipAction {
    /// Reading a channel: a subscription, a delivered reply, result or
    /// request.
    Read,
    /// Posting on a channel: a publish or an outbound service effect.
    Post,
}

impl MembershipAction {
    /// The contract's action word.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Read => "read",
            Self::Post => "post",
        }
    }

    /// The membership action a hot route needs: a publish and a call post,
    /// a subscription and a served request read. A call or serve still needs
    /// its own operation grant beside the membership; membership never
    /// implies it.
    #[must_use]
    pub const fn for_hot(action: HotAction) -> Self {
        match action {
            HotAction::Publish | HotAction::Call => Self::Post,
            HotAction::Subscribe | HotAction::Serve => Self::Read,
        }
    }
}

/// Why a protected act, delivery or page was refused.
///
/// Every variant has a
/// stable [`Self::name`]; none carries a payload, a credential, a resource
/// id or a channel name, so a refusal can cross any boundary whole.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum MembershipRefusal {
    /// The connection proved no Lys holder, so it has no membership.
    HolderUnverified,
    /// The admitted pass did not name its holder's directory kind.
    HolderKindUnknown,
    /// The verified holder's id or kind is empty or carries control
    /// characters.
    HolderMalformed,
    /// No ACCESS-005 grant authority is installed, so no membership is
    /// current.
    AuthorityMissing,
    /// The grant authority has not established its verified baseline.
    NotReady,
    /// The grant authority or the provider could not be asked, by cause.
    Unavailable {
        /// The stable cause, naming no response body.
        cause: String,
    },
    /// The grant authority's change feed has an uncovered interval.
    Gap {
        /// The last revision applied before the gap.
        after: u64,
        /// How many revisions are missing.
        missing: u64,
    },
    /// The grant log was reset; its revisions no longer compare.
    Reset,
    /// The authority moved to a generation this index has not yet resumed.
    ResumePending,
    /// A change or a resume is re-deciding this binding.
    ChangeApplying,
    /// The answer or the authority names another grant log or epoch.
    LogMismatch,
    /// The provider answered under a contract this liminal does not speak.
    ContractUnsupported {
        /// The contract version the provider named.
        contract: u32,
    },
    /// The answer echoes a request other than the one asked: a substituted
    /// subject, resource, workspace or action.
    AnswerSubstituted,
    /// The answer reflects an older revision than the one required.
    RevisionStale {
        /// The least revision required.
        required: u64,
        /// The revision decided at, if any.
        decided: Option<u64>,
    },
    /// The allowing scope is neither the mapped channel nor its workspace.
    ScopeOutsideMapping,
    /// Lys refused the membership by its stable name.
    Refused {
        /// The provider's stable refusal name.
        refusal: String,
    },
    /// A grant reaches the act, but only by approval.
    Held {
        /// The approval the grant requires.
        mode: Mode,
    },
    /// The index is at its configured bound; nothing was allocated.
    CapacityExceeded {
        /// The configured bound.
        limit: usize,
    },
    /// The index lock was poisoned by a panic elsewhere.
    IndexPoisoned,
    /// The workspace asked about is not in the product mapping.
    WorkspaceNotMapped,
    /// The channel asked about is not in the product mapping.
    ChannelNotMapped,
    /// The continuation was issued for another question.
    CursorForeign,
    /// The continuation was issued at another generation or log epoch.
    CursorStale,
    /// A page exceeded its row or byte bound.
    PageOverBound {
        /// The rows the page carried.
        rows: u64,
        /// The encoded bytes the rows carried.
        bytes: u64,
    },
    /// A page row answers another subject, kind, resource or workspace.
    PageForeignRow,
    /// A page listed one row twice.
    PageDuplicateRow,
    /// A page's counts, completion and continuation disagree.
    PageInconsistent,
}

impl MembershipRefusal {
    /// Whether this refusal holds rather than denies (ACCESS-006 R2): the
    /// authority or the index is not current, so a queued protected delivery
    /// waits, releasing nothing, rather than being retired. Every other
    /// refusal is a decided deny.
    #[must_use]
    pub const fn holds(&self) -> bool {
        matches!(
            self,
            Self::AuthorityMissing
                | Self::NotReady
                | Self::Unavailable { .. }
                | Self::Gap { .. }
                | Self::Reset
                | Self::ResumePending
                | Self::ChangeApplying
                | Self::IndexPoisoned
        )
    }

    /// The stable machine name.
    #[must_use]
    pub fn name(&self) -> &str {
        match self {
            Self::HolderUnverified => "membership_holder_unverified",
            Self::HolderKindUnknown => "membership_holder_kind_unknown",
            Self::HolderMalformed => "membership_holder_malformed",
            Self::AuthorityMissing => "membership_authority_missing",
            Self::NotReady => "membership_not_ready",
            Self::Unavailable { .. } => "membership_unavailable",
            Self::Gap { .. } => "membership_gap",
            Self::Reset => "membership_log_reset",
            Self::ResumePending => "membership_resume_pending",
            Self::ChangeApplying => "membership_change_applying",
            Self::LogMismatch => LOG_MISMATCH,
            Self::ContractUnsupported { .. } => CONTRACT_UNSUPPORTED,
            Self::AnswerSubstituted => "membership_answer_substituted",
            Self::RevisionStale { .. } => "membership_revision_stale",
            Self::ScopeOutsideMapping => "membership_scope_outside_mapping",
            Self::Refused { refusal } => refusal,
            Self::Held { .. } => "membership_held",
            Self::CapacityExceeded { .. } => "membership_capacity_exceeded",
            Self::IndexPoisoned => "membership_index_poisoned",
            Self::WorkspaceNotMapped => "membership_workspace_not_mapped",
            Self::ChannelNotMapped => "membership_channel_not_mapped",
            Self::CursorForeign => CURSOR_FOREIGN,
            Self::CursorStale => "membership_cursor_stale",
            Self::PageOverBound { .. } => "membership_page_over_bound",
            Self::PageForeignRow => "membership_page_foreign_row",
            Self::PageDuplicateRow => "membership_page_duplicate_row",
            Self::PageInconsistent => "membership_page_inconsistent",
        }
    }
}

impl std::fmt::Display for MembershipRefusal {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Unavailable { cause } => write!(formatter, "{}: {cause}", self.name()),
            Self::Gap { after, missing } => write!(
                formatter,
                "{}: {missing} revisions missing after {after}",
                self.name()
            ),
            Self::ContractUnsupported { contract } => write!(
                formatter,
                "{}: contract {contract}; this liminal speaks {CONTRACT_VERSION}",
                self.name()
            ),
            Self::RevisionStale { required, decided } => write!(
                formatter,
                "{}: required {required}, decided {decided:?}",
                self.name()
            ),
            Self::Held { mode } => write!(formatter, "{}: {mode:?}", self.name()),
            Self::CapacityExceeded { limit } => {
                write!(formatter, "{}: limit {limit}", self.name())
            }
            Self::PageOverBound { rows, bytes } => {
                write!(formatter, "{}: {rows} rows, {bytes} bytes", self.name())
            }
            _ => formatter.write_str(self.name()),
        }
    }
}

impl std::error::Error for MembershipRefusal {}

fn malformed(value: &str) -> bool {
    value.is_empty() || value.chars().any(char::is_control)
}

fn invalid(message: impl Into<String>) -> ServerError {
    ServerError::ConfigValidation {
        message: message.into(),
    }
}

fn bump(counter: &AtomicU64) {
    counter.fetch_add(1, Ordering::Relaxed);
}

fn bump_by(counter: &AtomicU64, count: u64) {
    counter.fetch_add(count, Ordering::Relaxed);
}

/// The verified subject a membership is asked about: the holder the one
/// `Connect` verification proved, with the directory kind its signed pass
/// states. Only [`Self::from_session`] makes one.
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord)]
pub struct MembershipSubject {
    id: String,
    kind: String,
}

impl MembershipSubject {
    /// The subject an admitted Lys session proved.
    ///
    /// # Errors
    /// [`MembershipRefusal::HolderKindUnknown`] when the session cannot name
    /// its holder's kind; [`MembershipRefusal::HolderMalformed`] when the id or
    /// kind is empty or carries control characters.
    pub fn from_session(session: &dyn LysSession) -> Result<Self, MembershipRefusal> {
        let kind = session
            .holder_kind()
            .ok_or(MembershipRefusal::HolderKindUnknown)?;
        let id = session.holder();
        if malformed(id) || malformed(kind) {
            return Err(MembershipRefusal::HolderMalformed);
        }
        Ok(Self {
            id: id.to_owned(),
            kind: kind.to_owned(),
        })
    }

    /// The directory kind, as the signed pass states it.
    #[must_use]
    pub fn kind(&self) -> &str {
        &self.kind
    }

    /// The contract's subject.
    #[must_use]
    pub fn to_contract(&self) -> Subject {
        Subject {
            id: self.id.clone(),
            kind: self.kind.clone(),
        }
    }

    fn is(&self, subject: &Subject) -> bool {
        self.id == subject.id && self.kind == subject.kind
    }
}

impl std::fmt::Debug for MembershipSubject {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("MembershipSubject")
            .field("id", &crate::auth_pass::ByteLength(self.id.len()))
            .field("kind", &self.kind)
            .finish()
    }
}

/// A protected channel's placement as the product mapping states it.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct MappedChannel {
    workspace: Resource,
    resource: Resource,
}

impl MappedChannel {
    /// The workspace the channel is placed under.
    #[must_use]
    pub const fn workspace(&self) -> &Resource {
        &self.workspace
    }

    /// The placed channel resource.
    #[must_use]
    pub const fn resource(&self) -> &Resource {
        &self.resource
    }
}

/// The validated product mapping.
#[derive(Debug)]
struct ChannelMap {
    channels: BTreeMap<String, MappedChannel>,
    conversations: BTreeMap<u64, String>,
    by_resource: BTreeMap<String, String>,
    workspaces: BTreeSet<String>,
}

fn channel_map(config: &MembershipConfig) -> Result<ChannelMap, ServerError> {
    if config.channels.is_empty() {
        return Err(invalid(
            "auth.lys.membership.channels: at least one protected channel is required",
        ));
    }
    let mut channels = BTreeMap::new();
    let mut by_resource = BTreeMap::new();
    let mut workspaces = BTreeSet::new();
    for entry in &config.channels {
        for (member, value) in [
            ("channel", entry.channel.as_str()),
            ("workspace", entry.workspace.as_str()),
            ("resource", entry.resource.as_str()),
        ] {
            if malformed(value) {
                return Err(invalid(format!(
                    "auth.lys.membership.channels: a {member} is empty or contains control characters"
                )));
            }
        }
        let mapped = MappedChannel {
            workspace: Resource {
                kind: config.workspace_kind.clone(),
                id: entry.workspace.clone(),
            },
            resource: Resource {
                kind: config.channel_kind.clone(),
                id: entry.resource.clone(),
            },
        };
        if channels.insert(entry.channel.clone(), mapped).is_some() {
            return Err(invalid(
                "auth.lys.membership.channels: a channel is listed twice",
            ));
        }
        if by_resource
            .insert(entry.resource.clone(), entry.channel.clone())
            .is_some()
        {
            return Err(invalid(
                "auth.lys.membership.channels: a placed resource is mapped to two channels",
            ));
        }
        workspaces.insert(entry.workspace.clone());
    }
    let mut conversations = BTreeMap::new();
    for entry in &config.conversations {
        if !channels.contains_key(&entry.channel) {
            return Err(invalid(
                "auth.lys.membership.conversations: a conversation names an unlisted channel",
            ));
        }
        if conversations
            .insert(entry.conversation, entry.channel.clone())
            .is_some()
        {
            return Err(invalid(
                "auth.lys.membership.conversations: a conversation is listed twice",
            ));
        }
    }
    Ok(ChannelMap {
        channels,
        conversations,
        by_resource,
        workspaces,
    })
}

fn validated_bounds(config: &MembershipConfig) -> Result<PageBounds, ServerError> {
    if config.contract != CONTRACT_VERSION {
        return Err(invalid(format!(
            "auth.lys.membership.contract {} is not served; this liminal speaks {CONTRACT_VERSION}",
            config.contract
        )));
    }
    for (member, value) in [
        ("log_identity", config.log_identity.as_str()),
        ("workspace_kind", config.workspace_kind.as_str()),
        ("channel_kind", config.channel_kind.as_str()),
    ] {
        if malformed(value) {
            return Err(invalid(format!(
                "auth.lys.membership.{member} is empty or contains control characters"
            )));
        }
    }
    if config.workspace_kind == config.channel_kind {
        return Err(invalid(
            "auth.lys.membership: workspace_kind and channel_kind must differ",
        ));
    }
    if config.page_rows == 0 || config.page_bytes == 0 || config.max_decisions == 0 {
        return Err(invalid(
            "auth.lys.membership: page_rows, page_bytes and max_decisions must each be positive",
        ));
    }
    Ok(PageBounds {
        rows: config.page_rows,
        bytes: config.page_bytes,
    })
}

/// A provider failure before a contract answer was received.
#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
#[error("membership provider could not be asked: {cause}")]
pub struct ProviderError {
    /// The stable cause, naming no response body and no credential.
    pub cause: String,
}

fn provider_cause(cause: &str) -> ProviderError {
    ProviderError {
        cause: cause.to_owned(),
    }
}

/// Lys's shared membership provider behind a synchronous seam
/// (ACCESS-006 R1, R4).
///
/// The live implementation is
/// [`HttpMembershipProvider`]; a fixture answers from a seeded world. No
/// implementation may answer from a liminal-side membership list.
pub trait MembershipProvider: Send + Sync + std::fmt::Debug {
    /// Asks one membership decision.
    ///
    /// # Errors
    /// [`ProviderError`] when Lys could not be asked or its answer could not
    /// be read.
    fn decide(&self, request: &MembershipRequest) -> Result<MembershipDecision, ProviderError>;

    /// Asks scoped guest admission to one workspace.
    ///
    /// # Errors
    /// [`ProviderError`] as for [`Self::decide`].
    fn admission(&self, request: &AdmissionRequest) -> Result<AdmissionDecision, ProviderError>;

    /// Asks one page of the resources a subject may act on.
    ///
    /// # Errors
    /// [`ProviderError`] as for [`Self::decide`].
    fn resources(&self, request: &ResourcePageRequest) -> Result<MembershipPage, ProviderError>;

    /// Asks one page of the subjects who may act on a resource.
    ///
    /// # Errors
    /// [`ProviderError`] as for [`Self::decide`].
    fn recipients(&self, request: &RecipientPageRequest) -> Result<MembershipPage, ProviderError>;
}

/// Lys's membership routes over HTTP, asked as this liminal's app with its
/// credential.
///
/// Each ask runs to completion on a scoped thread against the
/// server's runtime, as `HttpLysAsk` does; nothing is cached here and no
/// failure becomes an answer.
pub struct HttpMembershipProvider {
    http: reqwest::Client,
    base: reqwest::Url,
    credential: zeroize::Zeroizing<String>,
    handle: tokio::runtime::Handle,
}

impl std::fmt::Debug for HttpMembershipProvider {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("HttpMembershipProvider")
            .field("base", &self.base.as_str())
            .field(
                "credential",
                &crate::auth_pass::ByteLength(self.credential.len()),
            )
            .finish_non_exhaustive()
    }
}

impl HttpMembershipProvider {
    /// Asks the Lys `issuer` as the app holding `credential`, on `handle`'s
    /// runtime.
    ///
    /// # Errors
    /// [`ProviderError`] when the issuer is not an HTTP origin, the credential
    /// is empty, or the HTTP client cannot be built.
    pub fn new(
        issuer: &str,
        credential: zeroize::Zeroizing<String>,
        handle: tokio::runtime::Handle,
    ) -> Result<Self, ProviderError> {
        let mut base =
            reqwest::Url::parse(issuer).map_err(|_| provider_cause("issuer_not_a_url"))?;
        if !matches!(base.scheme(), "http" | "https")
            || base.host_str().is_none()
            || !base.username().is_empty()
            || base.password().is_some()
            || base.query().is_some()
            || base.fragment().is_some()
        {
            return Err(provider_cause("issuer_not_an_http_origin"));
        }
        if !base.path().ends_with('/') {
            let path = format!("{}/", base.path());
            base.set_path(&path);
        }
        if credential.trim().is_empty() {
            return Err(provider_cause("membership_credential_empty"));
        }
        let http = reqwest::Client::builder()
            .build()
            .map_err(|_| provider_cause("http_client_unavailable"))?;
        Ok(Self {
            http,
            base,
            credential,
            handle,
        })
    }

    fn post<Q, A>(&self, route: &str, question: &Q) -> Result<A, ProviderError>
    where
        Q: serde::Serialize + ?Sized,
        A: serde::de::DeserializeOwned + Send,
    {
        let url = self
            .base
            .join(route)
            .map_err(|_| provider_cause("membership_route_unjoinable"))?;
        let request = self
            .http
            .post(url)
            .bearer_auth(self.credential.as_str())
            .json(question);
        let handle = &self.handle;
        std::thread::scope(|scope| {
            scope
                .spawn(move || {
                    handle.block_on(async move {
                        let response = request
                            .send()
                            .await
                            .map_err(|_| provider_cause("lys_could_not_be_asked"))?;
                        let status = response.status();
                        if !status.is_success() {
                            return Err(ProviderError {
                                cause: format!("lys_http_status_{}", status.as_u16()),
                            });
                        }
                        response
                            .json::<A>()
                            .await
                            .map_err(|_| provider_cause("membership_answer_unreadable"))
                    })
                })
                .join()
        })
        .unwrap_or_else(|_| Err(provider_cause("membership_ask_did_not_complete")))
    }
}

impl MembershipProvider for HttpMembershipProvider {
    fn decide(&self, request: &MembershipRequest) -> Result<MembershipDecision, ProviderError> {
        self.post(DECISION_ROUTE, request)
    }

    fn admission(&self, request: &AdmissionRequest) -> Result<AdmissionDecision, ProviderError> {
        self.post(ADMISSION_ROUTE, request)
    }

    fn resources(&self, request: &ResourcePageRequest) -> Result<MembershipPage, ProviderError> {
        self.post(RESOURCES_ROUTE, request)
    }

    fn recipients(&self, request: &RecipientPageRequest) -> Result<MembershipPage, ProviderError> {
        self.post(RECIPIENTS_ROUTE, request)
    }
}

/// The ACCESS-005 grant authority's state, as the membership gate reads it
/// (ACCESS-006 R2).
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum AuthorityState {
    /// No verified baseline yet.
    NotReady,
    /// Baseline, cursor and indexes agree on `log` at `revision`, in resume
    /// generation `generation`.
    Ready {
        /// The grant log the authority is current on.
        log: GrantLog,
        /// The last applied receipt revision.
        revision: u64,
        /// The settled resume generation.
        generation: u64,
    },
    /// The change feed or the provider is lost.
    Unavailable {
        /// The stable cause.
        cause: String,
    },
    /// The change feed has an uncovered interval.
    Gap {
        /// The last revision applied before the gap.
        after: u64,
        /// How many revisions are missing.
        missing: u64,
    },
    /// The grant log was reset and the new baseline is not yet verified.
    Reset,
}

/// The seam ACCESS-005's versioned grant owner implements for membership
/// (ACCESS-006 R2): its current state, read locally with no network call and
/// no wait.
///
/// The one owner per grant log drives ordered changes and resumes
/// into the gate through [`MembershipListener`]; there is no second feed.
pub trait MembershipAuthority: Send + Sync + std::fmt::Debug {
    /// The authority's current state.
    fn state(&self) -> AuthorityState;
}

/// One applied grant change, in the owner's order.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct MembershipChange {
    /// The grant log the change belongs to.
    pub log: GrantLog,
    /// Its receipt revision.
    pub revision: u64,
    /// The subject whose grant changed; `None` when every holder on the
    /// resource is affected.
    pub subject: Option<Subject>,
    /// The resource the changed grant is held on: a channel or a workspace.
    pub resource: Resource,
}

/// What applying one change did.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct AppliedChange {
    /// The revision was already applied: nothing was touched.
    pub duplicate: bool,
    /// Retained bindings the change's resource reaches.
    pub affected: u64,
    /// Of those, re-decided at the change's revision.
    pub redecided: u64,
    /// Of those, released because the provider could not answer; the next
    /// act asks again.
    pub failed: u64,
}

/// What one resume did.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct ResumeReport {
    /// Retained bindings re-decided once for the generation.
    pub rechecked: u64,
    /// Of those, released because the provider could not answer.
    pub failed: u64,
    /// Whether every binding agreed and the gate adopted the generation:
    /// the owner republishes `Ready` only when this holds.
    pub settled: bool,
}

/// The gate's side of the ACCESS-005 owner's ordered consumer.
///
/// The owner
/// calls [`Self::changed`] after applying each change and before it admits
/// any later act, and [`Self::resumed`] after a resume's contiguous tail and
/// before it republishes `Ready`.
pub trait MembershipListener: Send + Sync {
    /// Re-decides the retained bindings `change` reaches.
    ///
    /// # Errors
    /// [`MembershipRefusal::LogMismatch`] for another log; the index's own
    /// refusals.
    fn changed(&self, change: &MembershipChange) -> Result<AppliedChange, MembershipRefusal>;

    /// Re-decides every retained binding once at `revision` and adopts
    /// `generation` when all agree.
    ///
    /// # Errors
    /// The index's own refusals.
    fn resumed(&self, generation: u64, revision: u64) -> Result<ResumeReport, MembershipRefusal>;
}

/// The result of a membership check.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Decided {
    /// The channel is not membership-protected: its prior contract applies.
    NotProtected,
    /// Allowed by `grant`, decided at `revision`.
    Allowed {
        /// The granting receipt.
        grant: String,
        /// The revision decided at.
        revision: u64,
    },
}

/// Every count the gate keeps, by owned phase (ACCESS-006 R6). There is no
/// history, roster or whole-state structure to traverse or clone, so no such
/// count exists; a page's rows are counted as received.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct MembershipCounts {
    /// Membership decisions asked of the provider.
    pub decisions_asked: u64,
    /// Admission decisions asked of the provider.
    pub admissions_asked: u64,
    /// Pages asked of the provider.
    pub pages_asked: u64,
    /// Page rows received.
    pub page_rows: u64,
    /// Checks answered from the index.
    pub local_lookups: u64,
    /// Release checks made at a delivery seam.
    pub release_checks: u64,
    /// Refused checks.
    pub refused: u64,
    /// Bindings a change's reverse index visited.
    pub affected_visited: u64,
    /// Bindings a resume re-decided.
    pub resume_rechecks: u64,
    /// Bindings released when their last connection closed.
    pub released: u64,
    /// Index lock acquisitions.
    pub locks: u64,
}

#[derive(Debug, Default)]
struct Counters {
    decisions_asked: AtomicU64,
    admissions_asked: AtomicU64,
    pages_asked: AtomicU64,
    page_rows: AtomicU64,
    local_lookups: AtomicU64,
    release_checks: AtomicU64,
    refused: AtomicU64,
    affected_visited: AtomicU64,
    resume_rechecks: AtomicU64,
    released: AtomicU64,
    locks: AtomicU64,
}

impl Counters {
    fn snapshot(&self) -> MembershipCounts {
        let read = |counter: &AtomicU64| counter.load(Ordering::Relaxed);
        MembershipCounts {
            decisions_asked: read(&self.decisions_asked),
            admissions_asked: read(&self.admissions_asked),
            pages_asked: read(&self.pages_asked),
            page_rows: read(&self.page_rows),
            local_lookups: read(&self.local_lookups),
            release_checks: read(&self.release_checks),
            refused: read(&self.refused),
            affected_visited: read(&self.affected_visited),
            resume_rechecks: read(&self.resume_rechecks),
            released: read(&self.released),
            locks: read(&self.locks),
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
struct DecisionKey {
    subject: MembershipSubject,
    channel: String,
    action: MembershipAction,
}

/// A retained decision, or the ticket of the ask settling it.
#[derive(Clone, Debug, PartialEq, Eq)]
enum Indexed {
    Applying { ticket: u64 },
    Allowed { grant: String, revision: u64 },
    Refused { refusal: String },
    Held { mode: Mode },
}

impl Indexed {
    fn result(&self) -> Result<Decided, MembershipRefusal> {
        match self {
            Self::Applying { .. } => Err(MembershipRefusal::ChangeApplying),
            Self::Allowed { grant, revision } => Ok(Decided::Allowed {
                grant: grant.clone(),
                revision: *revision,
            }),
            Self::Refused { refusal } => Err(MembershipRefusal::Refused {
                refusal: refusal.clone(),
            }),
            Self::Held { mode } => Err(MembershipRefusal::Held { mode: *mode }),
        }
    }
}

type ResourceKey = (String, String);

fn resource_key(resource: &Resource) -> ResourceKey {
    (resource.kind.clone(), resource.id.clone())
}

#[derive(Debug)]
struct Entry {
    verdict: Indexed,
    connections: BTreeSet<u64>,
    reached_by: [ResourceKey; 2],
}

#[derive(Debug, Default)]
struct DecisionIndex {
    generation: Option<u64>,
    applied: u64,
    next_ticket: u64,
    entries: BTreeMap<DecisionKey, Entry>,
    by_resource: BTreeMap<ResourceKey, BTreeSet<DecisionKey>>,
    by_connection: BTreeMap<u64, BTreeSet<DecisionKey>>,
}

impl DecisionIndex {
    const fn adopt(&mut self, generation: u64) -> Result<(), MembershipRefusal> {
        match self.generation {
            None => {
                self.generation = Some(generation);
                Ok(())
            }
            Some(current) if current == generation => Ok(()),
            Some(_) => Err(MembershipRefusal::ResumePending),
        }
    }

    fn ticket(&mut self) -> Result<u64, MembershipRefusal> {
        let ticket = self.next_ticket;
        self.next_ticket = ticket
            .checked_add(1)
            .ok_or(MembershipRefusal::CapacityExceeded { limit: usize::MAX })?;
        Ok(ticket)
    }

    /// The retained verdict for `key`, binding `pid` to it.
    fn known(&mut self, key: &DecisionKey, pid: u64) -> Option<Indexed> {
        let verdict = self.entries.get_mut(key).map(|entry| {
            entry.connections.insert(pid);
            entry.verdict.clone()
        })?;
        self.by_connection
            .entry(pid)
            .or_default()
            .insert(key.clone());
        Some(verdict)
    }

    /// Reserves `key` for one ask, refusing at `limit` before allocating.
    fn reserve(
        &mut self,
        key: &DecisionKey,
        mapped: &MappedChannel,
        pid: u64,
        limit: usize,
    ) -> Result<u64, MembershipRefusal> {
        if self.entries.len() >= limit {
            return Err(MembershipRefusal::CapacityExceeded { limit });
        }
        let ticket = self.ticket()?;
        let reached_by = [
            resource_key(&mapped.resource),
            resource_key(&mapped.workspace),
        ];
        for reach in &reached_by {
            self.by_resource
                .entry(reach.clone())
                .or_default()
                .insert(key.clone());
        }
        self.by_connection
            .entry(pid)
            .or_default()
            .insert(key.clone());
        self.entries.insert(
            key.clone(),
            Entry {
                verdict: Indexed::Applying { ticket },
                connections: BTreeSet::from([pid]),
                reached_by,
            },
        );
        Ok(ticket)
    }

    /// Re-reserves an existing binding for a change or resume ask.
    fn reapply(&mut self, key: &DecisionKey) -> Result<Option<u64>, MembershipRefusal> {
        if !self.entries.contains_key(key) {
            return Ok(None);
        }
        let ticket = self.ticket()?;
        if let Some(entry) = self.entries.get_mut(key) {
            entry.verdict = Indexed::Applying { ticket };
        }
        Ok(Some(ticket))
    }

    /// Settles the ask holding `ticket`; `false` when a later change or
    /// resume overtook it.
    fn settle(&mut self, key: &DecisionKey, ticket: u64, verdict: Indexed) -> bool {
        match self.entries.get_mut(key) {
            Some(entry) if entry.verdict == (Indexed::Applying { ticket }) => {
                entry.verdict = verdict;
                true
            }
            Some(_) | None => false,
        }
    }

    /// Forgets the binding still held by `ticket`'s ask.
    fn forget(&mut self, key: &DecisionKey, ticket: u64) {
        let held = self
            .entries
            .get(key)
            .is_some_and(|entry| entry.verdict == (Indexed::Applying { ticket }));
        if held {
            self.remove(key);
        }
    }

    fn remove(&mut self, key: &DecisionKey) {
        let Some(entry) = self.entries.remove(key) else {
            return;
        };
        for reach in &entry.reached_by {
            let emptied = self.by_resource.get_mut(reach).is_some_and(|keys| {
                keys.remove(key);
                keys.is_empty()
            });
            if emptied {
                self.by_resource.remove(reach);
            }
        }
        for pid in &entry.connections {
            let emptied = self.by_connection.get_mut(pid).is_some_and(|keys| {
                keys.remove(key);
                keys.is_empty()
            });
            if emptied {
                self.by_connection.remove(pid);
            }
        }
    }

    /// Releases every binding `pid` held; a binding no connection holds is
    /// removed. Visits only `pid`'s own bindings.
    fn release(&mut self, pid: u64) -> u64 {
        let Some(keys) = self.by_connection.remove(&pid) else {
            return 0;
        };
        let mut released = 0_u64;
        for key in keys {
            let unheld = self.entries.get_mut(&key).is_some_and(|entry| {
                entry.connections.remove(&pid);
                entry.connections.is_empty()
            });
            if unheld {
                self.remove(&key);
                released = released.saturating_add(1);
            }
        }
        released
    }
}

/// The question a page continuation is bound to.
#[derive(Clone, Debug, PartialEq, Eq)]
enum PageQuestion {
    Resources {
        workspace: String,
        subject: MembershipSubject,
        action: MembershipAction,
    },
    Recipients {
        channel: String,
        action: MembershipAction,
    },
}

/// An opaque page continuation, bound to its exact question, grant log and
/// generation; it grants nothing (ACCESS-006 R4).
#[derive(Clone, PartialEq, Eq)]
pub struct PageCursor {
    token: String,
    question: PageQuestion,
    generation: u64,
}

impl std::fmt::Debug for PageCursor {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("PageCursor")
            .field("token", &crate::auth_pass::ByteLength(self.token.len()))
            .field("generation", &self.generation)
            .finish_non_exhaustive()
    }
}

/// One page of the protected channels a holder may act on in a workspace.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ChannelPage {
    /// The liminal channels, in the provider's stable order.
    pub channels: Vec<String>,
    /// Rows the provider returned.
    pub returned: u64,
    /// Candidates the provider decided and left out.
    pub skipped: u64,
    /// Rows held for approval: a membership exists, the act still waits.
    pub held: u64,
    /// Rows naming a resource the product mapping does not list.
    pub unmapped: u64,
    /// The rows' encoded bytes.
    pub bytes: u64,
    /// The revision every row was decided at.
    pub revision: u64,
    /// Whether no row remains.
    pub complete: bool,
    /// The continuation, absent when complete.
    pub next: Option<PageCursor>,
}

/// One page of the subjects who may act on a protected channel.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RecipientPage {
    /// The subjects, in the provider's stable order.
    pub recipients: Vec<Subject>,
    /// Rows the provider returned.
    pub returned: u64,
    /// Candidates the provider decided and left out.
    pub skipped: u64,
    /// Rows held for approval.
    pub held: u64,
    /// The rows' encoded bytes.
    pub bytes: u64,
    /// The revision every row was decided at.
    pub revision: u64,
    /// Whether no row remains.
    pub complete: bool,
    /// The continuation, absent when complete.
    pub next: Option<PageCursor>,
}

/// A validated page's parts.
struct PageParts<'p> {
    rows: &'p [PageRow],
    skipped: u64,
    complete: bool,
    next: Option<String>,
    revision: u64,
    bytes: u64,
}

/// The membership gate: the product mapping, the provider seam, the
/// authority seam and the bounded decision index (ACCESS-006 R1, R2, R6).
pub struct MembershipGate {
    served: GrantLog,
    workspace_kind: String,
    channel_kind: String,
    map: ChannelMap,
    bounds: PageBounds,
    max_decisions: usize,
    connect_workspace: Option<String>,
    provider: Arc<dyn MembershipProvider>,
    authority: OnceLock<Arc<dyn MembershipAuthority>>,
    index: Mutex<DecisionIndex>,
    counters: Counters,
}

impl std::fmt::Debug for MembershipGate {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("MembershipGate")
            .field("served", &self.served)
            .field("channels", &self.map.channels.len())
            .field("bounds", &self.bounds)
            .field("max_decisions", &self.max_decisions)
            .field("authority", &self.authority.get().is_some())
            .finish_non_exhaustive()
    }
}

impl MembershipGate {
    /// Builds the gate from the validated product mapping over `provider`.
    /// No authority is installed: until one is, every protected act refuses
    /// [`MembershipRefusal::AuthorityMissing`].
    ///
    /// # Errors
    /// [`ServerError::ConfigValidation`] naming the first invalid member: a
    /// contract other than [`CONTRACT_VERSION`], an empty or control-bearing
    /// kind, log identity or mapping member, equal workspace and channel
    /// kinds, a zero bound, no protected channel, a channel or resource
    /// mapped twice, a conversation naming an unlisted channel, or a
    /// `connect_workspace` the mapping does not list.
    pub fn new(
        config: &MembershipConfig,
        provider: Arc<dyn MembershipProvider>,
    ) -> Result<Self, ServerError> {
        let bounds = validated_bounds(config)?;
        let map = channel_map(config)?;
        if let Some(workspace) = &config.connect_workspace
            && !map.workspaces.contains(workspace)
        {
            return Err(invalid(
                "auth.lys.membership.connect_workspace is not a mapped workspace",
            ));
        }
        Ok(Self {
            served: GrantLog {
                identity: config.log_identity.clone(),
                epoch: config.log_epoch,
            },
            workspace_kind: config.workspace_kind.clone(),
            channel_kind: config.channel_kind.clone(),
            map,
            bounds,
            max_decisions: config.max_decisions,
            connect_workspace: config.connect_workspace.clone(),
            provider,
            authority: OnceLock::new(),
            index: Mutex::new(DecisionIndex::default()),
            counters: Counters::default(),
        })
    }

    /// Installs the ACCESS-005 grant authority once.
    ///
    /// # Errors
    /// Returns the authority back when one is already installed, so one
    /// owner cannot be swapped for another under live bindings.
    pub fn install_authority(
        &self,
        authority: Arc<dyn MembershipAuthority>,
    ) -> Result<(), Arc<dyn MembershipAuthority>> {
        self.authority.set(authority)
    }

    /// The grant log this gate asks from.
    #[must_use]
    pub const fn served_log(&self) -> &GrantLog {
        &self.served
    }

    /// Whether `channel` is membership-protected.
    #[must_use]
    pub fn protects(&self, channel: &str) -> bool {
        self.map.channels.contains_key(channel)
    }

    /// The protected channel's placement, as the mapping states it.
    #[must_use]
    pub fn mapped(&self, channel: &str) -> Option<&MappedChannel> {
        self.map.channels.get(channel)
    }

    /// The protected channel a participant conversation is placed under, as
    /// the mapping states it; `None` is unmapped, never guessed.
    #[must_use]
    pub fn channel_for_conversation(&self, conversation: u64) -> Option<&str> {
        self.map
            .conversations
            .get(&conversation)
            .map(String::as_str)
    }

    /// Every count, as of now.
    #[must_use]
    pub fn counts(&self) -> MembershipCounts {
        self.counters.snapshot()
    }

    /// How many decisions the index retains.
    ///
    /// # Errors
    /// [`MembershipRefusal::IndexPoisoned`].
    pub fn retained(&self) -> Result<usize, MembershipRefusal> {
        Ok(self.lock()?.entries.len())
    }

    fn lock(&self) -> Result<MutexGuard<'_, DecisionIndex>, MembershipRefusal> {
        bump(&self.counters.locks);
        self.index
            .lock()
            .map_err(|_| MembershipRefusal::IndexPoisoned)
    }

    /// The authority's revision and generation while it is `Ready` on the
    /// served log; its exact cause otherwise.
    fn ready(&self) -> Result<(u64, u64), MembershipRefusal> {
        let authority = self
            .authority
            .get()
            .ok_or(MembershipRefusal::AuthorityMissing)?;
        match authority.state() {
            AuthorityState::Ready {
                log,
                revision,
                generation,
            } => {
                if log == self.served {
                    Ok((revision, generation))
                } else {
                    Err(MembershipRefusal::LogMismatch)
                }
            }
            AuthorityState::NotReady => Err(MembershipRefusal::NotReady),
            AuthorityState::Unavailable { cause } => Err(MembershipRefusal::Unavailable { cause }),
            AuthorityState::Gap { after, missing } => {
                Err(MembershipRefusal::Gap { after, missing })
            }
            AuthorityState::Reset => Err(MembershipRefusal::Reset),
        }
    }

    /// Decides `action` on `channel` for connection `pid`'s verified
    /// `holder`: [`Decided::NotProtected`] for a channel the mapping does
    /// not list; otherwise the current decision, from the index when it
    /// holds one at the authority's generation, asked of the provider once
    /// when it does not. The lock is never held across the ask.
    ///
    /// # Errors
    /// The exact [`MembershipRefusal`]: no verified holder, the authority's
    /// state, the index bound, the provider's refusal or failure, or an
    /// answer that is not the one asked.
    pub fn check(
        &self,
        pid: u64,
        holder: Option<&MembershipSubject>,
        channel: &str,
        action: MembershipAction,
    ) -> Result<Decided, MembershipRefusal> {
        let Some(mapped) = self.map.channels.get(channel) else {
            return Ok(Decided::NotProtected);
        };
        let decided = self.check_mapped(pid, holder, channel, mapped, action);
        if decided.is_err() {
            bump(&self.counters.refused);
        }
        decided
    }

    /// [`Self::check`] at a final data-release seam (ACCESS-006 R3), counted
    /// as a release check.
    ///
    /// # Errors
    /// As for [`Self::check`].
    pub fn check_release(
        &self,
        pid: u64,
        holder: Option<&MembershipSubject>,
        channel: &str,
        action: MembershipAction,
    ) -> Result<Decided, MembershipRefusal> {
        bump(&self.counters.release_checks);
        self.check(pid, holder, channel, action)
    }

    fn check_mapped(
        &self,
        pid: u64,
        holder: Option<&MembershipSubject>,
        channel: &str,
        mapped: &MappedChannel,
        action: MembershipAction,
    ) -> Result<Decided, MembershipRefusal> {
        let holder = holder.ok_or(MembershipRefusal::HolderUnverified)?;
        let (revision, generation) = self.ready()?;
        let key = DecisionKey {
            subject: holder.clone(),
            channel: channel.to_owned(),
            action,
        };
        let (ticket, at_least) = {
            let mut index = self.lock()?;
            index.adopt(generation)?;
            if let Some(verdict) = index.known(&key, pid) {
                drop(index);
                bump(&self.counters.local_lookups);
                return verdict.result();
            }
            let at_least = revision.max(index.applied);
            let ticket = index.reserve(&key, mapped, pid, self.max_decisions)?;
            drop(index);
            (ticket, at_least)
        };
        let asked = self.ask(holder, mapped, action, at_least);
        let mut index = self.lock()?;
        match asked {
            Ok(verdict) => {
                let current = index.generation == Some(generation)
                    && index.settle(&key, ticket, verdict.clone());
                if !current {
                    // A change or a resume overtook this ask: its answer is not
                    // the current one, and nothing is released on it.
                    index.forget(&key, ticket);
                    return Err(MembershipRefusal::ChangeApplying);
                }
                drop(index);
                verdict.result()
            }
            Err(refusal) => {
                index.forget(&key, ticket);
                drop(index);
                Err(refusal)
            }
        }
    }

    fn ask(
        &self,
        holder: &MembershipSubject,
        mapped: &MappedChannel,
        action: MembershipAction,
        at_least: u64,
    ) -> Result<Indexed, MembershipRefusal> {
        let request = MembershipRequest {
            contract: CONTRACT_VERSION,
            log: self.served.clone(),
            workspace: mapped.workspace.clone(),
            subject: holder.to_contract(),
            resource: mapped.resource.clone(),
            action: action.as_str().to_owned(),
            at_least,
        };
        bump(&self.counters.decisions_asked);
        let decision = self
            .provider
            .decide(&request)
            .map_err(|error| MembershipRefusal::Unavailable { cause: error.cause })?;
        judged(&request, &self.served, mapped, &decision)
    }

    /// Scoped guest admission at `Connect` (ACCESS-006 R1): when the mapping
    /// names a `connect_workspace`, the holder must hold a current grant
    /// somewhere in it. Admission opens no channel; every act after it is
    /// decided on its own. `Ok(None)` when no admission is configured.
    ///
    /// # Errors
    /// The authority's state, the provider's failure, a substituted or stale
    /// answer, or Lys's refusal by name (`membership_no_current_grant`).
    pub fn admit_connect(
        &self,
        holder: &MembershipSubject,
    ) -> Result<Option<u64>, MembershipRefusal> {
        let Some(workspace) = &self.connect_workspace else {
            return Ok(None);
        };
        let (revision, _) = self.ready()?;
        let request = AdmissionRequest {
            contract: CONTRACT_VERSION,
            log: self.served.clone(),
            workspace: Resource {
                kind: self.workspace_kind.clone(),
                id: workspace.clone(),
            },
            subject: holder.to_contract(),
            at_least: revision,
        };
        bump(&self.counters.admissions_asked);
        let decision = self
            .provider
            .admission(&request)
            .map_err(|error| MembershipRefusal::Unavailable { cause: error.cause })?;
        admitted(&request, &self.served, &decision).map(Some)
    }

    /// Releases every binding connection `pid` held (ACCESS-006 R6); a
    /// binding no connection holds leaves the index. A poisoned index is
    /// logged and left: it already refuses every check.
    pub fn release_connection(&self, pid: u64) {
        match self.lock() {
            Ok(mut index) => {
                let released = index.release(pid);
                drop(index);
                bump_by(&self.counters.released, released);
            }
            Err(refusal) => {
                tracing::warn!(
                    pid,
                    refusal = refusal.name(),
                    "membership bindings could not be released"
                );
            }
        }
    }

    /// Re-asks every key in `tickets` at `at_least`, settling each; returns
    /// how many settled and how many failed and were released.
    fn redecide(&self, tickets: Vec<(DecisionKey, u64)>, at_least: u64) -> (u64, u64) {
        let mut settled = 0_u64;
        let mut failed = 0_u64;
        for (key, ticket) in tickets {
            let asked = self
                .map
                .channels
                .get(&key.channel)
                .ok_or(MembershipRefusal::ChannelNotMapped)
                .and_then(|mapped| self.ask(&key.subject, mapped, key.action, at_least));
            let Ok(mut index) = self.lock() else {
                failed = failed.saturating_add(1);
                continue;
            };
            let kept = asked.is_ok_and(|verdict| index.settle(&key, ticket, verdict));
            if kept {
                settled = settled.saturating_add(1);
            } else {
                index.forget(&key, ticket);
                failed = failed.saturating_add(1);
            }
        }
        (settled, failed)
    }

    /// One page of the protected channels `holder` may `action` within
    /// `workspace` (ACCESS-006 R4): both bounds the operator's, every row
    /// checked against the question, the continuation bound to it.
    ///
    /// # Errors
    /// The exact [`MembershipRefusal`]: no verified holder, an unmapped
    /// workspace, the authority's state, a foreign or stale continuation,
    /// the provider's failure or refusal, or a page that is not the one
    /// asked (over bound, foreign or duplicate row, inconsistent counts).
    pub fn resource_page(
        &self,
        holder: Option<&MembershipSubject>,
        workspace: &str,
        action: MembershipAction,
        after: Option<&PageCursor>,
    ) -> Result<ChannelPage, MembershipRefusal> {
        let holder = holder.ok_or(MembershipRefusal::HolderUnverified)?;
        if !self.map.workspaces.contains(workspace) {
            return Err(MembershipRefusal::WorkspaceNotMapped);
        }
        let (revision, generation) = self.ready()?;
        let question = PageQuestion::Resources {
            workspace: workspace.to_owned(),
            subject: holder.clone(),
            action,
        };
        let request = ResourcePageRequest {
            contract: CONTRACT_VERSION,
            log: self.served.clone(),
            workspace: Resource {
                kind: self.workspace_kind.clone(),
                id: workspace.to_owned(),
            },
            subject: holder.to_contract(),
            kind: self.channel_kind.clone(),
            action: action.as_str().to_owned(),
            at_least: revision,
            bounds: self.bounds,
            after: continuation(after, &question, generation)?,
        };
        bump(&self.counters.pages_asked);
        let page = self
            .provider
            .resources(&request)
            .map_err(|error| MembershipRefusal::Unavailable { cause: error.cause })?;
        let parts = self.page_parts(&page, revision)?;
        let mut channels = Vec::with_capacity(parts.rows.len());
        let mut seen = BTreeSet::new();
        let mut held = 0_u64;
        let mut unmapped = 0_u64;
        for row in parts.rows {
            if !holder.is(&row.subject) || row.resource.kind != self.channel_kind {
                return Err(MembershipRefusal::PageForeignRow);
            }
            if !seen.insert(row.resource.id.as_str()) {
                return Err(MembershipRefusal::PageDuplicateRow);
            }
            let Some(channel) = self.map.by_resource.get(&row.resource.id) else {
                unmapped = unmapped.saturating_add(1);
                continue;
            };
            let placed_here = self
                .map
                .channels
                .get(channel)
                .is_some_and(|mapped| mapped.workspace.id == workspace);
            if !placed_here {
                return Err(MembershipRefusal::PageForeignRow);
            }
            if row.mode == Mode::Outright {
                channels.push(channel.clone());
            } else {
                held = held.saturating_add(1);
            }
        }
        Ok(ChannelPage {
            returned: u64::try_from(parts.rows.len())
                .map_err(|_| MembershipRefusal::PageInconsistent)?,
            channels,
            skipped: parts.skipped,
            held,
            unmapped,
            bytes: parts.bytes,
            revision: parts.revision,
            complete: parts.complete,
            next: parts.next.map(|token| PageCursor {
                token,
                question,
                generation,
            }),
        })
    }

    /// One page of the subjects who may `action` on the protected `channel`
    /// (ACCESS-006 R4), checked as [`Self::resource_page`] is.
    ///
    /// # Errors
    /// As for [`Self::resource_page`], with
    /// [`MembershipRefusal::ChannelNotMapped`] for an unmapped channel.
    pub fn recipient_page(
        &self,
        channel: &str,
        action: MembershipAction,
        after: Option<&PageCursor>,
    ) -> Result<RecipientPage, MembershipRefusal> {
        let mapped = self
            .map
            .channels
            .get(channel)
            .ok_or(MembershipRefusal::ChannelNotMapped)?;
        let (revision, generation) = self.ready()?;
        let question = PageQuestion::Recipients {
            channel: channel.to_owned(),
            action,
        };
        let request = RecipientPageRequest {
            contract: CONTRACT_VERSION,
            log: self.served.clone(),
            workspace: mapped.workspace.clone(),
            resource: mapped.resource.clone(),
            action: action.as_str().to_owned(),
            at_least: revision,
            bounds: self.bounds,
            after: continuation(after, &question, generation)?,
        };
        bump(&self.counters.pages_asked);
        let page = self
            .provider
            .recipients(&request)
            .map_err(|error| MembershipRefusal::Unavailable { cause: error.cause })?;
        let parts = self.page_parts(&page, revision)?;
        let mut recipients = Vec::with_capacity(parts.rows.len());
        let mut seen = BTreeSet::new();
        let mut held = 0_u64;
        for row in parts.rows {
            if row.resource != mapped.resource {
                return Err(MembershipRefusal::PageForeignRow);
            }
            if !seen.insert((row.subject.id.as_str(), row.subject.kind.as_str())) {
                return Err(MembershipRefusal::PageDuplicateRow);
            }
            if row.mode == Mode::Outright {
                recipients.push(row.subject.clone());
            } else {
                held = held.saturating_add(1);
            }
        }
        Ok(RecipientPage {
            returned: u64::try_from(parts.rows.len())
                .map_err(|_| MembershipRefusal::PageInconsistent)?,
            recipients,
            skipped: parts.skipped,
            held,
            bytes: parts.bytes,
            revision: parts.revision,
            complete: parts.complete,
            next: parts.next.map(|token| PageCursor {
                token,
                question,
                generation,
            }),
        })
    }

    fn page_parts<'p>(
        &self,
        page: &'p MembershipPage,
        at_least: u64,
    ) -> Result<PageParts<'p>, MembershipRefusal> {
        if page.contract != CONTRACT_VERSION {
            return Err(MembershipRefusal::ContractUnsupported {
                contract: page.contract,
            });
        }
        if page.log != self.served {
            return Err(MembershipRefusal::LogMismatch);
        }
        let (rows, returned, skipped, complete, next) = match &page.outcome {
            PageOutcome::Refused { refusal, .. } => return Err(page_refusal(refusal)),
            PageOutcome::Page {
                rows,
                returned,
                skipped,
                complete,
                next,
            } => (rows, *returned, *skipped, *complete, next),
        };
        let revision = page.revision.ok_or(MembershipRefusal::RevisionStale {
            required: at_least,
            decided: None,
        })?;
        if revision < at_least {
            return Err(MembershipRefusal::RevisionStale {
                required: at_least,
                decided: Some(revision),
            });
        }
        let count = u64::try_from(rows.len()).map_err(|_| MembershipRefusal::PageInconsistent)?;
        bump_by(&self.counters.page_rows, count);
        if count != returned || complete == next.is_some() {
            return Err(MembershipRefusal::PageInconsistent);
        }
        let bytes = encoded_bytes(rows)?;
        if count > u64::from(self.bounds.rows) || bytes > u64::from(self.bounds.bytes) {
            return Err(MembershipRefusal::PageOverBound { rows: count, bytes });
        }
        Ok(PageParts {
            rows,
            skipped,
            complete,
            next: next.clone(),
            revision,
            bytes,
        })
    }
}

impl MembershipListener for MembershipGate {
    fn changed(&self, change: &MembershipChange) -> Result<AppliedChange, MembershipRefusal> {
        if change.log != self.served {
            return Err(MembershipRefusal::LogMismatch);
        }
        let reached = resource_key(&change.resource);
        let tickets = {
            let mut index = self.lock()?;
            if change.revision <= index.applied {
                // A duplicate or older notification touches nothing, so no
                // withdrawn allow can return through it.
                return Ok(AppliedChange {
                    duplicate: true,
                    ..AppliedChange::default()
                });
            }
            index.applied = change.revision;
            let keys: Vec<DecisionKey> = index
                .by_resource
                .get(&reached)
                .map(|keys| {
                    keys.iter()
                        .filter(|key| {
                            change
                                .subject
                                .as_ref()
                                .is_none_or(|subject| key.subject.is(subject))
                        })
                        .cloned()
                        .collect()
                })
                .unwrap_or_default();
            let mut tickets = Vec::with_capacity(keys.len());
            for key in keys {
                if let Some(ticket) = index.reapply(&key)? {
                    tickets.push((key, ticket));
                }
            }
            tickets
        };
        let affected = u64::try_from(tickets.len()).unwrap_or(u64::MAX);
        bump_by(&self.counters.affected_visited, affected);
        let (redecided, failed) = self.redecide(tickets, change.revision);
        Ok(AppliedChange {
            duplicate: false,
            affected,
            redecided,
            failed,
        })
    }

    fn resumed(&self, generation: u64, revision: u64) -> Result<ResumeReport, MembershipRefusal> {
        let tickets = {
            let mut index = self.lock()?;
            let keys: Vec<DecisionKey> = index.entries.keys().cloned().collect();
            let mut tickets = Vec::with_capacity(keys.len());
            for key in keys {
                if let Some(ticket) = index.reapply(&key)? {
                    tickets.push((key, ticket));
                }
            }
            tickets
        };
        let rechecked = u64::try_from(tickets.len()).unwrap_or(u64::MAX);
        bump_by(&self.counters.resume_rechecks, rechecked);
        let (_, failed) = self.redecide(tickets, revision);
        let settled = failed == 0;
        if settled {
            let mut index = self.lock()?;
            index.generation = Some(generation);
            index.applied = index.applied.max(revision);
        }
        Ok(ResumeReport {
            rechecked,
            failed,
            settled,
        })
    }
}

/// The page continuation token for `after`, refusing one issued for another
/// question or at another generation.
fn continuation(
    after: Option<&PageCursor>,
    question: &PageQuestion,
    generation: u64,
) -> Result<Option<String>, MembershipRefusal> {
    let Some(cursor) = after else {
        return Ok(None);
    };
    if &cursor.question != question {
        return Err(MembershipRefusal::CursorForeign);
    }
    if cursor.generation != generation {
        return Err(MembershipRefusal::CursorStale);
    }
    Ok(Some(cursor.token.clone()))
}

fn page_refusal(refusal: &str) -> MembershipRefusal {
    match refusal {
        CURSOR_FOREIGN => MembershipRefusal::CursorForeign,
        CURSOR_RESET => MembershipRefusal::CursorStale,
        LOG_MISMATCH => MembershipRefusal::LogMismatch,
        other => MembershipRefusal::Refused {
            refusal: other.to_owned(),
        },
    }
}

fn encoded_bytes(rows: &[PageRow]) -> Result<u64, MembershipRefusal> {
    rows.iter().try_fold(0_u64, |total, row| {
        let length = serde_json::to_vec(row)
            .map(|encoded| encoded.len())
            .map_err(|_| MembershipRefusal::PageInconsistent)?;
        u64::try_from(length)
            .ok()
            .and_then(|length| total.checked_add(length))
            .ok_or(MembershipRefusal::PageInconsistent)
    })
}

/// Judges one provider answer against the exact request asked
/// (ACCESS-006 R1, R5): the contract, the grant log, the echoed request, the
/// revision fence and the allowing scope. A refusal decided before any
/// grant was asked carries no revision: it is returned, never retained, so
/// a transient provider state is never frozen into the index.
fn judged(
    request: &MembershipRequest,
    served: &GrantLog,
    mapped: &MappedChannel,
    decision: &MembershipDecision,
) -> Result<Indexed, MembershipRefusal> {
    if decision.contract != CONTRACT_VERSION {
        return Err(MembershipRefusal::ContractUnsupported {
            contract: decision.contract,
        });
    }
    if &decision.log != served {
        return Err(MembershipRefusal::LogMismatch);
    }
    if &decision.request != request {
        return Err(MembershipRefusal::AnswerSubstituted);
    }
    let Some(revision) = decision.revision else {
        return Err(match &decision.verdict {
            Verdict::Refused { refusal, .. } => page_refusal(refusal),
            Verdict::Allowed { .. } | Verdict::Held { .. } => MembershipRefusal::RevisionStale {
                required: request.at_least,
                decided: None,
            },
        });
    };
    if revision < request.at_least {
        return Err(MembershipRefusal::RevisionStale {
            required: request.at_least,
            decided: Some(revision),
        });
    }
    match &decision.verdict {
        Verdict::Allowed { grant, scope, .. } => {
            if scope == &mapped.resource || scope == &mapped.workspace {
                Ok(Indexed::Allowed {
                    grant: grant.clone(),
                    revision,
                })
            } else {
                Err(MembershipRefusal::ScopeOutsideMapping)
            }
        }
        Verdict::Held { mode, .. } => Ok(Indexed::Held { mode: *mode }),
        Verdict::Refused { refusal, .. } => Ok(Indexed::Refused {
            refusal: refusal.clone(),
        }),
    }
}

/// Judges one admission answer against the exact request asked.
fn admitted(
    request: &AdmissionRequest,
    served: &GrantLog,
    decision: &AdmissionDecision,
) -> Result<u64, MembershipRefusal> {
    if decision.contract != CONTRACT_VERSION {
        return Err(MembershipRefusal::ContractUnsupported {
            contract: decision.contract,
        });
    }
    if &decision.log != served {
        return Err(MembershipRefusal::LogMismatch);
    }
    if &decision.request != request {
        return Err(MembershipRefusal::AnswerSubstituted);
    }
    match (&decision.verdict, decision.revision) {
        (Verdict::Refused { refusal, .. }, _) => Err(page_refusal(refusal)),
        (Verdict::Held { mode, .. }, _) => Err(MembershipRefusal::Held { mode: *mode }),
        (Verdict::Allowed { .. }, Some(revision)) if revision >= request.at_least => Ok(revision),
        (Verdict::Allowed { .. }, decided) => Err(MembershipRefusal::RevisionStale {
            required: request.at_least,
            decided,
        }),
    }
}

/// One connection's membership binding (ACCESS-006 R1, R6): the gate, the
/// connection's id and the verified subject from its one `Connect`, or the
/// exact reason it has none.
///
/// Dropped with the connection's state on every
/// teardown path, releasing the bindings it held.
#[derive(Debug, Default)]
pub struct ConnectionMembership {
    bound: Option<BoundMembership>,
}

#[derive(Debug)]
struct BoundMembership {
    gate: Arc<MembershipGate>,
    pid: u64,
    holder: Result<MembershipSubject, MembershipRefusal>,
}

impl ConnectionMembership {
    /// Binds connection `pid` at `Connect`: the subject is the admitted Lys
    /// `session`'s holder, or none. Without a gate nothing is protected.
    #[must_use]
    pub fn bind(
        gate: Option<Arc<MembershipGate>>,
        pid: u64,
        session: Option<&dyn LysSession>,
    ) -> Self {
        Self {
            bound: gate.map(|gate| BoundMembership {
                gate,
                pid,
                holder: session.map_or(
                    Err(MembershipRefusal::HolderUnverified),
                    MembershipSubject::from_session,
                ),
            }),
        }
    }

    /// The gate, when one is bound.
    #[must_use]
    pub fn gate(&self) -> Option<&Arc<MembershipGate>> {
        self.bound.as_ref().map(|bound| &bound.gate)
    }

    /// The verified subject, or why there is none.
    ///
    /// # Errors
    /// [`MembershipRefusal::HolderUnverified`] without a gate or a holder;
    /// the subject's own refusal otherwise.
    pub fn holder(&self) -> Result<&MembershipSubject, MembershipRefusal> {
        self.bound
            .as_ref()
            .ok_or(MembershipRefusal::HolderUnverified)
            .and_then(|bound| bound.holder.as_ref().map_err(Clone::clone))
    }

    /// Scoped guest admission for this connection's holder at `Connect`.
    ///
    /// # Errors
    /// As for [`MembershipGate::admit_connect`]; the holder's own refusal.
    pub fn admit_connect(&self) -> Result<Option<u64>, MembershipRefusal> {
        let Some(bound) = &self.bound else {
            return Ok(None);
        };
        if bound.gate.connect_workspace.is_none() {
            return Ok(None);
        }
        let holder = bound.holder.as_ref().map_err(Clone::clone)?;
        bound.gate.admit_connect(holder)
    }

    /// Decides `action` on `channel` for this connection.
    ///
    /// # Errors
    /// As for [`MembershipGate::check`]; a protected channel on a connection
    /// with no verified subject refuses with that subject's cause.
    pub fn check(
        &self,
        channel: &str,
        action: MembershipAction,
    ) -> Result<Decided, MembershipRefusal> {
        self.decide(channel, action, false)
    }

    /// [`Self::check`] at a final data-release seam.
    ///
    /// # Errors
    /// As for [`Self::check`].
    pub fn release(
        &self,
        channel: &str,
        action: MembershipAction,
    ) -> Result<Decided, MembershipRefusal> {
        self.decide(channel, action, true)
    }

    /// [`Self::release`] for a participant conversation, on the channel the
    /// mapping places it under; an unmapped conversation is not protected.
    ///
    /// # Errors
    /// As for [`Self::check`].
    pub fn release_conversation(&self, conversation: u64) -> Result<Decided, MembershipRefusal> {
        let Some(bound) = &self.bound else {
            return Ok(Decided::NotProtected);
        };
        let Some(channel) = bound.gate.channel_for_conversation(conversation) else {
            return Ok(Decided::NotProtected);
        };
        self.decide(channel, MembershipAction::Read, true)
    }

    fn decide(
        &self,
        channel: &str,
        action: MembershipAction,
        release: bool,
    ) -> Result<Decided, MembershipRefusal> {
        let Some(bound) = &self.bound else {
            return Ok(Decided::NotProtected);
        };
        if !bound.gate.protects(channel) {
            return Ok(Decided::NotProtected);
        }
        let holder = match &bound.holder {
            Ok(holder) => holder,
            Err(refusal) => {
                bump(&bound.gate.counters.refused);
                return Err(refusal.clone());
            }
        };
        if release {
            bound
                .gate
                .check_release(bound.pid, Some(holder), channel, action)
        } else {
            bound.gate.check(bound.pid, Some(holder), channel, action)
        }
    }
}

impl ConnectionMembership {
    /// The final release check a queued delivery owner carries for
    /// `action` on the protected `channel` (ACCESS-006 R3): `None` when the
    /// channel is not protected or the connection has no verified subject
    /// (its admission was already refused, so no such owner exists).
    #[must_use]
    pub fn release_check(&self, channel: &str, action: MembershipAction) -> Option<ReleaseCheck> {
        let bound = self.bound.as_ref()?;
        if !bound.gate.protects(channel) {
            return None;
        }
        let holder = bound.holder.as_ref().ok()?;
        Some(ReleaseCheck {
            gate: Arc::clone(&bound.gate),
            pid: bound.pid,
            holder: holder.clone(),
            channel: channel.to_owned(),
            action,
        })
    }
}

/// The membership check a queued delivery owner carries to its final release
/// (ACCESS-006 R3).
///
/// It is the same gate, connection, verified subject, channel and action as
/// its admission, decided again at release against the current generation,
/// so a deny applied after admission releases nothing.
#[derive(Clone, Debug)]
pub struct ReleaseCheck {
    gate: Arc<MembershipGate>,
    pid: u64,
    holder: MembershipSubject,
    channel: String,
    action: MembershipAction,
}

impl ReleaseCheck {
    /// Decides the release now.
    ///
    /// # Errors
    /// As for [`MembershipGate::check`].
    pub fn check(&self) -> Result<Decided, MembershipRefusal> {
        self.gate
            .check_release(self.pid, Some(&self.holder), &self.channel, self.action)
    }
}

impl Drop for ConnectionMembership {
    fn drop(&mut self) {
        if let Some(bound) = &self.bound {
            bound.gate.release_connection(bound.pid);
        }
    }
}

/// Builds the membership gate `[auth.lys.membership]` configures over the
/// live HTTP provider at `issuer`, reading the app credential once.
///
/// # Errors
/// [`ServerError::ConfigValidation`] for an unreadable credential, a server
/// started outside its tokio runtime, a refused issuer, or an invalid
/// mapping (see [`MembershipGate::new`]).
pub fn configured_gate(
    config: &MembershipConfig,
    issuer: &str,
) -> Result<Arc<MembershipGate>, ServerError> {
    let raw = zeroize::Zeroizing::new(std::fs::read_to_string(&config.credential_path).map_err(
        |error| {
            invalid(format!(
                "auth.lys.membership.credential_path {}: {}",
                config.credential_path.display(),
                error.kind()
            ))
        },
    )?);
    let credential = zeroize::Zeroizing::new(raw.trim().to_owned());
    let handle = tokio::runtime::Handle::try_current()
        .map_err(|_| invalid("auth.lys.membership needs the server's tokio runtime at start"))?;
    let provider = HttpMembershipProvider::new(issuer, credential, handle).map_err(|error| {
        invalid(format!(
            "auth.lys.membership provider refused: {}",
            error.cause
        ))
    })?;
    MembershipGate::new(config, Arc::new(provider)).map(Arc::new)
}

/// Fixtures for the membership tests: a seeded world, a settable authority
/// and a verified session. Never compiled into a production build.
#[cfg(any(test, feature = "test-support"))]
pub mod fixture;

#[cfg(test)]
#[path = "auth_membership_tests.rs"]
mod tests;