gunnar-wire 1.5.1

//! `git-upload-pack`: serving fetch and clone.
//!
//! Protocol **v2** is the real implementation; **v0** is a fallback that has to
//! exist because real clients still open it. That is not hypothetical: `git`
//! 2.53 falls back to v0 whenever `GIT_PROTOCOL` does not reach the server
//! process, which is exactly what happens over the `ext::` transport. Measured
//! 2026-08-03, and the reason `tests/real_git_interop.rs` gets a genuine,
//! unforced v0 exchange for free.
//!
//! # Shape
//!
//! Nothing here owns a socket. [`UploadPack::serve`] takes `&mut dyn BufRead`
//! and `&mut dyn Write`; `gunnar-ssh` supplies a russh channel, a test supplies
//! a pipe or a `Vec<u8>`, and the state machine cannot tell.
//!
//! # `object-format`
//!
//! Echoed from the repository's own hash kind — carried in at construction,
//! from `RepoStore::hash_kind()` — and never from a default. Answering `sha1` to a sha256 repository does not error; it returns
//! an *empty* response, which is why it cost a session in `nornir-git` and why
//! [`crate::caps::check_object_format`] turns a mismatch into a hard error here.

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

use gix_hash::{Kind, ObjectId};
// `write_pack` and `WriteOptions` are gone from this module: the engine owns
// the format now, and the three knobs that used to be `WriteOptions` —
// `ofs_delta`, `thin`, the base-closure ceiling — cross the seam as
// `git_storage_trait::Caps`. `WriteOutcome` stays, because it is what this end
// reports and the server logs.
use gunnar_pktline::Packet;
use gunnar_store::git_storage_trait::GitServe;
use gunnar_store::object;

use crate::caps::{self, Offered, PeerCapabilities};
use crate::filter::Filter;
use crate::graph::{
    has_for_negotiation, reachable_closure, select_objects, Graph, Missing, SelectOptions,
};
use crate::limits::{Budget, Limits};
use crate::negotiate::{negotiate, HaveMemo};
use crate::refs::RefSnapshot;
use crate::shallow::{self, ShallowOutcome, ShallowRequest};
use crate::wire_io::{self, err_packet, progress, read_packet, read_section, PackOutput};
use crate::{Error, ProtocolVersion, Result, Session};

/// Knobs the SSH layer sets per connection.
#[derive(Debug, Clone)]
pub struct Config {
    /// The `agent=` string. Defaults to `gunnar/<version>`.
    pub agent: String,
    /// Which optional v2 features are on offer.
    pub offered: Offered,
    /// Which object ids a client may `want`.
    pub want_policy: WantPolicy,
    /// An operator's explicit cap on the objects one traversal may hold.
    ///
    /// **`None` by default, and that is not a guard left off.**
    /// [`Limits::walk_bytes`] is the ceiling that ships, in bytes, and this is
    /// `--fetch-budget` for an operator who wants a second one in objects.
    ///
    /// It used to default to [`crate::DEFAULT_BUDGET`], and leaving it there
    /// would have made the byte ceiling decorative: MEASURED 2026-08-05,
    /// ten million objects refuses a clone of a 14 042 000-object repository
    /// that the same machine serves in 5.59 GB, so the object cap and not the
    /// memory would still have been what decided.
    ///
    /// See [`crate::DEFAULT_BUDGET`] for why the unit was wrong and
    /// [`limits`](crate::limits) for the ceilings that are about bytes.
    pub budget: Option<usize>,
    /// The memory ceilings this connection serves under.
    ///
    /// Defaults to [`Limits::default`], which is **every ceiling on**. See
    /// [`limits`](crate::limits).
    pub limits: Limits,
    /// The server-wide purse this connection's traversals spend from.
    ///
    /// **`None` by default, and this is the one field where that is not a guard
    /// left off.** A pool is shared state between connections, so it can only be
    /// created by whatever owns them all: `gunnar serve` builds one sized to the
    /// machine and hands the same `Arc` to every session. A `Config` built by
    /// hand — a test, `gunnar mirror`, an embedding — has no siblings to share
    /// with, and giving it a private pool of its own would be a ceiling that
    /// bounds nothing while reading as if it did.
    ///
    /// [`Limits::walk_bytes`] is the per-request half and is on by default. See
    /// [`WalkPool`](crate::WalkPool) for why one without the other is not a
    /// bound on the server.
    pub walk_pool: Option<std::sync::Arc<crate::WalkPool>>,
}

impl Default for Config {
    fn default() -> Self {
        Config {
            agent: caps::default_agent(),
            offered: Offered::default(),
            want_policy: WantPolicy::default(),
            budget: None,
            limits: Limits::default(),
            walk_pool: None,
        }
    }
}

/// Which object ids a client is allowed to ask for.
///
/// git spells this as three separate `uploadpack.allow*SHA1InWant` settings,
/// and the middle one is not optional for a server that advertises `filter`: a
/// partial clone's **lazy fetch** asks for bare blob ids that were never
/// advertised, so [`WantPolicy::AdvertisedTips`] turns every `git checkout` in
/// a `blob:none` clone into "the remote end hung up unexpectedly".
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum WantPolicy {
    /// Only ids that appeared in the advertisement.
    ///
    /// The strictest setting, and the one that breaks partial clone.
    AdvertisedTips,
    /// Anything reachable from a reference. This is git's
    /// `uploadpack.allowReachableSHA1InWant`, and the default here.
    ///
    /// **The cost is honest and real:** proving reachability means walking the
    /// repository's whole object closure once per request that names a
    /// non-tip id. The `git` package's `__gunnar_reach__` bitmaps are what make this
    /// cheap; until they exist, this walk is the price of partial clone.
    #[default]
    ReachableFromRefs,
    /// Anything present in the store, reachable or not.
    ///
    /// Makes every object that was ever pushed fetchable by whoever can name
    /// it, including on branches that have since been deleted.
    Any,
}

impl WantPolicy {
    /// Does this policy make good on a capability row that
    /// [`requires`](crate::caps::Capability::requires) `promise`?
    ///
    /// # Why the answer is a `match` and not a `>=`
    ///
    /// The three variants happen to be orderable, and encoding that as an
    /// ordering would make adding a fourth silently inherit whichever
    /// neighbours it landed between. Written out, a new variant does not
    /// compile until someone decides what it promises — which is the property
    /// that was missing, not the ordering.
    ///
    /// The mapping is git's, bit for bit: `ALLOW_TIP_SHA1` prints
    /// `allow-tip-sha1-in-want`, `ALLOW_REACHABLE_SHA1` prints
    /// `allow-reachable-sha1-in-want`, and `ALLOW_ANY_SHA1` is both bits set,
    /// so it prints both (`upload-pack.c`, `send_ref`).
    ///
    /// [`WantPolicy::AdvertisedTips`] promises nothing, on purpose: gunnar
    /// hides no reference, so "only ids in the advertisement" is exactly what a
    /// v0 client assumes when neither capability is present.
    pub fn advertises(self, promise: WantPolicy) -> bool {
        match (self, promise) {
            // Reachability is promised by the policy that serves it and by the
            // one that serves more.
            (WantPolicy::ReachableFromRefs, WantPolicy::ReachableFromRefs) => true,
            (WantPolicy::Any, WantPolicy::ReachableFromRefs) => true,
            (WantPolicy::Any, WantPolicy::Any) => true,
            (WantPolicy::AdvertisedTips, _) => false,
            (WantPolicy::ReachableFromRefs, WantPolicy::Any) => false,
            (_, WantPolicy::AdvertisedTips) => false,
        }
    }
}

/// The server side of `git-upload-pack`.
///
/// One of these serves **one connection**. That matters for [`Self::fetch`],
/// which a v2 client calls once per negotiation round: the `have`s a client has
/// already been answered about are memoised here rather than re-resolved
/// against the object store every round. The memo holds no decisions, only
/// facts about the store (see `negotiate::HaveMemo`), so a shared or reused
/// instance answers the same, only slower.
pub struct UploadPack<'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 this crate cannot work without. So the
    /// objects and the references are the same handle, which is also the only
    /// way an advertisement and the pack it promises can be one snapshot.
    git: &'a dyn GitServe,
    /// The repository's hash kind.
    ///
    /// Carried rather than asked for: `GitServe` has no `hash_kind()` and the
    /// frozen signature leaves whether `GitOps` should grow one open. `RepoStore::hash_kind()`
    /// already exists and the server already holds one per repository, so
    /// nothing is being invented here — only carried the last hop, and the open
    /// question keeps one fewer consumer.
    hash_kind: Kind,
    config: Config,
    haves: HaveMemo,
    /// Whether a sideband is open on this connection, and therefore which of
    /// the two channels [`UploadPack::report_error`] must use.
    ///
    /// Set by [`UploadPack::send_pack`], which is the one place either protocol
    /// version opens one. Before that an `ERR` pkt-line is the only thing a
    /// client will read; after it, a bare pkt-line in the middle of a
    /// demultiplexed stream is a framing error rather than a message, so the
    /// reason has to ride band 3.
    ///
    /// An `AtomicBool` rather than a `Cell` to match [`HaveMemo`]'s `Mutex`:
    /// this type is built inside `spawn_blocking` and nothing shares it across
    /// threads today, but a `Cell` here would silently make it `!Sync` and the
    /// day someone does share it the compiler should be the one saying so.
    sideband_open: std::sync::atomic::AtomicBool,
}

/// One reference as it goes out in an advertisement.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AdvertisedRef {
    /// Full reference name.
    ///
    /// `String`, not the `FullName` newtype that died with `RefStore`: the
    /// contract speaks `RefRow::name: String` and minting a second validated
    /// name type in gunnar is the duplication the demolition removes. What the
    /// newtype carried *by construction* is now a rule about the producer —
    /// see `receive_pack::parse_command`, which is where a client-supplied name
    /// is validated before it can reach a store.
    pub name: String,
    /// What it resolves to, or `None` for an unborn `HEAD`.
    pub target: Option<ObjectId>,
    /// The commit an annotated tag points at, when peeling was asked for.
    pub peeled: Option<ObjectId>,
    /// The reference a symbolic reference points at.
    pub symref_target: Option<String>,
}

/// The arguments of `command=ls-refs`.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct LsRefsRequest {
    /// Report `symref-target:` for symbolic references.
    pub symrefs: bool,
    /// Report `peeled:` for annotated tags.
    pub peel: bool,
    /// Report an unborn `HEAD` instead of omitting it.
    pub unborn: bool,
    /// Only references starting with one of these; empty means all.
    pub ref_prefixes: Vec<String>,
}

impl LsRefsRequest {
    /// Parse the argument section of `command=ls-refs`.
    pub fn parse(lines: &[Vec<u8>]) -> Result<Self> {
        let mut out = LsRefsRequest::default();
        for line in lines {
            let text = String::from_utf8_lossy(line).into_owned();
            let (name, value) = split_argument(&text);
            let Some(argument) = caps::argument(caps::LS_REFS_ARGUMENTS, name) else {
                return Err(Error::UnknownArgument {
                    command: "ls-refs",
                    argument: text,
                });
            };
            match (argument.name, argument.value, value) {
                ("symrefs", _, _) => out.symrefs = true,
                ("peel", _, _) => out.peel = true,
                ("unborn", _, _) => out.unborn = true,
                ("ref-prefix", caps::Value::Required, Some(prefix)) => {
                    out.ref_prefixes.push(prefix.to_owned())
                }
                _ => {
                    return Err(Error::UnknownArgument {
                        command: "ls-refs",
                        argument: text,
                    })
                }
            }
        }
        Ok(out)
    }
}

/// The arguments of `command=fetch`, and of a v0 `want` section.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct FetchRequest {
    /// `want <oid>`.
    pub wants: Vec<ObjectId>,
    /// `want-ref <name>`, the `ref-in-want` feature.
    pub want_refs: Vec<String>,
    /// `have <oid>`.
    pub haves: Vec<ObjectId>,
    /// `done`: stop negotiating and send the pack.
    pub done: bool,
    /// `thin-pack`, and since the thin-pack ruling this one is **honoured**, not ignored.
    ///
    /// Pack-copy preserves the deltas the pushing client computed, but until
    /// 2026-08-04 only those whose base was also in the transfer: the rest
    /// were inflated and deflated whole, because `allow_thin_pack: false` will
    /// not point an entry at a base the receiver may not have. On a **clone**
    /// that costs nothing, since the transfer is the whole history and every
    /// base is in it. On a **negotiated fetch** it is the dominant cost —
    /// MEASURED 2026-08-04, 24 of 40 objects expanded and 3.98x stock
    /// git's thin pack of the same object set.
    ///
    /// What changed is not the caution, only what backs it: the walk that
    /// decides the transfer already knows what the client holds
    /// (`graph::Selection::client_has`), so a base can now be *checked*
    /// instead of assumed unsafe. A base that does not answer still costs its
    /// object the delta. See `send_pack` for the three gating conditions.
    pub thin_pack: bool,
    /// `no-progress`.
    pub no_progress: bool,
    /// `include-tag`.
    pub include_tag: bool,
    /// `ofs-delta`, and since pack-copy this one is **honoured**, not ignored.
    ///
    /// The pack-copy path emits an entry exactly as it found it in a pack, and
    /// most of those entries are `ofs-delta`s. A client that did not ask for
    /// `ofs-delta` cannot read them, so it gets whole objects instead. That is
    /// what `git-upload-pack` does too: it passes `--delta-base-offset` to
    /// `pack-objects` only when the client advertised the capability.
    ///
    /// Every git since 2005 sends it and stock `git 2.53.0` always does, so
    /// this is the fallback, not the path. It is a real fallback though, and
    /// `tests/pack_copy.rs` exercises it rather than assuming nobody lands
    /// there.
    pub ofs_delta: bool,
    /// `filter <spec>`.
    pub filter: Option<Filter>,
    /// `wait-for-done`.
    pub wait_for_done: bool,
    /// Everything shallow-related.
    pub shallow: ShallowRequest,
}

impl FetchRequest {
    /// Parse the argument section of `command=fetch`.
    ///
    /// Unknown arguments are an error rather than a shrug. The argument table
    /// in [`crate::caps`] is the same one the advertisement is generated from,
    /// so "accepted here" and "advertised there" cannot drift.
    pub fn parse(lines: &[Vec<u8>], hash_kind: Kind, offered: Offered) -> Result<Self> {
        let mut out = FetchRequest::default();
        for line in lines {
            let text = String::from_utf8_lossy(line).into_owned();
            let (name, value) = split_argument(&text);
            let argument = caps::argument(caps::FETCH_ARGUMENTS, name)
                .filter(|argument| offered.accepts(argument))
                .ok_or_else(|| Error::UnknownArgument {
                    command: "fetch",
                    argument: text.clone(),
                })?;
            let required = || {
                value.ok_or_else(|| Error::UnknownArgument {
                    command: "fetch",
                    argument: text.clone(),
                })
            };
            match argument.name {
                "want" => out.wants.push(parse_oid(required()?, hash_kind)?),
                "have" => out.haves.push(parse_oid(required()?, hash_kind)?),
                "want-ref" => out.want_refs.push(required()?.to_owned()),
                "done" => out.done = true,
                "thin-pack" => out.thin_pack = true,
                "no-progress" => out.no_progress = true,
                "include-tag" => out.include_tag = true,
                "ofs-delta" => out.ofs_delta = true,
                "wait-for-done" => out.wait_for_done = true,
                "filter" => out.filter = Some(Filter::parse(required()?)?),
                "shallow" => out
                    .shallow
                    .client_shallow
                    .push(parse_oid(required()?, hash_kind)?),
                "deepen" => {
                    out.shallow.depth = Some(required()?.parse::<usize>().map_err(|_| {
                        Error::protocol(format!("deepen wants a number, got {text:?}"))
                    })?)
                }
                "deepen-relative" => out.shallow.relative = true,
                "deepen-since" => {
                    out.shallow.since = Some(required()?.parse::<i64>().map_err(|_| {
                        Error::protocol(format!("deepen-since wants unix seconds, got {text:?}"))
                    })?)
                }
                "deepen-not" => out.shallow.not.push(required()?.to_owned()),
                other => {
                    return Err(Error::UnknownArgument {
                        command: "fetch",
                        argument: other.to_owned(),
                    })
                }
            }
        }
        Ok(out)
    }
}

/// What a whole `upload-pack` session did.
///
/// Returned rather than logged because the SSH layer wants it for metrics and
/// the interop tests want it as evidence: "the fetch succeeded" is true of a
/// server that negotiated properly and of one that shipped the entire
/// repository every time, and only `fetch_rounds` and `objects_sent` tell them
/// apart.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct UploadSummary {
    /// Which protocol was spoken.
    pub protocol: ProtocolVersion,
    /// How many `ls-refs` commands were served.
    pub ls_refs: usize,
    /// How many negotiation rounds the client took, the one that carried the
    /// packfile included.
    pub fetch_rounds: usize,
    /// Objects in the packfile, if one was sent.
    ///
    /// **This is what the ENGINE emitted, not what this end chose.** See
    /// [`objects_selected`](Self::objects_selected) for the other half; a
    /// reader that has only this one cannot tell the two apart.
    pub objects_sent: u32,
    /// Objects [`GitServe::select`] chose — the **exact set** `emit_pack` was
    /// handed.
    ///
    /// # Why both counts, and why this one is asserted rather than logged
    ///
    /// `emit_pack`'s first parameter is an exact set and the engine must not
    /// close it over reachability (DECIDED 2026-08-10). It may still emit
    /// *more* than it was given, legitimately: an `OFS_DELTA` names its base by
    /// a backwards distance **inside the pack**, so an engine copying stored
    /// entries has to bring along any delta base that fell outside the
    /// selection or emit something unparseable. `Caps::thin` is the allowance
    /// to skip that, and an engine is free not to take it.
    ///
    /// So the gap is expected and **unbounded**, which is precisely why it has
    /// to be reported rather than assumed small. MEASURED on the laptop 2026-08-10,
    /// the `h2h-linear-sha1-2048c-1024f-16k` corpus over the znippy engine, a
    /// negotiated fetch from a receiver `N` commits behind the tip:
    ///
    /// ```text
    ///   N commits behind    selected    sent    amplification
    ///                  1          18     157            8.72x
    ///                256        4382    7801            1.78x
    /// ```
    ///
    /// Until this field existed the left-hand column reached exactly one place
    /// — a human-readable sideband progress line that `git fetch -q` suppresses
    /// and no meter parses — so `bench/`'s `gunnar.vs_git_push` could report a
    /// 33.6% divergence with nothing anywhere able to attribute it. That is the
    /// failure a published number can hide: one describing a path other than the
    /// one its name claims.
    ///
    /// Zero for a session that sent no packfile. Summed over rounds in v2, for
    /// the same reason [`objects_sent`](Self::objects_sent) is.
    pub objects_selected: usize,
    /// How many `have`s turned out to be common on the final round.
    pub common: usize,
    /// Of [`objects_sent`](Self::objects_sent), how many left as a **copy** of
    /// the pack entry they were already stored as.
    ///
    /// Reported for the same reason `fetch_rounds` and `objects_sent` are: "the
    /// clone succeeded" is true of a server that shipped the client's own
    /// deltas back and of one that inflated and re-deflated 528 MiB to send
    /// 18× the bytes, and nothing else on this struct tells them apart. A byte
    /// count cannot either, since it moves with the corpus.
    pub objects_copied_from_pack: usize,
    /// Of [`objects_sent`](Self::objects_sent), how many had to be deflated
    /// from their decompressed content: everything, on a store with no packs,
    /// and on a packed one only the loose objects and the deltas whose base
    /// did not come along.
    pub objects_recompressed: usize,
    /// Whether the packfile round's object selection came from a `.bitmap`
    /// rather than from the walk.
    ///
    /// # Why this is on the summary and not only in a log line
    ///
    /// Because the two paths do **not** produce identical bytes and every
    /// measurement of this server is therefore a measurement of one of them.
    /// The bitmap path vouches for the client's whole closure
    /// (`ClientHas::Bits`) where `Exclusion::Edges`' walk vouches only for the
    /// frontier, so a delta base the walk expands the bitmap names — same
    /// request, same repository, fewer bytes. Nothing else a server reports
    /// says which one answered, and the failure is silent in the direction that
    /// matters: a repository with no `.bitmap` on disk falls through to the
    /// walk and every number taken on it is a walk number wearing the
    /// default's name.
    ///
    /// MEASURED 2026-08-04: the head-to-head fetch arm's server fixture is
    /// a `git clone --bare` carve, `git clone` writes no `.bitmap`, and so
    /// `gunnar.vs_git_fetch_divergent` — the one arm gunnar loses on bytes —
    /// had never once exercised the shipping path. `false` here is what says
    /// so.
    ///
    /// `false` for a session that sent no packfile, and for a request shape
    /// [`crate::bitmap::select`] declines (a filter, a shallow boundary, a
    /// deepening).
    pub answered_from_bitmap: bool,
    /// Thin-delta bases `gix` offered over the session, and how many
    /// `client_has` refused. See [`crate::pack::WriteOutcome::thin_bases_offered`].
    pub thin_bases_offered: usize,
    /// See [`crate::pack::WriteOutcome::thin_bases_vetoed`].
    pub thin_bases_vetoed: usize,
    /// See [`crate::pack::WriteOutcome::closure_bases_declined`]. Non-zero only
    /// on a clone out of a **multi-pack** store whose selection reaches one
    /// object by two routes, which is the shape no fixture in this tree had
    /// until 2026-08-05.
    pub closure_bases_declined: usize,
}

/// What one `fetch` produced.
#[derive(Debug, Clone, Default)]
pub struct FetchOutcome {
    /// The `have`s this repository holds.
    pub common: Vec<ObjectId>,
    /// Whether the server declared a cut point.
    pub ready: bool,
    /// Whether a packfile was sent, as opposed to another negotiation round.
    pub sent_pack: bool,
    /// How many objects went into it — what the **engine** emitted.
    pub objects_sent: u32,
    /// How many objects `select` chose. See
    /// [`UploadSummary::objects_selected`], which is where the two-column
    /// argument is written down.
    pub objects_selected: usize,
    /// How many of those were copied out of a pack rather than deflated
    /// out of an inflated object. See [`UploadSummary::objects_copied_from_pack`].
    pub objects_copied_from_pack: usize,
    /// How many had to be deflated. See
    /// [`UploadSummary::objects_recompressed`].
    pub objects_recompressed: usize,
    /// Whether a `.bitmap` answered this round's selection. See
    /// [`UploadSummary::answered_from_bitmap`].
    pub answered_from_bitmap: bool,
    /// See [`UploadSummary::thin_bases_offered`].
    pub thin_bases_offered: usize,
    /// See [`UploadSummary::thin_bases_vetoed`].
    pub thin_bases_vetoed: usize,
    /// See [`UploadSummary::closure_bases_declined`].
    pub closure_bases_declined: usize,
}

/// What [`UploadPack::send_pack`] wrote, and which selection path chose it.
///
/// A pair rather than two out-parameters, and its own type rather than a
/// tuple, because the second half is the thing a reader of the first half most
/// needs to know: a byte count from the walk and a byte count from the bitmap
/// are answers to different questions (see
/// [`UploadSummary::answered_from_bitmap`]).
struct SentPack {
    written: crate::pack::WriteOutcome,
    answered_from_bitmap: bool,
    /// The exact set handed to `emit_pack`. See
    /// [`UploadSummary::objects_selected`].
    objects_selected: usize,
}

impl<'a> UploadPack<'a> {
    /// Serve `objects`/`refs` with default configuration.
    pub fn new(git: &'a dyn GitServe, hash_kind: Kind) -> Self {
        let config = Config::default();
        UploadPack {
            haves: HaveMemo::with_capacity(memo_capacity(&config.limits)),
            git,
            hash_kind,
            config,
            sideband_open: std::sync::atomic::AtomicBool::new(false),
        }
    }

    /// Replace the configuration.
    ///
    /// The `have` memo is rebuilt, because its ceiling is derived from
    /// [`Config::limits`] — a configuration that lifted the request ceiling and
    /// left a memo capped at the old one would be honouring a limit nobody set.
    pub fn with_config(mut self, config: Config) -> Self {
        self.haves = HaveMemo::with_capacity(memo_capacity(&config.limits));
        self.config = config;
        self
    }

    /// The repository's hash kind, and the only source of `object-format`.
    pub fn hash_kind(&self) -> Kind {
        self.hash_kind
    }

    /// **One traversal scope for this call.** Its lifetime is what releases the
    /// [`WalkPool`](crate::WalkPool) reservation and the thread's id charge, so
    /// it is built per call and dropped where the walk ends — never held on
    /// `self`, which lives for the whole connection.
    fn graph(&self) -> Graph<'_> {
        Graph::new(self.git, self.hash_kind)
            .with_budget(self.config.budget)
            .with_walk_limit(self.config.limits.walk_bytes, self.config.walk_pool.clone())
    }

    /// Serve one connection at the given protocol version.
    pub fn serve(
        &self,
        version: ProtocolVersion,
        input: &mut dyn BufRead,
        out: &mut dyn Write,
    ) -> Result<UploadSummary> {
        self.serve_session(version, Session::Connected, input, out)
    }

    /// Serve one *call* at the given protocol version and [`Session`].
    ///
    /// [`Session::Connected`] is what [`serve`](Self::serve) does and what SSH
    /// uses. [`Session::StatelessRpc`] is one smart-HTTP `POST`: no
    /// advertisement is written (it was a separate `GET`), and the call answers
    /// one round and returns.
    pub fn serve_session(
        &self,
        version: ProtocolVersion,
        session: Session,
        input: &mut dyn BufRead,
        out: &mut dyn Write,
    ) -> Result<UploadSummary> {
        let served = match version {
            ProtocolVersion::V2 => self.serve_v2_session(session, input, out),
            ProtocolVersion::V0 => self.serve_v0_session(session, input, out),
        };
        // ── the reason goes to the CLIENT, not only to the log ───────────────
        //
        // Every refusal this crate makes — an unsupported `filter`, an unknown
        // v2 command, an `object-format` mismatch, a `want` that is not
        // reachable — used to reach the transport as an `Err` and stop there.
        // The transport logs it and closes the channel, so `git` printed
        //
        // ```text
        // fatal: the remote end hung up unexpectedly
        // ```
        //
        // and the operator had to be on the server to learn why. MEASURED on
        // 2026-08-05, `git -c protocol.version=2 fetch <gunnar> <unknown
        // oid>`: the client got that line and nothing else, while the server
        // log carried `err=object 0000…0001 is not in this repository (a want)`
        // in full. `report_error` was written for exactly this and had **zero**
        // call sites in the tree, tests included, which is why nothing noticed.
        //
        // Here rather than at either transport: `serve_session` is the one
        // entry point SSH and smart HTTP both come through, so one call covers
        // both and they cannot drift (one implementation).
        //
        // A failure to write the reason is discarded on purpose. The commonest
        // way to get here at all is a client that has already hung up, and
        // replacing the real error with "could not tell you about the real
        // error" would lose the only useful fact in the exchange.
        if let Err(err) = &served {
            let sideband = self
                .sideband_open
                .load(std::sync::atomic::Ordering::Relaxed);
            let _ = self.report_error(out, sideband, &err.client_message());
            let _ = out.flush();
        }
        served
    }

    /// The advertisement a smart-HTTP `GET /info/refs?service=git-upload-pack`
    /// answers with, for either protocol version.
    ///
    /// The `# service=git-upload-pack` preamble is **not** written here: it is
    /// an HTTP framing detail (and one protocol v2 does not use at all), so it
    /// belongs to `gunnar-http` and not to a state machine that also runs over
    /// SSH, where no such line exists.
    pub fn advertise(&self, version: ProtocolVersion, out: &mut dyn Write) -> Result<()> {
        match version {
            ProtocolVersion::V2 => self.write_capability_advertisement(out),
            ProtocolVersion::V0 => self.write_ref_advertisement(out),
        }
    }

    // ── protocol v2 ──────────────────────────────────────────────────────

    /// The v2 capability advertisement, as bytes.
    ///
    /// Sent before the client says anything, exactly as `git-upload-pack` does.
    pub fn capability_advertisement(&self) -> Result<Vec<u8>> {
        let mut out = Vec::new();
        self.write_capability_advertisement(&mut out)?;
        Ok(out)
    }

    /// Write the v2 capability advertisement.
    pub fn write_capability_advertisement(&self, out: &mut dyn Write) -> Result<()> {
        wire_io::text(out, b"version 2")?;
        // The list is `caps::UPLOAD_PACK_V2`, not five literals here. It used
        // to be five literals here — a fourth hand-written advertisement, in a
        // different file from the other three — which is how `server-option`
        // came to be advertised with nothing in the tree that had to agree
        // with it. `caps::ADVERTISED` now names every surface and
        // `tests/advertisement.rs` walks all of them.
        for line in caps::advertise(
            caps::UPLOAD_PACK_V2,
            self.hash_kind(),
            &self.config.agent,
            self.config.offered,
            // No v2 row carries a `requires`: protocol v2 has no
            // `allow-*-sha1-in-want` capability, because `upload_pack_v2` in
            // git sets `ALLOW_ANY_SHA1` unconditionally and the question never
            // reaches the wire. Passed anyway so the surface cannot acquire a
            // policy-dependent row without this call site being the one that
            // decides what it renders under.
            self.config.want_policy,
        ) {
            wire_io::text(out, line.as_bytes())?;
        }
        wire_io::flush(out)?;
        Ok(())
    }

    /// The whole v2 session: advertisement, then commands until the client
    /// stops.
    pub fn serve_v2(&self, input: &mut dyn BufRead, out: &mut dyn Write) -> Result<UploadSummary> {
        self.serve_v2_session(Session::Connected, input, out)
    }

    /// The v2 session, for either [`Session`].
    ///
    /// Under [`Session::StatelessRpc`] the advertisement is skipped and the
    /// call returns after **one** command, because that is what one HTTP `POST`
    /// carries. Returning after one is not merely equivalent to hitting EOF: a
    /// body carrying a second command would otherwise be served on a
    /// negotiation state the next `POST` cannot see, which is the shape of a
    /// clone that loops.
    pub fn serve_v2_session(
        &self,
        session: Session,
        input: &mut dyn BufRead,
        out: &mut dyn Write,
    ) -> Result<UploadSummary> {
        let mut summary = UploadSummary {
            protocol: ProtocolVersion::V2,
            ..Default::default()
        };
        if session.advertises_inline() {
            let _s = crate::stage::enter("advertise");
            self.write_capability_advertisement(out)?;
            out.flush()?;
        }
        loop {
            let command = {
                let _s = crate::stage::enter("request_read");
                self.read_command(input)?
            };
            let Some(command) = command else {
                return Ok(summary);
            };
            caps::check_object_format(
                self.hash_kind(),
                command.capabilities.value("object-format"),
            )?;
            match command.name.as_str() {
                "ls-refs" => {
                    let _s = crate::stage::enter("ls_refs");
                    let request = LsRefsRequest::parse(&command.arguments)?;
                    self.ls_refs(&request, out)?;
                    summary.ls_refs += 1;
                }
                "fetch" => {
                    let request = {
                        let _s = crate::stage::enter("request_parse");
                        FetchRequest::parse(
                            &command.arguments,
                            self.hash_kind(),
                            self.config.offered,
                        )?
                    };
                    let outcome = self.fetch(&request, out)?;
                    summary.fetch_rounds += 1;
                    summary.common = outcome.common.len();
                    summary.objects_sent += outcome.objects_sent;
                    summary.objects_selected += outcome.objects_selected;
                    summary.objects_copied_from_pack += outcome.objects_copied_from_pack;
                    summary.objects_recompressed += outcome.objects_recompressed;
                    // `|=` and not `=`: only the round that carries the
                    // packfile has a selection at all, and the rounds before it
                    // return `FetchOutcome::default()`, whose `false` would
                    // otherwise overwrite the one answer this field exists to
                    // report. Seen red on 2026-08-04 by omitting this line
                    // entirely: every arm of a five-arm A/B reported
                    // `answered_from_bitmap: false` including the ones served
                    // from a bitmap, which is the exact shape of the four
                    // instruments that spent that day measuring the wrong path.
                    summary.answered_from_bitmap |= outcome.answered_from_bitmap;
                    summary.thin_bases_offered += outcome.thin_bases_offered;
                    summary.thin_bases_vetoed += outcome.thin_bases_vetoed;
                    summary.closure_bases_declined += outcome.closure_bases_declined;
                }
                other => {
                    return Err(Error::protocol(format!("unknown v2 command {other:?}")));
                }
            }
            {
                let _s = crate::stage::enter("flush");
                out.flush()?;
            }
            if session.is_stateless() {
                return Ok(summary);
            }
        }
    }

    /// Read one v2 command — `command=…`, its capability lines, and, after the
    /// delim-pkt, its argument section.
    ///
    /// **One [`Budget`] spans the whole command**, capability lines and
    /// arguments together, because they land in the same two `Vec`s and a
    /// ceiling that only covered one half would be walked around by putting the
    /// bytes in the other. The budget is per *round*: a v2 client re-sends its
    /// whole `have` list on every `command=fetch`, so what is being bounded is
    /// what the server holds at once, not what the client says over a session.
    fn read_command(&self, input: &mut dyn BufRead) -> Result<Option<V2Command>> {
        let mut budget = Budget::new(self.config.limits.request_bytes);
        let mut name: Option<String> = None;
        let mut capability_lines: Vec<String> = Vec::new();
        let mut arguments: Vec<Vec<u8>> = Vec::new();
        let saw_delim;

        loop {
            match read_packet(input)? {
                // A clean hang-up, or the client's flush-pkt: both mean the
                // session is over. Over SSH a client that is done simply
                // closes, so EOF here is normal and must not be an error.
                None | Some(Packet::Flush) => {
                    return Ok(None);
                }
                Some(Packet::Delim) => {
                    saw_delim = true;
                    break;
                }
                Some(Packet::ResponseEnd) => {
                    return Err(Error::protocol("a response-end-pkt in a v2 request"))
                }
                Some(Packet::Data(data)) => {
                    budget.spend(data.len(), "v2 request")?;
                    let text = String::from_utf8_lossy(data.strip_suffix(b"\n").unwrap_or(&data))
                        .into_owned();
                    match text.strip_prefix("command=") {
                        Some(command) => name = Some(command.to_owned()),
                        None => capability_lines.push(text),
                    }
                }
            }
        }

        if saw_delim {
            arguments = read_section(input, "a v2 argument section", &mut budget)?;
        }

        let name = name.ok_or_else(|| Error::protocol("a v2 request with no command="))?;
        Ok(Some(V2Command {
            name,
            capabilities: PeerCapabilities::parse(&capability_lines.join(" ")),
            arguments,
        }))
    }

    /// `command=ls-refs`.
    pub fn ls_refs(&self, request: &LsRefsRequest, out: &mut dyn Write) -> Result<()> {
        let refs = self.collect_refs(&request.ref_prefixes, request.peel, request.symrefs)?;
        for entry in refs {
            let mut line = match entry.target {
                Some(id) => format!("{id} {}", entry.name),
                None if request.unborn => format!("unborn {}", entry.name),
                None => continue,
            };
            if let Some(symref) = &entry.symref_target {
                line.push_str(&format!(" symref-target:{symref}"));
            }
            if let Some(peeled) = entry.peeled {
                line.push_str(&format!(" peeled:{peeled}"));
            }
            wire_io::text(out, line.as_bytes())?;
        }
        wire_io::flush(out)?;
        Ok(())
    }

    /// `command=fetch`, one round.
    ///
    /// Returns without a packfile when negotiation is not finished; the client
    /// then sends another `command=fetch` with more `have`s.
    pub fn fetch(&self, request: &FetchRequest, out: &mut dyn Write) -> Result<FetchOutcome> {
        let graph = self.graph();
        // ONE snapshot of the reference namespace for the whole round: the
        // `want-ref`s, the shallow computation and `include-tag` all resolve
        // against the same one, so a reference that moves mid-request cannot
        // make this fetch promise an oid under a name it no longer has.
        let refs = RefSnapshot::read(self.git)?;
        let mut wants = request.wants.clone();
        let mut wanted_refs: Vec<(ObjectId, String)> = Vec::new();
        {
            let _s = crate::stage::enter("want_check");
            for name in &request.want_refs {
                gunnar_store::refname::validate(name.as_bytes())?;
                // `HEAD` is NOT a row, so `resolve` — a row lookup — cannot find
                // it. It has an accessor pair instead, and `ls-refs` already
                // advertises it through `head()`. Without this arm every
                // protocol-v2 `git clone` fails: the advertisement offers
                // `HEAD symref-target:refs/heads/…`, the client answers
                // `want-ref HEAD`, and the server says the ref does not exist.
                //
                // `refname::validate` admits `HEAD`, so it passes validation and
                // dies one line later — which is why nothing caught it: the wire
                // suites drive the state machine directly, and `ext::` negotiates
                // v0, where `want-ref` never appears. Found by running a real
                // `git clone` (2026-08-10).
                let resolved = if name == "HEAD" {
                    refs.head(self.git)?
                } else {
                    refs.resolve(name)?
                };
                let Some(id) = resolved else {
                    return Err(Error::protocol(format!(
                        "want-ref names {name:?}, which does not exist"
                    )));
                };
                wanted_refs.push((id, name.clone()));
                wants.push(id);
            }
            dedupe_oids(&mut wants);
            self.check_wants(&wants)?;
        }

        // `done` means the client has stopped negotiating, so the expensive
        // reachability question is not asked at all, only which `have`s are
        // present, which is what the pack needs anyway.
        let negotiation = {
            let _s = crate::stage::enter("negotiate");
            negotiate(
                &graph,
                &wants,
                &request.haves,
                request.wait_for_done || request.done,
                &self.haves,
            )?
        };

        if !request.done {
            wire_io::text(out, b"acknowledgments")?;
            // `ack`, not `common`, in **both** branches, which is git's
            // `send_acks` line for line: `if (!acks->nr) NAK` and then one
            // `ACK` per entry. The pack is still computed from the whole of
            // `common`; what goes on the wire is the subset the client had not
            // already told the server about, and that is 25 400 bytes on the
            // divergent fixture. See `Negotiation::ack`.
            //
            // A round that finds commons but no *new* ones therefore answers
            // `NAK`, exactly as git's does — and the section is never empty,
            // which an `ack.is_empty()` fall-through would have made it.
            if negotiation.ack.is_empty() {
                wire_io::text(out, b"NAK")?;
            } else {
                for id in &negotiation.ack {
                    wire_io::text(out, format!("ACK {id}").as_bytes())?;
                }
            }
            if negotiation.ready {
                wire_io::text(out, b"ready")?;
                wire_io::delim(out)?;
            } else {
                wire_io::flush(out)?;
                return Ok(FetchOutcome {
                    common: negotiation.common,
                    ready: false,
                    sent_pack: false,
                    ..Default::default()
                });
            }
        }

        let want_commits = {
            let _s = crate::stage::enter("peel_wants");
            self.commit_wants(&graph, &wants)?
        };
        let shallow_outcome = {
            let _s = crate::stage::enter("shallow");
            shallow::compute(&graph, &refs, &want_commits, &request.shallow)?
        };
        if shallow_outcome.has_info() {
            wire_io::text(out, b"shallow-info")?;
            for id in &shallow_outcome.newly_shallow {
                // No trailing newline, matching what `git-upload-pack` 2.53
                // puts on the wire (captured 2026-08-03).
                wire_io::data(out, format!("shallow {id}").as_bytes())?;
            }
            for id in &shallow_outcome.unshallow {
                wire_io::data(out, format!("unshallow {id}").as_bytes())?;
            }
            wire_io::delim(out)?;
        }

        if !wanted_refs.is_empty() {
            wire_io::text(out, b"wanted-refs")?;
            for (id, name) in &wanted_refs {
                wire_io::text(out, format!("{id} {name}").as_bytes())?;
            }
            wire_io::delim(out)?;
        }

        wire_io::text(out, b"packfile")?;
        let written = self.send_pack(
            &graph,
            &refs,
            &wants,
            &negotiation.common,
            request,
            &shallow_outcome,
            out,
            true,
        )?;
        {
            let _s = crate::stage::enter("flush");
            wire_io::flush(out)?;
        }

        Ok(FetchOutcome {
            common: negotiation.common,
            ready: negotiation.ready,
            sent_pack: true,
            objects_sent: written.written.num_objects,
            objects_selected: written.objects_selected,
            objects_copied_from_pack: written.written.objects_copied_from_pack,
            objects_recompressed: written.written.objects_recompressed,
            answered_from_bitmap: written.answered_from_bitmap,
            thin_bases_offered: written.written.thin_bases_offered,
            thin_bases_vetoed: written.written.thin_bases_vetoed,
            closure_bases_declined: written.written.closure_bases_declined,
        })
    }

    // ── protocol v0 ──────────────────────────────────────────────────────

    /// The v0 reference advertisement.
    pub fn write_ref_advertisement(&self, out: &mut dyn Write) -> Result<()> {
        let capabilities = caps::upload_pack_v0_capabilities(
            self.hash_kind(),
            &self.config.agent,
            self.config.offered,
            self.config.want_policy,
        );
        let refs_snapshot = RefSnapshot::read(self.git)?;
        let head = self.head_ref(&refs_snapshot)?;
        let mut extra = capabilities.clone();
        if let Some(symref) = head.as_ref().and_then(|h| h.symref_target.as_ref()) {
            extra.push(format!("symref=HEAD:{symref}"));
        }

        let refs = self.advertise_from(&refs_snapshot, &[], true, true)?;
        let mut first = true;
        for entry in refs {
            let Some(target) = entry.target else {
                continue;
            };
            let mut line = format!("{target} {}", entry.name).into_bytes();
            if first {
                line.push(0);
                line.extend_from_slice(extra.join(" ").as_bytes());
                first = false;
            }
            wire_io::text(out, &line)?;
            if let Some(peeled) = entry.peeled {
                wire_io::text(out, format!("{peeled} {}^{{}}", entry.name).as_bytes())?;
            }
        }
        if first {
            // An empty repository still has to say who it is, or the client
            // cannot learn the hash algorithm and reports nothing useful.
            let mut line = format!("{} capabilities^{{}}", self.hash_kind().null()).into_bytes();
            line.push(0);
            line.extend_from_slice(extra.join(" ").as_bytes());
            wire_io::text(out, &line)?;
        }
        wire_io::flush(out)?;
        Ok(())
    }

    /// The whole v0 session.
    pub fn serve_v0(&self, input: &mut dyn BufRead, out: &mut dyn Write) -> Result<UploadSummary> {
        self.serve_v0_session(Session::Connected, input, out)
    }

    /// The v0 session, for either [`Session`].
    ///
    /// # What stateless-rpc changes here, and it is one branch
    ///
    /// v0 has no `command=` boundary, so a stateless round is delimited by the
    /// client's flush-pkt after its `have` list. Under
    /// [`Session::StatelessRpc`] that flush ends the **call**: the server
    /// answers `ACK`/`NAK` and returns, and the client opens a new `POST`
    /// carrying its whole `want` list again plus the `have`s it has since
    /// decided to send. Under [`Session::Connected`] the same flush ends only a
    /// *round* and the loop keeps reading, which is what SSH needs.
    ///
    /// Getting this wrong is not a subtle bug: a server that kept reading would
    /// block on a body that has already ended, and the client would sit waiting
    /// for a response that is waiting for it.
    pub fn serve_v0_session(
        &self,
        session: Session,
        input: &mut dyn BufRead,
        out: &mut dyn Write,
    ) -> Result<UploadSummary> {
        let mut summary = UploadSummary {
            protocol: ProtocolVersion::V0,
            // The v0 advertisement IS the `ls-refs`, so it is counted where it
            // is written and not from a constant: a stateless `POST` writes no
            // advertisement, and a summary claiming one would make the metric
            // say a ref list went out when none did.
            ls_refs: usize::from(session.advertises_inline()),
            ..Default::default()
        };
        if session.advertises_inline() {
            self.write_ref_advertisement(out)?;
            out.flush()?;
        }

        // **One budget for the whole v0 session**, not one per section. v0 has
        // no `command=` boundary: the `want` list and every subsequent round of
        // `have`s accumulate into the *same* two vectors, held until the pack is
        // written, so a per-section ceiling would be spent afresh on every round
        // and bound nothing at all. This is the shape a hostile client reaches
        // for — open a connection, send `have <oid>` forever, never send `done`
        // — and it costs the server memory before any repository is touched.
        let mut budget = Budget::new(self.config.limits.request_bytes);
        let mut request = FetchRequest::default();
        let mut peer = PeerCapabilities::default();
        let mut first = true;
        loop {
            let packet = match read_packet(input)? {
                // `git ls-remote` over v0 takes the advertisement and hangs
                // up. That is a completed request, not a truncated one.
                None => return Ok(summary),
                Some(packet) => packet,
            };
            match packet {
                Packet::Flush => break,
                Packet::Data(data) => {
                    budget.spend(data.len(), "v0 want section")?;
                    let mut text: &[u8] = data.strip_suffix(b"\n").unwrap_or(&data);
                    if first {
                        // v0 upload-pack puts the client's capabilities on the
                        // first `want` line, **space-separated after the oid**,
                        // not after a NUL, which is receive-pack's form.
                        // Reading it as a NUL split leaves the capability
                        // string glued to the object id, and the request dies
                        // with "a hash sized 133 hexadecimal characters".
                        if let Some((head, caps)) = split_v0_capabilities(text) {
                            peer = PeerCapabilities::parse(&String::from_utf8_lossy(caps));
                            text = head;
                        }
                        first = false;
                    }
                    self.parse_v0_want_line(text, &mut request)?;
                }
                other => {
                    return Err(Error::protocol(format!(
                        "unexpected {other:?} in a v0 want section"
                    )))
                }
            }
        }

        caps::check_object_format(self.hash_kind(), peer.value("object-format"))?;
        if request.wants.is_empty() {
            return Ok(summary);
        }
        dedupe_oids(&mut request.wants);
        self.check_wants(&request.wants)?;

        // `side-band-64k` and NOT git's older plain `side-band`, which gunnar
        // does not advertise. Reading a capability that is not on offer is the
        // inverse of the drift `caps::Honoured` catches, and here it was not
        // merely dead: the two forms differ in the maximum payload a band may
        // carry — 65515 against 1000 — and `wire_io::sideband` frames at the
        // 64k size unconditionally. So a peer that guessed `side-band` and got
        // it honoured would have been handed packets it must reject. Advertise
        // what you keep, keep what you advertise, read nothing else;
        // `tests/advertisement.rs::every_capability_the_server_reads_is_one_it_advertises`
        // is what says so.
        let sideband = peer.has("side-band-64k");
        request.no_progress = request.no_progress || peer.has("no-progress");
        request.include_tag = request.include_tag || peer.has("include-tag");
        // v0 carries this as a capability rather than as a request line, and it
        // decides whether the pack may contain `ofs-delta` entries at all, so
        // whether pack-copy is usable on this connection. Reading it
        // here rather than in `parse_v0_want_line` is where the other two
        // capability-shaped flags are read.
        request.ofs_delta = request.ofs_delta || peer.has("ofs-delta");
        // The third capability-shaped flag, and it was missing.
        //
        // v0 advertises `thin-pack` (`caps::upload_pack_v0_capabilities`) and
        // `FetchRequest::thin_pack` was only ever set from a **v2** `thin-pack`
        // argument, so the `thin` gate in `send_pack` could not be true on a v0
        // connection: every v0 fetch was served as a whole pack against an
        // advertisement that promised otherwise.
        //
        // MEASURED 2026-08-05, `fx/knut` from the production-shaped corpus,
        // a client cut 40 commits back, the same 431 objects served on both
        // arms, stock git 2.53.0 as the client:
        //
        // | protocol | bytes_out |
        // |---|---|
        // | v0, before | 942 987 |
        // | v2 | 702 138 |
        //
        // 1.343x, 240 849 bytes, for no reason but this line. v2 has been git's
        // default since 2.26 so this is the compatibility path rather than the
        // common one, which is why it went four months unnoticed — and is also
        // why nothing else was going to notice it.
        request.thin_pack = request.thin_pack || peer.has("thin-pack");

        let graph = self.graph();
        let want_commits = self.commit_wants(&graph, &request.wants)?;
        let refs = RefSnapshot::read(self.git)?;
        let shallow_outcome = shallow::compute(&graph, &refs, &want_commits, &request.shallow)?;
        if request.shallow.is_deepening() {
            for id in &shallow_outcome.newly_shallow {
                wire_io::text(out, format!("shallow {id}").as_bytes())?;
            }
            for id in &shallow_outcome.unshallow {
                wire_io::text(out, format!("unshallow {id}").as_bytes())?;
            }
            wire_io::flush(out)?;
            out.flush()?;
        }

        // ── negotiation: two ACK modes, chosen by the client ────────────────
        //
        // Which one is in play is decided by what
        // [`caps::UPLOAD_PACK_V0`] advertised, and that table's comment carries
        // the measurement that settled it. In short:
        //
        // **`multi_ack_detailed`** is what any git since 1.6.6 picks, and the
        // only mode in which a multi-round negotiation over **stateless-rpc**
        // has a correct answer at all. A round answers `ACK <oid> common` per
        // newly-common `have`, `ACK <oid> ready` once the wants are reachable,
        // then `NAK` — git writes that `NAK` at every multi-ack round boundary
        // whether or not it is ready. With **`no-done`** the server then writes
        // the terminal `ACK` and the packfile into the same response instead of
        // waiting for the client to say `done`, which over HTTP saves a whole
        // request.
        //
        // **Single-ACK** is the fallback, for a client that offered neither.
        // It is `b9fee55`'s shape and not a reversion of it: one `ACK <oid>` at
        // the first recognised `have`, nothing at any later flush, and nothing
        // after `done`. That is git's `have_obj.nr` rule, and it is correct on
        // the connected sessions such a client can actually reach.
        let mut haves: Vec<ObjectId> = Vec::new();
        let multi_ack_detailed = peer.has("multi_ack_detailed");
        let no_done = multi_ack_detailed && peer.has("no-done");
        // In single-ACK this is git's `have_obj.nr > 0` — "an ACK has gone out,
        // say nothing more". In multi-ack it is git's `last_hex`, the LAST
        // `have` found common, which is what the terminal `ACK` names.
        let mut acked: Option<ObjectId> = None;
        // git's `sent_ready`: `ACK <oid> ready` is announced once per session.
        // Over stateless-rpc the session is one `POST`, so the latch resets
        // with it and every request that reaches readiness says so — which is
        // also what stock does, its upload-pack being a fresh process per
        // request.
        let mut ready_sent = false;
        // Set when `no-done` let this session write its terminal `ACK` inside
        // the round, so the block after the loop does not write a second one.
        let mut terminal_written = false;
        let mut done = false;
        while !done {
            match read_packet(input)? {
                None => return Ok(summary),
                Some(Packet::Flush) => {
                    summary.fetch_rounds += 1;
                    // ── one line per block, and NOTHING once an ACK has gone
                    //    out ────────────────────────────────────────────────
                    //
                    // The `_ => NAK` this replaces answered every later block
                    // with a `NAK` **after** the ACK had already ended the
                    // negotiation, and git's client stops reading the moment it
                    // sees a plain ACK. The surplus `NAK` therefore sat in the
                    // pipe until the client switched to sideband and read `N`
                    // as a band number:
                    //
                    // ```text
                    // fetch-pack: protocol error: bad band #78
                    // ```
                    //
                    // MEASURED 2026-08-05, `fx/knut`, client cut 40
                    // commits back: three blocks of `have`s, so gunnar wrote
                    // NAK / ACK / NAK and the third line corrupted the stream.
                    // `fx/funnel-core` at five commits back sends one block and
                    // was clean, which is why the small fixture in the interop
                    // suite never showed it — the trigger is a client with
                    // enough history to send a SECOND block.
                    //
                    // git's rule, `upload-pack.c:get_common_commits`, is on
                    // `have_obj.nr` and not on the round: with `multi_ack` off
                    // it writes `NAK` at a flush only while it has recognised
                    // **no** `have` at all, and the single `ACK` goes out inline
                    // at the first one it does recognise. `acked` is gunnar's
                    // `have_obj.nr > 0`.
                    if multi_ack_detailed {
                        // The SAME `negotiate` the v2 `command=fetch` path
                        // runs, over the same per-connection `HaveMemo`, so
                        // `ready` is decided once for both protocol versions
                        // and `Negotiation::ack`'s suppression of redundant
                        // parent-chain ACKs applies here too (one implementation).
                        let negotiation = negotiate(
                            &graph,
                            &request.wants,
                            &haves,
                            /* wait_for_done */ false,
                            &self.haves,
                        )?;
                        for id in &negotiation.ack {
                            wire_io::text(out, format!("ACK {id} common").as_bytes())?;
                        }
                        if let Some(id) = negotiation.common.last() {
                            acked = Some(*id);
                        }
                        let ready = negotiation.ready && acked.is_some();
                        if let (true, false, Some(id)) = (ready, ready_sent, acked) {
                            wire_io::text(out, format!("ACK {id} ready").as_bytes())?;
                            ready_sent = true;
                        }
                        // Unconditional, and NOT the surplus `NAK` the comment
                        // above warns about: that one followed a *terminal*
                        // single-ACK, after which git reads nothing. In
                        // multi-ack the client keeps reading until `done`, and
                        // git writes this `NAK` at every round boundary
                        // (`have_obj.nr == 0 || multi_ack`). A client that does
                        // not get one waits for a round that has ended.
                        wire_io::text(out, b"NAK")?;
                        out.flush()?;
                        if no_done && ready {
                            let id = acked.expect("`ready` is gated on a common `have`");
                            wire_io::text(out, format!("ACK {id}").as_bytes())?;
                            out.flush()?;
                            terminal_written = true;
                            break;
                        }
                    } else {
                        match (acked, self.first_common(&haves)) {
                            (Some(_), _) => {}
                            (None, Some(id)) => {
                                acked = Some(id);
                                wire_io::text(out, format!("ACK {id}").as_bytes())?;
                                out.flush()?;
                            }
                            (None, None) => {
                                wire_io::text(out, b"NAK")?;
                                out.flush()?;
                            }
                        }
                    }
                    if session.is_stateless() {
                        // The round IS the request. See the doc comment: the
                        // client's next `have` list arrives on a new
                        // connection, carrying its `want`s again.
                        summary.common = haves
                            .iter()
                            .filter(|id| has_for_negotiation(self.git, id))
                            .count();
                        return Ok(summary);
                    }
                }
                Some(Packet::Data(data)) => {
                    budget.spend(data.len(), "v0 negotiation")?;
                    let text = data.strip_suffix(b"\n").unwrap_or(&data);
                    if text == b"done" {
                        done = true;
                    } else if let Some(rest) = text.strip_prefix(b"have ") {
                        haves.push(parse_oid(&String::from_utf8_lossy(rest), self.hash_kind())?);
                    }
                }
                Some(other) => {
                    return Err(Error::protocol(format!(
                        "unexpected {other:?} during v0 negotiation"
                    )))
                }
            }
        }

        // ── after `done`, a single-ACK server that has already ACKed says
        //    NOTHING ─────────────────────────────────────────────────────────
        //
        // This used to be `acked.or_else(first_common)` and wrote an `ACK`
        // either way, so a negotiation that found a common commit sent the
        // **same ACK twice** — once from the flush arm above, once here. The
        // second one lands after the client has switched its reader into
        // sideband mode, which demultiplexes `A` as a band number:
        //
        // ```text
        // fetch-pack: protocol error: bad band #65
        // fatal: protocol error: bad pack header
        // ```
        //
        // MEASURED 2026-08-05 against stock git 2.53.0 over real SSH,
        // `fx/funnel-core` from the production-shaped corpus, client cut five
        // commits back:
        //
        // | fetch | v0 | v2 |
        // |---|---|---|
        // | full clone | 28 objects, OK | OK |
        // | **incremental** | **`bad band #65`, 0 objects** | 28 objects, OK |
        //
        // So **every** v0 fetch that shares history with the server was
        // corrupt, and only a clone — which sends no `have` and gets no ACK —
        // came through. `GIT_TRACE_PACKET` shows the duplicate plainly:
        // `fetch< ACK 36d2cb41…` followed by `sideband< ACK 36d2cb41…`.
        //
        // git's own rule, `upload-pack.c:get_common_commits`: with `multi_ack`
        // off, `done` returns without writing when any `have` was recognised,
        // and writes `NAK` only when none was. Its client is the other half of
        // the same contract — `fetch-pack.c:find_common` zeroes `flushes` on a
        // plain `ACK`, so `while (flushes || multi_ack)` after `done` reads
        // **no** lines at all and the next byte it expects is the pack.
        //
        // The `first_common` fallback is gone with it: it could only fire when
        // the flush arm had already run and decided otherwise, and reaching a
        // different verdict about the same `haves` two lines apart is the shape
        // of the bug rather than a safety net.
        //
        // In **multi-ack** the rule is the other one git writes at `done`: an
        // `ACK <last_hex>` when anything was common, `NAK` when nothing was.
        // The asymmetry is not an inconsistency — in single-ACK the client
        // stopped reading at the earlier terminal `ACK`, and in multi-ack it is
        // still reading and this line is the terminal one. `terminal_written`
        // is the `no-done` case, where it has already gone out with the pack.
        if !terminal_written {
            if multi_ack_detailed {
                match acked {
                    Some(id) => wire_io::text(out, format!("ACK {id}").as_bytes())?,
                    None => wire_io::text(out, b"NAK")?,
                }
            } else if acked.is_none() {
                wire_io::text(out, b"NAK")?;
            }
            out.flush()?;
        }

        request.haves = haves;
        let common: Vec<ObjectId> = request
            .haves
            .iter()
            .filter(|id| graph.contains_have(id))
            .copied()
            .collect();
        summary.fetch_rounds += 1;
        summary.common = common.len();
        let written = self.send_pack(
            &graph,
            &refs,
            &request.wants,
            &common,
            &request,
            &shallow_outcome,
            out,
            sideband,
        )?;
        summary.objects_sent = written.written.num_objects;
        summary.objects_selected = written.objects_selected;
        summary.objects_copied_from_pack = written.written.objects_copied_from_pack;
        summary.objects_recompressed = written.written.objects_recompressed;
        summary.answered_from_bitmap = written.answered_from_bitmap;
        summary.thin_bases_offered = written.written.thin_bases_offered;
        summary.thin_bases_vetoed = written.written.thin_bases_vetoed;
        summary.closure_bases_declined = written.written.closure_bases_declined;
        if sideband {
            // A sidebanded v0 response ends with an **unframed** flush-pkt.
            // Without it the client sits waiting for another band-1 line and
            // reports "unexpected disconnect while reading sideband packet"
            // when the pipe closes, a message that points at the transport
            // and not at the missing four bytes. (Found by the real-git
            // interop suite, 2026-08-03; the raw, non-sidebanded form needs no
            // terminator because the pack's own trailer ends it.)
            wire_io::flush(out)?;
        }
        out.flush()?;
        Ok(summary)
    }

    fn parse_v0_want_line(&self, text: &[u8], request: &mut FetchRequest) -> Result<()> {
        let line = String::from_utf8_lossy(text).into_owned();
        let (name, value) = split_argument(&line);
        let missing = || Error::protocol(format!("v0 line {line:?} is missing its value"));
        match name {
            "want" => request
                .wants
                .push(parse_oid(value.ok_or_else(missing)?, self.hash_kind())?),
            "shallow" => request
                .shallow
                .client_shallow
                .push(parse_oid(value.ok_or_else(missing)?, self.hash_kind())?),
            "deepen" => {
                request.shallow.depth = Some(
                    value
                        .ok_or_else(missing)?
                        .parse::<usize>()
                        .map_err(|_| Error::protocol(format!("deepen wants a number: {line:?}")))?,
                )
            }
            "deepen-since" => {
                request.shallow.since =
                    Some(value.ok_or_else(missing)?.parse::<i64>().map_err(|_| {
                        Error::protocol(format!("deepen-since wants unix seconds: {line:?}"))
                    })?)
            }
            "deepen-not" => request
                .shallow
                .not
                .push(value.ok_or_else(missing)?.to_owned()),
            "deepen-relative" => request.shallow.relative = true,
            "filter" => {
                if !self.config.offered.filter {
                    return Err(Error::UnsupportedFilter {
                        spec: value.unwrap_or_default().to_owned(),
                        reason: "this server does not offer the filter capability",
                    });
                }
                request.filter = Some(Filter::parse(value.ok_or_else(missing)?)?)
            }
            "" => {}
            other => {
                return Err(Error::protocol(format!(
                    "unknown v0 request line {other:?}"
                )))
            }
        }
        Ok(())
    }

    fn first_common(&self, haves: &[ObjectId]) -> Option<ObjectId> {
        haves
            .iter()
            .find(|id| has_for_negotiation(self.git, id))
            .copied()
    }

    // ── shared ───────────────────────────────────────────────────────────

    /// Report a fatal condition to the client in whichever way this point of
    /// the protocol allows: an `ERR` line before a pack has started, band 3
    /// once the sideband is open.
    pub fn report_error(&self, out: &mut dyn Write, sideband: bool, message: &str) -> Result<()> {
        if sideband {
            crate::wire_io::sideband_error(out, message)
        } else {
            err_packet(out, message)
        }
    }

    /// `HEAD`, which is **not a row in `refs()`**.
    ///
    /// `locked_on_target.md`'s `HEAD`-is-not-a-row question, settled from the gix arm's own words:
    /// `all()` walks `refs/` and *"deliberately excludes the pseudo-refs such
    /// as `HEAD`"*, which is the contract every backend honours. So it has its
    /// own accessor pair on `GitServe` and this asks the store, not the
    /// snapshot. Its target, though, is resolved **against the snapshot** — the
    /// same one the advertisement is built from — so `HEAD` cannot be
    /// advertised pointing at an oid no other line in the same advertisement
    /// agrees with.
    fn head_ref(&self, refs: &RefSnapshot) -> Result<Option<AdvertisedRef>> {
        let Some(row) = self.git.head().map_err(crate::graph::backend)? else {
            return Ok(None);
        };
        let symref_target = row.symref_target.clone();
        let resolved = match &row.oid {
            Some(bytes) => Some(crate::graph::oid(bytes)?),
            None => match &symref_target {
                Some(name) => refs.resolve(name)?,
                None => None,
            },
        };
        Ok(Some(AdvertisedRef {
            name: gunnar_store::HEAD.to_owned(),
            target: resolved,
            peeled: None,
            symref_target,
        }))
    }

    fn collect_refs(
        &self,
        prefixes: &[String],
        peel: bool,
        symrefs: bool,
    ) -> Result<Vec<AdvertisedRef>> {
        let refs = RefSnapshot::read(self.git)?;
        self.advertise_from(&refs, prefixes, peel, symrefs)
    }

    /// [`UploadPack::collect_refs`] against a snapshot the caller already took.
    fn advertise_from(
        &self,
        refs: &RefSnapshot,
        prefixes: &[String],
        peel: bool,
        symrefs: bool,
    ) -> Result<Vec<AdvertisedRef>> {
        let graph = self.graph();
        let mut out = Vec::new();

        // HEAD first, exactly as `git-upload-pack` orders it.
        if prefixes.is_empty() || prefixes.iter().any(|p| "HEAD".starts_with(p.as_str())) {
            if let Some(mut head) = self.head_ref(refs)? {
                if !symrefs {
                    head.symref_target = None;
                }
                out.push(head);
            }
        }

        for row in refs.rows() {
            if !prefixes.is_empty() && !prefixes.iter().any(|p| row.name.starts_with(p.as_str())) {
                continue;
            }
            let resolved = refs.resolve_row(row)?;
            // The store's own peel column when it has one, and only then the
            // walk. `RefRow::peeled` is *"for an annotated tag, the commit it
            // peels to"*; `None` means the backend does not say, never that
            // the tag peels to nothing, so the fallback is a real peel and not
            // an unpeeled advertisement.
            let peeled = match (peel, resolved) {
                (true, Some(id)) => match RefSnapshot::peeled(row)? {
                    Some(known) => Some(known),
                    None if graph.kind(&id)? == Some(object::Kind::Tag) => {
                        Some(graph.peel(&id)?.id)
                    }
                    None => None,
                },
                _ => None,
            };
            out.push(AdvertisedRef {
                name: row.name.clone(),
                target: resolved,
                peeled,
                symref_target: symrefs.then(|| row.symref_target.clone()).flatten(),
            });
        }
        Ok(out)
    }

    /// Every object id a client is allowed to `want`.
    fn advertised_oids(&self) -> Result<HashSet<ObjectId>> {
        let mut out = HashSet::new();
        for entry in self.collect_refs(&[], true, false)? {
            if let Some(id) = entry.target {
                out.insert(id);
            }
            if let Some(id) = entry.peeled {
                out.insert(id);
            }
        }
        Ok(out)
    }

    /// **Under a policy, "there is no such object" and "there is, and you may
    /// not have it" are ONE answer.**
    ///
    /// They were two. `wants_exist` ran first and answered `ObjectMissing`, and
    /// only a want that DID exist went on to be measured against the
    /// advertisement and answered `WantNotAdvertised`. The difference is an
    /// oracle, and the thing it answers is precisely what the policy exists to
    /// hide: `WantNotAdvertised`'s own doc says serving arbitrary ids "leaks
    /// objects that are not reachable from any reference, which is how
    /// 'deleted' branches stay fetchable" — and telling a caller that an oid IS
    /// here, one `want` at a time, leaks the same fact one bit at a time. An
    /// oid from a force-pushed branch, a deleted branch, or somebody else's
    /// paste can be tested against a public repository for free.
    ///
    /// This is the same merge `gunnar-ssh` makes between "no such repository"
    /// and "you have no grant": one string on the wire, the difference kept in
    /// the server's log.
    ///
    /// `wants_exist` is not called at all under a policy now, rather than
    /// called and its answer laundered. Existence is implied by the check that
    /// remains: an advertised oid is a reference target, and a reachable one
    /// was just walked to. A want that passes is servable by construction, and
    /// a reference pointing at an object the store does not hold is repository
    /// corruption that the pack build reports as the server's problem — which
    /// is what it is.
    ///
    /// [`WantPolicy::Any`] keeps the plain existence check, and it is not an
    /// oracle there: under `Any` everything that exists is servable, so
    /// "missing" is the whole truth and there is nothing to hide behind it.
    fn check_wants(&self, wants: &[ObjectId]) -> Result<()> {
        if wants.is_empty() || self.config.want_policy == WantPolicy::Any {
            return self.wants_exist(wants);
        }
        let advertised = self.advertised_oids()?;
        // The reachable closure is built at most once, and only if some want
        // was not itself advertised.
        let mut reachable: Option<HashSet<ObjectId>> = None;
        for want in wants {
            if advertised.contains(want) {
                continue;
            }
            if self.config.want_policy == WantPolicy::AdvertisedTips {
                return Err(Error::WantNotAdvertised { oid: *want });
            }
            let closure = match &reachable {
                Some(closure) => closure,
                None => reachable.insert(self.reachable_from_refs()?),
            };
            if !closure.contains(want) {
                return Err(Error::WantNotAdvertised { oid: *want });
            }
        }
        Ok(())
    }

    fn wants_exist(&self, wants: &[ObjectId]) -> Result<()> {
        for want in wants {
            if !self
                .git
                .has(want.as_bytes())
                .map_err(crate::graph::backend)?
            {
                return Err(Error::missing(*want, "a want"));
            }
        }
        Ok(())
    }

    /// Every object reachable from any reference.
    ///
    /// One repository-sized set, and only one. This used to be
    /// `select_objects(…).objects.into_iter().collect()`, which holds a `Vec` of
    /// every id in the repository *while* the `HashSet` built out of it fills —
    /// ~820 MB of peak at ten million sha256 ids where ~500 MB is needed. See
    /// [`reachable_closure`].
    fn reachable_from_refs(&self) -> Result<HashSet<ObjectId>> {
        let tips: Vec<ObjectId> = self
            .collect_refs(&[], false, false)?
            .into_iter()
            .filter_map(|entry| entry.target)
            .collect();
        let graph = self.graph();
        Ok(reachable_closure(&graph, &tips)?.into_inner())
    }

    /// The wants that peel to commits, deduplicated, in the order the client
    /// named them.
    ///
    /// # Both sets are here because the input is the client's
    ///
    /// `wants` is whatever the peer sent and is bounded only by what the
    /// request reader is willing to accept, so every per-want cost in here is a
    /// cost the peer chooses. Two of them used to be superlinear:
    ///
    /// * `asked` stops a repeated `want` being **peeled twice**. `peel` is an
    ///   object read, so a client naming the same tag ten thousand times used
    ///   to buy ten thousand store reads for one answer. This is the half a
    ///   counting store can see, and `tests/want_cost.rs` holds it.
    /// * `seen` replaces a `Vec::contains` over the answer so far, which was a
    ///   linear scan run once per want, so `O(wants²)` `ObjectId` comparisons.
    ///   Ten thousand distinct wants — an ordinary mirror clone — was fifty
    ///   million comparisons.
    ///
    /// The second half has **no deterministic instrument**, and that is worth
    /// stating rather than papering over: both spellings return identical
    /// output and touch the store identically, so the only difference is time,
    /// and this crate's cost tests deliberately count touches rather than
    /// timing anything (see `tests/connectivity_cost.rs`). It is fixed here and
    /// guarded only by the half above.
    fn commit_wants(&self, graph: &Graph<'_>, wants: &[ObjectId]) -> Result<Vec<ObjectId>> {
        let mut asked: HashSet<ObjectId> = HashSet::new();
        let mut seen: HashSet<ObjectId> = HashSet::new();
        let mut out = Vec::new();
        for want in wants {
            if !asked.insert(*want) {
                continue;
            }
            let peeled = graph.peel(want)?;
            if peeled.kind == object::Kind::Commit && seen.insert(peeled.id) {
                out.push(peeled.id);
            }
        }
        Ok(out)
    }

    #[allow(clippy::too_many_arguments)]
    fn send_pack(
        &self,
        graph: &Graph<'_>,
        refs: &RefSnapshot,
        wants: &[ObjectId],
        common: &[ObjectId],
        request: &FetchRequest,
        shallow_outcome: &ShallowOutcome,
        out: &mut dyn Write,
        sideband: bool,
    ) -> Result<SentPack> {
        // The one place either protocol version opens a sideband, and therefore
        // the one place that has to record it. Past this line a failure's reason
        // rides band 3; before it, an `ERR` pkt-line. See
        // [`UploadPack::sideband_open`].
        self.sideband_open
            .store(sideband, std::sync::atomic::Ordering::Relaxed);
        // The serving defaults live in one place, `SelectOptions::for_serving`,
        // and the request-specific fields are laid over them here. Building the
        // whole struct inline is what let a test measure a configuration the
        // server never runs in; see `Selection::answered_from_bitmap`.
        let options = SelectOptions {
            boundary: shallow_outcome.boundary.clone(),
            have_boundary: shallow_outcome.client_boundary.clone(),
            filter: request.filter.as_ref(),
            restrict: request
                .shallow
                .is_deepening()
                .then_some(&shallow_outcome.include),
            missing: Missing::Fail,
            ..SelectOptions::for_serving(&self.config.limits)
        };

        // The wants, plus the parents of everything this fetch just promised to
        // unshallow. Those parents are not a second want: they come from the
        // server's own walk of its own graph, already filtered by `restrict` to
        // what the deepening decided, so they widen the pack without widening
        // what the client is allowed to name. `check_wants` stays the gate on
        // the client's side of that line.
        let mut tips = wants.to_vec();
        if !shallow_outcome.unshallow_parents.is_empty() {
            // A set, not `tips.contains`: `wants` is the client's list.
            let mut named: HashSet<ObjectId> = tips.iter().copied().collect();
            for id in &shallow_outcome.unshallow_parents {
                if named.insert(*id) {
                    tips.push(*id);
                }
            }
        }
        let mut selection = {
            let _s = crate::stage::enter("select");
            select_objects(graph, &tips, common, &options)?
        };

        if request.include_tag {
            let _s = crate::stage::enter("include_tag");
            self.add_reachable_tags(graph, refs, &mut selection.objects)?;
        }

        let show_progress = sideband && !request.no_progress;
        progress(
            out,
            show_progress,
            &format!(
                "gunnar: packing {} objects from {} commits\n",
                selection.objects.len(),
                selection.commits.len()
            ),
        )?;

        // ── Copy, do not re-deflate ─────────────────────────────────────────
        //
        // `ObjectStore::packed()` is `None` for a store with no packs to copy
        // from: `mem`, and any cold tier that keeps inflated content per oid
        // instead. Those fall through to the whole-object writer they have
        // always used.
        //
        // `ofs_delta` is the client's own answer about whether it can decode
        // `OBJ_OFS_DELTA`, and until 2026-08-03 it gated the copy path
        // entirely: a client that did not advertise it got the pre-pack-copy
        // server, no delta reuse and no entry copying at all. MEASURED on the
        // churned fixture, that cost such a client **18.4x the bytes** — for
        // nothing, because `OBJ_REF_DELTA` needs no capability and carries the
        // identical delta bytes. The capability now decides how a copied delta
        // *names its base* (`WriteOptions::ofs_delta`), not whether copying
        // happens.
        // ── the unselected delta base ───────────────────────────────────────
        //
        // `WriteOptions::base_closure` lets `write_pack_copy` add a delta's
        // unselected base rather than expand the delta. MEASURED over the
        // corpus's single-branch clones: −2.82% wire bytes, −51.9% CPU.
        //
        // Three conditions, and each one is a different kind of reason:
        //
        //  1. **the negotiation found nothing in common.** This is the one that
        //     decides most of the value, and it is measured rather than argued.
        //     On a *clone*, an unselected base is an object nobody is getting,
        //     and shipping it buys back a whole exploded object. On a
        //     *negotiated* fetch it is usually an object the client already
        //     holds — it sits below the cut — so shipping it is the duplication
        //     thin packs exist to avoid. MEASURED 2026-08-05 on
        //     `tests/pack_copy.rs`'s churned fixture, one fetch 8 commits deep
        //     from a client that withheld `thin-pack`: **31 360 → 55 513 bytes,
        //     +77%**, adding 57 bases to recover 24 deltas. A fetch's answer to
        //     that delta is the thin voucher, not this.
        //  2. **a filtered request** turns it off. `--filter=blob:limit=1k` says
        //     which objects the client will accept, an unselected base is by
        //     construction an object the selection did not choose, and a small
        //     blob's delta base is usually a large one — so a closure under a
        //     filter ships exactly the object the filter excluded, silently.
        //  3. **the server's object ceiling** bounds how large a base may be. It
        //     is `Some` on every request, so it cannot be folded into the same
        //     answer as the other two; doing that turned the closure off
        //     everywhere and `tests/base_closure.rs` is what caught it.
        // ⚠ **THE THREE CONDITIONS ABOVE NO LONGER CROSS THE SEAM, and one of
        // them is a real loss.** `WriteOptions::base_closure` was a knob on
        // gunnar's own pack writer; the engine owns the format now and
        // `git_storage_trait::Caps` is deliberately the CLIENT's capabilities
        // (`thin`, `ofs_delta`) and not the server's policy, which is the right
        // line to draw. Conditions 1 and 2 survive by construction — a non-thin
        // pack must be closed over its bases whatever anyone prefers, and
        // `thin` below already carries "the client said it can take an open
        // one". Condition **3** does not: there is no longer a way to tell an
        // engine how large an unselected delta base it may pull in, so
        // `Limits::object_bytes` bounds what this end *selects* and no longer
        // bounds what the engine *adds*. The measurements above are kept
        // because they are the record of why the knob existed; the knob is
        // gone. Recorded here rather than silently dropped.
        // ── May this answer be a THIN pack? ─────────────────────────────────
        //
        // Pack-copy copies 100% of a clone's entries and only 97.6% of a
        // fetch's, and the missing 2.4% is not a rounding error — it is every
        // object whose delta base sits on the client's side of the negotiation
        // cut, each one inflated and re-deflated **whole**. MEASURED
        // 2026-08-04 on a 60-file churn fetched 8 commits deep: 24 of 40
        // objects expanded, 3.98x stock git's thin pack of the same objects.
        // Naming those bases instead of expanding them is the fix, and it
        // computes nothing: the deltas are the pushing client's, already in
        // gunnar's pack. `gunnar-delta` stays off this path.
        //
        // Three conditions, and every one of them is a correctness condition
        // rather than a tuning knob, because a thin pack whose bases the
        // receiver lacks is not a smaller answer but a **failed fetch**:
        //
        //  1. the client advertised `thin-pack`. Nothing else may be read as
        //     consent to receive one.
        //  2. the negotiation found something in common. With no `common` the
        //     request is a clone, the client holds nothing, and there is
        //     nothing a base could safely point at. `selection.client_has` is
        //     then **empty** — since 2026-08-05 it carries only what the
        //     receiver already held, never the objects being sent — so this
        //     would be a no-op. Saying it here means the clone path is provably
        //     untouched rather than incidentally so.
        //  3. the client is not shallow. A `have X` from a shallow client is
        //     not proof it holds `X`'s ancestry, which is the whole reason
        //     `ShallowOutcome::client_boundary` exists; the walk honours that
        //     boundary, and this declines to bet the fetch on it.
        //  4. the request carries **no filter**, and this one became live on
        //     2026-08-11. `have` is a set of commit *tips*, and every consumer
        //     of it — `selection.client_has` here, `client_bases` in the znippy
        //     engine — turns it into what the receiver holds by closing over it.
        //     For a partial clone that closure is a **lie in the dangerous
        //     direction**: a `blob:none` client's tips reach every blob under
        //     them and it was deliberately sent none of them, so a base vouched
        //     for this way can be an object it has never seen and its
        //     `index-pack --fix-thin` dies with `pack has N unresolved deltas`.
        //     It cost nothing while no engine read `have`; znippy's does now.
        //
        // Past the gate every individual base is still checked against
        // `selection.client_has`, and one that does not answer costs its
        // object the delta and nothing else.
        let thin = request.thin_pack
            && !common.is_empty()
            && request.filter.is_none()
            && shallow_outcome.boundary.is_empty()
            && shallow_outcome.client_boundary.is_empty();

        let outcome = {
            let _s = crate::stage::enter("pack_write");
            let mut pack = PackOutput::new(out, sideband);
            // ── the engine emits; this streams ────────────────────────────
            //
            // What stood here was gix: `write_pack_copy_thin` drove
            // `iter_from_counts`, and its `None` arm inflated every object
            // through gix's object traits. Both are gone with the packfile
            // engine, and the seam that chose between them (`ObjectStore::packed`)
            // went with them, because its return type named the pack writer.
            //
            // The engine owns the format now. It closes the request over its
            // delta bases, orders bases before deltas, re-encodes each entry
            // header and copies every payload byte for byte. This end knows
            // nothing about entry headers, which is the point.
            //
            // `None` is refused rather than improvised around: a whole-object
            // fallback produces a pack that passes `index-pack --strict` and
            // `fsck` while sending a measured 18.4x the wire bytes, which is
            // the failure nobody notices.
            // ── ✅ DECIDED 2026-08-10 — the first parameter is an EXACT SET ──
            //
            // The ruling went the way this call site needed. The
            // contract's first parameter is now named `objects` and reads
            // *"emit a packfile containing exactly `objects`"*. **The engine
            // must not compute a closure.** `GitServe::select` owns the
            // close-over-tips half and keeps it, so there is still exactly one
            // way to ask each question.
            //
            // This was not a rename. The znippy arm's `emit_pack` **was closing
            // over its input** — it opened with `self.select(objects, have)?`
            // and emitted that — so this call site was, as written, an
            // over-send waiting for the first engine to be wired up:
            //
            // | request | closure of the selection | the selection |
            // |---|---|---|
            // | clone, no filter | identical | identical |
            // | `--filter=blob:none` | **adds back every excluded blob** | correct |
            // | `--depth=N` | **adds back the grafted-away ancestry** | correct |
            //
            // An over-send passes `index-pack --strict` and `fsck`. It is the
            // failure nobody notices, and it is the failure class this whole
            // seam exists to abolish. It is now pinned on the engine side by
            // `znippy-plugin-git`'s
            // `emit_pack_emits_exactly_the_set_it_is_given_and_never_its_closure`,
            // which asserts an object **count** and was seen RED at 4 against
            // an expected 1.
            //
            // **`selected` below is genuinely the exact set to emit**, and that
            // is checked rather than assumed: `select_objects` applies the
            // partial-clone `filter` at insertion (`Graph::keeps_blob` /
            // `keeps_tree`), the shallow `boundary` and `have_boundary`, and a
            // deepening's `restrict`; `include_tag` then adds to
            // `selection.objects` above. Every one of those decisions is a
            // deliberate *exclusion*, and every one of them is undone by a
            // closure.
            //
            // `have` is the negotiated common **tips**, and that half is right
            // as frozen: it is what lets an engine derive for itself which
            // bases a thin delta may name — never to derive membership. Tips
            // and not the walk's exclusion closure, deliberately — the closure
            // is repository-sized, the engine can rebuild it from its own
            // index, and `ReachSet::client_has` is documented as *what the
            // receiver held BEFORE this transfer*, which is exactly what a set
            // of pre-transfer tips names and what `Selection::client_has` had
            // to be narrowed to on 2026-08-05.
            //
            // ✅ **The znippy arm reads `have` since 2026-08-11**, and this is
            // what it reads it for: with `caps.thin` set, a delta whose base
            // falls outside the request but inside the closure of these tips
            // goes out as a `REF_DELTA` naming that base, carrying the stored
            // delta stream unchanged, instead of being rebuilt whole. `vouched`
            // below is that set's size, and `recompressed` falling while it is
            // non-zero is the pass working.
            //
            // ⚠ **Do not assert `stats.objects == selected.len()` here.** The
            // contract's word is "exactly", and it means *no reachability
            // closure*; it does not mean the counts must match. A packfile's
            // `OFS_DELTA` names its base by a backwards distance **within the
            // pack**, so an engine copying stored entries has to pull in any
            // delta base that fell outside the selection or emit something
            // unparseable. That is a format requirement, not a decision about
            // what the client asked for, and `caps.thin` is the allowance to
            // skip it. `stats.objects` is therefore `selected.len()` plus the
            // bases the format forced, and it is logged rather than asserted.
            //
            // There is no `Ok(None)` and no fallback. An engine that cannot
            // emit says so as an error: a whole-object fallback produces a pack
            // that passes `fsck` while sending a measured **18.4x** the wire
            // bytes.
            let caps = gunnar_store::git_storage_trait::Caps {
                ofs_delta: request.ofs_delta,
                thin,
            };
            let selected: Vec<&[u8]> = selection.objects.iter().map(|id| id.as_bytes()).collect();
            let vouched: Vec<&[u8]> = common.iter().map(|id| id.as_bytes()).collect();
            // The trailer IS the pack's identity and it is the last bytes of
            // the stream, so it is read off what was emitted rather than
            // recomputed — the same property the buffered writer had, kept
            // through a streaming emitter by watching the tail go past.
            // `PackStats` deliberately does not carry a pack id, and it should
            // not have to: a receipt that *stated* the id would be a second
            // authority on it, and this one cannot disagree with the bytes.
            let mut tapped = TrailerTap::new(&mut pack, self.hash_kind.len_in_bytes());
            let stats = self
                .git
                .emit_pack(&selected, &vouched, &caps, &mut tapped)
                .map_err(crate::graph::backend)?;
            let pack_id = crate::graph::oid(&tapped.trailer()?)?;
            let outcome = crate::pack::WriteOutcome {
                num_objects: stats.objects as u32,
                pack_id,
                bytes_written: stats.bytes,
                // pack-copy's applied-output assertion, and it is READ BACK off
                // what the engine reported rather than asserted here. A byte
                // count and a wall clock cannot tell a pack-copy from a
                // re-deflate; these two can, and the day an engine starts
                // re-deflating, `recompressed` climbs instead of a test staying
                // green.
                objects_copied_from_pack: stats.copied as usize,
                objects_recompressed: stats.recompressed as usize,
                // Deltas are the pushing client's, preserved rather than
                // computed, so nothing here deltified anything.
                objects_deltified: 0,
                // Still zero, and now for a smaller reason: the narrowing pass
                // exists but `PackStats` does not carry its counts, so there is
                // nothing honest to put here. `recompressed` is the observable
                // — it falls as bases are offered — and inventing a number for
                // these three would be worse than reporting none.
                thin_bases_offered: 0,
                closure_bases_declined: 0,
                thin_bases_vetoed: 0,
            };
            pack.flush()?;
            outcome
        };

        tracing::debug!(
            // `gunnar::git`, not the module-path default. This line was on
            // `gunnar_wire::upload_pack` and was the LAST entry the observability
            // suite's `known` remainder held for this crate: it is the only line
            // the whole clone path emits, and `RUST_LOG=gunnar::git=debug` — the
            // filter this repository documents for looking at a transfer — did
            // not select it. An operator turning the git log up to find out what
            // a clone did got the push path and nothing else.
            target: crate::receive_pack::GIT_TARGET,
            event = "git.upload_pack.sent",
            objects = outcome.num_objects,
            // The other half of `objects`, and the reason it is here:
            // `objects` is what the ENGINE emitted and this is what
            // `select` chose. On a negotiated fetch over a packed store
            // the first was measured at 8.72x the second, and no line
            // anywhere said so.
            selected = selection.objects.len(),
            bytes = outcome.bytes_written,
            copied = outcome.objects_copied_from_pack,
            recompressed = outcome.objects_recompressed,
            // How much this transfer was allowed to point a delta at. The two
            // selection paths reach it differently — an id set on the walk, a
            // bitmap otherwise — and this is where a regression in either would
            // show up as `recompressed` climbing beside it.
            vouched = selection.client_has.len(),
            // WHICH path produced that vouch. Without it every line above is
            // ambiguous between two paths that measurably disagree; see
            // [`UploadSummary::answered_from_bitmap`].
            bitmap = selection.answered_from_bitmap,
            thin,
            sideband,
            "served a packfile"
        );
        Ok(SentPack {
            written: outcome,
            answered_from_bitmap: selection.answered_from_bitmap,
            objects_selected: selection.objects.len(),
        })
    }

    /// `include-tag`: add annotated tags whose target is already in the pack.
    fn add_reachable_tags(
        &self,
        graph: &Graph<'_>,
        refs: &RefSnapshot,
        objects: &mut Vec<ObjectId>,
    ) -> Result<()> {
        let mut selected: HashSet<ObjectId> = objects.iter().copied().collect();
        let mut extra = Vec::new();
        for row in refs.with_prefix(Some("refs/tags/")) {
            // A symbolic reference under `refs/tags/` is not a tag; the old
            // `Target::Object(id) else { continue }` said exactly this and this
            // says it against the row.
            let Some(bytes) = &row.oid else { continue };
            let id = crate::graph::oid(bytes)?;
            if selected.contains(&id) || !graph.contains(&id)? {
                continue;
            }
            if graph.kind(&id)? != Some(object::Kind::Tag) {
                continue;
            }
            let peeled = graph.peel(&id)?;
            if selected.contains(&peeled.id) {
                for tag in peeled.tags {
                    // `selected` absorbs what `extra` used to be scanned for
                    // — one set, one membership test, and a tag chain
                    // that appears under two references is still added once.
                    if selected.insert(tag) {
                        extra.push(tag);
                    }
                }
            }
        }
        objects.extend(extra);
        Ok(())
    }
}

/// How many `have` facts one connection may remember, derived from the request
/// ceiling rather than chosen separately (one implementation).
///
/// A `have` costs the client a whole pkt-line — four header bytes, `have `, and
/// a hex object id, so 45 bytes at sha1 and 77 at sha256. Dividing the request
/// ceiling by the *cheapest* of those is deliberate: it makes the memo's ceiling
/// the number of distinct `have`s a client could send in one ceiling's worth of
/// bytes, which is the honest ratio between what the peer spends and what the
/// server remembers.
fn memo_capacity(limits: &Limits) -> Option<usize> {
    /// `0004` + `have ` + 40 hex digits + `\n`.
    const CHEAPEST_HAVE_LINE: usize = 4 + 5 + 40 + 1;
    limits.request_bytes.map(|bytes| bytes / CHEAPEST_HAVE_LINE)
}

struct V2Command {
    name: String,
    capabilities: PeerCapabilities,
    arguments: Vec<Vec<u8>>,
}

/// Split a v0 `want <oid> <capabilities>` line into its command and its
/// capability string.
///
/// A NUL is honoured too, because that is what some older clients and every
/// `receive-pack` client uses, and accepting both costs one branch.
fn split_v0_capabilities(line: &[u8]) -> Option<(&[u8], &[u8])> {
    if let Some(nul) = line.iter().position(|&b| b == 0) {
        return Some((&line[..nul], &line[nul + 1..]));
    }
    let rest = line.strip_prefix(b"want ")?;
    let space = rest.iter().position(|&b| b == b' ')?;
    Some((&line[..b"want ".len() + space], &rest[space + 1..]))
}

/// Drop repeated ids, keeping the first occurrence of each.
///
/// # Once, here, rather than in each consumer
///
/// The want list is the peer's, and it is bounded only by what the request
/// reader accepts, so **every** per-line cost in it is a cost the peer chooses.
/// A repeated `want` used to be paid for four separate times over: an existence
/// probe in `wants_exist`, a `peel` in `commit_wants`, another `peel` as a tip
/// of the selection walk, and a pass in `negotiate`. Deduplicating in each of
/// those is four places to keep agreeing; deduplicating the list once, before
/// anybody reads it, is one (one implementation).
///
/// Order is preserved because it is observable: the selection walk emits
/// objects in tip order, so sorting here would quietly reorder packs.
fn dedupe_oids(ids: &mut Vec<ObjectId>) {
    if ids.len() < 2 {
        return;
    }
    let mut seen: HashSet<ObjectId> = HashSet::with_capacity(ids.len());
    ids.retain(|id| seen.insert(*id));
}

/// Split `name value` at the first space; a line with no space is all name.
fn split_argument(line: &str) -> (&str, Option<&str>) {
    match line.split_once(' ') {
        Some((name, value)) => (name, Some(value)),
        None => (line, None),
    }
}

fn parse_oid(hex: &str, hash_kind: Kind) -> Result<ObjectId> {
    let id = ObjectId::from_hex(hex.trim().as_bytes())
        .map_err(|err| Error::protocol(format!("{hex:?} is not an object id: {err}")))?;
    if id.kind() != hash_kind {
        return Err(Error::ObjectFormatMismatch {
            ours: caps::object_format_name(hash_kind),
            theirs: caps::object_format_name(id.kind()).to_owned(),
        });
    }
    Ok(id)
}

/// A pass-through writer that remembers the **last `n` bytes** it saw.
///
/// One use: a packfile's trailer is its identity and it is the last
/// `hash_kind.len_in_bytes()` bytes of the stream. `GitServe::emit_pack`
/// streams straight to the client and its
/// [`PackStats`](gunnar_store::git_storage_trait::PackStats) deliberately does
/// not carry a pack id — a receipt that *stated* the id would be a second
/// authority on it, and two authorities can disagree (one implementation). So
/// the id is read off the bytes that actually went out, which is what the old
/// buffered writer did and the property worth keeping.
///
/// It costs one `n`-byte array and one `copy_within` per write call, and nothing
/// per byte beyond the copy every writer already makes. It buffers **nothing**:
/// every byte is forwarded before it is recorded, so a slow client is not held
/// up and a large pack does not accumulate here.
struct TrailerTap<'w> {
    inner: &'w mut dyn Write,
    /// The tail, oldest first, never longer than `want`.
    tail: Vec<u8>,
    want: usize,
}

impl<'w> TrailerTap<'w> {
    fn new(inner: &'w mut dyn Write, want: usize) -> Self {
        TrailerTap {
            inner,
            tail: Vec::with_capacity(want),
            want,
        }
    }

    /// The last `want` bytes written.
    ///
    /// An error rather than a shorter slice when fewer went past: a pack
    /// without a full trailer is not a pack, and inventing an id for it would
    /// put a wrong pack id in a log line that exists to identify what was sent.
    fn trailer(&self) -> Result<Vec<u8>> {
        if self.tail.len() < self.want {
            return Err(Error::Store(gunnar_store::Error::backend(format!(
                "the engine emitted {} bytes, which is shorter than the {}-byte pack trailer \
                 every packfile ends with",
                self.tail.len(),
                self.want
            ))));
        }
        Ok(self.tail.clone())
    }
}

impl Write for TrailerTap<'_> {
    fn write(&mut self, data: &[u8]) -> std::io::Result<usize> {
        let written = self.inner.write(data)?;
        let fresh = &data[..written];
        if fresh.len() >= self.want {
            self.tail.clear();
            self.tail
                .extend_from_slice(&fresh[fresh.len() - self.want..]);
        } else {
            let overflow = (self.tail.len() + fresh.len()).saturating_sub(self.want);
            self.tail.drain(..overflow);
            self.tail.extend_from_slice(fresh);
        }
        Ok(written)
    }

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