alkgit 0.1.0

//! The wave-1 in-memory backend doubles — alkgit's contract-reference
//! test convention (docs/plans/implementation.md §Wave 1, the
//! test-convention bullet): the alkblobs mem-engine analog. Every
//! protocol wave's gate runs its machines against these doubles
//! (`resolve`/`put_repo`/`update_repo`/`remove_repo` on
//! [`MemRegistry`], advertisement data and CAS transactions on
//! [`MemRefs`], the ack/done-round seam on [`MemPackGen`]), so the
//! convention is established here even though the full harness
//! compiles with wave 2's first machines.
//!
//! Each test binary's `mod common;` compiles the whole shared module,
//! and no single binary exercises every double — dead_code is allowed
//! module-wide so one machine's harness never fails the next's.

#![allow(dead_code)]
//!
//! **Location decision (recorded, the OQ-03 freeze-inventory discipline):**
//! the doubles live *only* here — shared modules under `tests/` so the
//! integration tests of the wave-2/3 machines reach them with `mod
//! common;` — and never in the public API: no re-export, no feature
//! gate, nothing doubles-related reachable from the crate root. The
//! doubles' shapes are test surface, test-versioned freely; if a
//! `src/`-unit test ever needs them, the path is a `#[cfg(test)]`-only
//! module or a `doc(hidden)` one — never a public export.
//!
//! **The row/stamp discipline** (adapted from the sibling crates —
//! the alkstore ADR-017 §4.2 version-stamping convention as alkblobs'
//! tests/conformance suite carries it):
//!
//! - Assertions are **rows**: named, independently-runnable properties,
//!   `snake_case`, the name stating the *property*, not the test or the
//!   caller.
//! - Every row is **version-stamped** with a doc-comment list item
//!   beginning **`Contract stamp:`** citing the spec § / ADR § that pins
//!   the behavior tested — never a bare wave or date. Several stamps per
//!   row are allowed; a later change that amends a row's property
//!   appends to its stamp list; stamps accumulate, the ADRs record the
//!   history.
//! - The stamp marker is greppable —
//!   `grep -rn "Contract stamp:" tests/` is the inventory.
//!
//! Refs are rows with stamps, here too: [`MemRefs`]' store is a row
//! table keyed on the resolved id (backend.md §Pinned signatures —
//! impls key storage on `repo_id`; the record carries no path, ADR-019
//! §8), and the traits' contract rows carry the stamps.
//!
//! **CAS semantics the refs double encodes** (ADR-013 §7, §10; ADR-018
//! §6; the substrate row CAS per ADR-019 §2): one transaction per
//! `apply_updates` call. A per-ref `expected` comparison never errors —
//! a stale expected is a denied [`RefOutcome`] (a business result of
//! receive-pack), the reason string being the server-chosen, verbatim
//! `ng` text ([`STALE_REASON`] here). `atomic = true` prepares the whole
//! transaction and commits it only if every update's CAS succeeds: on
//! any denial nothing is applied, the denied refs report their own
//! reasons, and every other ref of the transaction reports
//! [`ATOMIC_FAILURE_REASON`] (the observed upstream shape,
//! ADR-013 §10). Updates inside one transaction evaluate sequentially
//! against the evolving row state, so intra-transaction compositions
//! (a create then an update of the same ref) compose like a real
//! transaction.
//!
//! Unknown-repo ≡ unauthorized (ADR-008): [`MemRegistry`] returns
//! [`RegistryError::NotFound`] for an unresolved id and nothing else —
//! the call sites that consume the double collapse `NotFound` with
//! `Forbidden` to the generic wire denial; the double itself never
//! carries the authorization decision.
//!
//! **The object-store skeleton's Find-shaped front** ([`MemObjects`]):
//! `contains` (the existence half of the predicate `common_haves`
//! recognizes — ADR-014 §5's exists + is-commit; the is-commit half
//! lives in [`MemPackGen`]) and `find` (the object lookup generation uses) fill
//! only what the seam needs at this wave. The gix-trait binding over them
//! is [`MemFind`] — the `gix_object::Find`/`Exists` + `gix_pack::Find`
//! surface ADR-004's generation pipeline needs; [`MemPackGen`]::
//! `generate` composes the real pipeline over it (the wave-2 task),
//! and [`MemPackIngest`]::`prepare` composes the real ingestion double
//! behavior (the wave-3 task).
//!
//! **The ingestion double's composition** (recorded — the wave-3 task's
//! shape decision): not the POC-O `Bundle::write_to_directory`-over-a-
//! working-area route (that needs a filesystem staging dir + a gix-odb
//! front, and is the wave-8 substrate impl's shape with the pin bridge),
//! but the **streaming input iterators landing objects directly** — the
//! same internals `Bundle::write_to_directory` composes:
//! `data::input::BytesToEntriesIter` wrapped by
//! `data::input::LookupRefDeltaObjectsIter` with [`MemFind`] as the
//! thin-pack base lookup (ADR-013 §5's server-odb resolution), every
//! pack object expanded and staged as a loose-form row keyed by its
//! re-derived oid under the decompressed-size cap
//! (`Limits::max_object_size`, review 004 R-2 — the expansion reads
//! incrementally, no allocation sized by a pack-declared size).
//! Delta application is a local applier (gix-pack's
//! `data::delta::apply` is crate-private), the zero-object pack's
//! trailer — never reached by the streaming machinery's verification —
//! is verified by the double itself, and nothing partial ever lands:
//! the connectivity walk runs over the staged rows plus the
//! pre-existing namespace, and the landing happens only once the walk
//! closes. The mem store has no GC sweep, so no pins: on the `unpack`
//! ng leg nothing lands at all — this double's strictest honest form
//! of the working-area discard (ADR-020 §2).

use std::collections::BTreeMap;
use std::collections::HashSet;
use std::io;
use std::io::{Read, Write};
use std::sync::atomic::AtomicBool;
use std::sync::Arc;

use async_trait::async_trait;
use gix_hash::ObjectId;
use gix_pack::data::entry::Header as PackHeader;
use gix_pack::data::input::BytesToEntriesIter;
use gix_pack::data::input::Mode as InputMode;
use gix_pack::data::input::{Entry as PackEntry, EntryDataMode, LookupRefDeltaObjectsIter};
use gix_pack::data::output::entry::iter_from_counts;
use gix_pack::data::output::entry::iter_from_counts::{Mode, Options, Outcome};
use gix_pack::data::output::{bytes::FromEntriesIter, count};
use gix_parallel::Finalize;
use gix_traverse::commit::{Parents, Simple};
use parking_lot::Mutex;

use alkgit::{
    GitPackGen, GitPackIngest, GitRefs, GitRegistry, GitRegistryStore, Limits, PreparedPush,
    PushOptions, RefLine, RefOutcome, RefUpdate, RegistryError, RepoRecord, StorageError,
};

/// The server-chosen verbatim `ng` reason of a per-ref CAS denial
/// (ADR-013 §7–8: CAS-stale failures carry a server-chosen reason,
/// upstream's `stale info`-style internal text).
pub const STALE_REASON: &str = "stale info";

/// The server-chosen verbatim `ng` reason a ref that would otherwise
/// have succeeded reports when an atomic transaction is aborted by any
/// denial (ADR-013 §10 — every *other* ref reports `atomic push
/// failure`; the denied refs keep their own reasons).
pub const ATOMIC_FAILURE_REASON: &str = "atomic push failure";

/// The mem registry double — [`GitRegistry`]/[`GitRegistryStore`] over
/// an in-memory id → record map (`BTreeMap` for deterministic
/// iteration, the record's byte-determinism rule carried over from
/// ADR-015 §8's shape), under a short-held [`parking_lot::Mutex`]
/// (convention 3; no poisoning recovery posture exists to apply).
///
/// `resolve` of an unknown id is [`RegistryError::NotFound`] — the
/// unknown-repo ≡ unauthorized posture's double form (ADR-008); the
/// double returns it and nothing else. `put_repo` rejects an id that
/// is already taken ([`RegistryError::AlreadyExists`]) and an empty
/// repo id ([`RegistryError::Invalid`]) — the record's key discipline.
#[derive(Clone, Default)]
pub struct MemRegistry {
    rows: Arc<Mutex<BTreeMap<String, RepoRecord>>>,
}

#[async_trait]
impl GitRegistry for MemRegistry {
    async fn resolve(&self, repo_id: &str) -> Result<RepoRecord, RegistryError> {
        match self.rows.lock().get(repo_id) {
            Some(record) => Ok(record.clone()),
            None => Err(RegistryError::NotFound),
        }
    }
}

#[async_trait]
impl GitRegistryStore for MemRegistry {
    async fn put_repo(&self, record: &RepoRecord) -> Result<(), RegistryError> {
        if record.repo_id.is_empty() {
            return Err(RegistryError::Invalid);
        }
        let mut rows = self.rows.lock();
        if rows.contains_key(&record.repo_id) {
            return Err(RegistryError::AlreadyExists);
        }
        rows.insert(record.repo_id.clone(), record.clone());
        Ok(())
    }

    async fn update_repo(&self, record: &RepoRecord) -> Result<(), RegistryError> {
        let mut rows = self.rows.lock();
        match rows.get_mut(&record.repo_id) {
            Some(row) => *row = record.clone(),
            None => return Err(RegistryError::NotFound),
        }
        Ok(())
    }

    async fn remove_repo(&self, repo_id: &str) -> Result<(), RegistryError> {
        match self.rows.lock().remove(repo_id) {
            Some(_) => Ok(()),
            None => Err(RegistryError::NotFound),
        }
    }
}

/// The mem refs double — [`GitRefs`] over a row table keyed on the
/// resolved repo id: `repo_id → (ref name → oid)`, both `BTreeMap`s
/// (deterministic advertisement order), under a short-held
/// [`parking_lot::Mutex`]. The
/// [`list_refs`][GitRefs::list_refs] and CAS semantics are the module
/// doc's CAS section; see its stamps.
///
/// The HEAD symref rows ([`set_head_symref`][MemRefs::set_head_symref])
/// carry the unborn-HEAD fixture (review 001 N-5): a repo whose HEAD
/// points to an unborn branch lists a `HEAD` row with `oid: None` and
/// the symref target — the row the wire layer maps to the unborn
/// ls-refs line (ADRs 003/009; the machine shape lands with the wave-2
/// ls-refs work).
#[derive(Clone, Default)]
pub struct MemRefs {
    rows: Arc<Mutex<BTreeMap<String, BTreeMap<String, ObjectId>>>>,
    head_symrefs: Arc<Mutex<BTreeMap<String, Option<String>>>>,
}

impl MemRefs {
    /// Sets (or, with `None`, clears) the repo's unborn-HEAD symref row
    /// — the fixture the unborn ls-refs shape composes against (the
    /// N-5 rider's `RefLine { oid: None, symref_target: Some(..) }`).
    pub fn set_head_symref(&self, repo: &RepoRecord, target: Option<String>) {
        match target {
            Some(target) => {
                self.head_symrefs
                    .lock()
                    .insert(repo.repo_id.clone(), Some(target));
            }
            None => {
                self.head_symrefs.lock().remove(&repo.repo_id);
            }
        }
    }
}

#[async_trait]
impl GitRefs for MemRefs {
    async fn list_refs(&self, repo: &RepoRecord) -> Result<Vec<RefLine>, StorageError> {
        let head = self
            .head_symrefs
            .lock()
            .get(&repo.repo_id)
            .cloned()
            .flatten();
        let rows = self.rows.lock();
        let Some(tier) = rows.get(&repo.repo_id) else {
            return Ok(match head {
                Some(target) => vec![RefLine {
                    name: "HEAD".to_owned(),
                    oid: None,
                    peeled: None,
                    symref_target: Some(target),
                }],
                None => Vec::new(),
            });
        };
        let mut listed = Vec::with_capacity(tier.len() + usize::from(head.is_some()));
        if let Some(target) = head {
            listed.push(RefLine {
                name: "HEAD".to_owned(),
                oid: None,
                peeled: None,
                symref_target: Some(target),
            });
        }
        listed.extend(tier.iter().map(|(name, oid)| RefLine {
            name: name.clone(),
            oid: Some(*oid),
            peeled: None,
            symref_target: None,
        }));
        Ok(listed)
    }

    async fn apply_updates(
        &self,
        repo: &RepoRecord,
        updates: Vec<RefUpdate>,
        atomic: bool,
    ) -> Result<Vec<RefOutcome>, StorageError> {
        let mut rows = self.rows.lock();
        let tier = rows.entry(repo.repo_id.clone()).or_default();
        let mut working = tier.clone();

        let mut evaluated: Vec<(RefUpdate, Result<(), String>)> = Vec::with_capacity(updates.len());
        for update in &updates {
            let outcome = evaluate(&mut working, update, STALE_REASON);
            evaluated.push((update.clone(), outcome));
        }

        let any_denial = evaluated.iter().any(|(_, result)| result.is_err());
        if !(atomic && any_denial) {
            *tier = working;
        }

        let outcomes = evaluated
            .into_iter()
            .map(|(update, result)| match result {
                Ok(()) => {
                    if atomic && any_denial {
                        RefOutcome {
                            name: update.name,
                            result: Err(ATOMIC_FAILURE_REASON.to_string()),
                        }
                    } else {
                        RefOutcome {
                            name: update.name,
                            result: Ok(()),
                        }
                    }
                }
                Err(reason) => RefOutcome {
                    name: update.name,
                    result: Err(reason),
                },
            })
            .collect();
        Ok(outcomes)
    }
}

fn evaluate(
    working: &mut BTreeMap<String, ObjectId>,
    update: &RefUpdate,
    stale: &str,
) -> Result<(), String> {
    let current = working.get(&update.name).copied();
    let matches = match (update.expected, current) {
        (None, None) => true,
        (None, Some(_)) => false,
        (Some(expected), Some(current)) => expected == current,
        (Some(_), None) => false,
    };
    if !matches {
        return Err(stale.to_string());
    }
    match update.new {
        Some(new) => {
            working.insert(update.name.clone(), new);
        }
        None => {
            working.remove(&update.name);
        }
    }
    Ok(())
}

type ObjectRows = BTreeMap<(String, ObjectId), Vec<u8>>;

/// The object-store double skeleton — an in-memory `repo_id → oid →
/// object-bytes` map (`BTreeMap`, short-held [`parking_lot::Mutex`])
/// with the Find-shaped front waves 2–3 compose their machines
/// against: [`contains`][MemObjects::contains] (the existence predicate
/// [`common_haves`][alkgit::GitPackGen::common_haves] recognizes —
/// ADR-014 §5) and [`find`][MemObjects::find] (the object lookup
/// generation uses). `put` is test support: the doubles' rows seed the
/// map before composing against it. The `gix_object::Find` trait
/// binding and the pack-level machinery are waves 2–3; the shape is
/// fixed here so that composition is mechanical.
#[derive(Clone, Default)]
pub struct MemObjects {
    entries: Arc<Mutex<ObjectRows>>,
}

impl MemObjects {
    /// The existence half of the seam's recognized predicate (ADR-014
    /// §5's exists + is-commit — the is-commit half is
    /// [`MemPackGen`]'s `common_haves`, decoding the object type from
    /// the stored bytes).
    pub fn contains(&self, repo: &RepoRecord, oid: &ObjectId) -> bool {
        self.entries
            .lock()
            .contains_key(&(repo.repo_id.clone(), *oid))
    }

    /// The object lookup for generation: the stored bytes, or `None`
    /// when the repo's namespace lacks the oid.
    pub fn find(&self, repo: &RepoRecord, oid: &ObjectId) -> Option<Vec<u8>> {
        self.entries
            .lock()
            .get(&(repo.repo_id.clone(), *oid))
            .cloned()
    }

    /// Seeds one object into the repo's namespace (test support).
    pub fn put(&self, repo: &RepoRecord, oid: ObjectId, bytes: Vec<u8>) {
        self.entries
            .lock()
            .insert((repo.repo_id.clone(), oid), bytes);
    }
}

/// The pack-level Find front over one repo's [`MemObjects`] namespace —
/// the gix-trait binding ADR-004's generation pipeline consumes. The
/// test-double form of the substrate's Find front (ADR-019 §6, the
/// `gix_object::Find`/`Exists` + `gix_pack::Find` shapes constructed once
/// per session): the stored rows are loose-object encodings, decoded back
/// through [`gix_object::loose_header`] on every lookup.
///
/// `location_by_oid`/`pack_offsets_and_oid`/`entry_by_location` answer
/// with "not in any pack": the doubles carry no packed objects, so every
/// entry is generated as a whole deflated base (ADR-004; delta synthesis
/// is the recorded optimization backlog).
///
/// A lookup miss is recorded — the last missed oid — so a failed
/// walk/count stage maps its not-found error back to the
/// [`StorageError::ObjectMissing`] the seam aborts with (ADR-004), the
/// recorded id naming the missing object rather than the want.
#[derive(Clone)]
pub struct MemFind {
    objects: MemObjects,
    repo: RepoRecord,
    missed: Arc<Mutex<Option<ObjectId>>>,
}

impl MemFind {
    /// The Find front over one repo's object namespace, for one
    /// generation call.
    pub fn new(objects: &MemObjects, repo: &RepoRecord) -> Self {
        MemFind {
            objects: objects.clone(),
            repo: repo.clone(),
            missed: Arc::default(),
        }
    }

    /// The oid recorded by the latest lookup miss, if any.
    fn missed_object(&self) -> Option<ObjectId> {
        self.missed.lock().take()
    }
}

impl gix_object::Exists for MemFind {
    fn exists(&self, id: &gix_hash::oid) -> bool {
        <Self as gix_pack::Find>::contains(self, id)
    }
}

impl gix_object::Find for MemFind {
    fn try_find<'a>(
        &self,
        id: &gix_hash::oid,
        buffer: &'a mut Vec<u8>,
    ) -> gix_error::Result<Option<gix_object::Data<'a>>> {
        let find = <Self as gix_pack::Find>::try_find_cached(
            self,
            id,
            buffer,
            &mut gix_pack::cache::Never,
        )?;
        Ok(find.map(|(data, _location)| data))
    }
}

impl gix_pack::Find for MemFind {
    fn contains(&self, id: &gix_hash::oid) -> bool {
        self.objects
            .contains(&self.repo, &(id as &gix_hash::oid).to_owned())
    }

    fn try_find_cached<'a>(
        &self,
        id: &gix_hash::oid,
        buffer: &'a mut Vec<u8>,
        _pack_cache: &mut dyn gix_pack::cache::DecodeEntry,
    ) -> gix_error::Result<
        Option<(
            gix_object::Data<'a>,
            Option<gix_pack::data::entry::Location>,
        )>,
    > {
        let Some(bytes) = self
            .objects
            .find(&self.repo, &(id as &gix_hash::oid).to_owned())
        else {
            self.missed.lock().replace(id.to_owned());
            return Ok(None);
        };
        let (kind, _size, header_len) = gix_object::decode::loose_header(&bytes)?;
        buffer.clear();
        buffer.extend_from_slice(&bytes[header_len..]);
        Ok(Some((
            gix_object::Data {
                kind,
                object_hash: gix_hash::Kind::Sha1,
                data: buffer,
            },
            None,
        )))
    }

    fn location_by_oid(
        &self,
        _id: &gix_hash::oid,
        _buf: &mut Vec<u8>,
    ) -> gix_error::Result<Option<gix_pack::data::entry::Location>> {
        Ok(None)
    }

    fn pack_offsets_and_oid(
        &self,
        _pack_id: u32,
    ) -> gix_error::Result<Option<Vec<(gix_pack::data::Offset, ObjectId)>>> {
        Ok(None)
    }

    fn entry_by_location(
        &self,
        _location: &gix_pack::data::entry::Location,
    ) -> Option<gix_pack::find::Entry> {
        None
    }
}

fn object_missing(find: &MemFind, fallback: ObjectId) -> StorageError {
    StorageError::ObjectMissing {
        oid: find.missed_object().unwrap_or(fallback),
    }
}

/// Maps a pipeline error to the seam's error model: a not-found is the
/// missing-object abort (ADR-004), anything else is a backing-store
/// failure (ADR-018 §7). `fallback` names the want the stage was
/// working on, only reached when no concrete miss was recorded.
fn from_pipeline_error(
    find: &MemFind,
    error: gix_error::Error,
    fallback: ObjectId,
) -> StorageError {
    if error.is_not_found() {
        object_missing(find, fallback)
    } else {
        StorageError::Io(error.to_string())
    }
}

/// The peel result of one want or have: the tag objects walked through
/// (the chain's closure members) and the terminal object's id.
struct PeeledTip {
    chain: Vec<ObjectId>,
    terminal: ObjectId,
    terminal_is_commit: bool,
}

/// Peels `tip` through its tag chain (ADR-004: tips must be peeled before
/// the commit-ancestry walk), returning the chain and the terminal kind.
///
/// Errors on the first missing object of the chain with
/// [`StorageError::ObjectMissing`] (ADR-004) — a tip that cannot be
/// resolved is the abort, never a partial pack.
fn peel_tip(find: &MemFind, tip: ObjectId) -> Result<PeeledTip, StorageError> {
    let mut chain = Vec::new();
    let mut current = tip;
    loop {
        let mut buf = Vec::new();
        let data = match gix_object::FindExt::find(find, &current, &mut buf) {
            Ok(data) => data,
            Err(error) => return Err(from_pipeline_error(find, error, tip)),
        };
        let decoded = data.decode();
        let next = match decoded {
            Ok(gix_object::ObjectRef::Tag(tag)) => {
                chain.push(current);
                tag.target()
            }
            Ok(gix_object::ObjectRef::Commit(_)) => {
                return Ok(PeeledTip {
                    chain,
                    terminal: current,
                    terminal_is_commit: true,
                });
            }
            Ok(_) => {
                return Ok(PeeledTip {
                    chain,
                    terminal: current,
                    terminal_is_commit: false,
                });
            }
            Err(error) => return Err(StorageError::Io(error.to_string())),
        };
        current = next;
    }
}

/// The generation pipeline over the mem Find — ADR-004's composition
/// (poc2-findings' validated shape): peel wants to commit tips, the
/// boundary-bounded commit-ancestry walk ([`Simple`], `Parents::All`),
/// the two `count::objects_unthreaded` stages over the gix Find, and the
/// entry/byte stages streaming the V2 sha1 pack into the sink.
///
/// The two count stages compose ADR-014 §2's closure subtraction: the
/// boundary-committed ancestry is excluded by the walk's predicate, and
/// non-commit want targets / peeled tag objects are counted against the
/// boundary's own peeled object set (a tag want's terminal commit being
/// a boundary have ships the tag alone). Counted objects are then
/// existence-checked before the byte pipeline opens, so a missing
/// closure member aborts with nothing ever written to the sink; the
/// entry stage's own `missing_objects` statistics remain the abort
/// posture ADR-004 pins.
fn generate_pack(
    objects: &MemObjects,
    repo: &RepoRecord,
    wants: Vec<ObjectId>,
    boundary_haves: Vec<ObjectId>,
    sink: Box<dyn io::Write + Send>,
) -> Result<(), StorageError> {
    let find = MemFind::new(objects, repo);
    let no_want = ObjectId::null(gix_hash::Kind::Sha1);
    let first_want = wants.first().copied().unwrap_or(no_want);

    let mut boundary_commits = HashSet::new();
    let mut boundary_objects = HashSet::new();
    for have in boundary_haves {
        let peeled = peel_tip(&find, have)?;
        boundary_objects.extend(peeled.chain.iter().copied());
        boundary_objects.insert(peeled.terminal);
        if peeled.terminal_is_commit {
            boundary_commits.insert(peeled.terminal);
        }
    }

    let mut commit_tips = Vec::new();
    let mut non_commit_tips = Vec::new();
    let mut tag_chains = Vec::new();
    for want in wants {
        let peeled = peel_tip(&find, want)?;
        for chain_oid in peeled.chain {
            if !boundary_objects.contains(&chain_oid) {
                tag_chains.push(chain_oid);
            }
        }
        if peeled.terminal_is_commit {
            commit_tips.push(peeled.terminal);
        } else if !boundary_objects.contains(&peeled.terminal) {
            non_commit_tips.push(peeled.terminal);
        }
    }

    let mut walk = Simple::filtered(commit_tips, find.clone(), |tip: &gix_hash::oid| {
        !boundary_commits.contains(tip)
    })
    .parents(Parents::All);
    let mut survivors = Vec::new();
    for info in walk.by_ref() {
        let info = info.map_err(|error| from_pipeline_error(&find, error, first_want))?;
        survivors.push(info.id);
    }

    let counter = gix_utils::progress::Discard;
    let interrupt = AtomicBool::new(false);
    let mut closure_inputs = survivors;
    closure_inputs.extend(non_commit_tips);
    let (mut counts, _count_outcome) = count::objects_unthreaded(
        &find,
        &mut closure_inputs
            .into_iter()
            .map(Ok::<ObjectId, gix_error::Error>),
        &counter,
        &interrupt,
        count::objects::ObjectExpansion::TreeContents,
    )
    .map_err(|error| from_pipeline_error(&find, error, first_want))?;
    let (tag_counts, _tag_outcome) = count::objects_unthreaded(
        &find,
        &mut tag_chains.into_iter().map(Ok::<ObjectId, gix_error::Error>),
        &counter,
        &interrupt,
        count::objects::ObjectExpansion::AsIs,
    )
    .map_err(|error| from_pipeline_error(&find, error, first_want))?;
    let mut counted: HashSet<ObjectId> = counts.iter().map(|count| count.id).collect();
    for tag_count in tag_counts {
        if counted.contains(&tag_count.id) {
            continue;
        }
        counted.insert(tag_count.id);
        counts.push(tag_count);
    }

    for count in &counts {
        if !gix_pack::Find::contains(&find, &count.id) {
            return Err(StorageError::ObjectMissing { oid: count.id });
        }
    }

    let counts_len = counts.len();
    let mut entries = gix_parallel::InOrderIter::from(
        iter_from_counts(
            counts,
            find.clone(),
            Box::new(gix_utils::progress::Discard),
            Options {
                thread_limit: Some(1),
                mode: Mode::PackCopyAndBaseObjects,
                allow_thin_pack: false,
                chunk_size: 10,
                version: gix_pack::data::Version::V2,
                compression: gix_zlib::Compression::DEFAULT,
            },
        )
        .map_err(|error| from_pipeline_error(&find, error, first_want))?,
    );
    let mut from_entries = FromEntriesIter::new(
        entries.by_ref(),
        sink,
        counts_len as u32,
        gix_pack::data::Version::V2,
        gix_hash::Kind::Sha1,
    );
    for step in from_entries.by_ref() {
        step.map_err(|error| from_pipeline_error(&find, error, first_want))?;
    }
    if from_entries.digest().is_none() {
        return Err(StorageError::Io(
            "the pack trailer was never written".to_string(),
        ));
    }
    let entry_statistics: Outcome = entries
        .inner
        .finalize()
        .map_err(|error| from_pipeline_error(&find, error, first_want))?;
    if entry_statistics.missing_objects > 0 {
        let missing = find
            .missed_object()
            .unwrap_or_else(|| ObjectId::null(gix_hash::Kind::Sha1));
        return Err(StorageError::ObjectMissing { oid: missing });
    }
    Ok(())
}

/// The pack-generation double — [`GitPackGen`] sharing the
/// [`MemObjects`] namespace: `common_haves` is wired (the seam's
/// recognized predicate — exists + is-commit, per ADR-014 §5; the
/// stored bytes' object type is decodable through
/// [`gix_object::decode::loose_header`], preserving the caller's order
/// verbatim) and `generate` composes ADR-004's real pipeline over
/// [`MemFind`] (sync test code — the blocking-thread posture is the
/// machine's, ADR-009).
#[derive(Clone, Default)]
pub struct MemPackGen {
    pub objects: MemObjects,
}

impl MemPackGen {
    /// The seam's recognized predicate (ADR-014 §5): exists in the
    /// repo's namespace **and** decodes as a commit. Unrecognizable
    /// stored bytes fail closed (never recognized).
    fn recognized(&self, repo: &RepoRecord, oid: &ObjectId) -> bool {
        let Some(bytes) = self.objects.find(repo, oid) else {
            return false;
        };
        matches!(
            gix_object::decode::loose_header(&bytes),
            Ok((gix_object::Kind::Commit, _, _))
        )
    }
}

#[async_trait]
impl GitPackGen for MemPackGen {
    async fn generate(
        &self,
        repo: &RepoRecord,
        wants: Vec<ObjectId>,
        boundary_haves: Vec<ObjectId>,
        _limits: Limits,
        sink: Box<dyn io::Write + Send>,
    ) -> Result<(), StorageError> {
        generate_pack(&self.objects, repo, wants, boundary_haves, sink)
    }

    async fn common_haves(
        &self,
        repo: &RepoRecord,
        haves: Vec<ObjectId>,
    ) -> Result<Vec<ObjectId>, StorageError> {
        Ok(haves
            .into_iter()
            .filter(|oid| self.recognized(repo, oid))
            .collect())
    }
}

/// One object expanded out of an ingested pack, staged (not yet
/// landed): the oid re-derived from the expanded body, the object kind
/// (a delta shares its base's kind), and the raw body.
#[derive(Clone)]
struct Staged {
    oid: ObjectId,
    kind: gix_object::Kind,
    body: Vec<u8>,
}

/// The impl-side pack-size budget (ADR-009's `max_pack_size` — the
/// wire-side relay count is the primary enforcement, the seam's
/// `limits` parameter is the impl's own defense): a counting reader
/// that errors once the stream passes the budget, failing the
/// ingestion into the `unpack` ng leg, never `Err`. EOF before the
/// budget is the ordinary stream end — under-length input is the
/// indexer's own parse error.
struct BudgetedReader<R> {
    inner: R,
    bytes_seen: u64,
    limit: u64,
}

impl<R> BudgetedReader<R> {
    fn new(inner: R, limit: u64) -> Self {
        BudgetedReader {
            inner,
            bytes_seen: 0,
            limit,
        }
    }
}

impl<R: Read> Read for BudgetedReader<R> {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        if self.bytes_seen >= self.limit {
            return Err(io::Error::other(format!(
                "pack stream exceeded the maximum pack size budget of {} bytes",
                self.limit
            )));
        }
        let read = self.inner.read(buf)?;
        self.bytes_seen += u64::try_from(read).unwrap_or(u64::MAX);
        Ok(read)
    }
}

/// One size field of a delta stream: the 7-bit little-endian groups
/// with the continuation bit, starting at `offset` — the consumed byte
/// count is returned alongside.
fn delta_size_at(delta: &[u8], offset: usize) -> Result<(u64, usize), String> {
    let mut size = 0u64;
    let mut shift = 0u32;
    let mut cursor = offset;
    loop {
        let Some(&byte) = delta.get(cursor) else {
            return Err("the delta header is truncated".to_string());
        };
        cursor += 1;
        size |= u64::from(byte & 0x7f) << shift;
        shift += 7;
        if byte & 0x80 == 0 {
            return Ok((size, cursor - offset));
        }
        if shift >= 64 {
            return Err("the delta header size overflow".to_string());
        }
    }
}

/// Applies a git delta stream to its base: the declared base size
/// must match, the declared result stays under `cap` (the ingestion
/// decompressed-size cap — the delta's materialized result is one of
/// the objects the pack produces),
/// and the instructions (copy from base, insert literal)
/// must produce exactly the declared result size. The result
/// accumulates incrementally — the declared size is checked against,
/// never pre-allocated (the size-bomb class): the loop itself aborts
/// once the produced bytes pass the declaration, so no delta stream
/// produces bytes beyond the cap unannounced.
fn apply_delta(base: &[u8], delta: &[u8], cap: u64) -> Result<Vec<u8>, String> {
    let (base_size, base_header_len) = delta_size_at(delta, 0)?;
    let (result_size, result_header_len) = delta_size_at(delta, base_header_len)?;
    if base_size != u64::try_from(base.len()).unwrap_or(u64::MAX) {
        return Err(format!(
            "the delta declares a base size of {base_size} but the base object is {} bytes",
            base.len()
        ));
    }
    if result_size > u64::from(u32::MAX) {
        return Err("the delta declares an out-of-bounds result size".to_string());
    }
    if result_size > cap {
        return Err(format!(
            "the delta declares a materialized result of {result_size} bytes beyond \
             the ingestion cap of {cap} bytes"
        ));
    }
    let mut out = Vec::new();
    let mut cursor = base_header_len + result_header_len;
    while let Some(&cmd) = delta.get(cursor) {
        cursor += 1;
        if cmd & 0x80 != 0 {
            let mut ofs = 0u64;
            let mut size = 0u64;
            for (bit, shift) in [(0x01, 0), (0x02, 8), (0x04, 16), (0x08, 24)] {
                if cmd & bit != 0 {
                    let Some(&byte) = delta.get(cursor) else {
                        return Err("the delta copy instruction is truncated".to_string());
                    };
                    cursor += 1;
                    ofs |= u64::from(byte) << shift;
                }
            }
            for (bit, shift) in [(0x10, 0), (0x20, 8), (0x40, 16)] {
                if cmd & bit != 0 {
                    let Some(&byte) = delta.get(cursor) else {
                        return Err("the delta copy instruction is truncated".to_string());
                    };
                    cursor += 1;
                    size |= u64::from(byte) << shift;
                }
            }
            if size == 0 {
                size = 0x10000;
            }
            let (ofs, size) = (
                usize::try_from(ofs).map_err(|_| "the delta copy offset overflows".to_string())?,
                usize::try_from(size).map_err(|_| "the delta copy size overflows".to_string())?,
            );
            let Some(end) = ofs.checked_add(size) else {
                return Err("the delta copy range overflows".to_string());
            };
            let Some(bytes) = base.get(ofs..end) else {
                return Err("the delta copy range exceeds the base object size".to_string());
            };
            out.extend_from_slice(bytes);
        } else if cmd == 0 {
            return Err("the delta command 0 is reserved and invalid".to_string());
        } else {
            let end = cursor + usize::from(cmd);
            let Some(bytes) = delta.get(cursor..end) else {
                return Err("the delta insert data is truncated".to_string());
            };
            cursor = end;
            out.extend_from_slice(bytes);
        }
        if u64::try_from(out.len()).unwrap_or(u64::MAX) > result_size {
            return Err(format!(
                "the delta instructions produce more than the declared {result_size} bytes"
            ));
        }
    }
    if out.len() as u64 != result_size {
        return Err(format!(
            "the delta instructions produce {} bytes but the delta declares {result_size}",
            out.len()
        ));
    }
    Ok(out)
}

/// Inflates one kept pack entry back to its bytes (the delta payload
/// for delta entries, the object body for base entries).
///
/// The decompressed-size cap (`cap`, the ingestion budget's
/// `max_object_size`) applies twice: a declaration beyond the cap is
/// rejected before any reading, and the inflation itself is
/// incremental — the produced bytes accumulate in bounded chunks and
/// the read aborts at the cap, so no allocation is ever sized by the
/// pack-declared size (the size-bomb class, review 004 R-2: a hostile
/// pack declares an entry and decompresses to gigabytes; the old
/// allocate-then-fill shape trusted the declaration).
fn inflate_entry(entry: &PackEntry, cap: u64) -> Result<Vec<u8>, String> {
    let compressed = entry
        .compressed
        .as_ref()
        .ok_or_else(|| "the pack entry lost its compressed bytes".to_string())?;
    if entry.decompressed_size > cap {
        return Err(format!(
            "the pack entry declares {} decompressed bytes beyond the ingestion cap of {cap} bytes",
            entry.decompressed_size
        ));
    }
    let mut out: Vec<u8> = Vec::new();
    let mut decompressor = gix_zlib::Decompress::new();
    let mut reader = io::BufReader::new(io::Cursor::new(compressed.as_slice()));
    let mut chunk = vec![0u8; INFLATE_CHUNK_BYTES];
    loop {
        let written = gix_zlib::stream::inflate::read(&mut reader, &mut decompressor, &mut chunk)
            .map_err(|error| format!("the pack entry does not inflate: {error}"))?;
        out.extend_from_slice(&chunk[..written]);
        if u64::try_from(out.len()).unwrap_or(u64::MAX) > cap {
            return Err(format!(
                "the pack entry inflates past the ingestion cap of {cap} bytes"
            ));
        }
        if written < chunk.len() {
            break;
        }
    }
    if out.len() as u64 != entry.decompressed_size {
        return Err(format!(
            "the pack entry inflates to {} bytes but {} were declared",
            out.len(),
            entry.decompressed_size
        ));
    }
    Ok(out)
}

/// The object kind of a base-entry header.
fn base_kind(header: &PackHeader) -> Option<gix_object::Kind> {
    match header {
        PackHeader::Blob => Some(gix_object::Kind::Blob),
        PackHeader::Tree => Some(gix_object::Kind::Tree),
        PackHeader::Commit => Some(gix_object::Kind::Commit),
        PackHeader::Tag => Some(gix_object::Kind::Tag),
        PackHeader::OfsDelta { .. } | PackHeader::RefDelta { .. } => None,
    }
}

/// The inflate chunk the expansion cap reads in (bounded output
/// slices — the incremental reading itself never buffers beyond the
/// chunk plus the produced bytes).
const INFLATE_CHUNK_BYTES: usize = 65536;

/// Expands one delta entry against the staged in-pack objects: the
/// base is resolved by pack offset (ofs delta) or oid (ref delta) —
/// `None` means the base is not staged (yet), leaving the entry
/// pending for the fixpoint pass.
fn expand_delta(
    staged_by_offset: &BTreeMap<u64, Staged>,
    staged_by_oid: &BTreeMap<ObjectId, Staged>,
    entry: &PackEntry,
    cap: u64,
) -> Result<Option<Staged>, String> {
    let base = match &entry.header {
        PackHeader::OfsDelta { base_distance } => {
            let Some(base_offset) = entry.pack_offset.checked_sub(*base_distance) else {
                return Err(format!(
                    "the ofs delta at pack offset {} names an impossible base distance {base_distance}",
                    entry.pack_offset
                ));
            };
            match staged_by_offset.get(&base_offset) {
                Some(base) => base,
                None => return Ok(None),
            }
        }
        PackHeader::RefDelta { base_id } => match staged_by_oid.get(base_id) {
            Some(base) => base,
            None => return Ok(None),
        },
        PackHeader::Blob | PackHeader::Tree | PackHeader::Commit | PackHeader::Tag => {
            return Err("a base-entry header cannot expand as a delta".to_string());
        }
    };
    let payload = inflate_entry(entry, cap)?;
    let body = apply_delta(&base.body, &payload, cap)?;
    let oid = gix_object::compute_hash(gix_hash::Kind::Sha1, base.kind, &body)
        .map_err(|error| format!("the expanded delta object does not hash: {error}"))?;
    Ok(Some(Staged {
        oid,
        kind: base.kind,
        body,
    }))
}

/// Stages one pack entry: base entries inflate and re-hash, delta
/// entries expand against the staged bases; an unresolvable delta is
/// parked for the fixpoint pass (in-pack forward references are legal).
fn stage_entry(
    staged_by_offset: &mut BTreeMap<u64, Staged>,
    staged_by_oid: &mut BTreeMap<ObjectId, Staged>,
    pending: &mut Vec<PackEntry>,
    entry: PackEntry,
    cap: u64,
) -> Result<(), String> {
    let staged = match &entry.header {
        PackHeader::Blob | PackHeader::Tree | PackHeader::Commit | PackHeader::Tag => {
            let kind = base_kind(&entry.header)
                .ok_or_else(|| "a base-entry header carries no object kind".to_string())?;
            let body = inflate_entry(&entry, cap)?;
            let oid = gix_object::compute_hash(gix_hash::Kind::Sha1, kind, &body)
                .map_err(|error| format!("the pack entry object does not hash: {error}"))?;
            Some(Staged { oid, kind, body })
        }
        _ => expand_delta(staged_by_offset, staged_by_oid, &entry, cap)?,
    };
    match staged {
        Some(staged) => {
            staged_by_offset.insert(entry.pack_offset, staged.clone());
            staged_by_oid.insert(staged.oid, staged);
        }
        None => pending.push(entry),
    }
    Ok(())
}

/// The server-chosen reason for a delta whose base resolved nowhere:
/// a ref delta may name the missing base oid (the oid came from the
/// client's own pack, no oracle risk — ADR-018 §7's `ObjectMissing`
/// note applies at the reason's discretion).
fn unresolved_delta_reason(entry: &PackEntry) -> String {
    match &entry.header {
        PackHeader::RefDelta { base_id } => {
            format!("the delta base {base_id} is neither in the pack nor in the repository")
        }
        PackHeader::OfsDelta { base_distance } => format!(
            "the ofs delta's base at distance {base_distance} does not resolve within the pack"
        ),
        _ => "a base entry cannot remain unresolved".to_string(),
    }
}

/// Lands the staged objects into the repo's namespace as loose-form
/// rows keyed by their oids — the verification posture the substrate
/// gets from the store's verification-on-put is test-side re-derivation
/// (the oids were already re-derived at expansion).
fn land_staged(objects: &MemObjects, repo: &RepoRecord, staged: &BTreeMap<ObjectId, Staged>) {
    for (oid, object) in staged {
        let header = gix_object::encode::loose_header(object.kind, object.body.len() as u64);
        let mut loose = Vec::new();
        loose.extend_from_slice(&header);
        loose.extend_from_slice(&object.body);
        objects.put(repo, *oid, loose);
    }
}

/// The full-graph connectivity walk from the pushed tips (ADR-020 §2,
/// parents included — `gix-fsck`'s `Connectivity::check_commit` does
/// not walk parents; the owning implementation walks): commits walk
/// parents and trees, trees walk entries, tags walk targets — over the
/// combined graph of the staged rows and the pre-existing namespace.
fn check_connectivity(
    objects: &MemObjects,
    repo: &RepoRecord,
    staged: &BTreeMap<ObjectId, Staged>,
    tips: &[ObjectId],
) -> Result<(), String> {
    let mut seen: HashSet<ObjectId> = HashSet::new();
    let mut stack: Vec<ObjectId> = tips.to_vec();
    while let Some(oid) = stack.pop() {
        if !seen.insert(oid) {
            continue;
        }
        let (kind, body) = if let Some(object) = staged.get(&oid) {
            (object.kind, object.body.clone())
        } else {
            let Some(bytes) = objects.find(repo, &oid) else {
                return Err(format!("missing object {oid}"));
            };
            let (kind, _size, header_len) = gix_object::decode::loose_header(&bytes)
                .map_err(|error| format!("corrupt object {oid}: {error}"))?;
            (kind, bytes[header_len..].to_vec())
        };
        match kind {
            gix_object::Kind::Commit => {
                let commit = gix_object::CommitRef::from_bytes(&body, gix_hash::Kind::Sha1)
                    .map_err(|error| format!("corrupt commit {oid}: {error}"))?;
                stack.push(commit.tree());
                for parent in commit.parents() {
                    stack.push(parent);
                }
            }
            gix_object::Kind::Tree => {
                let tree = gix_object::TreeRef::from_bytes(&body, gix_hash::Kind::Sha1)
                    .map_err(|error| format!("corrupt tree {oid}: {error}"))?;
                for entry in &tree.entries {
                    stack.push(entry.oid.into());
                }
            }
            gix_object::Kind::Tag => {
                let tag = gix_object::TagRef::from_bytes(&body, gix_hash::Kind::Sha1)
                    .map_err(|error| format!("corrupt tag {oid}: {error}"))?;
                stack.push(tag.target());
            }
            gix_object::Kind::Blob => {}
        }
    }
    Ok(())
}

/// The ingestion over one boxed pack stream (the ADR-020 §2
/// composition, the streaming-input form): the budgeted stream fails
/// into the ng leg past the size budget; the pack header, entries, and
/// trailer verify through the machinery (Mode::Verify) — except the
/// zero-object pack, whose trailer the machinery never reaches, so the
/// double verifies it itself; thin bases resolve from the repo's
/// namespace through [MemFind]; in-pack forward ref-delta references
/// resolve in a bounded fixpoint pass; and every expansion is bounded
/// by `entry_cap` — the decompressed-size cap (`Limits::max_object_size`)
/// the staged entries inflate and materialize under, the declared
/// size never pre-allocated (review 004 R-2's guard). Returns the
/// staged objects — the caller lands them only after the connectivity
/// walk closes. Every failure is pack-level validation and reports the
/// server- chosen `unpack ng` reason; `Err` (the `StorageError`) stays
/// for infrastructure failure only, which the mem double has none of.
fn ingest_pack(
    objects: &MemObjects,
    repo: &RepoRecord,
    pack: Box<dyn Read + Send>,
    budget: u64,
    entry_cap: u64,
) -> Result<BTreeMap<ObjectId, Staged>, String> {
    let mut budgeted = BudgetedReader::new(pack, budget);
    let mut header = [0u8; gix_pack::data::header::SIZE];
    budgeted
        .read_exact(&mut header)
        .map_err(|_| "the pack stream ended in the header".to_string())?;
    let (_version, object_count) = gix_pack::data::header::decode(&header)
        .map_err(|error| format!("the pack header does not decode: {error}"))?;

    let mut staged_by_offset: BTreeMap<u64, Staged> = BTreeMap::new();
    let mut staged_by_oid: BTreeMap<ObjectId, Staged> = BTreeMap::new();

    if object_count == 0 {
        // The header bytes were already consumed from `budgeted`, so the
        // trailer — which the streaming machinery never reaches for a
        // zero-object pack — is read from the stream directly.
        let mut trailer_reader = io::BufReader::new(budgeted);
        let mut trailer = [0u8; gix_hash::Kind::Sha1.len_in_bytes()];
        trailer_reader
            .read_exact(&mut trailer)
            .map_err(|_| "the zero-object pack stream ends before the trailer".to_string())?;
        let mut hasher = gix_hash::hasher(gix_hash::Kind::Sha1);
        hasher.update(&header);
        let digest = hasher
            .try_finalize()
            .map_err(|error| format!("the pack trailer does not hash: {error}"))?;
        if digest != ObjectId::from(trailer) {
            return Err(
                "the zero-object pack's trailer does not verify against the pack bytes".to_string(),
            );
        }
        return Ok(staged_by_oid);
    }

    let reader = io::BufReader::new(io::Cursor::new(header).chain(budgeted));
    let entries = BytesToEntriesIter::new_from_header(
        reader,
        InputMode::Verify,
        EntryDataMode::Keep,
        gix_hash::Kind::Sha1,
    )
    .map_err(|error| format!("the pack header does not decode: {error}"))?;
    let lookup = MemFind::new(objects, repo);
    let mut thin = LookupRefDeltaObjectsIter::new(entries, lookup, gix_zlib::Compression::DEFAULT);
    let mut pending: Vec<PackEntry> = Vec::new();
    for step in thin.by_ref() {
        let entry = step.map_err(|error| format!("the pack stream fails verification: {error}"))?;
        stage_entry(
            &mut staged_by_offset,
            &mut staged_by_oid,
            &mut pending,
            entry,
            entry_cap,
        )?;
    }
    while !pending.is_empty() {
        let drained = std::mem::take(&mut pending);
        let count = drained.len();
        for entry in drained {
            stage_entry(
                &mut staged_by_offset,
                &mut staged_by_oid,
                &mut pending,
                entry,
                entry_cap,
            )?;
        }
        if pending.len() == count {
            return Err(unresolved_delta_reason(&pending[0]));
        }
    }
    Ok(staged_by_oid)
}

/// The pack-ingestion double — [`GitPackIngest`] sharing the
/// [`MemObjects`] namespace: `prepare` composes the ADR-020 §2
/// ingestion over the streaming input iterators (the module doc's
/// recorded composition) — the budgeted stream, thin bases from
/// [`MemFind`] (ADR-013 §5), every pack object expanded, and the
/// full-graph connectivity walk from the pushed tips with parents
/// included. Sync test code — the blocking-thread posture is the
/// machine's (ADR-009, same as [`MemPackGen::generate`]). Pack-level
/// validation failures (parse, index, thin-base miss, connectivity,
/// budget, the decompressed-size cap —
/// `Limits::max_object_size`'s incremental inflate, review 004 R-2)
/// land in [`PreparedPush::unpack`]'s ng leg, never `Err`; the
/// session continues to the report per ADR-013 §8. `updates` returns
/// the input list verbatim on every path, including the empty pack
/// (zero objects is valid and reported as such) and every ng leg — the
/// wire layer needs the ref names for the all-refs unpacker-error
/// report (the amended seam's round-trip rule, ADR-018 §5).
#[derive(Clone, Default)]
pub struct MemPackIngest {
    pub objects: MemObjects,
}

impl MemPackIngest {
    /// The double's object namespace, for the ingested objects' loose
    /// rows to be inspected against.
    pub fn objects(&self) -> &MemObjects {
        &self.objects
    }
}

#[async_trait]
impl GitPackIngest for MemPackIngest {
    async fn prepare(
        &self,
        repo: &RepoRecord,
        updates: Vec<RefUpdate>,
        pack: Box<dyn Read + Send>,
        _push_options: Option<PushOptions>,
        limits: Limits,
    ) -> Result<PreparedPush, StorageError> {
        let unpack = match ingest_pack(
            &self.objects,
            repo,
            pack,
            limits.max_pack_size,
            limits.max_object_size,
        ) {
            Ok(staged) => {
                let mut tips: Vec<ObjectId> = Vec::new();
                for update in &updates {
                    if let Some(new) = update.new {
                        tips.push(new);
                    }
                }
                match check_connectivity(&self.objects, repo, &staged, &tips) {
                    Ok(()) => {
                        land_staged(&self.objects, repo, &staged);
                        Ok(())
                    }
                    Err(reason) => Err(reason),
                }
            }
            Err(reason) => Err(reason),
        };
        Ok(PreparedPush { unpack, updates })
    }
}

/// A sink whose written bytes the rows inspect after the fact — a
/// shared buffer wrapped in the [`io::Write`] the seam's boxed sink
/// carries (`Send`), so an aborted generation's emptiness is provable.
#[derive(Clone, Default)]
pub struct SharedSink {
    bytes: Arc<Mutex<Vec<u8>>>,
}

impl SharedSink {
    /// The bytes written so far, in write order.
    pub fn snapshot(&self) -> Vec<u8> {
        self.bytes.lock().clone()
    }
}

impl io::Write for SharedSink {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        self.bytes.lock().extend_from_slice(buf);
        Ok(buf.len())
    }

    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

fn fixture_signature() -> gix_actor::Signature {
    gix_actor::Signature {
        name: b"alkgit doubles".into(),
        email: b"doubles@alk.dev".into(),
        time: gix_object::date::Time {
            seconds: 1_700_000_000,
            offset: 0,
        },
    }
}

fn seeded_loose(kind: gix_object::Kind, body: &[u8]) -> (ObjectId, Vec<u8>) {
    let id = gix_object::compute_hash(gix_hash::Kind::Sha1, kind, body)
        .expect("fixture hashes its own object");
    let mut bytes = Vec::new();
    bytes.extend_from_slice(&gix_object::encode::loose_header(kind, body.len() as u64));
    bytes.extend_from_slice(body);
    (id, bytes)
}

/// Builds a blob fixture and returns `(oid, loose bytes)`.
pub fn blob_fixture(content: &[u8]) -> (ObjectId, Vec<u8>) {
    seeded_loose(gix_object::Kind::Blob, content)
}

/// Decodes one loose object's bytes back to `(kind, body)` — the
/// pack-entry fixture source (the fixture objects' loose encoding is
/// the seeded row; the entries encode from the body).
pub fn loose_body(bytes: &[u8]) -> (gix_object::Kind, Vec<u8>) {
    let (kind, _size, header_len) =
        gix_object::decode::loose_header(bytes).expect("fixture objects decode");
    (kind, bytes[header_len..].to_vec())
}

/// Builds a tree fixture from `(kind, filename, oid)` entries; the
/// entries are serialized in git's own entry order.
pub fn tree_fixture(
    entries: &[(gix_object::tree::EntryKind, &str, ObjectId)],
) -> (ObjectId, Vec<u8>) {
    let mut entries: Vec<gix_object::tree::Entry> = entries
        .iter()
        .map(|(kind, filename, oid)| gix_object::tree::Entry {
            mode: (*kind).into(),
            filename: (*filename).into(),
            oid: *oid,
        })
        .collect();
    entries.sort();
    let tree = gix_object::Tree { entries };
    let mut body = Vec::new();
    gix_object::WriteTo::write_to(&tree, &mut body).expect("fixture trees serialize");
    seeded_loose(gix_object::Kind::Tree, &body)
}

/// Builds a commit fixture over `tree` with `parents`; returns
/// `(oid, loose bytes)`.
pub fn commit_fixture(tree: ObjectId, parents: &[ObjectId], message: &str) -> (ObjectId, Vec<u8>) {
    let commit = gix_object::Commit {
        tree,
        parents: parents.iter().copied().collect(),
        author: fixture_signature(),
        committer: fixture_signature(),
        encoding: None,
        message: message.into(),
        extra_headers: Vec::new(),
    };
    let mut body = Vec::new();
    gix_object::WriteTo::write_to(&commit, &mut body).expect("fixture commits serialize");
    seeded_loose(gix_object::Kind::Commit, &body)
}

/// Builds an annotated tag fixture over `target`; returns
/// `(oid, loose bytes)`.
pub fn tag_fixture(
    target: ObjectId,
    target_kind: gix_object::Kind,
    name: &str,
) -> (ObjectId, Vec<u8>) {
    let tag = gix_object::Tag {
        target,
        target_kind,
        name: name.into(),
        tagger: Some(fixture_signature()),
        message: b"a fixture tag\n".into(),
        signature: None,
    };
    let mut body = Vec::new();
    gix_object::WriteTo::write_to(&tag, &mut body).expect("fixture tags serialize");
    seeded_loose(gix_object::Kind::Tag, &body)
}

/// A hostile delta stream: a valid base-size declaration, a declared
/// materialized result, and no instructions — the applier's
/// decompressed-size-cap check (and, past it, the exact-produced-size
/// check) fires before anything accumulates.
pub fn hostile_delta_stream(base_size: u64, declared_result: u64) -> Vec<u8> {
    let mut out = delta_size_varint(base_size);
    out.extend(delta_size_varint(declared_result));
    out
}

/// Deflates bytes as one zlib stream (the pack entries' compression).
fn deflate_bytes(bytes: &[u8]) -> Vec<u8> {
    let mut deflate =
        gix_zlib::stream::deflate::Write::new(Vec::new(), gix_zlib::Compression::DEFAULT);
    deflate.write_all(bytes).expect("fixture bytes deflate");
    deflate
        .flush()
        .expect("the fixture deflate stream finishes");
    deflate.into_inner()
}

/// One whole-base pack entry (the canonical encoding of `Commit`,
/// `Tree`, `Blob`, `Tag` entries) at `pack_offset`.
pub fn pack_entry_base(kind: gix_object::Kind, body: &[u8], pack_offset: u64) -> PackEntry {
    let header = match kind {
        gix_object::Kind::Blob => PackHeader::Blob,
        gix_object::Kind::Tree => PackHeader::Tree,
        gix_object::Kind::Commit => PackHeader::Commit,
        gix_object::Kind::Tag => PackHeader::Tag,
    };
    let compressed = deflate_bytes(body);
    PackEntry {
        header,
        header_size: header.size(body.len() as u64) as u16,
        pack_offset,
        compressed: Some(compressed.clone()),
        compressed_size: compressed.len() as u64,
        crc32: None,
        decompressed_size: body.len() as u64,
        trailer: None,
    }
}

/// One delta pack entry at `pack_offset`: the delta `payload` is the
/// zlib-deflated entry content; `header` carries the base reference
/// (a [`PackHeader::RefDelta`] thin-base or an in-pack
/// [`PackHeader::OfsDelta`]).
pub fn pack_entry_delta(header: PackHeader, payload: &[u8], pack_offset: u64) -> PackEntry {
    let compressed = deflate_bytes(payload);
    PackEntry {
        header,
        header_size: header.size(payload.len() as u64) as u16,
        pack_offset,
        compressed: Some(compressed.clone()),
        compressed_size: compressed.len() as u64,
        crc32: None,
        decompressed_size: payload.len() as u64,
        trailer: None,
    }
}

/// One size field of a delta stream header: 7-bit little-endian
/// groups with the continuation bit.
fn delta_size_varint(size: u64) -> Vec<u8> {
    let mut out = Vec::new();
    let mut rest = size;
    loop {
        let mut byte = rest as u8 & 0x7f;
        rest >>= 7;
        if rest != 0 {
            byte |= 0x80;
        }
        out.push(byte);
        if rest == 0 {
            return out;
        }
    }
}

/// A delta stream copying the whole base and inserting `result` as
/// literal instructions — the fixture form of a real client's delta:
/// the produced object is the whole base followed by `result` (the
/// copy instruction exercises the applier's copy path, the insertion
/// the insert path). Bases of exactly 65536 bytes (the copy size's
/// special case) are rejected — fixture bases are small.
pub fn delta_stream(base_size: u64, result: &[u8]) -> Vec<u8> {
    assert!(
        base_size < 0x10000,
        "fixture delta bases stay below the copy size's 65536 special case"
    );
    let mut out = delta_size_varint(base_size);
    out.extend(delta_size_varint(base_size + result.len() as u64));
    if base_size > 0 {
        let mut cmd = 0x80u8;
        let mut size_bytes = Vec::new();
        let mut rest = base_size;
        for bit in [0x10u8, 0x20, 0x40] {
            if rest & 0xff != 0 {
                cmd |= bit;
                size_bytes.push(rest as u8);
            }
            rest >>= 8;
        }
        out.push(cmd);
        out.extend(size_bytes);
    }
    for chunk in result.chunks(127) {
        out.push(chunk.len() as u8);
        out.extend_from_slice(chunk);
    }
    out
}

/// Assembles a V2 sha1 pack from `entries`: the `PACK` header, the
/// entries (header + compressed bytes each), and the sha1 trailer
/// over the whole stream.
pub fn pack_bytes(entries: &[PackEntry]) -> Vec<u8> {
    let mut out = Vec::new();
    out.extend_from_slice(&gix_pack::data::header::encode(
        gix_pack::data::Version::V2,
        entries.len() as u32,
    ));
    for entry in entries {
        entry
            .header
            .write_to(entry.decompressed_size, &mut out)
            .expect("fixture entry headers write");
        out.extend_from_slice(
            entry
                .compressed
                .as_ref()
                .expect("fixture entries keep bytes"),
        );
    }
    let mut hasher = gix_hash::hasher(gix_hash::Kind::Sha1);
    hasher.update(&out);
    let digest = hasher
        .try_finalize()
        .expect("fixture pack trailers hash over the pack bytes");
    out.extend_from_slice(digest.as_slice());
    out
}

/// The decoded pack contents the rows assert over: the parsed header,
/// the entry objects re-hashed from their decompressed bytes, and the
/// verified trailer.
pub struct DecodedPack {
    /// The pack data version the header declares.
    pub version: gix_pack::data::Version,
    /// The object count the header declares.
    pub object_count: u32,
    /// The entries as `(oid, kind, object bytes)`, in pack order.
    pub objects: Vec<(ObjectId, gix_object::Kind, Vec<u8>)>,
    /// The trailer checksum, verified over the whole pack stream.
    pub trailer: ObjectId,
}

/// The section flush of a push request's command section
/// (client framing).
pub const FLUSH: &[u8] = b"0000";

/// The V0-framed push request's framed command head (ADR-013 §3,
/// push-captures.md §Request): the command pkt-lines (caps
/// NUL-attached on the first line only) and the command-section flush —
/// the chunk list every substrate's client-side body builder chains the
/// pack (or the options section) onto.
pub fn push_commands(caps: &str, commands: &[String]) -> Vec<Vec<u8>> {
    let mut chunks = Vec::new();
    let mut commands = commands.iter();
    if let Some(first) = commands.next() {
        let line = if caps.is_empty() {
            first.clone()
        } else {
            format!("{first}\0{caps}")
        };
        chunks.push(pkt(line.as_bytes()));
    }
    for command in commands {
        chunks.push(pkt(command.as_bytes()));
    }
    chunks.push(FLUSH.to_vec());
    chunks
}

fn pkt(payload: &[u8]) -> Vec<u8> {
    let total = payload.len() + 4;
    let mut out = Vec::with_capacity(total);
    out.extend_from_slice(format!("{total:04x}").as_bytes());
    out.extend_from_slice(payload);
    out
}

pub fn create_command(name: &str, new: ObjectId) -> String {
    format!("{} {new} {name}", "0".repeat(40))
}

pub fn delete_command(name: &str, current: ObjectId) -> String {
    format!("{current} {} {name}", "0".repeat(40))
}

/// A fixture object: its oid and the (kind, body) its loose bytes
/// carry — the pack-entry source.
pub struct PackedFixture {
    pub oid: ObjectId,
    pub kind: gix_object::Kind,
    pub body: Vec<u8>,
}

pub fn blob_fix(content: &[u8]) -> PackedFixture {
    let (oid, bytes) = blob_fixture(content);
    let (kind, body) = loose_body(&bytes);
    PackedFixture { oid, kind, body }
}

pub fn tree_fix(entries: &[(gix_object::tree::EntryKind, &str, ObjectId)]) -> PackedFixture {
    let (oid, bytes) = tree_fixture(entries);
    let (kind, body) = loose_body(&bytes);
    PackedFixture { oid, kind, body }
}

pub fn commit_fix(tree: ObjectId, parents: &[ObjectId], message: &str) -> PackedFixture {
    let (oid, bytes) = commit_fixture(tree, parents, message);
    let (kind, body) = loose_body(&bytes);
    PackedFixture { oid, kind, body }
}

/// A two-commit chain + its self-contained V2 sha1 pack — the e2e
/// pushes' content.
pub struct ChainPack {
    pub pack: Vec<u8>,
    pub commit_a: ObjectId,
    pub commit_b: ObjectId,
}

pub fn chain_pack() -> ChainPack {
    let blob_a = blob_fix(b"alpha");
    let blob_b = blob_fix(b"beta");
    let tree_a = tree_fix(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob_a.oid)]);
    let tree_b = tree_fix(&[
        (gix_object::tree::EntryKind::Blob, "a.txt", blob_a.oid),
        (gix_object::tree::EntryKind::Blob, "b.txt", blob_b.oid),
    ]);
    let commit_a = commit_fix(tree_a.oid, &[], "alpha");
    let commit_b = commit_fix(tree_b.oid, &[commit_a.oid], "beta");
    let mut offset = 0u64;
    let entries: Vec<_> = [&blob_a, &blob_b, &tree_a, &tree_b, &commit_a, &commit_b]
        .iter()
        .map(|fixture| {
            let entry = pack_entry_base(fixture.kind, &fixture.body, offset);
            offset += entry.bytes_in_pack();
            entry
        })
        .collect();
    ChainPack {
        pack: pack_bytes(&entries),
        commit_a: commit_a.oid,
        commit_b: commit_b.oid,
    }
}

/// Pkt-line payloads of a framed wire body (a flush as a trailing
/// empty payload).
pub fn parse_pkt_lines(body: &[u8]) -> Vec<Vec<u8>> {
    let mut lines = Vec::new();
    let mut cursor = body;
    while !cursor.is_empty() {
        let (header, rest) = cursor.split_at(4);
        let len = usize::from_str_radix(std::str::from_utf8(header).expect("hex header"), 16)
            .expect("hex length");
        cursor = rest;
        if len < 4 {
            lines.push(Vec::new());
            continue;
        }
        let (payload, rest) = cursor.split_at(len - 4);
        cursor = rest;
        lines.push(payload.to_vec());
    }
    lines
}

pub fn wired_text(payload: &[u8]) -> String {
    String::from_utf8_lossy(payload)
        .trim_end_matches('\n')
        .to_owned()
}

/// The sideband-wrapped report's inner pkt-line lines: every data
/// packet is band-1, exactly one outer flush ends the response, the
/// band payload's own framing ends at its inner flush.
pub fn sideband_report_lines(body: &[u8]) -> Vec<String> {
    let mut payload = Vec::new();
    let mut outer_flushes = 0;
    for packet in &parse_pkt_lines(body) {
        if packet.is_empty() {
            outer_flushes += 1;
            continue;
        }
        assert_eq!(packet[0], 1, "band-1 framing carries the report");
        payload.extend_from_slice(&packet[1..]);
    }
    assert_eq!(outer_flushes, 1, "exactly one outer flush");
    let mut inner = parse_pkt_lines(&payload);
    let last = inner.pop().expect("the inner framing is non-empty");
    assert!(
        last.is_empty(),
        "the inner flush terminates the report: {last:?}"
    );
    inner.iter().map(|p| wired_text(p)).collect()
}

/// Decodes the sink's bytes back: the pack header (version, object
/// count), every entry inflated and re-hashed (deltas would fail — the
/// doubles carry no packs), and the trailer verified against the
/// stream. Entries are parsed with the pack machinery's own
/// `data::input` iterator in `Verify` mode.
pub fn decode_pack(pack: &[u8]) -> DecodedPack {
    let header: [u8; gix_pack::data::header::SIZE] = pack[..gix_pack::data::header::SIZE]
        .try_into()
        .expect("pack header present");
    let (version, object_count) =
        gix_pack::data::header::decode(&header).expect("the pack header decodes");
    let mut entries = gix_pack::data::input::BytesToEntriesIter::new_from_header(
        io::Cursor::new(pack),
        gix_pack::data::input::Mode::Verify,
        gix_pack::data::input::EntryDataMode::Keep,
        gix_hash::Kind::Sha1,
    )
    .expect("the pack stream starts with a decodable header");
    let mut objects = Vec::new();
    let mut trailer = None;
    let mut decompressor = gix_zlib::Decompress::new();
    for entry in &mut entries {
        let entry = entry.expect("every pack entry decodes");
        if entry.trailer.is_some() {
            trailer = entry.trailer;
        }
        let kind = match entry.header {
            gix_pack::data::entry::Header::Commit => gix_object::Kind::Commit,
            gix_pack::data::entry::Header::Tree => gix_object::Kind::Tree,
            gix_pack::data::entry::Header::Blob => gix_object::Kind::Blob,
            gix_pack::data::entry::Header::Tag => gix_object::Kind::Tag,
            delta @ (gix_pack::data::entry::Header::OfsDelta { .. }
            | gix_pack::data::entry::Header::RefDelta { .. }) => {
                panic!("the doubles' packs carry no delta entries, got {delta:?}")
            }
        };
        let compressed = entry.compressed.expect("keep mode retains the entry bytes");
        let mut object = vec![0; entry.decompressed_size as usize];
        let inflated = gix_zlib::stream::inflate::read(
            &mut io::Cursor::new(compressed),
            &mut decompressor,
            &mut object,
        )
        .expect("the entry inflates");
        decompressor.reset();
        assert_eq!(
            inflated,
            object.len(),
            "the inflate filled the entry's size"
        );
        let id = gix_object::compute_hash(gix_hash::Kind::Sha1, kind, &object)
            .expect("the object re-hashes");
        objects.push((id, kind, object));
    }
    DecodedPack {
        version,
        object_count,
        objects,
        trailer: trailer.unwrap_or_else(|| {
            let body_len = pack.len() - gix_hash::Kind::Sha1.len_in_bytes();
            let mut hasher = gix_hash::hasher(gix_hash::Kind::Sha1);
            hasher.update(&pack[..body_len]);
            let computed = hasher
                .try_finalize()
                .expect("the zero-entry trailer hashes over the header alone");
            let digest: [u8; 20] =
                <&[u8] as std::convert::TryInto<[u8; 20]>>::try_into(&pack[body_len..])
                    .expect("the trailer is a full digest");
            let stored = ObjectId::from(digest);
            assert_eq!(computed, stored, "the zero-entry pack's trailer verifies");
            stored
        }),
    }
}