didbot_serve/oauth/

decision.rs

//! Decision records: what a pushed authorization request becomes, and what
//! an agent answers.
//!
//! `plan/oauth.md`'s hard part is who consents. The answer this module
//! carries is that the pushed request itself is the question: at PAR, once
//! the client is known and `login_hint` has been resolved to an account,
//! this server works out what it would grant and writes that down as a
//! [`DecisionRecord`] the named account can read, approve or decline over
//! `bot.did.*`. Nothing about it needs a browser, and nothing about it
//! forbids one — `GET /oauth/authorize` renders the same record for a person.
//!
//! # What the record says, and what it keeps to itself
//!
//! [`DecisionRecord::to_json`] is the whole wire shape. The client's own copy
//! of its name, its home page and its logo is kept ([`ClientSelf`]) and never
//! serialized: `plan/app-allowlist.md`'s "record the client's own copy; do
//! not show it", because showing an untrusted client's self-description is
//! how impersonation works. What an agent sees about the client is the
//! origin it served its metadata from, the content key of that metadata, and
//! whether this account has approved that key before.
//!
//! # Narrowing is never silent
//!
//! A [`Verdict`] is one of three answers, and two of them say something was
//! taken away. [`Verdict::Narrow`] names the grant *and* the atoms the
//! ceiling did not give in full. See
//! [`crate::oauth::scope::ScopeSet::narrow`].
//!
//! An approval covers what the verdict granted, and that is what the login
//! is issued at. A token may do that set as the ceiling in force grants it
//! ([`crate::oauth::token::in_force`]), so a ceiling tightened after the
//! approval takes a granted atom back. A cut atom stays cut: widening the
//! ceiling cannot add one to a login the agent approved without it.
//!
//! # Nothing here is durable
//!
//! A record lives as long as the pushed request it was minted for
//! ([`crate::oauth::par::REQUEST_URI_TTL`]), answered or not, and is swept
//! when it expires: `GET /oauth/authorize` answers from it until then. An
//! approval whose code a browser carries holds its redirect here too, until
//! the page hands it to a browser signed in as the account
//! ([`DecisionStore::collect`]). [`MemoryDecisionStore`] is bounded twice
//! over — per account and in total — because PAR is unauthenticated, so an
//! unbounded queue of records would be a way to spend this server's memory
//! without holding any credential.
//!
//! # A remembered approval
//!
//! The account's approval of a client is remembered, per account and
//! [`ClientKey`], at the grant it approved, for [`REMEMBERED_TTL`]
//! ([`DecisionStore::remember`]). `GET /oauth/authorize` reads it to confirm
//! a later sign-in without asking; see
//! [`crate::oauth::authorize::confirms_unasked`]. A confirmation does not
//! write one, so it never renews itself. It lives in memory, so a restart
//! forgets every one. See `plan/remembered-consent.md`.

use std::collections::{HashMap, HashSet};
use std::sync::{Arc, RwLock};

use didbot_schema::format::datetime;
use rand::rngs::OsRng;
use rand::TryRngCore;
use serde_json::{json, Value};
use time::OffsetDateTime;
use tokio::sync::Notify;

use crate::oauth::client_metadata::{ClientKey, ClientMetadataDocument};
use crate::oauth::consent::ConsentReference;
use crate::oauth::scope::{Scope, ScopeSet};

/// How many decisions one account may have awaiting an answer.
///
/// A client that pushes a request and never comes back leaves one of these
/// behind for two minutes. Eight is more than any honest sign-in flow needs
/// at once and few enough that a spree against one account cannot crowd out
/// every other account's decisions.
///
/// Past it, the account's oldest undecided record is displaced rather than
/// the new push dropped — see [`DecisionStore::put`].
pub const DEFAULT_PENDING_PER_ACCOUNT: usize = 8;

/// How many decisions this server may hold in total. An answered one counts
/// until its pushed request expires.
pub const DEFAULT_PENDING_TOTAL: usize = 512;

/// How many `(account, client key)` approvals [`MemoryDecisionStore`]
/// remembers for `DecisionRecord::first_time`.
///
/// The set is cleared rather than trimmed when it fills, so every client
/// looks new again for a while. That is the failing-safe direction: `true`
/// tells an agent to look harder at a client it may in fact have seen, and
/// `false` would tell it not to bother with one it has not.
const SEEN_CLIENTS_MAX: usize = 4096;

/// How long the account's approval of a client is remembered: as long as
/// the login it approved can last.
pub const REMEMBERED_TTL: time::Duration = crate::oauth::token::REFRESH_TTL;

/// How many random bytes name one store's epoch; see [`Cursor`].
const EPOCH_BYTES: usize = 8;

/// A place in one store's arrival order, and which store's order it is.
///
/// `"<epoch>.<sequence>"` on the wire, opaque to a caller. The epoch is
/// minted when the store is created, so a cursor a daemon is holding across
/// a restart names an order that no longer exists and is read as "from the
/// beginning" rather than as a sequence this store has not reached yet.
///
/// Without it a restart strands every client: the counter begins again at
/// zero, a daemon holding `3` asks for everything after `3`, and the answer
/// is empty for as long as it takes the new store to mint four records —
/// which for a quiet account is never. The failure is silent on both sides,
/// which is what makes it worth a self-describing cursor rather than a
/// durable counter.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Cursor {
    /// Which store's arrival order this names.
    pub epoch: String,
    /// How far into that order the caller has read.
    pub sequence: u64,
}

impl Cursor {
    /// Reads a cursor a caller sent back, against the epoch in force now.
    ///
    /// Anything that is not this store's order — a different epoch, a
    /// malformed string, an absent cursor — is the beginning of this one.
    /// A caller is never told its cursor was discarded, because there is
    /// nothing it could usefully do differently: the answer it gets carries
    /// the current epoch, and sending that back works.
    pub fn read(raw: Option<&str>, epoch: &str) -> u64 {
        let Some((sent_epoch, sequence)) = raw.and_then(|raw| raw.split_once('.')) else {
            return 0;
        };
        if sent_epoch != epoch {
            return 0;
        }
        sequence.parse().unwrap_or(0)
    }

    /// The wire form.
    pub fn wire(epoch: &str, sequence: u64) -> String {
        format!("{epoch}.{sequence}")
    }
}

/// The rule a ceiling narrowing or refusal is attributed to.
///
/// `plan/scope-policy.md`'s ceiling is not one of the operator's numbered
/// policies — it is this server applying the enumeration it was configured
/// with — so it names itself rather than borrowing a policy id.
pub const CEILING_RULE: &str = "scope-ceiling";

/// What this server decided about one authorization request.
///
/// Derived, not chosen: [`verdict_for`] computes it from the requested scope
/// and the ceiling, and a [`didbot_pds::policy::PolicyGate`] refusal
/// overrides it. A policy denies and never grants (`plan/policy.md`), so
/// there is no variant here that widens anything.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Verdict {
    /// The ceiling grants the request exactly as asked.
    Allow,
    /// The ceiling grants part of the request.
    Narrow {
        /// What the ceiling grants now.
        granted: ScopeSet,
        /// The requested atoms no granted atom covers in full — see
        /// [`ScopeSet::narrow`]. Never part of the approval, so a login
        /// approved against this verdict never carries one.
        cut: ScopeSet,
        /// What did the narrowing: [`CEILING_RULE`], or a policy's own id.
        rule: String,
    },
    /// Nothing is granted.
    Deny {
        /// Why, in the refusing party's own words.
        reason: String,
        /// What refused: [`CEILING_RULE`], or a policy's own id.
        rule: String,
    },
}

impl Verdict {
    /// The wire word for this verdict's kind.
    pub fn kind(&self) -> &'static str {
        match self {
            Verdict::Allow => "allow",
            Verdict::Narrow { .. } => "narrow",
            Verdict::Deny { .. } => "deny",
        }
    }

    /// What a token issued under the ceiling this verdict was reached with
    /// would carry, given the scope that was requested.
    ///
    /// [`Verdict::Allow`] grants the request whole, which is what an empty
    /// `cut` means; [`Verdict::Deny`] grants nothing, which is why an
    /// approval of a denied record can never mint a code.
    pub fn granted(&self, requested: &ScopeSet) -> ScopeSet {
        match self {
            Verdict::Allow => requested.clone(),
            Verdict::Narrow { granted, .. } => granted.clone(),
            Verdict::Deny { .. } => ScopeSet::default(),
        }
    }

    /// Whether this verdict leaves anything to approve.
    pub fn is_deny(&self) -> bool {
        matches!(self, Verdict::Deny { .. })
    }

    fn to_json(&self) -> Value {
        match self {
            Verdict::Allow => json!({ "kind": "allow" }),
            Verdict::Narrow { granted, cut, rule } => json!({
                "kind": "narrow",
                "granted": atoms(granted),
                "cut": atoms(cut),
                "rule": rule,
            }),
            Verdict::Deny { reason, rule } => {
                json!({ "kind": "deny", "reason": reason, "rule": rule })
            }
        }
    }
}

/// Where one decision record is in its life.
///
/// The whole state machine, in one place: a record is minted
/// [`DecisionState::Pending`] at PAR and leaves that state exactly once —
/// the named account approves it, the named account declines it, or its
/// pushed request outlives its window and a sweep marks it
/// [`DecisionState::Expired`]. There is no path back to `Pending` and no
/// transition between the three terminal states. Only a `Pending` record is
/// listed to an agent, and only a `Pending` record can be approved.
///
/// A record can also stop existing without reaching any of the three: the
/// account's newest push displaces its oldest undecided record when the
/// account is at its bound ([`DecisionStore::put`]). A displaced record is
/// removed rather than settled, so everything downstream sees exactly what
/// it sees for one that expired and was swept.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DecisionState {
    /// Minted, and nobody has answered.
    Pending,
    /// The named account approved it, and a code was issued.
    Approved,
    /// The named account said no.
    Declined,
    /// Nobody answered before the pushed request's window closed.
    Expired,
}

impl DecisionState {
    /// The wire word for this state.
    pub fn as_str(self) -> &'static str {
        match self {
            DecisionState::Pending => "pending",
            DecisionState::Approved => "approved",
            DecisionState::Declined => "declined",
            DecisionState::Expired => "expired",
        }
    }
}

/// The client's own account of itself, kept for the audit trail and never
/// serialized to any caller.
///
/// `plan/app-allowlist.md`: the profile says an untrusted client's
/// `client_name`, `client_uri` and `logo_uri` are not displayed. Nothing in
/// [`DecisionRecord::to_json`] reads this type.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ClientSelf {
    /// The client's own name for itself.
    pub name: Option<String>,
    /// The client's own home page.
    pub uri: Option<String>,
    /// The client's own logo.
    pub logo: Option<String>,
}

impl ClientSelf {
    /// Reads the three fields out of a validated metadata document.
    pub fn of(document: &ClientMetadataDocument) -> Self {
        Self {
            name: document.client_name.clone(),
            uri: document.client_uri.clone(),
            logo: document.logo_uri.clone(),
        }
    }
}

/// One authorization request, as this server decided about it.
#[derive(Debug, Clone)]
pub struct DecisionRecord {
    /// The pushed request this decides. The record's identity.
    pub request_uri: String,
    /// The account the request named, resolved at PAR. Only this account may
    /// see, approve or decline the record.
    pub account: String,
    /// Where the client served its metadata document from — scheme, host and
    /// port of `client_id`, which is the only part of it that is a fact
    /// about the world.
    pub client_origin: String,
    /// The content key of that document; see
    /// [`crate::oauth::client_metadata::ClientKey`].
    pub client_key: ClientKey,
    /// Whether this account has not approved a request from this client key
    /// since this server started. Held in memory only, and forgotten
    /// wholesale once this server has recorded more approvals than it keeps
    /// track of — which makes every client look new again, the direction
    /// that fails safe.
    ///
    /// Always `true` for a client at the local origin
    /// ([`didbot_pds::policy::is_local_origin`]). Any process on that
    /// client's machine can send the same `client_id`, so an earlier
    /// approval says nothing about who sent this one.
    pub first_time: bool,
    /// The client's own copy of its name, home page and logo. Logged, never
    /// serialized; see [`ClientSelf`].
    pub client_self: ClientSelf,
    /// The scope the client asked for.
    pub requested: ScopeSet,
    /// What this server decided.
    pub verdict: Verdict,
    /// The one-time reference the named account approves or declines with.
    ///
    /// The same [`ConsentReference`] `plan/oauth.md`'s consent seam already
    /// mints, reused rather than joined by a second credential. Absent when
    /// the verdict is [`Verdict::Deny`]: there is nothing to approve, so
    /// nothing is minted that could approve it.
    pub token: Option<ConsentReference>,
    /// When this record was minted.
    pub created_at: OffsetDateTime,
    /// When it stops being answerable — the pushed request's own expiry,
    /// which is always sooner than the consent reference's, so this is the
    /// deadline that binds.
    pub expires_at: OffsetDateTime,
    /// Where it is in its life; see [`DecisionState`].
    pub state: DecisionState,
    /// Where it sits in this store's arrival order, for the long poll's
    /// cursor. Assigned by [`DecisionStore::put`].
    pub sequence: u64,
}

impl DecisionRecord {
    /// What the ceiling granted of the request when this record was minted.
    pub fn granted(&self) -> ScopeSet {
        self.verdict.granted(&self.requested)
    }

    /// Whether this record is still answerable at `now`.
    pub fn is_live(&self, now: OffsetDateTime) -> bool {
        self.state == DecisionState::Pending && now < self.expires_at
    }

    /// The record as the daemon and the consent page read it.
    ///
    /// [`ClientSelf`] is not in here, and must not be: see this module's doc
    /// and `plan/app-allowlist.md`.
    pub fn to_json(&self) -> Value {
        let mut out = json!({
            "requestUri": self.request_uri,
            "account": self.account,
            "client": {
                "origin": self.client_origin,
                "key": self.client_key.0,
                "firstTime": self.first_time,
            },
            "requested": atoms(&self.requested),
            "verdict": self.verdict.to_json(),
            "createdAt": datetime::render(self.created_at),
            "expiresAt": datetime::render(self.expires_at),
            "state": self.state.as_str(),
        });
        if let Some(token) = &self.token {
            out["token"] = json!(token.0);
        }
        out
    }
}

/// A scope set as the array of atoms the wire carries.
fn atoms(scopes: &ScopeSet) -> Vec<String> {
    scopes.0.iter().map(ToString::to_string).collect()
}

/// The account's latest approval of one client, remembered.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Remembered {
    /// The client it approved, so an operator's `end-logins` can name it.
    pub client_id: String,
    /// What it granted. A later sign-in is confirmed only at this grant or
    /// a narrower one.
    pub granted: ScopeSet,
    /// When it is forgotten.
    pub expires_at: OffsetDateTime,
}

/// Why a record could not be stored.
///
/// One reason, and it is server state — identity-independent, true for every
/// caller alike — which is what makes it safe to answer truthfully from an
/// unauthenticated endpoint: `oauth::par::ParError::TooManyPending`'s wire
/// body exists in exactly one shape. An account's own bound never refuses a
/// push, because a `403` that fires only for accounts already under load is
/// a `login_hint` oracle by another name; it displaces that account's oldest
/// undecided record instead — see [`DecisionStore::put`].
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum DecisionRefused {
    /// This server already holds as many undecided records as it may.
    #[error("this server already has {0} authorization requests awaiting an answer")]
    TooMany(usize),
}

/// The record one push displaced to make room for its own, and what must be
/// undone for it.
///
/// The caller owns the undo because the state is the caller's: the pushed
/// request lives in [`crate::oauth::par::ParStore`] and the reference in
/// [`crate::oauth::consent::ConsentStore`], neither of which this module can
/// reach. Leaving either behind is memory nothing will ever present,
/// reachable from an unauthenticated endpoint.
#[derive(Debug, Clone)]
pub struct Displaced {
    /// The pushed request the displaced record decided.
    pub request_uri: String,
    /// The reference that would have approved it, if it had one.
    pub token: Option<ConsentReference>,
}

/// What [`DecisionStore::put`] stored, and what it displaced to store it.
#[derive(Debug, Clone)]
pub struct Stored {
    /// The record as stored, with its `sequence` and `first_time` assigned.
    pub record: DecisionRecord,
    /// The account's oldest undecided record, if this push was over the
    /// account's bound and displaced one.
    pub displaced: Option<Displaced>,
}

/// Where decision records live between PAR and an answer.
pub trait DecisionStore: Send + Sync {
    /// Stores a freshly minted record, assigning its `sequence` and its
    /// `first_time`, and hands back what was stored.
    ///
    /// Sweeps expired records first, so a bound is measured against what is
    /// actually still answerable rather than against everything ever minted.
    ///
    /// **An account at its own bound displaces, it does not refuse.** PAR is
    /// unauthenticated and `login_hint` names whoever the caller likes, so a
    /// bound that dropped the newest push let a stranger keep an account from
    /// signing in at all: fill its eight slots every two minutes and the
    /// account's own sign-ins never reach it. Displacing the oldest undecided
    /// record instead keeps the bound — the account still holds no more than
    /// it may — while leaving an account's freshest attempt always answerable,
    /// which is the one an agent is actually waiting on. What that costs is in
    /// [`Displaced`], for the caller to undo.
    ///
    /// The server-wide bound still refuses a push that would grow this store;
    /// see [`DecisionRefused`]. A displacing push does not grow it, and is
    /// not measured against it: an account's own room is its own, so a full
    /// store cannot take an account's freshest attempt away from it.
    ///
    /// The account's bound counts its undecided records, and only one of
    /// those is ever displaced. An answer, and the redirect an approval
    /// holds, stay until the pushed request expires; the server-wide bound
    /// counts them.
    fn put(&self, record: DecisionRecord) -> Result<Stored, DecisionRefused>;

    /// Whether the server-wide bound has room right now, without writing
    /// anything and without checking any account's own bound — there is no
    /// account to check one against here.
    ///
    /// `oauth::par::push_for_unresolved_hint` calls this so a `login_hint`
    /// naming nobody answers exactly the `403` a real account's push would
    /// when this server itself is out of room — a fact about server load,
    /// which is fine to say identically to everyone — without that push
    /// ever costing the slot it checked for, since nothing is stored under
    /// it either way.
    fn check_room(&self) -> Result<(), DecisionRefused>;

    /// One record by the pushed request it decides, whatever state it is in.
    fn get(&self, request_uri: &str) -> Option<DecisionRecord>;

    /// One record by the token that approves it, if that token is still the
    /// token of a live record.
    fn by_token(&self, token: &str) -> Option<DecisionRecord>;

    /// This account's live records that arrived after `cursor`, oldest
    /// first, and the cursor a caller should send next.
    ///
    /// The cursor advances past records for other accounts too, so a poll
    /// that returns nothing still makes progress. A cursor this store did
    /// not mint — see [`Cursor`] — is read as the beginning of its order,
    /// so a client that outlived a restart is handed everything rather than
    /// nothing.
    fn pending_for(&self, account: &str, cursor: Option<&str>) -> (Vec<DecisionRecord>, String);

    /// Moves a live record out of [`DecisionState::Pending`], returning it as
    /// it was just before. `None` if it is unknown, expired or already
    /// settled — which is what makes an approval single-use even before the
    /// consent reference is consumed.
    ///
    /// An approval is what makes the client key no longer the account's
    /// `first_time`.
    fn settle(&self, request_uri: &str, state: DecisionState) -> Option<DecisionRecord>;

    /// [`DecisionStore::settle`]s a live record as
    /// [`DecisionState::Approved`] and holds `redirect` for a browser signed
    /// in as the account, in one step, so no load of the page finds the
    /// approval without it.
    fn approve(&self, request_uri: &str, redirect: Option<String>) -> Option<DecisionRecord>;

    /// Whether an approval holds a redirect for a browser.
    fn holds(&self, request_uri: &str) -> bool;

    /// Hands over the redirect an approval held for a browser, once.
    fn collect(&self, request_uri: &str) -> Option<String>;

    /// Drops every record whose pushed request has expired, answered or not,
    /// and the redirect it held.
    fn sweep(&self, now: OffsetDateTime);

    /// Woken every time a record arrives, so a long poll does not have to
    /// spin.
    fn arrivals(&self) -> Arc<Notify>;

    /// This store's arrival-order epoch; see [`Cursor`].
    fn epoch(&self) -> String;

    /// Remembers the account's approval of `client_key` at `granted`, until
    /// [`REMEMBERED_TTL`] from `now`. Replaces the one it had, rather than
    /// merging scopes into it.
    fn remember(
        &self,
        account: &str,
        client_key: &ClientKey,
        client_id: &str,
        granted: ScopeSet,
        now: OffsetDateTime,
    );

    /// The account's live remembered approval of `client_key`, if any.
    fn remembered(
        &self,
        account: &str,
        client_key: &ClientKey,
        now: OffsetDateTime,
    ) -> Option<Remembered>;

    /// Forgets the remembered approvals of `client_id`, on `account` or on
    /// every account, and says how many it forgot.
    fn forget_client(&self, client_id: &str, account: Option<&str>) -> usize;

    /// Forgets every remembered approval the account made.
    fn forget_account(&self, account: &str);

    /// Forgets every remembered approval.
    fn forget_all(&self);
}

/// The in-process decision store.
#[derive(Debug)]
pub struct MemoryDecisionStore {
    inner: RwLock<Inner>,
    per_account: usize,
    total: usize,
    arrivals: Arc<Notify>,
    epoch: String,
}

#[derive(Debug, Default)]
struct Inner {
    records: HashMap<String, DecisionRecord>,
    tokens: HashMap<String, String>,
    /// The redirect each approval holds for a browser, by `request_uri`.
    redirects: HashMap<String, String>,
    seen: HashSet<(String, String)>,
    /// Each account's remembered approval of a client, by the account's DID
    /// and the client's content key.
    remembered: HashMap<(String, String), Remembered>,
    sequence: u64,
}

impl Default for MemoryDecisionStore {
    fn default() -> Self {
        Self::with_bounds(DEFAULT_PENDING_PER_ACCOUNT, DEFAULT_PENDING_TOTAL)
    }
}

impl MemoryDecisionStore {
    /// An empty store with the default bounds.
    pub fn new() -> Self {
        Self::default()
    }

    /// An empty store with bounds of a deployment's own choosing; see
    /// `didbot_config::sections::OAuthSection`.
    pub fn with_bounds(per_account: usize, total: usize) -> Self {
        let mut bytes = [0u8; EPOCH_BYTES];
        OsRng
            .try_fill_bytes(&mut bytes)
            .expect("the OS random source is available");
        Self {
            inner: RwLock::new(Inner::default()),
            per_account,
            total,
            arrivals: Arc::new(Notify::new()),
            epoch: hex::encode(bytes),
        }
    }
}

impl Inner {
    /// Expires what has run out and forgets it, answered or not.
    fn sweep(&mut self, now: OffsetDateTime) {
        for record in self.records.values_mut() {
            if record.state == DecisionState::Pending && now >= record.expires_at {
                record.state = DecisionState::Expired;
            }
        }
        self.records.retain(|_, record| now < record.expires_at);
        let records = &self.records;
        self.redirects
            .retain(|request_uri, _| records.contains_key(request_uri));
        let live: HashSet<String> = self
            .records
            .values()
            .filter(|record| record.state == DecisionState::Pending)
            .filter_map(|record| record.token.as_ref().map(|token| token.0.clone()))
            .collect();
        self.tokens.retain(|token, _| live.contains(token));
        self.remembered
            .retain(|_, remembered| now < remembered.expires_at);
    }

    /// How many of this account's records await an answer. `sweep` has
    /// already run, so each of them is still live.
    fn live_for(&self, account: &str) -> usize {
        self.records
            .values()
            .filter(|record| record.account == account && record.state == DecisionState::Pending)
            .count()
    }

    /// Removes this account's oldest undecided record — arrival order, which
    /// `sequence` is — and says what it was, for the caller to undo.
    ///
    /// An answered record is never the one displaced: the bound counts only
    /// undecided records, and an answer stays for the authorize page to
    /// report until its request expires.
    fn displace_oldest_for(&mut self, account: &str) -> Option<Displaced> {
        let oldest = self
            .records
            .values()
            .filter(|record| record.account == account && record.state == DecisionState::Pending)
            .min_by_key(|record| record.sequence)?
            .request_uri
            .clone();
        let record = self.records.remove(&oldest)?;
        if let Some(token) = &record.token {
            self.tokens.remove(&token.0);
        }
        Some(Displaced {
            request_uri: record.request_uri,
            token: record.token,
        })
    }

    /// The server-wide bound, read from one place so
    /// [`DecisionStore::put`] and [`DecisionStore::check_room`] agree on it.
    fn has_room(&self, total: usize) -> bool {
        self.records.len() < total
    }

    /// [`DecisionStore::settle`], holding `redirect` when the answer is an
    /// approval that carries one.
    fn settle(
        &mut self,
        request_uri: &str,
        state: DecisionState,
        redirect: Option<String>,
        now: OffsetDateTime,
    ) -> Option<DecisionRecord> {
        let record = self.records.get_mut(request_uri)?;
        if !record.is_live(now) {
            return None;
        }
        let settled = record.clone();
        record.state = state;
        if let Some(token) = &settled.token {
            self.tokens.remove(&token.0);
        }
        if state == DecisionState::Approved {
            if self.seen.len() >= SEEN_CLIENTS_MAX {
                self.seen.clear();
            }
            self.seen
                .insert((settled.account.clone(), settled.client_key.0.clone()));
            if let Some(redirect) = redirect {
                self.redirects.insert(request_uri.to_owned(), redirect);
            }
        }
        Some(settled)
    }
}

impl DecisionStore for MemoryDecisionStore {
    fn put(&self, mut record: DecisionRecord) -> Result<Stored, DecisionRefused> {
        let mut inner = self.inner.write().unwrap_or_else(|p| p.into_inner());
        inner.sweep(OffsetDateTime::now_utc());

        // One record in, at most one out: every put adds exactly one, so an
        // account is only ever one over its bound by the time it is here.
        let displaced = (inner.live_for(&record.account) >= self.per_account)
            .then(|| inner.displace_oldest_for(&record.account))
            .flatten();
        // The server-wide bound is asked only of a push that will grow the
        // store. A displacing push does not: it takes one of this account's
        // own slots back first, so the count after is the count before.
        // Asking first made a full store refuse the very account whose slots
        // a stranger had filled — the lockout this displacement exists to
        // close, reached by the other bound instead.
        if displaced.is_none() && !inner.has_room(self.total) {
            return Err(DecisionRefused::TooMany(self.total));
        }

        // Read, not recorded: a push is anyone's to make, naming any
        // account. The account's own approval records it; see `settle`.
        record.first_time = didbot_pds::policy::is_local_origin(&record.client_origin)
            || !inner
                .seen
                .contains(&(record.account.clone(), record.client_key.0.clone()));

        inner.sequence += 1;
        record.sequence = inner.sequence;
        if let Some(token) = &record.token {
            inner
                .tokens
                .insert(token.0.clone(), record.request_uri.clone());
        }
        inner
            .records
            .insert(record.request_uri.clone(), record.clone());
        drop(inner);
        self.arrivals.notify_waiters();
        Ok(Stored { record, displaced })
    }

    fn check_room(&self) -> Result<(), DecisionRefused> {
        let mut inner = self.inner.write().unwrap_or_else(|p| p.into_inner());
        inner.sweep(OffsetDateTime::now_utc());
        if inner.has_room(self.total) {
            Ok(())
        } else {
            Err(DecisionRefused::TooMany(self.total))
        }
    }

    fn get(&self, request_uri: &str) -> Option<DecisionRecord> {
        let inner = self.inner.read().unwrap_or_else(|p| p.into_inner());
        inner.records.get(request_uri).cloned()
    }

    fn by_token(&self, token: &str) -> Option<DecisionRecord> {
        let inner = self.inner.read().unwrap_or_else(|p| p.into_inner());
        let request_uri = inner.tokens.get(token)?;
        inner.records.get(request_uri).cloned()
    }

    fn pending_for(&self, account: &str, cursor: Option<&str>) -> (Vec<DecisionRecord>, String) {
        let after = Cursor::read(cursor, &self.epoch);
        let now = OffsetDateTime::now_utc();
        let inner = self.inner.read().unwrap_or_else(|p| p.into_inner());
        let mut out: Vec<DecisionRecord> = inner
            .records
            .values()
            .filter(|record| {
                record.account == account && record.sequence > after && record.is_live(now)
            })
            .cloned()
            .collect();
        out.sort_by_key(|record| record.sequence);
        (out, Cursor::wire(&self.epoch, inner.sequence.max(after)))
    }

    fn settle(&self, request_uri: &str, state: DecisionState) -> Option<DecisionRecord> {
        self.inner
            .write()
            .unwrap_or_else(|p| p.into_inner())
            .settle(request_uri, state, None, OffsetDateTime::now_utc())
    }

    fn approve(&self, request_uri: &str, redirect: Option<String>) -> Option<DecisionRecord> {
        self.inner
            .write()
            .unwrap_or_else(|p| p.into_inner())
            .settle(
                request_uri,
                DecisionState::Approved,
                redirect,
                OffsetDateTime::now_utc(),
            )
    }

    fn holds(&self, request_uri: &str) -> bool {
        self.inner
            .read()
            .unwrap_or_else(|p| p.into_inner())
            .redirects
            .contains_key(request_uri)
    }

    fn collect(&self, request_uri: &str) -> Option<String> {
        self.inner
            .write()
            .unwrap_or_else(|p| p.into_inner())
            .redirects
            .remove(request_uri)
    }

    fn sweep(&self, now: OffsetDateTime) {
        self.inner
            .write()
            .unwrap_or_else(|p| p.into_inner())
            .sweep(now);
    }

    fn arrivals(&self) -> Arc<Notify> {
        self.arrivals.clone()
    }

    fn epoch(&self) -> String {
        self.epoch.clone()
    }

    fn remember(
        &self,
        account: &str,
        client_key: &ClientKey,
        client_id: &str,
        granted: ScopeSet,
        now: OffsetDateTime,
    ) {
        let mut inner = self.inner.write().unwrap_or_else(|p| p.into_inner());
        inner.sweep(now);
        let key = (account.to_owned(), client_key.0.clone());
        // Cleared rather than trimmed, as `seen` is: forgetting asks again,
        // which is the failing-safe direction.
        if inner.remembered.len() >= SEEN_CLIENTS_MAX && !inner.remembered.contains_key(&key) {
            inner.remembered.clear();
        }
        inner.remembered.insert(
            key,
            Remembered {
                client_id: client_id.to_owned(),
                granted,
                expires_at: now + REMEMBERED_TTL,
            },
        );
    }

    fn remembered(
        &self,
        account: &str,
        client_key: &ClientKey,
        now: OffsetDateTime,
    ) -> Option<Remembered> {
        let inner = self.inner.read().unwrap_or_else(|p| p.into_inner());
        inner
            .remembered
            .get(&(account.to_owned(), client_key.0.clone()))
            .filter(|remembered| now < remembered.expires_at)
            .cloned()
    }

    fn forget_client(&self, client_id: &str, account: Option<&str>) -> usize {
        let mut inner = self.inner.write().unwrap_or_else(|p| p.into_inner());
        let before = inner.remembered.len();
        inner.remembered.retain(|(did, _), remembered| {
            remembered.client_id != client_id || account.is_some_and(|account| account != did)
        });
        before - inner.remembered.len()
    }

    fn forget_account(&self, account: &str) {
        self.inner
            .write()
            .unwrap_or_else(|p| p.into_inner())
            .remembered
            .retain(|(did, _), _| did != account);
    }

    fn forget_all(&self) {
        self.inner
            .write()
            .unwrap_or_else(|p| p.into_inner())
            .remembered
            .clear();
    }
}

/// Whether this atom is one this server never grants, whatever a ceiling
/// says: `plan/scope-policy.md`'s hard-blocked capabilities, and any atom the
/// grammar cannot read. [`verdict_for`] drops each from the request whole, so
/// the rest can be granted and the agent sees it in `cut`.
///
/// `account` and `identity` are on that list because key rotation and
/// migrating out are boundaries an operator draws deliberately rather than
/// ones an app asks its way past. `blob` is not: an upload is checked
/// against the grant at `uploadBlob` and judged by policy twice, so a
/// `blob:` scope says what an app may upload rather than handing it the
/// account. `transition:*` is on it for
/// the opposite reason — it is not narrow but enormous, carrying records,
/// blobs and RPC in one string, which is what makes every granular rule
/// beside it decorative.
///
/// Deliberately here and not in [`crate::oauth::scope`]. That module is the
/// grammar — what a scope string means, and how two of them overlap — and
/// this is a policy about which of them this deployment will consider at
/// all. A grammar that knew the list could not be read by anything that
/// disagreed with it.
fn hard_blocked(atom: &Scope) -> bool {
    matches!(
        atom,
        Scope::Transition(_) | Scope::Identity { .. } | Scope::Account { .. } | Scope::Unknown(_)
    )
}

/// The verdict a ceiling alone gives for one request.
///
/// `granted` empty is a refusal, an empty `cut` is an unqualified yes, and
/// anything else is a narrowing that names what it took. A policy refusal is
/// applied by the caller on top of this, because a policy can only ever make
/// the answer worse.
pub fn verdict_for(requested: &ScopeSet, ceiling: &ScopeSet, rule: &str) -> Verdict {
    // Before the ceiling is consulted at all, and whole: a hard-blocked atom
    // is cut as asked for, never projected onto the ceiling in pieces.
    let (dropped, asked): (Vec<Scope>, Vec<Scope>) =
        requested.0.iter().cloned().partition(hard_blocked);
    let (granted, cut) = ScopeSet::new(asked).narrow(ceiling);
    let cut = ScopeSet::new(cut.0.into_iter().chain(dropped).collect());
    if granted.0.is_empty() {
        return Verdict::Deny {
            reason: format!("this account's ceiling grants none of `{requested}`"),
            rule: rule.to_owned(),
        };
    }
    if cut.0.is_empty() {
        return Verdict::Allow;
    }
    Verdict::Narrow {
        granted,
        cut,
        rule: rule.to_owned(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    fn record(account: &str, token: Option<&str>) -> DecisionRecord {
        let now = OffsetDateTime::now_utc();
        DecisionRecord {
            request_uri: format!("urn:{}", rand_suffix()),
            account: account.to_owned(),
            client_origin: "https://client.example".to_owned(),
            client_key: ClientKey("abcd".to_owned()),
            first_time: false,
            client_self: ClientSelf::default(),
            requested: ScopeSet::parse("atproto").unwrap(),
            verdict: Verdict::Allow,
            token: token.map(|t| ConsentReference(t.to_owned())),
            created_at: now,
            expires_at: now + time::Duration::minutes(2),
            state: DecisionState::Pending,
            sequence: 0,
        }
    }

    fn rand_suffix() -> String {
        use std::sync::atomic::{AtomicU64, Ordering};
        static NEXT: AtomicU64 = AtomicU64::new(0);
        NEXT.fetch_add(1, Ordering::Relaxed).to_string()
    }

    /// The bound `plan/oauth.md` needs because PAR is unauthenticated: one
    /// account cannot leave more than its share of undecided records behind.
    /// Past it the account's oldest makes way, so the bound holds and the
    /// newest push is the one that survives.
    #[test]
    fn an_account_at_its_bound_displaces_its_oldest_rather_than_refusing() {
        let store = MemoryDecisionStore::with_bounds(2, 100);
        let first = store
            .put(record("did:web:a", Some("tok-1")))
            .unwrap()
            .record;
        let second = store.put(record("did:web:a", None)).unwrap().record;
        let third = store.put(record("did:web:a", Some("tok-3"))).unwrap();

        let displaced = third.displaced.expect("the oldest made way");
        assert_eq!(displaced.request_uri, first.request_uri);
        assert_eq!(displaced.token.map(|t| t.0), Some("tok-1".to_owned()));
        assert!(store.get(&first.request_uri).is_none());
        assert!(store.by_token("tok-1").is_none());
        assert!(store.get(&second.request_uri).is_some());
        assert!(store.by_token("tok-3").is_some());

        let (pending, _) = store.pending_for("did:web:a", None);
        assert_eq!(pending.len(), 2, "the bound still holds: {pending:?}");

        // Another account is unaffected: the bound is per account, so a
        // spree against one cannot crowd out the rest.
        let other = store.put(record("did:web:b", None)).unwrap();
        assert!(other.displaced.is_none());
    }

    /// **The lockout this closes.** A stranger fills an account's slots by
    /// naming it in `login_hint`. The account's own sign-in, pushed last,
    /// must still be the one an agent can answer.
    #[test]
    fn the_newest_push_survives_a_stranger_filling_the_account() {
        let store = MemoryDecisionStore::with_bounds(2, 100);
        for _ in 0..8 {
            store.put(record("did:web:a", None)).unwrap();
        }
        let real = store.put(record("did:web:a", Some("mine"))).unwrap().record;
        let (pending, _) = store.pending_for("did:web:a", None);
        assert!(
            pending.iter().any(|r| r.request_uri == real.request_uri),
            "the account's own sign-in was crowded out: {pending:?}"
        );
        assert!(store.by_token("mine").is_some());
    }

    /// **The lockout the other bound reopened.** The server-wide bound was
    /// asked before an account's own slot was taken back, so once this
    /// store was full every account was refused — including the one whose
    /// eight slots a stranger had just filled, which is exactly the victim
    /// displacement exists for. A displacing push does not grow the store,
    /// so it is not measured against the store's total.
    #[test]
    fn a_full_store_still_answers_the_account_whose_own_slots_are_full() {
        let store = MemoryDecisionStore::with_bounds(2, 4);
        // Two accounts, two records each: this store is full.
        for account in ["did:web:a", "did:web:b"] {
            for _ in 0..2 {
                store.put(record(account, None)).unwrap();
            }
        }
        assert_eq!(
            store.put(record("did:web:c", None)).unwrap_err(),
            DecisionRefused::TooMany(4),
            "a third account asking for a slot this store does not have"
        );

        let mine = store
            .put(record("did:web:a", Some("mine")))
            .expect("an account at its own bound displaces, full store or not");
        assert!(mine.displaced.is_some());
        assert!(store.by_token("mine").is_some());
        let (pending, _) = store.pending_for("did:web:a", None);
        assert_eq!(pending.len(), 2, "the account's own bound still holds");
        assert_eq!(
            store.pending_for("did:web:b", None).0.len(),
            2,
            "and the other account's records were left alone"
        );
    }

    #[test]
    fn the_server_wide_bound_refuses_regardless_of_account() {
        let store = MemoryDecisionStore::with_bounds(100, 2);
        assert!(store.put(record("did:web:a", None)).is_ok());
        assert!(store.put(record("did:web:b", None)).is_ok());
        assert_eq!(
            store.put(record("did:web:c", None)).unwrap_err(),
            DecisionRefused::TooMany(2)
        );
    }

    /// `check_room` reads the same bound `put` enforces, without writing —
    /// `oauth::par::push_for_unresolved_hint` needs exactly this: an answer
    /// about server load that costs nothing, since there is no account for
    /// a write to be attributed to.
    #[test]
    fn check_room_reads_the_total_bound_without_writing() {
        let store = MemoryDecisionStore::with_bounds(100, 1);
        assert!(store.check_room().is_ok());
        assert!(
            store.check_room().is_ok(),
            "checking room does not use it up"
        );
        store.put(record("did:web:a", None)).unwrap();
        assert_eq!(store.check_room().unwrap_err(), DecisionRefused::TooMany(1));
    }

    /// A record leaves `Pending` once. The second answer finds nothing,
    /// which is what stops an approval and a decline from both landing.
    #[test]
    fn a_record_settles_once() {
        let store = MemoryDecisionStore::new();
        let stored = store.put(record("did:web:a", Some("tok"))).unwrap().record;
        assert!(store
            .settle(&stored.request_uri, DecisionState::Approved)
            .is_some());
        assert!(store
            .settle(&stored.request_uri, DecisionState::Declined)
            .is_none());
        assert_eq!(
            store.get(&stored.request_uri).unwrap().state,
            DecisionState::Approved
        );
        // The token stops resolving the moment the record settles.
        assert!(store.by_token("tok").is_none());
    }

    /// An approval holds its redirect for a browser, which takes it once.
    #[test]
    fn an_approval_holds_its_redirect_until_it_is_collected_once() {
        let store = MemoryDecisionStore::new();
        let stored = store.put(record("did:web:a", Some("tok"))).unwrap().record;
        assert!(!store.holds(&stored.request_uri));
        store
            .approve(
                &stored.request_uri,
                Some("https://client.example/cb?code=c".into()),
            )
            .expect("a live record approves");
        assert!(
            store
                .approve(
                    &stored.request_uri,
                    Some("https://elsewhere.example/".into())
                )
                .is_none(),
            "an approval settles once"
        );
        assert!(store.holds(&stored.request_uri));
        assert_eq!(
            store.collect(&stored.request_uri).as_deref(),
            Some("https://client.example/cb?code=c")
        );
        assert!(store.collect(&stored.request_uri).is_none());
        assert!(!store.holds(&stored.request_uri));
    }

    /// An answered record stays until its pushed request expires, so the
    /// authorize page can still say how the request was answered. Then it
    /// goes, with the redirect it held.
    #[test]
    fn an_answered_record_stays_until_its_request_expires() {
        let store = MemoryDecisionStore::new();
        let approved = store.put(record("did:web:a", Some("a"))).unwrap().record;
        let declined = store.put(record("did:web:a", Some("d"))).unwrap().record;
        store.approve(
            &approved.request_uri,
            Some("https://client.example/cb".into()),
        );
        store.settle(&declined.request_uri, DecisionState::Declined);

        store.sweep(OffsetDateTime::now_utc());
        assert_eq!(
            store.get(&approved.request_uri).unwrap().state,
            DecisionState::Approved
        );
        assert_eq!(
            store.get(&declined.request_uri).unwrap().state,
            DecisionState::Declined
        );

        store.sweep(declined.expires_at);
        assert!(store.get(&approved.request_uri).is_none());
        assert!(store.get(&declined.request_uri).is_none());
        assert!(store.collect(&approved.request_uri).is_none());
    }

    /// A stranger filling an account's bound displaces its undecided
    /// records, never one the account answered, so the page still reports
    /// that answer and still holds the code it waits to hand over.
    #[test]
    fn a_push_past_the_bound_never_displaces_an_answer() {
        let store = MemoryDecisionStore::with_bounds(2, 100);
        let approved = store.put(record("did:web:a", Some("mine"))).unwrap().record;
        store.approve(
            &approved.request_uri,
            Some("https://client.example/cb".into()),
        );
        for _ in 0..4 {
            let stranger = store.put(record("did:web:a", None)).unwrap();
            if let Some(displaced) = stranger.displaced {
                assert_ne!(displaced.request_uri, approved.request_uri);
            }
        }
        let (pending, _) = store.pending_for("did:web:a", None);
        assert_eq!(pending.len(), 2, "the account's own bound still holds");
        assert_eq!(
            store.get(&approved.request_uri).unwrap().state,
            DecisionState::Approved
        );
        assert_eq!(
            store.collect(&approved.request_uri).as_deref(),
            Some("https://client.example/cb")
        );
    }

    /// The server-wide bound counts what the store holds, answered records
    /// included, so keeping them does not let the store outgrow it.
    #[test]
    fn the_server_wide_bound_counts_answered_records() {
        let store = MemoryDecisionStore::with_bounds(100, 1);
        let answered = store.put(record("did:web:a", Some("tok"))).unwrap().record;
        store.settle(&answered.request_uri, DecisionState::Approved);
        assert_eq!(
            store.put(record("did:web:b", None)).unwrap_err(),
            DecisionRefused::TooMany(1)
        );
        assert_eq!(store.check_room().unwrap_err(), DecisionRefused::TooMany(1));
    }

    #[test]
    fn a_settled_record_is_not_listed_and_the_cursor_still_advances() {
        let store = MemoryDecisionStore::new();
        let first = store.put(record("did:web:a", None)).unwrap().record;
        let (pending, cursor) = store.pending_for("did:web:a", None);
        assert_eq!(pending.len(), 1);
        store.settle(&first.request_uri, DecisionState::Declined);
        let (pending, next) = store.pending_for("did:web:a", None);
        assert!(pending.is_empty());
        assert_eq!(next, cursor);
    }

    /// A cursor is per store, not per account: a record minted for somebody
    /// else still moves it, so a caller polling with the cursor it was last
    /// given never re-reads what it has already seen.
    #[test]
    fn a_cursor_only_returns_what_arrived_after_it() {
        let store = MemoryDecisionStore::new();
        store.put(record("did:web:a", None)).unwrap();
        let (_, cursor) = store.pending_for("did:web:a", None);
        let second = store.put(record("did:web:a", None)).unwrap().record;
        let (pending, _) = store.pending_for("did:web:a", Some(&cursor));
        assert_eq!(pending.len(), 1);
        assert_eq!(pending[0].request_uri, second.request_uri);
    }

    #[test]
    fn one_account_never_sees_another_s_record() {
        let store = MemoryDecisionStore::new();
        store.put(record("did:web:a", None)).unwrap();
        let (pending, _) = store.pending_for("did:web:b", None);
        assert!(pending.is_empty());
    }

    #[test]
    fn an_expired_record_is_swept_and_never_listed() {
        let store = MemoryDecisionStore::new();
        let mut stale = record("did:web:a", Some("tok"));
        stale.expires_at = OffsetDateTime::now_utc() - time::Duration::seconds(1);
        let stored = store.put(stale).unwrap().record;
        let (pending, _) = store.pending_for("did:web:a", None);
        assert!(pending.is_empty());
        store.sweep(OffsetDateTime::now_utc());
        assert!(store.get(&stored.request_uri).is_none());
        assert!(store.by_token("tok").is_none());
    }

    /// A client is new to an account until that account approves it. A
    /// push, which anyone may make naming any account, does not make it
    /// familiar, nor does a decline, nor another account's approval.
    #[test]
    fn first_time_lasts_until_the_account_approves() {
        let store = MemoryDecisionStore::new();
        let push = |account: &str| store.put(record(account, None)).unwrap().record;

        let declined = push("did:web:a");
        assert!(declined.first_time);
        assert!(push("did:web:a").first_time, "a second push");
        store.settle(&declined.request_uri, DecisionState::Declined);
        assert!(push("did:web:a").first_time, "after a decline");

        let approved = push("did:web:a");
        assert!(approved.first_time);
        store.settle(&approved.request_uri, DecisionState::Approved);
        assert!(!push("did:web:a").first_time, "after an approval");
        assert!(push("did:web:b").first_time, "another account");
    }

    /// A new approval replaces the grant remembered before it rather than
    /// adding to it, and each is forgotten [`REMEMBERED_TTL`] after the
    /// approval that wrote it.
    #[test]
    fn a_remembered_approval_is_the_latest_grant_until_it_expires() {
        let store = MemoryDecisionStore::new();
        let key = ClientKey("abcd".to_owned());
        let now = OffsetDateTime::now_utc();
        let wide = ScopeSet::parse("atproto repo:*").unwrap();
        let narrow = ScopeSet::parse("atproto").unwrap();

        store.remember("did:web:a", &key, "https://c.example", wide, now);
        store.remember("did:web:a", &key, "https://c.example", narrow.clone(), now);
        let remembered = store.remembered("did:web:a", &key, now).unwrap();
        assert_eq!(remembered.granted, narrow);
        assert!(store.remembered("did:web:b", &key, now).is_none());

        let later = now + REMEMBERED_TTL;
        assert!(store
            .remembered("did:web:a", &key, later - time::Duration::seconds(1))
            .is_some());
        assert!(store.remembered("did:web:a", &key, later).is_none());
    }

    /// An operator ending a client's logins on one account forgets that
    /// account's approval of it, and no other account's or client's.
    #[test]
    fn forgetting_a_client_on_one_account_spares_the_rest() {
        let store = MemoryDecisionStore::new();
        let (one, other) = (ClientKey("one".to_owned()), ClientKey("other".to_owned()));
        let now = OffsetDateTime::now_utc();
        let granted = ScopeSet::parse("atproto").unwrap();
        for account in ["did:web:a", "did:web:b"] {
            store.remember(account, &one, "https://one.example", granted.clone(), now);
            store.remember(
                account,
                &other,
                "https://other.example",
                granted.clone(),
                now,
            );
        }

        assert_eq!(
            store.forget_client("https://one.example", Some("did:web:a")),
            1
        );
        assert!(store.remembered("did:web:a", &one, now).is_none());
        assert!(store.remembered("did:web:a", &other, now).is_some());
        assert!(store.remembered("did:web:b", &one, now).is_some());

        assert_eq!(store.forget_client("https://one.example", None), 1);
        assert!(store.remembered("did:web:b", &one, now).is_none());
    }

    /// **The e2e defect.** A daemon that outlives a restart holds a cursor
    /// from a store that no longer exists. The new store's counter begins at
    /// zero, so `sequence > after` matches nothing and the daemon is handed
    /// an empty list forever — silently, since an empty answer is also what
    /// "nothing new" looks like. An epoch in the cursor makes the stale one
    /// legible: it names an order this store never had, so it is read as the
    /// beginning of the current one.
    #[test]
    fn a_cursor_from_before_a_restart_lists_from_the_beginning() {
        let before = MemoryDecisionStore::new();
        before.put(record("did:web:a", None)).unwrap();
        before.put(record("did:web:a", None)).unwrap();
        before.put(record("did:web:a", None)).unwrap();
        let (seen, stale) = before.pending_for("did:web:a", None);
        assert_eq!(seen.len(), 3);

        // The restart. A fresh store, a fresh epoch, counting from one.
        let after = MemoryDecisionStore::new();
        assert_ne!(before.epoch(), after.epoch());
        after.put(record("did:web:a", None)).unwrap();

        let (pending, fresh) = after.pending_for("did:web:a", Some(&stale));
        assert_eq!(
            pending.len(),
            1,
            "a cursor from the old store stranded the caller: {pending:?}"
        );
        assert!(fresh.starts_with(&after.epoch()));
        // And the cursor it was just handed works normally from here.
        let (pending, _) = after.pending_for("did:web:a", Some(&fresh));
        assert!(pending.is_empty());
    }

    /// A cursor is opaque, so anything a caller can send that is not one of
    /// this store's has to be read as the beginning rather than refused or
    /// believed. Refusing would strand a client on a string it cannot fix;
    /// believing a bare number is what the epoch exists to stop.
    #[test]
    fn a_cursor_this_store_did_not_mint_is_read_as_the_beginning() {
        let store = MemoryDecisionStore::new();
        store.put(record("did:web:a", None)).unwrap();
        for raw in [
            "",
            "3",
            "not-a-cursor",
            "deadbeefdeadbeef.3",
            ".",
            "..",
            &format!("{}.notanumber", store.epoch()),
        ] {
            let (pending, _) = store.pending_for("did:web:a", Some(raw));
            assert_eq!(pending.len(), 1, "`{raw}` was not read as the beginning");
        }
    }

    #[test]
    fn a_cursor_reads_back_the_sequence_it_carries() {
        assert_eq!(Cursor::read(Some("abc.7"), "abc"), 7);
        assert_eq!(Cursor::read(Some("abc.7"), "xyz"), 0);
        assert_eq!(Cursor::read(None, "abc"), 0);
        assert_eq!(Cursor::wire("abc", 7), "abc.7");
    }

    #[test]
    fn a_ceiling_that_grants_everything_asked_for_allows() {
        let requested = ScopeSet::parse("atproto repo:app.bsky.feed.post").unwrap();
        let ceiling = ScopeSet::parse("atproto repo:*").unwrap();
        assert_eq!(
            verdict_for(&requested, &ceiling, CEILING_RULE),
            Verdict::Allow
        );
    }

    #[test]
    fn a_ceiling_that_grants_nothing_denies() {
        let requested = ScopeSet::parse("repo:app.bsky.feed.post").unwrap();
        let verdict = verdict_for(&requested, &ScopeSet::default(), CEILING_RULE);
        assert!(verdict.is_deny());
        assert!(verdict.granted(&requested).0.is_empty());
    }

    /// The verdict the whole design exists for: part of the request is
    /// granted, and the record says which part was not.
    #[test]
    fn a_ceiling_that_grants_part_narrows_and_names_the_cut() {
        let requested = ScopeSet::parse("atproto repo:app.bsky.feed.post").unwrap();
        let ceiling = ScopeSet::parse("atproto").unwrap();
        let verdict = verdict_for(&requested, &ceiling, CEILING_RULE);
        let Verdict::Narrow { granted, cut, rule } = &verdict else {
            panic!("expected a narrowing, got {verdict:?}");
        };
        assert_eq!(granted.to_string(), "atproto");
        assert_eq!(cut.to_string(), "repo:app.bsky.feed.post");
        assert_eq!(rule, CEILING_RULE);
        assert_eq!(verdict.granted(&requested).to_string(), "atproto");
    }

    /// **The e2e defect.** A hard-blocked atom is cut whole, and nothing is
    /// paid out of it. `transition:generic` overlaps the compiled-in
    /// ceiling's `repo:*` and `rpc:*`, and projecting it onto them would hand
    /// back the entire ceiling. The rest of the request is granted.
    #[test]
    fn a_request_naming_a_hard_blocked_atom_is_granted_the_rest() {
        let requested = ScopeSet::parse("atproto transition:generic").unwrap();
        let ceiling = ScopeSet::parse("atproto repo:* rpc:*").unwrap();
        let verdict = verdict_for(&requested, &ceiling, CEILING_RULE);
        let Verdict::Narrow { granted, cut, rule } = &verdict else {
            panic!("expected a narrowing, got {verdict:?}");
        };
        assert_eq!(granted.to_string(), "atproto");
        assert_eq!(cut.to_string(), "transition:generic");
        assert_eq!(rule, CEILING_RULE);
    }

    /// Every atom this server never grants, in the spellings the permission
    /// spec gives it and some it does not, against a ceiling wide enough
    /// that only the drop can be cutting it. `blob:` is not on the list: an
    /// upload is checked against the grant and judged by policy.
    #[test]
    fn every_hard_blocked_atom_is_cut_however_it_is_spelled() {
        let ceiling = ScopeSet::parse("atproto repo:* rpc:*").unwrap();
        for (atom, named) in [
            ("transition:generic", "transition:generic"),
            ("transition:chat.bsky", "transition:chat.bsky"),
            ("transition:email", "transition:email"),
            ("identity:*", "identity:*"),
            ("identity:*?", "identity:*"),
            ("identity?attr=*", "identity:*"),
            ("identity:handle", "identity:handle"),
            ("identity?attr=handle", "identity:handle"),
            ("identity:handle?action=manage", "identity:handle"),
            ("account:email", "account:email"),
            ("account:email?action=read", "account:email"),
            ("account:repo?action=manage", "account:repo?action=manage"),
            ("account:status", "account:status"),
            (
                "space:*?authority=*&action=read",
                "space:*?authority=*&action=read",
            ),
        ] {
            let requested =
                ScopeSet::parse(&format!("atproto repo:app.bsky.feed.post {atom}")).unwrap();
            let verdict = verdict_for(&requested, &ceiling, CEILING_RULE);
            let Verdict::Narrow { granted, cut, rule } = &verdict else {
                panic!("`{atom}` was not cut: {verdict:?}");
            };
            assert_eq!(
                granted.to_string(),
                "atproto repo:app.bsky.feed.post",
                "`{atom}`"
            );
            assert_eq!(cut.to_string(), named, "`{atom}`");
            assert_eq!(rule, CEILING_RULE);
        }
    }

    /// The drop does not depend on the ceiling leaving these atoms out: one
    /// that holds them grants none of them either.
    #[test]
    fn a_ceiling_that_holds_a_hard_blocked_atom_still_grants_none() {
        let held = ScopeSet::parse(
            "atproto transition:generic transition:email identity:* account:email space:*",
        )
        .unwrap();
        let requested = ScopeSet::parse(
            "atproto transition:generic transition:email identity:handle account:email space:*",
        )
        .unwrap();
        let verdict = verdict_for(&requested, &held, CEILING_RULE);
        let Verdict::Narrow { granted, cut, .. } = &verdict else {
            panic!("expected a narrowing, got {verdict:?}");
        };
        assert_eq!(granted.to_string(), "atproto");
        assert_eq!(
            cut.to_string(),
            "transition:generic transition:email identity:handle account:email space:*"
        );
    }

    /// A request left with nothing grantable once those atoms are dropped is
    /// denied as a narrowing that grants nothing is.
    #[test]
    fn a_request_of_only_hard_blocked_atoms_is_denied_by_the_ceiling() {
        let requested = ScopeSet::parse("transition:email account:email").unwrap();
        let ceiling = ScopeSet::parse("atproto repo:* rpc:*").unwrap();
        assert_eq!(
            verdict_for(&requested, &ceiling, CEILING_RULE),
            Verdict::Deny {
                reason: "this account's ceiling grants none of \
                         `transition:email account:email`"
                    .to_owned(),
                rule: CEILING_RULE.to_owned(),
            }
        );
    }

    /// The granular scopes an app asks for beside `transition:generic` are
    /// granted on their own once it is dropped.
    #[test]
    fn scopes_asked_for_beside_a_transition_scope_survive_its_drop() {
        let requested =
            ScopeSet::parse("atproto transition:generic repo:app.bsky.feed.post blob:image/png")
                .unwrap();
        let ceiling = ScopeSet::parse("atproto repo:* rpc:* blob:*/*").unwrap();
        let verdict = verdict_for(&requested, &ceiling, CEILING_RULE);
        assert_eq!(
            verdict.granted(&requested).to_string(),
            "atproto repo:app.bsky.feed.post blob:image/png"
        );
    }

    /// The narrowing case the e2e run got right, kept as a regression: an
    /// ordinary request wider than its ceiling is met in part, and the part
    /// it did not get is named.
    #[test]
    fn a_request_wider_than_its_ceiling_narrows_to_the_ceiling() {
        let requested = ScopeSet::parse("atproto repo:* rpc:*").unwrap();
        let ceiling = ScopeSet::parse("atproto repo:*").unwrap();
        let verdict = verdict_for(&requested, &ceiling, CEILING_RULE);
        let Verdict::Narrow { granted, cut, .. } = &verdict else {
            panic!("expected a narrowing, got {verdict:?}");
        };
        assert_eq!(granted.to_string(), "atproto repo:*");
        assert_eq!(cut.to_string(), "rpc:*");
    }

    /// A request *narrower* than its ceiling is allowed at the form it
    /// asked for, never widened to the ceiling's. The direction
    /// `ScopeSet::narrow`'s projection rule exists to fix: a grant is a
    /// narrowing of the request, so a `repo:*` ceiling cannot answer a
    /// request for one collection with every collection.
    #[test]
    fn a_request_narrower_than_its_ceiling_is_granted_at_the_form_it_asked_for() {
        let requested = ScopeSet::parse("repo:com.example.thing?action=create").unwrap();
        let ceiling = ScopeSet::parse("repo:*").unwrap();
        let verdict = verdict_for(&requested, &ceiling, CEILING_RULE);
        assert_eq!(verdict, Verdict::Allow);
        assert_eq!(
            verdict.granted(&requested).to_string(),
            "repo:com.example.thing?action=create"
        );
    }

    /// `plan/app-allowlist.md`'s "record the client's own copy; do not show
    /// it", asserted on the one function that decides what leaves this
    /// server.
    #[test]
    fn the_wire_shape_carries_no_client_self_description() {
        let mut with_copy = record("did:web:a", Some("tok"));
        with_copy.client_self = ClientSelf {
            name: Some("Krillbridge".to_owned()),
            uri: Some("https://impersonator.example".to_owned()),
            logo: Some("https://impersonator.example/logo.png".to_owned()),
        };
        let text = with_copy.to_json().to_string();
        assert!(!text.contains("Krillbridge"));
        assert!(!text.contains("impersonator.example"));
        assert!(text.contains("https://client.example"));
        assert!(text.contains("\"firstTime\""));
    }
}