znippy-plugin-git 0.1.1

//! **The reading contract, answered out of Arrow IPC.** No gix, anywhere.
//!
//! ZNIPPY-GIT APACHE ARROW IPC IS LAW. [`crate::store`] holds the eleven — what
//! a caller *stores*; this file holds [`GitServe`] — what a caller *reads*, and
//! what a clone is served from. The split is the contract's, not this crate's:
//! `GitOps::get` hands back the pack entry **byte for byte as the client sent
//! it**, which for a delta is a delta, and nothing on the eleven inflates one.
//!
//! ```text
//!   read / header / sizes        →  the objects table, then §14's exploded table
//!   head / set_head              →  the ref log; HEAD is NOT a row in refs()
//!   select                       →  roaring bitmaps over the store's ordinals
//!   emit_pack                    →  emit_set → topological_order → byte-range copy
//! ```
//!
//! # 🚨 The trap this file exists to avoid, stated once
//!
//! The working reference for every method below is `gunnar-coldtier`'s
//! `git_store.rs`, and it carries **87 `gix_` references**. It reads objects
//! through `gix_pack::Find` / `decode_entry` / `decode_header`, and it emits
//! packs through `iter_from_counts` / `InOrderIter` / `FromEntriesIter`.
//! **Its intent is ported here. Not one line of its code is.**
//!
//! `znippy-plugin-git` links **zero gix in `[dependencies]`** — the four gix
//! crates it names are `[dev-dependencies]`, the differential oracle in
//! [`crate::pack_walk`] that proves this crate's parser agrees with gitoxide
//! *without shipping it*. That separation is the entire point of two
//! implementations behind one contract, and no method here may create an edge
//! into it. Not for a parser, not for `gix-hash`, not temporarily.
//!
//! Nor would copying it be desirable. That pack pipeline **is** `P-018`'s serial
//! tail — counting's serial reduce, a `sort_by` over counts, `InOrderIter`'s
//! `BTreeMap` reorder, `FromEntriesIter` hashing every byte on the consuming
//! thread — plus `P-025`'s residue, `data.to_owned()`, one heap allocation per
//! object served. The Arrow arm's answer to all of it is a **byte-range copy out
//! of the archive**, and [`crate::pack_walk::emit_pack`] is where it lives.
//!
//! # The serving tier is serial — `P-004`, DECIDED, with one bounded exception
//!
//! `sizes` does not fan out. `select` does not fan out. The two-tier split is
//! settled: **request concurrency, no intra-request fan-out**, one core's worth
//! per transfer behind an admission gate. At saturation, parallelising *one*
//! clone across N cores does not raise clones per second — it is the same work
//! rearranged — and the constellation's fan-out primitive has no reentrancy
//! detection, so a fan-out here inside a fan-out there silently spawns W².
//! `gatling` belongs in [`crate::indexer`]'s background tier, which is where
//! nearly all of it is.
//!
//! The exception, added 2026-08-14, is **phase 1 of `emit_oids`** —
//! [`GitStore::resolve_emit_payloads`](crate::git_ops::GitStore::resolve_emit_payloads),
//! which turns each entry's archive address into a slice of the mapped archive.
//! It is allowed for the reason the decision above is stated in terms of and not
//! against it:
//!
//!  * it is **bounded at 4 workers**, not at ncores, so 32 admitted transfers
//!    are 128 threads at worst rather than 32 × ncores — the admission gate
//!    stays in charge of the machine;
//!  * it **does not run at all** below 4096 entries, so every request small
//!    enough for thread-spawn to dominate is byte for byte the old serial path;
//!  * it takes **no lock**, because the archive is append-only and immutable
//!    behind a snapshot, so it is not the "same work rearranged" — it is work
//!    that has no serial dependency to rearrange around.
//!
//! Phase 2 — output offsets, `OFS_DELTA` distances, the writes and the running
//! hash — stays strictly serial, and that is not a bound anyone chose: an
//! `OFS_DELTA` names its base by distance back in the *output* pack, so entry
//! *n* cannot be encoded until every earlier entry's length is known.
//!
//! Every method **blocks**. Never call one from an async task without
//! `tokio::task::spawn_blocking`.

use anyhow::{anyhow, bail, Context as _, Result};

pub use git_storage_trait::{Caps, GitServe, PackStats, ReachSet};

use crate::git_ops::{GitOps, GitStore, RefRow, TxId};
use crate::index_layout::{ObjType, ObjectIndex};
use crate::object::GitObjectKind;
use crate::refs::RefUpdate;
use crate::store::Oid;

/// `HEAD`, the one pseudo-ref this contract names.
///
/// It is **not** a row in [`GitOps::refs`] — see [`GitStore::head`] for the
/// argument, which is a type constraint rather than a preference.
pub const HEAD: &str = "HEAD";

/// The resolved kind, in the contract's vocabulary.
///
/// A total function, and that is the guarantee: [`GitObjectKind`] has exactly
/// the four real types and no delta variants, so a value that came out of a
/// resolver **cannot** be labelled `OfsDelta` or `RefDelta` by accident. The
/// contract's `ObjType` has six codes and this is the one place the narrowing is
/// written down — [`crate::git_ops::GitStore::emit_set`] rebuilds whole entries
/// and needs the same narrowing, and calls this rather than repeating it.
pub(crate) fn resolved_type(kind: GitObjectKind) -> ObjType {
    match kind {
        GitObjectKind::Commit => ObjType::Commit,
        GitObjectKind::Tree => ObjType::Tree,
        GitObjectKind::Blob => ObjType::Blob,
        GitObjectKind::Tag => ObjType::Tag,
    }
}

/// How many bytes of an entry are read to find its type and its base.
///
/// A type/size varint is at most 10 bytes and an `OFS_DELTA` distance at most 9,
/// so 64 covers every header plus a 32-byte `REF_DELTA` base oid with room over.
/// It is a **ceiling on a `pread`**, not a promise about the entry: a shorter
/// entry reads short and the grammar still terminates.
const HEADER_PROBE: u64 = 64;

impl<S: ObjectIndex + 'static> GitStore<S> {
    /// The chain walk behind [`GitServe::header`]: the **resolved** type of the
    /// entry at `offset`, reading entry headers only.
    ///
    /// # Why this is not "just read the column"
    ///
    /// `objects.obj_type` is the entry's type **in the pack**, so a delta's is
    /// `OfsDelta` or `RefDelta` and the resolved kind is one or more entries
    /// away. `objects.uncompressed_size` has no such problem — it is written
    /// after delta application — which is why [`GitServe::sizes`] is a pure
    /// index pass and this is not.
    ///
    /// # What it costs, exactly
    ///
    /// One 64-byte `pread` per link of the chain, and **no inflate, no delta
    /// application, no allocation sized by the object**. gix answers the same
    /// question with `decode_header`, which walks the same links; the difference
    /// is that this one reads the offsets straight out of §13's `delta_base`
    /// column for the first hop rather than reconstructing them, and does not
    /// need a pack handle, a zlib scratch or a delta cache to do it.
    ///
    /// Chains are bounded by the depth the pushing client's `pack.depth` chose —
    /// 50 by default — but nothing in a *pushed* pack is under this server's
    /// control, so the walk carries its own ceiling and refuses rather than
    /// looping. A cycle in a pack's back-references is a corrupt pack, and
    /// saying so beats spinning.
    fn resolved_type_at(&self, offset: u64, obj_type: ObjType, delta_base: u64) -> Result<ObjType> {
        /// git's own `pack.depth` ceiling, doubled. A chain longer than this is
        /// not a deep delta, it is a loop.
        const MAX_LINKS: usize = 100;

        // One mapping for the whole chain walk. A probe used to be one `pread`
        // per link — an allocation and a kernel→user copy for 64 bytes that are
        // already in the page cache; through the mapping it is a bounds check
        // and a pointer, and the `Cow` is `Borrowed` on every link the snapshot
        // covers.
        let snap = self.archive_snapshot()?;
        let mut at = offset;
        let mut t = obj_type;
        let mut base = delta_base;
        for _ in 0..MAX_LINKS {
            match t {
                ObjType::Commit | ObjType::Tree | ObjType::Blob | ObjType::Tag => return Ok(t),
                ObjType::OfsDelta => {
                    // §13's column already holds the base's **absolute archive
                    // offset**, resolved at absorb time. No distance arithmetic
                    // and no second parse of this entry.
                    if base == 0 {
                        bail!(
                            "the entry at archive offset {at} is an ofs-delta whose base column is \
                             0, which is the no-base sentinel — the index disagrees with itself"
                        );
                    }
                    at = base;
                }
                ObjType::RefDelta => {
                    // A ref-delta's `delta_base` column is 0 by construction —
                    // its base is named by oid, inside the entry, directly after
                    // the type/size varint. The same few bytes
                    // `GitStore::emit_set` reads for the same reason.
                    let head = self.extent(&snap, at, HEADER_PROBE)?;
                    let (_, _, n) = crate::pack_walk::type_and_size_of(&head)?;
                    let oid_len = self.hash_kind().oid_len();
                    let base_oid = head.get(n..n + oid_len).ok_or_else(|| {
                        anyhow!(
                            "the ref-delta entry at archive offset {at} has no base oid after its \
                             header"
                        )
                    })?;
                    let Some(row) = self.index().lookup(base_oid) else {
                        bail!(
                            "the ref-delta entry at archive offset {at} names a base this \
                             repository does not have — its type cannot be resolved"
                        );
                    };
                    at = row.offset;
                    t = row.obj_type;
                    base = row.delta_base;
                    continue;
                }
            }
            // One header read at the new position. The type column would be a
            // second lookup keyed the wrong way (by offset, and the index is
            // keyed by oid), so the entry's own grammar answers it.
            let head = self.extent(&snap, at, HEADER_PROBE)?;
            let (next_t, _, n) = crate::pack_walk::type_and_size_of(&head)?;
            t = next_t;
            base = match next_t {
                ObjType::OfsDelta => {
                    let (distance, _) = crate::pack_walk::ofs_distance_of(&head[n..])?;
                    at.checked_sub(distance).ok_or_else(|| {
                        anyhow!(
                            "the ofs-delta at archive offset {at} names a base {distance} bytes \
                             back, which is before the start of the archive"
                        )
                    })?
                }
                _ => 0,
            };
        }
        bail!(
            "resolving the type of the entry at archive offset {offset} followed more than \
             {MAX_LINKS} delta links — the pack's back-references form a cycle"
        )
    }

    /// **Emit a packfile for an explicit object set**, closed over its delta
    /// bases — and over **nothing else**.
    ///
    /// This is the whole of [`GitServe::emit_pack`] as of 2026-08-10: that
    /// method is this call and a discarded `have`. It was written as the
    /// *inherent* escape route for a trait method that closed over its input,
    /// so that a caller told *"walk it yourself"* by [`GitServe::select`]'s
    /// `Ok(None)` had somewhere to put the set it had walked. The trait method
    /// stopped closing over its input, so the escape route and the front door
    /// are now the same door. It stays inherent because the gix arm needs no
    /// equivalent and the contract does not grow a method for it.
    ///
    /// # 🔴 No closure at all any more, and that is the 2026-08-11 fix
    ///
    /// [`GitStore::emit_set`] used to add a delta's base when the base fell
    /// outside the request, on the reasoning that an `OFS_DELTA` names its base
    /// by position and so cannot be encoded without it. That reasoning is
    /// correct about the *pack format* and wrong about the *clone*: a base
    /// pulled in is an object the client did not ask for, and if it is a tree it
    /// arrives owing children the pack does not contain. `git index-pack
    /// --check-self-contained-and-connected` — what a clone runs — then dies
    /// with `did not receive expected object`, while this server logs
    /// `git.upload_pack.served`. See `emit_set` for the measured numbers.
    ///
    /// So the set is now **exactly** the caller's, and the base decides how each
    /// entry is *encoded* rather than what the pack contains: base inside the
    /// request, copy the stored bytes; base outside it, rebuild the object
    /// whole. That is stock `pack-objects`' rule, and `recompressed` counts the
    /// second case honestly instead of being a literal zero.
    ///
    /// # Three steps, all of them znippy's own
    ///
    /// 1. [`GitStore::emit_set`] reads each requested entry and decides copy or
    ///    rebuild, never adding an object;
    /// 2. [`crate::pack_walk::topological_order`] puts every base before the
    ///    delta naming it, which backwards distances require;
    /// 3. [`crate::pack_walk::emit_pack`] re-encodes each entry header — the one
    ///    thing that must change, because a distance is relative to a position
    ///    in the *input* pack — and streams every payload **byte for byte**.
    ///
    /// `copied + recompressed == objects` always, and for a whole-repository
    /// clone `recompressed` is 0 because such a request contains every base.
    /// Both are counted off what happened rather than asserted: a byte count and
    /// a wall clock cannot tell a pack-copy from a re-deflate, and both pass
    /// `index-pack --strict`.
    pub fn emit_oids(
        &self,
        oids: &[Oid<'_>],
        have: &[Oid<'_>],
        caps: &Caps,
        out: &mut dyn std::io::Write,
    ) -> Result<PackStats> {
        // **Both halves, or nothing.** `caps.thin` is the client's *consent* to
        // receive a delta whose base it must supply itself; `have` is the
        // negotiation's answer to *which* bases those may be. Consent with no
        // negotiated tips is a clone — the client holds nothing, so there is
        // nothing a base could safely point at — and tips without consent is a
        // client that never agreed to run `index-pack --fix-thin`. Either alone
        // is a failed fetch rather than a smaller one, so either alone is `None`
        // and the boundary entries go out whole exactly as before.
        let thin_haves = (caps.thin && !have.is_empty()).then_some(have);

        let entries = self
            .emit_set(oids, caps.ofs_delta, thin_haves)
            .context("building the emit set")?;
        let (ordered, missing) = crate::pack_walk::topological_order(entries);
        if !missing.is_empty() {
            bail!(
                "the emit set is not closed: {} delta base(s) absent, first at archive offset {} \
                 — refusing to emit a pack whose closure does not hold",
                missing.len(),
                missing[0]
            );
        }

        // **The one capability that changes which bytes come out**, and it is
        // acted on in `emit_set` rather than here. A client that did not
        // advertise `ofs-delta` used to be REFUSED by name at this point,
        // because this engine copies stored entries and had no way to re-name a
        // base. It has one now: an `OFS_DELTA` whose base is in the request is
        // re-headed as a `REF_DELTA` carrying the same compressed delta stream,
        // so that client is served a pack it can parse without a single byte
        // being re-deflated. What remains here is the assertion that it worked —
        // emitting an `OFS_DELTA` to such a client is the failure that looks
        // like a working server.
        if !caps.ofs_delta {
            if let Some(e) = ordered.iter().find(|e| e.obj_type == ObjType::OfsDelta) {
                bail!(
                    "this client did not advertise ofs-delta and the entry at archive offset {} \
                     is still an ofs-delta after the emit set was built; refusing rather than \
                     answering with a pack the client cannot parse",
                    e.offset
                );
            }
        }
        // `caps.thin` IS acted on now, in `emit_set`, and the pass it needed
        // turned out to belong exactly where the note that used to sit here said
        // it would: beside the negotiation that produced `have`. What it does is
        // narrow, and deliberately so — it changes nothing about *which* objects
        // are sent, only whether a boundary entry may name a base the pack does
        // not carry instead of being rebuilt whole.

        // **Borrowed, not owned.** Every entry a clone sends is an
        // `EntryBytes::Extent` — an address, 16 bytes — which `emit_ordered`
        // resolves to a slice of one mapping of the archive (phase 1, parallel)
        // before streaming it (phase 2, serial). The `to_vec()` that used to sit
        // on this path is `P-025` exactly: one heap allocation per object
        // served, on the path whose whole claim is that it copies stored bytes
        // without touching them.
        let report = self
            .emit_ordered(&ordered, out)
            .context("emitting the pack")?;

        Ok(PackStats {
            bytes: report.bytes,
            objects: u64::from(report.written),
            copied: u64::from(report.copied),
            // **Counted, not asserted.** This was the literal `0` for as long as
            // nothing on the path could re-deflate; `emit_set` can now, for
            // exactly the entries whose delta base the request does not contain,
            // and the receipt says how many rather than continuing to claim
            // none. A whole-repository clone still reports 0 — and that is a
            // measurement of a full selection, not a promise.
            recompressed: u64::from(report.recompressed),
        })
    }
}

impl<S: ObjectIndex + 'static> GitServe for GitStore<S> {
    /// **The object, inflated and delta-resolved** — §14's exploded table, one
    /// point lookup.
    ///
    /// This is the question [`GitOps::get`] deliberately does not answer, and
    /// [`GitStore::content`] has answered it since §14 landed; the trait method
    /// is that call plus the kind narrowing. Two paths under it, returning
    /// identical bytes and distinguishable only by
    /// [`crate::exploded::ExplodedStats`]: the table, or a re-derivation from
    /// the verbatim truth when the table has been dropped or has not caught up.
    ///
    /// The returned [`ObjType`] can never be a delta — see [`resolved_type`].
    fn read(&self, oid: Oid<'_>) -> Result<Option<(ObjType, Vec<u8>)>> {
        Ok(self
            .content(oid)?
            .map(|(kind, bytes)| (resolved_type(kind), bytes)))
    }

    /// **Kind and post-resolution size, without the payload.**
    ///
    /// The size is a column, and the *right* column: `uncompressed_size` is
    /// written after delta application, so it is the fact a `.idx` and a `.rev`
    /// together cannot answer, and it is read rather than measured.
    ///
    /// The kind is not a column — `obj_type` is the entry's type in the pack —
    /// so for the common case (a non-delta entry) this touches nothing else, and
    /// for a delta it walks the chain's **headers** through
    /// [`GitStore::resolved_type_at`]. Still no inflate and still no payload.
    fn header(&self, oid: Oid<'_>) -> Result<Option<(ObjType, u64)>> {
        // Same preamble as `has`/`extents`, and for the same reason: a store
        // holding a pack whose bytes are durable and whose objects are not
        // indexed yet cannot answer "absent" without lying.
        if self.unindexed_packs() > 0 {
            self.absorb_pending()?;
        }
        let Some(row) = self.index().lookup(oid) else {
            return Ok(None);
        };
        let kind = self
            .resolved_type_at(row.offset, row.obj_type, row.delta_base)
            .with_context(|| format!("resolving the type of {}", hex::encode(oid)))?;
        Ok(Some((kind, row.uncompressed_size)))
    }

    /// **Post-resolution sizes in bulk: one index pass, no chain touched.**
    ///
    /// The half of [`header`](GitServe::header) that is already data. On the
    /// `h2h` fixture one clone paid **34 124** per-object header walks, every
    /// one of which this collapses into a single `lookup_batch`, to establish
    /// that a repository whose largest blob is 16 KiB holds nothing over the
    /// 1 GiB default ceiling.
    ///
    /// # The preamble is not optional
    ///
    /// This reaches [`GitStore::index`] directly, so it absorbs pending index
    /// work itself. Without it a pack that landed since the last drain reads as
    /// **absent**, and an absent size that a caller treated as a pass would
    /// silently skip the ceiling for exactly the objects a push had just
    /// introduced. `None` at a position is *"unknown, go ask `header`"* and
    /// never *"fine"*.
    fn sizes(&self, oids: &[Oid<'_>]) -> Result<Vec<Option<u64>>> {
        if self.unindexed_packs() > 0 {
            self.absorb_pending()?;
        }
        Ok(self
            .index()
            .lookup_batch(oids)
            .into_iter()
            .map(|row| row.map(|r| r.uncompressed_size))
            .collect())
    }

    /// **`HEAD`, which is not a row in [`GitOps::refs`].**
    ///
    /// The gix arm's `iter()` *"walks `refs/` (loose and packed) and deliberately
    /// excludes the pseudo-refs such as `HEAD`, which is exactly the contract
    /// every other backend honours"* — and as of this change so does this arm
    /// (see [`GitOps::refs`]). The reason is a type constraint rather than
    /// taste: a name type that admits `HEAD` also admits `MERGE_HEAD` and
    /// `FETCH_HEAD`, so putting pseudo-refs in the row stream means either
    /// widening the name type or filtering at every consumer.
    ///
    /// So it is read here, off the same fold [`GitOps::refs`] reads, rather than
    /// from a second place that could disagree with it. `None` for a repository
    /// that has never pointed one — an empty repository has no `HEAD`.
    fn head(&self) -> Result<Option<RefRow>> {
        let state = self.ref_state()?;
        let Some(s) = state.get(HEAD) else {
            return Ok(None);
        };
        let decode = |h: &Option<String>| -> Result<Option<Vec<u8>>> {
            match h {
                Some(h) => Ok(Some(hex::decode(h).map_err(|e| {
                    anyhow!("HEAD holds `{h}`, which is not a hex oid: {e}")
                })?)),
                None => Ok(None),
            }
        };
        Ok(Some(RefRow {
            name: HEAD.to_string(),
            oid: decode(&s.target)?,
            peeled: decode(&s.peeled)?,
            symref_target: s.symref_target.clone(),
        }))
    }

    /// Point `HEAD`, through the same log a push writes to.
    ///
    /// # The two shapes of a target, and how they are told apart
    ///
    /// `HEAD` is symbolic in every repository anyone serves — `ref:
    /// refs/heads/main` — and detached in the one case git also supports. The
    /// contract's parameter is one `&str` for both, so:
    ///
    /// * a target starting with `refs/` is a **symbolic** ref, written as one;
    /// * anything else must be a hex oid of exactly this store's width, written
    ///   as a direct target.
    ///
    /// The two cannot be confused: a ref name and a 40- or 64-character hex
    /// string are disjoint. Anything that is neither is **refused** rather than
    /// guessed at — a `HEAD` pointing at a name nothing resolves is an
    /// unclonable repository, and it is cheaper to say so here.
    ///
    /// A symbolic target is deliberately **not** checked for existence.
    /// `git init` points `HEAD` at `refs/heads/main` before that branch exists,
    /// and refusing it would make an empty repository unrepresentable. A direct
    /// target *is* checked, by [`GitOps::put_refs`], because a detached `HEAD`
    /// naming an absent object is dangling in exactly the sense that refuses.
    fn set_head(&self, target: &str) -> Result<TxId> {
        let update = if target.starts_with("refs/") {
            RefUpdate::symbolic(HEAD, target)
        } else {
            let hex_len = self.hash_kind().hex_len();
            if target.len() != hex_len || hex::decode(target).is_err() {
                bail!(
                    "HEAD can be pointed at a ref name under `refs/` or at a {hex_len}-character \
                     hex oid; `{target}` is neither, and guessing which was meant is how a \
                     repository ends up unclonable"
                );
            }
            RefUpdate::set(HEAD, target.to_ascii_lowercase())
        };
        self.put_refs(&[update])
    }

    /// **Emit a packfile containing exactly `objects`.**
    ///
    /// One line of delegation to [`GitStore::emit_oids`], and the absence of a
    /// second line is the contract.
    ///
    /// # 🔴 This must NOT compute a closure. Fixed 2026-08-10.
    ///
    /// Until that date this method opened with `self.select(objects, have)?` and
    /// emitted *that* — it **closed over its own input**. The caller's set came
    /// back larger than it went in, and the caller is upload-pack holding
    /// `selection.objects`: already post-filter, post-shallow, post-`include-tag`
    /// and *deliberately not closed*. So a `--filter=blob:none`,
    /// `--filter=tree:0` or `--depth=N` fetch was served **exactly the objects it
    /// had asked to be left out**.
    ///
    /// The reason it needed a rename and a guard rather than a shrug is that
    /// **nothing we own could see it**: the over-sent pack passes
    /// `git index-pack --strict` *and* `git fsck`, the clone succeeds, the exit
    /// code is zero, and the client just silently receives more than it asked
    /// for. `P-027`'s shape — a change no test can see. Pinned now by
    /// `emit_pack_emits_exactly_the_set_it_is_given_and_never_its_closure`,
    /// which asserts a **count** (seen RED at 4 against an expected 1).
    ///
    /// It had a second, louder failure mode too, and it is the one that proves
    /// the two questions were never the same question: [`select`](GitServe::select)'s tip check
    /// declines a `want` that is not a commit in the graph, so a selection
    /// containing a tree — which every real one does — could not be emitted at
    /// all.
    ///
    /// # Where the closure went
    ///
    /// Nowhere. [`select`](GitServe::select) still owns the close-over-tips half
    /// and still answers `Ok(None)` when its projection cannot cover a request;
    /// that hatch is untouched and still load-bearing. The two halves are now
    /// cleanly separated: *what to send* is asked of `select`, *send exactly
    /// this* is asked of here. A caller that `select` told to walk the graph
    /// itself hands the walked set straight back to this method.
    ///
    /// There is no closure left on this path at all: since 2026-08-11 the base
    /// decides how an entry is *encoded*, never what the pack contains.
    ///
    /// # `have` is READ, and only for thin-pack base selection
    ///
    /// The contract defines `have` as the negotiated common **tips**, *"used for
    /// thin-pack base selection, never to derive membership"*, and that is
    /// exactly and only what this arm does with it. Paired with `caps.thin`, it
    /// lets a boundary entry go out as a `REF_DELTA` naming a base the receiver
    /// already holds — carrying the stored delta stream unchanged — instead of
    /// being rebuilt whole. See [`GitStore::client_bases`] for the closure and
    /// for why the caller has to be able to stand behind the voucher.
    ///
    /// It is deliberately **not** repurposed into an exclusion set. Deriving
    /// membership from `have` is precisely the closure this method stopped
    /// doing, in the opposite direction: it would *remove* objects the caller's
    /// finished selection had already decided to send, which is an under-send,
    /// which is the failure that exits zero. The narrowing below changes the
    /// *encoding* of an entry and never whether it is emitted.
    fn emit_pack(
        &self,
        objects: &[Oid<'_>],
        have: &[Oid<'_>],
        caps: &Caps,
        out: &mut dyn std::io::Write,
    ) -> Result<PackStats> {
        self.emit_oids(objects, have, caps, out)
    }

    /// **`want` minus `have`, out of §13's bitmaps, with no object opened** —
    /// plus the two facts the caller cannot recompute cheaply.
    ///
    /// # The refusal is the part that matters
    ///
    /// [`GitOps::reachable`] treats a `want` it has no bitmap for as contributing
    /// **itself and nothing else**. That is the safe direction for its own
    /// caller — `live_set` over-keeps, and over-keeping is harmless — and a
    /// silent **under-send** for this one. A clone served that way exits zero
    /// with one object per branch.
    ///
    /// So every tip is checked against the store's commit graph first, and a tip
    /// that is not in it declines the whole request.
    ///
    /// # 🔴 *"In the graph"* and *"has a bitmap"* are NO LONGER the same fact
    ///
    /// They were, and this paragraph used to say so, because the live table was
    /// built with no commit cap precisely so that they would be. That cost 530
    /// 218 allocations a fetch (see [`crate::git_ops`]'s `LIVE_REACH_COMMITS`),
    /// and since 2026-08-14 the table is **sampled** at 512 like the sealed
    /// archive's.
    ///
    /// The check below is unchanged and is still exact, but it is now exact for
    /// a different reason. It is *"in the graph"* that matters, not *"has a
    /// bitmap"*: [`crate::reach::accumulate`] walks a bitmapless commit down to
    /// the first bitmapped one behind it, so every graph commit is answerable,
    /// with or without a bitmap of its own. A tip that is not a graph row is
    /// still refused — there is nothing to walk — and that is the case this
    /// check exists for.
    ///
    /// `Ok(None)` is therefore reachable in three ways, and all three are the
    /// same statement — *the projection does not cover this request*:
    /// an empty graph (which a clean restart produces), a tip that is not a
    /// commit in it, and a graph row whose oid does not parse.
    ///
    /// # The selection is NOT re-ordered, and that is measured
    ///
    /// `reachable` answers in ordinal order, which is oid-lexicographic and
    /// therefore unrelated to where the bytes are. Sorting by archive offset so
    /// a delta's base precedes it was written and could not be made to fail:
    /// **measured 2026-08-08 on a 154-object fixture — with the sort, without
    /// it, and with the whole selection deliberately reversed, all three give
    /// `copied=154 recompressed=0` and the identical object graph**, because
    /// [`crate::pack_walk::topological_order`] orders the emission itself. The
    /// sort was a batch index lookup per request buying a property the emitter
    /// already owns (LAW 5), and its absence is something no guard can see
    /// (LAW 2). It is gone.
    fn select(&self, want: &[Oid<'_>], have: &[Oid<'_>]) -> Result<Option<ReachSet>> {
        // The cheapest form of the same refusal, and the one a clean restart
        // takes: with no commit graph there is no bitmap for any want, so
        // `reachable` would contribute the wants alone.
        if self.commit_count() == 0 {
            return Ok(None);
        }

        // The graph as raw oids. Needed twice — to refuse a tip it does not
        // cover, and to name the commits in the answer.
        //
        // **Folded, not rebuilt per request.** This used to be
        // `graph_snapshot()` — a deep clone of every `CommitNode`, oid `String`,
        // parent list and all — followed by a `hex::decode` into a fresh
        // `Vec<u8>` per commit, on every single `select`. The set cannot change
        // between folds, so [`crate::git_ops::Derived::commit_raw`] holds it and
        // this borrows. `None` is the identical refusal the decode arm used to
        // produce: a graph row whose oid does not parse is a corrupt derivation,
        // and declining is safe because the caller's own walk still serves the
        // request.
        let Some(commits) = self.commit_oids_raw()? else {
            return Ok(None);
        };
        if want.iter().any(|tip| !commits.contains(*tip)) {
            return Ok(None);
        }

        let objects = self
            .reachable(want, have)
            .context("selecting want minus have")?;

        // What the client held BEFORE this transfer: the closure of `have`, and
        // only it. Empty for a clone, which is the request with no excludes at
        // all and therefore the one that pays nothing for this.
        //
        // **Raw bytes, in one buffer.** This is the repository-sized half of the
        // answer — the closure of what the client already holds, 12 112 oids to
        // serve a 69-object fetch on the measured corpus — and it used to be
        // built as oid hex and `hex::decode`d back into a `Vec<u8>` per object
        // on its way into a `Vec<Vec<u8>>` the caller only ever counts and
        // scans. `git_storage_trait::OidList` carries it in one allocation; see
        // `Derived::oids_raw`.
        let client_has = if have.is_empty() {
            git_storage_trait::OidList::new()
        } else {
            self.reachable_raw(have, &[])
                .context("closing over what the client already holds")?
        };

        let selected_commits = objects
            .iter()
            .filter(|oid| commits.contains(*oid))
            .cloned()
            .collect();

        Ok(Some(ReachSet {
            objects,
            commits: selected_commits,
            client_has,
        }))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::object::{canonical, GitHashKind};
    use crate::store::tests::{one_blob_pack, real_pack, tmpdir};
    use git_storage_trait::{Observed, RefCas, RefRejection};

    /// **`header` names the RESOLVED kind for every entry in a real pack —
    /// deltas included — and it does it with no gix and no inflate.**
    ///
    /// The oracle is [`crate::resolve::Resolved`], which carries both
    /// `stored_type` (what the entry is in the pack) and `kind` (what the chain
    /// resolves to). Asserting against `kind` is the whole point: a `header`
    /// that read the type column would agree on every non-delta and be wrong on
    /// exactly the deltas, which is the failure a spot-check misses.
    ///
    /// The test refuses to pass on a corpus that would not have caught that: it
    /// asserts the pack actually contained deltas first.
    #[test]
    fn header_resolves_every_delta_to_its_real_kind() {
        let dir = tmpdir("serve-header");
        let store = GitStore::open(&dir, "rickard").unwrap();
        let (pack, rows) = real_pack();
        store.put(&pack, &[]).unwrap();
        store.wait_indexed();
        store.absorb_pending().unwrap();

        let deltas = rows
            .iter()
            .filter(|r| matches!(r.stored_type, ObjType::OfsDelta | ObjType::RefDelta))
            .count();
        assert!(
            deltas > 0,
            "this corpus pack has no delta entries, so it cannot tell a chain walk from a column \
             read and the test proves nothing"
        );

        for r in &rows {
            let (kind, size) = GitServe::header(&store, &r.oid)
                .unwrap()
                .unwrap_or_else(|| panic!("{} is in the pack but header said absent", hex::encode(&r.oid)));
            assert_eq!(
                kind,
                resolved_type(r.kind),
                "{} is stored as {:?} and resolves to {:?}; header said {:?}",
                hex::encode(&r.oid),
                r.stored_type,
                r.kind,
                kind
            );
            assert_eq!(
                size,
                r.uncompressed_size,
                "{} post-resolution size",
                hex::encode(&r.oid)
            );
        }
    }

    /// `sizes` answers the same numbers as `header`, in one pass, positionally.
    #[test]
    fn sizes_is_header_in_bulk_and_stays_positional() {
        let dir = tmpdir("serve-sizes");
        let store = GitStore::open(&dir, "rickard").unwrap();
        let (pack, rows) = real_pack();
        store.put(&pack, &[]).unwrap();
        store.wait_indexed();

        // An oid nothing has, deliberately in the MIDDLE: a `sizes` that
        // filtered misses rather than answering `None` in place would shift
        // every later answer by one and still return the right count.
        let absent = vec![0xABu8; 20];
        let mut oids: Vec<&[u8]> = rows.iter().map(|r| r.oid.as_slice()).collect();
        let middle = oids.len() / 2;
        oids.insert(middle, &absent);

        let got = GitServe::sizes(&store, &oids).unwrap();
        assert_eq!(got.len(), oids.len(), "sizes must be positional");
        assert_eq!(got[middle], None, "an absent oid answers None IN PLACE");
        for (i, r) in rows.iter().enumerate() {
            let at = if i < middle { i } else { i + 1 };
            assert_eq!(got[at], Some(r.uncompressed_size), "row {i}");
        }
    }

    /// **`HEAD` is not a row in `refs()`, and it has its own accessor** — §3f
    /// Q1, and the conformance check the suite could not previously make.
    ///
    /// Seen RED by deleting the filter in [`GitOps::refs`]: `HEAD` appears in
    /// the ref stream and the first assertion fails. The two halves are both
    /// needed — a `refs()` that dropped `HEAD` and a `head()` that could not
    /// find it would pass the first assertion and leave the repository with no
    /// default branch to advertise.
    #[test]
    fn head_is_not_a_ref_row_but_is_reachable_through_its_own_accessor() {
        let dir = tmpdir("serve-head");
        let store = GitStore::open(&dir, "rickard").unwrap();
        let (pack, oid) = one_blob_pack(b"a blob to point at");
        store.put(&pack, &[]).unwrap();
        store.wait_indexed();
        store
            .put_refs(&[RefUpdate::set("refs/heads/main", hex::encode(&oid))])
            .unwrap();

        GitServe::set_head(&store, "refs/heads/main").unwrap();

        let rows = store.refs().unwrap();
        assert!(
            rows.iter().all(|r| r.name != HEAD),
            "HEAD is a pseudo-ref and must not appear in the ref stream: {:?}",
            rows.iter().map(|r| &r.name).collect::<Vec<_>>()
        );
        assert!(
            rows.iter().any(|r| r.name == "refs/heads/main"),
            "filtering HEAD must not have filtered anything else"
        );

        let head = GitServe::head(&store).unwrap().expect("HEAD was just set");
        assert_eq!(head.name, HEAD);
        assert_eq!(head.symref_target.as_deref(), Some("refs/heads/main"));
        assert_eq!(head.oid, None, "a symbolic HEAD has no direct target");

        // A store that has never been pointed has no HEAD, and that is `None`
        // rather than an error or an empty row.
        let dir2 = tmpdir("serve-head-empty");
        let fresh = GitStore::open(&dir2, "rickard").unwrap();
        assert!(GitServe::head(&fresh).unwrap().is_none());
    }

    /// `set_head` refuses a target that is neither a ref name nor an oid rather
    /// than guessing, and takes a detached oid when given one.
    #[test]
    fn set_head_takes_a_ref_or_an_oid_and_refuses_anything_else() {
        let dir = tmpdir("serve-sethead");
        let store = GitStore::open(&dir, "rickard").unwrap();
        let (pack, oid) = one_blob_pack(b"detached");
        store.put(&pack, &[]).unwrap();
        store.wait_indexed();

        let err = GitServe::set_head(&store, "main").unwrap_err();
        assert!(
            format!("{err:#}").contains("is neither"),
            "a bare branch name must be refused, not guessed at: {err:#}"
        );

        GitServe::set_head(&store, &hex::encode(&oid)).unwrap();
        let head = GitServe::head(&store).unwrap().unwrap();
        assert_eq!(head.oid.as_deref(), Some(oid.as_slice()));
        assert_eq!(head.symref_target, None);
    }

    /// **`select` declines rather than under-sending when it has no graph — and
    /// `emit_pack` is not affected by that at all.**
    ///
    /// The load-bearing `Ok(None)`. A store holding one blob and no commit has
    /// no reachability projection, and the wrong answer here is not an error —
    /// it is a `Some` containing the want alone, which serves a clone that exits
    /// zero and is short by everything.
    ///
    /// # What the second half of this test used to assert, and why it changed
    ///
    /// Until 2026-08-10 it asserted that `emit_pack` **refused** on the back of
    /// this same decline, because `emit_pack` began by calling `select`. It no
    /// longer does — that call was the over-send bug, see
    /// [`GitServe::emit_pack`] — so the assertion would now be asserting the
    /// defect. It is **replaced rather than deleted**, by the stronger statement
    /// the separation makes true: a store that cannot select *anything* still
    /// emits exactly the set it is handed. The refusal is not lost, it has an
    /// owner: `select` says `None`, and the caller decides.
    #[test]
    fn select_declines_when_the_projection_cannot_cover_the_request() {
        let dir = tmpdir("serve-select-none");
        let store = GitStore::open(&dir, "rickard").unwrap();
        let (pack, oid) = one_blob_pack(b"no commits here");
        store.put(&pack, &[]).unwrap();
        store.wait_indexed();
        store.absorb_pending().unwrap();

        assert_eq!(store.commit_count(), 0, "the fixture must have no commits");
        assert!(
            GitServe::select(&store, &[&oid], &[]).unwrap().is_none(),
            "a store with no commit graph must decline, not answer with the tip alone"
        );

        // And `emit_pack` is INDEPENDENT of that decline: it was handed one
        // object and it emits one object, out of the very store whose
        // projection cannot answer a thing. A single blob is a legitimate set
        // to emit — it is what a `--filter` fetch of one path looks like — and
        // an engine that consulted `select` here would refuse it.
        let mut sink = Vec::new();
        let stats = GitServe::emit_pack(&store, &[&oid], &[], &Caps::modern(), &mut sink).unwrap();
        assert_eq!(
            stats.objects, 1,
            "emit_pack emits what it is given; it does not ask select whether it may"
        );
        assert_eq!(
            u32::from_be_bytes(sink[8..12].try_into().unwrap()),
            1,
            "the emitted pack header must count what the receipt counts"
        );
    }

    /// One entry of a hand-built pack.
    ///
    /// [`one_blob_pack`] cannot express a graph and [`real_pack`] is whatever
    /// this machine happens to have. Two properties below need a pack whose
    /// **exact** shape is chosen rather than found: which objects point at which,
    /// and — for the narrowed-clone test — which entry is stored as a delta
    /// against which other entry.
    enum Item {
        Whole(GitObjectKind, Vec<u8>),
        /// An `OFS_DELTA` against the item at index `base`, reconstructing
        /// `body`.
        OfsDelta { base: usize, body: Vec<u8> },
    }

    impl Item {
        /// The resolved body, which is what a delta against this item states as
        /// its source size.
        fn body(&self) -> &[u8] {
            match self {
                Item::Whole(_, b) | Item::OfsDelta { body: b, .. } => b,
            }
        }
    }

    fn deflate(bytes: &[u8]) -> Vec<u8> {
        use std::io::Write;
        let mut e = flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::default());
        e.write_all(bytes).unwrap();
        e.finish().unwrap()
    }

    /// git's delta-header varint — LEB128, unlike the `OFS_DELTA` distance.
    fn delta_varint(out: &mut Vec<u8>, mut n: u64) {
        loop {
            let mut b = (n & 0x7f) as u8;
            n >>= 7;
            if n > 0 {
                b |= 0x80;
            }
            out.push(b);
            if n == 0 {
                return;
            }
        }
    }

    /// A delta stream that rebuilds `target` by **pure insertion**.
    ///
    /// Legal git delta encoding, and the reason it is used here rather than a
    /// copy-based one: the base's *content* becomes irrelevant while its
    /// *identity* stays load-bearing, so the fixture can name any base it likes
    /// and still produce a target of its own choosing. `source_size` must still
    /// match the base exactly — git checks it and refuses otherwise.
    fn insert_only_delta(base_len: usize, target: &[u8]) -> Vec<u8> {
        let mut d = Vec::new();
        delta_varint(&mut d, base_len as u64);
        delta_varint(&mut d, target.len() as u64);
        for chunk in target.chunks(0x7f) {
            d.push(chunk.len() as u8);
            d.extend_from_slice(chunk);
        }
        d
    }

    /// Assemble `items` into a packfile, in the order given.
    ///
    /// The trailer is zeroed: nothing on the absorb path verifies it, and a
    /// fixture that had to be re-hashed on every edit would be a second thing to
    /// get wrong.
    fn build_pack(items: &[Item]) -> Vec<u8> {
        let mut pack = b"PACK".to_vec();
        pack.extend_from_slice(&2u32.to_be_bytes());
        pack.extend_from_slice(&(items.len() as u32).to_be_bytes());
        let mut offsets: Vec<u64> = Vec::with_capacity(items.len());
        for (i, item) in items.iter().enumerate() {
            offsets.push(pack.len() as u64);
            match item {
                Item::Whole(kind, body) => {
                    crate::pack_walk::encode_type_and_size(
                        &mut pack,
                        resolved_type(*kind),
                        body.len() as u64,
                    );
                    pack.extend_from_slice(&deflate(body));
                }
                Item::OfsDelta { base, body } => {
                    let delta = insert_only_delta(items[*base].body().len(), body);
                    crate::pack_walk::encode_type_and_size(
                        &mut pack,
                        ObjType::OfsDelta,
                        delta.len() as u64,
                    );
                    crate::pack_walk::encode_ofs_distance(&mut pack, offsets[i] - offsets[*base]);
                    pack.extend_from_slice(&deflate(&delta));
                }
            }
        }
        pack.extend_from_slice(&[0u8; 20]);
        pack
    }

    /// A whole-object pack: every entry whole — no `OFS_DELTA`, no `REF_DELTA` —
    /// which is what makes "the emitted count equals the requested count" a
    /// statement about reachability closure and not about delta-base handling.
    fn whole_object_pack(objects: &[(GitObjectKind, Vec<u8>)]) -> Vec<u8> {
        let items: Vec<Item> = objects
            .iter()
            .map(|(k, b)| Item::Whole(*k, b.clone()))
            .collect();
        build_pack(&items)
    }

    /// One commit, its tree, and two blobs under that tree — the smallest graph
    /// in which the closure of a subset is strictly larger than the subset.
    ///
    /// Returns `(pack, commit_oid, tree_oid, blob_oids)`, all raw (not hex).
    fn commit_tree_two_blobs() -> (Vec<u8>, Vec<u8>, Vec<u8>, Vec<Vec<u8>>) {
        let hash = GitHashKind::Sha1;
        let bodies: [&[u8]; 2] = [b"first blob", b"second blob"];
        let mut blobs = Vec::new();
        let mut objects = Vec::new();
        let mut tree_payload = Vec::new();
        for (name, body) in ["a.txt", "b.txt"].iter().zip(bodies) {
            let canon = canonical(GitObjectKind::Blob, body);
            let oid = hash.oid_of(&canon);
            tree_payload.extend_from_slice(format!("100644 {name}\0").as_bytes());
            tree_payload.extend_from_slice(&oid);
            objects.push((GitObjectKind::Blob, body.to_vec()));
            blobs.push(oid);
        }
        let tree_canon = canonical(GitObjectKind::Tree, &tree_payload);
        let tree_oid = hash.oid_of(&tree_canon);

        let body = format!(
            "tree {}\nauthor A <a@x> 1700000000 +0000\ncommitter A <a@x> 1700000000 +0000\n\n\
             the closure fixture\n",
            hex::encode(&tree_oid)
        );
        let commit_canon = canonical(GitObjectKind::Commit, body.as_bytes());
        let commit_oid = hash.oid_of(&commit_canon);

        objects.push((GitObjectKind::Tree, tree_payload));
        objects.push((GitObjectKind::Commit, body.into_bytes()));
        (whole_object_pack(&objects), commit_oid, tree_oid, blobs)
    }

    /// 🔴 **`emit_pack` emits the set it is HANDED, and never that set's
    /// reachability closure.**
    ///
    /// This is the guard for the defect the parameter rename names. Until
    /// 2026-08-10 this method read
    ///
    /// ```text
    ///     let Some(set) = self.select(objects, have)? else { bail!(..) };
    ///     self.emit_oids(&set.objects, caps, out)
    /// ```
    ///
    /// — it **closed over its own input**. Upload-pack hands it
    /// `selection.objects`, which is already post-filter, post-shallow and
    /// post-`include-tag` and is *deliberately not closed*, so the closure added
    /// back exactly what `--filter=blob:none`, `--filter=tree:0` or `--depth=N`
    /// had excluded.
    ///
    /// **Nothing else in the tree can see that.** The over-sent pack passes
    /// `git index-pack --strict` and `git fsck`, the clone succeeds, the exit
    /// code is zero, and the client silently receives objects it asked not to
    /// have — `P-027`'s shape, a change no test can see. So the assertion is on
    /// a **count**: never on bytes, never on a clock.
    ///
    /// Two shapes, because the closure broke the method in two different
    /// directions and both had to stop:
    ///
    /// 1. `[commit]` alone — a real `--filter=tree:0` fetch. **RED before the
    ///    fix: `objects` was 4, not 1** (the tree and both blobs came back).
    /// 2. `[commit, tree]` — a set whose members are not all commits. **RED
    ///    before the fix with a refusal, not a count**: the old body routed
    ///    through `select`, whose tip check rejects a non-commit tip, so a
    ///    perfectly ordinary partial-clone selection could not be emitted at
    ///    all.
    ///
    /// The fixture asserts its own premise first — that the closure really is
    /// strictly larger than either input — or it would prove nothing on a
    /// repository where the two happened to coincide.
    #[test]
    fn emit_pack_emits_exactly_the_set_it_is_given_and_never_its_closure() {
        let dir = tmpdir("serve-emit-exact");
        let store = GitStore::open(&dir, "rickard").unwrap();
        let (pack, commit, tree, blobs) = commit_tree_two_blobs();
        store.put(&pack, &[]).unwrap();
        store.wait_indexed();
        store.absorb_pending().unwrap();

        // The premise. Without it the two assertions below are satisfied by a
        // store that happens to hold nothing beneath the commit.
        assert_eq!(
            store.commit_count(),
            1,
            "the fixture must have a commit graph"
        );
        let closure = GitServe::select(&store, &[&commit], &[])
            .unwrap()
            .expect("the projection covers a commit it folded");
        assert_eq!(
            closure.objects.len(),
            4,
            "the fixture's closure must be strictly larger than the subsets emitted below, or \
             this test cannot tell an exact emission from a closed one: {:?}",
            closure.objects.iter().map(hex::encode).collect::<Vec<_>>()
        );
        for b in &blobs {
            assert!(
                closure.objects.contains(b),
                "the closure must reach the blobs — that is what must NOT be emitted"
            );
        }
        // The mechanism behind shape 2 below, pinned as a live fact rather than
        // left as history: `select` declines a tip that is not a commit in its
        // graph, which is correct for `select` and is exactly why routing
        // `emit_pack` through it made an ordinary selection unservable.
        assert!(
            GitServe::select(&store, &[&commit, &tree], &[])
                .unwrap()
                .is_none(),
            "select is a close-over-TIPS question and a tree is not a tip it can answer for"
        );

        // 1. `--filter=tree:0`: the commit and nothing else.
        let mut out = Vec::new();
        let stats =
            GitServe::emit_pack(&store, &[&commit], &[], &Caps::modern(), &mut out).unwrap();
        assert_eq!(
            stats.objects, 1,
            "emit_pack was handed ONE object and must emit ONE; a closure here re-adds exactly \
             what the filter excluded, and index-pack --strict and fsck both accept the result"
        );
        // The receipt and the bytes must agree: the pack's own object-count
        // field is bytes 8..12, and reading it back is the check that the
        // counter is not simply reporting what it was asked for.
        assert_eq!(
            u32::from_be_bytes(out[8..12].try_into().unwrap()),
            1,
            "the emitted pack header must count what the receipt counts"
        );

        // 2. A post-filter selection whose members are not all commits.
        let mut out = Vec::new();
        let stats =
            GitServe::emit_pack(&store, &[&commit, &tree], &[], &Caps::modern(), &mut out).unwrap();
        assert_eq!(
            stats.objects, 2,
            "emit_pack was handed TWO objects and must emit TWO"
        );
        assert_eq!(
            u32::from_be_bytes(out[8..12].try_into().unwrap()),
            2,
            "the emitted pack header must count what the receipt counts"
        );
        assert_eq!(stats.copied, stats.objects, "still a byte-range copy");
        assert_eq!(stats.recompressed, 0, "still nothing re-deflated");
    }

    /// Two branches whose trees are near-identical, with the **second branch's
    /// tree stored as a delta against the first branch's tree**.
    ///
    /// That one fact is the whole fixture. Narrow a request to `base` and it
    /// contains a delta whose base is not in it — and the base is a *tree*, so
    /// pulling it in would drag an object owing children the request does not
    /// contain. It is the `h2h-linear-sha1-2048c-1024f-16k` failure in six
    /// objects.
    ///
    /// `shared_entries` is how many tree entries the two trees hold in **common**
    /// before the one they differ in. It is `0` for the connectivity test, which
    /// wants the smallest fixture that has the defect in it, and large for the
    /// thin-pack test, which has to be able to see the delta *win* in bytes: at
    /// 0 the whole tree is 33 bytes and a `REF_DELTA`'s 20-byte base oid costs
    /// more than the delta saves, so a byte assertion there would measure the
    /// fixture rather than the mechanism. The shared entries all name `b_main`,
    /// so nothing new becomes reachable and no tree owes a child that is not in
    /// one of the two closures.
    ///
    /// Returns the pack and, in order, `(c_main, t_main, b_main, c_base,
    /// t_base, b_base)`, all raw.
    #[allow(clippy::type_complexity)]
    fn two_branches_with_a_cross_branch_delta(shared_entries: usize) -> (Vec<u8>, [Vec<u8>; 6]) {
        let hash = GitHashKind::Sha1;
        let oid_of = |kind, body: &[u8]| hash.oid_of(&canonical(kind, body));

        let b_main = b"the payload that only main can reach\n".to_vec();
        let b_main_oid = oid_of(GitObjectKind::Blob, &b_main);
        let b_base = b"the payload that only base can reach\n".to_vec();
        let b_base_oid = oid_of(GitObjectKind::Blob, &b_base);

        let tree_for = |blob: &[u8]| {
            let mut t = Vec::new();
            // Sorted, and `zz.txt` sorts after every `fNN.txt`: an unsorted tree
            // is what `fsck --strict` rejects, and the oracles below run it.
            for i in 0..shared_entries {
                t.extend_from_slice(format!("100644 f{i:03}.txt\0").as_bytes());
                t.extend_from_slice(&b_main_oid);
            }
            t.extend_from_slice(b"100644 zz.txt\0");
            t.extend_from_slice(blob);
            t
        };
        let t_main = tree_for(&b_main_oid);
        let t_main_oid = oid_of(GitObjectKind::Tree, &t_main);
        let t_base = tree_for(&b_base_oid);
        let t_base_oid = oid_of(GitObjectKind::Tree, &t_base);

        let commit_for = |tree: &[u8], msg: &str| {
            format!(
                "tree {}\nauthor A <a@x> 1700000000 +0000\ncommitter A <a@x> 1700000000 +0000\n\n\
                 {msg}\n",
                hex::encode(tree)
            )
            .into_bytes()
        };
        let c_main = commit_for(&t_main_oid, "main");
        let c_main_oid = oid_of(GitObjectKind::Commit, &c_main);
        let c_base = commit_for(&t_base_oid, "base");
        let c_base_oid = oid_of(GitObjectKind::Commit, &c_base);

        // `t_base` is entry 4 and deltas against entry 1, `t_main`. Both trees
        // are 33 bytes and differ in 20 of them, which is what makes this the
        // delta a real packer would also choose.
        let pack = build_pack(&[
            Item::Whole(GitObjectKind::Blob, b_main),
            Item::Whole(GitObjectKind::Tree, t_main),
            Item::Whole(GitObjectKind::Commit, c_main),
            Item::Whole(GitObjectKind::Blob, b_base),
            Item::OfsDelta {
                base: 1,
                body: t_base,
            },
            Item::Whole(GitObjectKind::Commit, c_base),
        ]);
        (
            pack,
            [
                c_main_oid, t_main_oid, b_main_oid, c_base_oid, t_base_oid, b_base_oid,
            ],
        )
    }

    /// 🔴 **A NARROWED clone gets a pack that is self-contained *and connected*
    /// — the delta base is not smuggled in with it.**
    ///
    /// The defect, seen live on 2026-08-11 against
    /// `h2h-linear-sha1-2048c-1024f-16k`:
    ///
    /// ```text
    /// git clone --bare <url>                          → 36 172 objects, OK
    /// git clone --bare --single-branch --branch base  → FAILS
    ///     fatal: did not receive expected object 8601ec33920b7d701c9887fa04916501136d6e90
    ///     fatal: fetch-pack: invalid index-pack output
    /// event="git.upload_pack.served" objects=33773    ← the server logged SUCCESS
    /// ```
    ///
    /// `base` reaches 31 805 objects; `emit_set`'s delta-base closure added
    /// 1 968 more, giving exactly the 33 773 the server logged. **507 of the
    /// additions were trees**, and between them they named 204 objects the pack
    /// did not contain. `git index-pack --check-self-contained-and-connected` —
    /// what a clone runs, and which turns on `strict` — walks every received
    /// object's links and demands each one exist, so it died on the first, and
    /// nothing on the server ever knew.
    ///
    /// # Why the assertions are shaped the way they are
    ///
    /// The premise is asserted first and it is the load-bearing one: the request
    /// must actually contain a delta whose base is outside it. On a corpus where
    /// that happens not to hold — which is every full clone, by construction —
    /// this test cannot fail no matter what the code does.
    ///
    /// Then the **count**, not an exit code: the emitted pack must hold exactly
    /// the three objects `base` reaches. **RED before the fix at 4**, the fourth
    /// being `t_main`.
    ///
    /// Then stock git, in a **repository**: `index-pack --strict` outside one
    /// segfaults instead of judging (exit 139, no output), which is a green that
    /// was never looked at. Inside a fresh bare repo it is the exact arbiter that
    /// failed in production, and before the fix it prints `did not receive
    /// expected object <b_main>`.
    #[test]
    fn a_narrowed_clone_is_served_a_connected_pack_not_its_delta_bases() {
        let dir = tmpdir("serve-narrowed");
        let store = GitStore::open(&dir, "rickard").unwrap();
        let (pack, [c_main, t_main, b_main, c_base, t_base, b_base]) =
            two_branches_with_a_cross_branch_delta(0);
        store.put(&pack, &[]).unwrap();
        store.wait_indexed();
        store.absorb_pending().unwrap();
        assert_eq!(store.commit_count(), 2, "both branches must be in the graph");

        // ── the narrowed request ──────────────────────────────────────────────
        let set = GitServe::select(&store, &[&c_base], &[])
            .unwrap()
            .expect("the projection covers a commit it folded");
        let mut want = set.objects.clone();
        want.sort();
        let mut expected = vec![c_base.clone(), t_base.clone(), b_base.clone()];
        expected.sort();
        assert_eq!(
            want,
            expected,
            "`base` reaches its own commit, tree and blob and nothing of main's"
        );

        // ── THE PREMISE, or this test proves nothing ──────────────────────────
        //
        // `t_base` must really be stored as a delta against `t_main`, and
        // `t_main` must really be outside the request. Without both, an
        // implementation that pulls bases in is indistinguishable from one that
        // does not.
        let row = store
            .index()
            .lookup(&t_base)
            .expect("the tree base reaches is indexed");
        assert_eq!(
            row.obj_type,
            ObjType::OfsDelta,
            "the fixture must store base's tree as a delta or there is no base to leave out"
        );
        let base_row = store
            .index()
            .lookup(&t_main)
            .expect("the tree main reaches is indexed");
        assert_eq!(
            row.delta_base, base_row.offset,
            "the fixture's delta must name main's tree as its base"
        );
        assert!(
            !set.objects.contains(&t_main),
            "the base must be OUTSIDE the request, or nothing is being closed over"
        );

        // ── the count ─────────────────────────────────────────────────────────
        let oids: Vec<Oid<'_>> = set.objects.iter().map(Vec::as_slice).collect();
        let mut out = Vec::new();
        let stats = store.emit_oids(&oids, &[], &Caps::modern(), &mut out).unwrap();
        assert_eq!(
            stats.objects, 3,
            "a narrowed clone must be served exactly the objects it reaches; adding the delta \
             base makes it 4 and the fourth owes children the pack does not carry"
        );
        assert_eq!(
            u32::from_be_bytes(out[8..12].try_into().unwrap()),
            3,
            "the emitted pack header must count what the receipt counts"
        );
        // Applied output for the mechanism itself: two entries copied byte for
        // byte, and exactly the one whose base was left out rebuilt.
        assert_eq!(stats.copied, 2, "everything but the boundary entry is copied");
        assert_eq!(
            stats.recompressed, 1,
            "the entry whose base is outside the request is the one that must be rebuilt whole"
        );
        assert_eq!(stats.copied + stats.recompressed, stats.objects);

        // ── and the object main reaches must not be in the bytes ──────────────
        let walked = crate::pack_walk::walk(&out, 20).expect("our own walk reads what we emitted");
        assert_eq!(walked.entries.len(), 3);
        assert!(
            walked.entries.iter().all(|e| e.obj_type != ObjType::OfsDelta),
            "the boundary entry must be whole, not a delta naming a base that is not here"
        );

        // ── stock git, inside a repository, as the arbiter ────────────────────
        let scratch = tmpdir("serve-narrowed-idx");
        crate::git_oracle::assert_git_accepts(
            &scratch,
            "dst.git",
            &out,
            crate::git_oracle::Strictness::Connected,
        );

        // The full clone is unaffected, and that is asserted rather than
        // assumed: every base is inside a whole-repository request, so nothing
        // is rebuilt and the copy claim survives intact.
        let all = vec![
            c_main.as_slice(),
            t_main.as_slice(),
            b_main.as_slice(),
            c_base.as_slice(),
            t_base.as_slice(),
            b_base.as_slice(),
        ];
        let mut full = Vec::new();
        let stats = store.emit_oids(&all, &[], &Caps::modern(), &mut full).unwrap();
        assert_eq!(stats.objects, 6);
        assert_eq!(
            stats.recompressed, 0,
            "a full clone contains every base, so it must still be a pure byte-range copy"
        );
        assert_eq!(stats.copied, 6);
    }

    /// **Three revisions of one 16 KiB blob, the middle one off-branch**, which
    /// is the `h2h-linear-sha1-2048c-1024f-16k` boundary entry in nine objects.
    ///
    /// The chain is `v1 <- v2 <- v3`, stored in that order, and only `v2` is
    /// off-branch. So a request for `base` holds `v1` and `v3` and not `v2`:
    /// `v3`'s stored base is outside it, and `v3`'s **chain ancestor `v1` is
    /// inside it**. That second fact is the whole fixture — without it the
    /// re-delta has nothing to aim at, and the measured production shape is
    /// exactly this one (859 of 961 boundary entries have such an ancestor).
    ///
    /// The blob bodies are incompressible noise with a small splice between
    /// revisions, deliberately: on compressible filler a whole rebuild is nearly
    /// free and a byte assertion would be measuring zlib rather than the delta.
    /// At 16 KiB of noise, whole costs ~16.5 KB on the wire and a delta costs a
    /// few hundred bytes, so the two cannot be confused.
    ///
    /// Returns the pack and `(c_base_tip, t3, v3, c1, t1, v1, c_main, t2, v2)`.
    #[allow(clippy::type_complexity)]
    fn three_revisions_with_the_middle_one_off_branch() -> (Vec<u8>, [Vec<u8>; 9]) {
        let hash = GitHashKind::Sha1;
        let oid_of = |kind, body: &[u8]| hash.oid_of(&canonical(kind, body));

        // xorshift, so the bytes are reproducible and do not deflate.
        let noise = |n: usize, seed: u64| -> Vec<u8> {
            let mut s = seed | 1;
            (0..n)
                .map(|_| {
                    s ^= s << 13;
                    s ^= s >> 7;
                    s ^= s << 17;
                    (s >> 33) as u8
                })
                .collect::<Vec<u8>>()
        };
        let v1 = noise(16 * 1024, 0xC0FFEE);
        let mut v2 = v1.clone();
        v2[2048..2112].copy_from_slice(&noise(64, 2));
        let mut v3 = v2.clone();
        v3[8192..8256].copy_from_slice(&noise(64, 3));

        let v1_oid = oid_of(GitObjectKind::Blob, &v1);
        let v2_oid = oid_of(GitObjectKind::Blob, &v2);
        let v3_oid = oid_of(GitObjectKind::Blob, &v3);

        let tree_for = |blob: &[u8]| {
            let mut t = b"100644 f.bin\0".to_vec();
            t.extend_from_slice(blob);
            t
        };
        let (t1, t2, t3) = (tree_for(&v1_oid), tree_for(&v2_oid), tree_for(&v3_oid));
        let (t1_oid, t2_oid, t3_oid) = (
            oid_of(GitObjectKind::Tree, &t1),
            oid_of(GitObjectKind::Tree, &t2),
            oid_of(GitObjectKind::Tree, &t3),
        );

        let commit_for = |tree: &[u8], parent: Option<&[u8]>, msg: &str| {
            let mut c = format!("tree {}\n", hex::encode(tree));
            if let Some(p) = parent {
                c.push_str(&format!("parent {}\n", hex::encode(p)));
            }
            c.push_str(
                "author A <a@x> 1700000000 +0000\ncommitter A <a@x> 1700000000 +0000\n\n",
            );
            c.push_str(msg);
            c.push('\n');
            c.into_bytes()
        };
        let c1 = commit_for(&t1_oid, None, "r1");
        let c1_oid = oid_of(GitObjectKind::Commit, &c1);
        // `main`'s commit is parentless, so nothing of `base` becomes reachable
        // through it and the two closures stay the disjoint-plus-v1 shape the
        // assertions below rely on.
        let c_main = commit_for(&t2_oid, None, "r2 on main only");
        let c_main_oid = oid_of(GitObjectKind::Commit, &c_main);
        let c3 = commit_for(&t3_oid, Some(&c1_oid), "r3");
        let c3_oid = oid_of(GitObjectKind::Commit, &c3);

        // Entry 0 is `v1` whole; entry 1 is `v2` as an ofs-delta on it; entry 2
        // is `v3` as an ofs-delta on `v2`. Offsets increase along the chain,
        // which is the invariant the ancestor walk's acyclicity rests on.
        let pack = build_pack(&[
            Item::Whole(GitObjectKind::Blob, v1),
            Item::OfsDelta { base: 0, body: v2 },
            Item::OfsDelta { base: 1, body: v3 },
            Item::Whole(GitObjectKind::Tree, t1),
            Item::Whole(GitObjectKind::Tree, t2),
            Item::Whole(GitObjectKind::Tree, t3),
            Item::Whole(GitObjectKind::Commit, c1),
            Item::Whole(GitObjectKind::Commit, c_main),
            Item::Whole(GitObjectKind::Commit, c3),
        ]);
        (
            pack,
            [
                c3_oid, t3_oid, v3_oid, c1_oid, t1_oid, v1_oid, c_main_oid, t2_oid, v2_oid,
            ],
        )
    }

    /// 🔴 **A narrowed clone re-deltas its boundary entry against a base the
    /// pack DOES carry, instead of shipping it whole — and the pack still holds
    /// exactly the same objects, byte for byte.**
    ///
    /// # The gap this closes, measured before it existed
    ///
    /// `h2h-linear-sha1-2048c-1024f-16k`, one server process, oden 2026-08-11:
    /// a narrowed clone of `base` was **11 613 666** bytes for 31 805 objects
    /// where stock git sends 5 820 000 for the identical set — 2.0×. The cause
    /// is 961 entries (3.0 %) whose stored delta base falls outside the request:
    /// they are 16 KiB blobs stored as ~284-byte deltas, and shipping them whole
    /// costs ~5.8 KB each. Thin-pack narrowing (`d2ebb2c`) fixed the *fetch*
    /// half and provably cannot touch this one — a clone's receiver holds
    /// nothing, so there is no external base to name.
    ///
    /// # What is asserted, and why each part is needed
    ///
    /// * **The premise first.** `v3` must really be stored as a delta on `v2`,
    ///   `v2` must really be outside the request, and `v1` must really be inside
    ///   it and be `v2`'s base. On a corpus where any of those fails, every
    ///   implementation produces the same pack and this test is decoration.
    /// * **The count, held equal.** A smaller pack that dropped an object would
    ///   satisfy a byte assertion perfectly. The count is checked in the
    ///   receipt, in the pack's own header field, and in what git reads back.
    /// * **The bytes.** The load-bearing number. Seen red at **33 970** bytes
    ///   with `ZNIPPY_GIT_BOUNDARY_DELTA=0` — which is the pre-change behaviour
    ///   in this same binary, against the same store — and green at **16 839**.
    ///   The 2.02× between them is the same ratio the production fixture shows.
    /// * **The object graph, out of stock git.** A wrong copy offset produces a
    ///   pack `index-pack` accepts and whose objects are *different* — it files
    ///   each entry under the oid of whatever it decoded. Only reading the bytes
    ///   back and comparing them against the store catches it, so
    ///   [`crate::git_oracle::git_reads_back`] does exactly that for all six.
    /// * **The receipt.** `copied + recompressed == objects` still holds, and
    ///   the rebuilt entry is `deltified` rather than whole. Both are counted
    ///   off what happened; a byte count cannot tell the two rebuilds apart.
    #[test]
    fn a_narrowed_clone_re_deltas_its_boundary_entry_against_a_base_the_pack_carries() {
        let dir = tmpdir("serve-reboundary");
        let store = GitStore::open(&dir, "rickard").unwrap();
        let (pack, [c3, t3, v3, c1, t1, v1, c_main, _t2, v2]) =
            three_revisions_with_the_middle_one_off_branch();
        store.put(&pack, &[]).unwrap();
        store.wait_indexed();
        store.absorb_pending().unwrap();
        assert_eq!(store.commit_count(), 3, "all three commits must be folded");

        let set = GitServe::select(&store, &[&c3], &[])
            .unwrap()
            .expect("the projection covers a commit it folded");
        let mut want = set.objects.clone();
        want.sort();
        let mut expected = vec![
            c3.clone(),
            t3.clone(),
            v3.clone(),
            c1.clone(),
            t1.clone(),
            v1.clone(),
        ];
        expected.sort();
        assert_eq!(want, expected, "`base` reaches r1 and r3 and nothing of main");

        // ── THE PREMISE ───────────────────────────────────────────────────────
        let row_v3 = store.index().lookup(&v3).expect("v3 is indexed");
        let row_v2 = store.index().lookup(&v2).expect("v2 is indexed");
        let row_v1 = store.index().lookup(&v1).expect("v1 is indexed");
        assert_eq!(
            row_v3.obj_type,
            ObjType::OfsDelta,
            "v3 must be stored as a delta or there is no boundary entry"
        );
        assert_eq!(
            row_v3.delta_base, row_v2.offset,
            "v3's stored base must be v2 — the off-branch revision"
        );
        assert!(
            !set.objects.contains(&v2),
            "v2 must be OUTSIDE the request, or nothing is re-deltified"
        );
        assert_eq!(
            row_v2.delta_base, row_v1.offset,
            "v2's own base must be v1, or the chain ancestor this aims at does not exist"
        );
        assert!(
            set.objects.contains(&v1),
            "v1 must be INSIDE the request — it is the base the re-delta names"
        );
        assert!(
            row_v1.offset < row_v3.offset,
            "the ancestor must sit earlier in the archive; that ordering is what makes the new \
             delta edge acyclic by construction"
        );
        assert!(
            !c_main.is_empty(),
            "main's commit exists and is what keeps v2 in the repository but out of the request"
        );

        // ── the emission ──────────────────────────────────────────────────────
        let oids: Vec<Oid<'_>> = set.objects.iter().map(Vec::as_slice).collect();
        let mut out = Vec::new();
        let stats = store.emit_oids(&oids, &[], &Caps::modern(), &mut out).unwrap();

        assert_eq!(stats.objects, 6, "the count is held equal");
        assert_eq!(
            u32::from_be_bytes(out[8..12].try_into().unwrap()),
            6,
            "the emitted pack header must count what the receipt counts"
        );

        // ── THE BYTES ─────────────────────────────────────────────────────────
        //
        // 🔴 **33 970 with `ZNIPPY_GIT_BOUNDARY_DELTA=0`, 16 839 with it on** —
        // measured, same binary, same store. The floor is `v1` itself: it is
        // 16 KiB of noise and it is *copied* in both arms, so the pack can never
        // be small. What halves is the second copy of the same 16 KiB that `v3`
        // used to be, and the threshold sits between the two behaviours rather
        // than on the green figure, so a zlib version bump moves the number
        // without moving the verdict.
        assert!(
            stats.bytes < 20_000,
            "a 16 KiB boundary blob with an in-request chain ancestor must go out as a delta, not \
             whole: the pack is {} bytes, and shipping it whole measures 33 970",
            stats.bytes
        );
        assert_eq!(
            stats.bytes as usize,
            out.len(),
            "the receipt's byte count must be what was actually written"
        );

        // ── THE RECEIPT ───────────────────────────────────────────────────────
        assert_eq!(
            stats.copied + stats.recompressed,
            stats.objects,
            "every entry is one or the other, always"
        );
        assert_eq!(
            stats.recompressed, 1,
            "exactly the boundary entry is rebuilt; a re-delta is still a rebuild and must not be \
             counted as a copy"
        );
        let entries = store.emit_set(&oids, true, None).unwrap();
        let deltified: Vec<&crate::pack_walk::EmitEntry> =
            entries.iter().filter(|e| e.deltified).collect();
        assert_eq!(deltified.len(), 1, "one entry, and it is the boundary one");
        assert_eq!(deltified[0].oid, v3, "and it is v3");
        assert_eq!(
            deltified[0].delta_base, row_v1.offset,
            "the computed delta must name the in-request chain ancestor as its base"
        );
        // The mechanism, isolated from `v1`'s unavoidable 16 KiB: the entry
        // itself. Whole it is ~16.5 KB deflated; against `v1` it is a few
        // hundred bytes, and the gap between those two is the entire result.
        assert!(
            deltified[0].stored.len() < 1_000,
            "the re-deltified entry is {} bytes; whole it is ~16 500, and anything in between \
             means the delta was computed against the wrong base",
            deltified[0].stored.len()
        );
        assert!(
            entries.iter().all(|e| !e.deltified || e.recompressed),
            "`deltified` is a strict refinement of `recompressed`"
        );

        // ── THE OBJECT GRAPH, out of stock git ────────────────────────────────
        let scratch = tmpdir("serve-reboundary-idx");
        crate::git_oracle::assert_git_accepts(
            &scratch,
            "dst.git",
            &out,
            crate::git_oracle::Strictness::Connected,
        );
        let read_back = crate::git_oracle::git_reads_back(&scratch, "readback.git", &out)
            .expect("stock git reads back the pack it just accepted");
        assert_eq!(read_back.len(), 6, "git must hold exactly the six objects");
        for (oid_hex, body) in &read_back {
            let raw = hex::decode(oid_hex).expect("git prints hex oids");
            let (_, ours) = store
                .content(&raw)
                .unwrap()
                .unwrap_or_else(|| panic!("git read back {oid_hex}, which this store does not hold"));
            assert_eq!(
                &ours, body,
                "{oid_hex} came back from git with different bytes than the store holds — a \
                 computed delta with a wrong copy offset produces exactly this and passes \
                 index-pack"
            );
        }

        // ── and a full clone is untouched ─────────────────────────────────────
        let all: Vec<Oid<'_>> = [&c3, &t3, &v3, &c1, &t1, &v1, &c_main, &_t2, &v2]
            .iter()
            .map(|o| o.as_slice())
            .collect();
        let mut full = Vec::new();
        let stats = store.emit_oids(&all, &[], &Caps::modern(), &mut full).unwrap();
        assert_eq!(stats.objects, 9);
        assert_eq!(
            stats.recompressed, 0,
            "a whole-repository request contains every base, so nothing is rebuilt and nothing is \
             deltified — the copy claim is unchanged"
        );
        assert_eq!(stats.copied, 9);
    }

    /// A fresh bare repository, and the objects `pack` carries unpacked into it.
    ///
    /// `git unpack-objects` rather than `index-pack`, because the point is a
    /// receiver that **holds** these objects, not one that has a pack file
    /// sitting next to its repository.
    fn bare_repo_holding(scratch: &std::path::Path, name: &str, pack: &[u8]) -> std::path::PathBuf {
        use std::io::Write as _;
        let repo = scratch.join(name);
        let init = std::process::Command::new("git")
            .args(["init", "-q", "--bare"])
            .arg(&repo)
            .output()
            .expect("running git init");
        assert!(init.status.success(), "git init failed");
        if !pack.is_empty() {
            let mut child = std::process::Command::new("git")
                .args(["unpack-objects", "-q"])
                .current_dir(&repo)
                .stdin(std::process::Stdio::piped())
                .stdout(std::process::Stdio::piped())
                .stderr(std::process::Stdio::piped())
                .spawn()
                .expect("running git unpack-objects");
            child.stdin.take().unwrap().write_all(pack).unwrap();
            let done = child.wait_with_output().unwrap();
            assert!(
                done.status.success(),
                "seeding the receiver failed:\n{}",
                String::from_utf8_lossy(&done.stderr)
            );
        }
        repo
    }

    /// `git index-pack --stdin --fix-thin --strict` inside `repo`. The exact
    /// command a fetching client runs when it advertised `thin-pack`, and the
    /// only one that can judge a pack whose bases are somewhere else.
    fn fix_thin_into(repo: &std::path::Path, pack: &[u8]) -> std::process::Output {
        use std::io::Write as _;
        let mut child = std::process::Command::new("git")
            .args(["index-pack", "--stdin", "--fix-thin", "--strict"])
            .current_dir(repo)
            .stdin(std::process::Stdio::piped())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped())
            .spawn()
            .expect("running git index-pack --fix-thin");
        child.stdin.take().unwrap().write_all(pack).unwrap();
        child.wait_with_output().unwrap()
    }

    /// 🔴 **A thin FETCH names a base the receiver already holds instead of
    /// rebuilding the boundary entry whole — and stock git resolves it.**
    ///
    /// This is the cost `8c2679d` left behind, for the half of it that can be
    /// paid cheaply. A narrowed request cuts delta chains, and until now every
    /// entry on the cut went out **whole**: correct, connected, and measured at
    /// 11.6 MB against git's 5.82 MB on `h2h-linear-sha1-2048c-1024f-16k`. When
    /// the receiver already holds the base, none of that is necessary — a
    /// `REF_DELTA` naming an external base is exactly what `thin-pack` means,
    /// and the stored delta stream goes out unchanged.
    ///
    /// # Four assertions, and each one can fail on its own
    ///
    /// 1. **The premise.** `t_base` is stored as an `OFS_DELTA` on `t_main`,
    ///    `t_main` is outside the request, and the client holds it. Without all
    ///    three there is no thin entry to make and the rest proves nothing.
    /// 2. **The receipt.** `recompressed` must be **0** — RED before this
    ///    change at **1**, because the boundary entry was rebuilt.
    /// 3. **Applied output, on the bytes.** The emitted pack must contain a
    ///    `REF_DELTA` whose base oid is `t_main`, and `t_main` must **not** be
    ///    an entry in that pack. A receipt cannot tell a thin delta from a
    ///    copied one; the wire can.
    /// 4. **Stock git, both directions.** `index-pack --stdin --fix-thin
    ///    --strict` accepts it in a repository that holds `main`, and **fails**
    ///    in an empty one. The second half is what proves the pack is genuinely
    ///    thin rather than accidentally self-contained, and it is what makes the
    ///    first half worth anything.
    ///
    /// And the two arms that must NOT change: the same request without
    /// `caps.thin`, and the same request with no `have`, both still rebuild the
    /// boundary entry whole. A clone is the second of those.
    #[test]
    fn a_thin_fetch_names_a_base_the_client_holds_instead_of_rebuilding_it() {
        let dir = tmpdir("serve-thin");
        let store = GitStore::open(&dir, "rickard").unwrap();
        let (pack, [c_main, t_main, b_main, c_base, t_base, b_base]) =
            two_branches_with_a_cross_branch_delta(256);
        store.put(&pack, &[]).unwrap();
        store.wait_indexed();
        store.absorb_pending().unwrap();

        // The fetch: the client is on `main` and wants `base`.
        let set = GitServe::select(&store, &[&c_base], &[&c_main])
            .unwrap()
            .expect("the projection covers both commits it folded");
        let oids: Vec<Oid<'_>> = set.objects.iter().map(Vec::as_slice).collect();

        // ── THE PREMISE ───────────────────────────────────────────────────────
        assert!(
            !set.objects.contains(&t_main),
            "main's tree must be OUTSIDE the request or there is no external base"
        );
        assert!(
            set.objects.contains(&t_base),
            "base's tree must be IN the request — it is the boundary entry"
        );
        let row = store.index().lookup(&t_base).expect("indexed");
        assert_eq!(
            row.obj_type,
            ObjType::OfsDelta,
            "the fixture must store base's tree as a delta or there is no base to name"
        );
        assert_eq!(
            row.delta_base,
            store.index().lookup(&t_main).expect("indexed").offset,
            "the fixture's delta must name main's tree as its base"
        );
        assert!(
            set.client_has.contains(&t_main),
            "the negotiation must vouch that the client holds main's tree"
        );

        let thin_caps = Caps {
            thin: true,
            ofs_delta: true,
        };
        let have: Vec<Oid<'_>> = vec![&c_main];

        // ── the receipt ───────────────────────────────────────────────────────
        let mut thin = Vec::new();
        let stats = store
            .emit_oids(&oids, &have, &thin_caps, &mut thin)
            .unwrap();
        assert_eq!(stats.objects, 3, "the set is still exactly what was selected");
        assert_eq!(
            stats.recompressed, 0,
            "the boundary entry's base is one the client holds, so nothing may be rebuilt"
        );
        assert_eq!(stats.copied, 3, "every entry is a byte-for-byte copy");

        // ── applied output, on the emitted bytes ──────────────────────────────
        let walked = crate::pack_walk::walk(&thin, store.hash_kind().oid_len())
            .expect("our own walk reads what we emitted");
        assert_eq!(walked.entries.len(), 3);
        assert_eq!(
            walked.closure().external_refs,
            vec![t_main.clone()],
            "exactly one entry may name a base by oid, and it must be main's tree — which the \
             pack does not carry, because the pack carries only the three objects selected"
        );

        // ── stock git, both directions ────────────────────────────────────────
        let scratch = tmpdir("serve-thin-git");
        // What the client already has: main's three objects.
        let mut mains = Vec::new();
        store
            .emit_oids(
                &[&c_main, &t_main, &b_main],
                &[],
                &Caps::modern(),
                &mut mains,
            )
            .unwrap();
        let holder = bare_repo_holding(&scratch, "holder.git", &mains);
        let got = fix_thin_into(&holder, &thin);
        assert!(
            got.status.success(),
            "git refused a thin pack whose base it holds — status {:?}\n{}",
            got.status.code(),
            String::from_utf8_lossy(&got.stderr)
        );

        // …and it must FAIL where the base is absent — for the RIGHT reason.
        //
        // The receiver here holds `b_main` and nothing else, which is chosen
        // rather than convenient: an *empty* repository refuses this pack either
        // way, because `t_base` names `b_main` as a child and a strict index-pack
        // demands children exist, so a refusal there would prove connectivity and
        // say nothing about thinness. Holding exactly `b_main` makes the pack
        // connected and leaves only the delta base missing, so the only thing
        // git can complain about is the external base — and with the thin arm
        // switched off this same call **succeeds**, which is what makes it a
        // guard rather than a decoration.
        let blob_only = bare_repo_holding(&scratch, "blob-only.git", &{
            let mut p = Vec::new();
            store
                .emit_oids(&[&b_main], &[], &Caps::modern(), &mut p)
                .unwrap();
            p
        });
        let refused = fix_thin_into(&blob_only, &thin);
        assert!(
            !refused.status.success(),
            "a receiver without main's TREE accepted the pack, so nothing was actually \
             offered as external and this whole test is hollow"
        );
        let why = String::from_utf8_lossy(&refused.stderr).to_lowercase();
        assert!(
            why.contains("delta"),
            "the refusal must be about the delta base this pack does not carry, and it said: \
             {why}"
        );

        // ── and the two arms that must not have moved ─────────────────────────
        let mut not_thin = Vec::new();
        let stats = store
            .emit_oids(&oids, &have, &Caps::modern(), &mut not_thin)
            .unwrap();
        assert_eq!(
            stats.recompressed, 1,
            "a client that did not advertise thin-pack must still get the entry whole"
        );
        let mut no_have = Vec::new();
        let stats = store
            .emit_oids(&oids, &[], &thin_caps, &mut no_have)
            .unwrap();
        assert_eq!(
            stats.recompressed, 1,
            "consent with nothing negotiated is a clone, and a clone must be unchanged"
        );

        // The whole point, in bytes: the thin pack is smaller than the one that
        // rebuilt the entry.
        assert!(
            thin.len() < not_thin.len(),
            "a thin pack that is not smaller has bought nothing: {} vs {}",
            thin.len(),
            not_thin.len()
        );

        // And a FULL request is byte-identical whether or not thin is offered —
        // it has no base outside itself for any of this to apply to.
        let all: Vec<Oid<'_>> = vec![&c_main, &t_main, &b_main, &c_base, &t_base, &b_base];
        let mut full_thin = Vec::new();
        store
            .emit_oids(&all, &have, &thin_caps, &mut full_thin)
            .unwrap();
        let mut full_plain = Vec::new();
        store
            .emit_oids(&all, &[], &Caps::modern(), &mut full_plain)
            .unwrap();
        assert_eq!(
            full_thin, full_plain,
            "a full clone must be byte-identical with and without the thin allowance"
        );
    }

    /// **The emitted pack is a byte-range copy, and stock git accepts it.**
    ///
    /// Proven against `git index-pack --strict` rather than against this
    /// crate's own reader: our parser reading our writer would agree with itself
    /// even if both were wrong the same way.
    ///
    /// The receipt is asserted too, and that is not decoration: `copied ==
    /// objects` and `recompressed == 0` are the `P-001` applied-output
    /// assertion. A pipeline that inflated every object in order to deflate it
    /// again would pass `index-pack --strict` **and** `fsck` while sending a
    /// measured 18.4x the wire bytes, and only these two counters can see it.
    #[test]
    fn an_emitted_pack_is_copied_not_recompressed_and_stock_git_accepts_it() {
        let dir = tmpdir("serve-emit");
        let store = GitStore::open(&dir, "rickard").unwrap();
        let (pack, rows) = real_pack();
        store.put(&pack, &[]).unwrap();
        store.wait_indexed();
        store.absorb_pending().unwrap();

        let oids: Vec<Oid<'_>> = rows.iter().map(|r| r.oid.as_slice()).collect();
        let mut out = Vec::new();
        let stats = store.emit_oids(&oids, &[], &Caps::modern(), &mut out).unwrap();

        assert_eq!(
            stats.objects,
            rows.len() as u64,
            "every requested object must be emitted"
        );
        assert_eq!(
            stats.copied, stats.objects,
            "every payload must be COPIED, not re-deflated"
        );
        assert_eq!(stats.recompressed, 0, "nothing on this path re-deflates");
        assert_eq!(
            stats.bytes,
            out.len() as u64,
            "the receipt's byte count is what was written, not what was intended"
        );

        // 🔴 Through [`crate::git_oracle`], and that is the 2026-08-11 fix here:
        // this ran `index-pack --strict` in `scratch`, which is a plain
        // directory and not a repository, and outside a repository that command
        // **segfaults** on any pack it would have rejected — exit 139, no
        // output. The whole corpus is emitted, so the set is closed and
        // `Connected` is the honest question.
        let scratch = tmpdir("serve-emit-idx");
        crate::git_oracle::assert_git_accepts(
            &scratch,
            "emitted.git",
            &out,
            crate::git_oracle::Strictness::Connected,
        );
    }

    /// **A client without `ofs-delta` is SERVED — a ref-delta carrying the same
    /// bytes — and never handed an entry it cannot parse.**
    ///
    /// Until 2026-08-11 this asserted a refusal, because the engine copies
    /// stored entries and had no way to re-name a base; that is the
    /// `gunnar.clone_no_ofs_delta` arm, red with *"the selection contains at
    /// least one stored ofs-delta entry"*. It does have a way now, and it costs
    /// nothing: the two delta forms differ only in **how the base is named**, so
    /// re-heading an `OFS_DELTA` as a `REF_DELTA` is a header swap over an
    /// unchanged compressed payload.
    ///
    /// Both halves are asserted. Serving is not enough on its own — the
    /// silent-corruption direction is to emit the `OFS_DELTA` anyway and exit
    /// zero — so the emitted bytes are walked and every entry checked, and the
    /// receipt must still say `recompressed = 0` or the swap has quietly become
    /// a re-deflate.
    #[test]
    fn a_client_that_cannot_read_ofs_delta_is_served_ref_deltas_instead() {
        let dir = tmpdir("serve-caps");
        let store = GitStore::open(&dir, "rickard").unwrap();
        let (pack, rows) = real_pack();
        store.put(&pack, &[]).unwrap();
        store.wait_indexed();
        store.absorb_pending().unwrap();

        let stored_ofs = rows
            .iter()
            .filter(|r| r.stored_type == ObjType::OfsDelta)
            .count();
        assert!(
            stored_ofs > 0,
            "the corpus must contain an ofs-delta or this asserts nothing"
        );

        let oids: Vec<Oid<'_>> = rows.iter().map(|r| r.oid.as_slice()).collect();
        let caps = Caps {
            thin: false,
            ofs_delta: false,
        };
        let mut out = Vec::new();
        let stats = store.emit_oids(&oids, &[], &caps, &mut out).unwrap();
        assert_eq!(stats.objects, rows.len() as u64);
        assert_eq!(
            stats.recompressed, 0,
            "re-naming a base must not re-deflate a payload"
        );

        let walked = crate::pack_walk::walk(&out, store.hash_kind().oid_len())
            .expect("our own walk reads what we emitted");
        assert!(
            walked.entries.iter().all(|e| e.obj_type != ObjType::OfsDelta),
            "not one ofs-delta may reach a client that cannot parse one"
        );
        assert_eq!(
            walked
                .entries
                .iter()
                .filter(|e| e.obj_type == ObjType::RefDelta)
                .count(),
            stored_ofs
                + rows
                    .iter()
                    .filter(|r| r.stored_type == ObjType::RefDelta)
                    .count(),
            "every stored delta must still be a delta — re-headed, not flattened"
        );

        // And stock git reads it. The whole corpus is emitted, so the set is
        // closed and `Connected` is the question a clone would ask.
        let scratch = tmpdir("serve-caps-idx");
        crate::git_oracle::assert_git_accepts(
            &scratch,
            "dst.git",
            &out,
            crate::git_oracle::Strictness::Connected,
        );
    }

    /// **`put_refs_cas` applies every edit or none, and says which one lost.**
    ///
    /// Two properties in one test because they are one property: the batch that
    /// fails must leave the namespace exactly as it was, and the caller must be
    /// able to learn *which* ref and *what was there* **without reading the
    /// message**.
    #[test]
    fn an_atomic_batch_applies_every_edit_or_none_and_names_the_loser() {
        let dir = tmpdir("serve-cas");
        let store = GitStore::open(&dir, "rickard").unwrap();
        let (pack, oid) = one_blob_pack(b"cas fixture");
        store.put(&pack, &[]).unwrap();
        store.wait_indexed();
        store
            .put_refs(&[RefUpdate::set("refs/heads/taken", hex::encode(&oid))])
            .unwrap();

        // Edit 1 would succeed on its own; edit 2 must not. Neither may land.
        let edits = [
            RefCas {
                name: "refs/heads/fresh".into(),
                old: None,
                new: Some(&oid),
            },
            RefCas {
                name: "refs/heads/taken".into(),
                old: None, // "must not exist" — and it does
                new: Some(&oid),
            },
        ];
        let err = store.put_refs_cas(&edits).unwrap_err();

        let rejection = RefRejection::of(&err).expect(
            "a lost compare-and-swap must be TYPED — receive-pack cannot tell a lost race from a \
             broken disk by reading a message",
        );
        assert!(rejection.is_cas_failure());
        assert!(!rejection.is_lock_contention());
        assert_eq!(rejection.name(), "refs/heads/taken");
        match rejection {
            RefRejection::Cas {
                expected, actual, ..
            } => {
                assert_eq!(*expected, Observed::Nothing, "the caller claimed absent");
                assert_eq!(
                    *actual,
                    Observed::oid(&oid),
                    "the rejection must carry what was ACTUALLY there"
                );
            }
            other => panic!("wrong variant: {other:?}"),
        }

        let names: Vec<String> = store.refs().unwrap().into_iter().map(|r| r.name).collect();
        assert!(
            !names.iter().any(|n| n == "refs/heads/fresh"),
            "the first edit of a failed atomic batch must NOT have landed: {names:?}"
        );

        // And the same batch, with the expectation corrected, applies whole.
        let edits = [
            RefCas {
                name: "refs/heads/fresh".into(),
                old: None,
                new: Some(&oid),
            },
            RefCas {
                name: "refs/heads/taken".into(),
                old: Some(&oid),
                new: None, // delete
            },
        ];
        store.put_refs_cas(&edits).unwrap();
        let names: Vec<String> = store.refs().unwrap().into_iter().map(|r| r.name).collect();
        assert!(names.iter().any(|n| n == "refs/heads/fresh"));
        assert!(!names.iter().any(|n| n == "refs/heads/taken"));
    }

    /// **`S-023`: a create that must fail still fails when the new value equals
    /// the current one.**
    ///
    /// The defect this guards is a backend short-circuiting an edit whose new
    /// value already matches, never evaluating the caller's expectation and
    /// turning a `MustNotExist` into a **silent success**. It is not
    /// hypothetical — it was found in gix's file ref store, on the only backend
    /// that survives a restart — and it breaks *"exactly one creator wins"*,
    /// which is how receive-pack arbitrates two racing pushes.
    ///
    /// Seen RED by moving the expectation check after the write.
    #[test]
    fn s023_a_must_not_exist_create_fails_even_when_the_value_is_unchanged() {
        let dir = tmpdir("serve-s023");
        let store = GitStore::open(&dir, "rickard").unwrap();
        let (pack, oid) = one_blob_pack(b"s023");
        store.put(&pack, &[]).unwrap();
        store.wait_indexed();
        store
            .put_refs(&[RefUpdate::set("refs/heads/racy", hex::encode(&oid))])
            .unwrap();

        // Same value the ref already holds, with `old: None` — the exact shape a
        // short-circuiting backend turns into a no-op success.
        let edits = [RefCas {
            name: "refs/heads/racy".into(),
            old: None,
            new: Some(&oid),
        }];
        let err = store.put_refs_cas(&edits).unwrap_err();
        let rejection = RefRejection::of(&err).expect("must be a typed CAS rejection");
        assert!(
            rejection.is_cas_failure(),
            "a create-that-must-not-exist against an existing ref is a lost race, not a fault"
        );
    }

    /// An empty atomic batch is a no-op that succeeded.
    #[test]
    fn an_empty_atomic_batch_is_a_no_op_rather_than_an_error() {
        let dir = tmpdir("serve-cas-empty");
        let store = GitStore::open(&dir, "rickard").unwrap();
        assert_eq!(store.put_refs_cas(&[]).unwrap(), TxId::default());
    }

    /// The hash width the emitter trailers with is the store's, not a constant.
    #[test]
    fn the_emitted_trailer_is_the_stores_hash_width() {
        for hash in [GitHashKind::Sha1, GitHashKind::Sha256] {
            let dir = tmpdir(&format!("serve-trailer-{}", hash.oid_len()));
            let store = GitStore::open_with(&dir, "rickard", hash).unwrap();
            // No objects: the pack is header + trailer and nothing else, which
            // is exactly the shape that makes the widths comparable.
            let mut out = Vec::new();
            let stats = store.emit_oids(&[], &[], &Caps::modern(), &mut out).unwrap();
            assert_eq!(stats.objects, 0);
            assert_eq!(out.len(), 12 + hash.oid_len());
            assert_eq!(stats.bytes, out.len() as u64);
        }
    }
}