gunnar-server 3.0.0

//! **The join.** `gunnar-ssh`'s [`GitTransport`] seam, implemented over
//! `gunnar-wire`'s state machines, with `gunnar-wire`'s [`PushGate`] seam
//! implemented over `gunnar-policy`'s engine.
//!
//! Until this file existed, those three crates had each been proven alone:
//! `gunnar-wire` against stock `git` over pipes, `gunnar-ssh` against a
//! `BannerTransport`, `gunnar-policy` against crafted push fixtures. Nothing
//! had joined them, so nothing had proved a real `git clone` over a real SSH
//! connection. `tests/ssh_end_to_end.rs` does.
//!
//! # Async on one side, blocking on the other
//!
//! russh hands over a `tokio` duplex stream; `gunnar-wire` wants
//! `&mut dyn BufRead` and `&mut dyn Write`, deliberately, so it can be driven
//! by a pipe or a subprocess in tests (the git wire protocol). The stream is therefore
//! [`tokio::io::split`] into halves — one object cannot be borrowed mutably as
//! both reader and writer — each wrapped in a [`SyncIoBridge`], and the whole
//! transfer runs inside [`tokio::task::spawn_blocking`]. `SyncIoBridge` blocks
//! on the runtime, so running it on a runtime worker would deadlock the reactor
//! that has to feed it.
//!
//! # The protocol version is the client's to choose
//!
//! A v2 client announces itself by sending `GIT_PROTOCOL=version=2` as an SSH
//! **environment request**. `gunnar-ssh` accepts that one variable and no
//! other (`gunnar_ssh::server::ENV_ALLOWLIST`), and carries the raw value
//! on [`ServiceRequest::git_protocol`]. [`protocol_version`] turns it
//! into a [`ProtocolVersion`] with `gunnar-wire`'s own parser, so there is one
//! implementation of the grammar rather than two.
//!
//! A client that sends nothing gets **v0**, which is what `ProtocolVersion`'s
//! default already means and what `gunnar-wire`'s v0 fallback exists for. That
//! path is not vestigial: it is what an old client, or one behind a transport
//! that drops the variable, actually gets, and `tests/ssh_end_to_end.rs`
//! exercises it explicitly rather than assuming it.
//!
//! **What this unblocked, measured 2026-08-03.** While every session was v0,
//! `gunnar clone` could not clone from `gunnar serve` at all:
//!
//! ```text
//! gunnar clone ssh://git@127.0.0.1:2299/team/project.git
//!   → Could not decode server reply: Currently we require feature
//!     "multi_ack_detailed", which is not supported by the server
//! ```
//!
//! `gunnar-client` fetches through `nornir-git` → `gix`'s blocking transport,
//! which demands `multi_ack_detailed` in v0. `gunnar_wire::caps` used not to
//! advertise it, on the grounds that gunnar negotiated in single-ACK mode and
//! advertising more than is implemented is a bug. **It advertises and
//! implements it now** (smart HTTP): single-ACK cannot complete a stateless-rpc
//! negotiation, so smart HTTP forced the mode to be built properly rather than
//! declined. Delivering `GIT_PROTOCOL` still matters — v2 is the better
//! protocol and `gix` asks for it by itself — but the v0 fallback is no longer
//! a capability gunnar refuses `gix` on paper. That fallback is not exercised
//! here, so treat this paragraph as a statement about the advertisement, which
//! `gunnar-wire/tests/advertisement.rs` asserts, and not as a measurement of
//! `gix`'s v0 client.
//!
//! `receive-pack` is **not** given the negotiated version, and that is not an
//! oversight: protocol v2 has no push side. `git` sends `GIT_PROTOCOL` on a
//! push too and then speaks v0 anyway, so passing it on would be inventing a
//! protocol neither end implements.
//!
//! # This module is the event-stream producer, and the attestation one
//!
//! `Events.Watch` had a bus, a filter, a gap marker and its own tests. For
//! `PushAttestation` it had no producer at all. A stream with nothing behind it
//! connects, delivers nothing and is indistinguishable from a quiet server.
//! [`serve_receive_pack`] is where a push becomes events:
//!
//! ```text
//!   refs applied ── status reported ──┬── RefUpdated / RefDeleted / PushRejected
//!                                     └── PushAttestation ──┬── the event bus
//!                                                           └── attestations.jsonl
//! ```
//!
//! Everything is published **after** the report is on the wire. A push must
//! never fail because nobody was watching, and it must never be *reported* as
//! having done something the client has not yet been told about.
//!
//! ## Why there is no publish callback in `gunnar-wire`
//!
//! There does not need to be one. `ReceivePack::serve` already returns a
//! [`PushOutcome`] carrying every command, its status and the indexed pack,
//! which is the input an event needs. A `Publisher` trait threaded
//! through `gunnar-wire` would be a second path to the same facts, live at a
//! point where the report has *not* yet been written, and a second thing to
//! keep in step (one implementation). The coupling `gunnar-wire` must not grow is a
//! dependency on `gunnar-grpc`, and returning a value cannot create one.
//!
//! ## The dependency edge push attestation forbids
//!
//! gunnar does not link the Build Thing, nornir, skade, arrow or parquet for
//! any of this. The event is a proto message and the log is a file of JSON
//! lines; anything wanting a warehouse row subscribes and writes it itself.

use std::collections::HashMap;
use std::future::Future;
use std::io::{BufRead, BufReader, BufWriter, Write as _};
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};

use gix_hash::ObjectId;
use gunnar_core::observe::{ServerMetrics, TARGET_GIT, TARGET_POLICY, TARGET_STORE};
use gunnar_grpc::EventBus;
use gunnar_policy::{
    KeyDirectory, PolicyConfig, PolicyEngine, PushContext, RefUpdate, RepoId, StaticKeyDirectory,
};
use gunnar_proto::{event, PushAttestation, PushRejected, RefDeleted, RefUpdated, RepoPath};
use gunnar_ssh::seam::{Delivery, DuplexStream, GitTransport, ServiceRequest, TransportError};
use gunnar_ssh::GitService;
use gunnar_store::git_storage_trait::GitServe;
use gunnar_wire::{
    PackReceipt, ProtocolVersion, PushFacts, PushGate, PushOutcome, ReceivePack, RefCommand,
    RefStatus, Session, UploadPack,
};
use tokio_util::io::SyncIoBridge;

use crate::attest::AttestationLog;
use crate::storage::OpenRepo;
use crate::stores::ServerStores;

/// The per-repository policy document, read from the repository's git dir.
///
/// `gunnar-core` has no storage for branch protection or ref-name policy (see
/// the report in the crate docs), so the gate configuration comes from a file
/// beside the repository until it does. When the catalog grows a policy
/// document, this constant is the only thing that changes.
pub const POLICY_FILE: &str = "gunnar-policy.json";

/// Which protocol version to answer a request in.
///
/// The client chooses, by sending `GIT_PROTOCOL` as an SSH environment request;
/// `gunnar-ssh` allowlists that one name and carries the raw value here. The
/// grammar is parsed **once**, by `gunnar-wire`, so a v1 announcement cannot
/// mean one thing to the SSH layer and another to the wire (one implementation).
///
/// A request that carries nothing is v0, and not by a default that happens to
/// land on v0: that is what `from_git_protocol(None)` says.
fn protocol_version(request: &ServiceRequest) -> ProtocolVersion {
    ProtocolVersion::from_git_protocol(request.git_protocol.as_deref())
}

/// **The push's standing in its namespace's byte ledger, as the receive loop
/// asks about it.**
///
/// Four lines, and they are the whole of the third seam this file holds. The
/// other two are [`protocol_version`] and [`wire_session`], and this one exists
/// for the same reason they do: `gunnar-ssh` and `gunnar-wire` deliberately do
/// not depend on each other, and this crate is the only one that holds both.
///
/// The bug it closes is a time-of-check/time-of-use on a quota. Every push was
/// handed `PlanGate::pack_ceiling`'s number — the namespace's remaining room
/// plus slack — and held to the whole of it, so N pushes that arrived together
/// were each admitted against the same measurement and the room was spent N
/// times. `gunnar_lfs::Mark` makes the number a budget the namespace shares:
/// the receive loop charges every chunk before it takes it, against everything
/// admitted since this push measured.
///
/// # What happened to the slack
///
/// Nothing, and that is deliberate. [`crate::entitlement`] still computes
/// `remaining + max(cap / 16, 64 MiB)` and this file still passes that number
/// through untouched; the ledger adds no arithmetic of its own and never
/// re-adds the slack per connection. What changed is only WHO spends it: the
/// 6.25 % (floor 64 MiB) that covers the difference between a thin pack's wire
/// bytes and its stored bytes is now one allowance for the namespace, charged
/// once, instead of one allowance per simultaneous push. That is the direction
/// the slack was argued for in the first place — it exists so the push that is
/// landing is not cut by arithmetic nobody can do in advance, not so that eight
/// pushes may each have one.
///
/// # Why an adapter rather than the trait on `Mark` itself
///
/// Because `gunnar-lfs` would then have to name `gunnar_wire::PackLedger`, and
/// `gunnar-lfs` is a leaf (44 packages) that `gunnar-lfs-serve` ships on its
/// own; `gunnar-wire` cannot name `gunnar_lfs::Mark` either, being bound by the
/// no-OS rule and the dependency budget to plumbing. An `Arc<dyn PackLedger>`
/// built here costs one allocation per push, adds no edge to either crate's
/// graph, and puts the conversion where every other conversion between those two
/// already is.
#[derive(Debug)]
pub(crate) struct PackMark {
    pub(crate) mark: gunnar_lfs::Mark,
}

impl gunnar_wire::PackLedger for PackMark {
    /// Pure forwarding, on purpose: `room` is the receive loop's number (the
    /// ceiling plus the one byte that makes it reachable) and the arithmetic is
    /// the ledger's. Anything computed here would be a third place where a byte
    /// ceiling is decided.
    fn admit_upto(&self, want: u64, room: u64) -> u64 {
        self.mark.admit_upto(want, room)
    }
}

/// How much of a conversation this call carries, as `gunnar-wire` names it.
///
/// [`Delivery`] is the seam's word and [`Session`] is the wire's, and they are
/// two enums rather than one because `gunnar-ssh` deliberately does not depend
/// on `gunnar-wire` (see `ServiceRequest::git_protocol`). This is the whole of
/// the translation, in one place, so the mapping cannot be made differently in
/// two.
///
/// [`Delivery::Advertisement`] has no `Session` at all: it is not a session,
/// it is a `GET` that writes the advertisement and reads nothing, and it is
/// handled before this is consulted.
fn wire_session(request: &ServiceRequest) -> Session {
    match request.delivery {
        Delivery::Session | Delivery::Advertisement => Session::Connected,
        Delivery::StatelessRpc => Session::StatelessRpc,
    }
}

/// **What `gunnar-wire` is configured with, for both directions.**
///
/// `gunnar-wire` has had these knobs since it was written, and a
/// `with_config` on both state machines to receive them. Nothing ever called
/// either: this module said `UploadPack::new(…)` and `ReceivePack::new(…)`, so
/// every deployment ran on `Config::default()` and **an operator could not
/// decline `filter`** — the one capability the type's own doc comment says is
/// "the one capability an operator plausibly wants to decline".
///
/// A configuration surface that cannot affect behaviour is worse than none.
/// It reads as a control that works: an appliance that had to state it does
/// not serve partial clones had `Offered { filter: false }` sitting right
/// there, fully implemented in both protocol versions, and no way at all to
/// reach it. So the knobs are wired rather than deleted, and this struct is
/// the one place the two `gunnar_wire::…::Config` types are held (one implementation) — a
/// second copy of `agent` or `budget` on `ServeConfig` is how a value ends up
/// honoured on one direction and stubbed on the other.
#[derive(Debug, Clone)]
pub struct WireConfig {
    /// What `git-upload-pack` serves under: `offered`, `want_policy`, `budget`
    /// and `limits`.
    pub upload: gunnar_wire::upload_pack::Config,
    /// What `git-receive-pack` serves under: the connectivity budget.
    pub receive: gunnar_wire::receive_pack::Config,
    /// What `git-lfs-transfer` serves under.
    ///
    /// Held here and not beside the store for the same reason the two above
    /// are: the LFS session is a third service on this one listener, and a
    /// second home for its quota is how 8 GiB comes to be enforced by the SSH
    /// path and not by the next caller. `gunnar_lfs::Session::with_limits` and
    /// the three builders behind it had **zero** call sites outside
    /// `tests/quota.rs` — every deployment ran on `Limits::default()` and the
    /// numbers in the doc comments were not reachable from any command line.
    pub lfs: gunnar_lfs::Limits,
    /// Enumerate the objects each push introduced, so the push attestation can
    /// name them. **On by default.**
    ///
    /// The ids are not a by-product of storing a pack — a pack entry does not
    /// carry its oid — so answering this waits for the engine's indexer to
    /// drain, which is real latency on the push path
    /// (`storage::znippy_engine::ingest` has the measured multiples). Before
    /// this flag existed the enumeration happened only when a policy gate
    /// needed it, and an attestation on a repository with no gates therefore
    /// carried an empty object list that was indistinguishable from a push
    /// that introduced nothing.
    ///
    /// Turning it **off** does not restore that silence: the attestation then
    /// records the reason the set is absent (`new_object_oids_unknown`), because an
    /// absence that renders as a value is the defect this flag was added to
    /// remove, not the behaviour it restores.
    pub attest_objects: bool,
}

impl Default for WireConfig {
    /// Hand-written for one field. `#[derive(Default)]` would make
    /// [`WireConfig::attest_objects`] `false`, and the safe default here is the
    /// one that produces a complete attestation — the cheap default is the one
    /// an operator opts into, never the one they get by forgetting.
    fn default() -> Self {
        WireConfig {
            upload: Default::default(),
            receive: Default::default(),
            lfs: Default::default(),
            attest_objects: true,
        }
    }
}

impl WireConfig {
    /// The knobs an operator set on `gunnar serve`, translated once.
    ///
    /// `None` on every flag means "the operator did not say", which takes
    /// `gunnar-wire`'s default — the same convention `--max-concurrent-transfers`
    /// already uses. `0` is a *stated* value and is treated as one:
    ///
    /// * on the two byte ceilings it is the documented way to remove them,
    ///   because `Limits`' fields are `Option` and `None` means unbounded;
    /// * on `--fetch-budget` it is **refused**, because the walk stops when it
    ///   reaches the budget, so `0` would mean a server that answers no fetch
    ///   at all while reading as "no limit". An unbounded-looking flag that
    ///   silently bricks the server is the failure this whole item is about.
    pub fn from_serve_args(args: &gunnar_common::cli::ServeArgs) -> anyhow::Result<Self> {
        let mut wire = WireConfig::default();
        if args.no_attest_objects {
            wire.attest_objects = false;
        }
        if args.no_filter {
            wire.upload.offered.filter = false;
        }
        if let Some(policy) = args.want_policy {
            wire.upload.want_policy = policy.to_wire();
        }
        if let Some(budget) = args.fetch_budget {
            anyhow::ensure!(
                budget > 0,
                "--fetch-budget 0 would refuse every fetch rather than allow every fetch: the \
                 walk stops when it REACHES the budget. Leave the flag off for the default of \
                 {}, or give the number of objects a fetch may visit.",
                gunnar_wire::DEFAULT_BUDGET
            );
            wire.upload.budget = Some(budget);
        }
        if let Some(bytes) = args.max_fetch_request_bytes {
            wire.upload.limits.request_bytes = (bytes > 0).then_some(bytes);
        }
        if let Some(bytes) = args.max_fetch_object_bytes {
            wire.upload.limits.object_bytes = (bytes > 0).then_some(bytes);
        }
        if let Some(bytes) = args.max_fetch_walk_bytes {
            wire.upload.limits.walk_bytes = (bytes > 0).then_some(bytes);
        }
        // ── the ceiling that bounds the BOX, not the request ─────────────────
        //
        // Built here rather than defaulted in `gunnar-wire`, because a pool is
        // only a pool if every connection gets the same `Arc`. A default inside
        // `Config` would give each session a private purse of its own, which
        // reads exactly like a server-wide ceiling and is not one.
        //
        // `for_this_machine` reads `MemTotal`. The alternative — a constant —
        // would be a memory ceiling chosen without knowing how much memory
        // there is, which is the mistake `DEFAULT_BUDGET` made in objects.
        wire.upload.walk_pool = match args.max_walk_bytes_total {
            Some(0) => None,
            Some(bytes) => Some(gunnar_wire::WalkPool::new(bytes)),
            None => Some(gunnar_wire::WalkPool::for_this_machine()),
        };
        // The LFS three are `u64`/`usize` and not `Option`, so `0` is a stated
        // "refuse everything" rather than the "unbounded" the two flags above
        // mean by it. Same call as `--fetch-budget`: refuse it at startup,
        // because a flag that reads as "no limit" and bricks the service is the
        // failure this whole item is about.
        if let Some(bytes) = args.max_lfs_object_bytes {
            anyhow::ensure!(
                bytes > 0,
                "--max-lfs-object-bytes 0 would refuse every upload, not allow every upload. \
                 Leave the flag off for the default of {} bytes.",
                gunnar_lfs::DEFAULT_MAX_OBJECT_BYTES
            );
            wire.lfs = wire.lfs.max_object_bytes(bytes);
        }
        if let Some(objects) = args.max_lfs_batch_objects {
            anyhow::ensure!(
                objects > 0,
                "--max-lfs-batch-objects 0 would refuse every batch. Leave the flag off for the \
                 default of {}.",
                gunnar_lfs::DEFAULT_MAX_BATCH_OBJECTS
            );
            wire.lfs = wire.lfs.max_batch_objects(objects);
        }
        if let Some(bytes) = args.max_lfs_drain_bytes {
            anyhow::ensure!(
                bytes > 0,
                "--max-lfs-drain-bytes 0 would drop the channel on the first refusal instead of \
                 draining to the client's flush-pkt, so the refusal itself would not be \
                 delivered. Leave the flag off for the default."
            );
            wire.lfs = wire.lfs.max_drain_bytes(bytes);
        }
        Ok(wire)
    }

    /// **The one place an LFS session is built.**
    ///
    /// By construction rather than by a guard that watches two call sites
    /// agree (one implementation). The defect this closes was exactly a second construction
    /// path: `Session::new(…).serve(…)` in the SSH transport, with
    /// `with_limits` reachable only from `tests/quota.rs`. A test that built
    /// its own session could be green with the server on `Limits::default()`
    /// forever, so the test drives *this* and there is nothing else to drive.
    pub fn lfs_session<'a>(
        &self,
        store: &'a dyn gunnar_lfs::LfsStore,
        operation: gunnar_lfs::Operation,
    ) -> gunnar_lfs::Session<'a> {
        gunnar_lfs::Session::new(store, operation).with_limits(self.lfs)
    }
}

/// `gunnar-wire`, wired to the SSH listener.
pub struct WireTransport {
    stores: Arc<ServerStores>,
    bus: Option<Arc<EventBus>>,
    attestations: AttestationLog,
    metrics: Arc<ServerMetrics>,
    wire: Arc<WireConfig>,
}

impl WireTransport {
    /// Serve repositories out of `stores`.
    pub fn new(stores: Arc<ServerStores>) -> Self {
        let attestations = AttestationLog::under(stores.root());
        WireTransport {
            stores,
            bus: None,
            attestations,
            metrics: Arc::new(ServerMetrics::new()),
            wire: Arc::new(WireConfig::default()),
        }
    }

    /// Serve under `wire` rather than `gunnar-wire`'s defaults.
    ///
    /// The whole of what makes [`WireConfig`] reachable. Without a call to
    /// this, every knob on it is a comment.
    pub fn with_wire_config(mut self, wire: WireConfig) -> Self {
        self.wire = Arc::new(wire);
        self
    }

    /// Count into the **server's** counters rather than into a private set.
    ///
    /// `gunnar serve` mints one [`ServerMetrics`] and hands the same `Arc` here
    /// and to the SSH listener, so `connections_refused` and `pushes_rejected`
    /// are two readings of one server. A transport built without this still
    /// counts, into a set only it can see, which is what a unit test wants.
    pub fn with_metrics(mut self, metrics: Arc<ServerMetrics>) -> Self {
        self.metrics = metrics;
        self
    }

    /// Publish `RefUpdated` / `RefDeleted` / `PushRejected` / `PushAttestation`
    /// to the event bus.
    ///
    /// **The bus is optional; the attestation log is not.** A server started
    /// with `--no-grpc` has nobody to notify, but the attestation's durable record is not a
    /// notification. It is the thing the event stream explicitly does not
    /// promise to keep (see `EventBus::publish`), so the log is written either
    /// way and only the publish is conditional.
    pub fn with_bus(mut self, bus: Arc<EventBus>) -> Self {
        self.bus = Some(bus);
        self
    }

    /// Where the attestation's JSON lines land.
    pub fn attestation_log(&self) -> &AttestationLog {
        &self.attestations
    }
}

/// **The LFS object store, opened the one way** — for the SSH session and for
/// the HTTP batch API alike (one implementation).
///
/// The two surfaces used to be one call apart from having two answers to *where
/// does this repository keep its LFS objects*: `serve_blocking` reached for
/// `stores.repo_store(...).git_dir()` and anything else would have had to
/// re-derive it. It is `<git_dir>/lfs/`, the symlink check in
/// `ServerStores::repo_dir` is not optional on either path, and the handle the
/// store lock lives on is the same handle, so a repack and an LFS transfer
/// exclude each other the same way on both transports.
impl gunnar_ssh::seam::LfsTransport for WireTransport {
    fn open_lfs(
        &self,
        request: &ServiceRequest,
    ) -> Result<gunnar_ssh::seam::LfsAccess, TransportError> {
        // An upload into a repository that does not exist yet (the
        // first-repository door, 2026-09-28): the repository is made by its
        // first object, see `ProspectiveLfs`.
        if request.prospective {
            return Ok(gunnar_ssh::seam::LfsAccess {
                store: Arc::new(ProspectiveLfs::new(
                    Arc::clone(&self.stores),
                    request.clone(),
                )),
                limits: self.wire.lfs,
                allowance: request.lfs_allowance.clone(),
            });
        }
        let store = open_repo_store(&self.stores, request)?;
        let lfs = open_lfs_store(&self.stores, &store, request)?;
        Ok(gunnar_ssh::seam::LfsAccess {
            store: Arc::new(lfs),
            // **`self.wire.lfs`, not `Limits::default()`.** The one place an
            // operator's `--max-lfs-object-bytes` is held. A separate default
            // here is exactly how a quota comes to be enforced on SSH and not
            // on HTTP; see `WireConfig::lfs`.
            limits: self.wire.lfs,
            // **The tenant's plan**, decided by the gate chain in
            // `authorise_service` and carried on the request beside the pack
            // ceiling. Cloned rather than moved: `open_lfs` takes the request by
            // reference, and one `Allowance` is two small fields.
            allowance: request.lfs_allowance.clone(),
        })
    }
}

/// Open the repository behind an **already authorised** request, or say why not
/// in the sentence the client is owed.
///
/// Shared by `serve_blocking` and [`LfsTransport::open_lfs`]. The distinction
/// between the two failures is the load-bearing part and is why this is one
/// function: a held store lock is a running `gunnar gc` (or a second gunnar)
/// and clears on its own, so the answer is "retry"; every other open failure is
/// a fault an operator has to look at. Answering both with one sentence is how
/// a two-minute wait becomes a support ticket.
///
/// `TransportError::Refused` and not `Protocol`: both call sites reach this
/// before a single byte has been written, which is what that variant *means*.
fn open_repo_store(
    stores: &ServerStores,
    request: &ServiceRequest,
) -> Result<Arc<OpenRepo>, TransportError> {
    stores.repo_store(&request.repo).map_err(|err| {
        // Opening the repository failed *after* the grant check said yes, so
        // this is a storage fault and not a decision. It used to be returned
        // silently: the client saw a dead channel and the server said nothing
        // at all.
        tracing::error!(
            target: TARGET_GIT,
            error = format!("{err:#}"),
            "the repository could not be opened for a transfer that was already authorised"
        );
        TransportError::Refused(match locked_path(&err) {
            Some(_) => format!(
                "the repository {} is locked by maintenance on the server (a repack or a \
                 second gunnar holds its store lock); the transfer was not started, so retry",
                request.repo.key()
            ),
            None => format!(
                "the repository {} could not be opened on the server; the transfer was not \
                 started. The reason is in the server log",
                request.repo.key()
            ),
        })
    })
}

/// The LFS objects under a repository's own git dir. See
/// [`gunnar_lfs::fs_store`].
fn open_lfs_store(
    stores: &ServerStores,
    store: &OpenRepo,
    request: &ServiceRequest,
) -> Result<crate::txlog_lfs::LoggedLfs<gunnar_lfs::FsLfsStore>, TransportError> {
    let repo = format!("{}/{}", request.repo.namespace, request.repo.name);
    gunnar_lfs::FsLfsStore::open_or_init(store.git_dir())
        .map(|fs| {
            crate::txlog_lfs::LoggedLfs::new(fs, Arc::clone(stores.txlog()), repo)
                // Held to the data set's room, whole object at a time (cov-13).
                .with_space(stores.space_ledger())
        })
        .map_err(|err| {
            tracing::error!(
                target: TARGET_GIT,
                error = format!("{err}"),
                "the lfs object store could not be opened for an authorised transfer"
            );
            // The same `Refused`, for the same reason, at the twin of the seam
            // above: this is still before the first byte, and the LFS grammar has
            // no per-object error line to carry it later.
            TransportError::Refused(format!(
                "the lfs object store for {} could not be opened on the server; the \
             transfer was not started. The reason is in the server log",
                request.repo.key()
            ))
        })
}

/// **An LFS upload into a repository that does not exist yet** (2026-09-28):
/// the first-repository door for git-lfs's pre-push hook, which runs after
/// git took receive-pack's advertisement and BEFORE it sends a command.
/// MEASURED 2026-09-28: the first `git push` with an LFS object into a new
/// repository died in that hook — `batch request: exec request failed on
/// channel 0: exit status 255` — and neither the repository nor the ref was
/// created, while a precreated repository worked.
///
/// `Dispatch::authorise_service` admitted the upload under the SAME
/// push-to-create rules as the push (`Control::push_create_spec`: namespace
/// Admin, and server Admin for a namespace that does not exist; everyone
/// else the fused not-found). Nothing is written until an OBJECT arrives:
/// until then every object is absent (so the batch asks for all of them),
/// and a client that only batches, or hangs up, leaves nothing. The first
/// `begin_write` makes the repository through [`create_prospect`] — the
/// same writes a push-create makes, the grant asked again — or takes the one
/// a concurrent upload or push made first. After that it is the ordinary
/// store.
///
/// The repository is the push-create prospect, sha1, for the reason
/// `push_create_spec` gives; LFS objects are sha256 by their own spec
/// whatever the repository's hash is.
pub(crate) struct ProspectiveLfs<S> {
    stores: S,
    request: ServiceRequest,
    made: std::sync::OnceLock<(
        Arc<OpenRepo>,
        crate::txlog_lfs::LoggedLfs<gunnar_lfs::FsLfsStore>,
    )>,
    making: Mutex<()>,
}

impl<S: std::ops::Deref<Target = ServerStores> + Send + Sync> ProspectiveLfs<S> {
    pub(crate) fn new(stores: S, request: ServiceRequest) -> Self {
        ProspectiveLfs {
            stores,
            request,
            made: std::sync::OnceLock::new(),
            making: Mutex::new(()),
        }
    }

    fn made(&self) -> Option<&crate::txlog_lfs::LoggedLfs<gunnar_lfs::FsLfsStore>> {
        self.made.get().map(|(_, lfs)| lfs)
    }

    /// The store, made on first use.
    fn made_or_create(
        &self,
    ) -> gunnar_lfs::Result<&crate::txlog_lfs::LoggedLfs<gunnar_lfs::FsLfsStore>> {
        if let Some(lfs) = self.made() {
            return Ok(lfs);
        }
        let _one = self.making.lock().unwrap_or_else(|p| p.into_inner());
        if let Some(lfs) = self.made() {
            return Ok(lfs);
        }
        let stores: &ServerStores = &self.stores;
        let path = self.request.repo.path();
        let (repo, store) = create_prospect(stores, &self.request).map_err(|sentence| {
            if sentence.starts_with("repository not found") {
                gunnar_lfs::Error::NotFound(sentence)
            } else {
                gunnar_lfs::Error::Store(sentence)
            }
        })?;
        let mut admitted = self.request.clone();
        admitted.repo = repo;
        admitted.prospective = false;
        let lfs = open_lfs_store(stores, &store, &admitted)
            .map_err(|err| gunnar_lfs::Error::Store(err.to_string()))?;
        tracing::info!(
            target: TARGET_GIT,
            event = "git.lfs_transfer.created_repo",
            repo = %path,
            principal = %self.request.principal.id,
            "an LFS upload's first object made the repository (the push-create door)"
        );
        Ok(&self.made.get_or_init(|| (store, lfs)).1)
    }
}

impl<S: std::ops::Deref<Target = ServerStores> + Send + Sync> gunnar_lfs::LfsStore
    for ProspectiveLfs<S>
{
    fn size_of(&self, oid: &gunnar_lfs::Oid) -> gunnar_lfs::Result<Option<u64>> {
        match self.made() {
            Some(lfs) => lfs.size_of(oid),
            None => Ok(None),
        }
    }
    fn open(&self, oid: &gunnar_lfs::Oid) -> gunnar_lfs::Result<Option<gunnar_lfs::LfsObject>> {
        match self.made() {
            Some(lfs) => lfs.open(oid),
            None => Ok(None),
        }
    }
    fn begin_write(
        &self,
        oid: &gunnar_lfs::Oid,
        declared_size: u64,
    ) -> gunnar_lfs::Result<Box<dyn gunnar_lfs::PendingWrite + '_>> {
        self.made_or_create()?.begin_write(oid, declared_size)
    }
    fn delete(&self, oid: &gunnar_lfs::Oid) -> gunnar_lfs::Result<bool> {
        match self.made() {
            Some(lfs) => lfs.delete(oid),
            None => Ok(false),
        }
    }
}

impl GitTransport for WireTransport {
    fn serve<'a>(
        &'a self,
        request: ServiceRequest,
        stream: Box<dyn DuplexStream>,
    ) -> Pin<Box<dyn Future<Output = Result<(), TransportError>> + Send + 'a>> {
        let stores = Arc::clone(&self.stores);
        let bus = self.bus.clone();
        let attestations = self.attestations.clone();
        let metrics = Arc::clone(&self.metrics);
        let wire = Arc::clone(&self.wire);
        Box::pin(async move {
            // `spawn_blocking` does NOT carry the current span across (it is a
            // thread hand-off, not a task spawn), so without this the whole
            // transfer — every line `gunnar-wire`, the gate and this module
            // emit — would fall out of `git.transfer` and lose the connection
            // id, the principal and the repository. Captured here, where the
            // future is polled inside `gunnar-ssh`'s instrumented task.
            let parent = tracing::Span::current();
            // One object cannot be borrowed mutably as both `BufRead` and
            // `Write`, so the duplex is split before it is bridged.
            let (read_half, write_half) = tokio::io::split(stream);
            // ★ stinky row 1 (2026-10-10): the git work hears that its client
            // is gone. `gunnar-ssh` sets the flag when the session's grace runs
            // out and then WAITS for this future, so the fan worker is held
            // until the blocking task below has actually ended; the guard sets
            // it too if this future is dropped some other way, because a
            // dropped `JoinHandle` stops nothing.
            let cancel = request.cancel.clone();
            let _stop_if_dropped = CancelOnDrop(cancel.clone());
            let mut reader = Cancellable {
                inner: SyncIoBridge::new(read_half),
                cancel: cancel.clone(),
            };
            let mut writer = Cancellable {
                inner: SyncIoBridge::new(write_half),
                cancel,
            };
            // The halves are dropped here, at the end of the blocking task, and
            // that is now safe: `gunnar-ssh` splits the russh channel instead of
            // consuming it, so no part of this stream closes the channel when it
            // goes. A 250 ms `linger` used to hold them open to win a race
            // against `ChannelCloseOnDrop`; the race is gone and so is the
            // timer.
            tokio::task::spawn_blocking(move || {
                // The blocking thread is this transfer's alone for the whole
                // duration, so entering the span here cannot leak it into
                // another task's lines.
                let _entered = parent.enter();
                serve_blocking(
                    &Serving {
                        stores: &stores,
                        bus: bus.as_deref(),
                        attestations: &attestations,
                        metrics: &metrics,
                        wire: &wire,
                    },
                    request,
                    &mut reader,
                    &mut writer,
                )
            })
            .await
            .map_err(|err| {
                TransportError::Protocol(format!("the git transport task did not finish: {err}"))
            })?
        })
    }
}

/// Sets the transfer's [`Cancel`](gunnar_ssh::channel::Cancel) when dropped:
/// a transport future dropped before its blocking task ended tells the task.
struct CancelOnDrop(gunnar_ssh::channel::Cancel);

impl Drop for CancelOnDrop {
    fn drop(&mut self) {
        self.0.cancel();
    }
}

/// **The git work's safe points**: every read from and write to the client.
///
/// Once the transfer is told to stop, the next read or write fails with
/// `ConnectionAborted` — not `Interrupted`, which `read_exact`/`write_all`
/// retry — and the work unwinds from there. The ref updates of a receive-pack
/// transaction do no stream I/O between them, so a stop never lands inside
/// one: the transaction lands whole, or (stopped while the pack was still
/// arriving) not at all.
struct Cancellable<T> {
    inner: T,
    cancel: gunnar_ssh::channel::Cancel,
}

impl<T> Cancellable<T> {
    fn check(&self) -> std::io::Result<()> {
        if self.cancel.is_cancelled() {
            return Err(told_to_stop());
        }
        Ok(())
    }
}

fn told_to_stop() -> std::io::Error {
    std::io::Error::new(
        std::io::ErrorKind::ConnectionAborted,
        "the transfer was told to stop: its client's session ended and the transfer outlived the grace",
    )
}

impl<T: std::io::Read> std::io::Read for Cancellable<T> {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        self.check()?;
        self.inner.read(buf)
    }
}

impl<T: std::io::Write> std::io::Write for Cancellable<T> {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        self.check()?;
        self.inner.write(buf)
    }

    fn flush(&mut self) -> std::io::Result<()> {
        self.check()?;
        self.inner.flush()
    }
}

fn wire_error(err: gunnar_wire::Error) -> TransportError {
    TransportError::Protocol(err.to_string())
}

/// Counts the bytes that actually cross the wire.
///
/// **Measured, not inferred.** "Objects sent" comes out of the negotiation and
/// says nothing about what a client paid for: a fetch that sends 12 objects can
/// move 4 KiB or 40 MiB. `bytes_out` is the number an operator answering
/// *"why was that clone slow"* reaches for first, and the only place it can be
/// counted honestly is at the stream.
///
/// One relaxed atomic add per `read`/`write` call, and both are already
/// buffered (`BufReader`/`BufWriter` sit on top), so this is a handful of adds
/// per transfer rather than one per byte.
struct Metered<T> {
    inner: T,
    total: Arc<AtomicU64>,
}

impl<T> Metered<T> {
    fn new(inner: T) -> (Metered<T>, Arc<AtomicU64>) {
        let total = Arc::new(AtomicU64::new(0));
        (
            Metered {
                inner,
                total: Arc::clone(&total),
            },
            total,
        )
    }
}

impl<T: std::io::Read> std::io::Read for Metered<T> {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        let n = self.inner.read(buf)?;
        self.total.fetch_add(n as u64, Ordering::Relaxed);
        Ok(n)
    }
}

impl<T: std::io::Write> std::io::Write for Metered<T> {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        let n = self.inner.write(buf)?;
        self.total.fetch_add(n as u64, Ordering::Relaxed);
        Ok(n)
    }

    fn flush(&mut self) -> std::io::Result<()> {
        self.inner.flush()
    }
}

/// **THE counting point for per-namespace flow, both directions.**
///
/// It is a guard that records when it is DROPPED rather than a call at the end
/// of [`serve_blocking`], and that is the whole design. `serve_blocking` has
/// three service arms, an early return for the abandoned prospect, an early
/// return for the advertisement, and a `?` on every fallible step; a call at
/// "the end" would be five calls, and the sixth exit somebody adds next year
/// would silently stop counting a whole transport. A guard cannot be
/// forgotten, and it counts the failed transfers too — the bytes a client sent
/// before it hung up were still received, and a tenant who is failing is
/// exactly the tenant an operator wants to see.
///
/// It reads the SAME two counters [`Metered`] already keeps for
/// `ServerMetrics`, so a byte is counted once for the box and once for the
/// tenant off one measurement, and the two can never disagree about how much
/// moved. Nothing here counts bytes itself.
///
/// The three `Cell` fields are what the arm that ran decides:
///
/// * `fetched` — an upload-pack SESSION, set after the advertisement's early
///   return, so `git ls-remote` and a monitoring probe are bytes and not a
///   customer's fetch;
/// * `pushed` — a receive-pack session that came back `Ok`;
/// * the LFS pair — the session's own applied figures, which are a different
///   number from the wire bytes about the same transfer.
struct FlowRecording<'a> {
    flow: &'a gunnar_core::FlowLedger,
    namespace: String,
    principal: String,
    bytes_in: Arc<AtomicU64>,
    bytes_out: Arc<AtomicU64>,
    fetched: std::cell::Cell<bool>,
    pushed: std::cell::Cell<bool>,
    lfs_bytes_in: std::cell::Cell<u64>,
    lfs_bytes_out: std::cell::Cell<u64>,
}

impl Drop for FlowRecording<'_> {
    fn drop(&mut self) {
        self.flow.record(
            &self.namespace,
            gunnar_core::now_ms(),
            &gunnar_core::Transfer {
                bytes_in: self.bytes_in.load(Ordering::Relaxed),
                bytes_out: self.bytes_out.load(Ordering::Relaxed),
                pushed: self.pushed.get(),
                fetched: self.fetched.get(),
                lfs_bytes_in: self.lfs_bytes_in.get(),
                lfs_bytes_out: self.lfs_bytes_out.get(),
                principal: self.principal.clone(),
            },
        );
    }
}

/// The lock file, when `err` is a store whose exclusive lock is already held.
///
/// `gunnar gc` opens the repository the same way a transfer does, and the
/// `O_EXCL` lock the Arrow log takes is what makes "one writer" true across
/// processes — so a repack and a transfer on the same repository are mutually
/// exclusive **by that lock and nothing else**. Whichever loses gets
/// [`gunnar_store::Error::StoreLocked`] out of `open_or_init`, wrapped in
/// `anyhow` context by `ServerStores::repo_store`, which is why this walks the
/// chain rather than matching the top-level error.
///
/// Recognising it is not cosmetic: "retry in a moment" and "an operator has to
/// look at this" are opposite instructions, and the caller cannot tell them
/// apart from a formatted string without matching on prose.
fn locked_path(err: &anyhow::Error) -> Option<&std::path::Path> {
    match err.downcast_ref::<gunnar_store::Error>() {
        Some(gunnar_store::Error::StoreLocked { path }) => Some(path.as_path()),
        _ => None,
    }
}

/// Everything a transfer needs that belongs to the **server** rather than to
/// the request.
///
/// One borrow instead of five. `serve_blocking` grew an eighth argument when
/// `WireConfig` was wired through and clippy's `too_many_arguments` caught it,
/// which is the right moment to notice that five of those arguments are the
/// same five fields `WireTransport` holds and are passed together everywhere
/// they are passed at all. Bundling them is what the lint is for; an
/// `#[allow]` would have kept the shape that made a sixth easy to add.
struct Serving<'a> {
    stores: &'a ServerStores,
    bus: Option<&'a EventBus>,
    attestations: &'a AttestationLog,
    metrics: &'a Arc<ServerMetrics>,
    wire: &'a WireConfig,
}

fn serve_blocking(
    serving: &Serving<'_>,
    request: ServiceRequest,
    reader: &mut dyn std::io::Read,
    writer: &mut dyn std::io::Write,
) -> Result<(), TransportError> {
    // `bus` and `attestations` are not destructured here: they are used only by
    // `serve_receive_pack`, which takes the whole `Serving`.
    let Serving {
        stores,
        metrics,
        wire,
        ..
    } = *serving;
    // `Refused`, not `Protocol`: this is the one point in `serve_blocking` at
    // which **nothing has been written** — `Metered` is constructed on the next
    // lines and the state machine has not been entered — so the client can still
    // be handed a pkt-line, and `TransportError::Refused` is exactly the claim
    // "no byte has gone out yet". `open_repo_store` is shared with the HTTP LFS
    // seam so the two transports refuse a locked repository in one sentence and
    // not two (one implementation).
    //
    // A PROSPECTIVE request — the first-repository door — has no store to
    // open and must not get one here: `repo_store` would create the directory
    // for a push the client may abandon at the advertisement. That path goes
    // through `admit_prospective_push` below, once the buffers exist.
    if request.cancel.is_cancelled() {
        return Err(TransportError::Io(told_to_stop()));
    }
    let store = if request.prospective {
        None
    } else {
        Some(open_repo_store(stores, &request)?)
    };
    let (metered_in, bytes_in) = Metered::new(reader);
    let (metered_out, bytes_out) = Metered::new(writer);
    // **The per-namespace flow counting point.** Created here, beside the two
    // counters it reads, and recorded when it is dropped — on every exit from
    // this function, including the early returns and the failures. See
    // `FlowRecording`. It is created BEFORE the first-repository door below, so
    // that a client which took an advertisement of a repository it never
    // created still has its bytes charged to the namespace it asked about.
    let flow = FlowRecording {
        flow: stores.flow().as_ref(),
        namespace: request.repo.namespace.clone(),
        principal: request
            .key_fingerprint
            .as_ref()
            .map(|f| f.to_string())
            .unwrap_or_default(),
        bytes_in: Arc::clone(&bytes_in),
        bytes_out: Arc::clone(&bytes_out),
        fetched: std::cell::Cell::new(false),
        pushed: std::cell::Cell::new(false),
        lfs_bytes_in: std::cell::Cell::new(0),
        lfs_bytes_out: std::cell::Cell::new(0),
    };
    // **`BufWriter::new` is 8 KiB, and a sideband packet is 65 520.** So the
    // default capacity did not buffer the pack at all: every packet flushed the
    // 5-byte header as a write of its own and then passed 65 515 bytes straight
    // through. Two writes per packet, one of them five bytes — and under this
    // transport a write is not a syscall, it is a hand-off across a
    // `tokio::io::duplex` from a `spawn_blocking` thread, which is a `block_on`,
    // a waker and (whenever the reader is parked) a futex pair.
    // `gunnar_wire::TRANSPORT_BUFFER_BYTES` carries the arithmetic and the
    // write-counting guard.
    let mut input = BufReader::with_capacity(gunnar_wire::TRANSPORT_BUFFER_BYTES, metered_in);
    let mut output = BufWriter::with_capacity(gunnar_wire::TRANSPORT_BUFFER_BYTES, metered_out);
    let started = std::time::Instant::now();

    // ── the first-repository door: advertise, read, check, THEN create ──────
    //
    // Nothing has been written to the catalog or the disk for a prospective
    // request. The client is shown the empty repository it would be pushing
    // into; if it hangs up (a sha1 tree that read `object-format=sha256`,
    // `git push` with nothing to do) this returns and nothing exists. Only a
    // command list that passes the hash check turns into a repository — and
    // that create is `ServerStores::create_repo`, the same transactional verb
    // every other create uses.
    //
    // A prospective LFS UPLOAD (git-lfs's pre-push, before receive-pack's
    // commands) goes on with no store at all: `ProspectiveLfs` makes the
    // repository through the same writes when the first object arrives.
    let (request, store, preamble) = match store {
        Some(store) => (request, Some(store), None),
        None if matches!(request.service, GitService::LfsTransfer(_)) => (request, None, None),
        None => match admit_prospective_push(serving, &request, &mut input, &mut output)? {
            Some((request, store, preamble)) => (request, Some(store), Some(preamble)),
            None => {
                output.flush()?;
                metrics.add_bytes_out(bytes_out.load(Ordering::Relaxed));
                metrics.add_bytes_in(bytes_in.load(Ordering::Relaxed));
                tracing::info!(
                    target: TARGET_GIT,
                    event = "git.receive_pack.prospect_abandoned",
                    repo = %request.repo.key(),
                    elapsed_ms = started.elapsed().as_millis() as u64,
                    "the client took the advertisement of a repository that does not exist and \
                     sent no command; nothing was created"
                );
                return Ok(());
            }
        },
    };

    // Only a prospective LFS upload reaches here without a store.
    let git_store = || {
        store.clone().ok_or_else(|| {
            TransportError::Refused(format!(
                "repository not found, or you do not have access: {}",
                request.repo.path()
            ))
        })
    };
    match request.service {
        GitService::UploadPack => {
            let span = tracing::info_span!(target: TARGET_GIT, "git.upload_pack");
            let _entered = span.enter();
            let store = git_store()?;
            let version = protocol_version(&request);
            // ── the znippy engine answers reads ─────────────────────────────────
            //
            // When `GUNNAR_COLD_TIER=znippy` and the repository has an archive,
            // upload-pack serves pack ENTRIES out of that archive rather than
            // out of `objects/pack` — no unfold, no file. `packed()` is `Some`
            // on both arms, so the entry-copy path is the same code; only where
            // the bytes come from changes.
            //
            // receive-pack is deliberately NOT switched: a push lands on `fs`
            // and reaches the archive through the append hook. The archive is
            // immutable, and `ZnippyObjects::write` says so rather than
            // pretending.
            //
            // There is no second read side to choose between any more. The
            // engine *is* the archive on the znippy arm, so `GUNNAR_COLD_TIER`,
            // `cold_objects()` and the `Box<dyn ObjectStore>` that carried a
            // `None` for "this binary has no znippy engine" have all gone with
            // the old znippy engine crate: one store answers reads and writes, and which
            // engine it is was decided at startup.
            //
            // **One handle, not two.** `UploadPack` used to take an object
            // store and a ref store, and an advertisement built from one
            // snapshot could promise a pack built from another. `GitServe`
            // carries both halves, so the ref list and the pack it promises are
            // the same store.
            let git = store.serve();
            // `hash_kind` is a required argument, never a default: it comes
            // from the catalog entry through `OpenRepo`, and answering `sha1`
            // for a sha256 repository does not error — it decodes garbage.
            //
            // `with_config`, not `new`: this call is the whole of what makes
            // `--no-filter`, `--want-policy`, `--fetch-budget` and the byte
            // ceilings anything other than fields nobody reads.
            let upload = UploadPack::new(git, store.hash_kind()).with_config(wire.upload.clone());
            // `Delivery::Advertisement` is `GET /info/refs`: write the ref list
            // (v0) or the capability list (v2) and stop, reading nothing. The
            // same call the connected path makes inline, so the two transports
            // cannot advertise differently.
            if request.delivery == Delivery::Advertisement {
                upload.advertise(version, &mut output).map_err(wire_error)?;
                output.flush()?;
                let out = bytes_out.load(Ordering::Relaxed);
                metrics.add_bytes_out(out);
                tracing::info!(
                    target: TARGET_GIT,
                    event = "git.upload_pack.advertised",
                    protocol = version.number(),
                    bytes_out = out,
                    elapsed_ms = started.elapsed().as_millis() as u64,
                    "served the upload-pack advertisement"
                );
                return Ok(());
            }
            // Past the advertisement's early return above: this is a SESSION,
            // which is what `fetches` counts. A bare `git ls-remote` never
            // reaches here.
            flow.fetched.set(true);
            let summary = upload
                .serve_session(version, wire_session(&request), &mut input, &mut output)
                .map_err(|err| {
                    tracing::warn!(
                        target: TARGET_GIT,
                        event = "git.upload_pack.failed",
                        elapsed_ms = started.elapsed().as_millis() as u64,
                        bytes_out = bytes_out.load(Ordering::Relaxed),
                        %err,
                        "git-upload-pack failed"
                    );
                    wire_error(err)
                })?;
            output.flush()?;
            let out = bytes_out.load(Ordering::Relaxed);
            if summary.objects_sent > 0 {
                metrics.add_packs_served(1);
            }
            metrics.add_objects_sent(summary.objects_sent as u64);
            metrics.add_bytes_out(out);
            metrics.add_bytes_in(bytes_in.load(Ordering::Relaxed));
            // The applied outputs, and the duration. `repo` and `principal` are
            // NOT repeated: they are fields of `ssh.connection`, which this span
            // descends from.
            tracing::info!(
                target: TARGET_GIT,
                event = "git.upload_pack.served",
                // What was actually spoken, taken from the session's own
                // summary rather than from the request that asked for it; a v2
                // request served as v0 would otherwise log as v2.
                protocol = summary.protocol.number(),
                ls_refs = summary.ls_refs,
                rounds = summary.fetch_rounds,
                common = summary.common,
                objects = summary.objects_sent,
                // **WHICH PATH ANSWERED.** `Selection::answered_from_bitmap`
                // exists so a guard can tell a fetch served from the store's
                // reachability index from one served by walking, and until now
                // nothing anywhere read it: not a log line, not a metric, not a
                // test. Its only reader in the constellation was an example
                // binary.
                //
                // That blind spot has already cost something concrete. The
                // `--fetch-budget` control charged the ids a WALKER holds, and
                // on the store the appliance actually runs there is no walker —
                // the index answers whole — so the budget was compared against
                // nothing while a red test sat in front of three separate call
                // sites and each read it as a policy question. A field nobody can
                // observe is a behaviour nobody can assert.
                from_index = summary.answered_from_bitmap,
                bytes_out = out,
                elapsed_ms = started.elapsed().as_millis() as u64,
                "served git-upload-pack"
            );
        }
        GitService::ReceivePack => {
            let span = tracing::info_span!(target: TARGET_GIT, "git.receive_pack");
            let _entered = span.enter();
            let store = git_store()?;
            let outcome =
                serve_receive_pack(serving, &request, &store, preamble, &mut input, &mut output);
            let out = bytes_out.load(Ordering::Relaxed);
            let inn = bytes_in.load(Ordering::Relaxed);
            metrics.add_bytes_out(out);
            metrics.add_bytes_in(inn);
            // A push is a receive-pack session that CAME BACK. The bytes are
            // charged either way — they arrived — but a session that failed is
            // not a push, and counting it as one would make a tenant whose every
            // push is refused look like a tenant who is working.
            flow.pushed.set(outcome.is_ok());
            if let Err(err) = &outcome {
                tracing::warn!(
                    target: TARGET_GIT,
                    event = "git.receive_pack.failed",
                    elapsed_ms = started.elapsed().as_millis() as u64,
                    bytes_in = inn,
                    %err,
                    "git-receive-pack failed"
                );
            }
            outcome?;
        }
        // LFS over SSH, git-lfs 3.0's pure-SSH transfer. The grant was already decided
        // by `gunnar_ssh::command::parse`, from the same key table and the same
        // path sandbox the two verbs above went through: `operation` here is the
        // one that demanded Read or Write, not a second decision. It is passed
        // straight to the session so there is no second authorisation path.
        //
        // The objects live under the repository's own git dir, in `lfs/`. See
        // `gunnar_lfs::fs_store`.
        GitService::LfsTransfer(operation) => {
            let span = tracing::info_span!(target: TARGET_GIT, "git.lfs_transfer");
            let _entered = span.enter();
            let lfs: Box<dyn gunnar_lfs::LfsStore + '_> = match &store {
                Some(store) => Box::new(open_lfs_store(stores, store, &request)?),
                None => Box::new(ProspectiveLfs::new(stores, request.clone())),
            };
            let mut session = wire.lfs_session(lfs.as_ref(), operation);
            // **The tenant's remaining LFS room**, from the gate chain that ran
            // in `authorise_service`. `None` on a community build, on an
            // uncapped namespace and on every download; see
            // `gunnar_ssh::PlanGate::lfs_allowance`.
            if let Some(allowance) = request.lfs_allowance.clone() {
                session = session.with_allowance(allowance);
            }
            // `limits()` and `allowance()` read back off the session, not off
            // `wire` and not off the request: what is logged is what the session
            // will enforce, so a builder that stopped taking would show here
            // rather than in a support ticket.
            let effective = session.limits();
            tracing::debug!(
                target: TARGET_GIT,
                max_object_bytes = effective.max_object_bytes,
                max_batch_objects = effective.max_batch_objects,
                max_drain_bytes = effective.max_drain_bytes,
                lfs_allowance_bytes = session.allowance().map(|a| a.bytes),
                "lfs session limits"
            );
            let outcome = session.serve(&mut input, &mut output);
            output.flush()?;
            let out = bytes_out.load(Ordering::Relaxed);
            let inn = bytes_in.load(Ordering::Relaxed);
            metrics.add_bytes_out(out);
            metrics.add_bytes_in(inn);
            let summary = outcome.map_err(|err| {
                tracing::warn!(
                    target: TARGET_GIT,
                    event = "git.lfs_transfer.failed",
                    operation = %operation,
                    elapsed_ms = started.elapsed().as_millis() as u64,
                    %err,
                    "git-lfs-transfer failed"
                );
                TransportError::Protocol(err.to_string())
            })?;
            // The APPLIED figures, from the session's own count. They are a
            // different number from the wire bytes above about the same
            // transfer — the wire carries framing and the batch conversation —
            // so the flow ledger keeps both rather than one as a subset of the
            // other.
            flow.lfs_bytes_in.set(summary.bytes_received);
            flow.lfs_bytes_out.set(summary.bytes_sent);
            // Applied output: objects and bytes that really moved, taken from
            // the session's own count rather than from what was asked for.
            tracing::info!(
                target: TARGET_GIT,
                event = "git.lfs_transfer.served",
                operation = %operation,
                commands = summary.commands,
                batches = summary.batches,
                batch_objects = summary.batch_objects,
                objects_sent = summary.objects_sent,
                objects_received = summary.objects_received,
                lfs_bytes_out = summary.bytes_sent,
                lfs_bytes_in = summary.bytes_received,
                clean_quit = summary.clean_quit,
                bytes_out = out,
                bytes_in = inn,
                elapsed_ms = started.elapsed().as_millis() as u64,
                "served git-lfs-transfer"
            );
        }
    }
    Ok(())
}

// Eight, because the counters joined the six things a push already needed.
// Bundling them into a context struct would be one more type whose
// only job is to be unpacked on the first line of this function.
#[allow(clippy::too_many_arguments)]
/// The door's commit. Advertise the prospect, read the client's commands,
/// and — only if there are any and they pass the hash check — write the
/// catalog row (the namespace with it, if the admin's push names a new one)
/// and make the store. `None` means the client sent nothing and nothing was
/// made.
///
/// A refusal after the preamble (a stale store at the name, a grant revoked
/// since the exec, a store the engine cannot make) is answered in git's own
/// protocol — an `ERR` packet, which `git push` renders as `remote error:` —
/// and leaves nothing: the row is retracted, the namespace too if this call
/// conjured it, the directory removed.
fn admit_prospective_push(
    serving: &Serving<'_>,
    request: &ServiceRequest,
    input: &mut dyn BufRead,
    output: &mut dyn std::io::Write,
) -> Result<Option<(ServiceRequest, Arc<OpenRepo>, gunnar_wire::Preamble)>, TransportError> {
    let Serving { stores, wire, .. } = *serving;
    if request.service != GitService::ReceivePack {
        // An LFS upload's door is `ProspectiveLfs`, never this one; anything
        // else here is a programming error, refused rather than
        // served against a repository that does not exist.
        return Err(TransportError::Refused(format!(
            "repository not found, or you do not have access: {}",
            request.repo.path()
        )));
    }
    let hash_kind: gix_hash::Kind = request.repo.hash_kind.into();
    if wire_session(request).advertises_inline() || request.delivery == Delivery::Advertisement {
        ReceivePack::write_empty_advertisement(hash_kind, &wire.receive, output)
            .map_err(wire_error)?;
        if request.delivery == Delivery::Advertisement {
            return Ok(None);
        }
    }
    let Some(preamble) = ReceivePack::read_preamble(hash_kind, input).map_err(wire_error)? else {
        return Ok(None);
    };

    // The client means it. Now, and only now, the writes — in the order that
    // leaves nothing behind if any of them fails.
    match create_prospect(stores, request) {
        Ok((repo, store)) => {
            let mut admitted = request.clone();
            admitted.repo = repo;
            admitted.prospective = false;
            Ok(Some((admitted, store, preamble)))
        }
        Err(sentence) => {
            let _ = gunnar_wire::err_packet(output, &sentence);
            let _ = output.flush();
            Err(TransportError::Protocol(sentence))
        }
    }
}

/// **The first-repository door's writes**, for a prospective request whose
/// client has shown it means it — a receive-pack's commands past the hash
/// check, or an LFS upload's first object ([`ProspectiveLfs`]). The stale
/// store check, the catalog row (and the namespace, if the admin's push names
/// a new one) under [`gunnar_ssh::Control::admit_push_create`]'s re-asked
/// grant, then the store — and if the store cannot be made, the row and the
/// namespace are taken back and the directory removed. `Err` is the sentence
/// the client is owed; nothing is left behind by it.
fn create_prospect(
    stores: &ServerStores,
    request: &ServiceRequest,
) -> Result<(gunnar_core::Repo, Arc<OpenRepo>), String> {
    let path = request.repo.path();
    let control = stores.ssh_control();
    // **Made meanwhile, by this same push** (2026-09-28): git takes
    // receive-pack's advertisement (the prospect) BEFORE it runs the pre-push
    // hook, and git-lfs's upload in that hook makes the repository in
    // between. The stale-store check below would answer the push's own
    // commands with an ERR. A repository that now EXISTS is served, as the
    // lost race in `admit_push_create` is, after the same re-asked grant,
    // and only in the hash the client was shown.
    match control.resolve(&path) {
        Ok(Some(existing)) => {
            if existing.hash_kind != request.repo.hash_kind {
                return Err(format!(
                    "{path} was created meanwhile as a {:?} repository, and this push was shown {:?}; nothing was written — push again",
                    existing.hash_kind, request.repo.hash_kind
                ));
            }
            return match control.push_create_spec(&request.principal, &path) {
                Ok(Some(_)) => stores
                    .repo_store(&existing)
                    .map(|store| (existing, store))
                    .map_err(|err| {
                        format!("the repository {path} could not be opened on the server: {err:#}")
                    }),
                Ok(None) => Err(format!(
                    "repository not found, or you do not have access: {path}"
                )),
                Err(err) => Err(format!(
                    "the gunnar control plane could not read {path}: {err}"
                )),
            };
        }
        Ok(None) => {}
        Err(err) => {
            return Err(format!(
                "the gunnar control plane could not read {path}: {err}"
            ))
        }
    }
    if let Err(err) = stores.refuse_stale_store(&path, false, request.repo.hash_kind) {
        return Err(format!("{err:#}"));
    }
    let (repo, conjured) = match control.admit_push_create(&request.principal, &path) {
        Ok(Some(pair)) => pair,
        Ok(None) => {
            return Err(format!(
                "repository not found, or you do not have access: {path}"
            ))
        }
        Err(err) => {
            return Err(format!(
                "the gunnar control plane could not create {path}: {err}"
            ))
        }
    };
    match stores.repo_store(&repo) {
        Ok(store) => Ok((repo, store)),
        Err(err) => {
            stores.close_repo(&path);
            if let Err(undo) = control.retract_push_create(&path, conjured) {
                tracing::error!(
                    target: TARGET_GIT,
                    repo = %path,
                    error = %undo,
                    "retracting the catalog row of a push-create whose store failed"
                );
            }
            if let Ok(dir) = stores.repo_dir(&path) {
                let _ = std::fs::remove_dir_all(dir);
            }
            tracing::error!(
                target: TARGET_GIT,
                repo = %path,
                error = format!("{err:#}"),
                "the push-created repository's store could not be made; nothing was kept"
            );
            Err(format!("the repository {path} could not be created on the server; nothing was kept. The reason is in the server log"))
        }
    }
}

/// **A push into an encrypted repository on a build that cannot check it in quarantine**
/// (wiring tracer B 2026-10-08): the quarantine is the znippy engine's
/// (`encrypted_quarantine`), and without it a refused plaintext pack would be stored and framed
/// for the twin before the ref gate refused it. `Some(the line)`: refuse the push by name,
/// before a byte is read.
pub(crate) fn unchecked_encrypted_push(encrypted: bool, quarantine: bool) -> Option<String> {
    (encrypted && !quarantine).then(|| {
        "this repository is encrypted, and this Gunnar was built without the check that keeps \
         plaintext out of it, so it takes no push into it: push to a Gunnar built with its \
         standard storage engine. Nothing was received"
            .to_string()
    })
}

fn serve_receive_pack(
    serving: &Serving<'_>,
    request: &ServiceRequest,
    store: &OpenRepo,
    preamble: Option<gunnar_wire::Preamble>,
    input: &mut dyn BufRead,
    output: &mut dyn std::io::Write,
) -> Result<(), TransportError> {
    let Serving {
        stores,
        bus,
        attestations,
        metrics,
        wire,
    } = *serving;
    let started = std::time::Instant::now();
    // A build that cannot check an encrypted repository's pack in quarantine takes none.
    if let Some(line) =
        unchecked_encrypted_push(request.repo.encrypted, cfg!(feature = "engine-znippy"))
    {
        return Err(TransportError::Refused(line));
    }
    // **Everything below this line is a validation read.** A receive-pack
    // ingests, indexes and then walks every commit and tree it was sent to
    // prove the push closes — once each — and serves nobody. Those reads are
    // indistinguishable from a client's inside the store, so the caller is the
    // only thing that can say which they are, and this is the caller.
    //
    // Without it the push fills the engine's lazy read-cache for a reader that
    // does not exist: MEASURED 6 778 047 872 B on `torvalds/linux`, 31 % of the
    // finished store (measured 2026-09-03).
    //
    // Thread-scoped and dropped at the end of this function, so a concurrent
    // clone on another connection still warms the cache normally.
    #[cfg(feature = "engine-znippy")]
    let _validation_reads = crate::storage::znippy_engine::validation_reads();
    // **One handle.** `GitServe` carries the reading contract and, through its
    // `GitOps` supertrait, the eleven — so receive-pack advertises, checks
    // connectivity and writes references against a single store rather than
    // against an object handle and a ref handle that could disagree.
    let git = store.serve();
    let hash_kind = store.hash_kind();

    let gate = PolicyGate::open(store.git_dir(), request, stores, Arc::clone(metrics));

    // Opened before the ingest, and that ordering is load-bearing: whether the
    // policy has any gate at all is what decides if the ingest pays for an
    // object-set enumeration. It used to be built after, when nothing needed
    // to know.

    // No `objects/pack` is created. An empty one is what a later reader uses
    // to decide a repository is packfile-shaped, and this server never writes
    // a packfile into it.

    // ── who asks for the object set, and why it is not only the gates ──────
    //
    // `want_objects: !gate.engine.is_empty()` stood here, and it had **one**
    // consumer in mind. The push attestation is a second one, and it was not
    // counted: on a repository with no policy gates the ingest skipped the
    // enumeration and `attestation_for` then wrote `new_object_count: 0` with
    // an empty `new_object_oids` — a push of nothing and a push nobody
    // enumerated rendering as the same record. The log line thirty lines below
    // this already got that distinction right (`objects_unknown`); the
    // attestation, which is the durable artefact and the one a verifier reads,
    // did not. `tests/push_events.rs` saw it as *"the pushed commit is missing
    // from the attestation's object list: []"*.
    //
    // So the attestation asks too, and by default it gets an answer. That is
    // not free — see `znippy_engine::ingest`'s note on why the flag exists at
    // all: the ids are not a by-product of storing a pack, and asking for them
    // waits for the indexer to drain. An operator who would rather have the
    // latency than the object list turns it off with
    // `--no-attest-objects`, and the attestation then records **why** the set
    // is absent instead of implying it was empty.
    let objects = if gate.engine.reads_new_objects() {
        // A gate reads the set. (Not `!is_empty()`: the attribution-trailer
        // gate is live on every repository by default and decides from the
        // references and the object graph, so it must not make every push
        // demand an enumeration — the gix engine refuses `Required` outright.) An engine that cannot produce one must refuse
        // rather than let the gate decide on nothing.
        crate::storage::ObjectSet::Required
    } else if wire.attest_objects {
        // Only the attestation wants it, and the attestation can say it did not
        // get it. An engine that cannot enumerate still serves the push.
        crate::storage::ObjectSet::Wanted
    } else {
        crate::storage::ObjectSet::NotNeeded
    };
    let ingest = PushIngest {
        git_dir: store.git_dir().to_path_buf(),
        hash_kind,
        engine: stores.engine(),
        objects,
        // **A push has nobody watching an object count.** `git push` renders the
        // remote's progress band, and what it wants there is bytes; there is no
        // job record and no console poll on this path. The mirror is the caller
        // that has both, so the mirror is the caller that passes a sink.
        watching: None,
        // The pushed pack is a transaction (the twin): framed before the engine stores it. For a
        // first push `request.repo` is exactly the row `admit_push_create` writes first.
        frame: Some(PackFrame::of(stores, &request.repo)),
        // An encrypted repository's pack is checked before it is kept.
        encrypted: request.repo.encrypted.then_some(git),
        // Held to the data set's room, chunk by chunk (cov-13).
        space: stores.space_ledger(),
    };
    if gate.engine.is_empty() {
        tracing::info!(
            target: TARGET_POLICY,
            event = "policy.gates.none",
            policy = %store.git_dir().join(POLICY_FILE).display(),
            "no policy gates are configured for this repository; every well-formed push is \
             accepted. This is an empty policy, not a passing one."
        );
    } else {
        tracing::info!(
            target: TARGET_POLICY,
            event = "policy.gates.live",
            gates = ?gate.engine.gate_names(),
            "policy gates live for this push"
        );
    }

    // **receive-pack reads from the store it writes into.**
    //
    // This used to be `&objects` unconditionally, under a comment saying
    // receive-pack was "deliberately NOT switched" because a push landed on
    // `fs` and reached the tier through the append hook. That is no longer the
    // route: the pack goes straight into the tier, so checking connectivity
    // against the filesystem would ask an empty repository whether the objects
    // it just stored are present — and the answer is a refused push, which is
    // exactly what it did the first time this ran.
    //
    // A pack the ingest stored but the indexer has not absorbed yet is **not**
    // a stale answer here: the per-pack `indexed` bit makes an index miss fall back to the journal
    // rather than report absent, so the read side sees a push the moment
    // `put_pack` returns, which is before anything asks.
    //
    // `write_objects` and not `cold_objects`: the latter asks whether there is
    // anything to *serve*, and answers `None` for an empty repository — which
    // is every repository's first push.
    // The engine is the read side; there is nothing else to open.

    // No `#[allow(unused_mut)]` any more: it was here because the only
    // assignment below was `cfg(metered)`, so a community build had a `mut`
    // binding it never wrote. The ceiling is assigned on both builds, so the
    // `mut` is now load-bearing on both — and an allow that no longer suppresses
    // anything is a suppression nobody can see expire.
    let mut receive_config = wire.receive.clone();
    // The plan's warning (Grace), decided in `authorise_service` and carried
    // on the request: the wire prints it on the progress band beside the
    // report, so `git push` shows `remote: plan expired …; renew at …`.
    #[cfg(feature = "metered")]
    {
        receive_config.notice = request.plan_notice.clone();
    }
    // **The mid-upload ceiling**, decided by the same gate chain in
    // `authorise_service` and carried on the request beside the notice. Not
    // `cfg(metered)`: the field is on `ServiceRequest` unconditionally and a
    // community build simply carries `None`, so there is one path here rather
    // than a second one that only the paid build compiles (one implementation). The two
    // `PackCeiling`s are the deliberate twin `gunnar_ssh::PackCeiling`
    // documents — `gunnar-ssh` does not depend on `gunnar-wire` — and this is
    // the one seam that holds both and converts.
    //
    // The third field is the one this seam could not have been given by either
    // side alone: `gunnar-ssh` measured the namespace and marked its ledger,
    // `gunnar-wire` counts the bytes, and neither may name the other's type.
    // See `PackMark` below.
    receive_config.pack_ceiling =
        request
            .pack_ceiling
            .as_ref()
            .map(|ceiling| gunnar_wire::PackCeiling {
                bytes: ceiling.bytes,
                refusal: ceiling.refusal.clone(),
                ledger: request
                    .pack_mark
                    .clone()
                    .map(|mark| Arc::new(PackMark { mark }) as Arc<dyn gunnar_wire::PackLedger>),
            });
    let receive = ReceivePack::new(git, hash_kind).with_config(receive_config);
    // `GET /info/refs?service=git-receive-pack`: the ref list and its
    // capabilities, then stop. No command list, no pack, no gate — the push
    // itself arrives as a separate POST.
    if request.delivery == Delivery::Advertisement && preamble.is_none() {
        receive
            .write_ref_advertisement(output)
            .map_err(wire_error)?;
        output.flush()?;
        tracing::info!(
            target: TARGET_GIT,
            event = "git.receive_pack.advertised",
            elapsed_ms = started.elapsed().as_millis() as u64,
            "served the receive-pack advertisement"
        );
        return Ok(());
    }
    // A preamble already read (the first-repository door) skips the
    // advertisement and the command read: that half happened before the
    // repository existed.
    let outcome = match preamble {
        Some(preamble) => receive
            .serve_after_preamble(preamble, input, output, &ingest, &gate)
            .map(Some),
        None => receive.serve_session(wire_session(request), input, output, &ingest, &gate),
    }
    .map_err(wire_error)?;
    output.flush()?;

    let Some(outcome) = outcome else {
        tracing::info!(
            target: TARGET_GIT,
            event = "git.receive_pack.advertisement_only",
            elapsed_ms = started.elapsed().as_millis() as u64,
            "git-receive-pack: the client took the advertisement and had nothing to push"
        );
        return Ok(());
    };

    // The key this session authenticated with. Carried on the request since
    // The authenticating key itself, not recovered from the identity table.
    let fingerprint = session_key_fingerprint(request);
    // The answerable one. See `session_credential` and GUN-014.
    let credential = session_credential(request);
    let gates = gate.gate_that_refused();
    // **THE counting point for a policy refusal**, per namespace and per push.
    //
    // Once per PUSH and not once per refused reference, deliberately: a push
    // that names four protected branches met one door four times, and a console
    // that reported four refusals would make one mistake look like a campaign.
    // The refs themselves are already in the `git.ref.rejected` lines below and
    // in the `PushRejected` events, which is where per-reference detail belongs.
    if !gates.is_empty() {
        serving.stores.flow().refuse(
            &request.repo.namespace,
            gunnar_core::now_ms(),
            gunnar_core::Refusal::Policy,
        );
    }

    let mut refs_ok = 0usize;
    let mut refs_rejected = 0usize;
    for (command, status) in outcome.commands.iter().zip(&outcome.statuses) {
        match status {
            RefStatus::Ok => {
                refs_ok += 1;
                if command.is_delete() {
                    metrics.add_refs_deleted(1);
                } else {
                    metrics.add_refs_updated(1);
                }
                tracing::info!(
                    target: TARGET_GIT,
                    event = if command.is_delete() { "git.ref.deleted" } else { "git.ref.updated" },
                    ref_name = %command.name,
                    old = %command.old,
                    new = %command.new,
                    "ref updated"
                )
            }
            RefStatus::Rejected(reason) => {
                refs_rejected += 1;
                metrics.add_refs_rejected(1);
                tracing::warn!(
                    target: TARGET_GIT,
                    event = "git.ref.rejected",
                    ref_name = %command.name,
                    old = %command.old,
                    new = %command.new,
                    // The gate that said no, when one did. Empty means nothing
                    // named a gate: a connectivity failure, a lost
                    // compare-and-swap, or the collateral refusal every other
                    // reference gets in an `--atomic` push. The event stream publishes the
                    // same distinction; the log must not blur it.
                    gate = gates.get(&command.name.to_string()).map(String::as_str).unwrap_or(""),
                    %reason,
                    "ref rejected"
                )
            }
        }
        if let Some(bus) = bus {
            // Only a reference that actually moved has a descent question to
            // answer, so the walk is skipped for everything else.
            let forced = status.is_ok() && was_forced(git, command);
            publish(
                bus,
                request,
                command,
                status,
                outcome.atomic,
                &fingerprint,
                &credential,
                &gates,
                forced,
            );
        }
    }

    // A push that moved a reference is the pusher's operation (2026-10-21,
    // the admin's user list). In memory, at most once a minute per person.
    if refs_ok > 0 {
        stores
            .activity()
            .operated(&request.principal.id, gunnar_core::now_ms());
    }

    // ── a repository nobody could clone, made clonable ─────────────────────
    //
    // After the references have moved, and only for a repository whose `HEAD`
    // still resolves to nothing. See
    // [`ServerStores::adopt_default_branch_if_unborn`] for why a push is
    // allowed to decide this and for how narrow the condition is. A failure
    // here is logged and not fatal: the push itself succeeded and its
    // references are durable, and refusing it afterwards would be a lie about
    // what happened.
    let created: Vec<String> = outcome
        .commands
        .iter()
        .zip(&outcome.statuses)
        .filter(|(command, status)| status.is_ok() && !command.is_delete())
        .map(|(command, _)| command.name.to_string())
        .collect();
    if !created.is_empty() {
        if let Err(err) = stores.adopt_default_branch_if_unborn(&request.repo.path(), &created) {
            tracing::warn!(
                target: TARGET_GIT,
                event = "git.default_branch.adoption_failed",
                repo = %request.repo.key(),
                error = format!("{err:#}"),
                "the push landed, but this repository's HEAD names a branch that does not \
                 exist and adopting a pushed one failed; it cannot be cloned until \
                 `set-default-branch` is run against it"
            );
        }
    }

    // ── the one line per push an operator reads ────────────────────────────
    //
    // Applied outputs: how many references actually moved, how many were
    // refused, what the pack cost and how long the whole thing took. A push
    // summary that said only "done" would be indistinguishable from one that
    // moved nothing (measured, not asserted).
    //
    // `objects` is what the push INTRODUCED, and an ingest that could not work
    // that out logs the reason rather than a plausible zero, because a push of
    // nothing and a push whose contents could not be enumerated are different
    // facts.
    let objects_pushed: &[ObjectId] = outcome.new_objects.as_deref().unwrap_or(&[]);
    let objects_unknown = outcome.new_objects.as_ref().err().map(String::as_str);
    // Two different numbers, both worth having: `wire_bytes` is what the client
    // sent and is what a `max_pack_bytes` limit is enforced against;
    // `pack_bytes` is what landed, which is larger for a thin push because
    // completing it writes bases the client did not send.
    //
    // Neither may fall back to `0`. `PushFacts::pack_bytes` is `Option` on
    // purpose — "`Some(0)` for a deletions-only push … `None` only if the count
    // is genuinely unavailable" — and `unwrap_or(0)` threw away exactly the
    // distinction the type exists to carry. `stat` failing on a pack whose
    // bytes are then reported as `0` is the same disease: an absence that
    // renders as a value (and the `objects_unknown` field above).
    let (wire_bytes, wire_bytes_unknown) = measured(
        outcome.pack_bytes,
        "the ingest did not report how many packfile bytes were read off the wire",
    );
    let (pack_bytes, pack_bytes_unknown) = measured_pack_bytes(&outcome);
    if refs_rejected == 0 && outcome.unpack.is_ok() {
        metrics.add_pushes_accepted(1);
    } else {
        metrics.add_pushes_rejected(1);
    }
    tracing::info!(
        target: TARGET_GIT,
        event = "git.receive_pack.served",
        refs_ok,
        refs_rejected,
        atomic = outcome.atomic,
        unpack_ok = outcome.unpack.is_ok(),
        objects = objects_pushed.len(),
        objects_unknown,
        wire_bytes,
        wire_bytes_unknown,
        pack_bytes,
        pack_bytes_unknown,
        push_options = outcome.push_options.len(),
        elapsed_ms = started.elapsed().as_millis() as u64,
        "served git-receive-pack"
    );

    // ── the znippy engine took the objects at ingest ───────────────────────────
    //
    // There is no second step here any more, and its absence is the change.
    //
    // What used to sit at this line read the `.pack` and `.idx` that
    // `index_thin_pack` had written and appended them to an archive — so the
    // tier was a COPY of a filesystem write that had already happened, and the
    // packfile it copied from stayed in `objects/pack` for ever. That is what
    // made every push pay for git's on-disk format even on a server whose
    // storage is Arrow. `PushIngest::ingest_cold` now stores the bytes as they
    // arrive and no packfile is written at all.
    //
    // The `refs_ok > 0` guard that used to wrap this is gone with it, and its
    // reasoning does not transfer: it existed because archiving a
    // fully-rejected push would make the tier the only place that push
    // survived. Storage is no longer a place a push can survive *separately* —
    // it is where the push went — so a rejected ref list now leaves exactly
    // what a rejected ref list leaves on any git server: objects nothing names,
    // for `gunnar gc` to collect.

    // ── push attestation ───────────────────────────────────────────────────
    //
    // Built from what the push actually did, not from what it asked for: a
    // refused reference is not in `ref_updates`, so an attestation can never
    // claim a reference moved that did not.
    //
    // The `Result` goes in whole. Handing `objects_pushed` — which is the `Err`
    // arm already flattened to `&[]` for the log line above — is what made the
    // record claim an empty object list for every push into a repository with
    // no policy gates.
    let attested_objects = outcome.new_objects.as_deref().map_err(String::as_str);
    if let Some(attestation) = attestation_for(
        request,
        &outcome,
        attested_objects,
        &fingerprint,
        &credential,
    ) {
        tracing::info!(
            target: TARGET_GIT,
            event = "git.attestation",
            refs = attestation.ref_updates.len(),
            objects = attestation.new_object_count,
            // Absent from the line when the set WAS enumerated (`Option`
            // records nothing for `None`), so the field's presence is the fact.
            oids_unknown = (!attestation.new_object_oids_unknown.is_empty())
                .then_some(attestation.new_object_oids_unknown.as_str()),
            pack = %attestation.pack_hash,
            key_fingerprint = %attestation.key_fingerprint,
            // The field an HTTP push can be attributed by. `key_fingerprint`
            // above is empty on that transport (GUN-014).
            credential = %attestation.credential,
            "push attestation"
        );
        if let Err(err) = attestations.append(&attestation) {
            // the attestation's durable record failed, and the push has already been
            // reported to the client as done. Saying so loudly is the only
            // correct action; failing here would make gunnar and the client
            // disagree about what is in the repository.
            tracing::error!(
                target: TARGET_GIT,
                path = %attestations.path().display(),
                %err,
                "the push attestation could not be appended to the log; the push itself stands"
            );
        }
        if let Some(bus) = bus {
            bus.publish(event::Payload::PushAttestation(attestation));
        }
    }
    Ok(())
}

// ───────────────────────────────────────────────────────────────────────────
// Measurements, and the absence of one
// ───────────────────────────────────────────────────────────────────────────

/// A measurement split into **the number** and **why there isn't one**.
///
/// `tracing`'s `Value` impl for `Option<T>` records nothing at all for `None`,
/// so the numeric field is *absent* from the line when the measurement is,
/// rather than present and zero. That is the whole point. A reader who sees
/// `pack_bytes=0` has to be able to conclude the push carried no bytes, and
/// they cannot if `0` is also what a failed `stat` prints.
///
/// The same shape as `objects` / `objects_unknown` on the line above it:
/// one field for the fact, one for the reason it is missing, never
/// a plausible default standing in for either.
fn measured(value: Option<u64>, why: &str) -> (Option<u64>, Option<String>) {
    match value {
        Some(bytes) => (Some(bytes), None),
        None => (None, Some(why.to_owned())),
    }
}

/// How many bytes the pack that landed occupies, or why that is unknown.
///
/// Three different absences used to fold into one `0`:
///
/// * the push carried no pack at all, because it was deletions only;
/// * the pack was indexed without a directory, so nothing was written;
/// * `stat` failed on a path that *was* written.
///
/// Only the first is a zero, and it is a **measured** one. The other two are
/// unknowns and now say so, with the path in the reason: an operator chasing a
/// pack that is not where the server thinks it is needs the path more than they
/// need a number. This is a server log, not a client-facing status line, so the
/// path belongs here (the reverse mistake — a server path on the wire to
/// `git push` — was fixed separately).
fn measured_pack_bytes(outcome: &PushOutcome) -> (Option<u64>, Option<String>) {
    let Some(pack) = outcome.pack.as_ref() else {
        // Deletions only: nothing was written because nothing was sent.
        return (Some(0), None);
    };
    let Some(path) = pack.data_path.as_deref() else {
        // No file — which is the ordinary case, not a degraded one: the znippy
        // engine appends the pack verbatim into its archive and writes no
        // `.pack`. The store counted what it appended, so the measurement is
        // right here and only an engine that reported neither is an unknown.
        return match pack.stored_bytes {
            Some(bytes) => (Some(bytes), None),
            None => (
                None,
                Some(format!(
                    "pack {} was indexed without a directory and the store reported no size, so \
                     there is nothing to measure",
                    pack.pack_id
                )),
            ),
        };
    };
    match std::fs::metadata(path) {
        Ok(meta) => (Some(meta.len()), None),
        Err(err) => (
            None,
            Some(format!("{} could not be stat'ed: {err}", path.display())),
        ),
    }
}

/// How many objects a push introduced, for the attestation record.
///
/// **The count never depends on the list.** `Ok(ids)` is an enumerated set and
/// its length is the count, empty included — a push that introduced nothing is
/// a measured zero. `Err(_)` is "the set was never worked out", and the count
/// then comes off the stored pack's own 12-byte header, which is known either
/// way. Zero only for a push that carried no pack at all.
fn attested_object_count(
    new_objects: std::result::Result<&[ObjectId], &str>,
    outcome: &PushOutcome,
) -> u64 {
    match new_objects {
        Ok(ids) => ids.len() as u64,
        Err(_) => outcome
            .pack
            .as_ref()
            .map(|pack| u64::from(pack.num_objects))
            .unwrap_or(0),
    }
}

/// The push attestation record for one push, or `None` when nothing moved.
///
/// A push in which every reference was refused produces `PushRejected` events
/// and no attestation: there is no provenance to attest to, and a record with
/// an empty `ref_updates` would be indistinguishable from a bug in this
/// function.
fn attestation_for(
    request: &ServiceRequest,
    outcome: &PushOutcome,
    new_objects: std::result::Result<&[ObjectId], &str>,
    fingerprint: &str,
    credential: &str,
) -> Option<PushAttestation> {
    let ref_updates: Vec<gunnar_proto::RefUpdate> = outcome
        .commands
        .iter()
        .zip(&outcome.statuses)
        .filter(|(_, status)| status.is_ok())
        .map(|(command, _)| gunnar_proto::RefUpdate {
            ref_name: command.name.to_string(),
            old_oid: command.old.to_string(),
            new_oid: command.new.to_string(),
        })
        .collect();
    if ref_updates.is_empty() {
        return None;
    }

    // The pack as gunnar stored it. `pack_hash` is that pack's own trailing
    // checksum — the `<id>` in `pack-<id>.pack` — in the repository's hash
    // kind, so it is verifiable against the file on disk. It is deliberately
    // NOT a hash of the bytes the client sent: `index_thin_pack` completes a
    // thin pack from the store before writing it, so the received bytes and
    // the stored pack are different objects and only one of them still exists.
    //
    // **Through `measured_pack_bytes`, not through a `stat` of its own.** This
    // used to `std::fs::metadata(data_path).ok()…unwrap_or(0)`, which on an
    // engine that writes no packfile is `None.unwrap_or(0)` — so every
    // attestation this server produced said the push stored zero bytes. Same
    // disease as the empty object list below, in the same record, one field
    // along. The znippy store writes no pack FILE (`data_path: None`) but it
    // counts the bytes it appends, and that count is exactly what
    // `measured_pack_bytes` was taught to prefer over an absent file — one
    // function, so the log line and the durable record can never disagree
    // about how big a push was.
    //
    // The last resort is the wire count (`PushOutcome::pack_bytes`), for an
    // engine that reports neither a file nor a size: a byte count that could
    // not be established at all stays `0` here because the proto has no third
    // state for it, and `new_object_oids_unknown` is the only field that can
    // say the record is partial. What is gone is the case where the count was
    // *available* and thrown away for want of a file.
    let (pack_hash, pack_bytes) = match &outcome.pack {
        Some(pack) => {
            let bytes = measured_pack_bytes(outcome)
                .0
                .or(outcome.pack_bytes)
                .unwrap_or(0);
            (pack.pack_id.to_string(), bytes)
        }
        // A deletions-only push carries no pack at all, and that is a fact
        // worth attesting rather than a reason to skip the record.
        None => (String::new(), 0),
    };
    // ── the object set, and the reason there isn't one ──────────────────────
    //
    // The set is not free: on the znippy engine `oids_in_extent` WAITS for the
    // indexer to absorb the pack, and that wait is what the deferred indexer
    // exists to avoid paying on the push path (`znippy_engine::ingest`'s "why
    // this is a flag" note). So `PushIngest` says how badly it wants the ids
    // rather than always asking — `ObjectSet::Required` when a policy gate
    // reads them (`gate.engine` non-empty), `ObjectSet::Wanted` when only this
    // record does, and `ObjectSet::NotNeeded` under `--no-attest-objects`.
    // Where the set is not produced — an operator who bought the latency back,
    // or an engine that cannot enumerate for a `Wanted` — the ingest says so
    // with `Err(reason)` rather than an empty set.
    //
    // That `Err` used to be flattened to `&[]` by the caller. The record then
    // carried an empty list, and an empty list is what a push of zero objects
    // carries too, so the two were indistinguishable — and the fix before this
    // one made it worse in the honest direction only halfway: the COUNT became
    // the pack's, so a reader saw "3 objects" beside a list of none and no word
    // about why. The reason travels now, in its own field, and the record says
    // what it knows and no more (measured, and never a plausible value where an absence belongs).
    let new_object_count = attested_object_count(new_objects, outcome);
    let (new_object_oids, new_object_oids_unknown) = match new_objects {
        Ok(ids) => (ids.iter().map(ObjectId::to_string).collect(), String::new()),
        Err(reason) => (Vec::new(), reason.to_owned()),
    };

    let received_at_unix_ms = gunnar_grpc::bus::now_unix_ms();
    Some(PushAttestation {
        // Unique per push: the stored pack's identity is unique to the bytes,
        // and the millisecond disambiguates the same pack pushed twice (and a
        // deletions-only push, which has no pack at all).
        attestation_id: format!(
            "{}@{received_at_unix_ms}/{}",
            request.repo.key(),
            if pack_hash.is_empty() {
                "no-pack"
            } else {
                &pack_hash
            }
        ),
        repo: Some(RepoPath::new(&request.repo.namespace, &request.repo.name)),
        principal_id: request.principal.id.clone(),
        principal_name: request.principal.name.clone(),
        key_fingerprint: fingerprint.to_owned(),
        // Answerable on both transports, unlike the fingerprint above
        // (GUN-014). A verifier reading an older record sees an empty string,
        // which is what it saw before this field existed.
        credential: credential.to_owned(),
        hash_kind: crate::control_plane::hash_kind(request.repo.hash_kind) as i32,
        ref_updates,
        // The list is never a partial one: it is either the set the store
        // enumerated or empty beside a reason that says why nobody asked.
        // `new_object_oids_unknown` is what a reader honours; a reader that
        // ignores it sees a count that still came from the pack's own header
        // rather than from a list that was never built.
        new_object_count,
        new_object_oids,
        new_object_oids_unknown,
        new_object_bytes: pack_bytes,
        pack_hash,
        received_at_unix_ms,
        server_agent: gunnar_proto::AGENT.to_owned(),
        // The no-OS rule: git traffic has exactly one transport, and this module is the
        // SSH one. Not read from the request, because there is nothing else it
        // could be, and a field that can only ever hold one value should say
        // which one rather than sit empty.
        transport: "ssh".to_owned(),
    })
}

/// The SHA-256 fingerprint of the key this push authenticated with, when that
/// can be known without guessing.
///
/// The key that authenticated this session, for the attestation and the events.
///
/// # This used to be a guess, and the guess was usually wrong
///
/// The cross-crate gap is closed: `gunnar-ssh` computes the fingerprint in
/// `auth_publickey`, looks the principal up by it, and now **carries it** on
/// [`ServiceRequest`] instead of dropping it.
///
/// What this replaced read the identity table back and answered only when the
/// principal held **exactly one key** — so a machine account with a rotation
/// key, or any human with a laptop key and a desktop key, published an empty
/// fingerprint on every `RefUpdated`, every `RefDeleted` and every attestation.
/// "Which key pushed this" was therefore unanswerable for precisely the
/// accounts an audit trail exists for. Reading it back could never have been
/// made correct: the table cannot say which of two keys made a connection.
///
/// # An empty string here does NOT mean "nobody" (GUN-014)
///
/// It used to say so: *"an empty string means no key, and now means it for one
/// reason only — the anonymous, read-only path, which authenticates with no key
/// at all. That path cannot push, so it cannot reach an attestation."* That was
/// false for smart HTTP. HTTP authenticates with a **token**, so there is no
/// key and `ServiceRequest::key_fingerprint` is `None` on every HTTP request
/// including `git-receive-pack` — and the empty string went into `RefUpdated`,
/// `RefDeleted` and the attestation with nothing else on the record naming who
/// pushed. Two HTTP pushes by two different people were the same audit line.
///
/// The field is still exactly what it says: the SSH key, when there was one.
/// The question "which credential" is [`session_credential`]'s.
fn session_key_fingerprint(request: &ServiceRequest) -> String {
    match &request.key_fingerprint {
        Some(fingerprint) => fingerprint.as_str().to_owned(),
        None => {
            tracing::debug!(
                target: TARGET_GIT,
                principal = %request.principal.id,
                "no key fingerprint on this request: the session authenticated anonymously"
            );
            String::new()
        }
    }
}

/// **Which credential made this request**, as the audit trail names it:
/// `key:SHA256:…`, `token:<hash prefix>`, or empty for a caller that presented
/// none.
///
/// The field [`session_key_fingerprint`] could not be (GUN-014). It is derived
/// here rather than spelled by each front door so the two transports cannot
/// come to write the prefix differently: SSH already carries the fingerprint
/// and gets `key:` added to it; HTTP has no key and carries
/// [`ServiceRequest::credential`] instead.
///
/// Empty now genuinely has one meaning — no credential — and the anonymous read
/// path is the only thing that produces it, which is what the old note claimed
/// and this makes true.
fn session_credential(request: &ServiceRequest) -> String {
    match (&request.key_fingerprint, &request.credential) {
        (Some(fingerprint), _) => format!("key:{}", fingerprint.as_str()),
        (None, Some(credential)) => credential.clone(),
        (None, None) => String::new(),
    }
}

// `pack_object_ids` stood here: it opened a written `.idx` with
// `gix_pack::index::File::at` and collected every oid in it. It went with
// `index_thin_pack`, which is the only thing that ever wrote an `.idx` for it
// to read — a push is a verbatim `put_pack` into the engine now, and the ids a
// push introduced come from `PushReceipt::new_oids` off the engine's own index.
//
// It carried the last two live bumps of `gunnar_pack::oidprobe`'s
// `GIX_PACK_INDEX_OPEN` / `GIX_PACK_INDEX_ITER`, which `bench/examples/
// oidlookup-probe.rs` reports. Deleting it changes **no measurement**: the
// function had no caller, so those counters were already zero in every run that
// could be taken. A counter that reads 0 because the path is gone is the truth;
// keeping an uncalled function alive to hold a probe open would be the other
// way round — an instrument attached to code that cannot execute.

/// Taking a pushed pack into `objects/pack`, and saying which objects it
/// **introduced**.
///
/// # The subtraction, and why it is the only honest source
///
/// A gate cannot enforce a quota against a set that includes objects the
/// repository already held. This used to hand the gate every id in the written
/// `.idx`, which for a thin push is the client's objects *plus* the bases
/// `index_thin_pack` fetched from the store to complete it — so a
/// `max_objects` gate refused pushes it need not have, and no amount of
/// re-reading the index afterwards could tell the two apart, because by then
/// they are the same file.
///
/// The distinction exists in exactly one place: the `gix_object::Find` that
/// resolution asks for a missing base. Every id it is asked for and finds is,
/// by definition, an object the repository already had. So the lookup is
/// wrapped, the ids it served are recorded, and
///
/// ```text
///     introduced  =  everything in the .idx  −  the bases we handed back
/// ```
///
/// is exact rather than conservative, and needs no second enumeration of
/// anything (one implementation).
/// **Where a pushed or fetched pack is framed**: the box's transaction log, and the catalog
/// key and hash kind the twin's apply opens the repository by (`gunnar_core::txlog`).
#[derive(Clone)]
pub(crate) struct PackFrame {
    log: Arc<gunnar_core::txlog::Log>,
    repo: String,
    hash: String,
}

impl PackFrame {
    /// The frame for `repo` on `stores`.
    pub(crate) fn of(stores: &crate::stores::ServerStores, repo: &gunnar_core::Repo) -> PackFrame {
        PackFrame {
            log: Arc::clone(stores.txlog()),
            repo: repo.key(),
            hash: repo.hash_kind.to_string(),
        }
    }

    /// These exact bytes, as a `Pack` frame, durable before this returns.
    fn append(&self, bytes: &[u8]) -> anyhow::Result<()> {
        let meta = gunnar_core::txlog::Meta::Pack {
            repo: self.repo.clone(),
            hash: self.hash.clone(),
        };
        self.log
            .append(gunnar_core::txlog::Kind::Pack, &meta, bytes)?;
        Ok(())
    }
}

pub(crate) struct PushIngest<'a> {
    git_dir: PathBuf,
    hash_kind: gix_hash::Kind,
    /// Which engine stores this push. Carried on the value rather than read
    /// from a global, because the process's engine is a field on
    /// [`ServerStores`](crate::stores::ServerStores) now and a push is served
    /// out of one of those.
    engine: crate::storage::Engine,
    /// How badly this push needs the object set it introduced: a policy gate
    /// that reads the objects cannot decide without it, the push attestation
    /// would rather have it than apologise for it, and an operator can decline
    /// it altogether. Asking for it waits on index work the ack path defers on
    /// purpose, which is why this is three answers and not a `bool`.
    objects: crate::storage::ObjectSet,
    /// **Where the running object count goes, or `None`.**
    ///
    /// The numerator of a transfer's progress, and the reason it is on the
    /// *ingest* rather than on `gunnar-wire`'s `FetchProgress` beside the
    /// denominator: the count exists only inside the engine's pack walk, and
    /// there is exactly one of those (one implementation). `fetch_pack` reads the header and
    /// can publish `objects_total` because twelve bytes are all that costs;
    /// counting the objects themselves would be a second decoder, which its own
    /// module doc refuses. So the two halves of one progress pair are published
    /// from two places, and this is the half storage owns.
    watching: Option<&'a dyn git_storage_trait::PackProgress>,
    /// **The transaction log this pack is a frame in**, or `None` (a test, a tool). Handed
    /// to `Engine::ingest_push` as `before_store`: the pack's exact bytes are a `Pack` frame
    /// before the engine stores them — this path never passes the `OpenRepo::logged` tee.
    frame: Option<PackFrame>,
    /// **An encrypted repository's objects, or `None`** (crypt-level2 item 4,
    /// 2026-10-06): when set, the pushed pack is checked in quarantine
    /// ([`crate::encrypted_quarantine`]) BEFORE it is framed or stored — these
    /// are the bases a thin pack deltas against. A refusal keeps nothing.
    encrypted: Option<&'a dyn git_storage_trait::GitServe>,
    /// **The data set's space ledger**, or `None` (a test, a tool, a server
    /// whose doors were never wired): every chunk of the pack is admitted
    /// against it BEFORE it is read, and a pack that does not fit beside what
    /// is already arriving is refused by name — never failed by a full disk
    /// mid-receive (cov-13, [`crate::push_space`]).
    space: Option<Arc<gunnar_ssh::SpaceLedger>>,
}

impl<'a> PushIngest<'a> {
    /// **The ingest a MIRROR lands its fetched pack through** — this one, the
    /// push's own, and not a second sink.
    ///
    /// `fetch_pack`'s module doc states the principle it is built on: *a
    /// fetched pack and a pushed pack are the same bytes*, so both land through
    /// the identical [`gunnar_wire::PackIngest`] and znippy implements it once.
    /// A private constructor rather than a public struct because the fields are
    /// this module's business; what a mirror needs to say is the four facts
    /// below.
    pub(crate) fn new(
        git_dir: PathBuf,
        hash_kind: gix_hash::Kind,
        engine: crate::storage::Engine,
        objects: crate::storage::ObjectSet,
        watching: Option<&'a dyn git_storage_trait::PackProgress>,
        frame: Option<PackFrame>,
    ) -> Self {
        PushIngest {
            git_dir,
            hash_kind,
            engine,
            objects,
            watching,
            frame,
            // A mirror's fetched pack is not checked here: the repository it
            // lands in is the mirror's own (an encrypted repository is not a
            // mirror target in Level 1).
            encrypted: None,
            space: None,
        }
    }

    /// Hold the pack to the data set's room — `stores.space_ledger()`.
    pub(crate) fn with_space(mut self, space: Option<Arc<gunnar_ssh::SpaceLedger>>) -> Self {
        self.space = space;
        self
    }

    /// **The push, stored without a packfile.**
    ///
    /// The bytes go from the wire into the znippy engine's own storage and nothing
    /// is written to `objects/pack` — no `.pack`, no `.idx`, no `.keep`. That is
    /// the whole point of it, and it is why this does not call
    /// [`index_thin_pack`]: `gix_pack::Bundle::write_to_directory` takes a
    /// directory, so every route through it puts a git packfile on disk before
    /// the tier ever sees the bytes, and the tier's copy is then a second one.
    ///
    /// # What is given up, said out loud
    ///
    /// `index_thin_pack` **resolved** the pack: it completed thin deltas against
    /// the store and produced a real `.idx`. This does not. The closure check
    /// inside `put_pack` refuses a pack whose bases are missing, which is the
    /// correctness half, but the pack stays thin and stays verbatim — the whole
    /// claim of *verbatim is the truth*. A reader resolves chains on the way
    /// out instead.
    ///
    /// # The `.keep` that is not there
    ///
    /// `index_thin_pack` wrote one and the caller dropped it once the refs
    /// landed, guarding the new pack from a concurrent `gc` in the window
    /// between. There is no window here: `put_pack` returns only after the
    /// journal row is fsynced, and `gc` reads that journal. The marker's job is
    /// done by the ordering rather than by a file.
    fn ingest_engine(&self, pack: &mut dyn BufRead) -> gunnar_wire::Result<PackReceipt> {
        // ── space admission (cov-13) ─────────────────────────────────────
        //
        // The promise this pack holds on /data: spool + store, + the log's
        // frame when it is framed. Given back copy by copy as each lands (the
        // spool at the reader's `consume`, the frame in the hook below, the
        // store when this function returns and the hold drops).
        let hold = self.space.as_ref().map(|ledger| {
            ledger.hold(if self.frame.is_some() {
                gunnar_ssh::PUSH_COPIES
            } else {
                gunnar_ssh::PUSH_COPIES - 1
            })
        });
        let mut gated = None;
        let reader: &mut dyn BufRead = match &hold {
            Some(hold) => gated.insert(crate::push_space::Gated::new(
                pack,
                hold,
                crate::storage::SPOOL_ABOVE as u64,
            )),
            None => pack,
        };
        let hook = |bytes: &[u8]| -> anyhow::Result<()> {
            // ★ Before the frame and before the store: an encrypted
            // repository's pack is checked in quarantine, and a refusal keeps
            // nothing — not on this disk, not in the log the twin replays.
            #[cfg(feature = "engine-znippy")]
            if let Some(repo) = self.encrypted {
                let checked = crate::encrypted_quarantine::check(bytes, self.hash_kind, repo)?;
                tracing::info!(
                    target: TARGET_STORE,
                    event = "store.encrypted.quarantine_passed",
                    objects = checked,
                    "an encrypted repository's pushed pack passed its check in quarantine; it is framed and stored now"
                );
            }
            match &self.frame {
                Some(frame) => {
                    frame.append(bytes)?;
                    // On the disk now: the frame's promise is `statvfs`'s.
                    if let Some(hold) = &hold {
                        hold.realize(bytes.len() as u64);
                    }
                    Ok(())
                }
                None => Ok(()),
            }
        };
        let hooked = self.frame.is_some() || self.encrypted.is_some();
        // The mutation gate spans frame through store (`Log::mutation`): a snapshot's freeze
        // waits for this pack to be in the engine before it reads the head.
        let _framed = self.frame.as_ref().map(|f| f.log.mutation());
        let stored = self.engine.ingest_push(
            self.git_dir.as_path(),
            self.hash_kind,
            reader,
            self.objects,
            self.watching,
            hooked.then_some(&hook as &dyn Fn(&[u8]) -> anyhow::Result<()>),
        );
        // ★ Cut for room: the ledger's line, verbatim, whatever the ingest made
        // of the cut stream. Its spool is already gone (the failed read dropped
        // it), nothing was framed, nothing stored, and its promise is given back.
        let consumed_by_cut = gated
            .as_ref()
            .map(|g| (g.cut().map(str::to_owned), g.consumed()));
        drop(gated);
        if let Some((Some(line), sent)) = consumed_by_cut.clone() {
            tracing::warn!(
                target: TARGET_STORE,
                event = "store.push.no_room",
                pack_bytes = sent,
                refusal = %line,
                "a pushed pack was refused for room mid-receive: it did not fit beside the \
                 pushes already arriving; its spool is deleted and nothing of it was kept"
            );
            return Err(gunnar_wire::Error::Refused(line));
        }
        let received = stored
            .map_err(|err| {
                // ★ The disk filled ANYWAY (a writer outside the ledger): said in
                // the ledger's words, never `uring: op failed: Operation canceled`.
                if let Some(hold) = &hold {
                    let freed = consumed_by_cut.as_ref().map_or(0, |(_, sent)| *sent);
                    if crate::push_space::is_space_failure(&err, || hold.at_the_floor(freed)) {
                        let line = hold.full_now();
                        tracing::warn!(
                            target: TARGET_STORE,
                            event = "store.push.disk_full",
                            error = %format!("{err:#}"),
                            refusal = %line,
                            "the store failed a pushed pack because the data set filled; told by name"
                        );
                        return gunnar_wire::Error::Refused(line);
                    }
                }
                #[cfg(feature = "engine-znippy")]
                if let Some(refused) = crate::encrypted_quarantine::refusal_in(&err) {
                    tracing::info!(
                        target: TARGET_STORE,
                        event = "store.encrypted.quarantine_refused",
                        reason = %refused,
                        "an encrypted repository's pushed pack was refused in quarantine: nothing was framed or stored"
                    );
                    return gunnar_wire::Error::Refused(refused);
                }
                // `{err:#}`, not `{err}`. `anyhow`'s plain `Display` prints the
                // OUTERMOST context and nothing under it, so a failure three
                // `.context()` calls deep arrived as "storing the pushed pack:
                // enumerating the objects this push introduced" — the two
                // labels this code added, and never the cause the store gave.
                // The alternate form joins the whole chain, which is the only
                // version of this line an operator can act on.
                gunnar_wire::Error::Store(gunnar_store::Error::backend(format!(
                    "storing the pushed pack: {err:#}"
                )))
            })?;

        let pack_id = gix_hash::ObjectId::try_from(received.pack_id.as_slice()).map_err(|_| {
            gunnar_wire::Error::Store(gunnar_store::Error::backend(format!(
                "the stored pack's trailer is {} bytes, which is not a {} object id",
                received.pack_id.len(),
                self.hash_kind
            )))
        })?;

        tracing::info!(
            target: TARGET_STORE,
            // Renamed from `store.cold_tier.ingested` 2026-08-14. The cold tier
            // was purged and this event outlived it, so every push in the log
            // still announced a tier that does not exist. Nothing asserted on
            // the old name — only a doc-comment example, corrected with it.
            event = "store.znippy.ingested",
            objects = received.num_objects,
            bytes = received.pack_bytes,
            elapsed_ms = received.elapsed_ms,
            object_set = received.new_oids.is_some(),
            "the push was stored in the znippy-git store; no packfile was written"
        );

        let new_objects = match received.new_oids {
            Some(oids) => oids
                .iter()
                .map(|raw| {
                    gix_hash::ObjectId::try_from(raw.as_slice())
                        .map_err(|_| format!("the store returned a {}-byte object id", raw.len()))
                })
                .collect::<std::result::Result<Vec<_>, _>>(),
            // Not "no objects" — see `ColdReceipt::new_oids`. A gate reading
            // this refuses, which is the correct answer to "I was never asked
            // to find out".
            None => Err(
                "the policy has no gate over objects, so the object set was not \
                         enumerated"
                    .to_owned(),
            ),
        };

        Ok(PackReceipt {
            indexed: gunnar_wire::pack::IndexedPack {
                num_objects: received.num_objects,
                pack_id,
                // There is no index file, and `ObjectId::null` is this tree's
                // existing spelling for that — the same value the git-store
                // bench server and the store's own end-to-end test use.
                index_id: gix_hash::ObjectId::null(self.hash_kind),
                pack_version: gunnar_wire::pack::PackVersion::from_u32(received.version),
                hash_kind: self.hash_kind,
                // The three paths that no longer exist. `None` is not a
                // degraded answer here: it is the fact, and `cold_tier_append`
                // used to read exactly these two to build its second copy.
                index_path: None,
                data_path: None,
                keep_path: None,
                // …but the SIZE is not missing with them. The store counted the
                // bytes it appended, and until this line nobody carried the
                // number out, so `measured_pack_bytes` fell through to "there
                // is nothing on disk to measure" on every push into the engine
                // the appliance ships. There is no file; there is a
                // measurement. Nothing on disk has to be stat'ed to learn it,
                // and on this engine nothing on disk could be.
                stored_bytes: Some(received.pack_bytes),
            },
            new_objects,
        })
    }
}

impl gunnar_wire::PackIngest for PushIngest<'_> {
    /// Straight into the engine. There is no second arm.
    ///
    /// What stood beside this ran `index_thin_pack` — `gix_pack::Bundle::
    /// write_to_directory` — writing a real `.pack`, `.idx` and `.keep` into
    /// `objects/pack`, reading the `.idx` back to learn which objects arrived,
    /// and dropping the `.keep` afterwards. Every line of it was the packfile
    /// engine's, and the engine is amputated.
    fn ingest(&self, pack: &mut dyn BufRead) -> gunnar_wire::Result<PackReceipt> {
        self.ingest_engine(pack)
    }
}

/// Did this reference move to a tip that does **not** descend from the one it
/// held? The event stream's `RefUpdated.forced`.
///
/// # This is one ancestry walk, not a second one
///
/// [`gunnar_policy::ancestry::is_descendant_of`] is the same function
/// `BranchProtectionGate` decides a non-fast-forward with, and it is the one
/// cross-checked against `git merge-base --is-ancestor` in that crate's tests.
/// Re-deriving the answer here — `old != new`, a hex compare, a first-parent
/// walk — is the mistake that gate's module docs enumerate, and two
/// implementations that disagree would put a UI and a policy on different sides
/// of the same push (one implementation).
///
/// The gate cannot simply be asked, because it only runs where a protection
/// rule matches. The audit trail has to answer for **every** reference on every
/// repository, including the ones nothing protects, which is the case where
/// nobody is watching.
///
/// # An undecidable walk reports `true`
///
/// A missing object or an exhausted budget is not evidence of a fast-forward.
/// `BranchProtectionGate` refuses on undecided rather than allowing; the same
/// judgement here means flagging rather than clearing, with the reason logged at
/// `warn`. Reporting `false` for a walk that never finished is exactly the
/// wrong-answer this field is being fixed for.
fn was_forced(git: &dyn GitServe, command: &RefCommand) -> bool {
    // A create has no previous tip, and a delete has no new one. Neither is a
    // rewrite of anything.
    if command.is_create() || command.is_delete() {
        return false;
    }
    match gunnar_policy::ancestry::is_descendant_of(
        git,
        &command.new,
        &command.old,
        gunnar_policy::ancestry::DEFAULT_ANCESTRY_BUDGET,
    ) {
        Ok(descends) => !descends,
        Err(err) => {
            tracing::warn!(
                target: TARGET_POLICY,
                ref_name = %command.name,
                old = %command.old,
                new = %command.new,
                %err,
                "the ancestry walk behind RefUpdated.forced could not be completed; the update \
                 is published as forced, because `not forced` is a claim this could not make"
            );
            true
        }
    }
}

#[allow(clippy::too_many_arguments)]
fn publish(
    bus: &EventBus,
    request: &ServiceRequest,
    command: &RefCommand,
    status: &RefStatus,
    atomic: bool,
    fingerprint: &str,
    credential: &str,
    gates: &HashMap<String, String>,
    forced: bool,
) {
    let repo = Some(RepoPath::new(&request.repo.namespace, &request.repo.name));
    let payload = match status {
        RefStatus::Ok if command.is_delete() => event::Payload::RefDeleted(RefDeleted {
            repo,
            ref_name: command.name.to_string(),
            old_oid: command.old.to_string(),
            principal_id: request.principal.id.clone(),
            key_fingerprint: fingerprint.to_owned(),
            credential: credential.to_owned(),
        }),
        RefStatus::Ok => event::Payload::RefUpdated(RefUpdated {
            repo,
            ref_name: command.name.to_string(),
            old_oid: command.old.to_string(),
            new_oid: command.new.to_string(),
            principal_id: request.principal.id.clone(),
            key_fingerprint: fingerprint.to_owned(),
            credential: credential.to_owned(),
            // Answered by [`was_forced`], not assumed. This was a hardcoded
            // `false`, which made every force-push on the server
            // indistinguishable from an ordinary one on `Events.Watch`.
            forced,
        }),
        RefStatus::Rejected(reason) => event::Payload::PushRejected(PushRejected {
            repo,
            ref_name: command.name.to_string(),
            // Empty when nothing named a gate: a connectivity failure, a
            // compare-and-swap that lost, or the collateral refusal every other
            // reference gets in an `atomic` push. Those are real refusals and
            // they belong on the stream, but attributing them to a gate that
            // never ran would be a lie a UI would repeat.
            gate: gates
                .get(&command.name.to_string())
                .cloned()
                .unwrap_or_default(),
            reason: reason.clone(),
            principal_id: request.principal.id.clone(),
            whole_push_rejected: atomic,
        }),
    };
    bus.publish(payload);
}

// ───────────────────────────────────────────────────────────────────────────
// the policy seam
// ───────────────────────────────────────────────────────────────────────────

/// `gunnar-wire`'s [`PushGate`] over `gunnar-policy`'s [`PolicyEngine`].
struct PolicyGate {
    engine: PolicyEngine,
    /// Why this repository's policy could not be read, if it could not be.
    ///
    /// `Some` makes [`PolicyGate::check`] refuse **every** reference (GUN-007).
    /// An unreadable policy used to become `PolicyConfig::default()`, and an
    /// empty engine is let through by the `is_empty()` branch below — so a
    /// document with a typo, or a storage fault, quietly turned branch
    /// protection and `require_signature` off for the duration. This is the
    /// field that keeps "no gates are configured" and "the gates could not be
    /// read" from being the same value.
    unreadable: Option<String>,
    principal: gunnar_policy::Principal,
    repo: RepoId,
    keys: Box<dyn KeyDirectory>,
    metrics: Arc<ServerMetrics>,
    policy_path: PathBuf,
    /// Reference name → the `Gate::name()` that refused it.
    ///
    /// `PushGate::check` may only answer `Err(String)`, so the *name* of the
    /// gate that said no is lost at that boundary; it survives only inside the
    /// `Decision`. The event stream's `PushRejected.gate` needs it, so it is recorded here
    /// on the way past rather than parsed back out of the reason string; a
    /// reason is written for a human and its shape is free to change.
    refused_by: Mutex<HashMap<String, String>>,
}

/// `gunnar-core`'s principal kind as `gunnar-policy` names it.
///
/// # The fourth arm, which must never be taken
///
/// A [`PolicyGate`] is built only for a **push**, and a push needs
/// [`Level::Write`](gunnar_core::Level::Write), which
/// `gunnar_core::AuthTable::may` refuses to the anonymous principal
/// unconditionally, before `Dispatch::authorise` returns and long before this
/// runs. So the anonymous arm is unreachable through every route that exists.
///
/// It is loud rather than plausible. `gunnar-policy` has three kinds and none of
/// them means "nobody"; quietly answering `User` would put an anonymous push
/// through gates written about humans, and do it silently. If this line is ever
/// reached, the ACL has stopped being consulted before the gate. That is a
/// security incident, and it reads like one in the log.
pub(crate) fn policy_principal_kind(
    kind: gunnar_core::PrincipalKind,
) -> gunnar_policy::PrincipalKind {
    match kind {
        gunnar_core::PrincipalKind::User => gunnar_policy::PrincipalKind::User,
        gunnar_core::PrincipalKind::Machine => gunnar_policy::PrincipalKind::Machine,
        gunnar_core::PrincipalKind::Deploy => gunnar_policy::PrincipalKind::Deploy,
        // A group holds no keys and is never authenticated as (punkt 9), so a
        // push can never be attributed to one — the same unreachable-by-design
        // shape as the anonymous arm below, and loud for the same reason: a
        // group at the pre-receive gate means a keyless identity was handed a
        // push, which is the ACL being bypassed.
        gunnar_core::PrincipalKind::Group => {
            tracing::error!(
                target: gunnar_core::TARGET_AUDIT,
                event = "policy.gate.group_push",
                "a GROUP principal reached the pre-receive gate. A group holds no keys and \
                 cannot authenticate, so this cannot happen unless the grant check has been \
                 bypassed. Treating it as a machine identity for the gate; investigate \
                 immediately."
            );
            debug_assert!(
                false,
                "a group principal reached the pre-receive gate (punkt 9)"
            );
            gunnar_policy::PrincipalKind::Machine
        }
        gunnar_core::PrincipalKind::Anonymous => {
            tracing::error!(
                target: gunnar_core::TARGET_AUDIT,
                event = "policy.gate.anonymous_push",
                "an ANONYMOUS principal reached the pre-receive gate. Anonymous access is \
                 read-only and the ACL refuses it Write, so this cannot happen unless the \
                 grant check has been bypassed. Treating it as a machine identity for the \
                 gate; investigate immediately."
            );
            debug_assert!(false, "an anonymous principal reached the pre-receive gate");
            gunnar_policy::PrincipalKind::Machine
        }
    }
}

impl PolicyGate {
    fn open(
        git_dir: &Path,
        request: &ServiceRequest,
        stores: &ServerStores,
        metrics: Arc<ServerMetrics>,
    ) -> Self {
        let policy_path = git_dir.join(POLICY_FILE);
        let (config, unreadable) = match load_policy(stores, &request.repo, &policy_path) {
            Ok(config) => (config, None),
            // An empty engine on purpose: nothing should be evaluated against
            // a policy that could not be read, and `check` refuses before it
            // would be.
            Err(why) => (PolicyConfig::default(), Some(why)),
        };
        // An ENCRYPTED repository adds the plaintext check to whatever policy
        // it has: the flag is the account admin's, not the policy file's, so
        // no policy edit can switch it off.
        let mut gates = config.gates();
        if request.repo.encrypted {
            gates.push(Box::new(gunnar_policy::EncryptedGate::new()));
        }
        PolicyGate {
            engine: PolicyEngine::new(gates),
            unreadable,
            principal: gunnar_policy::Principal {
                id: request.principal.id.clone(),
                name: request.principal.name.clone(),
                kind: policy_principal_kind(request.principal.kind),
            },
            repo: RepoId::new(&request.repo.namespace, &request.repo.name),
            keys: Box::new(key_directory(stores)),
            metrics,
            policy_path,
            refused_by: Mutex::new(HashMap::new()),
        }
    }

    /// Which gate refused each reference, for the event-stream `PushRejected` events.
    ///
    /// Read after `ReceivePack::serve` has returned, so every `check` that was
    /// going to happen has happened.
    fn gate_that_refused(&self) -> HashMap<String, String> {
        self.refused_by
            .lock()
            .expect("the refusing-gate lock")
            .clone()
    }
}

impl PushGate for PolicyGate {
    fn check(
        &self,
        command: &RefCommand,
        facts: &PushFacts<'_>,
    ) -> std::result::Result<(), String> {
        // One span per gate evaluation. It is a child of
        // `git.receive_pack`, so the connection, principal and repository come
        // for free; what is new is the reference and how long deciding it took.
        // Branch protection walks ancestry, and an ancestry walk on a large
        // repository is one of the two places a push spends unexplained
        // seconds.
        let span = tracing::debug_span!(
            target: TARGET_POLICY,
            "policy.gate",
            ref_name = %command.name,
        );
        let _entered = span.enter();
        let started = std::time::Instant::now();
        // ── the policy could not be READ, which is not "there is none" ─────
        //
        // First, before `is_empty()`, because `is_empty()` is the branch that
        // lets a push through and an unreadable policy produced exactly that
        // value (GUN-007). A pre-receive refusal costs pushes and nothing
        // else: clone, fetch and browse are untouched, and the operator is
        // told which document and why.
        if let Some(why) = &self.unreadable {
            self.metrics.add_gate_refusals(1);
            tracing::error!(
                target: TARGET_POLICY,
                event = "policy.gate.unreadable",
                ref_name = %command.name,
                %why,
                "the policy for this repository could not be read, so every gate refuses"
            );
            return Err(format!(
                "gunnar could not read the policy for this repository, so it refuses this push \
                 rather than serving it with no gates at all: {why}. Reads are unaffected; fix \
                 the document (or `gunnar control policy set-protection`) and push again."
            ));
        }
        if self.engine.is_empty() {
            // No gates are configured, so there is nothing to decide. Said out
            // loud rather than reached by an engine that happens to return an
            // empty decision, because "allowed by every gate" and "there are no
            // gates" are different facts.
            return Ok(());
        }
        // A gate that cannot see what was pushed cannot decide, and an
        // undecidable gate refuses (the pre-receive gate). This is the branch that keeps a
        // configured limit from being silently satisfied by an empty object set.
        let new_objects = match facts.new_objects {
            Ok(ids) => ids,
            // No live gate reads the set, so its absence decides nothing.
            // The empty slice is never read: `reads_new_objects()` is exactly
            // the question "would any gate look at it".
            Err(_) if !self.engine.reads_new_objects() => &[],
            Err(reason) => {
                self.metrics.add_gate_refusals(1);
                tracing::error!(
                    target: TARGET_POLICY,
                    event = "policy.gate.undecidable",
                    %reason,
                    "the pushed objects could not be enumerated, so every configured gate refuses"
                );
                return Err(format!(
                    "gunnar could not enumerate the pushed objects, so the policy gates in {} \
                     could not decide: {reason}",
                    self.policy_path.display()
                ));
            }
        };
        let updates = [RefUpdate::from_wire(
            command.name.clone(),
            command.old,
            command.new,
        )];
        let mut ctx = PushContext::new(
            &self.principal,
            &self.repo,
            &updates,
            new_objects,
            facts.objects,
            self.keys.as_ref(),
        );
        // The byte count reaches the gate. Without it
        // `LimitsGate` refuses any push at all whenever `max_pack_bytes` is
        // configured, because it will not pretend an unmeasured pack is under a
        // limit.
        if let Some(bytes) = facts.pack_bytes {
            ctx = ctx.with_pack_bytes(bytes);
        }
        // The push options reach the gate. Unconditionally, including the empty
        // slice: `Some(&[])` is "the client sent none" and `None` is "nobody
        // looked", and `PushOptionsGate` refuses the second. Before this line
        // the whole tree's only consumer of a parsed push option was `.len()`
        // in the log field twenty lines below, so `push-options` was advertised
        // on the wire and discarded.
        ctx = ctx.with_push_options(facts.push_options);
        let decision = self.engine.evaluate(&ctx);
        let elapsed_ms = started.elapsed().as_millis() as u64;
        if decision.is_allowed() {
            tracing::debug!(
                target: TARGET_POLICY,
                event = "policy.gate.allowed",
                gates = self.engine.gate_names().len(),
                new_objects = new_objects.len(),
                elapsed_ms,
                "every gate allowed this reference"
            );
            Ok(())
        } else {
            // `evaluate` runs every gate, so a reference can be refused by more
            // than one. The first in gate order is the one named, and the
            // summary leads with the same one, so a UI and the client's `ng`
            // line agree about who said no.
            self.metrics.add_gate_refusals(1);
            if let Some(first) = decision.rejections.first() {
                // **`warn!`, and it NAMES THE GATE.** The audit trail's `Verify` asks for
                // exactly one error line per refused push naming the gate and
                // the reference; before this, a refused push produced a
                // `PushRejected` event with an empty `gate` field and no log
                // line at all that said who refused it.
                tracing::warn!(
                    target: TARGET_POLICY,
                    event = "policy.gate.refused",
                    gate = %first.gate,
                    reason = %first.reason,
                    refused_by_gates = decision.rejections.len(),
                    elapsed_ms,
                    "a policy gate refused this reference"
                );
                self.refused_by
                    .lock()
                    .expect("the refusing-gate lock")
                    .insert(command.name.to_string(), first.gate.clone());
            } else {
                // `is_allowed()` said no and named nobody. That is a bug in the
                // engine, not a refusal, and it must not read as a normal one.
                tracing::error!(
                    target: TARGET_POLICY,
                    event = "policy.gate.refused",
                    gate = "",
                    elapsed_ms,
                    "the policy engine refused a reference without naming a gate"
                );
            }
            Err(decision.summary())
        }
    }
}

/// Read the repository's policy document.
///
/// **`Err` is not an empty policy** (GUN-007). Until 2026-09-11 every failure
/// here — a document that did not parse, a control store that could not be
/// read, a file that could not be opened — returned `PolicyConfig::default()`,
/// and `PolicyGate::check` lets an empty engine through because "no gates are
/// configured" is a real and common answer. So a typo in a policy document, or
/// a transient storage fault, silently switched branch protection and the
/// signature requirement OFF for that push, at `error!` level in a log nobody
/// was reading at the time.
///
/// The old note here said refusing would be worse, because a typo would take
/// the repository offline. It does not: this is the PRE-RECEIVE gate, so a
/// refusal costs pushes and nothing else — clone, fetch, browse and every read
/// are untouched, and the operator is told which document and why. Degraded
/// operation must not be indistinguishable from full operation, which is the
/// same judgement `ancestry` already makes when a walk cannot finish.
///
/// `Ok(PolicyConfig::default())` survives for exactly one case and it is a
/// fact rather than a fault: nothing is stored and no file exists, so nothing
/// has been configured.
/// **The control store first, the file beside the repository second.**
///
/// `POLICY_FILE`'s constant used to carry the note *"When the catalog grows a
/// policy document, this constant is the only thing that changes."* The catalog
/// has grown one — `gunnar_core::PolicyStore`, at `refs/gunnar/policy` — and
/// this is that change. What did **not** change is the file: a repository
/// configured before the store existed keeps working, and an operator who edits
/// the JSON on disk still gets what they edited, until they write a policy over
/// the control plane and the stored one takes over.
///
/// The order is deliberate and is the only one that can be right. The stored
/// document is the one `Policy.SetProtection` writes and the one an operator
/// can see over gRPC; if a stale file could shadow it, turning protection on
/// through the console would appear to work and change nothing.
pub(crate) fn load_policy(
    stores: &ServerStores,
    repo: &gunnar_core::Repo,
    path: &Path,
) -> std::result::Result<PolicyConfig, String> {
    match stores.policies().get(&repo.path()) {
        Ok(Some(document)) => match PolicyConfig::from_json(&document) {
            Ok(config) => return Ok(config),
            Err(err) => {
                tracing::error!(
                    target: TARGET_POLICY,
                    repo = %repo.key(),
                    source = "control-store",
                    %err,
                    "the stored policy document did not parse; pushes to this repository are \
                     refused until it does"
                );
                return Err(format!("the stored policy document did not parse: {err}"));
            }
        },
        // Nothing stored. The file below is the pre-store shape and is still
        // the answer for a repository configured before `PolicyStore` existed.
        Ok(None) => {}
        Err(err) => {
            // NOT a fall-through to the file any more. The file "may be older
            // than what an operator configured" — its own words — so serving
            // it when the authoritative store is unreachable is a silent
            // downgrade to a policy nobody chose.
            tracing::error!(
                target: TARGET_POLICY,
                repo = %repo.key(),
                %err,
                "the policy store could not be read; pushes to this repository are refused \
                 rather than gated by a file that may be older than what an operator \
                 configured"
            );
            return Err(format!("the policy store could not be read: {err}"));
        }
    }
    let text = match std::fs::read_to_string(path) {
        Ok(text) => text,
        // The one `Ok(default)`: nothing stored and no file. Nothing has been
        // configured, which is a fact and not a fault.
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
            return Ok(PolicyConfig::default())
        }
        Err(err) => {
            tracing::error!(
                target: TARGET_POLICY,
                path = %path.display(),
                %err,
                "the policy document could not be read; pushes to this repository are refused"
            );
            return Err(format!(
                "the policy document at {} could not be read: {err}",
                path.display()
            ));
        }
    };
    match PolicyConfig::from_json(&text) {
        Ok(config) => Ok(config),
        Err(err) => {
            tracing::error!(
                target: TARGET_POLICY,
                path = %path.display(),
                %err,
                "the policy document did not parse; pushes to this repository are refused until \
                 it does"
            );
            Err(format!(
                "the policy document at {} did not parse: {err}",
                path.display()
            ))
        }
    }
}

/// Every key gunnar knows **that has not expired**, for the ssh-sig gate (the pre-receive gate).
///
/// `signing_keys()` and not `keys()`, and that one word is a defect fixed on
/// 2026-08-05: this function used to load every row in the table, so a key past
/// its `expires` could no longer authenticate an SSH session and could still
/// satisfy `require_signature` on a push. The expiry was enforced on exactly one
/// of the two things a key does.
pub(crate) fn key_directory(stores: &ServerStores) -> StaticKeyDirectory {
    let mut directory = StaticKeyDirectory::new();
    let keys = match stores.auth().signing_keys() {
        Ok(keys) => keys,
        Err(err) => {
            tracing::error!(
                target: TARGET_POLICY,
                %err,
                "the identity table could not be read for signature verification"
            );
            return directory;
        }
    };
    for key in keys {
        // The identity table holds two credential kinds in one hash space
        // (auth.rs, the mTLS plan): an `x509` row is the bootstrap certificate,
        // not an ssh key, and cannot verify an ssh-sig — skipped by name, so
        // the WARN below is kept for a key that IS ssh and DID go bad
        // (measured 2026-08-29: the band warned at every login).
        if !ssh_verifiable_kind(&key.algo) {
            tracing::debug!(target: TARGET_POLICY, algo = %key.algo, fingerprint = %key.fingerprint_sha256, "not an ssh key kind; not a signature verifier");
            continue;
        }
        match gunnar_core::ssh_key::PublicKey::from_openssh(&key.openssh) {
            Ok(parsed) => {
                directory.add(&key.principal_id, parsed);
            }
            Err(err) => tracing::warn!(
                target: TARGET_POLICY,
                fingerprint = %key.fingerprint_sha256,
                %err,
                "a registered key did not parse and cannot verify a signature"
            ),
        }
    }
    directory
}

/// Is a registered key's `algo` (the name as OpenSSH writes it) something
/// the ssh-sig verifier can hold? `ssh-*`, `ecdsa-*`, `sk-*` are; `x509`
/// (the bootstrap certificate stored beside them) is not.
pub fn ssh_verifiable_kind(algo: &str) -> bool {
    algo.starts_with("ssh-") || algo.starts_with("ecdsa-") || algo.starts_with("sk-")
}

// ───────────────────────────────────────────────────────────────────────────
// Guards
// ───────────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {

    /// ★ A build without the quarantine refuses an encrypted repository's push by name, and a
    /// plain repository's push is never refused for it (wiring tracer B 2026-10-08).
    #[test]
    fn a_build_without_the_quarantine_refuses_an_encrypted_push_by_name() {
        let line = super::unchecked_encrypted_push(true, false).expect("refused");
        assert!(
            line.contains("encrypted") && line.contains("this Gunnar"),
            "{line}"
        );
        assert_eq!(
            super::unchecked_encrypted_push(true, true),
            None,
            "the quarantine checks it"
        );
        assert_eq!(
            super::unchecked_encrypted_push(false, false),
            None,
            "a plain repository"
        );
    }
    use super::*;
    use gunnar_wire::pack::{IndexedPack, PackVersion};

    #[test]
    fn the_signature_directory_takes_ssh_kinds_and_skips_the_x509_row_by_name() {
        for ssh in [
            "ssh-ed25519",
            "ssh-rsa",
            "ecdsa-sha2-nistp256",
            "sk-ssh-ed25519@openssh.com",
            "sk-ecdsa-sha2-nistp256@openssh.com",
        ] {
            assert!(ssh_verifiable_kind(ssh), "{ssh}");
        }
        assert!(
            !ssh_verifiable_kind("x509"),
            "the bootstrap certificate is not tried as an ssh key"
        );
        assert!(!ssh_verifiable_kind(""));
    }

    /// A `PushOutcome` carrying whatever pack the caller wants to ask about.
    fn outcome_with(pack: Option<IndexedPack>) -> PushOutcome {
        PushOutcome {
            commands: Vec::new(),
            statuses: Vec::new(),
            unnameable: Vec::new(),
            push_options: Vec::new(),
            pack,
            pack_bytes: Some(0),
            new_objects: Ok(Vec::new()),
            atomic: false,
            unpack: Ok(()),
        }
    }

    /// `stored_bytes: None` on purpose: these arms are about what
    /// `measured_pack_bytes` does when the ENGINE reported no count, which is
    /// the only case where stat'ing a path is still the answer.
    fn indexed(data_path: Option<std::path::PathBuf>) -> IndexedPack {
        indexed_with(data_path, None)
    }

    fn indexed_with(
        data_path: Option<std::path::PathBuf>,
        stored_bytes: Option<u64>,
    ) -> IndexedPack {
        IndexedPack {
            num_objects: 1,
            pack_id: ObjectId::empty_blob(gix_hash::Kind::Sha1),
            index_id: ObjectId::empty_blob(gix_hash::Kind::Sha1),
            pack_version: PackVersion::V2,
            hash_kind: gix_hash::Kind::Sha1,
            index_path: None,
            data_path,
            keep_path: None,
            stored_bytes,
        }
    }

    /// **An engine that writes no packfile still knows how big the pack was.**
    ///
    /// The znippy store appends the received bytes verbatim and writes no
    /// `.pack`, so `data_path` is `None` — and every push into the engine the
    /// appliance ships logged `pack_bytes_unknown="pack … was indexed without a
    /// directory, so there is nothing on disk to measure"`. That sentence was
    /// FALSE: the store counted the bytes it appended
    /// (`PushReceipt::pack_bytes`) and nobody carried the number this far. An
    /// absence rendered where a measurement exists is the same defect as the
    /// reverse, and it is the one this whole family of tests is about.
    ///
    /// Both directions, or a `stored_bytes` that were ignored would still pass
    /// the first line: a store that reports NEITHER a file nor a size is still
    /// an honest unknown, and must still say so.
    #[test]
    fn a_pack_with_no_file_is_measured_by_the_store_that_stored_it() {
        let (bytes, why) = measured_pack_bytes(&outcome_with(Some(indexed_with(None, Some(4096)))));
        assert_eq!(
            bytes,
            Some(4096),
            "the store's own count is the measurement"
        );
        assert_eq!(
            why, None,
            "a measured pack carries no reason for not being measured"
        );

        let (bytes, why) = measured_pack_bytes(&outcome_with(Some(indexed_with(None, None))));
        assert_eq!(
            bytes, None,
            "neither a file nor a count is genuinely unknown"
        );
        assert!(
            why.as_deref().unwrap_or_default().contains("no size"),
            "and the reason must say the store reported no size, not merely that there is no \
             directory: {why:?}"
        );
    }

    /// **The defect this replaced.** `stat` failing rendered as `pack_bytes=0`,
    /// which is indistinguishable from a push that genuinely carried nothing.
    ///
    /// Seen red: with the body restored to
    /// `.and_then(|path| std::fs::metadata(path).ok()).map(len).unwrap_or(0)`
    /// this fails with *"a pack whose bytes could not be read reported
    /// Some(0)"* — the old code's exact answer.
    #[test]
    fn a_pack_that_cannot_be_stated_reports_no_number_and_names_the_path() {
        let missing = std::path::PathBuf::from("/nonexistent/gunnar/pack-deadbeef.pack");
        let outcome = outcome_with(Some(indexed(Some(missing.clone()))));
        let (bytes, why) = measured_pack_bytes(&outcome);
        assert_eq!(
            bytes, None,
            "a pack whose bytes could not be read reported {bytes:?}; the only honest answer is \
             no number at all, because 0 is a legal pack size for a deletions-only push"
        );
        let why = why.expect("an unreadable pack must say why it is unreadable");
        assert!(
            why.contains(missing.to_str().unwrap()),
            "the reason must name the path an operator has to go and look at, got {why:?}"
        );
    }

    /// The one arm that IS a zero, and must stay one: a deletions-only push
    /// carries no pack, and reporting that as *unknown* would be the mirror
    /// mistake.
    #[test]
    fn a_deletions_only_push_reports_a_measured_zero_and_no_reason() {
        let (bytes, why) = measured_pack_bytes(&outcome_with(None));
        assert_eq!(
            bytes,
            Some(0),
            "a push with no pack carries zero pack bytes"
        );
        assert_eq!(why, None, "zero bytes is a measurement, not an absence");
    }

    /// **Three different answers over the same empty list**, which is the whole
    /// reason the count takes a `Result` and not a slice.
    ///
    /// * not enumerated, 3-object pack → 3. Seen red on the rig's SSE feed
    ///   ("pushed 0 objects (0 B)" for a push that carried a commit).
    /// * not enumerated, no pack at all → 0, because there is nothing to count.
    /// * ENUMERATED and empty → 0 even beside a 3-object pack. That is a
    ///   measured zero: the store said which objects the push introduced and
    ///   the answer was none. Under the old `is_empty()` fallback this case
    ///   reported 3, i.e. the record disagreed with the store that had been
    ///   asked. It is the arm that makes this test unable to pass against the
    ///   code it replaces.
    #[test]
    fn the_object_count_reads_the_reason_not_the_length() {
        let mut pack = indexed(None);
        pack.num_objects = 3;
        let three = outcome_with(Some(pack));
        let none = outcome_with(None);
        let unknown = Err("the policy has no gate over objects");
        assert_eq!(attested_object_count(unknown, &three), 3);
        assert_eq!(attested_object_count(unknown, &none), 0);
        assert_eq!(
            attested_object_count(Ok(&[]), &three),
            0,
            "an enumerated empty set is a MEASURED zero and must not be overruled by the pack"
        );
        let oid = gix_hash::ObjectId::null(gix_hash::Kind::Sha1);
        assert_eq!(attested_object_count(Ok(&[oid, oid]), &none), 2);
    }

    /// A pack indexed with no directory wrote nothing, so there is nothing to
    /// measure — and that is not zero either.
    #[test]
    fn a_pack_never_written_to_disk_reports_no_number() {
        let (bytes, why) = measured_pack_bytes(&outcome_with(Some(indexed(None))));
        assert_eq!(bytes, None);
        assert!(
            why.expect("must say why").contains("without a directory"),
            "the reason must distinguish 'never written' from 'written and unreadable'"
        );
    }

    /// The number is the file's real length, not a plausible one. Seen red by
    /// writing 7 bytes and asserting 6.
    #[test]
    fn a_pack_on_disk_reports_the_bytes_the_file_actually_holds() {
        let dir = tempfile::tempdir().expect("a temp dir");
        let path = dir.path().join("pack-x.pack");
        std::fs::write(&path, b"1234567").expect("write");
        let (bytes, why) = measured_pack_bytes(&outcome_with(Some(indexed(Some(path)))));
        assert_eq!(bytes, Some(7), "the length must come off the filesystem");
        assert_eq!(why, None);
    }

    /// `wire_bytes` had the same `unwrap_or(0)`. `PushFacts::pack_bytes`
    /// documents `None` as "genuinely unavailable" and `Some(0)` as a
    /// deletions-only push; the log collapsed both onto `0`.
    #[test]
    fn an_uncounted_wire_is_not_a_wire_that_carried_nothing() {
        let (bytes, why) = measured(None, "the ingest did not report");
        assert_eq!(bytes, None, "an uncounted wire must not render as 0 bytes");
        assert_eq!(why.as_deref(), Some("the ingest did not report"));

        let (bytes, why) = measured(Some(0), "unused");
        assert_eq!(bytes, Some(0), "a counted zero stays a zero");
        assert_eq!(why, None, "a counted zero carries no reason");
    }

    /// The ssh-sig gate is handed **only unexpired** keys.
    ///
    /// The defect, found 2026-08-05: `key_directory` loaded `AuthTable::keys()`,
    /// which filters nothing, so a key past its `expires` could no longer open an
    /// SSH session and could still satisfy `require_signature` on a push. One
    /// credential with two lifetimes.
    ///
    /// `gunnar-core`'s `an_expired_key_signs_nothing_and_is_still_listed` guards
    /// the *predicate*. This guards the **wiring**, which is the half that was
    /// actually wrong: the predicate already existed and worked, and this call
    /// site simply did not ask it. A guard on the predicate alone would have been
    /// green over the whole bug.
    ///
    /// # Seen red
    ///
    /// With `signing_keys()` changed back to `keys()`, `all_keys()` has length 1
    /// and the first assertion fails. The valid-key half is the control: it fails
    /// if `key_directory` returns an empty directory for any other reason —
    /// an unreadable table, a key that will not parse — which would otherwise
    /// make the first assertion pass for the wrong reason.
    /// **Un-ignored 2026-08-10.** The reason it was skipped — *"this crate
    /// deliberately carries no storage engine"* — stopped being true when the
    /// engine became a feature of this crate with znippy as the default. There
    /// is no engine-free build of `gunnar-server` any longer to skip it for.
    #[test]
    fn the_signature_gate_is_handed_no_expired_key() {
        // The known-answer key `gunnar-core`'s own tests use, and a second one, so
        // the two principals below hold different key material.
        const EXPIRED_KEY: &str = "ssh-ed25519 \
            AAAAC3NzaC1lZDI1NTE5AAAAIMW4aqVr75w+3DWorOCbfwl5XCDCq1yCQ6puzU4qzol2 gunnar-kat";
        let minted = gunnar_core::mint_ed25519().expect("mint a key");
        let live_key = gunnar_core::ssh_key::PrivateKey::from_openssh(&minted)
            .expect("parse the minted key")
            .public_key()
            .to_openssh()
            .expect("render the public half");

        let dir = tempfile::tempdir().expect("a temp dir");
        let stores = ServerStores::open(dir.path()).expect("open the stores");
        for (id, key, expires) in [
            // Already expired: the unix epoch.
            ("stale", EXPIRED_KEY, Some(1_000)),
            ("live", live_key.as_str(), None),
        ] {
            stores
                .auth()
                .add_principal(
                    gunnar_core::Principal::new(id, id, gunnar_core::PrincipalKind::Machine)
                        .expect("a valid id"),
                )
                .expect("add the principal");
            stores
                .auth()
                .register_key(id, key, expires)
                .expect("register the key");
        }

        let directory = key_directory(&stores);
        assert!(
            directory.keys_for("stale").is_empty(),
            "an expired key must not be offered to the signature gate"
        );
        // The control. Without it an always-empty directory — an unreadable
        // table, a parse failure, a `key_directory` that returns early — passes
        // the line above while proving nothing.
        assert_eq!(
            directory.keys_for("live").len(),
            1,
            "an unexpired key must still be offered, or this test proves nothing"
        );
        assert_eq!(
            directory.all_keys().len(),
            1,
            "exactly the unexpired key, and nothing else, reaches the gate"
        );
    }
}