gunnar-core 3.0.0

//! The identity table and the ACL (the ACL model).
//!
//! **gunnar owns its own table.** No OS users, no `~/.ssh/authorized_keys`, no
//! `/etc/passwd`, no home directory. The server has to run as PID 1 in a
//! distroless image where none of those exist (the no-OS server rule). Identity comes from
//! the SSH key and nothing else; the SSH *username* is decorative, so
//! `git@host:ns/repo.git` works and `git` means nothing.
//!
//! The auth flow the SSH listener runs is three calls:
//!
//! ```text
//! russh auth_publickey  →  Fingerprint::of_key(offered)
//!                       →  AuthTable::principal_by_fingerprint(&fp)   // who
//!                       →  RepoCatalog::open(&sandboxed_path)         // what
//!                       →  AuthTable::may(&principal.id, &repo, level) // may they
//! ```
//!
//! `upload-pack` needs [`Level::Read`], `receive-pack` needs [`Level::Write`].
//!
//! # The one caller that never authenticates
//!
//! A **public** repository is readable with no key at all. Such a caller is
//! served as [`Principal::anonymous`], which is not in this table, cannot be
//! put in it, and holds neither key nor grant. The question *may it?* still
//! goes to [`AuthTable::may`], exactly as it does for everybody else. Rule 0
//! there is the whole of what anonymous can ever be allowed: `Read`, on a
//! `Public` repository, and nothing anywhere else. See
//! [`ANONYMOUS_PRINCIPAL_ID`](crate::model::ANONYMOUS_PRINCIPAL_ID).

use std::sync::Arc;

use crate::control::ControlStore;
use crate::doc::{impl_doc, DocRef, ShardedMap};
use crate::error::{Error, Result};
use crate::key_attestation::AttestingKey;
use crate::model::{
    now_ms, AccessToken, Grant, Level, Principal, PrincipalKind, PublicKey, Repo, Scope, UnixMillis,
};

mod sso_table;
pub use sso_table::{
    SsoImportOp, SsoImportRow, SsoLogin, SsoLoginDone, SsoProvision, SsoUserPatch,
};

/// **The name the identity table is stored under** in the control store.
///
/// It still reads `refs/gunnar/auth`. Until 2026-08-09 that WAS a git reference
/// (the certificate-identity plan); it is now a key in redb, and the string was kept because it is
/// what every error message, log line and runbook calls this document.
pub const AUTH_REF: &str = "refs/gunnar/auth";

/// Id of the principal [`AuthTable::bootstrap_admin`] installs when no name is
/// given (`--admin-key` on the command line).
pub const BOOTSTRAP_ADMIN_ID: &str = "admin";

/// **Clans** (owner, 2026-10-05): ordinary accounts grouped, each giving the
/// clan part of its own room; the clan's admin administers what is made in it.
mod clan_table;
pub use clan_table::{
    AccountRoom, Clan, ClanInvite, ClanMember, ClanRefusal, ClanRepoOffer, ClanRole,
};

/// Every fingerprint gunnar computes or accepts carries this prefix.
pub const SHA256_PREFIX: &str = "SHA256:";

/// The `algo` of a registered **quantum-safe encryption key** (crypt-pq): an
/// `age1pq1…` recipient (age's hybrid ML-KEM-768 + X25519), registered beside
/// a person's SSH keys so `Repos.RecipientKeys` can hand it to a member
/// wrapping a quantum-safe repository's key ring. It is never a login: no SSH
/// client can present a key with its fingerprint, and it cannot sign, so it is
/// never an attesting key either.
pub const PQ_RECIPIENT_ALGO: &str = "mlkem768x25519";

/// The fingerprint of an `age1pq1…` line — SHA-256 over its 1216 key bytes,
/// in the one fingerprint space — or `None` when the line is not one.
pub fn pq_recipient_fingerprint(line: &str) -> Option<Fingerprint> {
    gunnar_crypt::members::pq_public_key_bytes(line.trim()).map(|bytes| fingerprint_of_blob(&bytes))
}

/// **The one fingerprint: SHA-256 over the SSH wire-format public key
/// blob, rendered `SHA256:<base64>` exactly as `ssh-keygen -lf` prints it.**
///
/// `ssh_wire_blob` is the `string` an SSH public key is encoded as on the wire:
/// `ssh-ed25519` followed by its length-prefixed key material. Same bytes as
/// the base64 payload in the middle of an `authorized_keys` line. Both
/// `ssh-key` versions in gunnar's graph produce it with `PublicKey::to_bytes()`.
///
/// # Why bytes and not a type
///
/// `gunnar-ssh` links `ssh-key 0.7.0-rc.10`, because `russh 0.61.2` pins that
/// version exactly and `Handler::auth_publickey` hands over *its*
/// `PublicKey`. This crate links the workspace's `ssh-key 0.6.7`. They are
/// semver-incompatible and both are in the graph, so a key that arrives through
/// russh can never be passed to a 0.6 API. The workspace is deliberately
/// **not** bumped to a release candidate to paper over that.
///
/// A blob crosses the line. Define the fingerprint over the blob and one
/// fingerprint serves both crates: `gunnar-ssh` computes the same string the
/// [`AuthTable`] stores, and `tests/fingerprint.rs` pins it against literal
/// `ssh-keygen -lf` output rather than against gunnar's own arithmetic.
///
/// Any bytes at all are accepted: this is a hash, not a parser, and refusing to
/// hash a malformed blob would only turn "this key is not registered" into an
/// error path with a second answer in it.
pub fn fingerprint_of_blob(ssh_wire_blob: &[u8]) -> Fingerprint {
    use base64ct::Encoding as _;
    use sha2::Digest as _;

    let digest = sha2::Sha256::digest(ssh_wire_blob);
    Fingerprint(format!(
        "{SHA256_PREFIX}{}",
        base64ct::Base64Unpadded::encode_string(&digest)
    ))
}

/// The stored form of an access token secret: SHA-256, lowercase hex.
///
/// Hex rather than the `SHA256:<base64>` spelling [`Fingerprint`] uses, and the
/// difference is on purpose: a fingerprint is meant to be compared by eye
/// against what `ssh-keygen -lf` prints on somebody's laptop, and a token hash
/// is never compared against anything outside gunnar. Sharing the spelling
/// would invite the two to be confused in a log, where one identifies a key and
/// the other is all that is left of a password.
///
/// **Length is not a secret and neither is this value.** It is what
/// `token list` prints. What must never be printed is the input.
///
/// Public because the HTTP front end needs the same value to record a use and
/// to name the token in a log line, and computing it there would be a second
/// definition of the stored form of a credential (one implementation).
pub fn token_hash(secret: &str) -> String {
    use sha2::Digest as _;

    hex::encode(sha2::Sha256::digest(secret.as_bytes()))
}

/// How fresh a token's recorded `last_used` may be before another
/// authentication skips rewriting it — see [`AuthTable::touch_token`], where
/// the measurement that sets this is written down. One minute: coarse enough
/// that the shared-document CAS leaves the request path, fine enough that
/// `token list` still answers "is this credential alive" honestly.
pub const LAST_USED_GRANULARITY: UnixMillis = 60_000;

/// One user of a bulk import — see [`AuthTable::bulk_add_users`].
#[derive(Clone, Debug)]
pub struct BulkUser {
    /// The principal to create.
    pub principal: Principal,
    /// [`token_hash`] of an access-token secret, when the user gets one. The
    /// secret itself stays with the caller, as it does for
    /// [`AuthTable::register_token`].
    pub token_hash: Option<String>,
    /// The label stored beside the token; ignored without `token_hash`.
    pub token_label: String,
    /// An OpenSSH public key line to register for the user, when they get one.
    /// Parsed and fingerprinted here exactly as [`AuthTable::register_key`]
    /// does, so a bulk-imported key and a singly-registered one are the same
    /// record.
    pub openssh_public_key: Option<String>,
    /// Grants to install, replacing nothing — the principal is new.
    pub grants: Vec<Grant>,
}

/// Mint a token secret: 256 bits of OS entropy, `gunnar_pat_<base64url>`.
///
/// Separate from [`AuthTable::register_token`], which is handed a secret and
/// stores only its hash — the same split as [`mint_ed25519`](crate::mint_ed25519)
/// and [`AuthTable::register_key`]. Minting and storing are different acts, and
/// keeping them apart is what makes it possible to state that the table never
/// holds a usable credential.
///
/// **The prefix is not decoration.** It is what makes a leaked token findable:
/// a bare base64 blob in a log or a commit is indistinguishable from any other
/// blob, and every forge that has been through this ended up prefixing for the
/// same reason. It costs eleven bytes.
pub fn mint_token_secret() -> String {
    use base64ct::Encoding as _;
    use rand::Rng as _;

    let mut bytes = [0u8; 32];
    // The OS generator panics rather than returning short: a server that
    // could not get entropy must not mint a guessable credential.
    crate::os_rng().fill_bytes(&mut bytes);
    format!(
        "gunnar_pat_{}",
        base64ct::Base64UrlUnpadded::encode_string(&bytes)
    )
}

/// An OpenSSH `SHA256:<base64>` public key fingerprint.
///
/// This is exactly what `ssh-keygen -lf` prints and what OpenSSH shows on an
/// unknown-host prompt: unpadded base64 of the sha256 of the wire encoding.
/// gunnar never computes "something like" a fingerprint. An operator has to be
/// able to compare what gunnar lists against what `ssh-keygen` says on their
/// laptop, so `tests/fingerprint.rs` pins it to known answers taken from the
/// real OpenSSH binary.
///
/// Every constructor routes through [`fingerprint_of_blob`], so there is one
/// definition and it is a definition over bytes.
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize)]
pub struct Fingerprint(String);

/// A certificate's DER, from bytes that may be PEM-armoured or already DER.
///
/// Lives beside [`fingerprint_of_certificate`] because the two have to agree
/// about what "the certificate bytes" are. If the caller hashed the armoured
/// form and this unwrapped it, the same certificate would register under two
/// fingerprints depending only on which shape somebody happened to hand over.
pub fn certificate_der(raw: &[u8]) -> Result<Vec<u8>> {
    use base64ct::Encoding as _;

    let text = String::from_utf8_lossy(raw);
    const BEGIN: &str = "-----BEGIN CERTIFICATE-----";
    const END: &str = "-----END CERTIFICATE-----";
    let Some(start) = text.find(BEGIN) else {
        // No armour: it is already DER.
        return Ok(raw.to_vec());
    };
    let body = &text[start + BEGIN.len()..];
    let end = body
        .find(END)
        .ok_or_else(|| Error::Key("the PEM has a BEGIN line and no END line".to_owned()))?;
    let b64: String = body[..end].chars().filter(|c| !c.is_whitespace()).collect();
    base64ct::Base64::decode_vec(&b64)
        .map_err(|e| Error::Key(format!("the PEM body is not valid base64: {e}")))
}

/// The same SHA256 fingerprint, taken over an X.509 certificate's DER.
///
/// # Why mTLS reuses this type rather than growing its own
///
/// A `Fingerprint` is the whole of what gunnar knows about a credential: the
/// `AuthTable` maps one to a principal, and every `Grant` check downstream works
/// on the principal. So a second way to *arrive* at a fingerprint is the entire
/// mTLS change — no new identity, no second grant path, nothing downstream aware
/// there are now two kinds of credential (the mTLS plan).
///
/// # Over the DER, not the public key
///
/// The whole certificate, so that re-issuing the same subject with the same key
/// under a new validity window produces a **different** fingerprint. That is the
/// conservative direction: a re-issued certificate has to be registered again,
/// rather than silently inheriting the grants of the one it replaced. A
/// deployment that wants key continuity across renewal wants a different scheme
/// than a fingerprint, and should not get it by accident here.
///
/// Any bytes are accepted, for the same reason [`fingerprint_of_blob`] accepts
/// any: this is a hash, not a parser, and refusing to hash a malformed
/// certificate would only turn "not registered" into a second answer.
pub fn fingerprint_of_certificate(der: &[u8]) -> Fingerprint {
    use base64ct::Encoding as _;
    use sha2::Digest as _;

    let digest = sha2::Sha256::digest(der);
    Fingerprint(format!(
        "{SHA256_PREFIX}{}",
        base64ct::Base64Unpadded::encode_string(&digest)
    ))
}

impl Fingerprint {
    /// Fingerprint a parsed key.
    ///
    /// Convenience over [`fingerprint_of_blob`] for the crates that link *this*
    /// `ssh-key`. `gunnar-ssh` cannot use it (see [`fingerprint_of_blob`]) and
    /// calls that function with `russh`'s key blob instead.
    pub fn of_key(key: &ssh_key::PublicKey) -> Self {
        // Encoding a key that was already parsed cannot fail: `to_bytes` only
        // writes the decoded `KeyData` into a fresh `Vec`.
        let blob = key
            .to_bytes()
            .expect("a parsed public key always re-encodes to its own wire form");
        fingerprint_of_blob(&blob)
    }

    /// Parse an `authorized_keys`-style line and fingerprint it.
    ///
    /// Accepts what OpenSSH writes: `<algo> <base64> [comment]`.
    pub fn of_openssh(line: &str) -> Result<Self> {
        Ok(Self::of_key(&parse_public_key(line)?))
    }

    /// Accept a fingerprint that arrived as text: config file, gRPC request,
    /// an operator's terminal.
    ///
    /// Validated rather than trusted: `SHA256:` plus 43 base64 characters. A
    /// malformed fingerprint that gets stored anyway is a key that can never be
    /// matched and never be revoked.
    pub fn parse(text: &str) -> Result<Self> {
        let body = text.strip_prefix(SHA256_PREFIX).ok_or_else(|| {
            Error::BadFingerprint(format!("{text:?}: missing the {SHA256_PREFIX} prefix"))
        })?;
        if body.len() != 43 {
            return Err(Error::BadFingerprint(format!(
                "{text:?}: expected 43 base64 characters after the prefix, got {}",
                body.len()
            )));
        }
        if !body
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || b == b'+' || b == b'/')
        {
            return Err(Error::BadFingerprint(format!(
                "{text:?}: not unpadded base64"
            )));
        }
        Ok(Fingerprint(text.to_owned()))
    }

    /// The `SHA256:...` text.
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl std::fmt::Display for Fingerprint {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

impl std::fmt::Debug for Fingerprint {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "Fingerprint({})", self.0)
    }
}

impl std::str::FromStr for Fingerprint {
    type Err = Error;
    fn from_str(s: &str) -> Result<Self> {
        Fingerprint::parse(s)
    }
}

/// Parse an `authorized_keys` line into a key.
///
/// Pure Rust, through `ssh-key`. There is no `ssh-keygen` subprocess anywhere
/// in gunnar (the no-OS server rule).
pub fn parse_public_key(line: &str) -> Result<ssh_key::PublicKey> {
    let line = line.trim();
    if line.is_empty() {
        return Err(Error::Key("empty".to_owned()));
    }
    ssh_key::PublicKey::from_openssh(line).map_err(Error::from)
}

/// The identity table as it is stored.
///
/// All three maps are sharded (`doc.rs`): at forge scale this document holds
/// every principal, every public key and every grant on the server, and the
/// pre-2026-08-03 layout put all of them inline in one blob that was parsed
/// four times per SSH connection.
///
/// **`grants` is keyed by principal**, not a `Vec`. It was a flat list, so
/// [`AuthTable::may`] — three calls deep on every connection, through
/// `Control::access_for` — scanned every grant on the server to answer one
/// question. Keyed, the answer is one shard probe and a walk of that
/// principal's own handful of grants.
#[derive(Default, Clone, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct AuthDoc {
    #[serde(default)]
    principals: ShardedMap<Principal>,
    /// Keyed by fingerprint text, which is what an SSH handshake produces.
    #[serde(default)]
    keys: ShardedMap<PublicKey>,
    /// Keyed by [`Principal::id`]: one principal's grants, sorted.
    #[serde(default)]
    grants: ShardedMap<Vec<Grant>>,
    /// Keyed by [`AccessToken::hash`], so an HTTP request costs one probe and
    /// never a scan.
    #[serde(default)]
    tokens: ShardedMap<AccessToken>,
    /// **Group membership, keyed by member** (punkt 9): one principal's id maps
    /// to the sorted ids of the groups it belongs to.
    ///
    /// Keyed by member and not by group **for the same reason `grants` is keyed
    /// by principal**: [`AuthTable::may`] runs on every connection and asks one
    /// question — *what may this caller reach* — so the walk it needs is "which
    /// groups is this caller in", which is one shard probe here. The inverse
    /// question, *who is in this group*, is a listing
    /// ([`AuthTable::members_of`]) and walks, exactly as `grants(None)` walks:
    /// a listing is allowed to be O(rows), a connection is not.
    ///
    /// A group is itself a principal, so a group id can appear as a *member*
    /// here too — that is nesting, and [`AuthTable::grant_principals`] resolves
    /// it transitively and cycle-safely.
    #[serde(default)]
    memberships: ShardedMap<Vec<String>>,
    /// **Whether bootstrap has ever happened, as its own record.**
    ///
    /// Not a `ShardedMap`, not derived from any of the four above, and never
    /// touched by [`AuthTable::remove_principal`] or
    /// [`AuthTable::revoke_key`]: it is a field beside the identity table, not
    /// a row in it. That separation is the whole security property. "Bootstrap
    /// has happened" spelled as `!keys.is_empty()` would mean *deleting every
    /// key reopens the unauthenticated enrolment window*, which is the first
    /// thing an attacker who can reach `Access.RemovePrincipal` would try.
    ///
    /// It rides in the same document, and therefore in the same
    /// compare-and-swap on `refs/gunnar/auth`, as the identity it installs.
    /// That is deliberate and is what makes "exactly one of two concurrent
    /// enrolments wins" true by construction rather than by a lock: the loser
    /// re-reads inside [`DocRef::update`] and finds the seal already set. A
    /// separate reference would need two writes and could be interrupted
    /// between them.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    bootstrap: Option<BootstrapRecord>,
    /// **Accounts that are locked out**, keyed by principal id (see
    /// [`crate::gate`]).
    ///
    /// In this document and not beside it, for the reason `bootstrap` is: the
    /// gate is read on the same code path as the grants it overrides, and a
    /// second document would mean a second read, a second cache and a second
    /// version — three places for the lockout and the ACL to disagree, with the
    /// permissive answer winning silently (the one-sandbox rule). Here they are one blob and
    /// one compare-and-swap: a gate and the grants it overrides are never seen
    /// from different versions of the table.
    ///
    /// Sharded like the other maps, because [`AuthTable::may`] probes it on
    /// **every** connection and a whole-table parse per SSH handshake is the
    /// cost the sharding removed.
    #[serde(default)]
    gates: ShardedMap<crate::gate::AccountGate>,
    /// **Display name → principal id**, so a name is unique by construction
    /// and a lookup by name is one probe rather than a walk.
    ///
    /// Written in the same compare-and-swap as the `principals` row it points
    /// at — by [`AuthTable::add_principal`], [`AuthTable::rename_principal`]
    /// and [`AuthTable::remove_principal`] — so the index and the row are never
    /// seen from two versions of the table.
    ///
    /// **Rows from before this index existed have no entry**, and need none:
    /// every one of them was created with `name == id`, so a probe of
    /// `principals` by the name finds it. [`AuthTable::principal_by_name`] and
    /// the uniqueness check both take that second step, which is what makes
    /// the index additive — no migration, and a table written by an older
    /// server reads exactly as it did.
    #[serde(default)]
    names: ShardedMap<String>,
    /// **Employee → the account that enrolled it** (owner, 2026-10-03: "a user
    /// starts an account. he can choose to enroll new users=employees then it
    /// becomes an enterprise account").
    ///
    /// Written ONLY by [`AuthTable::enroll_employee`], in the same
    /// compare-and-swap as the principal row it creates, so a principal is an
    /// employee from the moment it exists or never: there is no writer that
    /// binds an EXISTING principal to an account ("they might have repos
    /// already, that would be a risk for the company"). Removed with the
    /// principal ([`AuthTable::remove_principal`]). An account with at least
    /// one row here IS an enterprise account; there is no flag to set.
    #[serde(default)]
    employers: ShardedMap<String>,
    /// **Employee → the room it holds out of its account's room** (owner,
    /// 2026-10-04, "Add users"). Written in the same compare-and-swap as the
    /// employee it belongs to, and removed with it (the room goes back to the
    /// account). An employee from before rooms existed has no row and holds
    /// nothing.
    #[serde(default)]
    rooms: ShardedMap<EmployeeRoom>,
    /// **Account → its SEATS**: places for employees not yet named, each
    /// holding its room, filled one by one by a seat claim (an invite, later
    /// an import). Lots, not a counter, because two orders may differ in the
    /// room per seat.
    #[serde(default)]
    seats: ShardedMap<Vec<SeatLot>>,
    /// **Order → what it granted**, so a seller that retries a grant after a
    /// timeout is answered with the first one instead of granting twice.
    #[serde(default)]
    room_orders: ShardedMap<RoomGrant>,
    /// **Clans** (owner, 2026-10-05), keyed by the clan's name — which is its
    /// namespace. Here, beside the grants, for the reason `gates` is: rule 2½
    /// of [`AuthTable::may`] reads it on the same path, from the same version.
    /// Not written at all until the first clan exists, so a table with none
    /// is byte-for-byte what it was before clans.
    #[serde(default, skip_serializing_if = "ShardedMap::is_empty")]
    clans: ShardedMap<clan_table::Clan>,
    /// **Account → the clans it is in or invited to**, by name: what it gives
    /// its clans is one probe on the push path (its own room is its cap less
    /// that), and "my clans" is a probe, not a walk.
    #[serde(default, skip_serializing_if = "ShardedMap::is_empty")]
    clan_index: ShardedMap<Vec<String>>,
    /// **Web credentials — the people's passkeys** (Stage L, 2026-10-04),
    /// keyed by the credential id (base64url), which is what a login names.
    /// In this document for the reason `keys` are: a passkey and the principal
    /// it logs in as are one compare-and-swap, it goes with the principal
    /// ([`AuthTable::remove_principal`]), and it reaches the twin with the
    /// rest of the identity table — a reinstalled front keeps every login.
    #[serde(default)]
    credentials: ShardedMap<crate::model::WebCredential>,
    /// **Invites — pending passkey enrolments** (Stage L), keyed by the
    /// invite id. Here for the reason `credentials` are: redeeming one and
    /// storing the passkey it enrols are one compare-and-swap, and an invite
    /// goes with its principal.
    #[serde(default)]
    invites: ShardedMap<crate::model::FirstInviteRow>,
    /// **Redeemed invites** (2026-10-19), keyed by the secret's hash: so
    /// `Access.CheckInvite` can say USED rather than UNKNOWN for an invite a
    /// passkey was registered with, and a console refuses it before a second
    /// passkey replaces the first on the person's device. Written in the
    /// redemption's own compare-and-swap, kept [`REDEEMED_INVITE_KEEP_MS`],
    /// removed with the principal. Written only once it holds a row, as the
    /// SSO maps are: an older gunnar still reads a table that never had one.
    #[serde(default, skip_serializing_if = "ShardedMap::is_empty")]
    redeemed_invites: ShardedMap<RedeemedInvite>,
    /// **The operator's looks into accounts** (2026-10-05), keyed by
    /// `account␟operator`: the latest request per pair. Here, in the table
    /// [`AuthTable::may`] reads, for the reason `gates` is: an approved look
    /// widens grants and must be read from the same version as they are.
    ///
    /// Written only once it holds a row, as `clans` and SSO's maps are: a
    /// table no operator ever asked into stays readable by an older gunnar.
    #[serde(default, skip_serializing_if = "ShardedMap::is_empty")]
    operator_access: ShardedMap<operator_access::OperatorAccess>,
    /// **Company SSO** (owner, 2026-10-05): each account's identity provider,
    /// keyed by the account. No secret is in it (see [`crate::sso`]).
    ///
    /// This map and the five SSO maps below are written only once they hold
    /// a row (as `clans` are), and dropped again with their last one: a table
    /// that never used SSO is byte-for-byte what it was before SSO, so an
    /// older gunnar — whose `AuthDoc` is `deny_unknown_fields` — still reads
    /// it on a downgrade or a twin not yet upgraded.
    #[serde(default, skip_serializing_if = "ShardedMap::is_empty")]
    sso_providers: ShardedMap<crate::sso::SsoProvider>,
    /// Email domain → the account whose IdP it leads to. One domain, one
    /// account; written with the provider row.
    #[serde(default, skip_serializing_if = "ShardedMap::is_empty")]
    sso_domains: ShardedMap<String>,
    /// **The companies' directories**: principal id → its directory row. An
    /// employee from SSO, SCIM or a CSV import has one; written in the same
    /// compare-and-swap as the employee, removed with it.
    #[serde(default, skip_serializing_if = "ShardedMap::is_empty")]
    sso_users: ShardedMap<crate::sso::SsoUser>,
    /// `s␟<account>␟<subject key>` and `u␟<account>␟<user name>` → principal
    /// id: a login resolves its subject in one probe, never across accounts.
    #[serde(default, skip_serializing_if = "ShardedMap::is_empty")]
    sso_index: ShardedMap<String>,
    /// SCIM bearers by the SHA-256 of their secret.
    #[serde(default, skip_serializing_if = "ShardedMap::is_empty")]
    scim_tokens: ShardedMap<crate::sso::ScimToken>,
    /// SCIM groups by id.
    #[serde(default, skip_serializing_if = "ShardedMap::is_empty")]
    sso_groups: ShardedMap<crate::sso::SsoGroup>,
}

impl_doc!(AuthDoc {
    principals,
    keys,
    grants,
    tokens,
    memberships,
    gates,
    names,
    clans,
    clan_index,
    employers,
    rooms,
    seats,
    room_orders,
    credentials,
    invites,
    redeemed_invites,
    operator_access,
    sso_providers,
    sso_domains,
    sso_users,
    sso_index,
    scim_tokens,
    sso_groups
});

/// How long a redeemed invite is remembered as USED: an invite lives at most
/// thirty days, so a link older than that reads expired anyway.
pub const REDEEMED_INVITE_KEEP_MS: UnixMillis = 30 * 24 * 60 * 60 * 1000;

/// **An invite that was redeemed** (2026-10-19): whose passkey it enrolled, and when.
#[derive(Clone, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
pub struct RedeemedInvite {
    /// The principal whose passkey it enrolled.
    pub principal_id: String,
    /// When it was redeemed.
    pub redeemed: UnixMillis,
}

/// **The room one employee holds** out of its account's room.
#[derive(Clone, PartialEq, Eq, Debug, Default, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EmployeeRoom {
    /// Bytes.
    pub room_bytes: u64,
    /// The order it came with (the seat's, for a filled seat); empty: none.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub order: String,
}

/// **A block of SEATS** an account holds: `total` places for employees not
/// yet named, each holding `room_bytes` of the account's room.
#[derive(Clone, PartialEq, Eq, Debug, Default, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SeatLot {
    /// Bytes per seat.
    pub room_bytes: u64,
    /// How many seats the lot was granted with.
    pub total: u32,
    /// The employees that filled a seat, in the order they came. A removed
    /// employee leaves this list, and its seat is open again.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub filled: Vec<String>,
    /// The order that granted it; empty: none.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub order: String,
    /// When.
    pub at: UnixMillis,
}

impl SeatLot {
    /// Seats not yet filled.
    pub fn open(&self) -> u32 {
        self.total
            .saturating_sub(u32::try_from(self.filled.len()).unwrap_or(u32::MAX))
    }
}

/// **What an order granted**, kept for its retry.
#[derive(Clone, PartialEq, Eq, Debug, Default, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RoomGrant {
    /// The account.
    pub account: String,
    /// The named employee, or empty for seats.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub principal: String,
    /// Seats granted (0 for a named employee).
    #[serde(default)]
    pub seats: u32,
    /// Bytes per employee or seat.
    pub room_bytes: u64,
    /// What it asked of the account, bytes.
    #[serde(default)]
    pub need_bytes: u64,
}

/// **What an enrolment asks of the account's room** ("Add users").
#[derive(Clone, PartialEq, Eq, Debug, Default)]
pub struct RoomAsk {
    /// Bytes per employee or seat. Ignored for a seat claim: the seat's.
    pub room_bytes: u64,
    /// Seats to grant, unnamed. 0: one named employee.
    pub seats: u32,
    /// Fill an open seat with the named employee instead of asking for room.
    pub from_seat: bool,
    /// The account's room: its cap. `None`: nothing caps this account on
    /// this server.
    pub cap: Option<u64>,
    /// What the account stores now, bytes (counted against `cap`).
    pub stored: u64,
    /// The order this belongs to (idempotency); empty: none.
    pub order: String,
    /// Ask, write nothing.
    pub dry_run: bool,
}

/// **The answer** to a [`RoomAsk`].
#[derive(Clone, PartialEq, Eq, Debug, Default)]
pub struct RoomAnswer {
    /// The named employee (unset for seats and for a dry run).
    pub principal: Option<Principal>,
    /// Bytes each new employee or seat holds.
    pub room_bytes: u64,
    /// The account's seats after this: every seat granted…
    pub seats: u32,
    /// …and the ones not yet filled.
    pub seats_open: u32,
    /// What was asked of the account's room, bytes.
    pub need_bytes: u64,
    /// The account's room neither stored nor promised, before this; `None`
    /// when nothing caps it.
    pub unused_bytes: Option<u64>,
    /// How much is missing (a dry run's answer; the real call refuses).
    pub shortfall_bytes: u64,
    /// Nothing was written: a dry run, or the retry of a granted order.
    pub unchanged: bool,
}

/// **The record that the unauthenticated enrolment window has been used.**
///
/// Written once, never rewritten, never removed. Every field is here so an
/// operator can answer *who opened this server, with which key, and when* from
/// the server itself rather than from a log that may have rotated.
#[derive(Clone, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct BootstrapRecord {
    /// When the window closed.
    pub at: UnixMillis,
    /// The principal that was created. It may since have been removed; the
    /// record still stands, which is the point.
    pub principal_id: String,
    /// The `SHA256:…` fingerprint of the key it was created with.
    pub fingerprint: String,
    /// The account (namespace) that principal is intrinsically admin of. Equal
    /// to `principal_id`; carried explicitly so a later change to that rule is
    /// a visible schema change and not a silent re-derivation.
    pub account: String,
}

/// What [`AuthTable::bootstrap_admin`] did.
#[derive(Clone, PartialEq, Eq, Debug)]
pub enum Bootstrap {
    /// The table was empty and this admin was installed.
    Installed(Principal),
    /// The table already held keys; nothing was changed.
    AlreadyProvisioned,
}

/// **What state the identity table is in at start-up, and therefore whether the
/// unauthenticated enrolment door is open.**
///
/// Until 2026-08-05 there were two states and one of them was *refuse to
/// start*: a server whose identity table was empty and which had been given no
/// `--admin-key` bailed out of `Server::start`. That is why there was no first
/// boot at all — the box could not be reached to be given its first key, and
/// the only way in was `gunnar key add` **with the server down**, which needs a
/// shell on the machine and there is no shell on the appliance.
///
/// The refusal is replaced by [`Readiness::AwaitingBootstrap`]: the server
/// starts, and `Bootstrap.Enroll` is servable **without a credential** for
/// exactly as long as [`AuthTable::bootstrap_record`] is `None`.
///
/// Three states, and each predicate below answers `false` for at least one of
/// them: a `may_serve()` that had become true for every state would be an
/// identity value, and a test sitting on it could not tell a working gate from
/// a dead one (measured, not asserted).
#[derive(Clone, PartialEq, Eq, Debug)]
pub enum Readiness {
    /// There is at least one registered key, and bootstrap has happened.
    /// Ordinary operation.
    Ready,
    /// **Nothing has ever been enrolled.** The identity table is empty *and*
    /// there is no [`BootstrapRecord`]. `Bootstrap.Enroll` is open, to anyone
    /// who can reach the control port. This must be said out loud at every
    /// start and reported in status; see `gunnar_grpc::service::bootstrap`.
    AwaitingBootstrap,
    /// **The enrolment window is shut and there is no key to authenticate
    /// with.** Nobody can reach this server over either listener until a key is
    /// put back with `gunnar key add`, offline — and the window stays shut
    /// regardless, which is the attack this design exists to refuse.
    SealedAndEmpty {
        /// Who bootstrapped it, and when, so the operator can tell an attack
        /// from an accident.
        ///
        /// `None` for a table that holds principals but no keys and carries no
        /// bootstrap record: a half-provisioned data path written before the
        /// record existed. That is a real state and must be described as
        /// itself rather than with an invented record.
        record: Option<Box<BootstrapRecord>>,
    },
}

impl Readiness {
    /// Is the unauthenticated `Bootstrap.Enroll` window open?
    ///
    /// True for exactly one of the three states.
    pub fn bootstrap_window_open(&self) -> bool {
        matches!(self, Readiness::AwaitingBootstrap)
    }

    /// Can anybody actually authenticate against this server right now?
    ///
    /// False for both non-`Ready` states, for two different reasons, which is
    /// why this is not the negation of [`Readiness::bootstrap_window_open`].
    pub fn has_an_identity(&self) -> bool {
        matches!(self, Readiness::Ready)
    }
}

impl std::fmt::Display for Readiness {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Readiness::Ready => f.write_str("ready"),
            Readiness::AwaitingBootstrap => f.write_str(
                "AWAITING BOOTSTRAP: no identity has ever been enrolled, so Bootstrap.Enroll is \
                 open to any caller that can reach the control port",
            ),
            Readiness::SealedAndEmpty {
                record: Some(record),
            } => write!(
                f,
                "bootstrapped at {} by {} but every identity has since been removed; the \
                 enrolment window stays SHUT and nobody can reach this server",
                record.at, record.principal_id
            ),
            Readiness::SealedAndEmpty { record: None } => f.write_str(
                "this identity table holds principals but no keys and carries no bootstrap \
                 record; the enrolment window stays SHUT and nobody can reach this server",
            ),
        }
    }
}

/// Principals, their keys, and the grants between them and repositories.
///
/// Cheap to clone; every clone reads and writes the same document.
#[derive(Clone, Debug)]
pub struct AuthTable {
    doc: DocRef<AuthDoc>,
}

impl AuthTable {
    /// Open the identity table held in `store` (the certificate-identity plan).
    pub fn new(store: Arc<dyn ControlStore>) -> Result<Self> {
        Ok(AuthTable {
            doc: DocRef::new(store, AUTH_REF)?,
        })
    }

    /// The name the table is stored under.
    pub fn ref_name(&self) -> &str {
        self.doc.ref_name()
    }

    /// **Every version this table has ever had**, oldest first (the certificate-identity plan).
    ///
    /// A ref update per change used to be a free audit trail; on a key/value
    /// store it is an explicit append-only table, and this is the door to it.
    /// Each entry's `blob` is still readable through the control store, so
    /// "what did the ACL look like before Tuesday" is answerable from the
    /// server rather than from a log that may have rotated.
    pub fn history(&self) -> Result<Vec<crate::control::DocVersion>> {
        self.doc.history()
    }

    // ── principals ─────────────────────────────────────────────────────────

    /// Add a principal.
    ///
    /// Refuses [`ANONYMOUS_PRINCIPAL_ID`](crate::model::ANONYMOUS_PRINCIPAL_ID)
    /// [`Principal::new`] refuses it too, but this is the door a
    /// [`Principal::anonymous`] *value* would arrive at. A principal sitting in
    /// the table under that name would be handed the public-read rule of
    /// [`AuthTable::may`] and could then be given keys and grants on top of it.
    ///
    /// **The table stamps `created`** with its own clock, whatever the caller
    /// put there: one writer for the field, so a row's creation time is the
    /// server's word and not the client's.
    ///
    /// The name is refused if another principal carries it — as its name or as
    /// its id ([`Error::NameExists`]). The id is what everything keys on; the
    /// name is the label; the label is still unique, in the same
    /// compare-and-swap as the row, so a listing can never show two rows a
    /// human reads as one.
    pub fn add_principal(&self, principal: Principal) -> Result<Principal> {
        crate::sandbox::check_component(&principal.id, &principal.id)?;
        if principal.id == crate::model::ANONYMOUS_PRINCIPAL_ID {
            return Err(Error::ReservedPrincipalId(principal.id));
        }
        let principal = Principal {
            created: now_ms(),
            ..principal
        };
        self.doc.update(|doc| {
            if doc.principals.contains_key(&principal.id)? {
                return Err(Error::PrincipalExists(principal.id.clone()));
            }
            Self::refuse_taken_name(doc, &principal.name, &principal.id)?;
            doc.principals
                .insert(principal.id.clone(), principal.clone())?;
            Self::index_name(doc, &principal)?;
            Ok(principal.clone())
        })
    }

    /// **Change a principal's display name and nothing else.**
    ///
    /// The id — and with it the account namespace, every key, grant,
    /// membership and gate — stays exactly where it is; only the label moves.
    /// This is what makes a rename an edit rather than a re-identification: a
    /// console that pinned the id sees the same principal afterwards, and a
    /// URL, a policy file or a grant that names the id is untouched.
    ///
    /// Refuses a name another principal already carries, as its name or as its
    /// id ([`Error::NameExists`]). Renaming to the name it already has is a
    /// no-op that succeeds.
    pub fn rename_principal(&self, id: &str, name: &str) -> Result<Principal> {
        self.doc.update(|doc| {
            let before = doc
                .principals
                .get(id)?
                .ok_or_else(|| Error::NoSuchPrincipal(id.to_owned()))?;
            if before.name == name {
                return Ok(before);
            }
            Self::refuse_taken_name(doc, name, id)?;
            let after = Principal {
                name: name.to_owned(),
                ..before.clone()
            };
            // The old index entry goes only if it was this principal's. A row
            // from before the index has no entry, and `remove` of an absent key
            // is the same no-op.
            if doc.names.get(&before.name)?.as_deref() == Some(id) {
                doc.names.remove(&before.name)?;
            }
            doc.principals.insert(id.to_owned(), after.clone())?;
            Self::index_name(doc, &after)?;
            Ok(after)
        })
    }

    /// **One principal, by its display name.** One index probe, then one row
    /// probe for a principal from before the index (whose name is its id).
    ///
    /// This is the lookup a console makes at the one moment a name is anybody's
    /// to give — registration — before it pins the id and stops asking by name.
    pub fn principal_by_name(&self, name: &str) -> Result<Option<Principal>> {
        let doc = self.doc.load()?;
        Self::resolve_name(&doc, name)
    }

    /// The principal `name` resolves to, through the index first and the
    /// pre-index fallback second. One reader for both
    /// [`AuthTable::principal_by_name`] and [`AuthTable::refuse_taken_name`], so
    /// "what does this name mean" and "is this name free" cannot disagree.
    fn resolve_name(doc: &AuthDoc, name: &str) -> Result<Option<Principal>> {
        if let Some(id) = doc.names.get(name)? {
            return doc.principals.get(&id);
        }
        // Before the index every row was `name == id`, so the name is the key.
        // The name check matters: a row whose name has since been changed is
        // still at this key, and it is not the principal called `name` now.
        Ok(doc
            .principals
            .get(name)?
            .filter(|principal| principal.name == name))
    }

    /// Refuse `name` unless it is free, or is `own_id`'s own.
    ///
    /// Taken means: another principal's name (through [`AuthTable::resolve_name`])
    /// or another principal's **id** — the second rule is what stops a display
    /// name from ever spelling somebody else's identity on the wire.
    fn refuse_taken_name(doc: &AuthDoc, name: &str, own_id: &str) -> Result<()> {
        let holder = match Self::resolve_name(doc, name)? {
            Some(principal) => Some(principal.id),
            None if name != own_id && doc.principals.contains_key(name)? => Some(name.to_owned()),
            None => None,
        };
        match holder {
            Some(id) if id != own_id => Err(Error::NameExists {
                name: name.to_owned(),
                principal_id: id,
            }),
            _ => Ok(()),
        }
    }

    /// Write the index entry for `principal`'s name. The one writer, used by
    /// every path that puts a row in `principals`.
    fn index_name(doc: &mut AuthDoc, principal: &Principal) -> Result<()> {
        // A clan's name is its namespace, and a principal is admin of the
        // namespace named after its id (`owns_account`): no principal is ever
        // made under a clan's name, on any path that makes one.
        if !doc.clans.is_empty() && doc.clans.contains_key(&principal.id)? {
            return Err(Error::PrincipalExists(principal.id.clone()));
        }
        doc.names
            .insert(principal.name.clone(), principal.id.clone())?;
        Ok(())
    }

    /// Remove a principal, its keys and its grants.
    ///
    /// All three together, by construction: a principal whose keys outlived it
    /// is an account that still authenticates.
    ///
    /// **O(keys on the server)**. Keys are stored by fingerprint, because that
    /// is what the connection path looks them up by, so finding one principal's
    /// keys is a walk. Deleting an account happens once a day, not once a
    /// connection; making it cheap would mean a second index to keep in step
    /// with the first, which is the twin that one implementation refuses.
    pub fn remove_principal(&self, id: &str) -> Result<Principal> {
        self.doc.update(|doc| {
            let principal = doc
                .principals
                .remove(id)?
                .ok_or_else(|| Error::NoSuchPrincipal(id.to_owned()))?;
            // The name goes with the row, in the same update, so the name is
            // claimable again exactly when the id is. Only if the entry is this
            // principal's: a pre-index row has none, and `remove` of an absent
            // key is the same no-op.
            if doc.names.get(&principal.name)?.as_deref() == Some(id) {
                doc.names.remove(&principal.name)?;
            }
            doc.keys.retain(|_, key| key.principal_id != id)?;
            // **Tokens go with the principal, for the reason the gate does**
            // (GUN-026).
            //
            // Leaving them was nearly harmless and not quite:
            // `principal_by_token_at` ends in `doc.principals.get(&token.\
            // principal_id)`, so an orphaned token authenticates as nobody and
            // the credential is dead the moment its holder is. But a principal
            // id is CLAIMABLE AGAIN the moment the principal is gone —
            // `add_principal` refuses only a name that is currently taken — and
            // a token row is keyed by its hash and names its principal by that
            // id. So recreating `alice` handed the new alice every access token
            // the old one had, and neither the operator who removed the
            // principal nor the one who created it saw a line about it. That is
            // the gate bug below in its other direction: there, a leftover row
            // locked the next principal out; here, it lets somebody else in.
            doc.tokens.retain(|_, token| token.principal_id != id)?;
            doc.grants.remove(id)?;
            // Membership, both directions. A principal whose membership edges
            // outlived it would keep inheriting grants (its own list) or keep
            // conferring them (if it was a group other principals were in), so
            // both are cut in the same update, by construction — the same rule
            // as keys and grants above.
            doc.memberships.remove(id)?;
            let strip: Vec<(String, Vec<String>)> = doc
                .memberships
                .entries()?
                .into_iter()
                .filter(|(_, groups)| groups.iter().any(|g| g == id))
                .map(|(member, mut groups)| {
                    groups.retain(|g| g != id);
                    (member, groups)
                })
                .collect();
            for (member, groups) in strip {
                if groups.is_empty() {
                    doc.memberships.remove(&member)?;
                } else {
                    doc.memberships.insert(member, groups)?;
                }
            }
            // **The gate goes with the principal, in the same update.**
            //
            // A gate is keyed by principal id, and a principal id is claimable
            // again the moment the principal is gone (`add_principal` only
            // refuses a name that is currently taken). A row left behind here
            // would therefore lock out *the next* principal created under this
            // name — an account bricked before it existed, by a decision taken
            // about somebody else. `tests/gate.rs` drives exactly that
            // sequence.
            doc.gates.remove(id)?;
            // An employee's binding to its account goes with it, so the id —
            // claimable again from here — never comes back as somebody's
            // employee.
            doc.employers.remove(id)?;
            // Its room goes back to the account, and the seat it filled (if
            // any) is open again; an account that goes takes its seats and its
            // orders with it.
            doc.rooms.remove(id)?;
            let reopen: Vec<(String, Vec<SeatLot>)> = doc
                .seats
                .entries()?
                .into_iter()
                .filter(|(_, lots)| lots.iter().any(|lot| lot.filled.iter().any(|f| f == id)))
                .map(|(account, mut lots)| {
                    for lot in &mut lots {
                        lot.filled.retain(|f| f != id);
                    }
                    (account, lots)
                })
                .collect();
            for (account, lots) in reopen {
                doc.seats.insert(account, lots)?;
            }
            doc.seats.remove(id)?;
            doc.room_orders
                .retain(|_, grant| grant.account != id && grant.principal != id)?;
            // Its clans (2026-10-05): out of every one, its contribution with
            // it; a clan it was the last admin of goes to its longest member.
            Self::forget_clan_member(doc, id)?;
            // Its passkeys go too (Stage L): a credential that outlived its
            // principal would log the NEXT principal of that id in.
            doc.credentials
                .retain(|_, credential| credential.principal_id != id)?;
            // …and its invites: a pending enrolment on a principal that is
            // gone would enrol the next principal of that id.
            doc.invites.retain(|_, invite| invite.principal_id != id)?;
            doc.redeemed_invites
                .retain(|_, row| row.principal_id != id)?;
            // …and the operator's looks into its account, or its own looks
            // into others': a grant keyed by a name that is claimable again.
            doc.operator_access
                .retain(|_, look| look.account != id && look.operator != id)?;
            // …and its directory row and the index entries that lead to it
            // (SSO): a subject bound to a principal that is gone would log
            // the NEXT principal of that id in. An account's own SSO rows go
            // with the account.
            doc.sso_users.remove(id)?;
            doc.sso_index.retain(|_, held| held != id)?;
            doc.sso_users.retain(|_, user| user.account != id)?;
            doc.sso_index
                .retain(|key, _| !Self::sso_index_of_account(key, id))?;
            if let Some(provider) = doc.sso_providers.remove(id)? {
                for domain in provider.domains {
                    doc.sso_domains.remove(&domain)?;
                }
            }
            doc.scim_tokens.retain(|_, token| token.account != id)?;
            doc.sso_groups.retain(|_, group| group.account != id)?;
            for (gid, mut group) in doc.sso_groups.entries()? {
                if group.members.iter().any(|m| m == id) {
                    group.members.retain(|m| m != id);
                    doc.sso_groups.insert(gid, group)?;
                }
            }
            Ok(principal)
        })
    }

    /// **Enroll an employee: a NEW user, created by and bound to `account`.**
    ///
    /// The only way a principal becomes an employee, and it can only make a
    /// new one: the id must not exist (an existing principal "might have repos
    /// already, that would be a risk for the company") and the name is checked
    /// as for every principal. In ONE compare-and-swap it writes the principal
    /// row, its name, the employee binding and a `Read` grant on the account's
    /// namespace — a member sees the account's repositories and events; what
    /// it may write is granted per repository by the account admin.
    ///
    /// An employee gets NO account of its own (no namespace named after it;
    /// `CoreControlPlane::enroll_employee` refuses an id that names one), so
    /// the intrinsic "your own account" rule never makes it an admin of
    /// anything, and it may not make a repository public anywhere: that is the
    /// account admin's ([`AuthTable::may_namespace`] at `Admin`).
    ///
    /// Who may enroll (the account's admin) is the caller's question; the
    /// account must exist as a namespace, which the control plane checks. Only
    /// a [`PrincipalKind::User`](crate::PrincipalKind::User) can be enrolled.
    pub fn enroll_employee(&self, account: &str, principal: Principal) -> Result<Principal> {
        let answer = self.enroll_with_room(account, Some(principal), &RoomAsk::default())?;
        answer.principal.ok_or_else(|| Error::InvalidInput {
            what: "employee",
            reason: "the enrolment named nobody".to_owned(),
        })
    }

    /// **Enroll with room** (owner, 2026-10-04, "Add users"): ONE named
    /// employee holding `room_bytes`, or `seats` unnamed seats each holding
    /// it, or a seat claim (`from_seat`: the named employee takes an open
    /// seat and its room). The room comes out of what the account holds and
    /// has neither stored nor promised — `budget` minus the room its
    /// employees and open seats hold — and in the SAME compare-and-swap as
    /// the rows it writes, so two enrolments can never both spend the last of
    /// it. Short: [`Error::RoomShort`] naming what is missing, nothing
    /// written. A dry run answers with the shortfall instead and writes
    /// nothing; a retried `order` is answered with its first grant.
    ///
    /// The rules [`AuthTable::enroll_employee`] keeps for a named employee
    /// hold here unchanged: a NEW principal only, a person, not the account.
    pub fn enroll_with_room(
        &self,
        account: &str,
        principal: Option<Principal>,
        ask: &RoomAsk,
    ) -> Result<RoomAnswer> {
        let principal = match principal {
            Some(principal) => {
                crate::sandbox::check_component(&principal.id, &principal.id)?;
                if principal.id == crate::model::ANONYMOUS_PRINCIPAL_ID {
                    return Err(Error::ReservedPrincipalId(principal.id));
                }
                if principal.kind != crate::model::PrincipalKind::User {
                    return Err(Error::InvalidInput {
                        what: "employee",
                        reason: format!(
                            "an employee is a person; {} is a {:?}, enroll a user",
                            principal.id, principal.kind
                        ),
                    });
                }
                if principal.id == account {
                    return Err(Error::InvalidInput {
                        what: "employee",
                        reason: format!("{account} is the account itself, not an employee of it"),
                    });
                }
                Some(Principal {
                    created: now_ms(),
                    ..principal
                })
            }
            None => None,
        };
        if principal.is_none() && (ask.seats == 0 || ask.from_seat) {
            return Err(Error::InvalidInput {
                what: "employee",
                reason: "one employee, or a seat filled, needs a name".to_owned(),
            });
        }
        if principal.is_some() && ask.seats > 0 {
            return Err(Error::InvalidInput {
                what: "employee",
                reason: "seats name nobody: a seat is named when it is filled".to_owned(),
            });
        }
        if ask.dry_run {
            let doc = self.doc.load()?;
            return Self::plan_room(&doc, account, principal.as_ref(), ask).map(|plan| plan.answer);
        }
        self.doc.update(|doc| {
            let plan = Self::plan_room(doc, account, principal.as_ref(), ask)?;
            if plan.answer.unchanged {
                return Ok(plan.answer);
            }
            if plan.answer.shortfall_bytes > 0 {
                return Err(Error::RoomShort {
                    account: account.to_owned(),
                    need: plan.answer.need_bytes,
                    unused: plan.answer.unused_bytes.unwrap_or(0),
                    short: plan.answer.shortfall_bytes,
                });
            }
            let mut answer = plan.answer.clone();
            let mut seats = doc.seats.get(account)?.unwrap_or_default();
            match &principal {
                Some(principal) => {
                    Self::write_employee(
                        doc,
                        account,
                        principal,
                        plan.lot,
                        &ask.order,
                        answer.room_bytes,
                        &mut seats,
                    )?;
                    answer.principal = Some(principal.clone());
                }
                None => seats.push(SeatLot {
                    room_bytes: answer.room_bytes,
                    total: ask.seats,
                    filled: Vec::new(),
                    order: ask.order.clone(),
                    at: now_ms(),
                }),
            }
            answer.seats = seats.iter().map(|lot| lot.total).sum();
            answer.seats_open = seats.iter().map(SeatLot::open).sum();
            if !seats.is_empty() {
                doc.seats.insert(account.to_owned(), seats)?;
            }
            if !ask.order.is_empty() {
                doc.room_orders.insert(
                    ask.order.clone(),
                    RoomGrant {
                        account: account.to_owned(),
                        principal: principal.as_ref().map(|p| p.id.clone()).unwrap_or_default(),
                        seats: ask.seats,
                        room_bytes: answer.room_bytes,
                        need_bytes: answer.need_bytes,
                    },
                )?;
            }
            Ok(answer)
        })
    }

    /// **The rows one NEW named employee is**, inside the caller's
    /// compare-and-swap: the principal and its name, the employee binding, a
    /// `Read` grant on the account's namespace, the seat it fills (when
    /// `lot`), and the room it holds. The one writer of an employee —
    /// [`AuthTable::enroll_with_room`] and an SSO first login both call it.
    fn write_employee(
        doc: &mut AuthDoc,
        account: &str,
        principal: &Principal,
        lot: Option<usize>,
        order: &str,
        room_bytes: u64,
        seats: &mut [SeatLot],
    ) -> Result<()> {
        doc.principals
            .insert(principal.id.clone(), principal.clone())?;
        Self::index_name(doc, principal)?;
        doc.employers
            .insert(principal.id.clone(), account.to_owned())?;
        doc.grants.insert(
            principal.id.clone(),
            vec![Grant {
                principal: principal.id.clone(),
                scope: Scope::Namespace(account.to_owned()),
                level: Level::Read,
            }],
        )?;
        let order = match lot {
            Some(lot) => {
                seats[lot].filled.push(principal.id.clone());
                seats[lot].order.clone()
            }
            None => order.to_owned(),
        };
        if room_bytes > 0 {
            doc.rooms
                .insert(principal.id.clone(), EmployeeRoom { room_bytes, order })?;
        }
        Ok(())
    }

    /// The decision, read-only: what the ask needs, what is unused, which seat
    /// a claim takes, and whether `order` was already granted.
    fn plan_room(
        doc: &AuthDoc,
        account: &str,
        principal: Option<&Principal>,
        ask: &RoomAsk,
    ) -> Result<RoomPlan> {
        let lots = doc.seats.get(account)?.unwrap_or_default();
        let seats = lots.iter().map(|lot| lot.total).sum();
        let seats_open = lots.iter().map(SeatLot::open).sum();
        if !ask.order.is_empty() {
            if let Some(grant) = doc.room_orders.get(&ask.order)? {
                if grant.account != account {
                    return Err(Error::InvalidInput {
                        what: "order",
                        reason: format!(
                            "order {} was granted to {}, not to {account}",
                            ask.order, grant.account
                        ),
                    });
                }
                let principal = match grant.principal.as_str() {
                    "" => None,
                    id => doc.principals.get(id)?,
                };
                return Ok(RoomPlan {
                    answer: RoomAnswer {
                        principal,
                        room_bytes: grant.room_bytes,
                        seats,
                        seats_open,
                        need_bytes: grant.need_bytes,
                        unused_bytes: None,
                        shortfall_bytes: 0,
                        unchanged: true,
                    },
                    lot: None,
                });
            }
        }
        if let Some(principal) = principal {
            if doc.principals.contains_key(&principal.id)? {
                return Err(Error::PrincipalExists(principal.id.clone()));
            }
            Self::refuse_taken_name(doc, &principal.name, &principal.id)?;
        }
        let lot = if ask.from_seat {
            Some(
                lots.iter()
                    .position(|lot| lot.open() > 0)
                    .ok_or_else(|| Error::NoOpenSeat(account.to_owned()))?,
            )
        } else {
            None
        };
        let room_bytes = match lot {
            Some(lot) => lots[lot].room_bytes,
            None => ask.room_bytes,
        };
        let need_bytes = match (lot, ask.seats) {
            (Some(_), _) => 0,
            (None, 0) => room_bytes,
            (None, seats) => room_bytes.saturating_mul(u64::from(seats)),
        };
        // What the account already holds that is NOT free: what it stores and
        // what its employees and open seats hold. Counted in full — an
        // account storing past its cap is short by that much too, so the
        // shortfall is everything a sale must add for the ask to fit.
        // What it gives its clans is promised too (2026-10-05): a byte given
        // to a clan is not a byte an employee may hold.
        let committed = match ask.cap {
            None => None,
            Some(_) => Some(
                Self::promised_by(doc, account)?
                    .saturating_add(Self::contributed_in(doc, account, None)?)
                    .saturating_add(ask.stored),
            ),
        };
        let unused_bytes = ask
            .cap
            .zip(committed)
            .map(|(cap, committed)| cap.saturating_sub(committed));
        let shortfall_bytes = ask.cap.zip(committed).map_or(0, |(cap, committed)| {
            need_bytes.saturating_add(committed).saturating_sub(cap)
        });
        Ok(RoomPlan {
            answer: RoomAnswer {
                principal: None,
                room_bytes,
                seats,
                seats_open,
                need_bytes,
                unused_bytes,
                shortfall_bytes,
                unchanged: ask.dry_run,
            },
            lot,
        })
    }

    /// **What an account has promised its people**: the room its employees
    /// hold and its open seats hold, bytes. The one count `plan_room` and a
    /// clan contribution both take.
    fn promised_by(doc: &AuthDoc, account: &str) -> Result<u64> {
        let lots = doc.seats.get(account)?.unwrap_or_default();
        let mut promised: u64 = lots
            .iter()
            .map(|lot| lot.room_bytes.saturating_mul(u64::from(lot.open())))
            .fold(0, u64::saturating_add);
        for (id, employer) in doc.employers.entries()? {
            if employer == account {
                if let Some(room) = doc.rooms.get(&id)? {
                    promised = promised.saturating_add(room.room_bytes);
                }
            }
        }
        Ok(promised)
    }

    /// **The account's seats**: every seat granted, and the open ones.
    pub fn seats_of(&self, account: &str) -> Result<(u32, u32)> {
        let lots = self.doc.load()?.seats.get(account)?.unwrap_or_default();
        Ok((
            lots.iter().map(|lot| lot.total).sum(),
            lots.iter().map(SeatLot::open).sum(),
        ))
    }

    /// The room an employee holds, if any.
    pub fn room_of(&self, principal_id: &str) -> Result<Option<EmployeeRoom>> {
        self.doc.load()?.rooms.get(principal_id)
    }

    /// The account that enrolled this principal, if it is an employee.
    pub fn employer_of(&self, principal_id: &str) -> Result<Option<String>> {
        self.doc.load()?.employers.get(principal_id)
    }

    /// Every employee an account has enrolled, by id. A listing: it walks the
    /// bindings, as `members_of` walks memberships.
    pub fn employees_of(&self, account: &str) -> Result<Vec<Principal>> {
        let doc = self.doc.load()?;
        let mut out = Vec::new();
        for (id, employer) in doc.employers.entries()? {
            if employer == account {
                if let Some(p) = doc.principals.get(&id)? {
                    out.push(p);
                }
            }
        }
        out.sort_by(|a, b| a.id.cmp(&b.id));
        Ok(out)
    }

    /// **Is this an enterprise account?** — derived, never set: an account
    /// whose admin has enrolled at least one employee.
    pub fn is_enterprise(&self, account: &str) -> Result<bool> {
        // Stops at the first binding (asked on every `Access.WhoAmI`), cloning nothing.
        self.doc
            .load()?
            .employers
            .any(|_, employer| employer == account)
    }

    /// One principal, if it exists.
    pub fn principal(&self, id: &str) -> Result<Option<Principal>> {
        self.doc.load()?.principals.get(id)
    }

    /// Every principal. The listing path, **O(principals)**.
    pub fn principals(&self) -> Result<Vec<Principal>> {
        self.doc.load()?.principals.values()
    }

    // ── keys ───────────────────────────────────────────────────────────────

    /// Register an `authorized_keys`-style public key line to a principal.
    ///
    /// Refuses an unknown principal, and refuses a key that is already
    /// registered. [`Error::KeyExists`] says why re-pointing it would be worse
    /// than failing.
    /// Register an **X.509 certificate** to a principal.
    ///
    /// The mTLS twin of [`register_key`](Self::register_key), and deliberately
    /// the same record: a `Fingerprint` is all gunnar knows about a credential,
    /// so a certificate and an SSH key differ in how they are *presented*, not
    /// in what is stored or in what any grant check afterwards does (the mTLS plan).
    ///
    /// `der` is the certificate itself. `algo` is recorded as `x509` so an
    /// operator listing keys can tell the two apart — which matters precisely
    /// because [`fingerprint_of_certificate`] and [`fingerprint_of_blob`] share
    /// one hash space, so the fingerprint alone cannot say which kind it is.
    ///
    /// **Not parsed.** gunnar does not validate the certificate here and must
    /// not: validity, chain and revocation are the TLS handshake's job, at
    /// connection time, against the configured CA. Checking a subset of that
    /// here would produce a second, weaker opinion about whether a credential is
    /// good — and the weaker opinion is the one an attacker picks.
    pub fn register_certificate(
        &self,
        principal_id: &str,
        der: &[u8],
        expires: Option<UnixMillis>,
    ) -> Result<PublicKey> {
        let fingerprint = fingerprint_of_certificate(der);
        let record = PublicKey {
            fingerprint_sha256: fingerprint.as_str().to_owned(),
            algo: "x509".to_owned(),
            principal_id: principal_id.to_owned(),
            added: now_ms(),
            last_used: None,
            expires,
            // The credential's own bytes, base64 as a PEM body carries them, so
            // an operator can see WHICH certificate was registered rather than
            // only its digest.
            openssh: {
                use base64ct::Encoding as _;
                base64ct::Base64::encode_string(der)
            },
        };
        self.doc.update(|doc| {
            if !doc.principals.contains_key(principal_id)? {
                return Err(Error::NoSuchPrincipal(principal_id.to_owned()));
            }
            // A fingerprint already registered is REFUSED rather than
            // reassigned, exactly as for an SSH key. Two principals sharing one
            // credential is not a convenience, it is an audit trail that cannot
            // say who acted.
            if let Some(existing) = doc.keys.get(fingerprint.as_str())? {
                return Err(Error::KeyExists {
                    fingerprint: fingerprint.as_str().to_owned(),
                    principal_id: existing.principal_id.clone(),
                });
            }
            doc.keys
                .insert(fingerprint.as_str().to_owned(), record.clone())?;
            Ok(record.clone())
        })
    }

    /// Register an OpenSSH public key line for `principal_id`, deriving the
    /// fingerprint here so no caller can store one that disagrees with the key.
    pub fn register_key(
        &self,
        principal_id: &str,
        openssh: &str,
        expires: Option<UnixMillis>,
    ) -> Result<PublicKey> {
        // A quantum-safe encryption key (crypt-pq) is kept beside the SSH
        // keys, in the same table under its own `algo`; see
        // [`PQ_RECIPIENT_ALGO`].
        let (fingerprint, algo, line) = match pq_recipient_fingerprint(openssh) {
            Some(fingerprint) => (
                fingerprint,
                PQ_RECIPIENT_ALGO.to_owned(),
                openssh.trim().to_owned(),
            ),
            None => {
                let parsed = parse_public_key(openssh)?;
                (
                    Fingerprint::of_key(&parsed),
                    parsed.algorithm().as_str().to_owned(),
                    parsed.to_openssh().map_err(Error::from)?.trim().to_owned(),
                )
            }
        };
        let record = PublicKey {
            fingerprint_sha256: fingerprint.as_str().to_owned(),
            algo,
            principal_id: principal_id.to_owned(),
            added: now_ms(),
            last_used: None,
            expires,
            openssh: line,
        };
        self.doc.update(|doc| {
            if !doc.principals.contains_key(principal_id)? {
                return Err(Error::NoSuchPrincipal(principal_id.to_owned()));
            }
            if let Some(existing) = doc.keys.get(fingerprint.as_str())? {
                return Err(Error::KeyExists {
                    fingerprint: fingerprint.as_str().to_owned(),
                    principal_id: existing.principal_id.clone(),
                });
            }
            doc.keys
                .insert(fingerprint.as_str().to_owned(), record.clone())?;
            Ok(record.clone())
        })
    }

    /// **Register a web credential (a passkey) on a person** (Stage L).
    ///
    /// Refuses an unknown principal, a principal that is not a USER (no human
    /// credential on a machine principal) and a credential id already
    /// registered — to anybody. The key's shape is the caller's to have
    /// checked; this stores what it is given.
    pub fn add_web_credential(
        &self,
        credential: crate::model::WebCredential,
    ) -> Result<crate::model::WebCredential> {
        self.doc.update(|doc| {
            let principal = doc
                .principals
                .get(&credential.principal_id)?
                .ok_or_else(|| Error::NoSuchPrincipal(credential.principal_id.clone()))?;
            if principal.kind != crate::model::PrincipalKind::User {
                return Err(Error::NotAPerson {
                    principal_id: principal.id.clone(),
                    kind: format!("{:?}", principal.kind).to_lowercase(),
                });
            }
            if let Some(existing) = doc.credentials.get(&credential.id)? {
                return Err(Error::CredentialExists {
                    id: credential.id.clone(),
                    principal_id: existing.principal_id,
                });
            }
            doc.credentials
                .insert(credential.id.clone(), credential.clone())?;
            Ok(credential.clone())
        })
    }

    /// **A passkey was used** (the identity core's login and step-up,
    /// IDENTITY-HOST 2026-10-27): its counter (when it counts) and last use, in
    /// one update; a counter that went back is refused, as a web login's is.
    pub fn touch_web_credential(
        &self,
        id: &str,
        presented: u32,
        now: UnixMillis,
    ) -> Result<crate::model::WebCredential> {
        self.doc.update(|doc| {
            let mut credential = doc
                .credentials
                .get(id)?
                .ok_or_else(|| Error::NoSuchCredential(id.to_owned()))?;
            if credential.sign_count != 0 && presented != 0 && presented <= credential.sign_count {
                return Err(Error::CounterRollback {
                    id: id.to_owned(),
                    stored: credential.sign_count,
                    presented,
                });
            }
            if presented != 0 {
                credential.sign_count = presented;
            }
            credential.last_used = Some(now);
            doc.credentials.insert(id.to_owned(), credential.clone())?;
            Ok(credential)
        })
    }

    /// One web credential by its id (base64url), or `None`.
    pub fn web_credential(&self, id: &str) -> Result<Option<crate::model::WebCredential>> {
        self.doc.load()?.credentials.get(id)
    }

    /// One principal's web credentials, or every one. A listing path.
    pub fn web_credentials(
        &self,
        principal_id: Option<&str>,
    ) -> Result<Vec<crate::model::WebCredential>> {
        let mut rows: Vec<crate::model::WebCredential> = self
            .doc
            .load()?
            .credentials
            .values()?
            .into_iter()
            .filter(|c| principal_id.is_none_or(|want| c.principal_id == want))
            .collect();
        rows.sort_by(|a, b| {
            (a.principal_id.as_str(), a.created, a.id.as_str()).cmp(&(
                b.principal_id.as_str(),
                b.created,
                b.id.as_str(),
            ))
        });
        Ok(rows)
    }

    /// Revoke a web credential **and the session keys its logins enrolled**,
    /// in one update — the person's console sessions on that passkey end with
    /// it. Returns the row removed, if there was one.
    pub fn revoke_web_credential(&self, id: &str) -> Result<Option<crate::model::WebCredential>> {
        let removed = self.doc.update(|doc| {
            let removed = doc.credentials.remove(id)?;
            if let Some(credential) = &removed {
                for fingerprint in &credential.sessions {
                    if doc
                        .keys
                        .get(fingerprint)?
                        .is_some_and(|key| key.principal_id == credential.principal_id)
                    {
                        doc.keys.remove(fingerprint)?;
                    }
                }
            }
            Ok(removed)
        })?;
        if let Some(removed) = &removed {
            tracing::info!(credential = %id, principal = %removed.principal_id, "web credential revoked");
        }
        Ok(removed)
    }

    /// **Commit an approved machine enrolment** (2026-10-05), in ONE update:
    /// the approver's step-up passkey's counter (re-checked: it
    /// must increase when both it and the stored one are non-zero) and last
    /// use; the MACHINE principal `machine` made when absent (a principal of
    /// that id that is not a machine is [`Error::NotAMachine`]); its
    /// server-wide grant set to `level`; and `credential` stored on it (an id
    /// already registered is [`Error::CredentialExists`]). The ONE way a web
    /// credential lands on a machine — [`AuthTable::add_web_credential`] and
    /// [`AuthTable::add_web_credential`] still refuse one. Returns the
    /// machine, the stored credential, and whether this update made the
    /// machine.
    pub fn approve_machine_credential(
        &self,
        machine: &str,
        level: Level,
        mut credential: crate::model::WebCredential,
        step_up: Option<(&str, u32)>,
        now: UnixMillis,
    ) -> Result<(Principal, crate::model::WebCredential, bool)> {
        crate::sandbox::check_component(machine, machine)?;
        if machine == crate::model::ANONYMOUS_PRINCIPAL_ID {
            return Err(Error::ReservedPrincipalId(machine.to_owned()));
        }
        self.doc.update(|doc| {
            if let Some(existing) = doc.credentials.get(&credential.id)? {
                return Err(Error::CredentialExists {
                    id: credential.id.clone(),
                    principal_id: existing.principal_id,
                });
            }
            let approver = match step_up {
                Some((id, presented)) => {
                    let mut held = doc
                        .credentials
                        .get(id)?
                        .ok_or_else(|| Error::NoSuchCredential(id.to_owned()))?;
                    if held.sign_count != 0 && presented != 0 && presented <= held.sign_count {
                        return Err(Error::CounterRollback {
                            id: id.to_owned(),
                            stored: held.sign_count,
                            presented,
                        });
                    }
                    if presented != 0 {
                        held.sign_count = presented;
                    }
                    held.last_used = Some(now);
                    Some((id.to_owned(), held))
                }
                None => None,
            };
            let (principal, created) = match doc.principals.get(machine)? {
                Some(held) if held.kind != crate::model::PrincipalKind::Machine => {
                    return Err(Error::NotAMachine {
                        principal_id: held.id.clone(),
                        kind: format!("{:?}", held.kind).to_lowercase(),
                    })
                }
                Some(held) => (held, false),
                None => {
                    Self::refuse_taken_name(doc, machine, machine)?;
                    let made = Principal {
                        id: machine.to_owned(),
                        name: machine.to_owned(),
                        kind: crate::model::PrincipalKind::Machine,
                        created: now,
                    };
                    doc.principals.insert(made.id.clone(), made.clone())?;
                    Self::index_name(doc, &made)?;
                    (made, true)
                }
            };
            if let Some((id, held)) = approver {
                doc.credentials.insert(id, held)?;
            }
            let grant = Grant::server(machine.to_owned(), level);
            let mut grants = doc.grants.get(machine)?.unwrap_or_default();
            grants.retain(|old| old.scope != grant.scope);
            grants.push(grant);
            grants.sort();
            doc.grants.insert(machine.to_owned(), grants)?;
            credential.principal_id = machine.to_owned();
            doc.credentials
                .insert(credential.id.clone(), credential.clone())?;
            Ok((principal, credential.clone(), created))
        })
    }

    /// Change a web credential's label. Returns the updated row, if there was
    /// one.
    pub fn rename_web_credential(
        &self,
        id: &str,
        label: &str,
    ) -> Result<Option<crate::model::WebCredential>> {
        self.doc.update(|doc| {
            let Some(mut credential) = doc.credentials.get(id)? else {
                return Ok(None);
            };
            credential.label = label.to_owned();
            doc.credentials.insert(id.to_owned(), credential.clone())?;
            Ok(Some(credential))
        })
    }

    /// The person an invite is for, checked: exists, and is a USER.
    fn invitable(doc: &AuthDoc, principal_id: &str) -> Result<Principal> {
        let principal = doc
            .principals
            .get(principal_id)?
            .ok_or_else(|| Error::NoSuchPrincipal(principal_id.to_owned()))?;
        if principal.kind != crate::model::PrincipalKind::User {
            return Err(Error::NotAPerson {
                principal_id: principal.id.clone(),
                kind: format!("{:?}", principal.kind).to_lowercase(),
            });
        }
        Ok(principal)
    }

    /// **Record the operator's first invite** (the installer's token by its
    /// hash; IDENTITY-HOST: the identity core reads it as a fed invite onto
    /// the operator). The person must be a USER; one per box, recorded again
    /// it is replaced.
    pub fn put_first_invite(&self, invite: crate::model::FirstInviteRow) -> Result<()> {
        self.doc.update(|doc| {
            Self::invitable(doc, &invite.principal_id)?;
            doc.invites.insert(invite.id.clone(), invite.clone())?;
            Ok(())
        })
    }

    /// The operator's first invite, if one is recorded (`id`), expired or not.
    pub fn first_invite(&self, id: &str) -> Result<Option<crate::model::FirstInviteRow>> {
        self.doc.load()?.invites.get(id)
    }

    /// Drop the operator's first invite (spent, withdrawn, or its person
    /// holds a passkey). Returns the row removed, if there was one.
    pub fn drop_first_invite(&self, id: &str) -> Result<Option<crate::model::FirstInviteRow>> {
        self.doc.update(|doc| doc.invites.remove(id))
    }

    /// Revoke a key. Returns the record that was removed, if there was one.
    pub fn revoke_key(&self, fingerprint: &Fingerprint) -> Result<Option<PublicKey>> {
        let removed = self
            .doc
            .update(|doc| doc.keys.remove(fingerprint.as_str()))?;
        if removed.is_some() {
            tracing::info!(%fingerprint, "public key revoked");
        }
        Ok(removed)
    }

    /// **Revoke every certificate registered to `principal_id` that certifies a
    /// key OTHER than `spki`** (a DER SubjectPublicKeyInfo), in one update.
    /// Returns the rows removed. SSH keys and certificates for `spki` itself (a
    /// renewal: the same key, a new validity window) are kept; a certificate row
    /// whose bytes do not parse certifies no key anyone can prove, and goes.
    ///
    /// What re-certifying a SINGLE-HOLDER principal means (the twin's
    /// `gunnar-twin`, twin reborn 2026-09-30): a new key is a new box, and the
    /// box it replaces — nuked, its disk possibly copied — must not keep a
    /// registered credential beside it.
    pub fn supersede_certificates(
        &self,
        principal_id: &str,
        spki: &[u8],
    ) -> Result<Vec<PublicKey>> {
        use base64ct::Encoding as _;
        self.doc.update(|doc| {
            let mut gone = Vec::new();
            for row in doc.keys.values()? {
                if row.principal_id != principal_id || row.algo != "x509" {
                    continue;
                }
                let same_key = base64ct::Base64::decode_vec(&row.openssh)
                    .ok()
                    .and_then(|der| crate::issuer::public_key_info_of_der(&der).ok())
                    .is_some_and(|key| key == spki);
                if !same_key {
                    if let Some(removed) = doc.keys.remove(&row.fingerprint_sha256)? {
                        gone.push(removed);
                    }
                }
            }
            Ok(gone)
        })
    }

    /// Every registered key, **O(keys)**. Also a listing path.
    ///
    /// **Includes expired keys**, deliberately: `gunnar key list` must show a key
    /// that has run out, or nobody can see why authentication stopped working.
    /// Anything making a *decision* wants [`AuthTable::signing_keys`] instead.
    pub fn keys(&self) -> Result<Vec<PublicKey>> {
        self.doc.load()?.keys.values()
    }

    /// Every key that may still **satisfy a signature requirement**, as of now.
    ///
    /// # Why this exists as its own method
    ///
    /// Found 2026-08-05: `gunnar-cli`'s `key_directory` — the whole of the
    /// ssh-sig gate's key material (the pre-receive gate) — built itself from
    /// [`AuthTable::keys`] with **no expiry check at all**. So a key past its
    /// `expires` could no longer open an SSH session, and could still sign a
    /// commit that `require_signature` accepted. That is one credential with two
    /// lifetimes, and the longer one was the one nobody had written down.
    ///
    /// A *revoked* key is a removed row and so cannot reach either path, which is
    /// exactly why the bug hid: the revocation case, which is the one everybody
    /// tests, was already correct.
    ///
    /// The fix is a named method rather than a `filter` at the call site because
    /// the call site is what got it wrong. "Every key" and "every key that may
    /// decide something" are different questions, and until now the second one
    /// had no way to be asked. [`AuthTable::principal_by_fingerprint_at`] answers
    /// the same question for authentication, through the same
    /// [`PublicKey::is_valid_at`] — one predicate, two consumers, so the two
    /// lifetimes cannot drift apart again (one implementation).
    pub fn signing_keys(&self) -> Result<Vec<PublicKey>> {
        self.signing_keys_at(now_ms())
    }

    /// [`AuthTable::signing_keys`] at an explicit time, so expiry can be tested
    /// without waiting for it.
    pub fn signing_keys_at(&self, now: UnixMillis) -> Result<Vec<PublicKey>> {
        Ok(self
            .doc
            .load()?
            .keys
            .values()?
            .into_iter()
            .filter(|key| key.is_valid_at(now))
            .collect())
    }

    /// **The keys one principal could attest an addition with**: its keys that
    /// are valid now and are OpenSSH key material rather than a certificate,
    /// since only those can make an `sshsig`. What
    /// [`crate::key_attestation::Requirement::of`] is handed.
    ///
    /// **O(keys on the server)**, for the reason [`AuthTable::keys_of`] is: it
    /// sits on `Access.AddKey`, which happens when a person gets a machine, not
    /// on a connection.
    pub fn attesting_keys_of(&self, principal_id: &str) -> Result<Vec<AttestingKey>> {
        let now = now_ms();
        Ok(self
            .keys_of(principal_id)?
            .into_iter()
            .filter(|key| {
                key.is_valid_at(now) && key.algo != "x509" && key.algo != PQ_RECIPIENT_ALGO
            })
            .map(|key| AttestingKey {
                fingerprint: key.fingerprint_sha256,
                openssh: key.openssh,
            })
            .collect())
    }

    /// Every key registered to one principal. `gunnar-policy` needs this to
    /// verify an ssh-sig signature (the pre-receive gate).
    ///
    /// **O(keys on the server)**, for the reason
    /// [`AuthTable::remove_principal`] gives. It sits on the
    /// signature-verification path of a push, not on a connection.
    pub fn keys_of(&self, principal_id: &str) -> Result<Vec<PublicKey>> {
        Ok(self
            .doc
            .load()?
            .keys
            .values()?
            .into_iter()
            .filter(|key| key.principal_id == principal_id)
            .collect())
    }

    /// **The keys a key ring may be wrapped to** (`Repos.RecipientKeys`, the
    /// forge's `gunnar-recipient-keys` door): the principal's keys valid NOW —
    /// an expired key is one its holder retired, and a ring wrapped to it
    /// opens for whoever holds that private half (wiring tracer B 2026-10-08).
    /// A certificate is never a recipient.
    pub fn recipient_keys_of(&self, principal_id: &str) -> Result<Vec<PublicKey>> {
        let now = now_ms();
        Ok(self
            .keys_of(principal_id)?
            .into_iter()
            .filter(|key| key.is_valid_at(now) && key.algo != "x509")
            .collect())
    }

    /// **The SSH auth answer**: who owns this key, as of now?
    ///
    /// `None` for an unregistered key, for a key whose principal has been
    /// removed, and for an expired key.
    pub fn principal_by_fingerprint(&self, fingerprint: &Fingerprint) -> Result<Option<Principal>> {
        self.principal_by_fingerprint_at(fingerprint, now_ms())
    }

    /// [`AuthTable::principal_by_fingerprint`] at an explicit time, so expiry
    /// can be tested without waiting for it.
    pub fn principal_by_fingerprint_at(
        &self,
        fingerprint: &Fingerprint,
        now: UnixMillis,
    ) -> Result<Option<Principal>> {
        let doc = self.doc.load()?;
        let Some(key) = doc.keys.get(fingerprint.as_str())? else {
            return Ok(None);
        };
        if !key.is_valid_at(now) {
            tracing::warn!(%fingerprint, "expired key offered");
            return Ok(None);
        }
        doc.principals.get(&key.principal_id)
    }

    /// Record that a key just authenticated.
    ///
    /// Separate from the lookup because it writes, and the lookup is on the
    /// connection path: a listener that cannot write must still be able to
    /// authenticate.
    ///
    /// # DELIBERATELY UNCALLED. Do not "fix" this by adding the call
    ///
    /// Nothing in `crates/*/src` calls this, so [`PublicKey::last_used`] is
    /// `None` for every key on the server and `Refs`/`Identity` hand a consumer
    /// a field that is never written. That is a real gap and it is filed. It is
    /// **not** closed by wiring this into `auth_publickey`.
    ///
    /// The cost is no longer the objection it was: sharding split the identity
    /// table, so this rewrites one leaf plus two trie nodes plus the document —
    /// single-digit KB below ~30 000 keys — and not the several hundred
    /// megabytes the finding was originally written against.
    ///
    /// What forbids it is the **shape**, not the size. [`DocRef::update`] is a
    /// compare-and-swap on **one** reference, `refs/gunnar/auth`, with
    /// `MAX_ATTEMPTS = 8`. Calling this per successful authentication puts every
    /// login on the server in contention for that single ref, and the failure
    /// mode is not slowness: it is an `Err` after eight lost races, **on the
    /// authentication path**. A key that authenticated correctly would be
    /// refused because a bookkeeping write lost. `last_used` reading `None` is
    /// strictly better than that.
    ///
    /// Either of two things closes it, and both are more than a call site: an
    /// append-only identity log, where "key K authenticated at T" is one
    /// appended row and needs no CAS; or a low-durability sidecar written
    /// best-effort off the connection path and allowed to lose updates. **No
    /// sidecar exists in the tree today**, which is why this is still here.
    pub fn touch_key(&self, fingerprint: &Fingerprint, now: UnixMillis) -> Result<()> {
        self.doc.update(|doc| {
            doc.keys
                .modify(fingerprint.as_str(), |key| key.last_used = Some(now))?;
            Ok(())
        })
    }

    // ── access tokens ──────────────────────────────────────────────────────

    /// Register an access token for `principal_id`, storing only its hash.
    ///
    /// **The secret is minted by the caller, not here**, exactly as
    /// [`AuthTable::register_key`] is handed an `authorized_keys` line rather
    /// than generating a key pair. That keeps a random number generator out of
    /// this crate, and it keeps the one moment the secret exists in a place the
    /// operator can see: whoever calls this is the only code that ever holds
    /// it, and after this returns nothing in gunnar can produce it again.
    ///
    /// Refuses an unknown principal, refuses
    /// [`ANONYMOUS_PRINCIPAL_ID`](crate::model::ANONYMOUS_PRINCIPAL_ID) (which
    /// cannot be in the table anyway, so this is the belt to
    /// [`AuthTable::add_principal`]'s braces), and refuses a secret that is
    /// already registered.
    pub fn register_token(
        &self,
        principal_id: &str,
        secret: &str,
        label: &str,
        expires: Option<UnixMillis>,
    ) -> Result<AccessToken> {
        if principal_id == crate::model::ANONYMOUS_PRINCIPAL_ID {
            return Err(Error::ReservedPrincipalId(principal_id.to_owned()));
        }
        let hash = token_hash(secret);
        let record = AccessToken {
            hash: hash.clone(),
            principal_id: principal_id.to_owned(),
            label: label.to_owned(),
            created: now_ms(),
            last_used: None,
            expires,
        };
        self.doc.update(|doc| {
            if !doc.principals.contains_key(principal_id)? {
                return Err(Error::NoSuchPrincipal(principal_id.to_owned()));
            }
            if let Some(existing) = doc.tokens.get(&hash)? {
                return Err(Error::TokenExists {
                    hash: hash.clone(),
                    principal_id: existing.principal_id.clone(),
                });
            }
            doc.tokens.insert(hash.clone(), record.clone())?;
            Ok(record.clone())
        })
    }

    /// Revoke a token by its **hash**, which is what listing prints.
    pub fn revoke_token(&self, hash: &str) -> Result<Option<AccessToken>> {
        let removed = self.doc.update(|doc| doc.tokens.remove(hash))?;
        if removed.is_some() {
            tracing::info!(token = %hash, "access token revoked");
        }
        Ok(removed)
    }

    /// Every registered token, **O(tokens)**. A listing path, never an auth
    /// path.
    pub fn tokens(&self) -> Result<Vec<AccessToken>> {
        self.doc.load()?.tokens.values()
    }

    /// **The HTTP auth answer**: who holds this secret, as of now?
    ///
    /// `None` for an unregistered secret, for a token whose principal has been
    /// removed, and for an expired token — the same three cases, in the same
    /// order, as [`AuthTable::principal_by_fingerprint`], because they are the
    /// same question asked with a different credential.
    pub fn principal_by_token(&self, secret: &str) -> Result<Option<Principal>> {
        self.principal_by_token_at(secret, now_ms())
    }

    /// [`AuthTable::principal_by_token`] at an explicit time, so expiry can be
    /// tested without waiting for it.
    pub fn principal_by_token_at(
        &self,
        secret: &str,
        now: UnixMillis,
    ) -> Result<Option<Principal>> {
        let doc = self.doc.load()?;
        let Some(token) = doc.tokens.get(&token_hash(secret))? else {
            return Ok(None);
        };
        if !token.is_valid_at(now) {
            // The hash, never the secret: this line goes to a log.
            tracing::warn!(token = %token.hash, "expired access token presented");
            return Ok(None);
        }
        doc.principals.get(&token.principal_id)
    }

    /// Record that a token just authenticated — **at most once per
    /// [`LAST_USED_GRANULARITY`] per token**.
    ///
    /// Separate from the lookup for the same reason [`AuthTable::touch_key`]
    /// is: the lookup is on the request path and must work on a table that
    /// cannot be written.
    ///
    /// The granularity is the fix for the serving tier's measured ceiling.
    /// `last_used` lives in ONE control document shared by every principal and
    /// every token, so writing it per request made every authenticated HTTP
    /// request in the process contend on one compare-and-swap — MEASURED on
    /// 2026-08-11 at 573 → 1416 req/s (16-way ls-refs, 4-CPU cpuset) with
    /// this call the only difference, and again on 2026-08-12 by
    /// `gunnar.multiuser_scaling`: 102 users plateaued at 82 ops/s with the
    /// server stuck below 2 of 32 cores, 8 630 lost CAS attempts and 862
    /// silently dropped updates in one 12-second window. A credential's
    /// "when was this last seen" is a minute-grained fact; a fresh
    /// `last_used` answers the question and the skipped write is the
    /// throughput. The first touch of a quiet token still writes immediately.
    pub fn touch_token(&self, hash: &str, now: UnixMillis) -> Result<()> {
        // The read is the same cached document load every request's auth
        // lookup already does; only a token whose record is stale (or never
        // written) proceeds to the shared-document CAS.
        if let Some(token) = self.doc.load()?.tokens.get(hash)? {
            if let Some(last) = token.last_used {
                if now.saturating_sub(last) < LAST_USED_GRANULARITY {
                    return Ok(());
                }
            }
        }
        self.doc.update(|doc| {
            doc.tokens.modify(hash, |token| {
                // Re-checked under the CAS: of the requests that race into
                // the same window boundary, one write wins and the rest
                // rewrite to the same bytes, which `DocRef::update` commits
                // as "unchanged" without moving the pointer.
                if token
                    .last_used
                    .is_none_or(|last| now.saturating_sub(last) >= LAST_USED_GRANULARITY)
                {
                    token.last_used = Some(now);
                }
            })?;
            Ok(())
        })
    }

    // ── bulk import ────────────────────────────────────────────────────────

    /// Import many users — principal, optional access token, grants — in **one
    /// document transaction**.
    ///
    /// The per-entity verbs commit one CAS each, and each CAS clones the
    /// document and rewrites its skeleton: at an appliance-scale import that is
    /// O(users²) work aimed at one shared pointer. This is the batch door the
    /// 1→100 000-user story needs: every insert lands in the in-memory
    /// document once, dirty shards are flushed once, and the whole population
    /// becomes durable in a single compare-and-swap — so the import is also
    /// **atomic**: a refused user refuses the batch, and a batch that lost the
    /// CAS race retries as one unit.
    ///
    /// Token secrets never enter this crate: callers hand over
    /// [`token_hash`]es, exactly as [`AuthTable::register_token`] stores and
    /// [`AuthTable::principal_by_token`] looks up.
    pub fn bulk_add_users(&self, users: &[BulkUser], now: UnixMillis) -> Result<usize> {
        for user in users {
            crate::sandbox::check_component(&user.principal.id, &user.principal.id)?;
            if user.principal.id == crate::model::ANONYMOUS_PRINCIPAL_ID {
                return Err(Error::ReservedPrincipalId(user.principal.id.clone()));
            }
        }
        // Keys parsed OUTSIDE the transaction: parsing 100k OpenSSH lines is
        // real work, and a CAS retry must not repeat it.
        let mut keys: Vec<Option<PublicKey>> = Vec::with_capacity(users.len());
        for user in users {
            keys.push(match &user.openssh_public_key {
                None => None,
                Some(line) => {
                    let parsed = parse_public_key(line)?;
                    let fingerprint = Fingerprint::of_key(&parsed);
                    Some(PublicKey {
                        fingerprint_sha256: fingerprint.as_str().to_owned(),
                        algo: parsed.algorithm().as_str().to_owned(),
                        principal_id: user.principal.id.clone(),
                        added: now,
                        last_used: None,
                        expires: None,
                        openssh: parsed.to_openssh().map_err(Error::from)?.trim().to_owned(),
                    })
                }
            });
        }
        self.doc.update(|doc| {
            for user in users {
                let id = &user.principal.id;
                if doc.principals.contains_key(id)? {
                    return Err(Error::PrincipalExists(id.clone()));
                }
                Self::refuse_taken_name(doc, &user.principal.name, id)?;
                let principal = Principal {
                    created: now,
                    ..user.principal.clone()
                };
                doc.principals.insert(id.clone(), principal.clone())?;
                Self::index_name(doc, &principal)?;
                if let Some(hash) = &user.token_hash {
                    if let Some(existing) = doc.tokens.get(hash)? {
                        return Err(Error::TokenExists {
                            hash: hash.clone(),
                            principal_id: existing.principal_id.clone(),
                        });
                    }
                    doc.tokens.insert(
                        hash.clone(),
                        AccessToken {
                            hash: hash.clone(),
                            principal_id: id.clone(),
                            label: user.token_label.clone(),
                            created: now,
                            last_used: None,
                            expires: None,
                        },
                    )?;
                }
                if !user.grants.is_empty() {
                    // The principal is new in this same transaction, so there
                    // is nothing to merge with; sorted for the same reason
                    // `grant` sorts.
                    let mut held = user.grants.clone();
                    held.sort();
                    doc.grants.insert(id.clone(), held)?;
                }
            }
            for record in keys.iter().flatten() {
                if let Some(existing) = doc.keys.get(&record.fingerprint_sha256)? {
                    return Err(Error::KeyExists {
                        fingerprint: record.fingerprint_sha256.clone(),
                        principal_id: existing.principal_id.clone(),
                    });
                }
                doc.keys
                    .insert(record.fingerprint_sha256.clone(), record.clone())?;
            }
            Ok(users.len())
        })
    }

    // ── grants ─────────────────────────────────────────────────────────────

    /// Add a grant, replacing any existing grant this principal has on the same
    /// scope.
    ///
    /// Replacing rather than accumulating means a downgrade is a downgrade: if
    /// grants piled up, "set them to Read" would leave the old Write in place
    /// and the ACL would take the maximum.
    pub fn grant(&self, grant: Grant) -> Result<Grant> {
        self.doc.update(|doc| {
            if !doc.principals.contains_key(&grant.principal)? {
                return Err(Error::NoSuchPrincipal(grant.principal.clone()));
            }
            let mut held = doc.grants.get(&grant.principal)?.unwrap_or_default();
            held.retain(|old| old.scope != grant.scope);
            held.push(grant.clone());
            held.sort();
            doc.grants.insert(grant.principal.clone(), held)?;
            Ok(grant.clone())
        })
    }

    /// Remove a principal's grant on one scope. `true` if there was one.
    pub fn revoke_grant(&self, principal_id: &str, scope: &Scope) -> Result<bool> {
        self.doc.update(|doc| {
            let Some(mut held) = doc.grants.get(principal_id)? else {
                return Ok(false);
            };
            let before = held.len();
            held.retain(|old| &old.scope != scope);
            if held.len() == before {
                return Ok(false);
            }
            if held.is_empty() {
                doc.grants.remove(principal_id)?;
            } else {
                doc.grants.insert(principal_id.to_owned(), held)?;
            }
            Ok(true)
        })
    }

    /// Every grant, or every grant held by one principal.
    ///
    /// Naming a principal is a shard probe; asking for all of them walks the
    /// whole ACL, which is what a listing is.
    pub fn grants(&self, principal_id: Option<&str>) -> Result<Vec<Grant>> {
        let doc = self.doc.load()?;
        match principal_id {
            Some(id) => Ok(doc.grants.get(id)?.unwrap_or_default()),
            None => Ok(doc.grants.values()?.into_iter().flatten().collect()),
        }
    }

    // ── groups (punkt 9) ───────────────────────────────────────────────────

    /// **Make `member_id` a member of the group `group_id`.**
    ///
    /// The group must exist and be a
    /// [`PrincipalKind::Group`](crate::PrincipalKind::Group); the member must
    /// exist, and may be a group itself (nesting). Idempotent: adding a member
    /// already in the group is a no-op, so a config that lists a membership
    /// twice does not create a duplicate edge.
    ///
    /// This is admin-gated at the RPC exactly like [`AuthTable::grant`], and for
    /// the same reason: adding a member to a group that holds a grant *is*
    /// giving that member the grant. The gate is the caller's; here we only
    /// refuse memberships that do not typecheck.
    pub fn add_to_group(&self, group_id: &str, member_id: &str) -> Result<()> {
        self.doc.update(|doc| {
            let group = doc
                .principals
                .get(group_id)?
                .ok_or_else(|| Error::NoSuchPrincipal(group_id.to_owned()))?;
            if group.kind != PrincipalKind::Group {
                return Err(Error::NotAGroup(group_id.to_owned()));
            }
            if !doc.principals.contains_key(member_id)? {
                return Err(Error::NoSuchPrincipal(member_id.to_owned()));
            }
            let mut held = doc.memberships.get(member_id)?.unwrap_or_default();
            if !held.iter().any(|g| g == group_id) {
                held.push(group_id.to_owned());
                held.sort();
                doc.memberships.insert(member_id.to_owned(), held)?;
            }
            Ok(())
        })
    }

    /// **Remove `member_id` from the group `group_id`.** `true` if it was a
    /// member.
    ///
    /// The access the group conferred is gone the moment this returns: the next
    /// [`AuthTable::may`] resolves the member's groups afresh from the document
    /// this write updated, so there is no cached membership and no window. The
    /// tests clone through SSH before and after to prove the applied effect, not
    /// the row.
    pub fn remove_from_group(&self, group_id: &str, member_id: &str) -> Result<bool> {
        self.doc.update(|doc| {
            let Some(mut held) = doc.memberships.get(member_id)? else {
                return Ok(false);
            };
            let before = held.len();
            held.retain(|g| g != group_id);
            if held.len() == before {
                return Ok(false);
            }
            if held.is_empty() {
                doc.memberships.remove(member_id)?;
            } else {
                doc.memberships.insert(member_id.to_owned(), held)?;
            }
            Ok(true)
        })
    }

    /// The groups `member_id` belongs to **directly** (one shard probe).
    pub fn groups_of(&self, member_id: &str) -> Result<Vec<String>> {
        Ok(self
            .doc
            .load()?
            .memberships
            .get(member_id)?
            .unwrap_or_default())
    }

    /// **Who is directly in this group.** A listing, **O(members on the
    /// server)**: membership is keyed by member for the connection path, so the
    /// inverse question is a walk. Sorted, for a stable listing.
    pub fn members_of(&self, group_id: &str) -> Result<Vec<String>> {
        let doc = self.doc.load()?;
        let mut members: Vec<String> = doc
            .memberships
            .entries()?
            .into_iter()
            .filter(|(_, groups)| groups.iter().any(|g| g == group_id))
            .map(|(member, _)| member)
            .collect();
        members.sort();
        Ok(members)
    }

    /// **Every group on the server** — the principals whose kind is
    /// [`Group`](crate::PrincipalKind::Group). Walks the principals, like the
    /// principal listing does.
    pub fn groups(&self) -> Result<Vec<Principal>> {
        Ok(self
            .doc
            .load()?
            .principals
            .values()?
            .into_iter()
            .filter(|p| p.kind == PrincipalKind::Group)
            .collect())
    }

    /// **Every principal whose grants `principal_id` may exercise**: itself,
    /// then the transitive closure of the groups it belongs to.
    ///
    /// This is the whole of group inheritance. A member is authorized by any
    /// grant its group holds; because a group is itself a principal that can be
    /// a member of another group, it is authorized by any grant a group's group
    /// holds, and so on — a breadth-first walk of the membership edges.
    ///
    /// # Bounded against a cycle, by construction
    ///
    /// Nothing at write time forbids a membership cycle (A ∈ B, B ∈ A), because
    /// forbidding it would need this very closure. `seen` is what makes the walk
    /// terminate anyway: a principal already expanded is never expanded again,
    /// so a cycle is visited once and stops. `group_cycle_terminates` in the
    /// tests drives A ∈ B ∈ A through here and asserts it returns rather than
    /// hangs — an unbounded resolver hangs that test red.
    ///
    /// The querying principal is always first and always present, so a
    /// principal in no group resolves to exactly itself and every `may*` below
    /// behaves precisely as it did before groups existed (one implementation).
    fn grant_principals(doc: &AuthDoc, principal_id: &str) -> Result<Vec<String>> {
        let mut seen = std::collections::BTreeSet::new();
        let mut queue = std::collections::VecDeque::new();
        let mut holders = Vec::new();
        queue.push_back(principal_id.to_owned());
        while let Some(id) = queue.pop_front() {
            if !seen.insert(id.clone()) {
                continue;
            }
            if let Some(groups) = doc.memberships.get(&id)? {
                for group in groups {
                    if !seen.contains(&group) {
                        queue.push_back(group);
                    }
                }
            }
            holders.push(id);
        }
        Ok(holders)
    }

    /// Every grant held by `principal_id` **or by any group it inherits from**.
    ///
    /// The one place group inheritance widens the grant set, so `may`,
    /// `may_namespace` and `may_server` share it and cannot drift: each applies
    /// its own scope filter to the same widened set (one implementation).
    fn holder_grants(doc: &AuthDoc, principal_id: &str) -> Result<Vec<Grant>> {
        let mut all = Vec::new();
        for holder in Self::grant_principals(doc, principal_id)? {
            if let Some(grants) = doc.grants.get(&holder)? {
                all.extend(grants);
            }
        }
        Ok(all)
    }

    // ── the ACL ────────────────────────────────────────────────────────────

    /// **May this principal do this to this repository?**
    ///
    /// The rules, in full:
    ///
    /// 0. The **anonymous** principal — the one an unauthenticated caller
    ///    is served as — may `Read` a
    ///    [`Visibility::Public`](crate::model::Visibility::Public) repository
    ///    and may do nothing else, anywhere. It is not in the table, cannot be
    ///    put in the table ([`Error::ReservedPrincipalId`]) and therefore can
    ///    never hold a grant; this line is the *whole* of what "anonymous" can
    ///    ever be allowed, and it is reached before the table is even read.
    /// 1. An unknown principal may do nothing. Not "read-only" — nothing.
    /// 2. **A principal is unconditionally admin of its own account**: the
    ///    namespace whose name is its own id, which is the `<account>` in
    ///    `<host>/<account>/<repo>`. See [`AuthTable::owns_account`].
    /// 3. A [`Visibility::Public`](crate::model::Visibility::Public)
    ///    repository is readable by any registered principal too, with no
    ///    grant.
    /// 4. Otherwise the highest grant covering the repository — server,
    ///    namespace or repository scope — held by this principal **or by any
    ///    group it is (transitively) a member of** must be at least `level`.
    ///    [`Level`] is ordered, so `Admin` implies `Write` implies `Read`.
    ///
    /// **Groups (punkt 9).** A group is a principal
    /// ([`PrincipalKind::Group`](crate::PrincipalKind::Group)) that holds no
    /// keys; a member inherits the group's grants and nothing else. Inheritance
    /// is of grants only — never of a group's intrinsic account ownership (rule
    /// 2 reads the *querying* principal's own id, so being in a group named
    /// `team` does not make you admin of the `team` namespace). Nesting is
    /// resolved transitively and cycle-safely by
    /// [`AuthTable::grant_principals`]. The maximum over the inherited grants is
    /// taken, so a group granted `Read` can never let a member `Write`: a
    /// read-only group is read-only for everyone in it, by the same ordering
    /// that stops a read grant from becoming a write one for a lone user.
    ///
    /// A namespace grant matches its namespace **exactly**: `team` does not
    /// reach `team/sub`. See [`Scope::Namespace`].
    ///
    /// # Why anonymous is a rule here and not a branch at the listener
    ///
    /// One implementation. The SSH dispatch, the gRPC gate and anything added later ask this
    /// one function; a listener that decided "public, so serve it" for itself
    /// would be a second ACL, and the failure mode of two ACLs is that the
    /// permissive one wins silently. `may` is the only place a `Visibility` is
    /// ever read for an access decision.
    ///
    /// **A server-scope grant stops at a customer's account** (2026-10-05,
    /// the operator sees aggregates only): inside a namespace a PERSON owns,
    /// a PERSON's server grant counts for nothing unless the account's admin
    /// approved a look, and then up to `Read` ([`operator_access`]). A
    /// machine's keeps its reach: it is the estate's plumbing.
    pub fn may(&self, principal_id: &str, repo: &Repo, level: Level) -> Result<bool> {
        self.may_at(principal_id, repo, level, now_ms())
    }

    /// [`AuthTable::may`] at a given instant (an operator's look ends on time).
    pub fn may_at(&self, principal_id: &str, repo: &Repo, level: Level, now: i64) -> Result<bool> {
        let doc = self.doc.load()?;
        Self::may_in(&doc, principal_id, repo, level, now)
    }

    /// **[`AuthTable::may`] for every repository of a listing, under ONE read
    /// of the identity table** (round 5.3.1, test5 2026-10-08: a listing asked
    /// `may` once per row, and each ask read the table's pointer from the
    /// control store again — two store reads and a handful of shared-cache
    /// locks per row, so concurrent listings queued on each other). The same
    /// rule, row by row: [`AuthTable::may_at`] is this with one row.
    pub fn may_each(&self, principal_id: &str, repos: &[Repo], level: Level) -> Result<Vec<bool>> {
        let doc = self.doc.load()?;
        let now = now_ms();
        repos
            .iter()
            .map(|repo| Self::may_in(&doc, principal_id, repo, level, now))
            .collect()
    }

    fn may_in(
        doc: &AuthDoc,
        principal_id: &str,
        repo: &Repo,
        level: Level,
        now: i64,
    ) -> Result<bool> {
        // **Rule −1: the gate, and it is read before anything else** (see
        // [`crate::gate`]). Both directions: a gated principal acts nowhere,
        // and nobody acts inside a gated account — including the anonymous
        // principal, which is why this sits ABOVE the anonymous rule rather
        // than after it. A stranger cloning a public repository out of an
        // account that is being emptied is exactly the reader that puts bytes
        // back behind a purge.
        if Self::gated(doc, principal_id)? || Self::gated(doc, &repo.namespace)? {
            return Ok(false);
        }
        if principal_id == crate::model::ANONYMOUS_PRINCIPAL_ID {
            return Ok(level == Level::Read && repo.visibility == crate::model::Visibility::Public);
        }
        if !doc.principals.contains_key(principal_id)? {
            return Ok(false);
        }
        // Rule 2½: a CLAN's own (2026-10-05) — its admin administers every
        // repository in the clan's namespace, its members read and write
        // them. Intrinsic like rule 2: no grant row a `Revoke` could take.
        if Self::clan_level(doc, principal_id, &repo.namespace)?.is_some_and(|have| have >= level) {
            return Ok(true);
        }
        // Rule 2: your own account. Intrinsic, so `level` is not consulted —
        // every level is allowed, including `Admin`. Never a MEMBER's: an
        // employee has no account of its own (2026-10-07).
        if Self::owns_account(principal_id, &repo.namespace) && !Self::is_member(doc, principal_id)?
        {
            return Ok(true);
        }
        if level == Level::Read && repo.visibility == crate::model::Visibility::Public {
            return Ok(true);
        }
        // Rule 4, now including inheritance: the highest grant covering the
        // repository held by this principal OR by any group it is (transitively)
        // a member of. `holder_grants` is `[principal] ∪ its group closure`, so
        // a principal in no group searches exactly its own grants as before.
        let covering = Self::holder_grants(doc, principal_id)?
            .into_iter()
            .filter(|grant| grant.scope.covers(repo));
        let best = Self::best_under(covering, || {
            Self::shield_in(doc, principal_id, &repo.namespace, now)
        })?;
        Ok(best.is_some_and(|have| have >= level))
    }

    /// [`AuthTable::may`], as the error the SSH and gRPC layers want to
    /// return.
    pub fn require(&self, principal_id: &str, repo: &Repo, level: Level) -> Result<()> {
        if self.may(principal_id, repo, level)? {
            return Ok(());
        }
        // **A gate is reported AS a gate**, and never as a missing grant. The
        // two mean different things to whoever is on the other end of the SSH
        // session: "ask for a grant" and "your account is locked and a grant
        // will not help". Reported for the principal's own gate first and for
        // the account's second, so a locked-out user is told about *their*
        // lockout rather than about the namespace they were reaching into.
        for account in [principal_id, repo.namespace.as_str()] {
            if let Some(gate) = self.gate_covering(account)? {
                return Err(Error::Gated {
                    account: gate.account,
                    reason: gate.reason.as_str().to_owned(),
                    note: gate.note,
                });
            }
        }
        Err(Error::Denied {
            principal_id: principal_id.to_owned(),
            level,
            repo: repo.path(),
        })
    }

    /// May this principal create and delete repositories in this namespace?
    ///
    /// Never, for the anonymous principal. Refused **by construction**
    /// rather than by the absence of a grant: absence is already an answer
    /// here, but an answer that depends on a table staying empty is a weaker
    /// promise than one that does not look at the table at all.
    pub fn may_namespace(&self, principal_id: &str, namespace: &str, level: Level) -> Result<bool> {
        if principal_id == crate::model::ANONYMOUS_PRINCIPAL_ID {
            return Ok(false);
        }
        let doc = self.doc.load()?;
        // The gate, both directions, exactly as in [`AuthTable::may`]. This is
        // the question `repo create` and `repo delete` ask, so a gate that
        // covered only `may` would let a locked-out user go on making
        // repositories in the account a purge is walking.
        if Self::gated(&doc, principal_id)? || Self::gated(&doc, namespace)? {
            return Ok(false);
        }
        if !doc.principals.contains_key(principal_id)? {
            return Ok(false);
        }
        // A clan's own, as in [`AuthTable::may`]'s rule 2½: its admin at
        // every level, its members up to `Write` (they create repositories
        // there; administering them is the admin's).
        if Self::clan_level(&doc, principal_id, namespace)?.is_some_and(|have| have >= level) {
            return Ok(true);
        }
        // Your own account, at every level. Same rule as [`AuthTable::may`]'s
        // rule 2 and the same one writer, so a repository question and a
        // namespace question can never disagree about who owns an account.
        if Self::owns_account(principal_id, namespace) && !Self::is_member(&doc, principal_id)? {
            return Ok(true);
        }
        let covering = Self::holder_grants(&doc, principal_id)?
            .into_iter()
            .filter(|grant| grant.scope.covers_namespace(namespace));
        let best = Self::best_under(covering, || {
            Self::shield_in(&doc, principal_id, namespace, now_ms())
        })?;
        Ok(best.is_some_and(|have| have >= level))
    }

    /// May this principal administer the server (create namespaces, register
    /// principals)?
    ///
    /// Never, for the anonymous principal, for the reason
    /// [`AuthTable::may_namespace`] gives.
    pub fn may_server(&self, principal_id: &str, level: Level) -> Result<bool> {
        if principal_id == crate::model::ANONYMOUS_PRINCIPAL_ID {
            return Ok(false);
        }
        let doc = self.doc.load()?;
        // Only the actor's own gate here: `Scope::Server` names no account, so
        // there is no second name to check. Without this line a gated principal
        // holding `Server`/`Admin` could lift their own gate through the front
        // door, which would make the whole table advisory.
        if Self::gated(&doc, principal_id)? {
            return Ok(false);
        }
        if !doc.principals.contains_key(principal_id)? {
            return Ok(false);
        }
        let best = Self::holder_grants(&doc, principal_id)?
            .into_iter()
            .filter(|grant| grant.scope == Scope::Server)
            .map(|grant| grant.level)
            .max();
        Ok(best.is_some_and(|have| have >= level))
    }

    /// **Does this principal hold a SERVER-scope grant** — any level, its own or through a
    /// group (2026-10-08)? The operator's mark: a namespace whose owner holds one is the
    /// SELLER's own (`Namespace.owner_server_grant`), never a customer a selling shop must give
    /// a fact. Unlike [`AuthTable::may_server`] the gate is not asked: a gated operator is still
    /// the seller (a fact given in the meantime would stay after the gate lifts). Only an
    /// operator grants `Scope::Server`; nothing a customer does — signing up, an invite, its
    /// own account, an employee's enrolment, a namespace or repository grant — makes one.
    pub fn holds_server_grant(&self, principal_id: &str) -> Result<bool> {
        if principal_id == crate::model::ANONYMOUS_PRINCIPAL_ID {
            return Ok(false);
        }
        let doc = self.doc.load()?;
        if !doc.principals.contains_key(principal_id)? {
            return Ok(false);
        }
        Ok(Self::holder_grants(&doc, principal_id)?
            .iter()
            .any(|grant| grant.scope == Scope::Server))
    }

    // ── accounts (`<host>/<account>/<repo>`) ───────────────────────────────

    /// **Does this principal own this account?**
    ///
    /// An account is a namespace, and a principal owns the namespace whose name
    /// is its own id — `github.com/<konto>/<repo>`, with `<konto>` the user.
    ///
    /// # Why this is a rule and not a grant
    ///
    /// Account admin is **intrinsic**, not granted. A `Grant { scope:
    /// Namespace(id), level: Admin }` row would look identical from the outside
    /// and would be *revocable*: `Access.Revoke` would take a user's own
    /// account away from them, and nothing would put it back, because nothing
    /// would remember that it was supposed to be there. This function reads no
    /// table, so there is no row to remove.
    ///
    /// The collision it depends on is closed at the other end: a principal's
    /// account namespace is created **with** the principal
    /// (`CoreControlPlane::add_principal`), so a namespace that already exists
    /// makes the principal fail to be created and a principal that already
    /// exists makes the namespace fail to be created. Neither name can be
    /// claimed twice, so this rule can never hand one person's namespace to
    /// somebody else.
    ///
    /// Associated rather than free so there is one spelling of the rule and
    /// both [`AuthTable::may`] and [`AuthTable::may_namespace`] use it (one implementation).
    pub fn owns_account(principal_id: &str, namespace: &str) -> bool {
        // Never for the anonymous principal, which holds no account. Both
        // callers already refuse it above; this is the belt, because the whole
        // rule is "a name equals a name" and `""` equalling `""` would be a
        // catastrophic reading of it.
        !principal_id.is_empty()
            && principal_id != crate::model::ANONYMOUS_PRINCIPAL_ID
            && principal_id == namespace
    }

    /// **Is `principal_id` an account's MEMBER** (an employee its admin
    /// enrolled)? A member has no account of its own (2026-10-07, the members
    /// model): rule 2 does not make it the admin of a namespace that happens
    /// to carry its id — the walkthrough's `ben` "owned" a namespace `ben`
    /// nobody had made, was offered it for New repository and Import, and
    /// bought room for it. Asked only when the names are equal, so the hot
    /// path pays one probe for the rare case.
    fn is_member(doc: &AuthDoc, principal_id: &str) -> Result<bool> {
        Ok(doc.employers.get(principal_id)?.is_some())
    }

    // ── the account gate (see `crate::gate`) ───────────────────────────────

    /// **Is `account` locked out right now?** The predicate every `may*` reads.
    ///
    /// Takes the already-loaded document rather than loading its own, because
    /// [`AuthTable::may`] asks it twice per call and a second `load()` would
    /// double the parse on the path the sharding exists to keep cheap — and, worse,
    /// could answer the two halves of one decision from two versions of the
    /// table.
    ///
    /// An **expired** lease is not a gate. That is the whole crash-safety
    /// story: see [`crate::gate`].
    /// **Namespaces nest, so this walks the ancestors.** `alice/team` is
    /// a different namespace from `alice` — [`Scope::Namespace`] matches
    /// exactly and [`AuthTable::owns_account`] is a string equality — but on
    /// the disk it is `repos/alice/team/…`, inside the account's own directory
    /// subtree, and `Accounts.Purge` reclaims it along with the rest. A gate
    /// keyed by exact name would leave open the one part of an account a purge
    /// is walking.
    ///
    /// On PATH COMPONENTS and never on characters: `alicehacks` is a different
    /// account and a gate on `alice` must not close it.
    ///
    /// The cost is one shard probe per component, on a path
    /// [`AuthTable::may`] runs twice per connection. Namespaces are one
    /// component deep in every deployment there has been, so this is one probe;
    /// the walk exists so that the rare nested case is correct rather than
    /// silently exempt.
    fn gated(doc: &AuthDoc, account: &str) -> Result<bool> {
        let now = crate::model::now_ms();
        let mut prefix = String::new();
        for component in account.split('/') {
            if !prefix.is_empty() {
                prefix.push('/');
            }
            prefix.push_str(component);
            if doc
                .gates
                .get(&prefix)?
                .is_some_and(|gate| gate.is_live_at(now))
            {
                return Ok(true);
            }
        }
        Ok(false)
    }

    /// **Lock an account out.** Upserts: the newest writer owns the row.
    ///
    /// Upsert rather than refuse-if-present on purpose. The two writers are a
    /// purge taking a lease and a person suspending an account, and in every
    /// ordering the *later* decision is the one that should stand — a refusal
    /// would mean an operator could not suspend an account a purge is holding,
    /// which is precisely when they most want to. What is protected instead is
    /// the *lift*: see [`AuthTable::lift_gate_held_by`].
    ///
    /// Returns the row as stored. Idempotent in the sense that matters: gating
    /// an already-gated account leaves it gated.
    pub fn gate(&self, gate: crate::gate::AccountGate) -> Result<crate::gate::AccountGate> {
        // A gate is a statement about a name, and the name has to be one this
        // server could ever hold — the same check `add_principal` makes, so a
        // path-traversal string cannot be parked in the table as a gate.
        crate::sandbox::check_component(&gate.account, &gate.account)?;
        if gate.account == crate::model::ANONYMOUS_PRINCIPAL_ID {
            // Gating "anonymous" would look like it closed public reads across
            // the whole server and would in fact do nothing of the sort, since
            // `may` reads the *repository's* namespace for the second check.
            // A refusal by name beats a row that lies.
            return Err(Error::ReservedPrincipalId(gate.account));
        }
        self.doc.update(|doc| {
            doc.gates.insert(gate.account.clone(), gate.clone())?;
            Ok(gate.clone())
        })
    }

    /// **The gate on `account`, if one is in force.**
    ///
    /// `None` covers both "never gated" and "the lease expired": to everything
    /// downstream those are the same fact, and reporting an expired lease as a
    /// lockout would tell an operator an account is shut when it is open.
    pub fn gate_of(&self, account: &str) -> Result<Option<crate::gate::AccountGate>> {
        Ok(self
            .doc
            .load()?
            .gates
            .get(account)?
            .filter(|gate| gate.is_live_at(crate::model::now_ms())))
    }

    /// **The gate that closes `account`**, which may be on an ancestor
    /// namespace rather than on `account` itself — see [`AuthTable::gated`],
    /// whose walk this is the reporting half of.
    ///
    /// [`AuthTable::gate_of`] stays an exact lookup because `Accounts.Ungate`
    /// and the purge's own release name a row, and lifting "whatever covers
    /// this" would let an ungate of `alice/team` silently lift the gate on
    /// `alice`. This one answers the different question a refusal has to
    /// answer: *why can I not do this*.
    pub fn gate_covering(&self, account: &str) -> Result<Option<crate::gate::AccountGate>> {
        let doc = self.doc.load()?;
        let now = crate::model::now_ms();
        let mut prefix = String::new();
        for component in account.split('/') {
            if !prefix.is_empty() {
                prefix.push('/');
            }
            prefix.push_str(component);
            if let Some(gate) = doc.gates.get(&prefix)?.filter(|g| g.is_live_at(now)) {
                return Ok(Some(gate));
            }
        }
        Ok(None)
    }

    /// Every gate in force, by account. The listing path, **O(gates)**.
    pub fn gates(&self) -> Result<Vec<crate::gate::AccountGate>> {
        let now = crate::model::now_ms();
        let mut live: Vec<crate::gate::AccountGate> = self
            .doc
            .load()?
            .gates
            .values()?
            .into_iter()
            .filter(|gate| gate.is_live_at(now))
            .collect();
        live.sort_by(|a, b| a.account.cmp(&b.account));
        Ok(live)
    }

    /// **Lift the gate on `account`, whoever holds it.** Returns what was
    /// there, or `None` if nothing was.
    ///
    /// The operator's override, and the reason a bricked account is always
    /// recoverable: it does not consult the holder, so a lease left behind by
    /// a process that no longer exists can still be cleared by hand.
    pub fn lift_gate(&self, account: &str) -> Result<Option<crate::gate::AccountGate>> {
        self.doc.update(|doc| doc.gates.remove(account))
    }

    /// **Lift the gate on `account` only if `holder` still holds it.**
    ///
    /// What a finishing purge calls. If an operator suspended the account
    /// while the purge was running, the row now names them, this returns
    /// `None`, and the suspension survives — a purge must not be able to
    /// un-suspend an account as a side effect of finishing.
    ///
    /// The read and the remove are inside one [`DocRef::update`], so there is
    /// no window between checking the holder and clearing the row.
    pub fn lift_gate_held_by(
        &self,
        account: &str,
        holder: &str,
    ) -> Result<Option<crate::gate::AccountGate>> {
        self.doc.update(|doc| {
            match doc.gates.get(account)? {
                Some(gate) if gate.holder == holder => doc.gates.remove(account),
                // Somebody else's row, or none at all. Both are "nothing of
                // mine to lift", and neither is an error: a purge that failed
                // its own lease renewal and had it expire finds no row here and
                // must still be able to finish reporting.
                _ => Ok(None),
            }
        })
    }

    /// **Push a lease's deadline out**, if `holder` still holds it.
    ///
    /// The heartbeat a running purge makes after each repository it reclaims.
    /// Returns whether the row was renewed — `false` means the purge no longer
    /// holds the gate (an operator replaced it, or its lease expired and
    /// something else took the account), which the worker logs rather than
    /// treats as a failure: it is still emptying an account that is still shut.
    pub fn renew_gate(&self, account: &str, holder: &str, expires: UnixMillis) -> Result<bool> {
        self.doc.update(|doc| {
            let renew = doc
                .gates
                .get(account)?
                .is_some_and(|gate| gate.holder == holder);
            if renew {
                doc.gates
                    .modify(account, |gate| gate.expires = Some(expires))?;
            }
            Ok(renew)
        })
    }

    // ── bootstrap (the secret-store seam) ──────────────────────────────────

    /// Are there no keys at all? A fresh server, in other words.
    ///
    /// **This is not "has bootstrap happened".** See [`AuthTable::bootstrap_record`].
    pub fn is_empty(&self) -> Result<bool> {
        Ok(self.doc.load()?.keys.is_empty())
    }

    /// **Has the unauthenticated enrolment window ever been used?**
    ///
    /// `Some` means it is shut, permanently, whatever the identity table now
    /// holds. `Err` means the table could not be read, and a caller must treat
    /// that as *unknown*, never as *no*: reading "cannot read" as "not
    /// bootstrapped" is precisely how the door reopens for real. Every caller
    /// in the tree propagates it; `bootstrap_persistence.rs` corrupts the
    /// document and asserts the enrolment RPC answers `UNAVAILABLE` rather than
    /// enrolling.
    pub fn bootstrap_record(&self) -> Result<Option<BootstrapRecord>> {
        Ok(self.doc.load()?.bootstrap.clone())
    }

    /// How many principals, keys and grants there are, for `Admin.Stats`.
    ///
    /// Principals and keys are counted, not walked. Grant *rows* are not grant
    /// *entries*; one entry holds a principal's whole list, so the third number
    /// costs a walk of the ACL. `Admin.Stats` is an operator asking a question,
    /// not a connection being served.
    pub fn counts(&self) -> Result<(usize, usize, usize)> {
        let doc = self.doc.load()?;
        let grants = doc.grants.values()?.iter().map(Vec::len).sum();
        Ok((doc.principals.len(), doc.keys.len(), grants))
    }

    /// **THE ONE WRITER of the first identity, and the one place the
    /// unauthenticated enrolment window is closed.**
    ///
    /// `--admin-key` at start-up and the `Bootstrap.Enroll` RPC both land here
    /// (one implementation). There is exactly one code path that can produce a
    /// [`BootstrapRecord`], and it does five things in **one** compare-and-swap
    /// on `refs/gunnar/auth`:
    ///
    /// 1. refuse if a [`BootstrapRecord`] already exists;
    /// 2. refuse if the identity table already holds a principal or a key —
    ///    belt to (1)'s braces, for a table written before this field existed;
    /// 3. create the principal, whose id **is** its account name;
    /// 4. register the key, and grant it `Server`/`Admin`, which is what
    ///    "gunnar-admin" is;
    /// 5. write the [`BootstrapRecord`].
    ///
    /// # Why one CAS is the whole concurrency argument
    ///
    /// [`DocRef::update`] re-reads the document on every attempt and commits
    /// with a compare-and-swap on the reference. Two enrolments racing both run
    /// step 1 against a table with no record; one of them commits; the other
    /// loses the CAS, is retried from a **fresh read**, sees the winner's
    /// record, and returns [`Error::AlreadyBootstrapped`]. There is no window
    /// between "check" and "write" because they are the same write. A lock, a
    /// flag file or a second reference would each have one.
    ///
    /// Partial failure cannot leave the door half-shut for the same reason: the
    /// principal, the key, the grant and the record are one blob and one ref
    /// move. Either all five happened or none did.
    fn install_first_identity(&self, id: &str, name: &str, openssh: &str) -> Result<Enrolled> {
        crate::sandbox::check_component(id, id)?;
        if id == crate::model::ANONYMOUS_PRINCIPAL_ID {
            return Err(Error::ReservedPrincipalId(id.to_owned()));
        }
        let parsed = parse_public_key(openssh)?;
        let fingerprint = Fingerprint::of_key(&parsed);
        let algo = parsed.algorithm().as_str().to_owned();
        let normalised = parsed.to_openssh().map_err(Error::from)?.trim().to_owned();

        let enrolled = self.doc.update(|doc| {
            if let Some(record) = &doc.bootstrap {
                return Err(Error::AlreadyBootstrapped {
                    at: record.at,
                    principal_id: record.principal_id.clone(),
                });
            }
            // A table with identities in it but no record predates this field.
            // Refuse anyway and say which of the two refusals fired, or the
            // upgrade path becomes a way to enrol a second admin on a live
            // server.
            if !doc.principals.is_empty() || !doc.keys.is_empty() {
                return Err(Error::AlreadyProvisioned {
                    principals: doc.principals.len(),
                    keys: doc.keys.len(),
                });
            }
            let principal = Principal {
                id: id.to_owned(),
                name: name.to_owned(),
                kind: PrincipalKind::User,
                created: now_ms(),
            };
            doc.principals.insert(id.to_owned(), principal.clone())?;
            Self::index_name(doc, &principal)?;
            let key = PublicKey {
                fingerprint_sha256: fingerprint.as_str().to_owned(),
                algo: algo.clone(),
                principal_id: id.to_owned(),
                added: now_ms(),
                last_used: None,
                expires: None,
                openssh: normalised.clone(),
            };
            doc.keys
                .insert(fingerprint.as_str().to_owned(), key.clone())?;
            // gunnar-admin. A `Scope::Server` Admin grant and nothing new:
            // it is the same row `Access.Grant --server … admin` writes, so
            // any gunnar-admin can hand it to anybody else through the ordinary
            // authenticated verb and there is no second path (one implementation).
            doc.grants.insert(
                id.to_owned(),
                vec![Grant::server(id.to_owned(), Level::Admin)],
            )?;
            let record = BootstrapRecord {
                at: now_ms(),
                principal_id: id.to_owned(),
                fingerprint: fingerprint.as_str().to_owned(),
                account: id.to_owned(),
            };
            doc.bootstrap = Some(record.clone());
            Ok(Enrolled {
                principal,
                key,
                record,
            })
        })?;

        tracing::warn!(
            target: "gunnar::audit",
            event = "bootstrap.enrolled",
            principal = %enrolled.principal.id,
            account = %enrolled.record.account,
            %fingerprint,
            "THE UNAUTHENTICATED ENROLMENT WINDOW IS NOW CLOSED. It cannot be reopened by \
             removing this principal or this key: the record lives beside the identity table, \
             not in it."
        );
        Ok(enrolled)
    }

    /// **Enrol the first identity over the wire.** The `Bootstrap.Enroll` RPC.
    ///
    /// One name, one public key. The principal that comes out is
    /// gunnar-admin (a `Server`/`Admin` grant) *and* intrinsically admin of the
    /// account `<name>`; everything after this is ordinary authenticated
    /// administration through `Access.*`.
    ///
    /// Refuses with [`Error::AlreadyBootstrapped`] once it has succeeded once,
    /// **including after every identity it created has been deleted**.
    pub fn enroll_first_user(&self, name: &str, openssh: &str) -> Result<Enrolled> {
        self.install_first_identity(name, name, openssh)
    }

    /// Install the first admin from a public key, on an empty table.
    ///
    /// This is the answer to "how does the first key get in on a box with no
    /// shell" **when the operator already has the key at boot time**: it comes
    /// from `--admin-key` and is installed here. It closes the enrolment window
    /// exactly as [`AuthTable::enroll_first_user`] does — otherwise a server
    /// started with `--admin-key` would still be standing there with an open
    /// unauthenticated door.
    ///
    /// Returns [`Bootstrap::AlreadyProvisioned`] and changes nothing if
    /// bootstrap has already happened, so it is safe to call on every boot.
    pub fn bootstrap_admin(&self, openssh: &str) -> Result<Bootstrap> {
        match self.install_first_identity(BOOTSTRAP_ADMIN_ID, "bootstrap admin", openssh) {
            Ok(enrolled) => Ok(Bootstrap::Installed(enrolled.principal)),
            Err(Error::AlreadyBootstrapped { .. }) | Err(Error::AlreadyProvisioned { .. }) => {
                tracing::debug!("identity table already provisioned; bootstrap key ignored");
                Ok(Bootstrap::AlreadyProvisioned)
            }
            Err(err) => Err(err),
        }
    }

    /// **The start-up gate.** Install the admin key if one was supplied and
    /// nothing has been enrolled, then say what state this server is in.
    ///
    /// | bootstrap record | keys | `admin_key` | result |
    /// |---|---|---|---|
    /// | present | any | any | [`Readiness::Ready`], key ignored |
    /// | absent | any | `Some` | admin installed, [`Readiness::Ready`] |
    /// | absent | none | `None` | [`Readiness::AwaitingBootstrap`] — **door open** |
    /// | present | none | any | [`Readiness::SealedAndEmpty`] — door stays shut |
    ///
    /// The third row used to be a refusal to start, which is why gunnar had no
    /// first boot; see [`Readiness`].
    ///
    /// A read fault propagates as `Err`. It is never reported as "nothing is
    /// enrolled": that would open the door on a corrupt or unreadable document,
    /// which is the single worst failure this design can have.
    pub fn prepare_to_serve(&self, admin_key: Option<&str>) -> Result<Readiness> {
        if let Some(key) = admin_key {
            // Idempotent, and it seals. A no-op when bootstrap has happened.
            self.bootstrap_admin(key)?;
        }
        let doc = self.doc.load()?;
        if !doc.keys.is_empty() {
            return Ok(Readiness::Ready);
        }
        match &doc.bootstrap {
            Some(record) => Ok(Readiness::SealedAndEmpty {
                record: Some(Box::new(record.clone())),
            }),
            // A table with principals but no keys and no record is not a fresh
            // server either: `install_first_identity` refuses it, so reporting
            // the window as open here would be a status that disagrees with the
            // gate. `bootstrap_window_open` is the one predicate both use.
            None if !doc.principals.is_empty() => Ok(Readiness::SealedAndEmpty { record: None }),
            None => Ok(Readiness::AwaitingBootstrap),
        }
    }

    /// **Is `Bootstrap.Enroll` servable without a credential right now?**
    ///
    /// The single predicate the gate and the status report both read, so a
    /// server can never say "closed" and then enrol, or say "open" and then
    /// refuse. Its three clauses are the three refusals
    /// [`AuthTable::install_first_identity`] performs, in the same order.
    pub fn bootstrap_window_open(&self) -> Result<bool> {
        let doc = self.doc.load()?;
        Ok(doc.bootstrap.is_none() && doc.principals.is_empty() && doc.keys.is_empty())
    }
}

/// What [`AuthTable::enroll_first_user`] created: the principal, its key, and
/// the record that shut the door behind it.
#[derive(Clone, PartialEq, Eq, Debug)]
pub struct Enrolled {
    /// The principal that was created. Its id is also its account name.
    pub principal: Principal,
    /// The key it was created with.
    pub key: PublicKey,
    /// The record that shut the enrolment window.
    pub record: BootstrapRecord,
}

#[cfg(test)]
mod name_index_tests {
    use super::*;

    /// A row written by a server from before the `names` index existed has no
    /// index entry — and it must still be found by name, and its name must
    /// still be refused to a newcomer. Driven by writing the row the old way,
    /// straight into `principals` with no index, which nothing public can do
    /// any more; that is why this test lives here.
    #[test]
    fn a_row_from_before_the_index_resolves_by_name_and_keeps_it() {
        let store = crate::memory_control_store().expect("store");
        let auth = AuthTable::new(store).expect("table");
        let legacy = Principal::new("alice", "alice", PrincipalKind::User).unwrap();
        auth.doc
            .update(|doc| {
                doc.principals.insert("alice".into(), legacy.clone())?;
                Ok(())
            })
            .expect("write the old way");
        assert!(
            auth.doc.load().unwrap().names.is_empty(),
            "the fixture must be a row with NO index entry, or the test proves nothing"
        );

        assert_eq!(
            auth.principal_by_name("alice").expect("read").map(|p| p.id),
            Some("alice".to_owned()),
            "a pre-index row must be found by its name"
        );
        let err = auth
            .add_principal(Principal::new("alice2", "alice", PrincipalKind::User).unwrap())
            .expect_err("a pre-index row's name is still taken");
        assert!(matches!(err, Error::NameExists { .. }), "{err}");

        // Renaming it brings it into the index, and frees nothing that is an id.
        auth.rename_principal("alice", "Alice").expect("rename");
        assert_eq!(
            auth.principal_by_name("Alice").expect("read").map(|p| p.id),
            Some("alice".to_owned())
        );
        assert!(auth.principal_by_name("alice").expect("read").is_none());
    }
}

#[cfg(test)]
mod certificate_fingerprint_tests {
    use super::{fingerprint_of_blob, fingerprint_of_certificate};

    /// **Different certificates must fingerprint differently, and the same one
    /// must fingerprint the same.**
    ///
    /// The first half is what makes the value an identity at all. The second is
    /// what makes it storable: a fingerprint that varied per call would put a
    /// principal in the `AuthTable` that nothing could ever match.
    #[test]
    fn a_certificate_fingerprint_is_stable_and_distinguishing() {
        let a = b"-----BEGIN CERTIFICATE-----AAAA-----END CERTIFICATE-----";
        let b = b"-----BEGIN CERTIFICATE-----AAAB-----END CERTIFICATE-----";

        assert_eq!(
            fingerprint_of_certificate(a),
            fingerprint_of_certificate(a),
            "the same certificate must fingerprint the same, or nothing could store one"
        );
        assert_ne!(
            fingerprint_of_certificate(a),
            fingerprint_of_certificate(b),
            "two certificates differing by one byte must not share a fingerprint"
        );
    }

    /// **A certificate and an SSH key must never collide**, even on identical
    /// bytes.
    ///
    /// They cannot today — both are SHA256 over the input, so identical input
    /// gives identical output — and that is worth *asserting* rather than
    /// assuming, because the consequence is not cosmetic: a collision would let
    /// a credential of one kind be presented as the other and match a principal
    /// it was never issued to.
    ///
    /// This test therefore documents a REAL property of the current scheme: the
    /// two spaces are not separated. Registering a fingerprint registers the
    /// bytes, not the credential type. If that is ever not wanted, the fix is a
    /// domain separator in the hash input, and this test is where it will be
    /// noticed.
    #[test]
    fn the_two_credential_kinds_share_one_hash_space_and_that_is_recorded() {
        let bytes = b"identical bytes, two meanings";
        assert_eq!(
            fingerprint_of_certificate(bytes),
            fingerprint_of_blob(bytes),
            "if these ever differ, the hash grew a domain separator and the AuthTable's \
             existing entries are all invalidated — which is a migration, not a tidy-up"
        );
    }
}

/// [`AuthTable::plan_room`]'s answer, and which seat lot a claim takes.
struct RoomPlan {
    answer: RoomAnswer,
    lot: Option<usize>,
}

/// **The operator's server grant stops at an account's door; the account
/// admin approves a look** (2026-10-05). A child module so the identity table
/// keeps one file per concern.
pub mod operator_access;

/// **The seller's mark** (2026-10-08): only a SERVER-scope grant makes a principal's namespace
/// the seller's own (`Namespace.owner_server_grant`) — what a selling shop leaves out when it
/// gives every person's account a fact.
#[cfg(test)]
mod seller_mark_tests {
    use super::*;
    use crate::model::{Grant, Level};

    fn table() -> AuthTable {
        AuthTable::new(crate::memory_control_store().expect("store")).expect("table")
    }

    fn person(auth: &AuthTable, id: &str) {
        auth.add_principal(Principal::new(id, id, PrincipalKind::User).unwrap())
            .expect("a person");
    }

    /// ★ **A normal person never holds the seller's mark**: their own account, a namespace
    /// grant (even Admin, even on another account), a repository grant, being an employee —
    /// none of it is a server grant. A server grant, direct or through a group, is.
    #[test]
    fn only_a_server_grant_marks_the_seller_and_a_normal_person_never_holds_it() {
        let auth = table();
        for id in ["cust", "boss", "emp", "op", "staff", "ops-group"] {
            if id == "ops-group" {
                auth.add_principal(Principal::new(id, id, PrincipalKind::Group).unwrap())
                    .unwrap();
            } else {
                person(&auth, id);
            }
        }
        auth.grant(Grant::namespace("cust", "cust", Level::Admin))
            .unwrap();
        auth.grant(Grant::namespace("cust", "boss", Level::Admin))
            .unwrap();
        let _ = auth.enroll_employee(
            "boss",
            Principal::new("emp2", "emp2", PrincipalKind::User).unwrap(),
        );
        for normal in [
            "cust",
            "boss",
            "emp",
            "emp2",
            "nobody",
            crate::model::ANONYMOUS_PRINCIPAL_ID,
        ] {
            assert!(
                !auth.holds_server_grant(normal).unwrap(),
                "{normal} is a normal person (or nobody): never the seller"
            );
        }
        auth.grant(Grant::server("op", Level::Read)).unwrap();
        assert!(
            auth.holds_server_grant("op").unwrap(),
            "a server grant at any level is the seller's mark"
        );
        auth.grant(Grant::server("ops-group", Level::Write))
            .unwrap();
        auth.add_to_group("ops-group", "staff").unwrap();
        assert!(
            auth.holds_server_grant("staff").unwrap(),
            "through a group too"
        );
        assert!(
            !auth.holds_server_grant("cust").unwrap(),
            "and still never the customer"
        );
    }
}