gunnar-wire 1.5.1

//! `git-receive-pack`, i.e. serving push. Protocol v0, which is the only
//! protocol push has ever had.
//!
//! # The order that matters
//!
//! ```text
//! advertise refs+caps → command list → push-options → PACK → index →
//! connectivity → policy gate → ref transaction → report
//! ```
//!
//! The two steps in the middle are the ones that separate a forge from a
//! directory of files:
//!
//! * **Connectivity is verified before any reference moves.** A thin pack's
//!   deltas resolve against objects the receiver already had, so "the pack
//!   indexed cleanly" does not mean "the new tip is complete". A reference
//!   pointing at an incomplete commit is a repository that every subsequent
//!   clone fails on, and nothing in the push itself reports it.
//! * **`atomic` is all-or-nothing at the storage layer**, not a loop that stops
//!   early. It is one `GitOps::put_refs_cas`, which is where the batch and the
//!   compare-and-swap are the *same* call; a per-reference loop with a break
//!   leaves the references it already applied applied.
//!
//!   That method is why `--atomic` needed a contract change rather than a
//!   rearrangement here (`locked_on_target.md`'s `GitServe` reading contract): `put_refs` is a batch
//!   **without** CAS — `RefUpdate` has no expected-old field — and
//!   `update_ref` is CAS **without** a batch, and git defines `--atomic` as
//!   both at once. Neither composes into the other, so the non-atomic path
//!   below uses `update_ref` (one reference, still compare-and-swap) and the
//!   atomic path uses `put_refs_cas`.
//!
//! # Where the packfile goes
//!
//! Not decided here. `gunnar-wire` must work with no filesystem, and
//! "keep the pushed pack" is inherently a storage decision: an appliance folds
//! it into a znippy archive, a normal-OS server drops it in `objects/pack`. So
//! the caller supplies a [`PackIngest`], and this module's only requirement is
//! the one it can check: **after `ingest` returns, the objects must be readable
//! from the store.** The connectivity walk proves it — and it needs them
//! *decoded*, which is why the handle is a `GitServe` and not a `GitOps`.

use std::io::{BufRead, Write};

use crate::pack::IndexedPack;
use gix_hash::ObjectId;
use gunnar_pktline::{Packet, BAND_PACK};
use gunnar_store::git_storage_trait::{GitServe, Observed, RefCas, RefRejection, RefTarget};

use crate::caps::{self, PeerCapabilities};
use crate::graph::{verify_connectivity, Frontier, Graph};
use crate::refs::RefSnapshot;
use crate::wire_io::{self, read_packet, read_section};
use crate::{Error, Result, Session};

/// The tracing target every line in this module carries, so a push is
/// filterable with the same `RUST_LOG=gunnar::git=info` that selects the
/// server's own transfer log.
///
/// # Why the string and not `gunnar_core::observe::TARGET_GIT`
///
/// That is where it is *named*, and this is a second spelling of it, which one
/// implementation is normally against. The edge that would remove the second
/// spelling runs the wrong way: `gunnar-core` carries the identity table, the
/// ACL and the path sandbox, and this crate is bound by the no-OS rule and the
/// crate layout to plumbing —
/// pulling it in to reach one `&str` would put all of it in the appliance's
/// crate budget.
///
/// So the two spellings are held together by a guard where both crates are
/// already linked (`gunnar-cli`'s `observability` test), which costs nothing
/// and cannot be forgotten by a build that does not link them.
pub const GIT_TARGET: &str = "gunnar::git";

/// One `<old> <new> <ref>` line from the client.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RefCommand {
    /// What the client believes the reference currently holds; null for a
    /// create.
    pub old: ObjectId,
    /// What it should hold afterwards; null for a delete.
    pub new: ObjectId,
    /// The reference.
    ///
    /// `String`, not the `FullName` newtype that died with `RefStore`, because
    /// `RefUpdate::name` and `RefRow::name` on the contract are `String` and a
    /// second validated name type in gunnar is the duplication the 2026-08-10
    /// demolition removes.
    ///
    /// **The validation did not go with the type.** It moved into
    /// `ReceivePack::parse_command`, which routes every client-supplied name
    /// through `gunnar_store::refname::validate` before one of these can
    /// exist — and that is a path-traversal boundary
    /// (`refs/heads/../../etc/passwd`), not a tidiness rule. What is gone is
    /// only the *compile-time* guarantee that a value of this type was
    /// checked; the runtime check is in one place and it is the same rule set.
    pub name: String,
}

impl RefCommand {
    /// Is this a deletion?
    pub fn is_delete(&self) -> bool {
        self.new.is_null()
    }

    /// Is this a creation?
    pub fn is_create(&self) -> bool {
        self.old.is_null()
    }
}

/// What the client sends first: its ref commands (and the ones it could not
/// name) plus the capabilities it chose. Read by [`ReceivePack::read_preamble`]
/// — with the hash check already applied — and consumed by
/// [`ReceivePack::serve_after_preamble`]. The split exists so the
/// first-repository door can read this **before** a store exists.
#[derive(Debug)]
pub struct Preamble {
    /// The nameable commands, in wire order.
    pub commands: Vec<RefCommand>,
    /// The commands whose names gunnar refused.
    pub unnameable: Vec<UnnameableRef>,
    /// What the client asked for.
    pub peer: PeerCapabilities,
}

/// A command line this server will not accept a **name** from.
///
/// # Why an invalid name is not an error
///
/// It used to be. `read_commands` called `FullName::new` with `?`, so one bad
/// name aborted the whole session before the pack had been read: the client was
/// still writing and got a broken pipe, no `report-status` was ever produced,
/// and every *other* reference in the same push was lost with no line saying
/// which one was at fault. That is the `the remote end hung up unexpectedly`
/// shape, and it is the same defect the shallow-clone fix in `read_commands`
/// closed for one particular cause without closing the class.
///
/// **Stock git defers it to the report, and this is measured, not assumed.**
/// `git receive-pack` 2.53.0, 2026-08-06, fed a raw two-command push of
/// `refs/heads/goodone` and `refs/heads/bad~name` over one connection, answered:
///
/// ```text
/// unpack ok
/// ok refs/heads/goodone
/// ng refs/heads/bad~name funny refname
/// ```
///
/// exit 0, `refs/heads/goodone` created. So the good reference lands, the bad
/// one is named in the report, and the session ends cleanly. gunnar now does
/// the same.
///
/// The name is kept as **bytes**: it is precisely a name the validator refused,
/// so there is nothing to hold it in, and it is echoed back verbatim because a
/// client matches the `ng` line against the name it sent. Since the contract's
/// `RefRow::name` is a `String`, a name that is *valid* but not UTF-8 lands
/// here too — see `ReceivePack::parse_command`, which records that narrowing
/// rather than hiding it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UnnameableRef {
    /// The reference name exactly as the client spelled it.
    pub name: Vec<u8>,
    /// The one-line reason, rendered into the `ng` line.
    pub reason: String,
    /// Where the line sat in the client's command list, so the report can be
    /// emitted in the order the client sent — which is what git does and what a
    /// client zipping its own list against the report needs.
    pub position: usize,
    /// Whether the line asked for a deletion. A push carrying only unnameable
    /// *deletes* sends no pack, and a server that waited for one would hang.
    pub deletes: bool,
}

/// What happened to one reference.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RefStatus {
    /// Applied.
    Ok,
    /// Refused, with the reason the client is told.
    Rejected(String),
}

impl RefStatus {
    /// Was it applied?
    pub fn is_ok(&self) -> bool {
        matches!(self, RefStatus::Ok)
    }
}

/// Everything `receive-pack` learned about the push, handed to every gate.
///
/// # Why this is a struct and not three arguments
///
/// The gate seam used to be `(&RefCommand, &dyn ObjectStore)` and nothing else,
/// so a gate could not see which objects the push introduced or how many bytes
/// the client sent. `gunnar-policy`'s size and quota gates
/// (`gunnar_policy::gates::limits`) are the things standing between a forge and
/// a client pushing until the disk fills, and neither could be enforced: the
/// object set was recovered downstream by reading the written `.idx` back,
/// which counts a thin pack's resolved bases as if the client had sent them,
/// and the byte count was never produced at all.
///
/// A struct rather than a wider tuple so the *next* fact does not re-break
/// every implementation of [`PushGate`].
pub struct PushFacts<'a> {
    /// The store, after the pack has been taken in.
    ///
    /// `GitServe`: a gate reads objects, and reading means *inflated and
    /// delta-resolved*. `gunnar-policy`'s ancestry walk and its signature gate
    /// were two of the three errors that crate was blocked on, and they are the
    /// same missing capability the connectivity walk here needs.
    pub objects: &'a dyn GitServe,
    /// The objects this push genuinely introduced, or the reason they could not
    /// be enumerated.
    ///
    /// **`Err` is not "none"**: a gate that cannot see what was pushed cannot
    /// decide, and an undecidable gate refuses. Handing an empty slice for
    /// "I do not know" is the shape that lets a configured limit be silently
    /// satisfied, which is what this type exists to prevent.
    pub new_objects: std::result::Result<&'a [ObjectId], &'a str>,
    /// Packfile bytes the client sent, counted off the wire by
    /// [`ReceivePack::serve`] rather than measured from what landed on disk.
    ///
    /// `Some(0)` for a deletions-only push, which carries no pack: that is a
    /// measured zero, not an absent measurement, and a byte limit must not
    /// refuse it. `None` only if the count is genuinely unavailable.
    pub pack_bytes: Option<u64>,
    /// The `push-option` lines the client sent, in wire order.
    ///
    /// # Why this field exists
    ///
    /// `push-options` was **advertised** by
    /// [`caps::receive_pack_capabilities`](crate::caps::receive_pack_capabilities),
    /// **parsed** by [`ReceivePack::serve`] and then thrown away: the only
    /// consumer in the whole tree was `.len()` in one log line. A client that
    /// sent `-o ci.skip` got the capability acknowledged on the wire and the
    /// option silently ignored, which is a wrong answer on the wire rather than
    /// a missing feature — the gate could not see them, so no policy could act
    /// on one even in principle.
    ///
    /// Stock git hands push options to the `pre-receive` hook. gunnar has no
    /// hooks and no shell; [`PushGate`] is what replaces them, so
    /// this is where they have to arrive.
    ///
    /// **Empty means the client sent none**, which is also what a client that
    /// never negotiated the capability produces. The two are not told apart
    /// here on purpose: a gate that requires an option must refuse both, and
    /// a gate that forbids unknown ones must accept both.
    pub push_options: &'a [String],
}

/// What taking the pack into storage produced.
///
/// [`PackIngest`] returns this rather than a bare [`IndexedPack`] because the
/// only place that can answer *"which objects did this push introduce"*
/// honestly is the code that resolved the pack: it is the one that knows which
/// entries came off the wire and which bases it pulled out of the store to
/// complete a thin pack. Recovering it afterwards from the written index cannot
/// tell those apart.
///
/// `Debug` and `Clone` because both of its fields already are, and because
/// [`crate::fetch_pack::FetchReport`] carries one: a report a caller cannot
/// print is a report a caller logs by hand, field by field, and gets wrong.
#[derive(Debug, Clone)]
pub struct PackReceipt {
    /// The pack as indexed.
    pub indexed: IndexedPack,
    /// The genuinely-new objects, or why they are unknown. See
    /// [`PushFacts::new_objects`] for why this is not an `Option`.
    pub new_objects: std::result::Result<Vec<ObjectId>, String>,
}

impl PackReceipt {
    /// A receipt from an ingest that does not enumerate the pushed objects.
    ///
    /// The reason, rather than an empty set, so gates refuse instead of
    /// deciding on nothing.
    pub fn without_object_set(indexed: IndexedPack) -> Self {
        PackReceipt {
            indexed,
            new_objects: Err(
                "this PackIngest does not report which objects the push introduced".to_owned(),
            ),
        }
    }
}

/// A pre-receive gate.
///
/// `gunnar-policy` supplies the real ones; this crate only defines
/// the seam so that receive-pack has somewhere to ask. **Measuring rather than
/// asserting applies hardest here**: a gate that always returns `Ok` is
/// indistinguishable from a working gate in any test that only pushes valid
/// commits, so every implementation must be seen to reject before it is
/// trusted to allow.
pub trait PushGate {
    /// `Err(reason)` rejects the reference; the reason is sent verbatim.
    fn check(&self, command: &RefCommand, facts: &PushFacts<'_>)
        -> std::result::Result<(), String>;
}

/// The gate that lets everything through. This is the default, and the thing
/// `gunnar-policy` replaces.
pub struct AllowAll;

impl PushGate for AllowAll {
    fn check(
        &self,
        _command: &RefCommand,
        _facts: &PushFacts<'_>,
    ) -> std::result::Result<(), String> {
        Ok(())
    }
}

/// How a pushed packfile is taken into storage.
///
/// The stream handed to [`PackIngest::ingest`] is positioned on the packfile's
/// first byte (the `P` of `PACK`) because [`crate::receive_pack`] reads
/// pkt-lines with exact-length reads and never buffers past the flush.
pub trait PackIngest {
    /// Consume the pack and make its objects readable from the store.
    fn ingest(&self, pack: &mut dyn BufRead) -> Result<PackReceipt>;
}

/// A bare closure is still an ingest, and it reports **no** object set.
///
/// Kept because most of this crate's own tests are closures that only need the
/// pack indexed. The receipt says so out loud
/// ([`PackReceipt::without_object_set`]) rather than answering "no new
/// objects", so a real gate behind such an ingest refuses rather than passing a
/// limit it never evaluated.
impl<F> PackIngest for F
where
    F: Fn(&mut dyn BufRead) -> Result<IndexedPack>,
{
    fn ingest(&self, pack: &mut dyn BufRead) -> Result<PackReceipt> {
        self(pack).map(PackReceipt::without_object_set)
    }
}

/// Counts the bytes an ingest consumes, so [`PushFacts::pack_bytes`] is the
/// number that came off the wire.
///
/// Wrapping the reader is what makes the count true by construction: it is not
/// `IndexedPack`'s business, it is not the storage backend's, and the only code
/// that sees every byte of the packfile go past is the code that hands the
/// stream over.
pub(crate) struct Counting<'a> {
    inner: &'a mut dyn BufRead,
    pub(crate) bytes: u64,
    /// The ceiling this push is being held to, if any. See [`PackCeiling`].
    ceiling: Option<PackCeiling>,
    /// **Bytes charged to the shared ledger that this push has not yet taken.**
    ///
    /// `fill_buf` charges what it is about to OFFER and `consume` discharges
    /// what was actually taken, and the two are not the same call: a consumer
    /// may peek at a bufferful three times and consume half of it. Without this
    /// counter each peek would charge the same bytes again and a push would
    /// spend its namespace's room several times over on bytes it never took.
    ///
    /// Charging at `consume` alone — after the bytes are already out — was the
    /// alternative, and it does not close the race: every one of N pushes would
    /// still hand its ingest a whole buffer measured against a room nobody had
    /// spent yet. The charge has to happen BEFORE the bytes are offered, which
    /// means it has to be remembered until they are taken.
    ///
    /// Left-over credit at the end of a transfer (bytes peeked and never
    /// consumed) stays charged, which is the same over-conservative,
    /// self-healing direction `gunnar_lfs::ledger` documents for an aborted
    /// upload: it is bounded by one buffer, it costs only the holders alive
    /// beside it, and it errs towards refusing rather than towards a byte the
    /// operator pays for.
    charged_ahead: u64,
    /// Set the moment the count passed the ceiling, and never cleared. The
    /// caller reads it rather than trying to recognise the abort in whatever
    /// the ingest turned the cut stream into — an indexer is entitled to
    /// report a truncated pack as a malformed one, and "your plan is full"
    /// and "your pack is corrupt" are not the same sentence to a tenant.
    pub(crate) cut: bool,
}

impl<'a> Counting<'a> {
    /// A reader over `inner` held to `ceiling` — the receive loop's own guard,
    /// lent to a FETCH so a mirror is cut at its namespace's room by the same
    /// code that cuts a push (`fetch_pack::FetchPack::with_ceiling`).
    pub(crate) fn new(inner: &'a mut dyn BufRead, ceiling: Option<PackCeiling>) -> Self {
        Counting {
            inner,
            bytes: 0,
            ceiling,
            charged_ahead: 0,
            cut: false,
        }
    }
}

impl Counting<'_> {
    /// Has the count passed the ceiling?
    ///
    /// **Strictly greater**, so a push of exactly `ceiling` bytes lands. The
    /// ceiling is the last byte that is allowed, not the first that is not —
    /// see [`PackCeiling`] on why the number already carries slack.
    fn over(&self) -> bool {
        matches!(&self.ceiling, Some(ceiling) if self.bytes > ceiling.bytes)
    }

    /// How many more bytes may be handed out before the ceiling is passed —
    /// the room under it **plus the one byte that trips it**, so the trip is
    /// reachable. `None` when there is no ceiling.
    ///
    /// # Why the slice is truncated to this, rather than trusted
    ///
    /// `fill_buf` hands out whatever the source has buffered, and the source
    /// decides how much that is: a `BufReader` gives a bufferful, a `Cursor`
    /// over a `Vec` gives **the whole rest of the push in one call**. Without
    /// this clamp the ceiling's precision would be the source's buffer size —
    /// the guard would still fire, but a pack read out of memory would cross
    /// it by its entire length before anything looked, and the guard's own
    /// test would be measuring the fixture rather than the code (measure,
    /// never assert). With it, a push that overshoots costs the server exactly
    /// `bytes + 1` and the number is the same on every reader.
    fn room(&self) -> Option<usize> {
        let ceiling = self.ceiling.as_ref()?.bytes;
        let room = ceiling.saturating_sub(self.bytes).saturating_add(1);
        // A ceiling past `usize` on a 32-bit box is not a ceiling this read can
        // reach in one call, so clamping to `usize::MAX` is the honest answer
        // rather than a truncating cast.
        Some(usize::try_from(room).unwrap_or(usize::MAX))
    }

    /// **How many of `want` bytes this push may hand on, asking the shared
    /// ledger for whatever it has not already been credited.**
    ///
    /// `want` is what the caller would otherwise offer, already clamped to
    /// [`Self::room`]. With no ceiling, or a ceiling with no ledger behind it,
    /// the answer is `want` and nothing is charged — the community build and
    /// every gunnar before 2026-09-08 take this path, and it is the same code
    /// path, not a second one.
    ///
    /// The room the ledger is asked about is `ceiling + 1`: the ceiling plus
    /// the one byte that trips it. A budget of exactly `ceiling` would starve
    /// the trip byte, and a push of exactly its ceiling would then end as a
    /// stream that simply stopped — which the indexer reports as a malformed
    /// pack, and telling a tenant their pack is corrupt when their plan filled
    /// up is the refusal this file's `refuse` exists to prevent.
    fn admit(&mut self, want: usize) -> usize {
        let Some(ledger) = self.ceiling.as_ref().and_then(|c| c.ledger.as_ref()) else {
            return want;
        };
        // Unwrap-free: the `let else` above already proved there is a ceiling.
        let room = self
            .ceiling
            .as_ref()
            .map(|c| c.bytes.saturating_add(1))
            .unwrap_or(u64::MAX);
        let want = want as u64;
        // Only the part that is not already paid for. On a re-peek of a buffer
        // nothing was consumed from, this is 0 and the ledger is not touched.
        let owing = want.saturating_sub(self.charged_ahead);
        if owing > 0 {
            self.charged_ahead = self
                .charged_ahead
                .saturating_add(ledger.admit_upto(owing, room));
        }
        // `min`: credit carried over from an earlier, larger offer does not
        // make this offer bigger than it is.
        usize::try_from(self.charged_ahead.min(want)).unwrap_or(usize::MAX)
    }

    /// Cut the stream: mark the push and return the error the ingest sees.
    ///
    /// The text here is for the **server log**, not the client: what the client
    /// is told is [`PackCeiling::refusal`], installed over the ingest's own
    /// message by [`ReceivePack::serve_after_preamble`]. Two sentences because
    /// they answer two people.
    fn refuse(&mut self) -> std::io::Error {
        self.cut = true;
        let ceiling = self.ceiling.as_ref().map(|c| c.bytes).unwrap_or_default();
        std::io::Error::other(format!(
            "the push passed its {ceiling}-byte ceiling after {} byte(s); the pack read was \
             aborted mid-transfer and nothing was stored",
            self.bytes
        ))
    }
}

impl std::io::Read for Counting<'_> {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        if self.cut || self.over() {
            let err = self.refuse();
            return Err(err);
        }
        // **What the source actually has, before anything is charged for it —
        // and asked for ONLY when something is going to be charged.**
        //
        // The ledger has to be asked about bytes that EXIST: charging
        // `buf.len()` and reading fewer would leave the difference charged to
        // the namespace with nothing to discharge it, and
        // `simultaneous_pushes_that_all_fit_are_all_admitted` measured that as
        // seven of eight legitimate pushes admitted where all eight fit. A
        // quota that refuses a push nobody's bytes displaced is the same defect
        // as one that admits too many, pointing the other way.
        //
        // `fill_buf` blocks exactly where `read` would — until one byte is
        // available or the stream ends — so it costs no latency. It does cap
        // this call at what the source has buffered, which is why it is skipped
        // entirely with no ledger to ask: an unmetered `read` still goes
        // straight through to the source and may still hand back more than one
        // bufferful in a call, exactly as it did before this field existed.
        let available = match self.ceiling.as_ref().and_then(|c| c.ledger.as_ref()) {
            Some(_) => {
                let available = self.inner.fill_buf()?.len();
                if available == 0 {
                    // The client is done. Not a refusal: nothing to refuse.
                    return Ok(0);
                }
                available
            }
            None => usize::MAX,
        };
        let want = match self.room() {
            Some(room) => room.min(buf.len()).min(available),
            None => buf.len().min(available),
        };
        // The namespace's answer, not just this push's: `admit` returns `want`
        // unchanged when there is no ledger, so the unmetered path is
        // arithmetically untouched.
        let allowed = self.admit(want);
        if allowed == 0 && !buf.is_empty() {
            // Nothing left in the namespace, and the client is still sending.
            // Cut it here rather than returning `Ok(0)`: a zero-length read is
            // EOF to every consumer in std, and "your plan is full" must not
            // reach the indexer as "the pack ended early".
            let err = self.refuse();
            return Err(err);
        }
        let read = self.inner.read(&mut buf[..allowed])?;
        self.bytes += read as u64;
        self.charged_ahead = self.charged_ahead.saturating_sub(read as u64);
        Ok(read)
    }
}

impl BufRead for Counting<'_> {
    /// **Where the ceiling actually bites**, because this is how the pack is
    /// read: `read_pushed_pack_spooling` peeks with `fill_buf` and takes with
    /// `consume`, so this is the path the shipping ingest takes and `read`
    /// above is the path one that does not peek takes. Both are guarded and
    /// both clamp to [`Counting::room`], so the overshoot is **one byte**, not
    /// one buffer — a push that means to send 400 GiB past its ceiling costs
    /// the server `ceiling + 1` and nothing more.
    fn fill_buf(&mut self) -> std::io::Result<&[u8]> {
        if self.cut || self.over() {
            let err = self.refuse();
            return Err(err);
        }
        // `room()` is never `Some(0)` — `over()` above has already returned for
        // that — so this never turns a live stream into a spurious EOF.
        let room = self.room();
        // **The namespace's half, and it fills the buffer TWICE.** The ledger
        // has to be asked about the bytes actually in hand — asking it about
        // `room` instead would reserve a tenant's whole remaining plan for the
        // first connection to say hello and refuse a sibling push that would
        // have fitted twice over — and it cannot be asked while a slice
        // borrowed out of `self.inner` is alive. A second `fill_buf` with no
        // `consume` between is a peek: it returns the same buffer it just
        // returned, reads nothing more and cannot block. The `min` below makes
        // even a source that ignores that safe.
        //
        // Skipped entirely with no ledger to ask, so the unmetered path fills
        // once and clamps to `room`, exactly as it did before this existed.
        let allowed = match self.ceiling.as_ref().and_then(|c| c.ledger.as_ref()) {
            Some(_) => {
                let available = self.inner.fill_buf()?.len();
                let want = match room {
                    Some(room) if available > room => room,
                    _ => available,
                };
                let allowed = self.admit(want);
                if allowed == 0 && available > 0 {
                    // The namespace's room is gone and the client is still
                    // sending. An empty slice would be EOF to the indexer, i.e.
                    // "the pack is truncated"; the cut says whose bytes filled
                    // the plan.
                    let err = self.refuse();
                    return Err(err);
                }
                Some(allowed)
            }
            None => None,
        };
        let filled = self.inner.fill_buf()?;
        Ok(match (allowed, room) {
            (Some(allowed), _) => &filled[..allowed.min(filled.len())],
            (None, Some(room)) if filled.len() > room => &filled[..room],
            _ => filled,
        })
    }

    /// The only place `fill_buf`'s bytes are charged. `read` charges its own,
    /// and the two never see the same byte: `BufRead::read`'s default goes
    /// through `fill_buf`/`consume`, but this type overrides `read` to delegate
    /// straight to the inner reader instead, so a byte is counted once whichever
    /// way the consumer asks for it.
    fn consume(&mut self, amount: usize) {
        self.inner.consume(amount);
        self.bytes += amount as u64;
        // Taken, so no longer owed: the credit `fill_buf` bought for these
        // bytes is spent. What is left is credit for bytes still in the buffer,
        // which the next `fill_buf` will offer again and must not pay for
        // twice.
        self.charged_ahead = self.charged_ahead.saturating_sub(amount as u64);
    }
}

/// Everything one push did.
#[derive(Debug)]
pub struct PushOutcome {
    /// The commands as sent.
    pub commands: Vec<RefCommand>,
    /// One status per command, in the same order.
    pub statuses: Vec<RefStatus>,
    /// Command lines refused for their reference **name**, which therefore have
    /// no [`RefCommand`] to carry a status. See [`UnnameableRef`].
    pub unnameable: Vec<UnnameableRef>,
    /// `push-option` lines, if the capability was negotiated.
    pub push_options: Vec<String>,
    /// The indexed pack, or `None` when the push was deletions only.
    pub pack: Option<IndexedPack>,
    /// Packfile bytes the client sent, counted off the wire. See
    /// [`PushFacts::pack_bytes`].
    pub pack_bytes: Option<u64>,
    /// The objects the push introduced, or why they are unknown. Same value the
    /// gates saw; see [`PushFacts::new_objects`].
    pub new_objects: std::result::Result<Vec<ObjectId>, String>,
    /// Whether the client asked for `--atomic`.
    pub atomic: bool,
    /// What the client was told about unpacking: `Ok` or a one-line reason.
    pub unpack: std::result::Result<(), String>,
}

impl PushOutcome {
    /// Did every reference land?
    ///
    /// An unnameable reference counts as a refusal: it is reported as `ng`, so
    /// a caller reading this to mean *the push succeeded* must see false.
    pub fn all_ok(&self) -> bool {
        self.unpack.is_ok()
            && self.unnameable.is_empty()
            && self.statuses.iter().all(RefStatus::is_ok)
    }
}

/// **How many packfile bytes this push may send, and what the client is told
/// when it sends more.**
///
/// # Why a ceiling exists at all
///
/// `git-receive-pack`'s pack header carries an **object count, not a byte
/// count**, so the size of a push is not knowable at the moment it is
/// authorised. A quota asked before the pack (`gunnar_ssh::PlanGate::pack_quota`)
/// refuses a tenant who is *already* at the ceiling, and a limit asked after it
/// (`gunnar_policy::gates::limits`, whose own header says so plainly) refuses a
/// pack that has *already been received* — so between those two there was
/// nothing at all in front of one enormous push. A tenant one byte under a
/// 10 GiB cap could send 400 GiB and the box would take every byte of it before
/// anything looked.
///
/// This is that missing thing. It is enforced in the receive loop itself
/// ([`Counting`]), so an unbounded push costs the server the ceiling and not
/// the push.
///
/// # It is a backstop, not a meter
///
/// The polite refusal is `pack_quota`'s, before a byte moves. This fires only
/// for a push that was under the cap when it started and crossed it in flight,
/// which is why whoever sets it is expected to have added **slack** to it: the
/// bytes on the wire are not the bytes that land (a thin pack is completed with
/// bases the client never sent; the engine stores compressed), and a ceiling
/// with no room refuses a push that would have fitted. `gunnar-wire` does not
/// choose that number — it does not know what a plan is — it enforces the one
/// it is handed.
/// # …and it is one number shared by everything pushing at once
///
/// [`PackCeiling::bytes`] is measured for ONE push, and two pushes that arrive
/// together get the same measurement — so before [`PackCeiling::ledger`] each
/// of N simultaneous pushes was held to the whole of the namespace's remaining
/// room, and the room was spent N times. `tests/concurrent_pack_ceiling.rs`
/// measured it at eight packs of 41016 bytes into 65536 bytes of room. The
/// ledger is what makes the number a shared budget rather than a per-connection
/// allowance.
#[derive(Debug, Clone)]
pub struct PackCeiling {
    /// Packfile bytes past which the read is aborted. A push of exactly this
    /// many bytes is **accepted**; the next byte cuts it.
    pub bytes: u64,
    /// The line the client is told, verbatim, in place of whatever the ingest
    /// made of the truncated stream. Composed by whoever set the ceiling — the
    /// metered appliance names the plan, the cap and the billing URL, the same
    /// shape every other entitlement refusal uses — because this crate knows
    /// nothing about plans, caps or money.
    pub refusal: String,
    /// **Who else is spending [`Self::bytes`] right now.** `None` — the
    /// default, and what a build with no byte cap installs — means the number
    /// is this push's alone, which is what every gunnar before 2026-09-08 did
    /// and is correct exactly as long as one push is the only push.
    ///
    /// `Some` makes the ceiling a budget the namespace shares: the receive loop
    /// asks before every chunk it hands to the ingest, and takes what it is
    /// granted. See [`PackLedger`].
    ///
    /// `PartialEq`/`Eq` came off this struct when the field arrived. Two
    /// ceilings are not comparable any more, because a shared standing in a
    /// counter is not a value — and nothing compared them: the equality the
    /// tests use is `gunnar_ssh::PackCeiling`'s, which is the plain twin the
    /// gates return and still derives both.
    pub ledger: Option<std::sync::Arc<dyn PackLedger>>,
}

/// **The namespace's shared standing, asked one chunk at a time.**
///
/// A packfile's header carries an object count and no byte count, so a push
/// cannot declare its size and the door cannot admit it up front the way the
/// LFS door admits an object. The receive loop therefore asks about the chunk
/// in its hand, before it hands it on, and takes what it is granted.
///
/// # Why a trait here instead of the ledger itself
///
/// The implementation is `gunnar_lfs::Mark` (a monotone `u64` per namespace and
/// a per-holder mark — see that module for the whole design, including why
/// there is deliberately nothing to give back), wired up by
/// `gunnar_server::transport`, which is the one place that already holds both
/// halves of the [`PackCeiling`] twin. This crate must not link `gunnar-lfs`
/// and must not link `gunnar-core`: it is bound by the no-OS rule and the crate
/// layout to plumbing, and
/// its own [`GIT_TARGET`] refuses `gunnar-core` for a single `&str` on exactly
/// that argument. So the edge is inverted rather than added, and `gunnar-wire`
/// resolves the same 51 packages it resolved before the ledger existed
/// (MEASURED 2026-09-08, `cargo tree -e normal`).
pub trait PackLedger: Send + Sync + std::fmt::Debug {
    /// **Charge up to `want` bytes against everything admitted to this
    /// namespace since this push measured it, inside a total of `room`, and
    /// return how many were charged.**
    ///
    /// Partial credit, not a verdict: a grant short of `want` is a chunk the
    /// caller may take that much of, and a grant of `0` means the room is gone
    /// and the push must be cut. An implementation that answered all-or-nothing
    /// would refuse a push that fits, for asking about a buffer bigger than the
    /// remainder.
    ///
    /// `room` is passed in on every call rather than held by the
    /// implementation because it is the RECEIVE LOOP's number: it is
    /// [`PackCeiling::bytes`] plus the one byte that makes the ceiling
    /// reachable ([`Counting::room`] explains why that byte exists), and
    /// spelling that `+ 1` in two crates is how two guards start disagreeing.
    ///
    /// **Must not block on I/O.** It is called once per chunk of a transfer at
    /// wire speed; the shipping implementation takes a `Mutex` across a hash
    /// lookup and an integer add, and nothing else.
    fn admit_upto(&self, want: u64, room: u64) -> u64;
}

/// Knobs for one push.
#[derive(Debug, Clone)]
pub struct Config {
    /// The `agent=` string.
    pub agent: String,
    /// Traversal budget for the connectivity check.
    pub budget: Option<usize>,
    /// **A line the client is told beside the report**, on the progress band
    /// (`remote: …`), for a push that is ALLOWED but carries a warning — the
    /// metered appliance's `plan expired <date>; renew at <url>` for a tenant
    /// in Grace. `None` prints nothing. Sent only when the client negotiated a
    /// sideband, which every git since 1.7 does; a client without one gets the
    /// bare report, never a line where the report should be.
    pub notice: Option<String>,
    /// **The mid-upload ceiling.** `None` — the default, and what a community
    /// build with no byte cap installs — reads the pack to its end whatever
    /// its size, which is what every gunnar before 2026-09-05 did. See
    /// [`PackCeiling`].
    pub pack_ceiling: Option<PackCeiling>,
}

impl Default for Config {
    fn default() -> Self {
        Config {
            agent: caps::default_agent(),
            budget: Some(crate::graph::CONNECTIVITY_BUDGET),
            notice: None,
            pack_ceiling: None,
        }
    }
}

/// The server side of `git-receive-pack`.
pub struct ReceivePack<'a> {
    /// **One store, not two.** `ObjectStore` and `RefStore` were separate
    /// gunnar traits and are deleted; `GitOps` says both, and `GitServe`
    /// extends it with the decoded read the connectivity walk needs.
    git: &'a dyn GitServe,
    /// The repository's hash kind.
    ///
    /// Carried in rather than asked for — `GitServe` has no `hash_kind()` and
    /// the frozen signature leaves whether `GitOps` should grow one open. It is not a
    /// convenience here: `parse_command` refuses a command line whose oids are
    /// in the wrong hash, and answering that from a default would accept a
    /// sha1 push into a sha256 repository.
    hash_kind: gix_hash::Kind,
    config: Config,
}

/// An [`Observed`] in git's own vocabulary for a ref-status line.
///
/// git says `is at <oid> but expected <oid>`, and for a ref that did not exist
/// it says the all-zero oid. `Unreadable` has no git spelling at all — the
/// backend could not decode what the ref held — so it is named rather than
/// flattened into "absent", which would tell a client the ref was empty when it
/// was in fact corrupt.
fn render(o: &Observed) -> String {
    match o {
        Observed::Nothing => "0000000000000000000000000000000000000000".to_owned(),
        // A symbolic ref is spelled `ref: <name>`, as git does, so a client is
        // never handed a branch name where it expects an oid.
        Observed::Value(RefTarget::Object(oid)) => {
            // Written out rather than pulling in a `hex` dependency for one
            // call: this crate has none and a ref-status line is not worth one.
            oid.iter()
                .fold(String::with_capacity(oid.len() * 2), |mut acc, b| {
                    use std::fmt::Write as _;
                    let _ = write!(acc, "{b:02x}");
                    acc
                })
        }
        Observed::Value(RefTarget::Symbolic(name)) => format!("ref: {name}"),
        Observed::Unreadable(what) => format!("unreadable ({what})"),
    }
}

impl<'a> ReceivePack<'a> {
    /// Receive into `git` with default configuration.
    pub fn new(git: &'a dyn GitServe, hash_kind: gix_hash::Kind) -> Self {
        ReceivePack {
            git,
            hash_kind,
            config: Config::default(),
        }
    }

    /// Replace the configuration.
    pub fn with_config(mut self, config: Config) -> Self {
        self.config = config;
        self
    }

    /// The repository's hash kind.
    pub fn hash_kind(&self) -> gix_hash::Kind {
        self.hash_kind
    }

    /// Write the reference advertisement and its capabilities.
    pub fn write_ref_advertisement(&self, out: &mut dyn Write) -> Result<()> {
        let capabilities = caps::receive_pack_capabilities(self.hash_kind(), &self.config.agent);
        // ONE snapshot: `RefStore::iter` followed by a `resolve` per symbolic
        // row was N+1 reads of a namespace that can move between them, so an
        // advertisement could name an oid for a reference that had already
        // been deleted. A client then pushes against what it was told.
        let snapshot = RefSnapshot::read(self.git)?;
        let mut refs = Vec::new();
        for row in snapshot.rows() {
            let Some(id) = snapshot.resolve_row(row)? else {
                continue;
            };
            refs.push(gunnar_sendpack::RemoteRef::new(row.name.as_bytes(), id));
        }
        // Rendered by the shared grammar, including the `capabilities^{}`
        // pseudo-ref an empty repository has to emit so a client can still
        // learn the hash algorithm. `gunnar-client` parses these lines with the
        // same module.
        for line in gunnar_sendpack::advertisement::lines(&refs, &capabilities, self.hash_kind()) {
            wire_io::text(out, &line)?;
        }
        wire_io::flush(out)?;
        Ok(())
    }

    /// Serve one push, from the advertisement through the status report.
    pub fn serve(
        &self,
        input: &mut dyn BufRead,
        out: &mut dyn Write,
        ingest: &dyn PackIngest,
        gate: &dyn PushGate,
    ) -> Result<Option<PushOutcome>> {
        self.serve_session(Session::Connected, input, out, ingest, gate)
    }

    /// Serve one push for either [`Session`].
    ///
    /// **Push is the easy half of stateless-rpc, and it is worth saying why.**
    /// `receive-pack` has no negotiation: the client is told the refs once, and
    /// then sends its command list, its options and its packfile in one go. So
    /// a smart-HTTP push is exactly one `POST` and the only difference from SSH
    /// is that the advertisement was a separate `GET`. There is no round to cut
    /// short here, and protocol v2 has no push side to complicate it.
    pub fn serve_session(
        &self,
        session: Session,
        input: &mut dyn BufRead,
        out: &mut dyn Write,
        ingest: &dyn PackIngest,
        gate: &dyn PushGate,
    ) -> Result<Option<PushOutcome>> {
        if session.advertises_inline() {
            let _s = crate::stage::enter("advertise");
            self.write_ref_advertisement(out)?;
            out.flush()?;
        }
        let Some(preamble) = Self::read_preamble(self.hash_kind(), input)? else {
            return Ok(None);
        };
        self.serve_after_preamble(preamble, input, out, ingest, gate)
            .map(Some)
    }

    /// The advertisement of a repository that **does not exist yet**: no
    /// references, the capabilities and `object-format` of `hash_kind`. The
    /// first-repository door writes this instead of [`Self::write_ref_advertisement`]
    /// because there is no store to snapshot — and must not be one, until the
    /// client has sent a command that passes the hash check.
    pub fn write_empty_advertisement(
        hash_kind: gix_hash::Kind,
        config: &Config,
        out: &mut dyn Write,
    ) -> Result<()> {
        let capabilities = caps::receive_pack_capabilities(hash_kind, &config.agent);
        for line in gunnar_sendpack::advertisement::lines(&[], &capabilities, hash_kind) {
            wire_io::text(out, &line)?;
        }
        wire_io::flush(out)?;
        out.flush()?;
        Ok(())
    }

    /// The client's half of the opening: its command list and capabilities,
    /// with the hash check applied. `None` when the client took the
    /// advertisement and sent nothing, which is `git push` with nothing to do
    /// — and a sha1 working tree that read `object-format=sha256` and gave up
    ///. Neither is an error and neither is a push.
    ///
    /// An associated function, not a method: the first-repository door calls
    /// it **before** any store exists, so it must not need one.
    pub fn read_preamble(
        hash_kind: gix_hash::Kind,
        input: &mut dyn BufRead,
    ) -> Result<Option<Preamble>> {
        let (commands, unnameable, peer) = {
            let _s = crate::stage::enter("command_read");
            Self::read_commands(hash_kind, input)?
        };
        if commands.is_empty() && unnameable.is_empty() {
            tracing::info!(
                target: GIT_TARGET,
                event = "git.receive_pack.no_commands",
                "receive-pack: the client sent no ref commands"
            );
            // The client took the advertisement and hung up; `git push` with
            // nothing to do does exactly this. Not an error, and not a push.
            return Ok(None);
        }
        caps::check_object_format(hash_kind, peer.value("object-format"))?;
        Ok(Some(Preamble {
            commands,
            unnameable,
            peer,
        }))
    }

    /// Serve a push whose [`Preamble`] has been read: options, the pack, the
    /// gates, the ref updates and the status report.
    pub fn serve_after_preamble(
        &self,
        preamble: Preamble,
        input: &mut dyn BufRead,
        out: &mut dyn Write,
        ingest: &dyn PackIngest,
        gate: &dyn PushGate,
    ) -> Result<PushOutcome> {
        let Preamble {
            commands,
            unnameable,
            peer,
        } = preamble;
        let _s_opts = crate::stage::enter("push_options");
        let push_options = if peer.has("push-options") {
            // The push side has its own ceiling for the same reason the fetch
            // side does: the section ends when the *peer* sends a flush-pkt.
            // The number is `gunnar-wire`'s one request ceiling, not a second
            // constant that could drift from it (one implementation).
            let mut budget =
                crate::limits::Budget::new(Some(crate::limits::DEFAULT_REQUEST_LIMIT_BYTES));
            read_section(input, "the push-options section", &mut budget)?
                .into_iter()
                .map(|line| String::from_utf8_lossy(&line).into_owned())
                .collect()
        } else {
            Vec::new()
        };

        crate::stage::end(_s_opts);
        let sideband = peer.has("side-band-64k");
        let report_requested = peer.has("report-status") || peer.has("report-status-v2");
        let atomic = peer.has("atomic");

        // Deletions carry no objects, so the client sends no pack at all. A
        // server that reads one anyway blocks until the connection times out.
        //
        // An unnameable reference counts here too. The client does not know
        // gunnar refused its name and streams the pack regardless, so a server
        // that skipped the read because every *nameable* command was a delete
        // would leave a pack in the pipe, write its report into the middle of
        // it, and desync the session.
        let expects_pack = commands.iter().any(|command| !command.is_delete())
            || unnameable.iter().any(|refused| !refused.deletes);

        // ── the phase log ───────────────────────────────────────────────────
        //
        // Everything below this point can take minutes on a repository the
        // size of a kernel, and until now it said **nothing at all** until it
        // was over. The Fet Kaos Monkey rig's `linux.git` import (13 816 503
        // objects, 8 170 MB pack, 2 872 refs; 2026-08-05, load 25–33)
        // wrote its pack and its index in ~30 min and had still logged no line
        // at 40 min. That is an operability defect on its own terms and
        // separate from the walk being slow: an operator watching a working
        // import and one watching a hang were reading the identical empty log.
        //
        // So each phase announces itself on **entry** and reports on **exit**.
        // The entry line is the load-bearing one — an exit-only log still
        // leaves the ten minutes in between unattributed, which is precisely
        // the state that was unreadable.
        //
        // `info`, not `debug`: a phase that can run for ten minutes is not
        // diagnostic detail, and a default-level log an operator has to know to
        // turn on is one that is off during the incident.
        tracing::info!(
            target: GIT_TARGET,
            event = "git.receive_pack.commands",
            commands = commands.len(),
            deletes = commands.iter().filter(|c| c.is_delete()).count(),
            push_options = push_options.len(),
            atomic,
            expects_pack,
            "receive-pack: read the ref command list"
        );
        let mut pack = None;
        // `Some(0)`, not `None`: a deletions-only push carries no pack, and that
        // is a MEASURED zero. A byte limit must refuse an oversized pack, not a
        // `git push --delete`.
        let mut pack_bytes = Some(0u64);
        let mut new_objects: std::result::Result<Vec<ObjectId>, String> = Ok(Vec::new());
        let mut unpack: std::result::Result<(), String> = Ok(());
        if expects_pack {
            // Indexing and verifying what the client sent. This is the stage
            // the receipt's `num_objects` is about, and on a push it is where
            // the money is.
            let _s = crate::stage::enter("pack_ingest");
            tracing::info!(
                target: GIT_TARGET,
                event = "git.receive_pack.receiving",
                "receiving and indexing the pushed pack"
            );
            let started = std::time::Instant::now();
            // Counted HERE rather than by measuring the pack that landed on
            // disk. What a limit is defending against is what the client sent;
            // the file that lands is a different number, because completing a
            // thin pack adds bases the client did not send, and on a store with
            // no filesystem there is no file to measure at all.
            let mut counting = Counting {
                inner: input,
                bytes: 0,
                ceiling: self.config.pack_ceiling.clone(),
                charged_ahead: 0,
                cut: false,
            };
            let received = ingest.ingest(&mut counting);
            pack_bytes = Some(counting.bytes);
            let cut = counting.cut;
            match received {
                Ok(receipt) => {
                    tracing::info!(
                        target: GIT_TARGET,
                        event = "git.receive_pack.indexed",
                        objects = receipt.indexed.num_objects,
                        pack = %receipt.indexed.pack_id,
                        pack_bytes = counting.bytes,
                        elapsed_ms = started.elapsed().as_millis() as u64,
                        "indexed a pushed pack"
                    );
                    new_objects = receipt.new_objects;
                    pack = Some(receipt.indexed);
                }
                Err(err) => {
                    let message = err.client_message();
                    tracing::warn!(
                        target: GIT_TARGET,
                        event = "git.receive_pack.index_failed",
                        pack_bytes = counting.bytes,
                        elapsed_ms = started.elapsed().as_millis() as u64,
                        // **The unredacted error, in the server's own log.**
                        //
                        // This logged `client_message()` — the redacted
                        // sentence the *client* is told — so a push that failed
                        // to index left `the repository is unavailable; try
                        // again` in the log and nothing anywhere said what the
                        // store had actually refused. `Error::client_message`'s
                        // own doc promises the opposite: "the unredacted text is
                        // still what the server logs". The fetch side kept that
                        // promise (`git.upload_pack.failed` carries `%err`);
                        // this side did not, and an operator debugging a refused
                        // push had the client's sentence twice and the cause
                        // never.
                        error = %err,
                        // Kept beside it, so a log reader can match what the
                        // operator was shown to what actually happened.
                        client_message = %message,
                        "the pushed pack did not index"
                    );
                    unpack = Err(message);
                }
            }
            // ── the ceiling, installed over whatever the ingest concluded ──
            //
            // Checked AFTER the match and unconditionally, not inside the `Err`
            // arm, because an ingest that swallowed the cut and answered `Ok`
            // would otherwise hand a partially-read push to `apply` — and a
            // guard that only fires when the code under it already failed is
            // the guard a measured assertion forbids.
            //
            // What the client is told is the ceiling's own line
            // (`PackCeiling::refusal`) and not `Error::client_message`'s: the
            // indexer's honest answer to a truncated stream is "the pack is
            // malformed", and telling a paying tenant their pack was corrupt
            // when what happened is that their plan filled up is the refusal an
            // operator gets a ticket about.
            if cut {
                let line = self
                    .config
                    .pack_ceiling
                    .as_ref()
                    .map(|ceiling| ceiling.refusal.clone())
                    // Unreachable — `Counting::cut` is set only where a ceiling
                    // exists — and still spelled out rather than
                    // `unwrap_or_default()`, because an empty `unpack` line is
                    // a report that says a push failed and will not say why.
                    .unwrap_or_else(|| {
                        "the push was cut at its byte ceiling".to_owned()
                    });
                tracing::warn!(
                    target: GIT_TARGET,
                    event = "git.receive_pack.ceiling",
                    pack_bytes = counting.bytes,
                    ceiling = self.config.pack_ceiling.as_ref().map(|c| c.bytes),
                    refusal = %line,
                    "the pushed pack passed its byte ceiling and the transfer was aborted \
                     mid-stream; nothing was indexed and no reference was moved"
                );
                // The pack is not `Some`: the ingest never completed one, and a
                // receipt here would make `PushOutcome` claim a pack that does
                // not exist.
                pack = None;
                new_objects = Err(
                    "the push was cut at its byte ceiling before the pack was indexed"
                        .to_owned(),
                );
                unpack = Err(line);
            }
        }

        let statuses = if unpack.is_err() {
            vec![RefStatus::Rejected("unpacker error".to_owned()); commands.len()]
        } else {
            let facts = PushFacts {
                objects: self.git,
                new_objects: new_objects.as_deref().map_err(String::as_str),
                pack_bytes,
                push_options: &push_options,
            };
            // `--atomic` is all-or-nothing across the batch, and an unnameable
            // reference is a refusal *in* the batch. `apply` cannot see one —
            // it is only ever handed the commands that have a name — so the
            // fact is passed in, and it must be passed **in** rather than
            // collapsed afterwards: an after-the-fact collapse would rewrite
            // statuses for references the transaction had already applied, and
            // report as refused what is on disk.
            self.apply(&commands, atomic, !unnameable.is_empty(), gate, &facts)?
        };

        let outcome = PushOutcome {
            commands,
            statuses,
            unnameable,
            push_options,
            pack,
            pack_bytes,
            new_objects,
            atomic,
            unpack,
        };

        if report_requested {
            let _s = crate::stage::enter("report");
            self.write_report(out, &outcome, sideband)?;
            out.flush()?;
        }
        Ok(outcome)
    }

    fn read_commands(
        hash_kind: gix_hash::Kind,
        input: &mut dyn BufRead,
    ) -> Result<(Vec<RefCommand>, Vec<UnnameableRef>, PeerCapabilities)> {
        let mut commands = Vec::new();
        let mut unnameable = Vec::new();
        let mut peer = PeerCapabilities::default();
        let mut first = true;
        let mut position = 0usize;
        loop {
            let packet = match read_packet(input)? {
                None => {
                    if first {
                        return Ok((commands, unnameable, peer));
                    }
                    return Err(Error::UnexpectedEof("the push command list"));
                }
                Some(packet) => packet,
            };
            match packet {
                Packet::Flush => break,
                Packet::Data(data) => {
                    let mut text: &[u8] = data.strip_suffix(b"\n").unwrap_or(&data);
                    // ── a shallow announcement is not the first COMMAND ──────
                    //
                    // A shallow client sends `shallow <oid>` for each graft
                    // before its commands. They are not commands, which this
                    // already knew — and they are not the line the capabilities
                    // ride on, which it did not.
                    //
                    // `first` used to mean "the first packet", so a shallow
                    // pusher spent it on a `shallow` line. That line carries no
                    // NUL, so no capabilities were parsed; `first` went false;
                    // and the real first command was then handed to
                    // `parse_command` with its whole capability section still
                    // attached:
                    //
                    // ```text
                    // invalid reference name "refs/heads/x\0 report-status-v2 side-band-64k
                    //   quiet object-format=sha1 agent=git/2.53.0-Linux": control character or space
                    // ```
                    //
                    // So EVERY push from a shallow clone was refused, and the
                    // client saw only `the remote end hung up unexpectedly`.
                    // That is not an exotic client: `actions/checkout` clones at
                    // `fetch-depth: 1` by default, so a CI job that pushes back
                    // is the shape that hits it.
                    //
                    // ISOLATED 2026-08-05 against stock git 2.53.0 over
                    // real SSH, one client, one commit, one server:
                    //
                    // | client                         | push        |
                    // |--------------------------------|-------------|
                    // | `clone --depth 3`              | **refused** |
                    // | the same clone, `fetch --unshallow` first | accepted |
                    //
                    // The second failure hid behind the first: `peer` stayed
                    // empty, so `report-status`, `side-band-64k`, `atomic`,
                    // `push-options` and the `object-format` check were all
                    // silently off for a shallow pusher even where the ref name
                    // happened to validate.
                    //
                    // git does it this way round too — `receive-pack.c`'s
                    // `read_head_info` takes the capabilities off the first
                    // **ref command**, having consumed any `shallow` lines
                    // ahead of it.
                    if text.starts_with(b"shallow ") {
                        continue;
                    }
                    if first {
                        if let Some(nul) = text.iter().position(|&b| b == 0) {
                            peer =
                                PeerCapabilities::parse(&String::from_utf8_lossy(&text[nul + 1..]));
                            text = &text[..nul];
                        }
                        first = false;
                    }
                    match Self::parse_command(hash_kind, text, position)? {
                        Ok(command) => commands.push(command),
                        Err(refused) => unnameable.push(refused),
                    }
                    position += 1;
                }
                other => {
                    return Err(Error::protocol(format!(
                        "unexpected {other:?} in the push command list"
                    )))
                }
            }
        }
        Ok((commands, unnameable, peer))
    }

    /// One `<old> <new> <ref>` line, decoded by the shared grammar and then
    /// validated as a reference name by this server.
    ///
    /// **The split is on bytes, not on a lossy `String`,** and that is
    /// [`gunnar_sendpack::command::parse_line`]'s job now rather than this
    /// function's. Reference names are bytes in git: `refs/heads/\xff` is a
    /// name git will create and serve. Routing the line through
    /// `String::from_utf8_lossy` first replaced every invalid byte with U+FFFD
    /// *before* the name was validated, so a non-UTF-8 name was accepted and
    /// stored under a name the client never asked for — and any two names
    /// differing only in their invalid bytes collapsed onto the same reference,
    /// making a push to one silently overwrite the other.
    ///
    /// What stays here is the part that is this **server's** policy rather than
    /// the format's: **validating the name**, and
    /// reporting a hash-kind mismatch as [`Error::ObjectFormatMismatch`] so the
    /// client is told which two formats disagreed. The grammar refuses the same
    /// oid; it just cannot know what gunnar wants to call the refusal.
    /// The grammar's own failures stay hard errors — a line that is not
    /// `<old> <new> <ref>`, or an oid in the wrong hash, means the two ends
    /// disagree about the *format* and there is nothing left to report over.
    /// A name that parses and this server will not have is a different thing:
    /// the conversation is intact and the client is owed a line about that one
    /// reference, so it comes back as [`UnnameableRef`] rather than as `Err`.
    fn parse_command(
        hash_kind: gix_hash::Kind,
        line: &[u8],
        position: usize,
    ) -> Result<std::result::Result<RefCommand, UnnameableRef>> {
        let command =
            gunnar_sendpack::command::parse_line(line, hash_kind).map_err(|err| match err {
                gunnar_sendpack::Error::ObjectFormat { ours, theirs } => {
                    Error::ObjectFormatMismatch {
                        ours,
                        theirs: theirs.to_owned(),
                    }
                }
                other => Error::protocol(other.to_string()),
            })?;
        // ── the validator moved; the rule set did not ───────────────────────
        //
        // This was `FullName::new`, whose constructor ran the rules. The type
        // is gone with `RefStore` and the rules live in
        // `gunnar_store::refname::validate` — the same rules, in the place
        // `locked_on_target.md`'s checklist names, reached the same way
        // `gunnar-core::check_branch` reaches them. **This call is
        // load-bearing**: `gunnar-policy`'s gate layer no longer catches
        // `refs/heads/../../etc/passwd`, because its `RefUpdate::name` is a
        // plain `String` now, so this is the path-traversal boundary.
        //
        // ⚠ **A narrowing, recorded rather than hidden.** Reference names are
        // bytes in git and `FullName` held bytes; `RefRow::name` and
        // `RefUpdate::name` on the contract are `String`, so a name that
        // passes validation but is not UTF-8 — `refs/heads/\xff`, which git
        // will create and serve — can no longer be represented and comes back
        // as an [`UnnameableRef`] instead. That refuses **one** reference with
        // a line the client can read, and leaves the rest of the push
        // untouched, which is the failure direction this whole type exists to
        // produce. It is a real interop narrowing and it belongs to the
        // contract, not to this function: widening it means widening
        // `RefRow::name`, a two-implementation decision.
        let name = match validate_ref_name(command.name.as_slice()) {
            Ok(name) => name,
            Err(err) => {
                tracing::warn!(
                    target: GIT_TARGET,
                    event = "git.receive_pack.unnameable_ref",
                    ref_name = %String::from_utf8_lossy(&command.name),
                    error = %err,
                    "refusing one reference for its name; the rest of the push continues"
                );
                return Ok(Err(UnnameableRef {
                    name: command.name.to_vec(),
                    // git says `funny refname` and nothing else; gunnar's
                    // validator already knows *which* rule was broken and the
                    // client can only act on being told.
                    //
                    // **The rule and not the error.** `Error::InvalidRefName`
                    // renders the offending name inside its own message, so
                    // `format!("{err}")` here put the name in the `ng` line
                    // twice — and the `ng` line is one pkt-line. A 60 000-byte
                    // name then produced a 120 105-byte packet, `write_text`
                    // refused it with `TooLong`, and the whole report failed to
                    // write: the refusal path had its own way of dropping the
                    // session, which is the exact defect this type exists to
                    // remove. MEASURED 2026-08-06.
                    reason: format!("funny refname: {err}"),
                    position,
                    deletes: command.new.is_null(),
                }));
            }
        };
        Ok(Ok(RefCommand {
            old: command.old,
            new: command.new,
            name,
        }))
    }

    /// Connectivity, policy, then the reference transaction.
    fn apply(
        &self,
        commands: &[RefCommand],
        atomic: bool,
        already_refused: bool,
        gate: &dyn PushGate,
        facts: &PushFacts<'_>,
    ) -> Result<Vec<RefStatus>> {
        let mut statuses = vec![RefStatus::Ok; commands.len()];
        let _conn = crate::stage::enter("connectivity");
        tracing::info!(
            target: GIT_TARGET,
            event = "git.receive_pack.checking",
            commands = commands.len(),
            "checking that the push closes, before any reference moves"
        );
        let started = std::time::Instant::now();

        // ── THE INDEX, BEFORE THE WALK AND NOT DURING IT (`S-0xx`) ─────────
        //
        // The znippy arm acks a push's *bytes* before its *oids* exist: a pack
        // entry does not carry an oid, so one is knowable only after its delta
        // chain has been applied, and doing 13.9 M of those is what the indexer
        // puts behind the ack. Everything below reads objects by oid.
        //
        // Without this call the walk's first `peel` fell into that backend's
        // read-time fallback, which resolves the whole pending pack **on this
        // thread and throws the rows away** — a second resolver beside the
        // indexer's, over the same objects, each holding its own O(pack) tables.
        // MEASURED on the rig 2026-08-30, gunnar v1.1.3-180-g01ce6c58,
        // torvalds/linux --mirror at 16 GB and at 8 GB:
        //
        // ```text
        // 07:43:49  git.receive_pack.checking   commands=2874
        // 09:20:21  index.scanned  "a reader scanned a pending pack for itself:
        //                           13948244 objects in 5792442 ms"
        // ```
        //
        // 1h36m32s and 1h44m14s for the scan alone, and `git.receive_pack.checked`
        // never fired on either box: the clients sat unacked 2h14m30s and
        // 2h12m24s and the run ended only because the repository was deleted.
        // Stock git acks the same mirror in ~11 min, and it does it by running
        // `index-pack` first and `check_connected` second — which is exactly
        // this line and then the rest of this function.
        //
        // One call, before any read, so the resolve happens ONCE and everything
        // after it is a point lookup. A backend that indexes inside `put_pack`
        // (the gix arm) answers `Ok(0)` and pays nothing.
        match self.git.index_now() {
            Ok(0) => {}
            Ok(packs) => tracing::info!(
                target: GIT_TARGET,
                event = "git.receive_pack.indexed_for_check",
                packs = packs,
                elapsed_ms = started.elapsed().as_millis() as u64,
                "finished the deferred index before checking that the push closes"
            ),
            // **The push is REFUSED, and this used to be a warning it walked
            // past.** MEASURED on the rig 2026-08-30 at 8 GB, which is
            // what changed my mind: the backend's watchdog gave up by name
            // after 300 s (`index.gave_up ... 1.5G available, 1.6G needed`) —
            // exactly as designed, five minutes instead of the 1h45m park it
            // replaced — and then this arm logged a warning and went on to the
            // walk anyway. The walk's first read then fell into the very
            // read-time fallback `index_now` exists to keep it out of: a
            // whole-pack resolve on the connection thread, on a box that had
            // just said in writing that it cannot resolve this pack. The client
            // was back to hanging, one layer down.
            //
            // "Failing the push on this line would turn a slow index into a
            // refused mirror" — the old comment — mistook the error class. This
            // is not the backend being slow; it is the backend saying *I could
            // not make these objects addressable*. A check that cannot look
            // objects up cannot say whether the push closes, and the only
            // honest answers are "refused, here is why" or an unbounded wait.
            //
            // The backend's own sentence is carried through verbatim, because
            // it is the one that names what was available, what the floor is,
            // how far it got and what the operator can do. The bytes stay
            // durable and nothing partial is indexed, so the same push retried
            // on a bigger box is a no-op transfer and an ack.
            //
            // **A DIFFERENT event name from the ingest's**, deliberately.
            // `git.receive_pack.index_failed` already means *the pushed pack
            // could not be stored*; this one means *the pack is stored and
            // could not be indexed*. One name over two outcomes is a log
            // nobody can read.
            Err(err) => {
                tracing::warn!(
                    target: GIT_TARGET,
                    event = "git.receive_pack.index_for_check_failed",
                    commands = commands.len(),
                    elapsed_ms = started.elapsed().as_millis() as u64,
                    error = %format_args!("{err:#}"),
                    "the deferred index did not finish, so the push cannot be checked and is \
                     refused; the pack's bytes are durable and nothing partial was indexed"
                );
                // **The ROOT cause on the `ng` line, the whole chain in the
                // log.** `{err:#}` prints outermost-first, and on the rig at
                // 8 GB the outermost frame is the absorber's own context —
                // "absorbing pack 0 at (0, 8723679817) — its bytes are durable
                // and it stays un-indexed; reads will keep falling back rather
                // than answer absent" — which is true for a READER and is not
                // what happened to this push, and which buried the sentence
                // that matters ("this box cannot index this pack: the memory
                // watchdog gave up (deadline) with 1.2G available against a
                // 1.6G floor") 150 characters into a one-line pkt.
                //
                // A `ng` line is read by a person at a terminal, once. It gets
                // the cause; the log keeps the chain.
                let reason = format!("the push cannot be checked: {}", err.root_cause());
                return Ok(commands
                    .iter()
                    .map(|command| {
                        // A delete needs no object of this push to be
                        // addressable, and refusing one because an unrelated
                        // pack would not index is a refusal with no cause.
                        if command.is_delete() {
                            RefStatus::Ok
                        } else {
                            RefStatus::Rejected(reason.clone())
                        }
                    })
                    .collect());
            }
        }

        // The frontier is what the repository already pointed at. It is built
        // ONCE, not once per command: a push of a thousand references used to
        // peel every reference in the repository a thousand times.
        let snapshot = RefSnapshot::read(self.git)?;
        let mut frontier_values: Vec<ObjectId> = Vec::with_capacity(snapshot.rows().len());
        for row in snapshot.rows() {
            // Direct rows only, exactly as before: a symbolic reference points
            // at one of the others, so following it would put the same oid in
            // the frontier twice and add nothing.
            if let Some(bytes) = &row.oid {
                frontier_values.push(crate::graph::oid(bytes)?);
            }
        }
        let graph = Graph::new(self.git, self.hash_kind).with_budget(self.config.budget);
        let frontier = Frontier::new(&graph, &frontier_values)?;

        // ── ONE walk for the whole push, not one per reference ──────────────
        //
        // `verify_connectivity` already takes a *slice* of tips and shares one
        // `seen` set across all of them, and `Frontier` is already built once
        // for the whole push. What was still per-command was the
        // call, so N references over one shared history walked that history N
        // times — the second axis the earlier fix never touched.
        //
        // A mirror push is exactly that shape and it is the ordinary case for
        // an import: `linux.git` is 13 816 503 objects under **2 872**
        // references on one mainline, so the union of their closures is the
        // closure of the newest and the per-command loop was doing it 2 872
        // times. MEASURED 2026-08-05 on the in-memory fixture
        // (`tests/connectivity_cost.rs`): 64 references over one 200-commit
        // history cost **54.2x** what pushing the newest alone costs, 33 376
        // object touches against 616. The ratio is the reference count, so at
        // 2 872 it is the reason a kernel-sized push writes its pack and its
        // index and then never reports.
        //
        // And in wall clock, on a real repository over the `fs` backend, with
        // every object already present and an empty pack so that the
        // connectivity walk is the only thing in the bracket. `small.git`,
        // 12 328 objects, sha1; N references taken down one first-parent line,
        // which is `linux.git`'s shape; 2026-08-05, `taskset -c 0-7`, best
        // of three, **box not idle** (load average 33–48):
        //
        // ```text
        //   refs      per-command        batched      ratio
        //      1          0.086 s        0.084 s       1.0x
        //     16          0.561 s        0.104 s       5.4x
        //     64          1.978 s        0.141 s      14.0x
        //    256          6.903 s        0.122 s      56.6x
        //    512         10.424 s        0.153 s      68.1x
        // ```
        //
        // The left column is linear in the reference count and the right one is
        // flat, which is the whole claim. `njord.git` (10 468 objects, one
        // 966 MB blob) is the same shape at 17.0x for 256 references.
        //
        // # Why the per-command walk is still here
        //
        // Because the event stream reports **per reference**, and one batched walk can only
        // say that *something* in the push does not close. So the batch is the
        // fast path and the per-command loop is the attribution path, entered
        // only when the batch has already failed. That keeps the common case
        // O(union) and pays O(sum) only on a push that is being refused
        // anyway — which is also the one case where an operator needs the
        // precision. git makes the cheaper choice here and refuses the whole
        // batch with one message (`check_connected` over the entire command
        // list, `receive-pack.c`); this keeps gunnar's finer report at a cost
        // that is only ever paid on failure.
        let new_tips: Vec<ObjectId> = commands
            .iter()
            .filter(|command| !command.is_delete())
            .map(|command| command.new)
            .collect();
        let batch = if new_tips.is_empty() {
            Ok(0)
        } else {
            verify_connectivity(&graph, &new_tips, &frontier)
        };
        match &batch {
            Ok(objects) => tracing::info!(
                target: GIT_TARGET,
                event = "git.receive_pack.checked",
                tips = new_tips.len(),
                objects = *objects,
                elapsed_ms = started.elapsed().as_millis() as u64,
                "connectivity verified"
            ),
            Err(err) => tracing::warn!(
                target: GIT_TARGET,
                event = "git.receive_pack.check_failed",
                tips = new_tips.len(),
                elapsed_ms = started.elapsed().as_millis() as u64,
                error = %err,
                "the push does not close; re-walking one reference at a time to say which"
            ),
        }

        for (index, command) in commands.iter().enumerate() {
            // **A deletion skips the WALK, not the GATE.**
            //
            // `if command.is_delete() { continue; }` stood here, and the half
            // of it that is right is the walk: a deletion introduces no
            // objects, `command.new` is the zero oid, and there is nothing for
            // `verify_connectivity` to reach. The `continue` carried
            // `gate.check` out with it — so **every deletion bypassed the
            // pre-receive gate entirely**.
            //
            // What that made unenforceable is not a corner: `allow_delete` is
            // the field whose whole purpose is refusing this push, and
            // `allowed_principal_ids` and the ref-name denylist were bypassable
            // the same way. A branch could be protected against a force-push,
            // visibly and correctly, and deleted outright by anyone with WRITE
            // — which is a strictly worse outcome than the force-push beside
            // it, and was invisible behind it because the force-push guard
            // worked.
            //
            // `gunnar-policy`'s protection gate has always had a
            // `UpdateKind::Delete` arm with its own refusal message, and
            // `tests/protection.rs` exercises `allow_delete` directly. The gate
            // was written for this and simply never handed one.
            //
            // Found by `gunnar-ui/uitest`'s `a-protected-branch-refuses-a-deletion`
            // going red while the force-push row beside it passed; pinned in
            // `gunnar-cli/tests/policy_mirror.rs::a_protection_rule_refuses_a_real_branch_deletion`,
            // which drives stock git over real ssh.
            if !command.is_delete() && batch.is_err() {
                if let Err(err) = verify_connectivity(&graph, &[command.new], &frontier) {
                    tracing::warn!(target: GIT_TARGET, ref_name = %command.name, error = %err, "push failed connectivity");
                    statuses[index] =
                        RefStatus::Rejected(format!("missing necessary objects: {err}"));
                    continue;
                }
            }
            if let Err(reason) = gate.check(command, facts) {
                statuses[index] = RefStatus::Rejected(reason);
            }
        }

        crate::stage::end(_conn);
        if atomic && (already_refused || statuses.iter().any(|s| !s.is_ok())) {
            // One refusal fails the batch. Every other reference is reported
            // as refused too, or the client is told a lie about what landed.
            // `already_refused` carries in the refusals that never became a
            // command at all — a reference whose *name* this server will not
            // have (see [`UnnameableRef`]).
            for status in statuses.iter_mut() {
                if status.is_ok() {
                    *status = RefStatus::Rejected("atomic transaction failed".to_owned());
                }
            }
            return Ok(statuses);
        }

        // `edit_for` peels a new annotated-tag tip **once, here on the write
        // path**, while the objects the push just delivered are warm and the
        // connectivity walk above has already proven them present, and hands the
        // terminal commit to the store as the ref row's peeled column. That is
        // the write half of the stored peel: with it stored, `advertise_from` reads the
        // peeled oid straight out of the row and never re-peels, so every later
        // clone, fetch and `ls-remote` pays zero object reads per tag instead of
        // one. The graph is the same one `verify_connectivity` just used.
        let edits: Vec<(usize, RefCas<'_>)> = commands
            .iter()
            .enumerate()
            .filter(|(index, _)| statuses[*index].is_ok())
            .map(|(index, command)| Ok((index, edit_for(command, &graph)?)))
            .collect::<Result<_>>()?;

        let _txn = crate::stage::enter("ref_txn");
        tracing::info!(
            target: GIT_TARGET,
            event = "git.receive_pack.updating",
            edits = edits.len(),
            atomic,
            "applying the reference updates"
        );
        let started = std::time::Instant::now();
        if atomic {
            // ONE call. `put_refs_cas` applies every edit or none and names the
            // first whose expected value did not match; a loop with a break
            // leaves what it already applied applied, which is the thing
            // `--atomic` promises does not happen.
            let batch: Vec<RefCas<'_>> = edits
                .iter()
                .map(|(_, edit)| RefCas {
                    name: edit.name.clone(),
                    old: edit.old,
                    new: edit.new,
                    peeled: edit.peeled.clone(),
                })
                .collect();
            if let Err(err) = self.git.put_refs_cas(&batch) {
                let reason = format!("atomic transaction failed: {err:#}");
                for (index, _) in &edits {
                    statuses[*index] = RefStatus::Rejected(reason.clone());
                }
            }
        } else {
            for (index, edit) in edits {
                // A **one-element `put_refs_cas`**, not `update_ref`: both are
                // the contract's compare-and-swap-one-reference call and a
                // single-edit atomic batch has identical single-ref semantics —
                // it applies that one ref or rejects it, raising the same typed
                // `RefRejection::Cas` a lost race needs. `update_ref` cannot
                // carry a peel (its signature is `name/old/new`), and `put_refs`
                // would lose the CAS; `put_refs_cas` keeps the CAS **and** takes
                // the `RefCas::peeled` this edit computed, so the non-atomic path
                // stores a tag's peel exactly as the atomic path above does
                // (one implementation: one writer for both, not a peel-losing
                // second one).
                // Per-ref independence is preserved because each edit is its own
                // call — a rejection on one does not touch the others.
                if let Err(err) = self.git.put_refs_cas(std::slice::from_ref(&edit)) {
                    // ── THE TYPED REJECTION, RECOVERED ──────────────────────
                    //
                    // This used to render `{err:#}` and the comment here said
                    // so: "A LOST COMPARE-AND-SWAP AND A BROKEN DISK ARE THE
                    // SAME STRING HERE … Nothing may recover the distinction by
                    // matching on this text; the fix is a typed rejection on
                    // the contract."
                    //
                    // The contract had `RefRejection` all along
                    // (`git-storage-trait`); `put_refs_cas` raised it and
                    // `update_ref` raised a bare `bail!`. That is fixed, so the
                    // distinction is a `downcast_ref` rather than a substring
                    // match — and this is deliberately NOT parsing the message,
                    // which the old comment forbade in as many words.
                    //
                    // A lost race is rendered in git's own words so a client
                    // can act on it; anything else keeps the full chain,
                    // because a broken disk deserves every line it has.
                    statuses[index] = match err.downcast_ref::<RefRejection>() {
                        Some(RefRejection::Cas {
                            name,
                            expected,
                            actual,
                        }) => RefStatus::Rejected(format!(
                            "cannot lock ref '{name}': is at {} but expected {}",
                            render(actual),
                            render(expected)
                        )),
                        Some(RefRejection::Locked { name }) => RefStatus::Rejected(format!(
                            "cannot lock ref '{name}': held by another writer, try again"
                        )),
                        _ => RefStatus::Rejected(format!("{err:#}")),
                    };
                }
            }
        }
        tracing::info!(
            target: GIT_TARGET,
            event = "git.receive_pack.updated",
            // Counted off `statuses`, which is what the client is about to be
            // told, rather than off `edits`, which is only what was attempted.
            // A transaction that lost its compare-and-swap turns an edit into a
            // refusal, and a log reporting the attempt would report a reference
            // as moved that did not (measure, never assert).
            refs_ok = statuses.iter().filter(|s| s.is_ok()).count(),
            refs_rejected = statuses.iter().filter(|s| !s.is_ok()).count(),
            elapsed_ms = started.elapsed().as_millis() as u64,
            "the reference updates are applied"
        );

        Ok(statuses)
    }

    fn write_report(
        &self,
        out: &mut dyn Write,
        outcome: &PushOutcome,
        sideband: bool,
    ) -> Result<()> {
        // The lines come from the shared grammar (`gunnar-sendpack`), which is
        // the same code `gunnar-client` parses them with. What stays here is
        // the framing and the sideband, which belong to this transport.
        //
        // The two lists are merged back into the order the client sent, using
        // the position [`UnnameableRef`] carries. A client zips the report
        // against its own command list — `gunnar-client` does, and so does
        // `git push`'s per-ref output — so a report that silently reordered
        // would attribute a refusal to the wrong reference.
        let mut statuses: Vec<(usize, gunnar_sendpack::RefStatus)> = Vec::new();
        // The nameable commands sit at the positions the unnameable ones left
        // free, in order, which is enough to recover the original interleaving
        // without threading a position through `RefCommand` as well.
        let taken: std::collections::BTreeSet<usize> =
            outcome.unnameable.iter().map(|r| r.position).collect();
        let mut free = (0..).filter(|position| !taken.contains(position));
        for (command, status) in outcome.commands.iter().zip(&outcome.statuses) {
            let position = free.next().expect("the position sequence is infinite");
            statuses.push((
                position,
                match status {
                    RefStatus::Ok => gunnar_sendpack::RefStatus::ok(command.name.as_bytes()),
                    RefStatus::Rejected(reason) => {
                        gunnar_sendpack::RefStatus::rejected(command.name.as_bytes(), reason)
                    }
                },
            ));
        }
        for refused in &outcome.unnameable {
            statuses.push((
                refused.position,
                gunnar_sendpack::RefStatus::rejected(
                    echoable_name(&refused.name, refused.reason.len()),
                    &refused.reason,
                ),
            ));
        }
        statuses.sort_by_key(|(position, _)| *position);
        let statuses: Vec<gunnar_sendpack::RefStatus> =
            statuses.into_iter().map(|(_, status)| status).collect();
        let mut report = Vec::new();
        for line in gunnar_sendpack::report::lines(
            outcome.unpack.as_ref().copied().map_err(String::as_str),
            &statuses,
        ) {
            gunnar_pktline::write_text(&mut report, &line)?;
        }
        gunnar_pktline::write_flush(&mut report)?;

        if sideband {
            // The notice first, on band 2, so `git push` prints it as
            // `remote: …` ABOVE the per-ref lines — where a person reads before
            // the summary — and separately framed so a client that ignores band
            // 2 still parses the report untouched.
            if let Some(notice) = &self.config.notice {
                let mut line = notice.clone();
                if !line.ends_with('\n') {
                    line.push('\n');
                }
                wire_io::progress(out, true, &line)?;
            }
            // The whole report, its terminating flush-pkt included, rides band
            // 1; the connection is then ended by an *unframed* flush. That is
            // what `git-receive-pack` 2.53 puts on the wire (captured
            // 2026-08-03), and a client reading a demultiplexed stream cannot
            // see the difference between the two flushes any other way.
            wire_io::sideband(out, BAND_PACK, &report)?;
            wire_io::flush(out)?;
        } else {
            out.write_all(&report)?;
        }
        Ok(())
    }
}

/// The most of a refused reference name that can go into an `ng` line without
/// the line outgrowing a pkt-line.
///
/// A refused name is the one name on the wire that is **not** bounded by
/// anything gunnar chose: it arrived because a client sent it, and the only
/// ceiling it ever passed is the command packet's own 65 516 bytes. Echoing it
/// whole into `ng <name> <reason>\n` therefore does not fit whenever the name is
/// within a few dozen bytes of that ceiling, and `write_text` refuses the packet
/// rather than emitting a header that lies. The report then does not get
/// written at all — the session dies on the refusal path, which is precisely the
/// failure [`UnnameableRef`] exists to remove.
///
/// So a name too big to echo is truncated and **says so**, rather than costing
/// the client the whole report. A name this long cannot match anything the
/// client will keep anyway: every legal name is at most
/// [`gunnar_store::MAX_REF_COMPONENT_BYTES`] per component.
fn echoable_name(name: &[u8], reason_len: usize) -> Vec<u8> {
    const MARKER: &[u8] = b"\xe2\x80\xa6[truncated]";
    // `ng ` + name + ` ` + reason + `\n`.
    let overhead = 3 + 1 + reason_len + 1;
    let budget = gunnar_pktline::MAX_PAYLOAD.saturating_sub(overhead);
    if name.len() <= budget {
        return name.to_vec();
    }
    let keep = budget.saturating_sub(MARKER.len());
    let mut out = name[..keep].to_vec();
    out.extend_from_slice(MARKER);
    out
}

/// The rules `FullName::new` used to run, plus the one the contract adds.
///
/// Two failures with one shape, because both produce the same `ng` line and a
/// caller that had to tell them apart would be re-deriving the difference from
/// a message: the name is not a legal reference name, or it is legal and is not
/// UTF-8 and therefore cannot be a `RefRow::name`. See
/// [`ReceivePack::parse_command`] for why the second is a recorded narrowing
/// rather than a bug.
fn validate_ref_name(name: &[u8]) -> std::result::Result<String, String> {
    gunnar_store::refname::validate(name).map_err(|err| match err {
        // The validator renders the offending name inside its own message and
        // the `ng` line already carries the name, so only the *rule* is taken.
        // A 60 000-byte name once produced a 120 105-byte packet, `write_text`
        // refused it, and the whole report failed to write. MEASURED
        // 2026-08-06.
        gunnar_store::Error::InvalidRefName { reason, .. } => reason.to_owned(),
        other => other.to_string(),
    })?;
    String::from_utf8(name.to_vec())
        .map_err(|_| "a reference name gunnar can store must be UTF-8".to_owned())
}

/// One command as a compare-and-swap edit, with a tag's peeled target attached.
///
/// The three shapes `PreviousValue`/`Change` used to spell out collapse into
/// two `Option`s, because that is what the contract says and it says the same
/// thing: `old: None` is *"must not exist"* (a create) and `new: None` is a
/// delete. A non-create carries `command.old`, which is the client's claim
/// about what the reference holds and the only thing that makes this a race
/// the server can lose safely.
///
/// The fourth field, `peeled`, is the one this fn computes rather than copies:
/// an annotated tag's terminal commit, so the store persists it and the
/// advertisement never re-peels. It needs the object graph, which is why
/// this takes `graph` and returns a `Result` — a peel is a store read.
fn edit_for<'a>(command: &'a RefCommand, graph: &Graph<'_>) -> Result<RefCas<'a>> {
    // The peeled column, filled only for a create/update whose new tip is an
    // annotated tag. `peel` follows the tag chain to the first non-tag object
    // (a commit, in the common case of one tag over a commit) and reports the
    // tags it walked through: `tags` is non-empty **iff** the tip really was a
    // tag object, which is the only case git's `packed-refs` writes a `^peeled`
    // line for. A delete has no tip; a branch head or a lightweight tag peels to
    // itself with no tag walked, so both carry `None` and the store keeps no
    // peel — matching packed-refs exactly. This is the one peel per new tag the
    // whole stored-peel fix costs, paid once at ingest instead of on every
    // advertisement.
    //
    // The oid is **owned** here (`RefCas::peeled` is `Option<Vec<u8>>`): unlike
    // `old`/`new`, which are slices of the command line the caller still holds,
    // the peel is derived from the store and borrows nothing.
    let peeled = if command.is_delete() {
        None
    } else {
        let peeled = graph.peel(&command.new)?;
        (!peeled.tags.is_empty()).then(|| peeled.id.as_bytes().to_vec())
    };
    Ok(RefCas {
        name: command.name.clone(),
        old: (!command.is_create()).then(|| command.old.as_bytes()),
        new: (!command.is_delete()).then(|| command.new.as_bytes()),
        peeled,
    })
}

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

    /// **[`echoable_name`] is a fix, not a guard, and it is not reachable from
    /// the wire today. Both facts, with the arithmetic that decides them.**
    ///
    /// A command line is one pkt-line, so the longest reference name a client
    /// can physically send is `MAX_PAYLOAD - (40 + 1 + 40 + 1 + 1)` =
    /// **65 433** bytes. The `ng` line it comes back in is
    /// `"ng " + name + " " + reason + "\n"`, so the name fits whenever the
    /// reason is at most `65 511 - 65 433` = **78** bytes. Every reason
    /// [`ReceivePack::parse_command`] can produce is `"funny refname: "` plus
    /// one of the validator's fixed strings, and the longest of those makes
    /// **51**. 51 < 78, so no push can currently reach the truncation.
    ///
    /// That margin is 27 bytes of coincidence between two constants nobody
    /// holds together. One longer refusal string and the report becomes
    /// unwritable — `write_text` refuses the packet, `write_report` returns
    /// `Err`, and the session dies **on the refusal path**, which is the exact
    /// failure [`UnnameableRef`] exists to remove. So the truncation stays, and
    /// this is where it is proven, since the wire cannot prove it.
    #[test]
    fn a_refused_name_too_long_to_echo_is_truncated_rather_than_costing_the_report() {
        let name = vec![b'x'; gunnar_pktline::MAX_PAYLOAD - 40];
        let reason = "funny refname: ".to_owned() + &"y".repeat(200);

        let echoed = echoable_name(&name, reason.len());

        // The property that matters, asserted FIRST so a red run names the
        // consequence and not a proxy for it: the line frames.
        let mut line = b"ng ".to_vec();
        line.extend_from_slice(&echoed);
        line.push(b' ');
        line.extend_from_slice(reason.as_bytes());
        let mut out = Vec::new();
        gunnar_pktline::write_text(&mut out, &line).unwrap_or_else(|err| {
            panic!(
                "the `ng` line for a refused {}-byte name does not fit in a pkt-line ({err}), \
                 so the whole report fails to write and the session dies on the refusal path",
                name.len()
            )
        });
        assert!(
            echoed.ends_with("…[truncated]".as_bytes()),
            "the name was cut to fit and the client is not told; a truncated name that \
             does not say so is a name the client cannot match against its own list"
        );

        // And the acceptance half: a name that fits is not touched.
        let short = b"refs/heads/main".to_vec();
        assert_eq!(
            echoable_name(&short, reason.len()),
            short,
            "a name that fits was truncated anyway"
        );
    }
}