gunnar_grpc/

seam.rs

//! **The storage seam: temporary, and marked so.**
//!
//! # Why this trait exists at all
//!
//! `gunnar-store` (the real `ObjectStore` / `RefStore` of the storage seam) and `gunnar-core`
//! (the repo model, identity table and ACL) are being written in
//! parallel with this crate. Blocking on them would have meant writing no
//! services at all, so the service implementations here are written against
//! [`ControlPlane`], a small local trait describing exactly what the seven
//! gRPC services need and nothing more.
//!
//! # SEAM: what happens when `gunnar-core` / `gunnar-store` land
//!
//! This trait is **not** the intended long-term shape and must not calcify:
//!
//! - The repo/namespace/principal/key/grant model belongs to `gunnar-core`.
//!   When it exists, [`ControlPlane`] becomes a thin *adapter* over it (one
//!   impl in this crate) and stops being an independent definition of the
//!   model. It must not become a second source of truth for what a `Repo` is
//!   (one implementation, never a twin).
//! - The ref/object methods ([`ControlPlane::list_refs`],
//!   [`ControlPlane::log`], [`ControlPlane::read_object`],
//!   [`ControlPlane::diff`]) are exactly the `RefStore`/`ObjectStore` surface
//!   of the storage seam seen from the control plane. They will delegate, not reimplement.
//! - [`MemoryControlPlane`] is a **test double**, not a backend. It is not the
//!   `mem` backend of the storage seam and must not be mistaken for one.
//!
//! # Why it is synchronous
//!
//! `ObjectStore` and `RefStore` in the storage seam are synchronous traits, and this one is
//! shaped to match so the adapter is a straight delegation. That also keeps it
//! object-safe without pulling in `async-trait`. A backend whose work is long
//! enough to matter belongs behind `spawn_blocking` at the call site, which is
//! a decision the real store can make once it knows its own costs.

use std::collections::BTreeMap;
use std::sync::Mutex;

use crate::auth::Access;

use gunnar_proto::{
    AddWebhookRequest, BranchProtection, Commit, FileChange, Grant, GrantLevel, HashKind,
    MirrorConfig, MirrorCredentialKind, MirrorState,
    Namespace, Principal, PrincipalKind, PublicKey, PullRequest, ReadLfsObjectResponse,
    ReadObjectResponse, RefNamePolicy,
    RefTarget, ReplicationStatusResponse, ReplicationTarget, Repo, RepoPath,
    IssueState, RepoStorage, ResyncReplicationResponse, ReviewState, SignedForgeEvent,
    SignedKeyAttestation, SignedMembershipEvent, StorageInfoResponse, WebhookListResponse,
    Visibility,
};

/// **The two arguments a credential write must agree about**, checked once for
/// every implementation of the seam.
///
/// A `kind` with no `secret`, or a `secret` with no `kind`, are the same
/// request read two ways — and one of those readings SILENTLY FORGETS a stored
/// credential. That is the failure this refuses: an operator who meant to set a
/// token and left the kind unset would otherwise clear the key they already
/// had, and find out at the next sync.
///
/// Free function rather than a default body, because `CoreControlPlane` and the
/// memory double must apply the same rule and a default body is the one thing
/// an implementation can silently not call.
pub fn credential_arguments(
    kind: MirrorCredentialKind,
    secret: &str,
) -> ControlResult<()> {
    match (kind, secret.is_empty()) {
        (MirrorCredentialKind::Unspecified, true) => Ok(()),
        (MirrorCredentialKind::Unspecified, false) => Err(ControlError::InvalidArgument(
            "a credential value was given with no kind. `kind = UNSPECIFIED` is how a stored \
             credential is FORGOTTEN, and it must carry no value — otherwise \"set this token\" \
             and \"forget my key\" are the same request".to_owned(),
        )),
        (_, true) => Err(ControlError::InvalidArgument(format!(
            "a credential of kind {kind:?} was asked for with an empty value. To forget the \
             stored credential, send `kind = UNSPECIFIED`"
        ))),
        (_, false) => Ok(()),
    }
}

/// Everything that can go wrong behind the seam.
///
/// Deliberately small: it maps one-to-one onto the gRPC status codes the
/// services return, in [`crate::status`]: one converter, not a `match` copied
/// into seven service impls (one implementation).
#[derive(Debug, thiserror::Error)]
pub enum ControlError {
    #[error("{0} not found")]
    NotFound(String),
    #[error("{0} already exists")]
    AlreadyExists(String),
    #[error("invalid argument: {0}")]
    InvalidArgument(String),
    #[error("failed precondition: {0}")]
    FailedPrecondition(String),
    /// The request is well formed and would be answered, but answering it would
    /// cost the server more than it is willing to spend.
    ///
    /// Distinct from [`ControlError::InvalidArgument`] on purpose: the caller
    /// did nothing wrong, and a retry with a narrower request is the right
    /// response, which is what `RESOURCE_EXHAUSTED` tells a generated client and
    /// `INVALID_ARGUMENT` does not.
    #[error("resource exhausted: {0}")]
    ResourceExhausted(String),
    /// The seam is not wired to a real store yet. This is the honest answer for
    /// every ref/object operation until `gunnar-store` lands, and it is the
    /// value that must NOT quietly turn into an empty successful response. An
    /// empty ref list is indistinguishable from a working server with no refs.
    #[error("not implemented: {0}; the storage seam is not wired to gunnar-store yet")]
    Unimplemented(&'static str),
    #[error("internal error: {0}")]
    Internal(String),
    /// **The state could not be read.** A storage fault, not an answer.
    ///
    /// Distinct from [`ControlError::Internal`] because a caller must be able
    /// to tell "I could not find out" from "I found out and here it is". It
    /// exists for `Bootstrap.Status` above all: an unreadable identity table
    /// reported as `window_open: false` would be the most dangerous lie this
    /// server can tell, and one reported as `window_open: true` would open the
    /// door on a disk fault.
    #[error("unavailable: {0}")]
    Unavailable(String),
    /// No verified caller. Distinct from [`ControlError::PermissionDenied`] on
    /// purpose: "I do not know who you are" and "I know who you are and the
    /// answer is no" are different facts, and collapsing them makes an
    /// operator's first question unanswerable.
    #[error("unauthenticated: {0}")]
    Unauthenticated(String),
    #[error("permission denied: {0}")]
    PermissionDenied(String),
}

pub type ControlResult<T> = Result<T, ControlError>;

/// **A `Mirror.LogFetch` range, ready to stream** ([`ControlPlane::log_fetch_stream`]): run on a
/// blocking thread, it hands the range to the sink as `LogChunk` pieces and returns once the
/// range is out (twin-streaming, 2026-10-06).
pub type LogFetchProducer =
    Box<dyn FnOnce(&mut dyn FnMut(gunnar_proto::LogChunk) -> ControlResult<()>) -> ControlResult<()> + Send>;

/// The most blob bytes one `Mirror.LogFetch` piece carries: under every client's default 4 MiB
/// message ceiling with room for the meta.
pub const LOG_PIECE_BYTES: usize = 1 << 20;

/// One whole entry as `LogChunk` pieces into `sink`: the blob in [`LOG_PIECE_BYTES`] slabs,
/// `meta` on the first piece, `last` on the last; an empty blob is one piece.
pub fn log_entry_pieces(
    e: gunnar_proto::LogEntry,
    sink: &mut dyn FnMut(gunnar_proto::LogChunk) -> ControlResult<()>,
) -> ControlResult<()> {
    log_entry_pieces_from(e, 0, sink)
}

/// [`log_entry_pieces`] from byte `offset` of the blob on (restore-large-entry, 2026-10-07):
/// the pieces below it are not sent, every piece says its `offset`, and an offset at the blob's
/// end is one empty last piece. An offset PAST the end is not this entry as the asker knew it
/// (an entry erased since): it is sent whole from 0, and the asker sees a length that differs.
/// No `entry_crc`: a plane holding entries in memory has no stored crc to vouch with.
pub fn log_entry_pieces_from(
    e: gunnar_proto::LogEntry,
    offset: u64,
    sink: &mut dyn FnMut(gunnar_proto::LogChunk) -> ControlResult<()>,
) -> ControlResult<()> {
    let blob_len = e.blob.len() as u64;
    let start = if offset <= blob_len { offset as usize } else { 0 };
    let rest = &e.blob[start..];
    let n = rest.len().div_ceil(LOG_PIECE_BYTES).max(1);
    let mut meta = Some(e.meta);
    for i in 0..n {
        let piece = &rest[(i * LOG_PIECE_BYTES).min(rest.len())..((i + 1) * LOG_PIECE_BYTES).min(rest.len())];
        sink(gunnar_proto::LogChunk {
            seq: e.seq,
            kind: e.kind,
            meta: meta.take().unwrap_or_default(),
            blob_len,
            blob: piece.to_vec(),
            last: i + 1 == n,
            restore_head: 0,
            restore_bytes: 0,
            log_id: String::new(),
            offset: Some((start + (i * LOG_PIECE_BYTES).min(rest.len())) as u64),
            entry_crc: None,
            snapshot_item: None,
            snapshot_id: String::new(),
        })?;
    }
    Ok(())
}

/// **One `Bootstrap.Restore` stream, taken a chunk at a time** (restore-large-entry,
/// 2026-10-07) — what the handler feeds every chunk of the stream into, so where an entry's
/// bytes live while it arrives is the plane's business and not the wire's.
///
/// The server's intake stages an entry larger than one piece on the box's disk as it arrives
/// and keeps what it holds when the stream ends inside it, so the next stream resumes the entry
/// at the byte it reached (`Bootstrap.Status.restore_partial_*`, `LogFetchRequest.from_offset`);
/// the default one ([`ControlPlane::restore_intake`]) gathers each entry in memory, as every
/// restore did before.
pub trait RestoreIntake: Send {
    /// One chunk: a piece of an entry (`seq` > 0) or of a snapshot (`seq` 0). An entry is
    /// applied when its last piece arrives whole.
    fn take(&mut self, chunk: &gunnar_proto::LogChunk) -> ControlResult<()>;
    /// The stream ended cleanly. `Err` when it ended inside an entry — naming what this box
    /// holds of it and where the next stream resumes.
    fn end_of_stream(&mut self) -> ControlResult<()>;
    /// `(applied, lost)` by this stream: entries applied, and replayed CAS claims that lost.
    fn counts(&self) -> (u64, u64);
}

/// The intake every plane had before: each entry gathered whole in memory, then
/// [`ControlPlane::restore_apply`]. An entry the stream ends inside is lost with the stream.
struct GatherIntake<'a, P: ?Sized> {
    plane: &'a P,
    next: u64,
    open: Option<gunnar_proto::LogEntry>,
    want_len: u64,
    applied: u64,
    lost: u64,
}

impl<P: ControlPlane + ?Sized> RestoreIntake for GatherIntake<'_, P> {
    fn take(&mut self, chunk: &gunnar_proto::LogChunk) -> ControlResult<()> {
        // A snapshot piece (seq 0): staged by the plane, its terminator installs it.
        if chunk.seq == 0 {
            if let Some(e) = &self.open {
                return Err(ControlError::InvalidArgument(format!("a snapshot piece arrived with seq {} unfinished", e.seq)));
            }
            if let Some(at) = self.plane.restore_snapshot_chunk(chunk)? {
                self.next = at + 1;
            }
            return Ok(());
        }
        let entry = match self.open.as_mut() {
            Some(e) if e.seq == chunk.seq => e,
            Some(e) => return Err(ControlError::InvalidArgument(format!("seq {} arrived with seq {} unfinished", chunk.seq, e.seq))),
            None => {
                self.want_len = chunk.blob_len;
                self.open.insert(gunnar_proto::LogEntry { seq: chunk.seq, kind: chunk.kind, meta: chunk.meta.clone(), blob: Vec::with_capacity(gunnar_proto::reserve_for(chunk.blob_len)) })
            }
        };
        entry.blob.extend_from_slice(&chunk.blob);
        if !chunk.last {
            return Ok(());
        }
        let done = self.open.take().expect("open entry");
        if done.blob.len() as u64 != self.want_len {
            return Err(ControlError::InvalidArgument(format!("seq {}: {} of {} blob bytes arrived", done.seq, done.blob.len(), self.want_len)));
        }
        if done.seq < self.next {
            return Ok(());
        }
        if done.seq > self.next {
            return Err(ControlError::FailedPrecondition(format!("seq {} arrived where {} was due; resend from {}", done.seq, self.next, self.next)));
        }
        if self.plane.restore_apply(done)? {
            self.lost += 1;
        }
        self.applied += 1;
        self.next += 1;
        Ok(())
    }

    fn end_of_stream(&mut self) -> ControlResult<()> {
        if self.open.is_some() {
            return Err(ControlError::InvalidArgument("the stream ended inside an entry; resend from the box's head + 1".into()));
        }
        Ok(())
    }

    fn counts(&self) -> (u64, u64) {
        (self.applied, self.lost)
    }
}

// ───────────────────────────────────────────────────────────────────────────
// The admission gate, reached through a seam (the gRPC half of the shared gate)
// ───────────────────────────────────────────────────────────────────────────
//
// The git front doors — SSH and smart HTTP — put every transfer through ONE
// gate object, `gunnar_ssh::Admission`: max-min-fair per-principal share, a
// server-wide concurrency ceiling, and a memory ledger. `Repos.ArchiveOut`
// streams a whole repository's stored bytes down the control-plane channel; it
// is a transfer in everything but name and must take the SAME permit a
// `git-upload-pack` does, or the third front door is the one with no gate.
//
// `gunnar_ssh::Admission` cannot be named here: `gunnar-grpc` is on the
// client/control graph (`gunnar-cli --features control` pulls this crate WITHOUT
// `gunnar-server`), and a `gunnar-ssh` dependency here would drag the whole SSH
// server — russh and all — into every client. So the gate is reached through a
// seam, exactly as the storage backend is reached through `ControlPlane`:
// `gunnar-server` implements [`TransferGate`] over the real `Arc<Admission>` the
// SSH and HTTP fronts already hold. This is a seam TO the one gate, not a second
// gate — no admission logic lives here.

/// A held admission permit for a control-plane transfer.
///
/// It is opaque on purpose: the real permit is a `gunnar_ssh::AdmissionGuard`,
/// which this crate may not name. Dropping it releases the server-wide slot, the
/// per-principal count and the memory reservation — so the caller holds it for
/// the whole stream and lets it fall at the end, the same lifetime the SSH and
/// HTTP paths give their guard.
pub struct TransferPermit(pub Box<dyn TransferHold>);

/// What a held [`TransferPermit`] can still be told.
///
/// One thing: that the transfer's real work has begun, which is the sample the
/// governor's CoDel controller reads. The SSH and HTTP paths call the guard's
/// own `entered_service` at the first line of the spawned transfer; this is that
/// call, across the seam.
pub trait TransferHold: Send {
    /// File the sojourn sample: everything before this was the transfer waiting.
    fn entered_service(&self);
}

/// Why the gate refused a control-plane transfer, carrying the same fields the
/// SSH and HTTP refusals log.
///
/// The counters are read from the gate at the moment of refusal so the audit
/// line names the state that refused — `in_flight`, `peak_in_flight`, the
/// governed `limit`, `reserved_bytes` and the running `refused_total` — rather
/// than a bare "busy". `message` is the gate's own `client_message`, so the
/// sentence the operator reads is the one the log recorded.
#[derive(Clone, Debug)]
pub struct TransferRefused {
    /// The gate's own client-facing sentence.
    pub message: String,
    /// Transfers in flight across all fronts when this was refused.
    pub in_flight: u64,
    /// The high-water mark of `in_flight`.
    pub peak_in_flight: u64,
    /// The governed concurrency limit in force at the refusal.
    pub limit: usize,
    /// Bytes the memory ledger had reserved to in-flight transfers.
    pub reserved_bytes: u64,
    /// How many transfers this gate has refused since start-up, this one
    /// included.
    pub refused_total: u64,
}

/// The seam to the ONE admission gate (`gunnar_ssh::Admission`).
///
/// `gunnar-server` implements it over the real `Arc<Admission>` the SSH and HTTP
/// fronts hold; a plaintext test router that wires no gate leaves it `None` and
/// the transfer-shaped RPCs run ungated, which is exactly the ungated state
/// the red half of the test drives.
pub trait TransferGate: Send + Sync {
    /// Admit `principal` for one control-plane transfer of `namespace`'s data,
    /// or refuse it.
    ///
    /// **Never blocks** — it is `gunnar_ssh::Admission::try_admit` across the
    /// seam. On success the returned [`TransferPermit`] is held for the whole
    /// stream; on refusal the [`TransferRefused`] carries the audit fields and
    /// the client sentence.
    ///
    /// `namespace` is carried for the flow ledger on the other side, which
    /// counts a refusal against the tenant that met it. It was not here before
    /// the ledger was, and the call site always had it: the refusal was logged
    /// with `repo = %path` in the same breath as it was counted against nobody.
    fn admit_transfer(
        &self,
        principal: &str,
        namespace: &str,
    ) -> Result<TransferPermit, TransferRefused>;
}

// ───────────────────────────────────────────────────────────────────────────
// Identity: the ACL model, asked through the seam, decided in gunnar-core
// ───────────────────────────────────────────────────────────────────────────

/// A key as the gate needs it: whose it is, and the OpenSSH line to verify a
/// signature against.
///
/// The line is carried rather than a parsed key because `gunnar-ssh` and
/// `gunnar-core` link semver-incompatible `ssh-key` versions and no key
/// *type* can cross that line. Text can.
/// **A web credential (a WebAuthn passkey) as the plane stores it** (Stage L,
/// 2026-10-04): the wire row, and the public key the wire never carries —
/// P-256, SEC1 uncompressed (`0x04 || X || Y`), which is what gunnar verifies a
/// login's assertion with.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StoredCredential {
    pub credential: gunnar_proto::WebCredential,
    pub public_key: Vec<u8>,
}

/// **An invite to be stored** (Stage L): a pending passkey enrolment on an
/// existing principal. The plane is handed the secret's HASH only — the secret
/// itself is shown once to whoever created the invite and is never stored.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NewInvite {
    pub id: String,
    pub principal_id: String,
    pub created_by: String,
    pub created_at_unix_ms: i64,
    pub expires_at_unix_ms: i64,
    pub note: String,
    /// The server-wide grant a redemption applies; `Unspecified` for none.
    pub server_level: GrantLevel,
    /// Lowercase hex SHA-256 of the secret.
    pub secret_hash: String,
}

/// **When a principal last signed in, and last changed something**
/// (2026-10-21) — what `Access.ListPrincipals` answers beside each principal.
/// `None`: never, as far as this box knows.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct PrincipalSeen {
    /// Unix ms of the last console session issued to it.
    pub last_sign_in_unix_ms: Option<i64>,
    /// Unix ms of its last authenticated call that changed something.
    pub last_operation_unix_ms: Option<i64>,
}

/// **What an invite secret is now** (2026-10-19, `Access.CheckInvite`):
/// read by the plane, nothing spent.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum InviteCheck {
    /// Redeemable now: the invite's row.
    Valid(gunnar_proto::Invite),
    /// A passkey was registered with it.
    Used,
    /// Past its expiry, never redeemed.
    Expired,
    /// No invite has this secret.
    Unknown,
}

/// **What a verified web login commits** (Stage L), in one update: the
/// passkey's new counter and last use, and the console's session key on the
/// passkey's principal, expiring at `expires_at_unix_ms`. No attestation is
/// owed for the session key — the verified passkey assertion IS the person's
/// consent.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WebLoginCommit {
    pub credential_id: Vec<u8>,
    /// The counter the verified assertion presented.
    pub sign_count: u32,
    pub now_unix_ms: i64,
    /// One OpenSSH public key line.
    pub session_openssh: String,
    pub comment: String,
    pub expires_at_unix_ms: i64,
}

/// What a committed web login answers.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WebLoginOutcome {
    pub principal: Principal,
    pub credential: gunnar_proto::WebCredential,
    /// The session key's fingerprint.
    pub fingerprint: String,
}

/// **A passkey registered from a creation ceremony** (Stage L), in one
/// update: the invite (if any) redeemed and removed, the credential stored,
/// the invite's server grant applied.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WebRegistration {
    /// Lowercase hex SHA-256 of the invite's secret; `None` when the relay
    /// names the principal itself.
    pub invite_secret_hash: Option<String>,
    /// Required without an invite; with one, empty or the invite's own.
    pub principal_id: Option<String>,
    pub credential_id: Vec<u8>,
    pub public_key: Vec<u8>,
    pub label: String,
    pub sign_count: u32,
    pub now_unix_ms: i64,
}

/// **An approved machine enrolment** (`Access.ApproveMachineEnrolment`,
/// 2026-10-05), committed in ONE update: the MACHINE principal made
/// when it is not there (a principal of that id that is not a machine is
/// `FailedPrecondition`); its server-wide grant set; the credential stored
/// (an id already registered is `AlreadyExists`); and the approver's step-up
/// passkey's counter and last use, when a person approved.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MachineApproval {
    /// The machine principal: id = name (`kit-*`).
    pub machine: String,
    /// `Admin` or `Write`.
    pub server_level: GrantLevel,
    pub credential_id: Vec<u8>,
    pub public_key: Vec<u8>,
    pub label: String,
    pub sign_count: u32,
    pub now_unix_ms: i64,
    /// The approver's step-up passkey and the counter it presented.
    pub step_up: Option<(Vec<u8>, u32)>,
}

/// What an approval committed.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MachineApproved {
    pub principal: Principal,
    pub credential: gunnar_proto::WebCredential,
    /// The machine principal was made by this approval.
    pub created: bool,
}

/// What a registration answers.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WebRegistered {
    pub credential: gunnar_proto::WebCredential,
    pub principal: Principal,
    /// The server-wide grant the redeemed invite applied; `Unspecified` for
    /// none.
    pub server_level: GrantLevel,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RegisteredKey {
    /// The ACL principal the key belongs to.
    pub principal_id: String,
    /// `<algo> <base64> [comment]`, exactly as `authorized_keys` writes it.
    pub openssh: String,
    /// The principal is a `Machine` (the ACL model): a server's deploy key rather than a
    /// person's. The one caller a browse replica's control plane still lets
    /// write (`Repos.Delete`, the browse replica).
    pub machine: bool,
}

/// Who a **client certificate** belongs to, once the handshake has already
/// proven the peer holds its private key.
///
/// The difference from [`RegisteredKey`] is the whole reason this is a separate
/// type: a `RegisteredKey` carries an `openssh` line **because the gate still
/// has to verify a signature against it**. A certificate has no such step left.
/// TLS proved possession during the handshake and the issuer proved provenance,
/// so what remains is one table lookup: are these bytes registered, and to
/// whom. Handing back an `openssh` field here would invite a second
/// verification against material that is not the credential, which is how a
/// check comes to pass against the wrong thing.
///
/// The certificate is **not** an identity model of its own (the mTLS and
/// certificate-identity plans): the fingerprint lands in the same ACL key table
/// as every SSH key, under `algo: "x509"`, and the same `AuthTable::may` decides
/// what the principal can do. A certificate the CA signed and nobody registered
/// reaches the gate and is refused there.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CertificateIdentity {
    /// The ACL principal the certificate was registered to.
    pub principal_id: String,
    /// The `SHA256:…` fingerprint of the DER, as
    /// `gunnar_core::auth::fingerprint_of_certificate` computes it. Carried so
    /// the audit record can name the credential without re-hashing it in a
    /// crate that must not own that decision.
    pub fingerprint: String,
}

/// **The two ACL questions the gRPC gate asks, and neither of them is answered
/// in this crate.**
///
/// Split out of [`ControlPlane`] rather than folded into it so that the
/// distinction is visible at the type level: a control plane that has an
/// identity table implements this, and one that does not cannot be mistaken
/// for one that does.
///
/// # Why there is no default implementation that allows
///
/// [`Identity::may`] could be *derived* from [`ControlPlane::list_grants`]:
/// filter by principal, take the maximum level. That derivation would be a
/// **second grant decision**, living beside `gunnar_core::AuthTable::may`, free
/// to disagree with it. The one-sandbox rule records what two security checks that disagree
/// cost the last time: the safe one gives false confidence about the permissive
/// one. So this is a delegation, and the only implementation with an opinion is
/// `AuthTable`'s.
pub trait Identity: Send + Sync {
    /// The registered key for a `SHA256:…` fingerprint, or `None` if there is
    /// no such key, the key has expired, or its principal is gone.
    ///
    /// `Result<Option<_>>`, not `Option<_>`: the table lives in storage and a
    /// read can fail, and a bare `Option` has to invent an answer for that. The
    /// safe-looking invention (`None`) turns a storage fault into a silent
    /// authentication failure (the ACL model, the same reason `may` returns a `Result`).
    fn registered_key(&self, fingerprint: &str) -> ControlResult<Option<RegisteredKey>>;

    /// **A key this box was FED to enrol and has not enrolled yet**: its
    /// OpenSSH line, for a fingerprint [`Identity::registered_key`] does not
    /// know (cov-E r3 2026-10-06, run 3). A primary's fed principals — the
    /// front's certify key, the console's, monetize's — are enrolled by its
    /// pairing loop after the installer's feed lands: a moment later on a
    /// healthy box, MINUTES later on a starved one (measured: 4 m 47 s, the
    /// certify key sixth in line). A caller holding one of those keys is not a
    /// stranger, and the gate answers it `Unavailable` (ask again), never
    /// `Unauthenticated`, once its signature verifies against the fed key
    /// ([`crate::auth::AuthFailure::Enrolling`]).
    ///
    /// Defaulted to `None` — every unknown fingerprint unknown, the answer
    /// before — because `None` IS the refusal: an adapter that never says a
    /// key is awaited answers exactly as it always did, and nothing is let in
    /// by this answer either way.
    fn awaiting_enrolment(&self, _fingerprint: &str) -> Option<String> {
        None
    }

    /// Who these **client-certificate** DER bytes were registered to, or `None`
    /// if nobody registered them, the registration expired, or its principal is
    /// gone.
    ///
    /// # Why the DER and not a fingerprint
    ///
    /// Hashing the credential is the one step that decides *what the credential
    /// is*, and `gunnar_core::auth::fingerprint_of_certificate` already owns
    /// that decision for both the SSH and the X.509 side of the ACL table. This
    /// crate deliberately does not link `gunnar-core`
    /// ([`crate::peer_certificate_der`] says why), so a fingerprint computed on
    /// this side of the seam would be a *second* opinion about what bytes a
    /// certificate is — and the two would agree right up until one of them
    /// changed. The seam therefore carries the bytes and the adapter hashes.
    ///
    /// # Why this is not a second identity model
    ///
    /// It returns an ACL principal id, from the same table
    /// [`Identity::registered_key`] reads, and every authorisation question is
    /// still answered by [`Identity::may`]. The certificate is a *credential*,
    /// like an SSH key is a credential. What the certificate's subject claims
    /// is never consulted: two certificates can claim the same subject, and the
    /// bytes are what was registered (the mTLS plan, and the tension with the
    /// certificate-identity plan's cert-CN shape is resolved this way on purpose).
    ///
    /// **Required, not defaulted.** A default returning `Ok(None)` would be a
    /// silent "no certificate identity here" for any adapter that forgot to
    /// implement it, and an adapter that forgot is exactly the one whose
    /// operators would then be told their registered certificate is unknown.
    /// The two implementations in this workspace both answer.
    fn registered_certificate(&self, der: &[u8]) -> ControlResult<Option<CertificateIdentity>>;

    /// **May this principal do this, at this level?** `gunnar_core::AuthTable::may`
    /// (the ACL model) seen through the seam.
    fn may(&self, principal_id: &str, access: &Access, level: GrantLevel) -> ControlResult<bool>;

    /// **[`ControlPlane::may`] at `level` for every repository of a listing**
    /// (round 5.3.1): one answer per row, in order. Defaulted to one `may` per
    /// row; the core plane answers the whole listing under one read of the
    /// catalog and one of the identity table, so concurrent listings no longer
    /// queue on each other's per-row store reads (test5 2026-10-08).
    fn may_each_repo(&self, principal_id: &str, paths: &[RepoPath], level: GrantLevel) -> ControlResult<Vec<bool>> {
        paths.iter().map(|p| self.may(principal_id, &Access::repo(p), level)).collect()
    }

    /// **The methods a NARROWED principal may call** (f-principal-reach), or
    /// `None` for a principal judged by its grants. A narrowed principal holds
    /// no grant: [`crate::service::ServiceContext::require`] lets it through
    /// exactly the methods listed and refuses every other by name. Defaulted
    /// to `None` because it only ever RESTRICTS a principal nobody narrowed —
    /// an adapter that does not know narrowing judges every caller as before.
    fn narrowed_methods(&self, _principal_id: &str) -> ControlResult<Option<Vec<String>>> {
        Ok(None)
    }
}

// ───────────────────────────────────────────────────────────────────────────
// First boot: the one unauthenticated door, seen through the seam
// ───────────────────────────────────────────────────────────────────────────

/// What a server can say about its own enrolment, without a credential.
///
/// Every field is a fact the server holds, never a derivation this crate makes:
/// [`BootstrapState::window_open`] is `gunnar_core::AuthTable::bootstrap_window_open`,
/// the one predicate that also gates the write. A second opinion computed here
/// out of `principals == 0` would be a status that could disagree with the
/// gate, and the permissive one would win silently (the one-sandbox rule).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct BootstrapState {
    /// `Bootstrap.Enroll` will be served without a credential.
    pub window_open: bool,
    /// The window has been used. Not the negation of `window_open`.
    pub bootstrapped: bool,
    pub bootstrapped_at_unix_ms: i64,
    pub bootstrapped_principal: String,
    pub bootstrapped_fingerprint: String,
    pub principals: u64,
    pub keys: u64,
    /// **The git listener is serving**, i.e. the boot window is closed.
    ///
    /// `false` while the server's start-up disk test is still saturating the
    /// disk a repository's first post-crash open has to read; git transfers are
    /// refused with `EX_TEMPFAIL` and a sentence containing `starting` for that
    /// whole window. It rides on `Bootstrap.Status` and not on `Admin.Health`
    /// because this is the one call an unenrolled server answers, and a boot
    /// script is exactly the caller with no credential.
    ///
    /// `Default` is `false`: an implementation that forgets it says "not yet",
    /// and "wait" is the safe thing to say wrongly.
    pub serving_git: bool,
    /// One sentence about [`BootstrapState::serving_git`].
    pub readiness: String,
    /// **The build stamp** — `gunnar_common::BUILD_REF`, the same string an
    /// SSH banner's comments field carries. Empty when the plane does not
    /// know it (a test double); the server fills it.
    pub build: String,
    /// `primary` or `twin`: the `--role` this box was started with, or a box
    /// that has since become a twin. What a door proves a twin medium by, since
    /// a twin serves no git and so has no banner.
    pub role: String,
    /// **The twin a RECHARGED main awaits** (`Bootstrap.Status.recharge_from`): set while the
    /// box is a re-admitted main that is fresh or restoring; empty otherwise.
    pub recharge_from: String,
    /// A restore is under way (`txlog-restoring`).
    pub restoring: bool,
    /// The box's transaction-log head.
    pub log_head_seq: u64,
    /// **The entry a restoring box holds part of**, staged on its disk (restore-large-entry,
    /// 2026-10-07): `(seq, bytes held durably)`; `None` when nothing is staged.
    pub restore_partial: Option<(u64, u64)>,
    /// **This box stages a twin's snapshot across streams** (restore-after-snapshot, 2026-10-07).
    pub restore_snapshot_resumable: bool,
    /// What a restoring box holds of a twin's snapshot: `(id, items whole, bytes of the next,
    /// whole)`; `None` when nothing is staged.
    pub restore_snapshot: Option<(String, u64, u64, bool)>,
}

/// What [`FirstBoot::enroll_first_identity`] created.
#[derive(Debug, Clone, PartialEq)]
pub struct Enrolled {
    pub principal: Principal,
    pub key: PublicKey,
    /// The namespace this principal is intrinsically admin of.
    pub account: String,
    pub bootstrapped_at_unix_ms: i64,
    /// A client certificate minted for the principal by the server's issuer,
    /// with its private key and the issuer certificate to pin. All three empty
    /// when the server has no issuer — enrolment does not fail on their
    /// account, because the window is already shut when minting runs.
    pub client_cert_pem: String,
    pub client_key_pem: String,
    pub issuer_cert_pem: String,
}

/// **The trust-on-first-use door.**
///
/// A supertrait of [`ControlPlane`] for the same reason [`Identity`] is one: a
/// control plane that cannot answer *has anybody been enrolled here* cannot be
/// served safely, and a compile error is cheaper than a runtime hole. There is
/// no default implementation, and in particular no default that answers
/// `window_open: true`.
pub trait FirstBoot: Send + Sync {
    /// The enrolment state, as the server holds it.
    ///
    /// `Result`, so a table that cannot be read is an error and never a
    /// cheerful `window_open: false` **or** `true`. Both inventions are wrong
    /// and one of them opens the door.
    fn bootstrap_state(&self) -> ControlResult<BootstrapState>;

    /// Create the first identity and shut the window, atomically.
    ///
    /// [`ControlError::FailedPrecondition`] once it has succeeded once,
    /// including after every identity it created has been deleted.
    fn enroll_first_identity(
        &self,
        name: &str,
        openssh_public_key: &str,
    ) -> ControlResult<Enrolled>;
}

/// Counts for `Admin.Stats`. Event counters come from the bus, not from here.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct ControlCounts {
    pub namespaces: u64,
    pub repos: u64,
    pub principals: u64,
    pub keys: u64,
    pub grants: u64,
}

/// One repository's index work in flight, for `Admin.Stats` (`index`) and the
/// dashboard's `index` row. Information, never a gate.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct IndexProgressRow {
    pub namespace: String,
    pub name: String,
    pub pack_id: u64,
    pub objects_done: u64,
    pub objects_total: u64,
    pub bytes_done: u64,
    pub bytes_total: u64,
    pub phase: String,
    pub started_at_unix_ms: i64,
    pub resumed: bool,
    pub reinflations: u64,
    pub pauses: u64,
}

impl IndexProgressRow {
    pub fn percent(&self) -> u32 {
        (self.objects_done * 100)
            .checked_div(self.objects_total)
            .map_or(0, |percent| percent.min(100) as u32)
    }
}

/// What a merge moved: the fact the `RefUpdated` event needs, handed back so
/// the SERVICE publishes it — the seam stays free of the event bus.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MergedRef {
    /// Full target ref name.
    pub ref_name: String,
    /// Where it was.
    pub old_oid: String,
    /// Where it is now: the source tip.
    pub new_oid: String,
}

/// What a new repository is created with. Server defaults are applied by the
/// implementation, not by the caller.
#[derive(Debug, Clone)]
pub struct NewRepo {
    pub path: RepoPath,
    pub hash_kind: HashKind,
    /// Full ref name. Empty means the server default.
    pub default_branch: String,
    pub visibility: Visibility,
    pub description: String,
    /// Take over a store already on disk at this name; see
    /// `CreateRepoRequest.adopt`.
    pub adopt: bool,
}

/// What a `Repos.Delete` was asked to do.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct DeleteOptions {
    /// Reclaim the bytes NOW instead of tombstoning them.
    pub purge: bool,
    /// How long a tombstone keeps them. `None` = the server's default.
    /// `Some(0)` is "at the next sweep" and is a decision, not an absence.
    pub retention_days: Option<u32>,
    /// Who asked. Recorded in the tombstone.
    pub principal_id: String,
}

/// What a `Repos.Delete` did. **Not a unit**: "the bytes went" and "the bytes
/// are being kept until a date" are different outcomes and a caller that could
/// not tell them apart could not report either one honestly.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DeleteOutcome {
    /// The bytes were reclaimed now.
    Purged {
        /// What came back.
        bytes_reclaimed: u64,
    },
    /// The bytes are held until the deadline.
    Tombstoned {
        /// When a sweep may take them.
        purge_after_unix_ms: i64,
        /// What is being held — and still charged for.
        bytes_kept: u64,
    },
}

/// **The two things `Admin.Stats` reports that are not counts of rows**:
/// bytes held by tombstones, and what replication is doing.
///
/// The standby rows are `gunnar_proto::StandbyStatus` and not a local twin:
/// `storage_info` already returns the proto type straight through, and a
/// second definition of "what a standby looks like" is exactly what one
/// implementation forbids. The default is all-zero and `configured: false`, which is what a
/// control plane with no replication *is* — a fact, not an absence.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct StatsExtras {
    /// Repositories deleted and not yet purged.
    pub tombstoned_repos: u64,
    /// Bytes they are holding — still charged to the account.
    pub tombstoned_bytes: u64,
    /// Is any standby configured?
    pub replication_configured: bool,
    /// replication counters.
    pub replication_enqueued: u64,
    /// Attempts a standby accepted.
    pub replication_synced: u64,
    /// Attempts that failed. Never resets.
    pub replication_failed: u64,
    /// Times the event subscription fell behind.
    pub replication_bus_gaps: u64,
    /// Every ref on this server, and how many of them no standby carries.
    /// Replication copies `refs/heads/*` and `refs/tags/*` only.
    pub replication_refs_total: u64,
    pub replication_refs_unmirrored: u64,
    /// One row per standby.
    pub standbys: Vec<gunnar_proto::StandbyStatus>,
    /// The tenants row: namespaces by ENFORCED plan state. Zero on a build
    /// that holds no facts.
    pub tenants_paid: u64,
    pub tenants_grace: u64,
    pub tenants_suspended: u64,
}

/// One soft-deleted repository, as `Repos.ListDeleted` reports it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DeletedRepoRow {
    /// Where it was.
    pub path: RepoPath,
    /// When the delete was asked for.
    pub deleted_at_unix_ms: i64,
    /// When a sweep may reclaim it; 0 = no deadline.
    pub purge_after_unix_ms: i64,
    /// Who asked.
    pub principal_id: String,
    /// Bytes still on disk, and still on the account.
    pub bytes: u64,
}

/// **One store directory with no catalog row and no soft-delete tombstone**,
/// as `Repos.ListOrphans` reports it.
///
/// The invisible class. Every reclaim path the server had keyed off a record —
/// `Repos.Delete` off the catalog, `Purge`/`Undelete`/`ListDeleted` and the
/// expiry sweep off a tombstone — so a directory with neither was reachable by
/// nothing at all, while `Repos.Create` refused its name because it was there.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OrphanRow {
    /// The name the directory stands at.
    pub path: RepoPath,
    /// Where it is on the server's disk. There is no shell on the appliance,
    /// so this is reported rather than looked up.
    pub dir: String,
    /// What it is holding.
    pub bytes: u64,
    /// `sha1`, `sha256`, or **empty for a store written before the
    /// `OBJECT_FORMAT` marker existed** — which is what a create with `adopt`
    /// takes the caller's hash at their word about.
    pub hash_marker: String,
    /// Newest mtime under the directory.
    pub modified_unix_ms: i64,
    /// Non-empty when a tombstone records a reclaim that FAILED here.
    pub failed_reclaim: String,
}

/// What a finished [`ControlPlane::archive_out`] cut did — the last message of
/// a `Repos.ArchiveOut` stream.
///
/// It is a `gunnar-grpc` type rather than the plugin's `ArchiveOutReceipt`
/// because this crate must not link a storage engine (the dependency budget): the seam is where
/// the engine's answer becomes the control plane's.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct ArchiveCut {
    /// The watermark this cut started above.
    pub from_generation: u64,
    /// The watermark to pass as `from_generation` next time.
    pub to_generation: u64,
    /// Payload bytes written into the sink.
    pub bytes_out: u64,
    /// SHA-256 of every byte written.
    pub file_sha256: [u8; 32],
}

/// Where a queued maintenance run has got to.
///
/// The seam's own vocabulary rather than the generated `gunnar.v1.GcState`,
/// for the same reason every other type in this file is: an adapter implements
/// this trait without linking the protos, and the mapping to the wire lives in
/// exactly one place (`service/repos.rs`).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GcState {
    /// Waiting its turn behind other repositories' gcs.
    Queued,
    /// The one gc the server is running.
    Running,
    /// Finished. `bytes_reclaimed` is the answer.
    Done,
    /// Finished badly. The repository was NOT maintained.
    Failed,
}

/// One queued or completed maintenance run, as a caller sees it.
#[derive(Debug, Clone)]
pub struct GcJob {
    /// Stable for the life of the job, so a console can tell its own run from
    /// somebody else's that started after it.
    pub job_id: String,
    pub path: RepoPath,
    pub state: GcState,
    /// How many jobs are ahead of this one. 0 unless [`GcState::Queued`].
    pub queue_position: u32,
    /// Bytes that left the filesystem. Only meaningful when [`GcState::Done`].
    pub bytes_reclaimed: u64,
    /// Only set when [`GcState::Failed`].
    pub error: Option<String>,
    pub enqueued_at_unix_ms: i64,
    /// 0 until the job leaves [`GcState::Running`].
    pub finished_at_unix_ms: i64,
}

// ───────────────────────────────────────────────────────────────────────────
// Accounts: the gate, and the per-user hard purge
// ───────────────────────────────────────────────────────────────────────────

/// Why an account is locked out. Mirrors `gunnar.v1.GateReason`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GateReason {
    /// Somebody decided. No deadline.
    Administrative,
    /// A purge is holding it, under a lease it re-stamps as it goes.
    Purge,
}

/// **What `Accounts.ClaimOperator` answered** (2026-10-04): the estate's
/// operator, as gunnar holds it now. See the proto for the rule; the plane is
/// where it is kept.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OperatorClaim {
    /// The operator: a USER, id = name, the account it is named after.
    pub principal: Principal,
    /// This call created the principal (and claimed its namespace).
    pub created: bool,
    /// The namespace already existed with no principal of that id and was
    /// adopted as the operator's account rather than refused.
    pub adopted_namespace: bool,
    /// `fed` — the installer's `operator_name` — or `asked`, the request's.
    pub named_by: &'static str,
}

/// **The id of the operator's first invite** — the installer's break-glass
/// token, recorded by its hash (`Accounts.ClaimOperator`'s
/// `bootstrap_invite_sha256`, the feed's `operator_invite_sha256`). One per
/// box: recording it again replaces it.
pub const OPERATOR_INVITE_ID: &str = "operator-bootstrap";

/// How long the operator's first invite stays redeemable from the moment it
/// is recorded — the most any invite lives. Every feed (an install, an
/// upgrade's re-admission) records it again while the operator holds no
/// passkey, so a box that waits longer than this for its owner is renewed by
/// its next feed.
pub const OPERATOR_INVITE_TTL_MS: i64 = 30 * 24 * 60 * 60 * 1000;

/// **What became of the operator's first invite.**
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BootstrapInvite {
    /// Recorded now (none was pending, or a different hash was).
    Recorded,
    /// The same hash was pending; its expiry is renewed.
    Renewed,
    /// The operator already holds a passkey: the bootstrap is spent, nothing
    /// is recorded, and a leftover one is withdrawn.
    Spent,
}

/// **The least a first invite may live**: a minute. A link that dies before
/// a person can open it is no way in.
pub const OPERATOR_INVITE_MIN_TTL_MS: i64 = 60 * 1000;

/// **A first invite's lifetime, checked** (private ownership, 2026-10-07):
/// at least [`OPERATOR_INVITE_MIN_TTL_MS`], at most
/// [`OPERATOR_INVITE_TTL_MS`] — the most any invite lives. `INVALID_ARGUMENT`
/// otherwise, by name; nothing is recorded.
pub fn bootstrap_ttl(ttl_ms: i64) -> ControlResult<i64> {
    if (OPERATOR_INVITE_MIN_TTL_MS..=OPERATOR_INVITE_TTL_MS).contains(&ttl_ms) {
        Ok(ttl_ms)
    } else {
        Err(ControlError::InvalidArgument(format!(
            "the first invite's lifetime is {ttl_ms} ms: it lives at least a minute and at most {} days",
            OPERATOR_INVITE_TTL_MS / 86_400_000
        )))
    }
}

/// A bootstrap hash as the wire carries it: 64 lowercase hex digits (what
/// `webauthn::sha256_hex` writes and an invite row keeps). `INVALID_ARGUMENT`
/// otherwise — a token sent in place of its hash is refused, never stored.
pub fn bootstrap_hash(raw: &str) -> ControlResult<String> {
    let hash = raw.trim();
    if hash.len() == 64 && hash.bytes().all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) {
        Ok(hash.to_owned())
    } else {
        Err(ControlError::InvalidArgument(
            "the bootstrap invite is the lowercase hex SHA-256 of the break-glass token (64 \
             characters), never the token itself"
                .into(),
        ))
    }
}

/// **Which name is the operator's**, the one rule both planes keep: the fed
/// name wins; the asked one only when nothing was fed; neither is
/// `FAILED_PRECONDITION`. Blank is absent.
pub fn operator_name<'a>(
    fed: Option<&'a str>,
    asked: &'a str,
) -> ControlResult<(&'a str, &'static str)> {
    match (fed.map(str::trim).filter(|n| !n.is_empty()), asked.trim()) {
        (Some(fed), _) => Ok((fed, "fed")),
        (None, asked) if !asked.is_empty() => Ok((asked, "asked")),
        _ => Err(ControlError::FailedPrecondition(
            "no operator is named for this box: the installer fed no `operator_name` and the \
             request named nobody"
                .into(),
        )),
    }
}

/// **One account, locked**, as `Accounts` reports it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AccountGate {
    /// A principal id, which is also the namespace that principal owns.
    pub account: String,
    pub reason: GateReason,
    /// A principal id for a person, a job id for a purge. Only a matching
    /// holder may lift its own gate.
    pub holder: String,
    /// The principal that asked, for the audit trail.
    pub actor: String,
    /// Shown to the locked-out user in the refusal they get.
    pub note: String,
    pub at_unix_ms: i64,
    /// **0 = no deadline.** Non-zero is a purge's lease, which is what stops a
    /// dead purge from bricking an account.
    pub expires_unix_ms: i64,
}

/// Where an account purge has got to. Mirrors `gunnar.v1.PurgeState`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PurgeState {
    /// Waiting its turn behind other accounts' purges.
    Queued,
    /// The one purge the server is running.
    Running,
    /// Finished, and everything it set out to reclaim is gone.
    Done,
    /// Finished badly. The account is PARTIAL and `failures` says where.
    Failed,
}

/// One queued or completed account purge, as a caller sees it.
#[derive(Debug, Clone)]
pub struct PurgeJob {
    /// Stable for the life of the job, and also the gate's `holder` while it
    /// runs.
    pub job_id: String,
    pub account: String,
    pub state: PurgeState,
    /// How many jobs are ahead of this one. 0 unless [`PurgeState::Queued`].
    pub queue_position: u32,
    /// Stores the walk found. **0 until the job starts** — counted when its
    /// turn comes, because an account can grow while a job waits.
    pub stores_found: u32,
    /// How many of those are gone.
    pub stores_purged: u32,
    pub bytes_reclaimed: u64,
    pub principal_removed: bool,
    /// **Whether the lockout this job took has been released.** The property
    /// that decides whether the account is usable again, on the wire rather
    /// than only in a log. False on a job that found an administrative gate
    /// already in place: it did not take that one and must not lift it.
    pub gate_released: bool,
    /// Only set when [`PurgeState::Failed`].
    pub error: Option<String>,
    /// One line per store that could not be reclaimed. The recovery list.
    pub failures: Vec<String>,
    pub enqueued_at_unix_ms: i64,
    /// 0 until the job leaves [`PurgeState::Queued`].
    pub started_at_unix_ms: i64,
    /// 0 until the job leaves [`PurgeState::Running`].
    pub finished_at_unix_ms: i64,
}

/// **What an enrolment asks** (`Accounts.EnrollEmployee`; owner, 2026-10-04,
/// "Add users"): ONE named employee holding `room_bytes`, `seats` unnamed
/// seats each holding it, or a seat claim (`from_seat`). The service has
/// already applied the rules a person types against (count range, names,
/// the room's default and floor); the plane keeps the rules about what is
/// STORED: a new principal only, an open seat for a claim, and the room.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct EmployeeAsk {
    /// The new principal's id and display name; both empty for seats.
    pub id: String,
    pub name: String,
    /// Bytes per employee or seat. Ignored for a claim: the seat decides.
    pub room_bytes: u64,
    /// Unnamed seats to grant; 0: one named employee (or a claim).
    pub seats: u32,
    /// Fill an open seat with the named employee.
    pub from_seat: bool,
    /// The account's room — its `pack_bytes` cap. `None`: nothing caps it on
    /// this server.
    pub cap: Option<u64>,
    /// What the account stores now (counted against `cap`).
    pub stored: u64,
    /// The order this belongs to, for its retry; empty: none.
    pub order: String,
    /// Ask, write nothing.
    pub dry_run: bool,
}

/// **The plane's answer** to an [`EmployeeAsk`]. A short account is an
/// answer, not an error, at this layer: `shortfall_bytes > 0` with
/// `unchanged`, nothing written — the SERVICE turns it into the refusal, in
/// one sentence for every plane.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct EmployeeAnswer {
    pub principal: Option<Principal>,
    pub room_bytes: u64,
    pub seats: u32,
    pub seats_open: u32,
    pub need_bytes: u64,
    pub unused_bytes: Option<u64>,
    pub shortfall_bytes: u64,
    pub unchanged: bool,
}

/// What `Accounts.Purge` was asked for, carried whole rather than as three
/// positional booleans nobody can read at a call site.
#[derive(Debug, Clone, Default)]
pub struct PurgeAccountOptions {
    /// The principal that asked. Recorded on the gate.
    pub principal_id: String,
    /// Why. Required at the edge; shown to the locked-out user.
    pub reason: String,
    /// Also remove the principal, its keys, grants, tokens and memberships.
    pub remove_principal: bool,
    /// Refuse unless the account holds no live repositories.
    pub expect_empty: bool,
}

/// The control plane's view of storage. See the module docs: this is a seam.
///
/// [`Identity`] is a supertrait, not an optional extra: a control plane that
/// cannot say who a key belongs to and whether they may act cannot be served
/// safely, and making that a compile error is cheaper than making it a runtime
/// hole.
pub trait ControlPlane: Identity + FirstBoot + Send + Sync + 'static {
    // ── namespaces (multi-tenancy) ─────────────────────────────────────────
    fn create_namespace(&self, name: &str, description: &str) -> ControlResult<Namespace>;
    fn list_namespaces(&self) -> ControlResult<Vec<Namespace>>;
    /// **One namespace, if it exists** — what the account settings and the
    /// outsider's view read (`Namespace.public`, `Namespace.events_public`).
    ///
    /// The default walks [`ControlPlane::list_namespaces`], which is right for
    /// the test double; the real adapter answers it with one catalog probe.
    fn namespace(&self, name: &str) -> ControlResult<Option<Namespace>> {
        Ok(self
            .list_namespaces()?
            .into_iter()
            .find(|namespace| namespace.name == name))
    }
    /// **`name` as an outsider may be shown it** — the namespace whose
    /// `public` / `events_public` its account admin opened, read for a caller
    /// who is NOT inside it. `None` when it does not exist, and also while the
    /// account or the caller is gated: the gate outranks the admin's setting
    /// exactly as it outranks `Visibility::Public` in `AuthTable::may`, so a
    /// public account that is being emptied is not listed and its events are
    /// not shown, and a locked-out principal is shown no public account at all.
    ///
    /// The default is [`ControlPlane::namespace`], which is right only for a
    /// double that cannot gate anybody (its `gate_account` is
    /// `Unimplemented`). An adapter that gates answers it with the gate read.
    fn namespace_for_outsider(&self, principal_id: &str, name: &str) -> ControlResult<Option<Namespace>> {
        let _ = principal_id;
        self.namespace(name)
    }
    /// **The account admin's two settings** (owner, 2026-10-03: namespaces and
    /// emitted events are not public by default, "such things has to be
    /// set"): is the namespace public, are its events public. `None` leaves a
    /// setting as it is. Who may (the account admin) is the service's question.
    fn set_namespace_settings(
        &self,
        name: &str,
        public: Option<bool>,
        events_public: Option<bool>,
    ) -> ControlResult<Namespace>;
    /// Returns how many repos went with it. Refuses a non-empty namespace
    /// unless `delete_repos`.
    fn delete_namespace(&self, name: &str, delete_repos: bool) -> ControlResult<u32>;

    // ── repos ──────────────────────────────────────────────────────────────
    fn create_repo(&self, spec: NewRepo) -> ControlResult<Repo>;

    /// **Has `namespace` room for a NEW repository, by its plan?** — the
    /// questions a push-to-create meets at the front door, asked for
    /// `Repos.Create` (owner 2026-10-06: the console's New repository is the
    /// account admin's, by the push's rules): the plan's verdict on a write,
    /// then whether the namespace is already at its cap. `machine`: the caller
    /// is a machine principal, which a plan does not bind unless it says so
    /// (as on the SSH door). The refusal is the plan's own line.
    ///
    /// Defaulted to `Ok(())`: a plane with no plan (the memory double, a
    /// community build) has room, which is what it answered before.
    fn create_room(&self, _namespace: &str, _machine: bool) -> ControlResult<()> {
        Ok(())
    }
    fn get_repo(&self, path: &RepoPath) -> ControlResult<Repo>;
    fn list_repos(&self, namespace: Option<&str>) -> ControlResult<Vec<Repo>>;
    /// **Delete a repository: a dated tombstone by default, a reclaim only
    /// when [`DeleteOptions::purge`] says so.**
    ///
    /// The signature carries the whole decision rather than a bare path
    /// because there is no safe default for the caller to omit: `purge` is the
    /// difference between "the name is free again in thirty days" and "181 GiB
    /// went in two seconds", and the principal is what the tombstone records
    /// so an undelete can say who deleted it.
    fn delete_repo(&self, path: &RepoPath, options: DeleteOptions) -> ControlResult<DeleteOutcome>;

    /// **Put a soft-deleted repository back**, inside its window. Returns the
    /// restored catalog row, with the recorded hash kind and default branch.
    fn undelete_repo(&self, path: &RepoPath) -> ControlResult<Repo> {
        let _ = path;
        Err(ControlError::Unimplemented("Repos.Undelete"))
    }

    /// **Reclaim a soft-deleted repository's bytes now.** Returns the bytes.
    fn purge_repo(&self, path: &RepoPath) -> ControlResult<u64> {
        let _ = path;
        Err(ControlError::Unimplemented("Repos.Purge"))
    }

    /// **Ask for a LIVE repository to be maintained** (`Repos.Gc`): seal the
    /// live logs, compact the archive, unlink the exploded read cache.
    ///
    /// **Returns when the job is ACCEPTED, not when the gc has run.** A gc is
    /// hundreds of thousands of unlinks and the box has one disk, so the plane
    /// runs one at a time and queues the rest; an RPC that blocked for the
    /// duration is what makes a console button feel broken. The returned
    /// [`GcJob`] is the one now responsible for the repository — a fresh one,
    /// or the one that was already queued or running, because a button gets
    /// pressed twice and twice is one gc.
    ///
    /// Named apart from [`ControlPlane::purge_repo`] because they are opposite
    /// in consequence and identical only in shape. A purge ends a retention
    /// window and there is no undo; this one reclaims what the server can
    /// re-derive, so the worst it costs is a slower next read.
    fn gc_repo(&self, path: &RepoPath) -> ControlResult<GcJob> {
        let _ = path;
        Err(ControlError::Unimplemented("Repos.Gc"))
    }

    /// **How a repository's gc is going** (`Repos.GcStatus`).
    ///
    /// `Ok(None)` is a real answer and not a stub: the plane is holding no job
    /// for this repository, because none was ever asked for or the last one
    /// finished long enough ago to have been dropped. A plane that cannot see
    /// the queue at all still answers `Unimplemented` — "nothing is running" and
    /// "I cannot tell" must not look alike.
    fn gc_status(&self, path: &RepoPath) -> ControlResult<Option<GcJob>> {
        let _ = path;
        Err(ControlError::Unimplemented("Repos.GcStatus"))
    }

    /// What is deleted and not yet purged. Empty is a legitimate answer here —
    /// unlike a ref list, "nothing is deleted" is a fact a caller can act on —
    /// but the default is still `Unimplemented`, because a control plane that
    /// cannot see tombstones must not report that there are none.
    fn list_deleted(&self, namespace: Option<&str>) -> ControlResult<Vec<DeletedRepoRow>> {
        let _ = namespace;
        Err(ControlError::Unimplemented("Repos.ListDeleted"))
    }
    /// **Every store directory with no catalog row and no soft-delete
    /// tombstone** (`Repos.ListOrphans`).
    ///
    /// Empty is a legitimate answer — "nothing is stranded" is a fact an
    /// operator can act on — but the default is still `Unimplemented`, for the
    /// same reason [`ControlPlane::list_deleted`]'s is: a control plane that
    /// cannot walk the disk must not report that the disk is clean.
    fn list_orphans(&self, namespace: Option<&str>) -> ControlResult<Vec<OrphanRow>> {
        let _ = namespace;
        Err(ControlError::Unimplemented("Repos.ListOrphans"))
    }

    /// **Reclaim an orphan's bytes** (`Repos.ReclaimOrphan`). Returns them.
    ///
    /// Refuses a catalogued or soft-deleted repository by name: each of those
    /// has its own verb, and a second door onto destroying a live repository
    /// is not a recovery tool.
    fn reclaim_orphan(&self, path: &RepoPath) -> ControlResult<u64> {
        let _ = path;
        Err(ControlError::Unimplemented("Repos.ReclaimOrphan"))
    }

    fn set_default_branch(&self, path: &RepoPath, default_branch: &str) -> ControlResult<Repo>;
    fn set_visibility(&self, path: &RepoPath, visibility: Visibility) -> ControlResult<Repo>;

    /// Turn a repository's "encrypted" flag on or off (`Repos.SetEncrypted`).
    /// The caller has already checked that the account admin asks.
    fn set_encrypted(&self, path: &RepoPath, encrypted: bool) -> ControlResult<Repo> {
        let _ = (path, encrypted);
        Err(ControlError::Unimplemented("Repos.SetEncrypted"))
    }

    /// A principal's SSH public keys as stored OpenSSH lines
    /// (`Repos.RecipientKeys`). The caller has already checked access.
    fn recipient_keys(&self, principal_id: &str) -> ControlResult<Vec<gunnar_proto::RecipientKey>> {
        let _ = principal_id;
        Err(ControlError::Unimplemented("Repos.RecipientKeys"))
    }

    // ── accounts: the gate, and the per-user hard purge ────────────────────

    /// **Lock `account` out, completely** (`Accounts.Gate`).
    ///
    /// Returns the gate as stored, and whether the account was **already**
    /// gated when the request arrived. Idempotent; the newest writer owns the
    /// row, because an operator must be able to suspend an account a purge is
    /// already holding.
    ///
    /// The default is [`ControlError::Unimplemented`] and not a silent success:
    /// an adapter that cannot lock anybody out must not answer "locked", or
    /// [`ControlPlane::purge_account`] would walk an account that is still
    /// live and a caller would be told the user was shut out when they were
    /// not.
    fn gate_account(
        &self,
        account: &str,
        actor: &str,
        note: &str,
    ) -> ControlResult<(AccountGate, bool)> {
        let _ = (account, actor, note);
        Err(ControlError::Unimplemented("Accounts.Gate"))
    }

    /// **Let `account` back in** (`Accounts.Ungate`). Returns what was lifted,
    /// or `None` if it was not gated.
    ///
    /// Unconditional — it does not consult the holder — because this is the
    /// operator's override, and an override that could be blocked by a lease
    /// belonging to a process that no longer exists is not an override.
    fn ungate_account(&self, account: &str) -> ControlResult<Option<AccountGate>> {
        let _ = account;
        Err(ControlError::Unimplemented("Accounts.Ungate"))
    }

    /// **Every account locked out right now** (`Accounts.ListAccountGates`).
    ///
    /// Empty is a legitimate answer; the default is still `Unimplemented`, for
    /// the reason [`ControlPlane::list_deleted`]'s is. A plane that cannot read
    /// the gate table must not report that nobody is locked out.
    fn account_gates(&self) -> ControlResult<Vec<AccountGate>> {
        Err(ControlError::Unimplemented("Accounts.ListAccountGates"))
    }

    /// **Empty an account: gate it NOW, enqueue the walk** (`Accounts.Purge`).
    ///
    /// Returns when the account is locked out and the job is accepted, **not**
    /// when the account is empty. The two halves are deliberately different in
    /// timing: the gate is synchronous so there is no window in which a purge
    /// has been accepted and the user can still push, and the walk is queued
    /// because it is minutes of unlink I/O and an RPC that held a tonic worker
    /// for it would make a console button look broken.
    ///
    /// The returned [`AccountGate`] is the one now in force — which may be an
    /// administrative gate that was already there rather than this job's lease.
    fn purge_account(
        &self,
        account: &str,
        options: PurgeAccountOptions,
    ) -> ControlResult<(PurgeJob, AccountGate)> {
        let _ = (account, options);
        Err(ControlError::Unimplemented("Accounts.Purge"))
    }

    /// **How an account's purge is going** (`Accounts.PurgeStatus`), and the
    /// lockout as it stands.
    ///
    /// `Ok((None, None))` is a real answer: no job is being held for this
    /// account and it is not gated. A plane that cannot see the queue answers
    /// `Unimplemented` instead — "nothing is running" and "I cannot tell" must
    /// not look alike.
    fn purge_account_status(
        &self,
        account: &str,
    ) -> ControlResult<(Option<PurgeJob>, Option<AccountGate>)> {
        let _ = account;
        Err(ControlError::Unimplemented("Accounts.PurgeStatus"))
    }

    /// **Write one repository's archive bytes above `from_generation` into
    /// `sink`** (`Repos.ArchiveOut`).
    ///
    /// The bytes are exactly what `gunnar repo archive-out --to FILE` writes,
    /// so what comes off the wire restores with `archive-in` and nothing new
    /// has to understand a second format.
    ///
    /// **Blocking, on purpose.** It reads files and writes a sink; the service
    /// runs it on the blocking pool rather than pretending it is async, which
    /// is the same shape `replicate_once` already uses for send-pack.
    ///
    /// The default is [`ControlError::Unimplemented`] and not an empty
    /// success: an archive stream that ends immediately looks exactly like a
    /// repository with no bytes, and a backup that silently archived nothing is
    /// the failure this whole surface exists to prevent.
    fn archive_out(
        &self,
        path: &RepoPath,
        from_generation: u64,
        sink: &mut dyn std::io::Write,
    ) -> ControlResult<ArchiveCut> {
        let _ = (path, from_generation, sink);
        Err(ControlError::Unimplemented("Repos.ArchiveOut"))
    }

    // ── refs and objects (the storage seam: RefStore / ObjectStore) ────────
    fn list_refs(&self, path: &RepoPath, prefix: &str, peel: bool)
        -> ControlResult<Vec<RefTarget>>;
    fn get_ref(&self, path: &RepoPath, ref_name: &str, peel: bool) -> ControlResult<RefTarget>;
    fn log(
        &self,
        path: &RepoPath,
        start: &str,
        limit: u32,
        touching: &str,
    ) -> ControlResult<Vec<Commit>>;
    /// [`ControlPlane::log`] walked from every branch and tag tip (and `HEAD`)
    /// at once, newest committer time first, at most `limit` commits.
    fn log_all_refs(
        &self,
        path: &RepoPath,
        limit: u32,
        touching: &str,
    ) -> ControlResult<Vec<Commit>> {
        let _ = (path, limit, touching);
        Err(ControlError::Unimplemented("Refs.Log all_refs"))
    }

    // ── pull requests ──────────────────────────────────────────────────────
    /// Open a pull request. `author` is the authenticated principal.
    fn create_pull(
        &self,
        path: &RepoPath,
        author: &str,
        title: &str,
        description: &str,
        source_ref: &str,
        target_ref: &str,
    ) -> ControlResult<PullRequest>;
    /// Every pull in one repository; `state` zero means all.
    fn list_pulls(&self, path: &RepoPath, state: i32) -> ControlResult<Vec<PullRequest>>;
    fn get_pull(&self, path: &RepoPath, number: u64) -> ControlResult<PullRequest>;
    fn close_pull(&self, path: &RepoPath, number: u64) -> ControlResult<PullRequest>;
    /// FAST-FORWARD ONLY. Moves the target ref through the store's CAS and
    /// returns what moved, so the service can publish the same `RefUpdated`
    /// a push would.
    ///
    /// # Why the caller's identity crosses this seam
    ///
    /// It did not, until GUN-004. `merge_pull(path, number)` moved a
    /// **protected** ref and the principal-dependent half of branch protection
    /// — `allowed_principals` — was impossible to evaluate on the far side,
    /// because the far side did not know who was asking. So `PullRequests.Merge`
    /// read the same policy document the push path reads, took
    /// `forbid_self_approval` and `required_approvals` out of it, and wrote the
    /// ref with `update_ref`: an explicitly configured per-ref protection,
    /// bypassed by a side door, while the push path refused correctly and the
    /// protection *looked* like it worked.
    ///
    /// `principal_id` is `Caller::principal_id`, the same string
    /// `ServiceContext::require` just checked a grant for.
    fn merge_pull(
        &self,
        path: &RepoPath,
        number: u64,
        principal_id: &str,
    ) -> ControlResult<(PullRequest, MergedRef)>;
    /// **A new repository's first commit, written by the server**
    /// (`CreateRepoRequest.readme`): `README.md` holding `readme` on the default
    /// branch. Answers the commit id (hex). A control plane that keeps no git
    /// store refuses by name.
    fn write_first_commit(&self, path: &RepoPath, readme: &str) -> ControlResult<String> {
        let _ = readme;
        Err(ControlError::FailedPrecondition(format!("this control plane writes no commit into {path}: it keeps no git store")))
    }
    fn read_object(
        &self,
        path: &RepoPath,
        oid: &str,
        max_bytes: u64,
    ) -> ControlResult<ReadObjectResponse>;

    /// **One LFS object, by the sha256 its pointer names** (`Refs.ReadLfsObject`).
    ///
    /// A file kept in LFS is a POINTER in the tree, so a caller that reads the
    /// tree gets 131 bytes of text where the file should be. This is the way
    /// to the bytes for a caller that already holds Read on the repository and
    /// speaks gRPC rather than one of `git lfs`'s transfer adapters.
    ///
    /// Whole object or nothing: a prefix of a git blob is a readable prefix of
    /// a file, and a prefix of an LFS object is a broken PDF. Past the ceiling
    /// it is refused, and the caller that wants it anyway wants `git lfs`.
    ///
    /// `range` is `Some((offset, length))` for a ranged read (the request's
    /// `offset`/`length`, not both 0): the bytes from `offset`, at most
    /// `length` (0 = to the ceiling), the ceiling applied to the range, an
    /// offset past the end refused. `None` is the whole-or-refused read.
    ///
    /// Defaulted to `Unimplemented` so a control plane that has no LFS store —
    /// the seams in this crate's tests — needs no stub.
    fn read_lfs_object(
        &self,
        path: &RepoPath,
        oid: &str,
        max_bytes: u64,
        range: Option<(u64, u64)>,
    ) -> ControlResult<ReadLfsObjectResponse> {
        let _ = (path, oid, max_bytes, range);
        Err(ControlError::Unimplemented("Refs.ReadLfsObject"))
    }
    fn diff(
        &self,
        path: &RepoPath,
        from: &str,
        to: &str,
        path_prefixes: &[String],
    ) -> ControlResult<Vec<FileChange>>;

    /// **Blame a file at a revision** (`Refs.Blame`).
    ///
    /// The walk itself lives in `gunnar-server`'s `blame` module — the ONE
    /// implementation the store-local CLI verb already uses (one implementation) — and it
    /// runs here against the store the running server already holds, so a
    /// no-shell appliance can be blamed over the control plane while the
    /// exclusive lock that verb takes makes it impossible to run down.
    ///
    /// The default is `Unimplemented`, not an empty blame: a control plane that
    /// cannot walk must refuse, because zero lines reads as "an empty file",
    /// which is a wrong answer a console reader would believe.
    fn blame(
        &self,
        path: &RepoPath,
        rev: &str,
        file: &str,
    ) -> ControlResult<gunnar_proto::BlameResponse> {
        let _ = (path, rev, file);
        Err(ControlError::Unimplemented("Refs.Blame"))
    }

    /// **Search one revision's files for a literal string** (`Refs.Search`).
    ///
    /// The same store-shared, walk-once-live implementation as [`blame`]: the
    /// `search` module in `gunnar-server`, reached against the running server's
    /// open store. `max_file_bytes` and `max_matches` of zero mean the module's
    /// defaults. Empty is a legitimate answer (a string that is not there), so
    /// the refusal below is the walk that could not START, never a miss.
    ///
    /// [`blame`]: ControlPlane::blame
    fn search(
        &self,
        path: &RepoPath,
        rev: &str,
        pattern: &str,
        max_file_bytes: u64,
        max_matches: u32,
    ) -> ControlResult<gunnar_proto::SearchResponse> {
        let _ = (path, rev, pattern, max_file_bytes, max_matches);
        Err(ControlError::Unimplemented("Refs.Search"))
    }

    // ── identity and access (the ACL model) ────────────────────────────────
    /// Create a principal with `id` as its identity (and account) and `name`
    /// as its display name. The service resolves an empty wire `id` to the
    /// name before calling, so this never sees an empty id.
    fn add_principal(&self, id: &str, name: &str, kind: PrincipalKind)
        -> ControlResult<Principal>;
    /// **Enroll an employee: a NEW user created by and bound to `account`**
    /// (`Accounts.EnrollEmployee`). `AlreadyExists` for an id or name that is
    /// taken — an existing principal never becomes an employee — and for an id
    /// that names an existing namespace (its owner-by-name would be the
    /// employee). `NotFound` when `account` is not a namespace. The employee
    /// gets no namespace of its own and `Read` on the account's.
    ///
    /// With room ("Add users"): see [`EmployeeAsk`]. `FailedPrecondition`
    /// for a claim with no open seat; a short account is answered with
    /// `shortfall_bytes` and nothing written.
    fn enroll_employee(&self, account: &str, ask: &EmployeeAsk) -> ControlResult<EmployeeAnswer>;
    /// The account's employees, by id.
    fn employees(&self, account: &str) -> ControlResult<Vec<Principal>>;
    /// **Has the account enrolled anyone?** What `Access.WhoAmI` asks on every call; a plane
    /// that can answer without listing (CoreControlPlane) says so.
    fn has_employees(&self, account: &str) -> ControlResult<bool> {
        Ok(!self.employees(account)?.is_empty())
    }
    /// The account's seats: every seat granted, and the open ones.
    fn seats(&self, account: &str) -> ControlResult<(u32, u32)> {
        let _ = account;
        Ok((0, 0))
    }
    /// **Clans** (2026-10-05, [`crate::clans::ClanPlane`]), when this plane
    /// keeps an identity table that holds them. `None`: every `Clans` verb is
    /// UNIMPLEMENTED and no namespace is a clan's.
    fn clans(&self) -> Option<&dyn crate::clans::ClanPlane> {
        None
    }
    /// **The estate's operator, created on first claim** (`Accounts.ClaimOperator`,
    /// 2026-10-04). `asked` is the caller's name for it, used only when nothing
    /// was fed ([`operator_name`]). Idempotent; an ownerless namespace of that
    /// name is adopted; a machine under the name is `FAILED_PRECONDITION`. The
    /// principal is created keyless and gunnar-admin. The default is the
    /// honest answer of a plane that keeps no operator.
    fn claim_operator(&self, asked: &str) -> ControlResult<OperatorClaim> {
        let _ = asked;
        Err(ControlError::Unimplemented("Accounts.ClaimOperator"))
    }
    /// **The operator's first invite** (Stage L, 2026-10-04): record
    /// `secret_sha256` (checked by [`bootstrap_hash`]) as the invite
    /// [`OPERATOR_INVITE_ID`] for `operator`'s first passkey — no server
    /// grant, expiring `ttl_ms` after `now` (checked by [`bootstrap_ttl`]:
    /// [`OPERATOR_INVITE_TTL_MS`] for the kit's own estates, shorter for a
    /// SOLD estate's first-login link) — unless the operator already holds a
    /// passkey ([`BootstrapInvite::Spent`]). The console's first-admin
    /// registration, or the buyer's first-login link, redeems it through
    /// `Access.WebRegister`.
    fn operator_bootstrap_invite(
        &self,
        operator: &str,
        secret_sha256: &str,
        now_unix_ms: i64,
        ttl_ms: i64,
    ) -> ControlResult<BootstrapInvite> {
        let _ = (operator, secret_sha256, now_unix_ms, ttl_ms);
        Err(ControlError::Unimplemented("Accounts.ClaimOperator (bootstrap invite)"))
    }
    fn list_principals(&self) -> ControlResult<Vec<Principal>>;
    /// **When each principal last signed in and last changed something**
    /// (2026-10-21), by principal id; a principal with no entry was never seen.
    /// The default knows nothing — a plane that keeps no stamps answers
    /// "never" for everybody, never an error.
    fn principal_activity(&self) -> ControlResult<std::collections::BTreeMap<String, PrincipalSeen>> {
        Ok(std::collections::BTreeMap::new())
    }
    /// One principal by its display name, or `None`.
    fn principal_by_name(&self, name: &str) -> ControlResult<Option<Principal>>;
    /// One principal by its ID, or `None` (`Access.WhoAmI`). The default walks
    /// [`ControlPlane::list_principals`]; a plane with an id index may answer
    /// it directly.
    fn principal(&self, id: &str) -> ControlResult<Option<Principal>> {
        Ok(self.list_principals()?.into_iter().find(|p| p.id == id))
    }
    /// **Does `principal_id` hold a SERVER-scope grant** (the seller's mark,
    /// `Namespace.owner_server_grant`)? The default: no grant is known here.
    fn holds_server_grant(&self, principal_id: &str) -> ControlResult<bool> {
        let _ = principal_id;
        Ok(false)
    }
    /// Change the display name and nothing else. `AlreadyExists` when another
    /// principal carries the name (as name or id); `NotFound` for an unknown id.
    fn rename_principal(&self, principal_id: &str, name: &str) -> ControlResult<Principal>;
    /// **Register a web credential on a principal** (`Access.AddCredential`,
    /// Stage L). `NotFound` for an unknown principal; `FailedPrecondition` for
    /// one that is not a USER (no human credential on a machine principal);
    /// `AlreadyExists` for a credential id already registered. The public key
    /// is validated by the service before it gets here. Removed with the
    /// principal. Defaults to `Unimplemented`: a plane that keeps no identity
    /// table keeps no passkeys either.
    fn add_credential(
        &self,
        principal_id: &str,
        credential_id: &[u8],
        public_key: &[u8],
        label: &str,
        sign_count: u32,
    ) -> ControlResult<gunnar_proto::WebCredential> {
        let _ = (principal_id, credential_id, public_key, label, sign_count);
        Err(ControlError::Unimplemented("Access.AddCredential"))
    }
    /// One principal's web credentials, or every one (`None`).
    fn credentials(&self, principal_id: Option<&str>) -> ControlResult<Vec<gunnar_proto::WebCredential>> {
        let _ = principal_id;
        Err(ControlError::Unimplemented("Access.ListCredentials"))
    }
    /// One web credential with its public key, or `None`.
    fn credential(&self, credential_id: &[u8]) -> ControlResult<Option<StoredCredential>> {
        let _ = credential_id;
        Err(ControlError::Unimplemented("Access.RevokeCredential"))
    }
    /// Change a web credential's label; the updated row, or `None` if there
    /// is no such credential.
    fn rename_credential(
        &self,
        credential_id: &[u8],
        label: &str,
    ) -> ControlResult<Option<gunnar_proto::WebCredential>> {
        let _ = (credential_id, label);
        Err(ControlError::Unimplemented("Access.RenameCredential"))
    }
    /// **Store an invite** (`Access.CreateInvite`, Stage L). `NotFound` for an
    /// unknown principal, `FailedPrecondition` for one that is not a USER,
    /// `AlreadyExists` for an id taken. Removed with the principal.
    fn create_invite(&self, invite: NewInvite) -> ControlResult<gunnar_proto::Invite> {
        let _ = invite;
        Err(ControlError::Unimplemented("Access.CreateInvite"))
    }
    /// Every stored invite, expired ones included — the service decides what
    /// is pending. Never a secret: the plane holds only hashes.
    fn invites(&self) -> ControlResult<Vec<gunnar_proto::Invite>> {
        Err(ControlError::Unimplemented("Access.ListInvites"))
    }
    /// Withdraw an invite; the row it was, or `None`.
    fn revoke_invite(&self, id: &str) -> ControlResult<Option<gunnar_proto::Invite>> {
        let _ = id;
        Err(ControlError::Unimplemented("Access.RevokeInvite"))
    }
    /// **What the invite with this secret hash is now** (`Access.CheckInvite`,
    /// 2026-10-19): valid, used, expired or unknown. Read-only: never spends it.
    fn check_invite(&self, secret_hash: &str, now_unix_ms: i64) -> ControlResult<InviteCheck> {
        let _ = (secret_hash, now_unix_ms);
        Err(ControlError::Unimplemented("Access.CheckInvite"))
    }
    /// **Commit a verified web login** (`Access.WebLogin`, Stage L): see
    /// [`WebLoginCommit`]. `NotFound` for a credential that is gone;
    /// `PermissionDenied` when its counter did not increase (re-checked in
    /// the update); `AlreadyExists` for a session key already registered.
    fn web_login(&self, commit: WebLoginCommit) -> ControlResult<WebLoginOutcome> {
        let _ = commit;
        Err(ControlError::Unimplemented("Access.WebLogin"))
    }
    /// **Register a passkey from a creation ceremony** (`Access.WebRegister`,
    /// Stage L): see [`WebRegistration`]. An invite that is unknown, used or
    /// expired — or names another principal — is `PermissionDenied`, in the
    /// same words for all of them.
    fn web_register(&self, registration: WebRegistration) -> ControlResult<WebRegistered> {
        let _ = registration;
        Err(ControlError::Unimplemented("Access.WebRegister"))
    }
    /// **Commit an approved machine enrolment** (2026-10-05): see
    /// [`MachineApproval`]. The ONE way a web credential lands on a MACHINE
    /// principal — `add_credential` and `web_register` still refuse one. A
    /// step-up counter that did not increase is `PermissionDenied`. The
    /// operator's first invite is never touched: it is the owner's.
    fn approve_machine_enrolment(&self, approval: MachineApproval) -> ControlResult<MachineApproved> {
        let _ = approval;
        Err(ControlError::Unimplemented("Access.ApproveMachineEnrolment"))
    }
    /// The account that enrolled `principal_id` as an employee, if any.
    fn employer_of(&self, principal_id: &str) -> ControlResult<Option<String>> {
        let _ = principal_id;
        Err(ControlError::Unimplemented("Accounts.EnrollEmployee"))
    }
    /// **Company SSO's storage** ([`crate::sso::plane::SsoPlane`]), when this
    /// plane has an identity table that holds it. `None`: every `Sso` verb
    /// answers UNIMPLEMENTED, and `Access.WebLogin` knows of no `require_sso`.
    fn sso(&self) -> Option<&dyn crate::sso::plane::SsoPlane> {
        None
    }
    /// Remove a web credential; the row it was, or `None` if there was none.
    fn revoke_credential(&self, credential_id: &[u8]) -> ControlResult<Option<gunnar_proto::WebCredential>> {
        let _ = credential_id;
        Err(ControlError::Unimplemented("Access.RevokeCredential"))
    }

    /// Returns (keys revoked, grants revoked).
    fn remove_principal(&self, principal_id: &str) -> ControlResult<(u32, u32)>;
    /// Register a key on `principal_id`.
    ///
    /// `caller` is the principal on the connection (empty under allow-all), and
    /// `attestation` the target's signed word that this key may act as them.
    /// The rule for when the attestation is required is
    /// `gunnar_core::key_attestation::Requirement::of`, one writer both planes
    /// ask; the plane derives the new key's fingerprint itself and checks the
    /// attestation against it before anything is written.
    #[allow(clippy::too_many_arguments)]
    fn add_key(
        &self,
        principal_id: &str,
        openssh_public_key: &str,
        comment: &str,
        expires_at_unix_ms: Option<i64>,
        caller: &str,
        attestation: Option<SignedKeyAttestation>,
    ) -> ControlResult<PublicKey>;
    fn revoke_key(&self, fingerprint: &str) -> ControlResult<()>;
    fn list_keys(&self, principal_id: Option<&str>) -> ControlResult<Vec<PublicKey>>;
    fn grant(
        &self,
        principal_id: &str,
        namespace: &str,
        repo: &str,
        level: GrantLevel,
    ) -> ControlResult<Grant>;
    fn revoke(&self, principal_id: &str, namespace: &str, repo: &str) -> ControlResult<()>;
    fn list_grants(
        &self,
        principal_id: Option<&str>,
        namespace: Option<&str>,
        repo: Option<&str>,
    ) -> ControlResult<Vec<Grant>>;

    // ── groups (punkt 9) ───────────────────────────────────────────────────
    //
    // A group is a principal (`add_principal` with `PrincipalKind::Group`);
    // these manage its membership. A member inherits the group's grants
    // through `may`, so the RPC gates each of these on Admin exactly as it gates
    // `grant`. Default bodies return `Unimplemented` so a control plane that has
    // no membership store (the in-memory test double) compiles unchanged.
    //
    // Add and Remove take the acting admin's `signed` event and, like
    // `forge_append`, verify it against the ACL key table before applying and
    // append it to the signed membership log afterwards — the verification
    // lives behind this seam so there is one implementation of "is this
    // evidence" and not one per transport (one implementation). They return the log index
    // the event landed at, the index an inclusion proof is taken at.
    /// Verify the acting admin's signed event, make `member_id` a member of
    /// `group_id`, and append the event to the membership log.
    fn add_group_member(
        &self,
        group_id: &str,
        member_id: &str,
        signed: SignedMembershipEvent,
    ) -> ControlResult<u64> {
        let _ = (group_id, member_id, signed);
        Err(ControlError::Unimplemented("Access.AddGroupMember"))
    }
    /// Verify the signed event, remove `member_id` from `group_id`, and append
    /// it. Returns whether it was a member and the log index.
    fn remove_group_member(
        &self,
        group_id: &str,
        member_id: &str,
        signed: SignedMembershipEvent,
    ) -> ControlResult<(bool, u64)> {
        let _ = (group_id, member_id, signed);
        Err(ControlError::Unimplemented("Access.RemoveGroupMember"))
    }
    /// The direct members of a group.
    fn list_group_members(&self, group_id: &str) -> ControlResult<Vec<Principal>> {
        let _ = group_id;
        Err(ControlError::Unimplemented("Access.ListGroupMembers"))
    }
    /// The signed membership log, optionally filtered to one group, in append
    /// order — the evidence behind the live table.
    fn membership_events(
        &self,
        group_id: Option<&str>,
    ) -> ControlResult<Vec<SignedMembershipEvent>> {
        let _ = group_id;
        Err(ControlError::Unimplemented("Access.ListGroupMembershipEvents"))
    }

    // ── policy (the pre-receive gate) ──────────────────────────────────────
    fn get_protection(
        &self,
        path: &RepoPath,
        ref_pattern: Option<&str>,
    ) -> ControlResult<Vec<BranchProtection>>;
    fn set_protection(
        &self,
        path: &RepoPath,
        protection: BranchProtection,
    ) -> ControlResult<BranchProtection>;
    fn delete_protection(&self, path: &RepoPath, ref_pattern: &str) -> ControlResult<()>;
    fn get_ref_name_policy(&self, path: &RepoPath) -> ControlResult<RefNamePolicy>;
    fn set_ref_name_policy(
        &self,
        path: &RepoPath,
        policy: RefNamePolicy,
    ) -> ControlResult<RefNamePolicy>;

    // ── forge: the review half (roadmap phase 1) ───────────────────────────
    //
    // The server never signs. `forge_append` takes an event the caller signed,
    // verifies it against the ACL key table, and appends it; the verification
    // lives behind this seam rather than in the service so that there is one
    // implementation of "is this evidence" and not one per transport (one implementation).
    //
    // There is no `next_pull` here: the number belongs to the record, and
    // `create_pull` assigns it. A second numbering authority is exactly the
    // twin this split exists to avoid.
    /// Verify a signed review event against the registered keys and append it.
    ///
    /// Returns the index it landed at — the index an inclusion proof is taken
    /// at. `InvalidArgument` for an event that does not verify, because a
    /// signature that does not check out is a malformed request and not a
    /// server fault.
    fn forge_append(&self, event: SignedForgeEvent) -> ControlResult<u64>;
    /// One pull request's review, folded, with every approval re-checked now.
    fn forge_review(
        &self,
        repo: &str,
        pull: u64,
    ) -> ControlResult<(ReviewState, Vec<SignedForgeEvent>)>;
    /// The review on every pull request in `repo` that has one.
    fn forge_reviews(&self, repo: &str) -> ControlResult<Vec<ReviewState>>;

    // ── forge: the issue half ──────────────────────────────────────────────
    //
    // The SAME log and the SAME `forge_append`: an issue event is signed,
    // verified against the ACL key table and appended exactly as a review event
    // is. What differs is only the fold — an issue owns its open/closed state
    // where a review defers it to the pull request record — so these read the
    // one log through a different fold rather than opening a second store.
    /// One issue, folded from its events, or `None` if nothing opened that
    /// number. The whole thread is returned alongside — the opening, every
    /// comment, and the close if there is one.
    fn forge_issue(
        &self,
        repo: &str,
        number: u64,
    ) -> ControlResult<Option<(IssueState, Vec<SignedForgeEvent>)>>;
    /// Every issue in `repo`, folded, in the order they were opened.
    fn forge_issues(&self, repo: &str) -> ControlResult<Vec<IssueState>>;

    // ── admin ──────────────────────────────────────────────────────────────
    fn counts(&self) -> ControlResult<ControlCounts>;

    /// **Tombstones and replication, for `Admin.Stats`.**
    ///
    /// The default is the honest empty one and NOT `Unimplemented`: unlike a
    /// ref list, "no tombstones and no standbys" is a real state a caller can
    /// act on, and a `Stats` that failed because a test double has no
    /// replicator would make the whole call unusable.
    fn stats_extras(&self) -> ControlResult<StatsExtras> {
        Ok(StatsExtras::default())
    }
    fn storage_info(&self, path: Option<&RepoPath>) -> ControlResult<StorageInfoResponse>;

    /// **"Which CA does this box mint with?"** (`Admin.IssuerInfo`).
    ///
    /// A read of the PKI control document: the live issuer's certificate and
    /// fingerprint, and the trusted-CA list. `gunnar-server` answers it
    /// through [`ServerStores::pki_readonly`], whose secret store refuses
    /// every call — so this seam cannot mint, cannot rotate, and cannot reach
    /// the private key even if someone later edits the body of the one
    /// implementation.
    ///
    /// The default is `Unimplemented` and NOT `has_issuer: false`, for the
    /// reason [`Self::capacity`] gives above: a control plane with no PKI
    /// behind it has nothing to read, and "this box has no issuer" is a
    /// SPECIFIC and actionable fact — it is what a never-served data root
    /// looks like. A caller told that when the truth is "I cannot say" would
    /// go and mint one over the top of a CA their whole fleet pins.
    fn issuer_info(&self) -> ControlResult<gunnar_proto::IssuerInfoResponse> {
        Err(ControlError::Unimplemented("Admin.IssuerInfo"))
    }

    /// **"Is the record of what happened here still complete?"**
    /// (`Admin.AuditTrail`, lane T11, 2026-09-21).
    ///
    /// The shape of the control plane's tamper-evident audit chain — entry
    /// count, tip hash, whether it verifies, whether it is replicated into the
    /// transaction log, and optionally the hash at one sequence number. Never
    /// an entry body: this answers completeness, not contents.
    ///
    /// The default is `Unimplemented` and NOT `present: false`, for the reason
    /// [`Self::issuer_info`] gives: "this box keeps no trail" is a specific and
    /// alarming fact, and a caller told it when the truth is "I cannot say"
    /// would report a box as unaudited that is not.
    fn audit_trail(&self, _at_seq: u64) -> ControlResult<gunnar_proto::AuditTrailResponse> {
        Err(ControlError::Unimplemented("Admin.AuditTrail"))
    }

    /// **"If a tenant's cap rose to N bytes, could you serve it?"**
    /// (`Admin.Capacity`) — the question a seller must ask BEFORE the money
    /// moves, and OPEN SOURCE, ALWAYS ON: a self-hoster wants to know whether
    /// their disk can grow as much as a service provider does, so this is not
    /// behind `metered`.
    ///
    /// Not "how much free space is there". The answer carries a verdict AND,
    /// independently, whether the data set can ever hold more than it does —
    /// because on the appliance the second answer is NO and a caller that read
    /// a single small number would buy a disk that is billed and invisible.
    /// `gunnar-server`'s `capacity` module is the ONE implementation (one implementation);
    /// it measures with the same `statvfs` the push door refuses under.
    ///
    /// The default is `Unimplemented`, not a plausible answer: a control plane
    /// with no filesystem behind it has nothing to measure, and "0 bytes free"
    /// from it would read as a full disk rather than as an unanswered
    /// question — which is exactly the blind guard to avoid.
    fn capacity(
        &self,
        _request: &gunnar_proto::CapacityRequest,
    ) -> ControlResult<gunnar_proto::CapacityResponse> {
        Err(ControlError::Unimplemented("Admin.Capacity"))
    }

    // ── The runtime data-set growth (`Admin.Grow*`, `Bootstrap.GrowStatus`) ──
    //
    // `DATA-SET-GROWTH-FLOW.md` §3, ruling 1. `gunnar-server`'s `grow` module
    // is the ONE implementation; a control plane without it answers an honest
    // `idle` with `engine_present: false` on status — never `Unimplemented`,
    // because the credential-free status is what an outside actor reads to
    // decide about the volumes, and "I cannot say" must not look like "ask
    // again later" — and refuses the two mutators by name.

    /// Begin a runtime growth: drain, unmount, wait, grow, mount, reopen.
    fn grow_start(
        &self,
        _request: &gunnar_proto::GrowStartRequest,
    ) -> ControlResult<gunnar_proto::GrowStatusResponse> {
        Err(ControlError::Unimplemented(
            "Admin.GrowStart: this control plane has no runtime grow engine (gunnar-server built without `grow-runtime`)",
        ))
    }

    /// The growth's phase, members and journal. Answered with and without a
    /// credential (`Admin.GrowStatus`, `Bootstrap.GrowStatus`).
    fn grow_status(&self) -> ControlResult<gunnar_proto::GrowStatusResponse> {
        Ok(gunnar_proto::GrowStatusResponse {
            phase: "idle".to_owned(),
            engine_present: false,
            serving: true,
            server_unix_ms: crate::bus::now_unix_ms(),
            detail: "this control plane has no runtime grow engine; the data set grows only at a restart, if at all".to_owned(),
            ..Default::default()
        })
    }

    /// The outside actor says "look again" (or the operator gives up).
    fn grow_resume(&self, _give_up: bool) -> ControlResult<gunnar_proto::GrowStatusResponse> {
        Err(ControlError::Unimplemented(
            "Admin.GrowResume: this control plane has no runtime grow engine (gunnar-server built without `grow-runtime`)",
        ))
    }

    /// **Per-namespace meters** (`Usage.Read` / `Usage.ReadAll`), the numbers a
    /// bill is made of — and OPEN SOURCE, ALWAYS ON: a self-hoster wants them
    /// as much as a service provider does, so this is not behind `metered`.
    ///
    /// `Some(ns)` answers one namespace and is `NotFound` for one the catalog
    /// does not hold; `None` answers every namespace, including ones with no
    /// repositories (`repos = 0`, everything else 0) — a tenant with nothing
    /// is a real row, not an absent one. What each meter counts, file by
    /// file, is written in `proto/gunnar/v1/usage.proto`; the ONE
    /// implementation is `gunnar-server`'s (one implementation), which measures the same
    /// directories `Admin.StorageInfo` walks, attributed per file.
    ///
    /// The default is `Unimplemented`, not an empty answer: a control plane
    /// with no store behind it has no bytes to count, and "0 bytes" from it
    /// would read as a free tenant rather than as an unanswered question.
    fn usage(&self, namespace: Option<&str>) -> ControlResult<Vec<gunnar_proto::NamespaceUsage>> {
        let _ = namespace;
        Err(ControlError::Unimplemented("Usage"))
    }

    /// **One namespace's repositories, each with what it stores**
    /// (`Usage.Read`'s `repos`, 2026-10-04) — measured by the same walk that
    /// sums [`ControlPlane::usage`], so the rows' `stored_bytes` add up to the
    /// namespace's `pack_bytes`. Sorted by repository name. `NotFound` for a
    /// namespace the catalog does not hold; the default is `Unimplemented`, for
    /// [`ControlPlane::usage`]'s reason.
    fn repo_usage(&self, namespace: &str) -> ControlResult<Vec<gunnar_proto::RepoStored>> {
        let _ = namespace;
        Err(ControlError::Unimplemented("Usage"))
    }

    /// **One page of every namespace's meters** (`UsageBatch.ReadMany`,
    /// 2026-10-06), in namespace order, at most `page_size` rows, resuming
    /// after `page_token` ([`usage_page_token`]); with a `changed_since` cursor
    /// it may answer only the namespaces whose row changed.
    ///
    /// The default is CORRECT and not cheap: it pages over
    /// [`ControlPlane::usage`]`(None)` and never honours `changed_since`
    /// (`changed_only = false`, no cursor) — every plane that can read usage can
    /// page it. `gunnar-server` overrides it with its usage ledger, which is the
    /// whole point of the call.
    fn usage_page(
        &self,
        page_token: &str,
        page_size: usize,
        changed_since: &str,
    ) -> ControlResult<gunnar_proto::ReadManyUsageResponse> {
        let _ = changed_since;
        let (_, after) = usage_page_token::decode(page_token);
        let mut rows = self.usage(None)?;
        rows.sort_by(|a, b| a.namespace.cmp(&b.namespace));
        Ok(usage_page_token::page(rows, &after, page_size, String::new(), false))
    }

    /// **The FLOW series** (`Usage.Activity`): one row per UTC day this
    /// namespace moved anything, oldest first — and OPEN SOURCE, ALWAYS ON,
    /// for [`ControlPlane::usage`]'s reason.
    ///
    /// The window is inclusive at both ends and resolved to whole UTC days;
    /// `0` for `since_unix_ms` means the whole kept window, `0` for
    /// `until_unix_ms` means now. A day with no traffic is ABSENT rather than
    /// returned as a row of zeroes: the caller knows which days it asked for,
    /// and a day that was never written and a day on which nothing happened
    /// are the same fact.
    ///
    /// The second half of the answer is how many days this server keeps, which
    /// is what lets a caller tell an empty series that means *nothing
    /// happened* from one that means *you asked about a time this box no
    /// longer remembers*.
    ///
    /// The default is `Unimplemented`, not an empty series — for exactly the
    /// reason [`ControlPlane::usage`]'s is. A control plane with no flow ledger
    /// behind it counted nothing, and "no days" from it would read as a silent
    /// tenant rather than as an unanswered question. That distinction is the
    /// whole point of this call.
    fn activity(
        &self,
        namespace: &str,
        since_unix_ms: i64,
        until_unix_ms: i64,
    ) -> ControlResult<(Vec<gunnar_proto::DayActivity>, i64)> {
        let _ = (namespace, since_unix_ms, until_unix_ms);
        Err(ControlError::Unimplemented("Usage.Activity"))
    }

    /// **The admission plan for a namespace** (`Usage.AdmitPlan`): its usage
    /// over the rolling window, the budget, and per kind of work the class
    /// the gate would give the next transfer, the learnt estimate and the
    /// plan's sentence. Open source and always on — a build without
    /// `admit-share` answers `pass-through`, which is the truth about it.
    ///
    /// The default is `Unimplemented`, for [`ControlPlane::usage`]'s reason: a
    /// control plane with no admission gate behind it has no plan to report.
    fn admit_plan(&self, namespace: &str) -> ControlResult<gunnar_proto::AdmitPlanResponse> {
        let _ = namespace;
        Err(ControlError::Unimplemented("Usage.AdmitPlan"))
    }

    // ── the METERED appliance: `Entitlement` ────────────────────────────────
    //
    // Compiled only with the `metered` feature, and then the defaults are a
    // refusal BY NAME — the in-memory test double has no monetize key to
    // verify a fact against and no control document to keep it in. Off, these
    // methods do not exist: a `ControlPlane` with no `entitlement_set` is the
    // whole point (the metered build split), not a stub that returns
    // `Unimplemented` from a binary that still carries the code.

    /// **The fact being ENFORCED for one namespace**, or `None` = Free.
    ///
    /// # The contract, and the one implementation that does not keep it
    ///
    /// "Enforced", not "stored". Two things read a namespace's plan on this
    /// box and they read different places:
    ///
    /// * `gunnar_ssh::PlanGate` — the push door, `Entitlement.Set`'s whole
    ///   purpose — asks the CACHE (`EntitlementCache::allows`), which holds
    ///   only facts that verified under the configured monetize key;
    /// * this method, on `gunnar-server`, read the STORE until 2026-09-21
    ///   (`EntitlementSlot::get` → `EntitlementStore::get`), which holds
    ///   whatever was written, unverified on the way out. **It now reads the
    ///   cache** — the fix described below is made, and
    ///   `gunnar-server/tests/entitlement.rs`
    ///   `a_box_restored_under_the_wrong_key_reports_no_plan_it_does_not_enforce`
    ///   pins it. The rest of this note is the reason, kept.
    ///
    /// They agreed on every path except one, and it is a path that runs on
    /// every restart. `EntitlementSlot::open` loads each stored fact and pushes
    /// it into the cache **through `monetize_embed::signing::verify_fact`**; a
    /// fact that no longer verifies — the monetize key was rotated, the control
    /// document was edited — is logged (`entitlement.boot.refused`) and left
    /// out, and that namespace is Free for everything the cache answers.
    ///
    /// The store still has it. So `Usage.Read` reports its caps as the
    /// namespace's `Allowance`, `Entitlement.Get` returns it, and
    /// `Entitlement.List` lists it, while NOTHING on the box enforces any of
    /// it. That is precisely the defect `Allowance.enforcement` was added to
    /// close, coming back through a different door and worse: there the tenant
    /// was shown a measured number labelled as a limit, here they are shown a
    /// `pack_bytes` cap labelled `ENFORCED` — by
    /// [`crate::service::meters::METERS`], correctly, since that table
    /// describes the CODE — with no fact in force behind it.
    ///
    /// **A rejected fact must not be reported as an allowance.** The fix is in
    /// the implementation and not here: `EntitlementSlot::get` must answer out
    /// of the same snapshot `PlanGate` reads, so that a fact refused at boot is
    /// `None` to both. Reporting a plan the box will not act on is a promise
    /// nobody keeps, and it is silent — the only trace today is one
    /// `tracing::error!` at startup.
    ///
    /// Refusing to serve `Usage.Read` at all in that state would be the wrong
    /// repair: the METERS are still true, and a tenant whose signature broke
    /// still deserves to see what they hold.
    #[cfg(feature = "metered")]
    fn entitlement_get(
        &self,
        namespace: &str,
    ) -> ControlResult<Option<gunnar_proto::EntitlementFact>> {
        let _ = namespace;
        Err(ControlError::Unimplemented("Entitlement.Get"))
    }

    /// Every fact, one per namespace.
    #[cfg(feature = "metered")]
    fn entitlement_list(&self) -> ControlResult<Vec<gunnar_proto::EntitlementFact>> {
        Err(ControlError::Unimplemented("Entitlement.List"))
    }

    /// The stored facts this box refused when it booted (`Entitlement.List`'s
    /// `refused_at_boot`). Default `None`: nothing refused, or nothing held.
    #[cfg(feature = "metered")]
    fn entitlement_refused_at_boot(&self) -> ControlResult<Option<gunnar_proto::RefusedAtBoot>> {
        Ok(None)
    }

    /// **Verify, log, store, swap.** Returns the state now enforced and the
    /// sequence number (line index) of the attestation line the accepted
    /// fact was appended as. A fact that does not verify is
    /// [`ControlError::PermissionDenied`] naming why, and NOTHING moves — not
    /// the snapshot, not the document, not the log.
    #[cfg(feature = "metered")]
    fn entitlement_set(
        &self,
        fact: gunnar_proto::EntitlementFact,
    ) -> ControlResult<(gunnar_proto::EntitlementState, u64)> {
        let _ = fact;
        Err(ControlError::Unimplemented("Entitlement.Set"))
    }

    // ── the METERED appliance: `Billing` ────────────────────────────────────
    //
    // Two methods and a deliberate shape: this side of the seam holds the
    // KEY, and the other side holds the ticket FORMAT.
    //
    // `crate::service::billing` builds the ticket, asks here for a signature
    // over bytes, and pastes the result in. It is never handed key material.
    // The alternative — a seam method that returns the private key — would put
    // a signing key into a crate whose job is to parse requests, and would add
    // a second place on this box that can produce the appliance's signature
    // over anything at all. The alternative in the other direction — a seam
    // method that mints the whole ticket — would move the format into
    // `gunnar-server` and leave the grant check and the thing it guards in two
    // different crates.

    /// **The public half of the actor-ticket key, 64 lowercase hex characters**
    /// (`Billing.ApplianceIdentity`).
    ///
    /// The default is a refusal BY NAME rather than a fabricated key: a control
    /// plane that cannot reach a secret store has no key, and answering with
    /// something that looks like one would be advertising a key that verifies
    /// nothing this box will ever sign.
    #[cfg(feature = "metered")]
    fn actor_ticket_public_key_hex(&self) -> ControlResult<String> {
        Err(ControlError::Unimplemented("Billing.ApplianceIdentity"))
    }

    /// **Sign `message` with the actor-ticket key** — raw Ed25519, 64 bytes, no
    /// envelope.
    ///
    /// `message` is `monetize_embed::ticket::ticket_message` of an unsigned
    /// ticket, and this method has no opinion about that: it takes bytes,
    /// because the half that holds the key is not the half that knows the
    /// format, and an opinion here would be a second place the format is
    /// written down.
    ///
    /// **Every call to this is a ticket that exists.** The caller must have
    /// passed its grant check before it gets here — `service::billing` does,
    /// and `tests/billing.rs` counts the calls to prove a refused request makes
    /// none.
    #[cfg(feature = "metered")]
    fn actor_ticket_sign(&self, message: &[u8]) -> ControlResult<[u8; 64]> {
        let _ = message;
        Err(ControlError::Unimplemented("Billing.MintActorTicket"))
    }

    /// **The named sealed slot `name`** (2026-10-09): its owner's principal id
    /// and its value, or `None` when there is no such slot. The ACL is the
    /// service's (`service::sealed`); the plane stores and reads.
    #[cfg(feature = "metered")]
    fn sealed_get(&self, name: &str) -> ControlResult<Option<(String, Vec<u8>)>> {
        let _ = name;
        Err(ControlError::Unimplemented("Sealed.Get"))
    }

    /// **Store `value` in the slot `name`, owned by `owner`** (a principal id):
    /// sealed under the KEK, one `Secret` frame. Answers whether it was created.
    #[cfg(feature = "metered")]
    fn sealed_put(&self, name: &str, owner: &str, value: &[u8]) -> ControlResult<bool> {
        let _ = (name, owner, value);
        Err(ControlError::Unimplemented("Sealed.Put"))
    }

    /// **Walk the provenance chain** (`Admin.VerifyChain`, roadmap phase 2).
    ///
    /// The walk itself lives in `gunnar-server`'s `verify` module — the one
    /// implementation the CLI verb already uses (one implementation) — so the default here
    /// is `Unimplemented`, not an empty chain: a control plane that cannot
    /// walk the chain must refuse, because an empty chain with no BROKEN link
    /// reads as merely incomplete, and "the backend cannot verify anything"
    /// must never be dressed up as "nothing was disproven".
    fn verify_chain(
        &self,
        request: &gunnar_proto::VerifyChainRequest,
    ) -> ControlResult<gunnar_proto::VerifyChainResponse> {
        let _ = request;
        Err(ControlError::Unimplemented("Admin.VerifyChain"))
    }

    /// Index work in flight. Default: none — an engine without a background
    /// indexer has nothing to report, and that is not an error.
    fn index_progress(&self) -> ControlResult<Vec<IndexProgressRow>> {
        Ok(Vec::new())
    }

    // ── mirror (mirroring) ─────────────────────────────────────────────────
    fn mirror_add(
        &self,
        path: &RepoPath,
        remote_url: &str,
        prune: bool,
        create_repo: bool,
    ) -> ControlResult<MirrorConfig>;
    /// **Enqueue a sync, and return without syncing.**
    ///
    /// Returns the registration with [`MirrorConfig::job`] set to the job now
    /// responsible for the repository — a new one, or the one that was already
    /// queued or running for it. It does NOT return what moved: a fetch is
    /// minutes of somebody else's network and a unary RPC that waited for one
    /// would time out. Poll `mirror_status` until the job is DONE.
    ///
    /// It used to return `(MirrorConfig, Vec<RefUpdate>)` and to be
    /// `Unimplemented` in every implementation, so nothing ever read that pair.
    fn mirror_sync(&self, path: &RepoPath) -> ControlResult<MirrorConfig>;
    /// **Store, replace or forget the credential this mirror dials with.**
    ///
    /// `secret` is the value and is **write-only**: nothing returns it, and the
    /// `MirrorConfig` that comes back says only which kind is set. A `kind` of
    /// `Unspecified` forgets it, and is the only spelling of that; see
    /// `mirror.proto` for why this is a verb of its own rather than two more
    /// fields on `mirror_add`.
    fn mirror_set_credential(
        &self,
        path: &RepoPath,
        kind: MirrorCredentialKind,
        secret: &str,
        username: &str,
    ) -> ControlResult<MirrorConfig>;
    fn mirror_status(
        &self,
        namespace: Option<&str>,
        path: Option<&RepoPath>,
    ) -> ControlResult<Vec<MirrorConfig>>;
    fn mirror_remove(&self, path: &RepoPath) -> ControlResult<()>;

    // ── replication: the OPERATOR half ─────────────────────────────────────
    //
    // Default bodies, and the default is a REFUSAL BY NAME. Two callers cannot
    // do this: a build without `gunnar-server`'s `replication` feature, and the
    // in-memory test double, which has no replicator to reconfigure. Both must
    // say so out loud — an empty `Ok` here would be a control plane that
    // accepts a standby and silently keeps none, which is the exact shape of
    // the defect replication's whole file is written against.
    //
    // They exist at all because the appliance has NO SHELL. `replication.json`
    // is the truth, and until these three verbs there was no way to write it on
    // an OS-less image that runs as PID 1.

    /// Add or replace one standby by name, persist it, and start replicating.
    fn replication_set(
        &self,
        _target: ReplicationTarget,
    ) -> ControlResult<ReplicationStatusResponse> {
        Err(ControlError::Unimplemented(
            "this server was built without replication (`--features replication`), so it              cannot be given a standby",
        ))
    }
    /// Drop one standby by name, persist that, and stop replicating to it.
    fn replication_remove(&self, _name: &str) -> ControlResult<ReplicationStatusResponse> {
        Err(ControlError::Unimplemented(
            "this server was built without replication (`--features replication`), so it              has no standby to remove",
        ))
    }
    /// Drop EVERY standby at once — the failover demote. Persist the empty
    /// roster and stop replicating anywhere.
    fn replication_demote(&self) -> ControlResult<ReplicationStatusResponse> {
        Err(ControlError::Unimplemented(
            "this server was built without replication (`--features replication`), so it              has no standby to demote",
        ))
    }
    /// **Set this box's role**: `STANDBY` freezes client writes (a read-only
    /// standby); `PRIMARY` is a writable primary — the takeover flip; `BROWSE`
    /// is a frozen standby that serves anonymous reads and can never be
    /// promoted (the browse replica) — an implementation refuses
    /// `PRIMARY` on a browse box with `FailedPrecondition`. Persist it and
    /// flip the shared write-gate the front doors read.
    fn replication_set_role(
        &self,
        _role: gunnar_proto::ReplicationRole,
    ) -> ControlResult<ReplicationStatusResponse> {
        Err(ControlError::Unimplemented(
            "this server was built without replication (`--features replication`), so it              has no role to set",
        ))
    }
    /// **Abdicate**: freeze client writes, flush pending to the standbys, then
    /// demote — the master steps down cleanly, in that order.
    fn replication_abdicate(&self) -> ControlResult<ReplicationStatusResponse> {
        Err(ControlError::Unimplemented(
            "this server was built without replication (`--features replication`), so it              cannot abdicate",
        ))
    }
    /// What is configured, and what each running standby has done.
    ///
    /// `include_refs` asks for the expensive half — the count of refs a mirror
    /// does not carry, which opens every repository's reference database. False
    /// (a plain `replication list`) answers the cheap operational signal and
    /// opens no store; true (`--refs`) takes the walk.
    fn replication_status(
        &self,
        _include_refs: bool,
    ) -> ControlResult<ReplicationStatusResponse> {
        Err(ControlError::Unimplemented(
            "replication: this server was built without the `replication` feature",
        ))
    }
    /// **Backfill**: queue every repository one standby covers, or one named
    /// repository, so a standby added AFTER the repositories were written gets
    /// them. Adding a standby enqueues nothing.
    fn replication_resync(
        &self,
        _name: &str,
        _repo: Option<&RepoPath>,
        _except: &[RepoPath],
    ) -> ControlResult<ResyncReplicationResponse> {
        Err(ControlError::Unimplemented(
            "replication: this server was built without the `replication` feature, so it              has no standby to backfill",
        ))
    }

    /// **The replication watermark** (`Mirror.FlushReplication`,
    /// `DATA-SET-GROWTH-FLOW.md` G2/T11): push everything pending to every
    /// standby, wait — bounded — and report the tuple that proves (or does
    /// not) that the twin is caught up. Changes no role and no roster.
    fn replication_flush(
        &self,
        _request: &gunnar_proto::FlushReplicationRequest,
    ) -> ControlResult<gunnar_proto::FlushReplicationResponse> {
        Err(ControlError::Unimplemented(
            "replication: this server was built without the `replication` feature, so it has no standby to flush to",
        ))
    }

    // ── The transaction log: what the twin fetches ──────────────────────────

    /// The log's head — `head_seq`, `first_seq`, bytes, and on a twin what it
    /// has applied. Default `Unimplemented` for a plane with no log.
    fn log_head(&self) -> ControlResult<gunnar_proto::LogHeadResponse> {
        Err(ControlError::Unimplemented(
            "transaction log: this control plane keeps none",
        ))
    }

    /// Entries `from..=to`, whole, at most `max_bytes` of blob in total but
    /// never fewer than one entry. `from` below the oldest retained sequence
    /// is `FailedPrecondition` naming it.
    fn log_fetch(
        &self,
        _from: u64,
        _to: u64,
        _max_bytes: u64,
    ) -> ControlResult<Vec<gunnar_proto::LogEntry>> {
        Err(ControlError::Unimplemented(
            "transaction log: this control plane keeps none",
        ))
    }

    /// **[`ControlPlane::log_fetch`], streamed** (twin-streaming, 2026-10-06): the same
    /// refusals, answered NOW — before the call's first piece, so `Mirror.LogFetch` keeps its
    /// status codes (`FailedPrecondition` for a range retention dropped, and the rest) — then a
    /// producer that hands the range to its sink as `LogChunk` pieces of at most
    /// [`LOG_PIECE_BYTES`], `meta` on an entry's first, `last` on its last. The service runs it
    /// on a blocking thread behind a few pieces of window, so the sink's backpressure paces it.
    ///
    /// The default gathers through `log_fetch` and cuts each entry into pieces, which is right
    /// for a plane that holds its entries in memory anyway. A plane that keeps a real log
    /// overrides it and reads each entry off the disk a piece at a time: a 512 MiB pack must not
    /// cost the primary 512 MiB (it cost twice that before: the entry, then its pieces).
    fn log_fetch_stream(&self, from: u64, to: u64, max_bytes: u64) -> ControlResult<LogFetchProducer> {
        let entries = self.log_fetch(from, to, max_bytes)?;
        Ok(Box::new(move |sink| {
            for e in entries {
                log_entry_pieces(e, sink)?;
            }
            Ok(())
        }))
    }

    /// **[`ControlPlane::log_fetch_stream`] with the FIRST entry started at byte `from_offset`
    /// of its blob** (`LogFetchRequest.from_offset`, restore-large-entry 2026-10-07): a restore
    /// resuming an entry the box being filled holds part of. Every piece says its `offset`.
    /// 0 is [`ControlPlane::log_fetch_stream`] exactly. The default gathers through
    /// `log_fetch` and cuts the first entry from the offset; a plane with a real log overrides it,
    /// reads the blob from the offset only and stamps the stored crc on the last piece for the
    /// receiver to check.
    fn log_fetch_stream_at(&self, from: u64, from_offset: u64, to: u64, max_bytes: u64) -> ControlResult<LogFetchProducer> {
        if from_offset == 0 {
            return self.log_fetch_stream(from, to, max_bytes);
        }
        let entries = self.log_fetch(from, to, max_bytes)?;
        Ok(Box::new(move |sink| {
            for e in entries {
                let at = if e.seq == from { from_offset } else { 0 };
                log_entry_pieces_from(e, at, sink)?;
            }
            Ok(())
        }))
    }

    /// **On a twin: its primary's heartbeat** (when the primary last answered),
    /// with the silence bound `silent_after_ms`. Default `None`: not a twin.
    fn primary_heartbeat(&self, _silent_after_ms: u64) -> Option<gunnar_proto::PrimaryHeartbeat> {
        None
    }

    /// **What `Admin.Topology` needs from the plane** (2026-10-04): the
    /// pairing feed's peers, which principals run on the front, the plugin
    /// registration, this box's own address and data volume. Default: a box
    /// that knows only itself — which is the truthful answer for a plane with
    /// no pairing (a solo gunnar, a test).
    fn topology_facts(&self) -> crate::topology::Facts {
        crate::topology::Facts::default()
    }

    /// **Is this box a twin right now?** Default `false`. Read by the browse refusal: a twin is
    /// frozen like a browse replica, yet it must take its OWN data-set growth exactly as main
    /// does (owner ruling 2026-09-22, T21), so the grow verbs pass on a twin and on no browse box.
    fn is_twin(&self) -> bool {
        false
    }

    /// **Does this box's transaction log FOLLOW another's** (a twin, or a main
    /// being restored)? It originates no transaction until it opens as the
    /// primary. Default `false` (wiring tracer B 2026-10-08: `Entitlement.Set`
    /// answers such a box's "try again" before it asks whether a namespace
    /// exists — a namespace made after the twin's snapshot is not replayed yet).
    fn log_follows(&self) -> bool {
        false
    }

    /// **States this box is in that an operator must act on and no console reaches**
    /// (`Admin.Health.alarms`). Default: none.
    fn health_alarms(&self) -> Vec<gunnar_proto::HealthAlarm> {
        Vec::new()
    }

    /// **Why this box reports `maintenance`** (`Admin.Health.status`), one reason each; empty is
    /// `serving`. It still serves either way. Default: none.
    fn maintenance_reasons(&self) -> Vec<String> {
        Vec::new()
    }

    /// Is `principal_id` a member of `group`? The twin's log gate asks it
    /// (`Mirror.LogHead` / `LogFetch` for a member of group `twin`). Default
    /// `false`: a plane with no groups has no twin members.
    fn is_group_member(&self, _principal_id: &str, _group: &str) -> ControlResult<bool> {
        Ok(false)
    }

    /// `Bootstrap.BecomeTwin`: a fresh box becomes a twin of `primary`;
    /// answers the twin's public key line.
    fn become_twin(&self, _primary: &str) -> ControlResult<String> {
        Err(ControlError::Unimplemented("twin: this control plane cannot become a twin"))
    }

    /// `Bootstrap.AssignTunnelAddress` (2026-09-21, T17): `address` empty =
    /// answer only; otherwise record it as this box's WireGuard tunnel address
    /// and raise `wg0`. Default: a plane with no tunnel has no address to take.
    fn assign_tunnel_address(&self, _address: &str) -> ControlResult<gunnar_proto::AssignTunnelAddressResponse> {
        Err(ControlError::Unimplemented("tunnel: this control plane has no WireGuard tunnel (built without wg-client)"))
    }

    /// `Admin.SetTunnelAddress` (2026-09-21, T17): the SAME answer as
    /// [`ControlPlane::assign_tunnel_address`] for a box past its bootstrap
    /// window. The caller has already been held to server-wide Admin by the
    /// service; the plane does not re-check the window.
    fn set_tunnel_address(&self, _address: &str) -> ControlResult<gunnar_proto::AssignTunnelAddressResponse> {
        Err(ControlError::Unimplemented("tunnel: this control plane has no WireGuard tunnel (built without wg-client)"))
    }

    /// `Bootstrap.TwinCsr`: this twin's replication certificate request,
    /// minting its TLS key on the box on first call (lane T9).
    fn twin_csr(&self) -> ControlResult<String> {
        Err(ControlError::Unimplemented("Bootstrap.TwinCsr"))
    }

    /// `Bootstrap.InstallTwinCertificate`: take the primary-signed certificate
    /// and pin its issuer; answers whole days left.
    fn install_twin_certificate(&self, _cert_pem: &str, _ca_pem: &str, _replication_port: u16, _replication_host: &str) -> ControlResult<i64> {
        Err(ControlError::Unimplemented("Bootstrap.InstallTwinCertificate"))
    }

    /// `Admin.IssueTwinCertificate`, on the primary: `(cert, issuer, port, host)`
    /// — `host` is the address the listener is bound on when that is not the
    /// joined one (its tunnel address), empty otherwise.
    fn issue_twin_certificate(&self, _principal: &str, _csr_pem: &str) -> ControlResult<(String, String, u16, String)> {
        Err(ControlError::Unimplemented("Admin.IssueTwinCertificate"))
    }

    /// [`ControlPlane::log_head`], **signed over the caller's nonce** — what a
    /// twin's poll actually calls (2026-09-21, lane T4). The default ignores the
    /// nonce and answers unsigned, which is the honest answer for a plane with
    /// no key of its own; a twin that has pinned a key refuses it.
    fn log_head_signed(&self, _nonce: &str) -> ControlResult<gunnar_proto::LogHeadResponse> {
        self.log_head()
    }

    /// **A twin's proof that a NEWER generation of this box's log exists**
    /// (`LogHeadRequest.generation_proof`, stinky row 18, the main's half).
    /// Called before the answer is made. A box that finds the proof genuine — its
    /// own log, its own host key, a generation above its own — fences itself and
    /// originates nothing from then on; anything else is ignored. The default
    /// keeps no log and has nothing to fence.
    fn fence_by_generation(&self, _proof: &gunnar_proto::GenerationProof) -> ControlResult<()> {
        Ok(())
    }

    /// **Is this box's git listener still bound where a primary's was?**
    ///
    /// A separate question from `become_twin`'s answer because it stays true
    /// until the box restarts, and can be asked again afterwards. The bind is
    /// chosen once at start-up; a box conferred the twin role over the wire is
    /// past that point. Nothing is SERVED through that socket — the front door
    /// refuses every git service on a twin from the instant the role lands —
    /// but it is still bound on an address a twin's socket does not belong on.
    ///
    /// `false` for a plane that has no git listener to be wrong about.
    fn twin_git_listener_needs_restart(&self) -> bool {
        false
    }

    /// `Bootstrap.Restore`, start: the next sequence this box takes. `owner` is
    /// who the stream's credential proved the caller to be
    /// ([`crate::auth::authenticate_restore`]): a box already restoring resumes
    /// only for the owner that started it, and refuses anybody else
    /// `PERMISSION_DENIED` (h-v1-6a, 2026-09-28).
    fn restore_begin(&self, _owner: &str) -> ControlResult<u64> {
        Err(ControlError::Unimplemented("restore: this control plane keeps no log to restore into"))
    }

    /// **The keys this box takes a `Bootstrap.Restore` from** (h-v1-6a,
    /// 2026-09-28), as OpenSSH public key lines. A fresh box has no identity
    /// table to check a credential against, so these are the keys it was FED
    /// before anybody could enrol: what the installer's door handed it, and
    /// what the medium named. Default: none — a plane that was fed nothing
    /// refuses every restore by name, which is the fail-closed answer.
    fn restore_keys(&self) -> ControlResult<Vec<String>> {
        Ok(Vec::new())
    }

    /// `Bootstrap.Restore`, the room check: has this box's data set space for a stream of
    /// `announced_bytes` of log? Asked ONCE, on the first chunk, before anything of the restore
    /// is written — the restore-onto-different-hardware case. `0` = the sender did not say, and
    /// no guess is made. A default `Ok(())`: a plane with no data set of its own cannot run out.
    fn restore_room(&self, _announced_bytes: u64) -> ControlResult<()> {
        Ok(())
    }

    /// **`Bootstrap.Restore`, the stream's intake** (restore-large-entry, 2026-10-07): what every
    /// chunk of one stream is fed into. Starting one is [`ControlPlane::restore_begin`] — the
    /// owner check, the next sequence — and, on a plane that stages, holding the box's ONE intake
    /// until it is dropped. `Ok(None)`: another stream's intake still holds the box (a stream the
    /// sender abandoned whose apply has not returned yet); the caller waits a little and asks
    /// again rather than racing it over the same staged entry.
    ///
    /// The default gathers every entry whole in memory and applies it with
    /// [`ControlPlane::restore_apply`]: right for a plane with no disk of its own.
    fn restore_intake<'a>(&'a self, owner: &str) -> ControlResult<Option<Box<dyn RestoreIntake + 'a>>> {
        let next = self.restore_begin(owner)?;
        Ok(Some(Box::new(GatherIntake { plane: self, next, open: None, want_len: 0, applied: 0, lost: 0 })))
    }

    /// `Bootstrap.Restore`, one whole entry: `true` when a replayed CAS lost.
    fn restore_apply(&self, _entry: gunnar_proto::LogEntry) -> ControlResult<bool> {
        Err(ControlError::Unimplemented("restore: this control plane keeps no log to restore into"))
    }

    /// `Bootstrap.Restore`, end: the head reached. `announced_head` is the largest
    /// `LogChunk.restore_head` the stream carried (0 = none): a box whose head is below it
    /// refuses to finish and stays restoring.
    fn restore_finish(&self, _announced_head: u64, _log_id_of_source: &str) -> ControlResult<u64> {
        Err(ControlError::Unimplemented("restore: this control plane keeps no log to restore into"))
    }

    /// `Mirror.SnapshotFetch`: a consistent snapshot streamed into `sink` as `LogChunk` pieces —
    /// items with `seq` 0, at most 1 MiB of blob a piece, `blob_len` the item's total on its
    /// last piece and `u64::MAX` before it, the `SnapshotAt` terminator last. Never gathered:
    /// the sink's backpressure paces the producer. Answers the sequence it was taken at.
    fn snapshot_stream(&self, _sink: &mut dyn FnMut(gunnar_proto::LogChunk) -> ControlResult<()>) -> ControlResult<u64> {
        Err(ControlError::Unimplemented("snapshot: this control plane keeps no log"))
    }

    /// **[`ControlPlane::snapshot_stream`] from item `from_item`, byte `from_offset` on**
    /// (`SnapshotFetchRequest`, restore-after-snapshot 2026-10-07): a twin's kept base, resumed
    /// for a box being restored across capped calls; every piece names its item, offset and base
    /// (`LogChunk.snapshot_item`, `offset`, `snapshot_id`). `expect_id`: the base the asker holds
    /// part of — another one now is refused `[snapshot-changed]`. The default resumes nothing:
    /// asked from its start it is `snapshot_stream`, otherwise refused by name.
    fn snapshot_stream_from(&self, from_item: u64, from_offset: u64, expect_id: &str, sink: &mut dyn FnMut(gunnar_proto::LogChunk) -> ControlResult<()>) -> ControlResult<u64> {
        if from_item == 0 && from_offset == 0 && expect_id.trim().is_empty() {
            return self.snapshot_stream(sink);
        }
        Err(ControlError::Unimplemented("snapshot: this control plane cannot resume a snapshot"))
    }

    /// `Mirror.LogRetain`: drop every whole segment below `from` (0 = head + 1); answers the
    /// head as `LogHead` does. Refused on a twin.
    fn log_retain(&self, _from: u64) -> ControlResult<gunnar_proto::LogHeadResponse> {
        Err(ControlError::Unimplemented("retention: this control plane keeps no log"))
    }

    /// ★ **The log's segments, for a byte-bounded retention** (cov-13, 2026-10-06):
    /// `(first, last, bytes)`, oldest first, the open one last (`last = first - 1` while it
    /// holds nothing). Only whole segments are dropped, so this is what a byte budget is
    /// decided over. Default `Unimplemented`: a plane with no log.
    fn log_layout(&self) -> ControlResult<Vec<(u64, u64, u64)>> {
        Err(ControlError::Unimplemented("retention: this control plane keeps no log"))
    }

    /// ★ **The data volume the log lives on**: `(total, used)` bytes, statvfs — what the
    /// retention's byte budget is a share of, and what a twin silent through main's growth
    /// is owed (cov-13). `None` when unknown, and while the set is released for a growth
    /// (statvfs would read the box's root, not the set).
    fn data_volume(&self) -> Option<(u64, u64)> {
        None
    }

    /// ★ `twin-owed.json` (cov-13): what each twin is owed, as last written; empty = never.
    /// Default `Unimplemented`: the book then lives in memory only.
    fn twin_owed_read(&self) -> ControlResult<Vec<u8>> {
        Err(ControlError::Unimplemented("twin-owed: this control plane keeps no data root"))
    }

    /// ★ Write `twin-owed.json` (replace, crash-safe).
    fn twin_owed_write(&self, _bytes: &[u8]) -> ControlResult<()> {
        Err(ControlError::Unimplemented("twin-owed: this control plane keeps no data root"))
    }

    /// `Bootstrap.Restore`, one snapshot piece (`seq` 0), staged on disk: `Some(at)` once the
    /// terminator installed the snapshot, so the restore continues from `at + 1`.
    fn restore_snapshot_chunk(&self, _chunk: &gunnar_proto::LogChunk) -> ControlResult<Option<u64>> {
        Err(ControlError::Unimplemented("restore: this control plane keeps no log to restore into"))
    }

    // ── Outbound webhooks: delivery to an external URL ─────────────────────
    //
    // A webhook is a subscriber to the SAME event bus the replicator and the SSE
    // fan-out consume (one implementation), which POSTs each matching event to a URL, signed
    // with an sshsig the receiver verifies against the server's published key.
    // Configuration lives in `<data-root>/webhooks.json` and is set over the
    // control plane because the appliance has no shell — exactly the reason the
    // replication verbs above exist. Default `Unimplemented`, so a server built
    // without the `webhooks` feature says so rather than silently accepting a
    // configuration it will never deliver on (measured, not asserted — the silent green).

    /// Add or replace one endpoint by name, persist it, and start delivering.
    fn webhook_add(&self, _request: AddWebhookRequest) -> ControlResult<WebhookListResponse> {
        Err(ControlError::Unimplemented(
            "outbound webhooks: this server was built without the `webhooks` feature, so it              cannot be given an outbound endpoint",
        ))
    }

    /// What is configured, plus the signing public key a receiver verifies with.
    fn webhook_list(&self) -> ControlResult<WebhookListResponse> {
        Err(ControlError::Unimplemented(
            "outbound webhooks: this server was built without the `webhooks` feature, so it              has no outbound endpoints",
        ))
    }

    /// Drop one endpoint by name, persist that, and stop delivering to it.
    fn webhook_remove(&self, _name: &str) -> ControlResult<WebhookListResponse> {
        Err(ControlError::Unimplemented(
            "outbound webhooks: this server was built without the `webhooks` feature, so it              has no outbound endpoint to remove",
        ))
    }

    /// **The owner's enterprise list** (2026-10-05): the account audit log,
    /// the operator's aggregates and look, the name filter
    /// ([`crate::enterprise::EnterprisePlane`]), when this plane has the
    /// stores behind it. `None`: nothing is recorded, nothing is shielded,
    /// every name is clean, and every `Enterprise` verb answers UNIMPLEMENTED.
    fn enterprise(&self) -> Option<&dyn crate::enterprise::EnterprisePlane> {
        None
    }
}

// ───────────────────────────────────────────────────────────────────────────
// The test double
// ───────────────────────────────────────────────────────────────────────────

/// An in-memory [`ControlPlane`]. **A test double, not a backend.**
///
/// It implements the catalogue and identity halves so the services can be
/// exercised end to end, and answers [`ControlError::Unimplemented`] for
/// everything that genuinely needs `gunnar-store`: refs, objects, diffs and
/// mirror sync. Returning an empty success for those would be worse than
/// useless: an empty ref list looks exactly like a working server with no refs
/// (measured, not asserted — the silent green).
#[derive(Debug, Default)]
pub struct MemoryControlPlane {
    inner: Mutex<Inner>,
    /// **A slow synchronous call, on demand** (R5.1-G): milliseconds every
    /// `bootstrap_state` sleeps on the calling thread before it answers — the
    /// shape of a real control-plane call (a disk walk, an fsync) that holds a
    /// worker without yielding. Zero, the default, answers at once.
    stall_ms: std::sync::atomic::AtomicU64,
}

/// One seat lot: `(room per seat, total, filled ids, order)`.
type SeatLot = (u64, u32, Vec<String>, String);

#[derive(Debug, Default)]
struct Inner {
    namespaces: BTreeMap<String, Namespace>,
    repos: BTreeMap<(String, String), Repo>,
    principals: BTreeMap<String, Principal>,
    /// Employee id → the account that enrolled it (`enroll_employee`).
    employers: BTreeMap<String, String>,
    /// Employee id → the room it holds and the order it came with.
    rooms: BTreeMap<String, (u64, String)>,
    /// Account → its seat lots `(room per seat, total, filled ids, order)`.
    seats: BTreeMap<String, Vec<SeatLot>>,
    /// Order → `(account, principal or "", seats, room, need)`.
    room_orders: BTreeMap<String, (String, String, u32, u64, u64)>,
    /// namespace → what a test says gunnar measured for it (`usage`).
    /// Empty in every other use, and then `usage` answers `Unimplemented`
    /// exactly as before: the double has no store to count.
    usage: BTreeMap<String, gunnar_proto::NamespaceUsage>,
    /// Web credentials (Stage L): credential id (base64url) → the row and
    /// its public key.
    credentials: BTreeMap<String, StoredCredential>,
    /// Web credential (base64url) → the session keys its logins enrolled.
    credential_sessions: BTreeMap<String, Vec<String>>,
    /// Invites (Stage L): id → the row and the secret's hash.
    invites: BTreeMap<String, (gunnar_proto::Invite, String)>,
    /// Redeemed invites: the secret's hash (the plane's USED answer).
    redeemed_invites: std::collections::BTreeSet<String>,
    keys: BTreeMap<String, PublicKey>,
    grants: Vec<Grant>,
    protections: BTreeMap<(String, String), BTreeMap<String, BranchProtection>>,
    ref_policies: BTreeMap<(String, String), RefNamePolicy>,
    mirrors: BTreeMap<(String, String), MirrorConfig>,
    /// fingerprint → the `authorized_keys` line, so a credential can be
    /// verified. The proto `PublicKey` deliberately does not carry key
    /// material; the gate needs it, so the double keeps it beside.
    openssh: BTreeMap<String, String>,
    /// Client-certificate DER → the identity it was registered under.
    ///
    /// Keyed by the **bytes**, not by a fingerprint the double computed: the
    /// real adapter hashes with `gunnar_core::auth::fingerprint_of_certificate`
    /// and a second hash spelled here would be a twin of that decision (one implementation).
    /// The double is handed the fingerprint by whoever registers, so the two
    /// can never disagree — there is only one of them.
    certificates: BTreeMap<Vec<u8>, CertificateIdentity>,
    /// `Scope::Server` grants (the ACL model). The proto `Grant` cannot express one, since
    /// it requires a namespace, which is a real gap recorded in
    /// `gunnar-cli`'s adapter; the double works around it rather than
    /// pretending a server-wide grant is a namespace grant on `""`.
    server_grants: BTreeMap<String, GrantLevel>,
    /// Narrowed principals → the methods each may call (f-principal-reach).
    narrowed: BTreeMap<String, Vec<String>>,
    /// See [`MemoryControlPlane::force_may`]. `None` in every real use.
    forced_may: Option<bool>,
    /// See [`MemoryControlPlane::set_twin`]. `false` unless a test says otherwise.
    twin: bool,
    /// `set_log_follows`: a main being restored.
    log_follows: bool,
    /// See [`MemoryControlPlane::await_enrolment`]: fingerprint → the fed key's
    /// OpenSSH line. Empty unless a test feeds one.
    awaiting: BTreeMap<String, String>,
    /// See [`MemoryControlPlane::forge_principal_id`]. `None` in every real use.
    forged_principal_id: Option<String>,
    /// The bootstrap record: `(at, principal, fingerprint)`. Beside the
    /// identity maps above and never removed with them, which is the whole
    /// property the double has to reproduce.
    bootstrap: Option<(i64, String, String)>,
    /// namespace → the fact this box holds for it (the METERED appliance).
    ///
    /// Seeded by a test; empty in every other use. `entitlement_get` answers
    /// out of it, which is the truthful thing for a double to do — "no fact"
    /// is Free, and that is a real answer rather than a missing one. Writing a
    /// fact is still [`ControlError::Unimplemented`]: the double has no
    /// monetize key to VERIFY one with, and a `set` that stored whatever it
    /// was given would be a double that is more permissive than the thing it
    /// stands in for.
    #[cfg(feature = "metered")]
    entitlements: BTreeMap<String, gunnar_proto::EntitlementFact>,
    /// How many times the actor-ticket key has been asked to sign.
    ///
    /// **An instrument, and the reason it exists is a specific defect.** A
    /// handler that signed a ticket and then checked the grant would satisfy a
    /// test that only looked at the returned `Status`, while having put a
    /// valid, spendable ticket for somebody else's tenant into whatever sits
    /// between the handler and the caller. Counting the signatures is how a
    /// refusal is held to "nothing was minted" rather than to "nothing was
    /// returned".
    #[cfg(feature = "metered")]
    actor_tickets_signed: u64,
    /// The double's actor-ticket key, minted on first use, exactly as the real
    /// appliance's is — a real ed25519 key and never a stand-in, because the
    /// signature it makes is checked by the OTHER repository's verifier in
    /// `tests/billing.rs`.
    #[cfg(feature = "metered")]
    actor_ticket_key: Option<ed25519_dalek::SigningKey>,
}

impl Inner {
    /// The double's actor-ticket key, minted on first use.
    ///
    /// Minted from `OsRng` and not from a constant seed. A fixed key would make
    /// every run of `tests/billing.rs` produce byte-identical tickets, which
    /// reads as reassuring and is the opposite: the contract test would then
    /// be checking one recorded signature rather than the agreement between a
    /// signer and a verifier.
    #[cfg(feature = "metered")]
    fn actor_ticket_key(inner: &mut Inner) -> &ed25519_dalek::SigningKey {
        inner.actor_ticket_key.get_or_insert_with(|| {
            use rand::Rng as _;
            let mut seed = [0u8; 32];
            rand::rand_core::UnwrapErr(rand::rngs::SysRng).fill_bytes(&mut seed);
            ed25519_dalek::SigningKey::from_bytes(&seed)
        })
    }

    /// Who, other than `own_id`, holds `name` — as their name or as their id.
    /// The double's spelling of `AuthTable::refuse_taken_name`.
    fn name_holder(inner: &Inner, name: &str, own_id: &str) -> Option<String> {
        inner
            .principals
            .values()
            .find(|p| p.id != own_id && (p.name == name || p.id == name))
            .map(|p| p.id.clone())
    }
}

/// The default branch a repo gets when the request does not name one.
pub const DEFAULT_BRANCH: &str = "refs/heads/main";

/// sha256 is the default hash kind for new repos.
pub const DEFAULT_HASH_KIND: HashKind = HashKind::Sha256;

impl MemoryControlPlane {
    /// **Say this plane is a twin** (T21): what `CoreControlPlane::is_twin` reads off
    /// `twin.json`, for the gate's twin-only grow verbs.
    pub fn set_twin(&self, yes: bool) {
        self.inner.lock().expect("memory control plane").twin = yes;
    }

    /// **Say this plane's log follows another's** (`ControlPlane::log_follows`):
    /// a main being restored, as `CoreControlPlane` reads it off its txlog.
    pub fn set_log_follows(&self, yes: bool) {
        self.inner.lock().expect("memory control plane").log_follows = yes;
    }

    /// **Say this key was fed and is not enrolled yet** (`Identity::awaiting_enrolment`):
    /// what a primary's installer feed says of the front's keys until its pairing loop
    /// enrolled them.
    pub fn await_enrolment(&self, fingerprint: &str, openssh: &str) {
        self.inner.lock().expect("memory control plane").awaiting.insert(fingerprint.to_owned(), openssh.to_owned());
    }

    pub fn new() -> Self {
        Self::default()
    }

    /// Make every `bootstrap_state` (`Bootstrap.Status`) hold its thread for
    /// `by` before it answers: a synchronous call that blocks the worker it
    /// runs on, which is what the keepalive test needs every handler worker
    /// doing at once.
    pub fn stall_bootstrap_state(&self, by: std::time::Duration) {
        self.stall_ms.store(
            by.as_millis().min(u64::MAX as u128) as u64,
            std::sync::atomic::Ordering::SeqCst,
        );
    }

    /// Convenience for tests: a namespace and a repo in one call.
    pub fn with_repo(namespace: &str, repo: &str) -> Self {
        let plane = Self::new();
        plane
            .create_namespace(namespace, "")
            .expect("fresh namespace");
        plane
            .create_repo(NewRepo {
                path: RepoPath::new(namespace, repo),
                hash_kind: DEFAULT_HASH_KIND,
                default_branch: String::new(),
                visibility: Visibility::Private,
                description: String::new(),
                adopt: false,
            })
            .expect("fresh repo");
        plane
    }

    fn key(path: &RepoPath) -> (String, String) {
        (path.namespace.clone(), path.repo.clone())
    }

    /// **Narrow a principal to a method list** (f-principal-reach).
    pub fn narrow(&self, principal_id: &str, methods: &[&str]) {
        self.inner.lock().unwrap().narrowed.insert(principal_id.to_owned(), methods.iter().map(|m| m.to_string()).collect());
    }

    /// Give a principal a `Scope::Server` grant (the ACL model).
    ///
    /// Not on [`ControlPlane`]: the proto `Grant` cannot express a server-wide
    /// scope (it requires a namespace), which is a gap in `gunnar-proto`
    /// recorded in `gunnar-cli`'s adapter. Until that is fixed the double
    /// carries server grants beside the namespace/repo ones rather than
    /// smuggling one in as a namespace grant on `""`.
    pub fn grant_server(&self, principal_id: &str, level: GrantLevel) {
        self.inner
            .lock()
            .unwrap()
            .server_grants
            .insert(principal_id.to_owned(), level);
    }

    /// Seed what gunnar measured for one namespace (`Usage.Read`'s row).
    pub fn set_usage(&self, usage: gunnar_proto::NamespaceUsage) {
        self.inner
            .lock()
            .unwrap()
            .usage
            .insert(usage.namespace.clone(), usage);
    }

    /// Seed the fact this box holds for one namespace (the METERED appliance).
    ///
    /// The double's stand-in for a fact monetize signed and `Entitlement.Set`
    /// verified and stored. It takes the fact as given and checks no signature,
    /// which is why it is a test-only setter and not `entitlement_set`.
    #[cfg(feature = "metered")]
    pub fn set_entitlement_fact(&self, fact: gunnar_proto::EntitlementFact) {
        self.inner
            .lock()
            .unwrap()
            .entitlements
            .insert(fact.namespace.clone(), fact);
    }

    /// **How many times the actor-ticket key has been asked to sign.**
    ///
    /// The instrument that turns "the request was refused" into "nothing was
    /// minted". See the field's own comment for the defect it exists to catch.
    #[cfg(feature = "metered")]
    pub fn actor_tickets_signed(&self) -> u64 {
        self.inner.lock().unwrap().actor_tickets_signed
    }

    /// **A red-first instrument.** Force [`Identity::may`] to a constant answer.
    ///
    /// `Some(false)` must make every *allow* test in `tests/auth.rs` go red,
    /// and `Some(true)` must make every *permission-denied* test go red. Both
    /// are asserted there. Without them a handler that had been hard-wired to
    /// allow, or a gate that refused everything, would pass a naive suite just
    /// as well as a correct one. That is the exact failure this instrument
    /// exists to make impossible.
    ///
    /// `None` restores the real answer.
    pub fn force_may(&self, forced: Option<bool>) {
        self.inner.lock().unwrap().forced_may = forced;
    }

    /// **A red-first instrument, for anonymous read.** Make
    /// [`Identity::registered_key`] name
    /// a principal of the caller's choosing, whatever the key table says.
    ///
    /// `AuthGate` refuses a credential that resolves to the anonymous principal
    /// id. Through `gunnar-core` that state is unreachable (the reserved id
    /// cannot be registered and cannot be given a key), so without this the
    /// guard could never be *seen* refusing, and a guard nobody has watched
    /// fail is worth nothing.
    ///
    /// Nor is it a contrivance. `AuthGate` takes [`ControlPlane`] as a
    /// **trait**, so any adapter, present or future, can return whatever
    /// principal id it likes. That is the situation this reproduces.
    ///
    /// `None` restores the real answer.
    pub fn forge_principal_id(&self, id: Option<&str>) {
        self.inner.lock().unwrap().forged_principal_id = id.map(str::to_owned);
    }
}

/// `Admin` implies `Write` implies `Read` (the ACL model). One ordering, used by every
/// comparison below.
fn level_rank(level: GrantLevel) -> u8 {
    match level {
        GrantLevel::Unspecified => 0,
        GrantLevel::Read => 1,
        GrantLevel::Write => 2,
        GrantLevel::Admin => 3,
    }
}

/// The **test double's** ACL.
///
/// It is a double of `gunnar_core::AuthTable::may`, not a second production
/// decision: the real server routes every `may` through `AuthTable` via the
/// adapter, and this exists so the seven services can be exercised without a
/// storage backend. Its rules are deliberately the same three: unknown
/// principal may nothing, a covering grant must be at least `level`, and a
/// namespace grant matches its namespace **exactly** (`team` does not reach
/// `team/sub`).
impl Identity for MemoryControlPlane {
    fn awaiting_enrolment(&self, fingerprint: &str) -> Option<String> {
        self.inner.lock().unwrap().awaiting.get(fingerprint).cloned()
    }

    fn narrowed_methods(&self, principal_id: &str) -> ControlResult<Option<Vec<String>>> {
        Ok(self.inner.lock().unwrap().narrowed.get(principal_id).cloned())
    }

    fn registered_key(&self, fingerprint: &str) -> ControlResult<Option<RegisteredKey>> {
        let inner = self.inner.lock().unwrap();
        let Some(key) = inner.keys.get(fingerprint) else {
            return Ok(None);
        };
        if let Some(expires) = key.expires_at_unix_ms {
            if now_ms() >= expires {
                return Ok(None);
            }
        }
        let Some(openssh) = inner.openssh.get(fingerprint) else {
            return Ok(None);
        };
        Ok(Some(RegisteredKey {
            principal_id: inner
                .forged_principal_id
                .clone()
                .unwrap_or_else(|| key.principal_id.clone()),
            openssh: openssh.clone(),
            machine: inner
                .principals
                .get(&key.principal_id)
                .is_some_and(|p| p.kind == PrincipalKind::Machine as i32),
        }))
    }

    fn registered_certificate(&self, der: &[u8]) -> ControlResult<Option<CertificateIdentity>> {
        let inner = self.inner.lock().unwrap();
        Ok(inner.certificates.get(der).cloned())
    }

    fn may(&self, principal_id: &str, access: &Access, level: GrantLevel) -> ControlResult<bool> {
        if level == GrantLevel::Unspecified {
            return Ok(false);
        }
        let inner = self.inner.lock().unwrap();
        if let Some(forced) = inner.forced_may {
            return Ok(forced);
        }
        if !inner.principals.contains_key(principal_id) {
            return Ok(false);
        }
        // A principal is intrinsically admin of the account named after it.
        // `gunnar_core::AuthTable::owns_account` is the real decision; this is
        // the double keeping the same shape, and `tests/bootstrap.rs` asserts
        // both directions of it so the double cannot quietly allow more.
        // A MEMBER owns no account by its name (2026-10-07), as in the table.
        let member = inner.employers.contains_key(principal_id);
        let owns_account = |namespace: &str| !member && !namespace.is_empty() && namespace == principal_id;
        match access {
            Access::Namespace(name) if owns_account(name) => return Ok(true),
            Access::Repo(path) if owns_account(&path.namespace) => return Ok(true),
            _ => {}
        }
        // A PUBLIC repository is readable by every registered principal —
        // `AuthTable::may`'s rule for it, kept here so a double cannot quietly
        // hide what the real table shows (the outsider's view of an account,
        // `tests/account_model.rs`).
        if let Access::Repo(path) = access {
            if level == GrantLevel::Read
                && inner
                    .repos
                    .get(&(path.namespace.clone(), path.repo.clone()))
                    .is_some_and(|repo| repo.visibility == Visibility::Public as i32)
            {
                return Ok(true);
            }
        }
        let want = level_rank(level);
        // **A server grant stops at a customer's account** (2026-10-05):
        // `gunnar_core::AuthTable`'s shield, kept here so the double cannot
        // quietly show an operator what the real table refuses it.
        // The box's own first identity (bootstrap) is the estate, as a
        // machine is: `gunnar_core::AuthTable::is_box_identity`.
        let box_identity = inner.bootstrap.as_ref().map(|(_, id, _)| id.as_str());
        let machine = box_identity == Some(principal_id)
            || inner
                .principals
                .get(principal_id)
                .is_some_and(|p| p.kind == PrincipalKind::Machine as i32);
        let shielded = |namespace: &str| {
            let account = namespace.split('/').next().unwrap_or("");
            !machine
                && !account.is_empty()
                && box_identity != Some(account)
                && account != principal_id
                && inner
                    .principals
                    .get(account)
                    .is_some_and(|p| p.kind == PrincipalKind::User as i32)
                && !inner.employers.contains_key(account)
                && inner.employers.get(principal_id).map(String::as_str) != Some(account)
        };
        let inside_closed_account = match access {
            Access::Server => false,
            Access::Namespace(name) => shielded(name),
            Access::Repo(path) => shielded(&path.namespace),
        };
        let server = inner
            .server_grants
            .get(principal_id)
            .copied()
            .map(level_rank)
            .unwrap_or(0);
        if server >= want && !inside_closed_account {
            return Ok(true);
        }
        let scoped = inner
            .grants
            .iter()
            .filter(|grant| grant.principal_id == principal_id)
            .filter(|grant| match access {
                // A server-scoped question is never answered by a namespace or
                // repo grant. Widening it there would be silent escalation.
                Access::Server => false,
                Access::Namespace(name) => grant.repo.is_empty() && &grant.namespace == name,
                Access::Repo(path) => {
                    grant.namespace == path.namespace
                        && (grant.repo.is_empty() || grant.repo == path.repo)
                }
            })
            .filter_map(|grant| GrantLevel::try_from(grant.level).ok())
            .map(level_rank)
            .max()
            .unwrap_or(0);
        Ok(scoped >= want)
    }
}

/// The double's first-boot door.
///
/// It mirrors `gunnar_core::AuthTable::install_first_identity` in the one thing
/// that matters: the record is checked and written **under the same lock** as
/// the identity it creates, so two concurrent enrolments cannot both win. In
/// the real server that lock is a compare-and-swap on `refs/gunnar/auth`; here
/// it is the `Mutex` this whole double is behind. `tests/bootstrap.rs` races
/// both.
impl FirstBoot for MemoryControlPlane {
    fn bootstrap_state(&self) -> ControlResult<BootstrapState> {
        let stall = self.stall_ms.load(std::sync::atomic::Ordering::SeqCst);
        if stall > 0 {
            // Blocking on purpose: `std::thread::sleep`, not a timer, so the
            // worker is held exactly as a synchronous store call holds it.
            std::thread::sleep(std::time::Duration::from_millis(stall));
        }
        let inner = self.inner.lock().unwrap();
        let (at, principal, fingerprint) =
            inner
                .bootstrap
                .clone()
                .unwrap_or((0, String::new(), String::new()));
        Ok(BootstrapState {
            window_open: inner.bootstrap.is_none()
                && inner.principals.is_empty()
                && inner.keys.is_empty(),
            bootstrapped: inner.bootstrap.is_some(),
            bootstrapped_at_unix_ms: at,
            bootstrapped_principal: principal,
            bootstrapped_fingerprint: fingerprint,
            principals: inner.principals.len() as u64,
            keys: inner.keys.len() as u64,
            // An in-memory control plane has no disk to be racing, so it is
            // always serving. The boot window is `gunnar-server`'s fact.
            serving_git: true,
            readiness: "serving git".to_owned(),
            build: String::new(),
            role: "primary".to_owned(),
            ..Default::default()
        })
    }

    fn enroll_first_identity(
        &self,
        name: &str,
        openssh_public_key: &str,
    ) -> ControlResult<Enrolled> {
        let fingerprint = fingerprint_of_openssh(openssh_public_key).ok_or_else(|| {
            ControlError::InvalidArgument(format!(
                "{openssh_public_key:?} is not an authorized_keys line"
            ))
        })?;
        let mut inner = self.inner.lock().unwrap();
        if let Some((at, principal, _)) = &inner.bootstrap {
            return Err(ControlError::FailedPrecondition(format!(
                "this server was bootstrapped at {at} by \"{principal}\"; the window is closed \
                 permanently and deleting identities does not reopen it"
            )));
        }
        if !inner.principals.is_empty() || !inner.keys.is_empty() {
            return Err(ControlError::FailedPrecondition(format!(
                "this server already holds {} principals and {} keys, so it is not a fresh server",
                inner.principals.len(),
                inner.keys.len()
            )));
        }
        if name.is_empty() {
            return Err(ControlError::InvalidArgument("`name` is required".into()));
        }
        let at = now_ms();
        let principal = Principal {
            id: name.to_owned(),
            name: name.to_owned(),
            kind: PrincipalKind::User as i32,
            created_at_unix_ms: at,
        };
        let key = PublicKey {
            fingerprint: fingerprint.clone(),
            algo: String::new(),
            principal_id: name.to_owned(),
            comment: String::new(),
            added_at_unix_ms: at,
            last_used_unix_ms: None,
            expires_at_unix_ms: None,
        };
        inner.principals.insert(name.to_owned(), principal.clone());
        inner.keys.insert(fingerprint.clone(), key.clone());
        inner
            .openssh
            .insert(fingerprint.clone(), openssh_public_key.trim().to_owned());
        // gunnar-admin, the ordinary `Scope::Server` Admin grant.
        inner
            .server_grants
            .insert(name.to_owned(), GrantLevel::Admin);
        inner.namespaces.insert(
            name.to_owned(),
            Namespace {
                name: name.to_owned(),
                description: String::new(),
                created_at_unix_ms: at,
                owner_kind: 0,
                ..Default::default()
            },
        );
        inner.bootstrap = Some((at, name.to_owned(), fingerprint));
        Ok(Enrolled {
            principal,
            key,
            account: name.to_owned(),
            bootstrapped_at_unix_ms: at,
            // The in-memory control plane has no PKI; the degraded (empty)
            // shape is exactly what it should exercise.
            client_cert_pem: String::new(),
            client_key_pem: String::new(),
            issuer_cert_pem: String::new(),
        })
    }
}

fn now_ms() -> i64 {
    crate::bus::now_unix_ms()
}

/// The `SHA256:<base64>` fingerprint OpenSSH prints, computed by `ssh-key`
/// itself.
///
/// This used to be a deliberate `PLACEHOLDER-…` string, because computing
/// "something like" a fingerprint and letting an integration believe it would
/// have been a lie. Since gRPC authentication landed this crate links
/// `ssh-key`, so the double can produce the **real** value. It must, because a
/// credential carries the fingerprint the client's key actually has and the
/// double has to be able to find it.
///
/// It is not a second definition of gunnar's fingerprint: `ssh-key`'s own
/// `PublicKey::fingerprint` and `gunnar_core::fingerprint_of_blob` are both
/// sha256 over the same wire blob, and `gunnar-core` only spells it out by hand
/// because russh pins an incompatible `ssh-key`. A mismatch between them
/// would be caught the first time a real key was used against a real table.
fn fingerprint_of_openssh(line: &str) -> Option<String> {
    let key = ssh_key::PublicKey::from_openssh(line).ok()?;
    Some(key.fingerprint(ssh_key::HashAlg::Sha256).to_string())
}

/// **The double's spelling of the key-attestation rule**, structurally.
///
/// The rule and its arithmetic are `gunnar_core::key_attestation` — one
/// writer, which the shipped plane asks. This crate does not depend on
/// `gunnar-core` (see the note on `fingerprint_of_openssh`'s neighbour), so the
/// double reproduces the three arms — nobody to ask, the caller is the target,
/// otherwise required — and checks the attestation's SHAPE: present, about
/// this principal and this fingerprint, signed by a key the target holds. It
/// does **not** verify the signature; the real record is verified on the real
/// plane, and `gunnar-cli/tests/add_key_attested.rs` drives that red. The same
/// stance the double takes on membership events, which it does not implement
/// at all.
fn require_attestation(
    target: &str,
    caller: &str,
    holders: &[String],
    adding: &str,
    attestation: Option<&SignedKeyAttestation>,
) -> ControlResult<()> {
    if holders.is_empty() || (!caller.is_empty() && caller == target) {
        return Ok(());
    }
    let Some(signed) = attestation else {
        return Err(ControlError::FailedPrecondition(format!(
            "principal {target} already holds {} key(s), so adding another needs {target}'s own \
             attestation over the new key's fingerprint ({adding})",
            holders.len()
        )));
    };
    let Some(subject) = signed.attestation.as_ref() else {
        return Err(ControlError::InvalidArgument("`attestation.attestation` is required".into()));
    };
    if subject.principal_id != target || subject.fingerprint != adding {
        return Err(ControlError::InvalidArgument(format!(
            "the attestation is for key {} on {:?} but the request adds key {adding} to {target:?}",
            subject.fingerprint, subject.principal_id
        )));
    }
    if !holders.iter().any(|held| held == &signed.key_fingerprint) {
        return Err(ControlError::InvalidArgument(format!(
            "the attestation is signed by {}, which is not a key {target:?} holds",
            signed.key_fingerprint
        )));
    }
    Ok(())
}

impl ControlPlane for MemoryControlPlane {
    fn usage(&self, namespace: Option<&str>) -> ControlResult<Vec<gunnar_proto::NamespaceUsage>> {
        let inner = self.inner.lock().unwrap();
        if inner.usage.is_empty() {
            return Err(ControlError::Unimplemented("Usage"));
        }
        Ok(inner
            .usage
            .values()
            .filter(|row| namespace.is_none_or(|ns| row.namespace == ns))
            .cloned()
            .collect())
    }

    fn is_twin(&self) -> bool {
        self.inner.lock().expect("memory control plane").twin
    }

    fn log_follows(&self) -> bool {
        let inner = self.inner.lock().expect("memory control plane");
        inner.twin || inner.log_follows
    }

    /// The fact this box holds, or `None` = Free.
    ///
    /// The double answers this one rather than refusing, because it now has
    /// somewhere to keep a fact ([`MemoryControlPlane::set_entitlement_fact`])
    /// and "no fact" is a real answer and not a missing one. `entitlement_set`
    /// stays `Unimplemented`: storing is not the half the double is missing,
    /// VERIFYING is, and a double that stored an unverified fact would be more
    /// permissive than the thing it stands in for.
    #[cfg(feature = "metered")]
    fn entitlement_get(
        &self,
        namespace: &str,
    ) -> ControlResult<Option<gunnar_proto::EntitlementFact>> {
        Ok(self
            .inner
            .lock()
            .unwrap()
            .entitlements
            .get(namespace)
            .cloned())
    }

    #[cfg(feature = "metered")]
    fn entitlement_list(&self) -> ControlResult<Vec<gunnar_proto::EntitlementFact>> {
        Ok(self
            .inner
            .lock()
            .unwrap()
            .entitlements
            .values()
            .cloned()
            .collect())
    }

    /// The double's actor-ticket key, minted on first use and stable after —
    /// which is what the real store's `get_or_create_with` gives the appliance,
    /// reproduced here under this double's one `Mutex`.
    #[cfg(feature = "metered")]
    fn actor_ticket_public_key_hex(&self) -> ControlResult<String> {
        let mut inner = self.inner.lock().unwrap();
        let key = Inner::actor_ticket_key(&mut inner);
        Ok(key
            .verifying_key()
            .to_bytes()
            .iter()
            .map(|byte| format!("{byte:02x}"))
            .collect())
    }

    #[cfg(feature = "metered")]
    fn actor_ticket_sign(&self, message: &[u8]) -> ControlResult<[u8; 64]> {
        use ed25519_dalek::Signer as _;
        let mut inner = self.inner.lock().unwrap();
        let signature = Inner::actor_ticket_key(&mut inner).sign(message).to_bytes();
        // Counted AFTER the signature exists, so the number means "a ticket
        // this box signed" and not "a call that was attempted".
        inner.actor_tickets_signed += 1;
        Ok(signature)
    }

    fn create_namespace(&self, name: &str, description: &str) -> ControlResult<Namespace> {
        if name.is_empty() {
            return Err(ControlError::InvalidArgument("namespace is empty".into()));
        }
        let mut inner = self.inner.lock().unwrap();
        if inner.namespaces.contains_key(name) {
            return Err(ControlError::AlreadyExists(format!("namespace {name}")));
        }
        let namespace = Namespace {
            name: name.to_string(),
            description: description.to_string(),
            created_at_unix_ms: now_ms(),
            owner_kind: 0,
            ..Default::default()
        };
        inner.namespaces.insert(name.to_string(), namespace.clone());
        Ok(namespace)
    }

    fn list_namespaces(&self) -> ControlResult<Vec<Namespace>> {
        let inner = self.inner.lock().unwrap();
        Ok(inner
            .namespaces
            .values()
            .cloned()
            .map(|mut ns| {
                ns.employees = inner
                    .employers
                    .values()
                    .filter(|account| **account == ns.name)
                    .count() as u32;
                ns
            })
            .collect())
    }

    fn set_namespace_settings(
        &self,
        name: &str,
        public: Option<bool>,
        events_public: Option<bool>,
    ) -> ControlResult<Namespace> {
        {
            let mut inner = self.inner.lock().unwrap();
            let ns = inner
                .namespaces
                .get_mut(name)
                .ok_or_else(|| ControlError::NotFound(format!("namespace {name}")))?;
            if let Some(public) = public {
                ns.public = public;
            }
            if let Some(events_public) = events_public {
                ns.events_public = events_public;
            }
        }
        Ok(self
            .namespace(name)?
            .expect("the namespace was just written"))
    }

    fn delete_namespace(&self, name: &str, delete_repos: bool) -> ControlResult<u32> {
        let mut inner = self.inner.lock().unwrap();
        if !inner.namespaces.contains_key(name) {
            return Err(ControlError::NotFound(format!("namespace {name}")));
        }
        let held: Vec<(String, String)> = inner
            .repos
            .keys()
            .filter(|(ns, _)| ns == name)
            .cloned()
            .collect();
        if !held.is_empty() && !delete_repos {
            return Err(ControlError::FailedPrecondition(format!(
                "namespace {name} holds {} repos; deleting repos is never implicit",
                held.len()
            )));
        }
        for key in &held {
            inner.repos.remove(key);
            inner.mirrors.remove(key);
            inner.protections.remove(key);
            inner.ref_policies.remove(key);
        }
        inner.namespaces.remove(name);
        inner.grants.retain(|g| g.namespace != name);
        Ok(held.len() as u32)
    }

    fn create_repo(&self, spec: NewRepo) -> ControlResult<Repo> {
        spec.path
            .validate()
            .map_err(|e| ControlError::InvalidArgument(e.to_string()))?;
        let mut inner = self.inner.lock().unwrap();
        if !inner.namespaces.contains_key(&spec.path.namespace) {
            return Err(ControlError::NotFound(format!(
                "namespace {}",
                spec.path.namespace
            )));
        }
        let key = MemoryControlPlane::key(&spec.path);
        if inner.repos.contains_key(&key) {
            return Err(ControlError::AlreadyExists(format!(
                "repository {}",
                spec.path
            )));
        }
        let hash_kind = if spec.hash_kind == HashKind::Unspecified {
            DEFAULT_HASH_KIND
        } else {
            spec.hash_kind
        };
        let default_branch = if spec.default_branch.is_empty() {
            DEFAULT_BRANCH.to_string()
        } else {
            spec.default_branch.clone()
        };
        let visibility = if spec.visibility == Visibility::Unspecified {
            Visibility::Private
        } else {
            spec.visibility
        };
        let repo = Repo {
            path: Some(spec.path.clone()),
            hash_kind: hash_kind as i32,
            default_branch,
            visibility: visibility as i32,
            created_at_unix_ms: now_ms(),
            description: spec.description,
            encrypted: false,
        };
        inner.repos.insert(key, repo.clone());
        Ok(repo)
    }

    fn get_repo(&self, path: &RepoPath) -> ControlResult<Repo> {
        self.inner
            .lock()
            .unwrap()
            .repos
            .get(&MemoryControlPlane::key(path))
            .cloned()
            .ok_or_else(|| ControlError::NotFound(format!("repository {path}")))
    }

    fn list_repos(&self, namespace: Option<&str>) -> ControlResult<Vec<Repo>> {
        Ok(self
            .inner
            .lock()
            .unwrap()
            .repos
            .iter()
            .filter(|((ns, _), _)| namespace.is_none_or(|want| want == ns))
            .map(|(_, repo)| repo.clone())
            .collect())
    }

    /// The test double holds no bytes, so both arms of the real verb collapse
    /// to removing the row — but the OUTCOME still says which was asked for,
    /// because that is the fact the service reports and a double that always
    /// said `Purged` would let a soft-delete regression pass.
    fn delete_repo(&self, path: &RepoPath, options: DeleteOptions) -> ControlResult<DeleteOutcome> {
        let key = MemoryControlPlane::key(path);
        let mut inner = self.inner.lock().unwrap();
        if inner.repos.remove(&key).is_none() {
            return Err(ControlError::NotFound(format!("repository {path}")));
        }
        inner.mirrors.remove(&key);
        inner.protections.remove(&key);
        inner.ref_policies.remove(&key);
        inner
            .grants
            .retain(|g| !(g.namespace == path.namespace && g.repo == path.repo));
        Ok(if options.purge {
            DeleteOutcome::Purged { bytes_reclaimed: 0 }
        } else {
            DeleteOutcome::Tombstoned {
                purge_after_unix_ms: now_ms()
                    + i64::from(options.retention_days.unwrap_or(30)) * 86_400_000,
                bytes_kept: 0,
            }
        })
    }

    fn set_default_branch(&self, path: &RepoPath, default_branch: &str) -> ControlResult<Repo> {
        if default_branch.is_empty() {
            return Err(ControlError::InvalidArgument(
                "default_branch is empty".into(),
            ));
        }
        let mut inner = self.inner.lock().unwrap();
        let repo = inner
            .repos
            .get_mut(&MemoryControlPlane::key(path))
            .ok_or_else(|| ControlError::NotFound(format!("repository {path}")))?;
        repo.default_branch = default_branch.to_string();
        Ok(repo.clone())
    }

    fn set_visibility(&self, path: &RepoPath, visibility: Visibility) -> ControlResult<Repo> {
        if visibility == Visibility::Unspecified {
            return Err(ControlError::InvalidArgument(
                "visibility unspecified".into(),
            ));
        }
        let mut inner = self.inner.lock().unwrap();
        let repo = inner
            .repos
            .get_mut(&MemoryControlPlane::key(path))
            .ok_or_else(|| ControlError::NotFound(format!("repository {path}")))?;
        repo.visibility = visibility as i32;
        Ok(repo.clone())
    }

    fn set_encrypted(&self, path: &RepoPath, encrypted: bool) -> ControlResult<Repo> {
        let mut inner = self.inner.lock().unwrap();
        let repo = inner
            .repos
            .get_mut(&MemoryControlPlane::key(path))
            .ok_or_else(|| ControlError::NotFound(format!("repository {path}")))?;
        repo.encrypted = encrypted;
        Ok(repo.clone())
    }

    fn list_refs(
        &self,
        path: &RepoPath,
        _prefix: &str,
        _peel: bool,
    ) -> ControlResult<Vec<RefTarget>> {
        self.get_repo(path)?;
        Err(ControlError::Unimplemented("Refs.List"))
    }

    fn get_ref(&self, path: &RepoPath, _ref_name: &str, _peel: bool) -> ControlResult<RefTarget> {
        self.get_repo(path)?;
        Err(ControlError::Unimplemented("Refs.Get"))
    }

    fn log(
        &self,
        path: &RepoPath,
        _start: &str,
        _limit: u32,
        _touching: &str,
    ) -> ControlResult<Vec<Commit>> {
        self.get_repo(path)?;
        Err(ControlError::Unimplemented("Refs.Log"))
    }

    fn create_pull(
        &self,
        path: &RepoPath,
        _author: &str,
        _title: &str,
        _description: &str,
        _source_ref: &str,
        _target_ref: &str,
    ) -> ControlResult<PullRequest> {
        self.get_repo(path)?;
        Err(ControlError::Unimplemented("PullRequests.Create"))
    }

    fn list_pulls(&self, path: &RepoPath, _state: i32) -> ControlResult<Vec<PullRequest>> {
        self.get_repo(path)?;
        Err(ControlError::Unimplemented("PullRequests.List"))
    }

    fn get_pull(&self, path: &RepoPath, _number: u64) -> ControlResult<PullRequest> {
        self.get_repo(path)?;
        Err(ControlError::Unimplemented("PullRequests.Get"))
    }

    fn close_pull(&self, path: &RepoPath, _number: u64) -> ControlResult<PullRequest> {
        self.get_repo(path)?;
        Err(ControlError::Unimplemented("PullRequests.Close"))
    }

    fn merge_pull(
        &self,
        path: &RepoPath,
        _number: u64,
        _principal_id: &str,
    ) -> ControlResult<(PullRequest, MergedRef)> {
        self.get_repo(path)?;
        Err(ControlError::Unimplemented("PullRequests.Merge"))
    }

    fn read_object(
        &self,
        path: &RepoPath,
        _oid: &str,
        _max_bytes: u64,
    ) -> ControlResult<ReadObjectResponse> {
        self.get_repo(path)?;
        Err(ControlError::Unimplemented("Refs.ReadObject"))
    }

    fn diff(
        &self,
        path: &RepoPath,
        _from: &str,
        _to: &str,
        _path_prefixes: &[String],
    ) -> ControlResult<Vec<FileChange>> {
        self.get_repo(path)?;
        Err(ControlError::Unimplemented("Refs.Diff"))
    }

    fn add_principal(&self, id: &str, name: &str, kind: PrincipalKind) -> ControlResult<Principal> {
        if name.is_empty() || id.is_empty() {
            return Err(ControlError::InvalidArgument(
                "principal name and id are required".into(),
            ));
        }
        let mut inner = self.inner.lock().unwrap();
        if inner.principals.contains_key(id) {
            return Err(ControlError::AlreadyExists(format!("principal {id}")));
        }
        // The same rule as the real table: a name is taken by another name
        // AND by another id.
        if let Some(holder) = Inner::name_holder(&inner, name, id) {
            return Err(ControlError::AlreadyExists(format!(
                "the name \"{name}\" is already taken by principal {holder}"
            )));
        }
        let principal = Principal {
            id: id.to_string(),
            name: name.to_string(),
            kind: kind as i32,
            created_at_unix_ms: now_ms(),
        };
        inner.principals.insert(id.to_string(), principal.clone());
        Ok(principal)
    }

    fn enroll_employee(&self, account: &str, ask: &EmployeeAsk) -> ControlResult<EmployeeAnswer> {
        let named = ask.seats == 0;
        if named && (ask.name.is_empty() || ask.id.is_empty()) {
            return Err(ControlError::InvalidArgument(
                "an employee's id and name are required".into(),
            ));
        }
        let mut inner = self.inner.lock().unwrap();
        if !inner.namespaces.contains_key(account) {
            return Err(ControlError::NotFound(format!("account {account}")));
        }
        let open = |lots: &[(u64, u32, Vec<String>, String)]| -> u32 {
            lots.iter()
                .map(|(_, total, filled, _)| total.saturating_sub(filled.len() as u32))
                .sum()
        };
        let lots = inner.seats.get(account).cloned().unwrap_or_default();
        let total: u32 = lots.iter().map(|(_, total, _, _)| *total).sum();
        if !ask.order.is_empty() {
            if let Some((owner, principal, _, room, need)) = inner.room_orders.get(&ask.order).cloned() {
                if owner != account {
                    return Err(ControlError::InvalidArgument(format!(
                        "order {} was granted to {owner}, not to {account}",
                        ask.order
                    )));
                }
                return Ok(EmployeeAnswer {
                    principal: inner.principals.get(&principal).cloned(),
                    room_bytes: room,
                    seats: total,
                    seats_open: open(&lots),
                    need_bytes: need,
                    unused_bytes: None,
                    shortfall_bytes: 0,
                    unchanged: true,
                });
            }
        }
        if named {
            if ask.id == account {
                return Err(ControlError::InvalidArgument(format!(
                    "{account} is the account itself, not an employee of it"
                )));
            }
            if inner.principals.contains_key(&ask.id)
                || inner.principals.values().any(|p| p.name == ask.name)
            {
                return Err(ControlError::AlreadyExists(format!(
                    "principal {} already exists, and an existing principal can never become \
                     an employee",
                    ask.id
                )));
            }
            if inner.namespaces.contains_key(&ask.id) {
                return Err(ControlError::AlreadyExists(format!(
                    "namespace {} exists, so an employee may not be called that",
                    ask.id
                )));
            }
        }
        let lot = if ask.from_seat {
            Some(
                lots.iter()
                    .position(|(_, total, filled, _)| (filled.len() as u32) < *total)
                    .ok_or_else(|| {
                        ControlError::FailedPrecondition(format!(
                            "{account} has no open seat: every seat it holds is filled. Its \
                             admin adds users first"
                        ))
                    })?,
            )
        } else {
            None
        };
        let room_bytes = lot.map_or(ask.room_bytes, |lot| lots[lot].0);
        let need_bytes = match (lot, ask.seats) {
            (Some(_), _) => 0,
            (None, 0) => room_bytes,
            (None, seats) => room_bytes.saturating_mul(u64::from(seats)),
        };
        let committed = ask.cap.map(|_| {
            let promised: u64 = lots
                .iter()
                .map(|(room, total, filled, _)| room * u64::from(total.saturating_sub(filled.len() as u32)))
                .sum::<u64>()
                + inner
                    .employers
                    .iter()
                    .filter(|(_, employer)| *employer == account)
                    .filter_map(|(id, _)| inner.rooms.get(id).map(|(room, _)| *room))
                    .sum::<u64>();
            promised.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));
        let mut answer = EmployeeAnswer {
            principal: None,
            room_bytes,
            seats: total,
            seats_open: open(&lots),
            need_bytes,
            unused_bytes,
            shortfall_bytes,
            unchanged: ask.dry_run || shortfall_bytes > 0,
        };
        if answer.unchanged {
            return Ok(answer);
        }
        let mut lots = lots;
        if named {
            let principal = Principal {
                id: ask.id.clone(),
                name: ask.name.clone(),
                kind: PrincipalKind::User as i32,
                created_at_unix_ms: now_ms(),
            };
            inner.principals.insert(ask.id.clone(), principal.clone());
            inner.employers.insert(ask.id.clone(), account.to_owned());
            inner.grants.push(Grant {
                principal_id: ask.id.clone(),
                namespace: account.to_owned(),
                repo: String::new(),
                level: GrantLevel::Read as i32,
                ..Default::default()
            });
            let order = match lot {
                Some(lot) => {
                    lots[lot].2.push(ask.id.clone());
                    lots[lot].3.clone()
                }
                None => ask.order.clone(),
            };
            if room_bytes > 0 {
                inner.rooms.insert(ask.id.clone(), (room_bytes, order));
            }
            answer.principal = Some(principal);
        } else {
            lots.push((room_bytes, ask.seats, Vec::new(), ask.order.clone()));
        }
        answer.seats = lots.iter().map(|(_, total, _, _)| *total).sum();
        answer.seats_open = open(&lots);
        if !lots.is_empty() {
            inner.seats.insert(account.to_owned(), lots);
        }
        if !ask.order.is_empty() {
            inner.room_orders.insert(
                ask.order.clone(),
                (
                    account.to_owned(),
                    if named { ask.id.clone() } else { String::new() },
                    ask.seats,
                    room_bytes,
                    need_bytes,
                ),
            );
        }
        Ok(answer)
    }

    fn seats(&self, account: &str) -> ControlResult<(u32, u32)> {
        let inner = self.inner.lock().unwrap();
        let lots = inner.seats.get(account).cloned().unwrap_or_default();
        Ok((
            lots.iter().map(|(_, total, _, _)| *total).sum(),
            lots.iter()
                .map(|(_, total, filled, _)| total.saturating_sub(filled.len() as u32))
                .sum(),
        ))
    }

    fn employees(&self, account: &str) -> ControlResult<Vec<Principal>> {
        let inner = self.inner.lock().unwrap();
        Ok(inner
            .employers
            .iter()
            .filter(|(_, employer)| *employer == account)
            .filter_map(|(id, _)| inner.principals.get(id).cloned())
            .collect())
    }

    /// The double keeps no feed: the operator is the ASKED name, by the same
    /// [`operator_name`] rule (a plane-level test of the fed half lives where
    /// the feed is, in `gunnar-server`).
    /// The double keeps the same rule as the shipped plane: the operator must
    /// be a person; once he holds a passkey the bootstrap is spent (and a
    /// leftover one withdrawn); otherwise the invite is (re)recorded.
    fn operator_bootstrap_invite(
        &self,
        operator: &str,
        secret_sha256: &str,
        now_unix_ms: i64,
        ttl_ms: i64,
    ) -> ControlResult<BootstrapInvite> {
        let hash = bootstrap_hash(secret_sha256)?;
        let ttl_ms = bootstrap_ttl(ttl_ms)?;
        let mut inner = self.inner.lock().unwrap();
        let principal = inner
            .principals
            .get(operator)
            .cloned()
            .ok_or_else(|| ControlError::NotFound(format!("principal {operator}")))?;
        if principal.kind != PrincipalKind::User as i32 {
            return Err(ControlError::FailedPrecondition(format!(
                "principal {operator} is not a person: the operator's first invite is a person's"
            )));
        }
        if inner
            .credentials
            .values()
            .any(|stored| stored.credential.principal_id == operator)
        {
            inner.invites.remove(OPERATOR_INVITE_ID);
            return Ok(BootstrapInvite::Spent);
        }
        let renewed = inner
            .invites
            .get(OPERATOR_INVITE_ID)
            .is_some_and(|(row, held)| *held == hash && row.principal_id == operator);
        let row = gunnar_proto::Invite {
            invite_id: OPERATOR_INVITE_ID.to_owned(),
            principal_id: operator.to_owned(),
            created_by: "installer".to_owned(),
            created_at_unix_ms: now_unix_ms,
            expires_at_unix_ms: now_unix_ms + ttl_ms,
            note: "the operator's first passkey: the installer's first invite".to_owned(),
            server_level: GrantLevel::Unspecified as i32,
        };
        inner.invites.insert(OPERATOR_INVITE_ID.to_owned(), (row, hash));
        Ok(if renewed { BootstrapInvite::Renewed } else { BootstrapInvite::Recorded })
    }
    fn claim_operator(&self, asked: &str) -> ControlResult<OperatorClaim> {
        let (name, named_by) = operator_name(None, asked)?;
        let held = {
            let inner = self.inner.lock().unwrap();
            inner
                .principals
                .get(name)
                .cloned()
                .or_else(|| inner.principals.values().find(|p| p.name == name).cloned())
        };
        let (principal, created) = match held {
            Some(p) if p.kind != PrincipalKind::User as i32 => {
                return Err(ControlError::FailedPrecondition(format!(
                    "the operator's name {name:?} is held by a principal that is not a person"
                )))
            }
            Some(p) => (p, false),
            None => (self.add_principal(name, name, PrincipalKind::User)?, true),
        };
        let mut inner = self.inner.lock().unwrap();
        let adopted_namespace = created && inner.namespaces.contains_key(&principal.id);
        if !inner.namespaces.contains_key(&principal.id) {
            inner.namespaces.insert(
                principal.id.clone(),
                Namespace {
                    name: principal.id.clone(),
                    created_at_unix_ms: now_ms(),
                    ..Default::default()
                },
            );
        }
        inner
            .server_grants
            .insert(principal.id.clone(), GrantLevel::Admin);
        Ok(OperatorClaim {
            principal,
            created,
            adopted_namespace,
            named_by,
        })
    }

    fn list_principals(&self) -> ControlResult<Vec<Principal>> {
        Ok(self
            .inner
            .lock()
            .unwrap()
            .principals
            .values()
            .cloned()
            .collect())
    }

    fn principal_by_name(&self, name: &str) -> ControlResult<Option<Principal>> {
        Ok(self
            .inner
            .lock()
            .unwrap()
            .principals
            .values()
            .find(|p| p.name == name)
            .cloned())
    }

    fn add_credential(
        &self,
        principal_id: &str,
        credential_id: &[u8],
        public_key: &[u8],
        label: &str,
        sign_count: u32,
    ) -> ControlResult<gunnar_proto::WebCredential> {
        let mut inner = self.inner.lock().unwrap();
        let principal = inner
            .principals
            .get(principal_id)
            .cloned()
            .ok_or_else(|| ControlError::NotFound(format!("principal {principal_id}")))?;
        if principal.kind != PrincipalKind::User as i32 {
            return Err(ControlError::FailedPrecondition(format!(
                "principal {principal_id} is not a person: no human credential on a machine principal"
            )));
        }
        let key = crate::webauthn::b64url(credential_id);
        if inner.credentials.contains_key(&key) {
            return Err(ControlError::AlreadyExists(format!("web credential {key}")));
        }
        let credential = gunnar_proto::WebCredential {
            credential_id: credential_id.to_vec(),
            principal_id: principal_id.to_owned(),
            label: label.to_owned(),
            created_at_unix_ms: now_ms(),
            last_used_unix_ms: None,
            sign_count,
        };
        inner.credentials.insert(
            key,
            StoredCredential {
                credential: credential.clone(),
                public_key: public_key.to_vec(),
            },
        );
        Ok(credential)
    }

    fn credentials(&self, principal_id: Option<&str>) -> ControlResult<Vec<gunnar_proto::WebCredential>> {
        Ok(self
            .inner
            .lock()
            .unwrap()
            .credentials
            .values()
            .filter(|c| principal_id.is_none_or(|want| c.credential.principal_id == want))
            .map(|c| c.credential.clone())
            .collect())
    }

    fn credential(&self, credential_id: &[u8]) -> ControlResult<Option<StoredCredential>> {
        Ok(self
            .inner
            .lock()
            .unwrap()
            .credentials
            .get(&crate::webauthn::b64url(credential_id))
            .cloned())
    }

    fn revoke_credential(&self, credential_id: &[u8]) -> ControlResult<Option<gunnar_proto::WebCredential>> {
        let key = crate::webauthn::b64url(credential_id);
        let mut inner = self.inner.lock().unwrap();
        let removed = inner.credentials.remove(&key).map(|c| c.credential);
        // Its session keys go with it.
        for fingerprint in inner.credential_sessions.remove(&key).unwrap_or_default() {
            inner.keys.remove(&fingerprint);
            inner.openssh.remove(&fingerprint);
        }
        Ok(removed)
    }

    fn rename_credential(
        &self,
        credential_id: &[u8],
        label: &str,
    ) -> ControlResult<Option<gunnar_proto::WebCredential>> {
        let mut inner = self.inner.lock().unwrap();
        Ok(inner
            .credentials
            .get_mut(&crate::webauthn::b64url(credential_id))
            .map(|stored| {
                stored.credential.label = label.to_owned();
                stored.credential.clone()
            }))
    }

    fn create_invite(&self, invite: NewInvite) -> ControlResult<gunnar_proto::Invite> {
        let mut inner = self.inner.lock().unwrap();
        let principal = inner
            .principals
            .get(&invite.principal_id)
            .cloned()
            .ok_or_else(|| ControlError::NotFound(format!("principal {}", invite.principal_id)))?;
        if principal.kind != PrincipalKind::User as i32 {
            return Err(ControlError::FailedPrecondition(format!(
                "principal {} is not a person: no human credential on a machine principal",
                invite.principal_id
            )));
        }
        if inner.invites.contains_key(&invite.id) {
            return Err(ControlError::AlreadyExists(format!("invite {}", invite.id)));
        }
        let row = gunnar_proto::Invite {
            invite_id: invite.id.clone(),
            principal_id: invite.principal_id,
            created_by: invite.created_by,
            created_at_unix_ms: invite.created_at_unix_ms,
            expires_at_unix_ms: invite.expires_at_unix_ms,
            note: invite.note,
            server_level: invite.server_level as i32,
        };
        inner.invites.insert(invite.id, (row.clone(), invite.secret_hash));
        Ok(row)
    }

    fn invites(&self) -> ControlResult<Vec<gunnar_proto::Invite>> {
        let mut rows: Vec<gunnar_proto::Invite> =
            self.inner.lock().unwrap().invites.values().map(|(row, _)| row.clone()).collect();
        rows.sort_by(|a, b| (a.created_at_unix_ms, &a.invite_id).cmp(&(b.created_at_unix_ms, &b.invite_id)));
        Ok(rows)
    }

    fn revoke_invite(&self, id: &str) -> ControlResult<Option<gunnar_proto::Invite>> {
        Ok(self.inner.lock().unwrap().invites.remove(id).map(|(row, _)| row))
    }

    fn check_invite(&self, secret_hash: &str, now_unix_ms: i64) -> ControlResult<InviteCheck> {
        let inner = self.inner.lock().unwrap();
        if let Some((row, _)) = inner.invites.values().find(|(_, held)| held == secret_hash) {
            return Ok(if row.expires_at_unix_ms > now_unix_ms { InviteCheck::Valid(row.clone()) } else { InviteCheck::Expired });
        }
        Ok(if inner.redeemed_invites.contains(secret_hash) { InviteCheck::Used } else { InviteCheck::Unknown })
    }

    fn holds_server_grant(&self, principal_id: &str) -> ControlResult<bool> {
        Ok(self.inner.lock().unwrap().server_grants.contains_key(principal_id))
    }

    fn employer_of(&self, principal_id: &str) -> ControlResult<Option<String>> {
        Ok(self.inner.lock().unwrap().employers.get(principal_id).cloned())
    }

    fn web_login(&self, commit: WebLoginCommit) -> ControlResult<WebLoginOutcome> {
        let key = crate::webauthn::b64url(&commit.credential_id);
        let mut fields = commit.session_openssh.split_whitespace();
        let algo = fields.next().unwrap_or_default().to_owned();
        let fingerprint = fingerprint_of_openssh(&commit.session_openssh).ok_or_else(|| {
            ControlError::InvalidArgument("the session key is not an OpenSSH public key".into())
        })?;
        let mut inner = self.inner.lock().unwrap();
        let mut stored = inner
            .credentials
            .get(&key)
            .cloned()
            .ok_or_else(|| ControlError::NotFound(format!("web credential {key}")))?;
        let principal = inner
            .principals
            .get(&stored.credential.principal_id)
            .cloned()
            .ok_or_else(|| ControlError::NotFound(format!("principal {}", stored.credential.principal_id)))?;
        let held = stored.credential.sign_count;
        if held != 0 && commit.sign_count != 0 && commit.sign_count <= held {
            return Err(ControlError::PermissionDenied(format!(
                "the signature counter of web credential {key} went from {held} to {}: a cloned \
                 authenticator, or a replay",
                commit.sign_count
            )));
        }
        if inner.keys.contains_key(&fingerprint) {
            return Err(ControlError::AlreadyExists(format!("key {fingerprint}")));
        }
        if commit.sign_count != 0 {
            stored.credential.sign_count = commit.sign_count;
        }
        stored.credential.last_used_unix_ms = Some(commit.now_unix_ms);
        inner.credentials.insert(key.clone(), stored.clone());
        inner
            .credential_sessions
            .entry(key)
            .or_default()
            .push(fingerprint.clone());
        inner
            .openssh
            .insert(fingerprint.clone(), commit.session_openssh.trim().to_owned());
        inner.keys.insert(
            fingerprint.clone(),
            PublicKey {
                fingerprint: fingerprint.clone(),
                algo,
                principal_id: principal.id.clone(),
                comment: commit.comment,
                added_at_unix_ms: commit.now_unix_ms,
                last_used_unix_ms: None,
                expires_at_unix_ms: Some(commit.expires_at_unix_ms),
            },
        );
        Ok(WebLoginOutcome {
            principal,
            credential: stored.credential,
            fingerprint,
        })
    }

    fn web_register(&self, registration: WebRegistration) -> ControlResult<WebRegistered> {
        const INVALID: &str =
            "this invite is not valid: it is unknown, already used, expired, or for somebody else";
        let mut inner = self.inner.lock().unwrap();
        let (target, level, redeemed) = match &registration.invite_secret_hash {
            Some(hash) => {
                let (row, _) = inner
                    .invites
                    .values()
                    .find(|(row, held)| held == hash && row.expires_at_unix_ms > registration.now_unix_ms)
                    .cloned()
                    .ok_or_else(|| ControlError::PermissionDenied(INVALID.into()))?;
                if registration
                    .principal_id
                    .as_deref()
                    .is_some_and(|asked| asked != row.principal_id)
                {
                    return Err(ControlError::PermissionDenied(INVALID.into()));
                }
                (
                    row.principal_id.clone(),
                    GrantLevel::try_from(row.server_level).unwrap_or(GrantLevel::Unspecified),
                    Some(row.invite_id),
                )
            }
            None => (
                registration.principal_id.clone().ok_or_else(|| {
                    ControlError::InvalidArgument("name the principal, or redeem an invite".into())
                })?,
                GrantLevel::Unspecified,
                None,
            ),
        };
        let principal = inner
            .principals
            .get(&target)
            .cloned()
            .ok_or_else(|| ControlError::NotFound(format!("principal {target}")))?;
        if principal.kind != PrincipalKind::User as i32 {
            return Err(ControlError::FailedPrecondition(format!(
                "principal {target} is not a person: no human credential on a machine principal"
            )));
        }
        let key = crate::webauthn::b64url(&registration.credential_id);
        if inner.credentials.contains_key(&key) {
            return Err(ControlError::AlreadyExists(format!("web credential {key}")));
        }
        let credential = gunnar_proto::WebCredential {
            credential_id: registration.credential_id.clone(),
            principal_id: target.clone(),
            label: registration.label,
            created_at_unix_ms: registration.now_unix_ms,
            last_used_unix_ms: None,
            sign_count: registration.sign_count,
        };
        inner.credentials.insert(
            key,
            StoredCredential {
                credential: credential.clone(),
                public_key: registration.public_key,
            },
        );
        if let Some(id) = redeemed {
            inner.invites.remove(&id);
            if let Some(hash) = registration.invite_secret_hash.clone() {
                inner.redeemed_invites.insert(hash);
            }
        }
        if level != GrantLevel::Unspecified {
            inner.server_grants.insert(target, level);
        }
        Ok(WebRegistered {
            credential,
            principal,
            server_level: level,
        })
    }

    /// The double keeps the shipped plane's rule, in one lock: the machine
    /// made or found (a person under the id refused), the grant, the
    /// credential, the approver's counter.
    fn approve_machine_enrolment(&self, approval: MachineApproval) -> ControlResult<MachineApproved> {
        let mut inner = self.inner.lock().unwrap();
        let key = crate::webauthn::b64url(&approval.credential_id);
        if inner.credentials.contains_key(&key) {
            return Err(ControlError::AlreadyExists(format!("web credential {key}")));
        }
        let mut step_up = None;
        if let Some((id, presented)) = &approval.step_up {
            let held = crate::webauthn::b64url(id);
            let stored = inner
                .credentials
                .get(&held)
                .cloned()
                .ok_or_else(|| ControlError::NotFound(format!("web credential {held}")))?;
            let count = stored.credential.sign_count;
            if count != 0 && *presented != 0 && *presented <= count {
                return Err(ControlError::PermissionDenied(format!(
                    "the signature counter of web credential {held} went from {count} to {presented}: \
                     a cloned authenticator, or a replay"
                )));
            }
            step_up = Some((held, stored, *presented));
        }
        let (principal, created) = match inner.principals.get(&approval.machine).cloned() {
            Some(p) if p.kind != PrincipalKind::Machine as i32 => {
                return Err(ControlError::FailedPrecondition(format!(
                    "principal {} is not a machine: a machine's credential never lands on a person",
                    approval.machine
                )))
            }
            Some(p) => (p, false),
            None => {
                if let Some(holder) = Inner::name_holder(&inner, &approval.machine, &approval.machine) {
                    return Err(ControlError::AlreadyExists(format!(
                        "the name \"{}\" is already taken by principal {holder}",
                        approval.machine
                    )));
                }
                let p = Principal {
                    id: approval.machine.clone(),
                    name: approval.machine.clone(),
                    kind: PrincipalKind::Machine as i32,
                    created_at_unix_ms: approval.now_unix_ms,
                };
                inner.principals.insert(p.id.clone(), p.clone());
                (p, true)
            }
        };
        if let Some((held, mut stored, presented)) = step_up {
            if presented != 0 {
                stored.credential.sign_count = presented;
            }
            stored.credential.last_used_unix_ms = Some(approval.now_unix_ms);
            inner.credentials.insert(held, stored);
        }
        inner
            .server_grants
            .insert(principal.id.clone(), approval.server_level);
        let credential = gunnar_proto::WebCredential {
            credential_id: approval.credential_id.clone(),
            principal_id: principal.id.clone(),
            label: approval.label,
            created_at_unix_ms: approval.now_unix_ms,
            last_used_unix_ms: None,
            sign_count: approval.sign_count,
        };
        inner.credentials.insert(
            key,
            StoredCredential {
                credential: credential.clone(),
                public_key: approval.public_key,
            },
        );
        Ok(MachineApproved {
            principal,
            credential,
            created,
        })
    }

    fn rename_principal(&self, principal_id: &str, name: &str) -> ControlResult<Principal> {
        if name.is_empty() {
            return Err(ControlError::InvalidArgument("the new name is empty".into()));
        }
        let mut inner = self.inner.lock().unwrap();
        if !inner.principals.contains_key(principal_id) {
            return Err(ControlError::NotFound(format!("principal {principal_id}")));
        }
        if let Some(holder) = Inner::name_holder(&inner, name, principal_id) {
            return Err(ControlError::AlreadyExists(format!(
                "the name \"{name}\" is already taken by principal {holder}"
            )));
        }
        let row = inner.principals.get_mut(principal_id).expect("checked above");
        row.name = name.to_string();
        Ok(row.clone())
    }

    fn remove_principal(&self, principal_id: &str) -> ControlResult<(u32, u32)> {
        let mut inner = self.inner.lock().unwrap();
        inner.employers.remove(principal_id);
        if inner.principals.remove(principal_id).is_none() {
            return Err(ControlError::NotFound(format!("principal {principal_id}")));
        }
        let before_keys = inner.keys.len();
        inner.keys.retain(|_, k| k.principal_id != principal_id);
        let live: Vec<String> = inner.keys.keys().cloned().collect();
        inner.openssh.retain(|fp, _| live.contains(fp));
        let before_grants = inner.grants.len();
        inner.grants.retain(|g| g.principal_id != principal_id);
        inner.server_grants.remove(principal_id);
        inner
            .credentials
            .retain(|_, c| c.credential.principal_id != principal_id);
        inner
            .invites
            .retain(|_, (row, _)| row.principal_id != principal_id);
        Ok((
            (before_keys - inner.keys.len()) as u32,
            (before_grants - inner.grants.len()) as u32,
        ))
    }

    fn add_key(
        &self,
        principal_id: &str,
        openssh_public_key: &str,
        comment: &str,
        expires_at_unix_ms: Option<i64>,
        caller: &str,
        attestation: Option<SignedKeyAttestation>,
    ) -> ControlResult<PublicKey> {
        let mut fields = openssh_public_key.split_whitespace();
        let (Some(algo), Some(_blob)) = (fields.next(), fields.next()) else {
            return Err(ControlError::InvalidArgument(
                "not an OpenSSH public key line: expected `<algo> <base64> [comment]`".into(),
            ));
        };
        if !(algo.starts_with("ssh-") || algo.starts_with("ecdsa-") || algo.starts_with("sk-")) {
            return Err(ControlError::InvalidArgument(format!(
                "unrecognised public key algorithm {algo}"
            )));
        }
        let mut inner = self.inner.lock().unwrap();
        // Principal first, then key material: "there is no such principal" is
        // the more useful answer, and the ordering is pinned by
        // `access_rejects_the_things_that_would_break_the_ssh_side`.
        if !inner.principals.contains_key(principal_id) {
            return Err(ControlError::NotFound(format!("principal {principal_id}")));
        }
        let fingerprint = fingerprint_of_openssh(openssh_public_key).ok_or_else(|| {
            ControlError::InvalidArgument(format!(
                "the {algo} key does not parse as an OpenSSH public key line; the fingerprint \
                 the SSH listener will match against cannot be computed from it"
            ))
        })?;
        if inner.keys.contains_key(&fingerprint) {
            return Err(ControlError::AlreadyExists(format!("key {fingerprint}")));
        }
        let holders: Vec<String> = inner
            .keys
            .values()
            .filter(|key| key.principal_id == principal_id)
            .filter(|key| key.expires_at_unix_ms.is_none_or(|expires| now_ms() < expires))
            .map(|key| key.fingerprint.clone())
            .collect();
        require_attestation(principal_id, caller, &holders, &fingerprint, attestation.as_ref())?;
        inner
            .openssh
            .insert(fingerprint.clone(), openssh_public_key.trim().to_owned());
        let key = PublicKey {
            fingerprint: fingerprint.clone(),
            algo: algo.to_string(),
            principal_id: principal_id.to_string(),
            comment: comment.to_string(),
            added_at_unix_ms: now_ms(),
            last_used_unix_ms: None,
            expires_at_unix_ms,
        };
        inner.keys.insert(fingerprint, key.clone());
        Ok(key)
    }

    fn revoke_key(&self, fingerprint: &str) -> ControlResult<()> {
        let mut inner = self.inner.lock().unwrap();
        // Both together, by construction: a key whose material outlived its
        // record is a credential that still verifies.
        inner.openssh.remove(fingerprint);
        inner
            .keys
            .remove(fingerprint)
            .map(|_| ())
            .ok_or_else(|| ControlError::NotFound(format!("key {fingerprint}")))
    }

    fn list_keys(&self, principal_id: Option<&str>) -> ControlResult<Vec<PublicKey>> {
        Ok(self
            .inner
            .lock()
            .unwrap()
            .keys
            .values()
            .filter(|k| principal_id.is_none_or(|want| want == k.principal_id))
            .cloned()
            .collect())
    }

    fn grant(
        &self,
        principal_id: &str,
        namespace: &str,
        repo: &str,
        level: GrantLevel,
    ) -> ControlResult<Grant> {
        if level == GrantLevel::Unspecified {
            return Err(ControlError::InvalidArgument(
                "grant level unspecified".into(),
            ));
        }
        if namespace.is_empty() {
            return Err(ControlError::InvalidArgument(
                "a grant must name a namespace; multi-tenancy has no ungrouped repos".into(),
            ));
        }
        let mut inner = self.inner.lock().unwrap();
        if !inner.principals.contains_key(principal_id) {
            return Err(ControlError::NotFound(format!("principal {principal_id}")));
        }
        if !inner.namespaces.contains_key(namespace) {
            return Err(ControlError::NotFound(format!("namespace {namespace}")));
        }
        if !repo.is_empty()
            && !inner
                .repos
                .contains_key(&(namespace.to_string(), repo.to_string()))
        {
            return Err(ControlError::NotFound(format!(
                "repository {namespace}/{repo}"
            )));
        }
        let grant = Grant {
            principal_id: principal_id.to_string(),
            namespace: namespace.to_string(),
            repo: repo.to_string(),
            level: level as i32,
            granted_at_unix_ms: now_ms(),
        };
        inner.grants.retain(|g| {
            !(g.principal_id == principal_id && g.namespace == namespace && g.repo == repo)
        });
        inner.grants.push(grant.clone());
        Ok(grant)
    }

    fn revoke(&self, principal_id: &str, namespace: &str, repo: &str) -> ControlResult<()> {
        let mut inner = self.inner.lock().unwrap();
        let before = inner.grants.len();
        inner.grants.retain(|g| {
            !(g.principal_id == principal_id && g.namespace == namespace && g.repo == repo)
        });
        if inner.grants.len() == before {
            return Err(ControlError::NotFound(format!(
                "grant for {principal_id} on {namespace}/{repo}"
            )));
        }
        Ok(())
    }

    fn list_grants(
        &self,
        principal_id: Option<&str>,
        namespace: Option<&str>,
        repo: Option<&str>,
    ) -> ControlResult<Vec<Grant>> {
        Ok(self
            .inner
            .lock()
            .unwrap()
            .grants
            .iter()
            .filter(|g| principal_id.is_none_or(|w| w == g.principal_id))
            .filter(|g| namespace.is_none_or(|w| w == g.namespace))
            .filter(|g| repo.is_none_or(|w| w == g.repo))
            .cloned()
            .collect())
    }

    fn get_protection(
        &self,
        path: &RepoPath,
        ref_pattern: Option<&str>,
    ) -> ControlResult<Vec<BranchProtection>> {
        self.get_repo(path)?;
        let inner = self.inner.lock().unwrap();
        let Some(rules) = inner.protections.get(&MemoryControlPlane::key(path)) else {
            return Ok(Vec::new());
        };
        Ok(rules
            .values()
            .filter(|p| ref_pattern.is_none_or(|w| w == p.ref_pattern))
            .cloned()
            .collect())
    }

    fn set_protection(
        &self,
        path: &RepoPath,
        protection: BranchProtection,
    ) -> ControlResult<BranchProtection> {
        if protection.ref_pattern.is_empty() {
            return Err(ControlError::InvalidArgument("ref_pattern is empty".into()));
        }
        self.get_repo(path)?;
        let mut inner = self.inner.lock().unwrap();
        inner
            .protections
            .entry(MemoryControlPlane::key(path))
            .or_default()
            .insert(protection.ref_pattern.clone(), protection.clone());
        Ok(protection)
    }

    fn delete_protection(&self, path: &RepoPath, ref_pattern: &str) -> ControlResult<()> {
        self.get_repo(path)?;
        let mut inner = self.inner.lock().unwrap();
        let removed = inner
            .protections
            .get_mut(&MemoryControlPlane::key(path))
            .and_then(|rules| rules.remove(ref_pattern));
        if removed.is_none() {
            return Err(ControlError::NotFound(format!(
                "protection {ref_pattern} on {path}"
            )));
        }
        Ok(())
    }

    fn get_ref_name_policy(&self, path: &RepoPath) -> ControlResult<RefNamePolicy> {
        self.get_repo(path)?;
        Ok(self
            .inner
            .lock()
            .unwrap()
            .ref_policies
            .get(&MemoryControlPlane::key(path))
            .cloned()
            .unwrap_or_default())
    }

    fn set_ref_name_policy(
        &self,
        path: &RepoPath,
        policy: RefNamePolicy,
    ) -> ControlResult<RefNamePolicy> {
        self.get_repo(path)?;
        self.inner
            .lock()
            .unwrap()
            .ref_policies
            .insert(MemoryControlPlane::key(path), policy.clone());
        Ok(policy)
    }

    // ── forge ──────────────────────────────────────────────────────────────
    //
    // **Unimplemented, on purpose.** Verifying an `sshsig` is `gunnar-core`'s
    // job and this crate does not depend on it; a memory plane that appended
    // without verifying would be a second, weaker version of the one rule that
    // makes a review event evidence. `Unimplemented` is the honest answer, and
    // it is the value that must not quietly become an empty successful
    // response — an empty approval list is indistinguishable from a pull
    // request nobody has approved, and one of those may merge.
    fn forge_append(&self, _event: SignedForgeEvent) -> ControlResult<u64> {
        Err(ControlError::Unimplemented("Forge.Append"))
    }

    fn forge_review(
        &self,
        _repo: &str,
        _pull: u64,
    ) -> ControlResult<(ReviewState, Vec<SignedForgeEvent>)> {
        Err(ControlError::Unimplemented("Forge.GetReview"))
    }

    fn forge_reviews(&self, _repo: &str) -> ControlResult<Vec<ReviewState>> {
        Err(ControlError::Unimplemented("Forge.ListReviews"))
    }

    fn forge_issue(
        &self,
        _repo: &str,
        _number: u64,
    ) -> ControlResult<Option<(IssueState, Vec<SignedForgeEvent>)>> {
        Err(ControlError::Unimplemented("Forge.GetIssue"))
    }

    fn forge_issues(&self, _repo: &str) -> ControlResult<Vec<IssueState>> {
        Err(ControlError::Unimplemented("Forge.ListIssues"))
    }

    fn counts(&self) -> ControlResult<ControlCounts> {
        let inner = self.inner.lock().unwrap();
        Ok(ControlCounts {
            namespaces: inner.namespaces.len() as u64,
            repos: inner.repos.len() as u64,
            principals: inner.principals.len() as u64,
            keys: inner.keys.len() as u64,
            grants: inner.grants.len() as u64,
        })
    }

    fn storage_info(&self, path: Option<&RepoPath>) -> ControlResult<StorageInfoResponse> {
        let repos = match path {
            Some(path) => vec![self.get_repo(path)?],
            None => self.list_repos(None)?,
        };
        Ok(StorageInfoResponse {
            // The test double holds no bytes and therefore no tombstones.
            total_tombstoned_bytes: 0,
            tombstoned: Vec::new(),
            // **No machine behind this double, so no storage reality.** `None`
            // is the honest answer: a fabricated one here would let a console
            // render an installation that does not exist, which is the exact
            // failure the real implementation refuses by never inferring.
            reality: None,
            // Honest: this test double has no store behind it. It must not
            // claim "tiered(redb,znippy)" and let a UI report a backend that
            // is not there.
            object_backend: "none (MemoryControlPlane test double)".into(),
            ref_backend: "none (MemoryControlPlane test double)".into(),
            // No engine behind a test double: none is named.
            engine: None,
            total_hot_bytes: 0,
            total_cold_bytes: 0,
            // No store, so no cold tier measured.
            cold_measured: false,
            // No store behind this double, so no cache and no budget. Zero is
            // what is measured, not a plausible number.
            total_explode_bytes: 0,
            total_explode_budget_bytes: 0,
            repos: repos
                .into_iter()
                .map(|repo| RepoStorage {
                    repo: repo.path.clone(),
                    hash_kind: repo.hash_kind,
                    ..Default::default()
                })
                .collect(),
        })
    }

    fn mirror_add(
        &self,
        path: &RepoPath,
        remote_url: &str,
        prune: bool,
        create_repo: bool,
    ) -> ControlResult<MirrorConfig> {
        if remote_url.is_empty() {
            return Err(ControlError::InvalidArgument("remote_url is empty".into()));
        }
        if self.get_repo(path).is_err() {
            if !create_repo {
                return Err(ControlError::NotFound(format!("repository {path}")));
            }
            self.create_repo(NewRepo {
                path: path.clone(),
                hash_kind: HashKind::Unspecified,
                default_branch: String::new(),
                visibility: Visibility::Private,
                description: format!("mirror of {remote_url}"),
                adopt: false,
            })?;
        }
        let mut inner = self.inner.lock().unwrap();
        let key = MemoryControlPlane::key(path);
        if inner.mirrors.contains_key(&key) {
            return Err(ControlError::AlreadyExists(format!("mirror for {path}")));
        }
        let mirror = MirrorConfig {
            repo: Some(path.clone()),
            remote_url: remote_url.to_string(),
            prune,
            state: MirrorState::Idle as i32,
            added_at_unix_ms: now_ms(),
            last_sync_at_unix_ms: None,
            last_error: String::new(),
            sync_count: 0,
            credential: MirrorCredentialKind::Unspecified as i32,
            credential_username: String::new(),
            job: None,
            repo_pending: false,
            lfs_not_copied: false,
        };
        inner.mirrors.insert(key, mirror.clone());
        Ok(mirror)
    }

    fn mirror_sync(&self, path: &RepoPath) -> ControlResult<MirrorConfig> {
        let inner = self.inner.lock().unwrap();
        if !inner.mirrors.contains_key(&MemoryControlPlane::key(path)) {
            return Err(ControlError::NotFound(format!("mirror for {path}")));
        }
        // Enqueuing needs the real queue, the real client and a real store to
        // write into. Answering with a job that will never run — or with "0
        // refs changed", which is what this used to be built to avoid — would
        // be a silent green of exactly the kind that proves nothing.
        Err(ControlError::Unimplemented("Mirror.Sync"))
    }

    /// **The credential is recorded here as a KIND and the value is dropped**,
    /// which is not a stub: this plane has no sealed store to put it in, and
    /// keeping a private key in a test double's `BTreeMap` is a plaintext
    /// secret in a process that had no reason to hold one. What the double owes
    /// its callers is the `MirrorConfig` shape, and that it gives exactly.
    fn mirror_set_credential(
        &self,
        path: &RepoPath,
        kind: MirrorCredentialKind,
        secret: &str,
        username: &str,
    ) -> ControlResult<MirrorConfig> {
        credential_arguments(kind, secret)?;
        let mut inner = self.inner.lock().unwrap();
        let mirror = inner
            .mirrors
            .get_mut(&MemoryControlPlane::key(path))
            .ok_or_else(|| ControlError::NotFound(format!("mirror for {path}")))?;
        mirror.credential = kind as i32;
        mirror.credential_username = if kind == MirrorCredentialKind::Token {
            username.to_owned()
        } else {
            String::new()
        };
        Ok(mirror.clone())
    }

    fn mirror_status(
        &self,
        namespace: Option<&str>,
        path: Option<&RepoPath>,
    ) -> ControlResult<Vec<MirrorConfig>> {
        Ok(self
            .inner
            .lock()
            .unwrap()
            .mirrors
            .iter()
            .filter(|((ns, _), _)| namespace.is_none_or(|w| w == ns))
            .filter(|((ns, repo), _)| path.is_none_or(|w| w.namespace == *ns && w.repo == *repo))
            .map(|(_, mirror)| mirror.clone())
            .collect())
    }

    fn mirror_remove(&self, path: &RepoPath) -> ControlResult<()> {
        self.inner
            .lock()
            .unwrap()
            .mirrors
            .remove(&MemoryControlPlane::key(path))
            .map(|_| ())
            .ok_or_else(|| ControlError::NotFound(format!("mirror for {path}")))
    }
}

/// **`UsageBatch.ReadMany`'s page token and page cut** (2026-10-06), shared by
/// the default [`ControlPlane::usage_page`] and `gunnar-server`'s ledger.
///
/// The token is opaque to the caller and carries two things: the CURSOR the
/// read began at (so every page of one read answers the same cursor, and a
/// change that lands mid-read is answered again next time), and the last
/// namespace served (the read resumes after it). `v1;<cursor>;<after>` — a
/// namespace name cannot hold `;`, and a token that does not parse starts the
/// read over rather than failing it.
pub mod usage_page_token {
    /// `(cursor, after)` out of a token; `("", "")` for the first page or a
    /// token this build did not write.
    pub fn decode(token: &str) -> (String, String) {
        let mut parts = token.splitn(3, ';');
        match (parts.next(), parts.next(), parts.next()) {
            (Some("v1"), Some(cursor), Some(after)) => (cursor.to_owned(), after.to_owned()),
            _ => (String::new(), String::new()),
        }
    }

    /// The token resuming after `after`, for a read that began at `cursor`.
    pub fn encode(cursor: &str, after: &str) -> String {
        format!("v1;{cursor};{after}")
    }

    /// **Cut one page** out of rows SORTED by namespace: the rows after
    /// `after`, at most `page_size`, and the token to the next page (empty on
    /// the last).
    pub fn page(
        rows: Vec<gunnar_proto::NamespaceUsage>,
        after: &str,
        page_size: usize,
        cursor: String,
        changed_only: bool,
    ) -> gunnar_proto::ReadManyUsageResponse {
        let page_size = page_size.max(1);
        let start = if after.is_empty() { 0 } else { rows.partition_point(|r| r.namespace.as_str() <= after) };
        let mut rest: Vec<gunnar_proto::NamespaceUsage> = rows.into_iter().skip(start).collect();
        let more = rest.len() > page_size;
        rest.truncate(page_size);
        let next_page_token = match (more, rest.last()) {
            (true, Some(last)) => encode(&cursor, &last.namespace),
            _ => String::new(),
        };
        gunnar_proto::ReadManyUsageResponse {
            usages: rest,
            next_page_token,
            cursor,
            changed_only,
            as_of_unix_ms: crate::bus::now_unix_ms(),
        }
    }

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

        fn row(ns: &str) -> gunnar_proto::NamespaceUsage {
            gunnar_proto::NamespaceUsage { namespace: ns.into(), ..Default::default() }
        }

        #[test]
        fn pages_resume_after_the_last_namespace_and_the_last_page_has_no_token() {
            let rows = || vec![row("a"), row("b"), row("c"), row("d"), row("e")];
            let one = page(rows(), "", 2, "E.1.7".into(), true);
            assert_eq!(one.usages.iter().map(|r| r.namespace.as_str()).collect::<Vec<_>>(), ["a", "b"]);
            let (cursor, after) = decode(&one.next_page_token);
            assert_eq!((cursor.as_str(), after.as_str()), ("E.1.7", "b"));
            let two = page(rows(), &after, 2, cursor.clone(), true);
            assert_eq!(two.usages.iter().map(|r| r.namespace.as_str()).collect::<Vec<_>>(), ["c", "d"]);
            let (_, after) = decode(&two.next_page_token);
            let three = page(rows(), &after, 2, cursor, true);
            assert_eq!(three.usages.iter().map(|r| r.namespace.as_str()).collect::<Vec<_>>(), ["e"]);
            assert!(three.next_page_token.is_empty(), "the last page says so");
            assert_eq!(decode("garbage"), (String::new(), String::new()), "a foreign token starts over");
        }
    }
}