gunnar-store 1.6.0

//! **One** conformance suite, run against **every** [`GitServe`] backend.
//!
//! Written once here rather than copied per backend, because the point of the
//! contract is that a caller cannot tell the backends apart, and the only honest
//! way to claim that is to run identical assertions against all of them.
//!
//! # Why this is the most valuable thing in the crate
//!
//! There are two implementations of [`GitOps`] — Arrow-IPC in
//! `znippy/plugins/native/znippy-plugin-git` and gix in
//! `edda/crates/storage-git-gix` — and the whole point of having two is to be
//! able to compare them. **Both running this same suite is what makes that a
//! comparison rather than two anecdotes.** A number taken from a backend that
//! has not been shown to implement the same contract is a number about a
//! different program.
//!
//! # Both layers, because a backend is not two backends
//!
//! It was a [`GitOps`] suite until 2026-08-10 and it is a [`GitServe`] one now.
//! That is not scope creep: `GitServe: GitOps`, so this is the same single
//! contract family and the same single implementation answering it, and the
//! reading half is where the two most expensive disagreements live — whether
//! `HEAD` is a row, and whether `emit_pack` closes over the graph. A suite that
//! stopped at the storage half could not see either.
//!
//! # What it pins
//!
//! * a packfile in, every object out: `has` / `get` / `size` / `extents` over
//!   every entry, with the **verbatim entry bytes** asserted byte-identical,
//!   because that is what `Stored::bytes` promises;
//! * that `uncompressed_size` and `size()` are **post-resolution**, asserted on
//!   an entry that is deliberately an `OFS_DELTA` whose stored length differs
//!   from its resolved length — the one bug in this contract that passes every
//!   wire-byte and object-count test and only shows up as a quota gate that
//!   never fires;
//! * **sha1 and sha256**, every check run twice, so the sha256 path is exercised
//!   rather than inferred from sha1 being green;
//! * object ids checked against **ground truth taken from coreutils**, not from
//!   gix and not from a previous run of gunnar (see [`EMPTY_BLOB_SHA1`]);
//! * `update_ref` compare-and-swap **succeeding** and **failing**, with the
//!   failure asserted to have left the stored value untouched;
//! * `reachable` returning the whole closure, subtracting what the client has,
//!   tolerating a `have` it does not know, and **not** following a submodule;
//! * `gc` keeping everything a reference points at;
//! * `put_refs_cas` applying **every edit or none**, rejecting with a typed
//!   [`RefRejection`] rather than a message, and refusing a `must not exist`
//!   whose proposed value *equals* the value already there — the same-value shape,
//!   which is a silent success on a backend that hands the expectation through;
//! * `read` returning the **inflated, delta-resolved** object where `get`
//!   returns the stored entry, and `header`/`sizes` agreeing with it;
//! * **`HEAD` is not a row in `refs()` and `head()` answers it** — the two arms
//!   disagreed about this until 2026-08-10 and the suite could not see it,
//!   because it never exercised `HEAD` at all (the `GitServe` HEAD ruling);
//! * `emit_pack` emitting **the set it was given**, plus only the delta bases
//!   the format forces — never the graph closure of that set;
//! * `select` agreeing with `reachable` when it answers, and **declining** rather
//!   than inventing an answer for a tip it does not hold;
//! * `SecretStore::get_or_create` idempotent across a simulated restart, with a
//!   byte-identical key, and atomic under a thread race.
//!
//! # What it deliberately does not pin, and why
//!
//! * **That `gc` removes anything.** [`GitOps::gc`] is *"reachability, drop the
//!   dead index rows, then compact"*, but a backend that keeps a dead object is
//!   wasteful and a backend that drops a live one is broken. The suite asserts
//!   only the direction where a bug is fatal. Same reasoning as the old
//!   `contains_without_rescan` note: a stale answer is safe in exactly one
//!   direction, so only that direction is a contract.
//! * **That an `Extent` means anything in particular.** The contract says the
//!   unit is *"wherever this backend keeps the bytes"* and promises only that the
//!   pair is stable — the gix arm's is pack-local and cannot say which pack. So
//!   the suite asserts positional correspondence, absence, self-agreement and
//!   stability, and never compares an extent against the fixture's own offsets.
//! * **`REF_DELTA` ingestion.** gitoxide's thin-pack resolver never looks inside
//!   the pack it is reading, so the gix arm cannot take a `REF_DELTA` whose base
//!   is in the same pack. That is a recorded limitation of one arm, not a
//!   contract, so the fixture emits `OFS_DELTA`
//!   — which is what every modern client sends anyway.
//! * **Reference-name rules.** [`GitOps`] takes `&str` names and validates
//!   nothing. The rules and their measurements live in [`crate::refname`], with
//!   their own unit tests, and are the *caller's* to apply.
//!
//! # Demonstrably capable of failing
//!
//! A conformance suite that cannot catch a broken backend is worse than no
//! suite, because it manufactures confidence. Deliberately broken backends live
//! in `conformance/red.rs` and are run through this same suite, which asserts
//! **specific named checks go red** and, just as importantly, that the rest stay
//! green. Those tests failing means the suite has gone blind.
//!
//! # Using it from another crate
//!
//! ```no_run
//! use gunnar_store::conformance::{self, MemHarness};
//! let passed = conformance::assert_all_pass(&conformance::run_all(&MemHarness::default()));
//! assert!(passed > 0);
//! ```

pub mod pack;

use std::panic::AssertUnwindSafe;
use std::sync::atomic::{AtomicUsize, Ordering};

use anyhow::{bail, ensure, Context};
use git_storage_trait::{Caps, GitOps, GitServe, ObjType, RefCas, RefRejection, RefUpdate};
use gix_hash::ObjectId;
use gix_object::Kind;

use crate::refname::HEAD;
use crate::secret::{SecretStore, SecretStoreExt};

/// Ground truth: `printf 'blob 0\0' | sha1sum`.
///
/// Taken from coreutils, not from gix and not from gunnar. A round-trip that
/// only proves the store agrees with itself proves nothing; this is the outside
/// reference that makes the object-id assertion real.
pub const EMPTY_BLOB_SHA1: &str = "e69de29bb2d1d6434b8b29ae775ad8c2e48c5391";
/// Ground truth: `printf 'blob 0\0' | sha256sum`.
pub const EMPTY_BLOB_SHA256: &str =
    "473a0f4c3be8a93681a267e3b1e9a7dcda1185436fe141f7749120a303721813";
/// Ground truth: `printf 'tree 0\0' | sha1sum`.
pub const EMPTY_TREE_SHA1: &str = "4b825dc642cb6eb9a060e54bf8d69288fbee4904";
/// Ground truth: `printf 'tree 0\0' | sha256sum`.
pub const EMPTY_TREE_SHA256: &str =
    "6ef19b41225c5369f1c104d45d8d85efa9b057b53b14b4b9b939dd74decc5321";

/// The two hash kinds every backend is exercised against.
pub const HASH_KINDS: [gix_hash::Kind; 2] = [gix_hash::Kind::Sha1, gix_hash::Kind::Sha256];

/// One backend's two stores, as the suite sees them.
///
/// Two, where there were three. `objects` and `refs` were separate gunnar traits
/// and are now one [`GitOps`]; `secrets` is not on that contract and never will
/// be, because a secret is a server concern with no git-object equivalent
/// (the storage-abstraction census).
pub struct Backend<G, S> {
    /// The git store under test: the twelve, and the reading half above them.
    pub git: G,
    /// The secret store under test.
    pub secrets: S,
}

/// How the suite gets hold of a backend.
///
/// `slot` gives each check its own isolated store, and calling [`Harness::open`]
/// **twice with the same slot** must return a store that sees the first one's
/// data. That is how a restart is simulated without a process boundary. The
/// caller drops the first handle before reopening, so a backend holding an
/// exclusive file lock is free to take one.
pub trait Harness {
    /// The git store this harness produces.
    ///
    /// [`GitServe`] and not [`GitOps`], since 2026-08-10: the two are one
    /// contract family and one implementation, and a harness that could only
    /// produce the storage half would silence every check above the seam.
    type Git: GitServe;
    /// The secret store this harness produces.
    type Secrets: SecretStore;

    /// A name for reports.
    fn name(&self) -> &str;

    /// Open, or reopen, the store for `slot` at `hash_kind`.
    fn open(
        &self,
        slot: &str,
        hash_kind: gix_hash::Kind,
    ) -> anyhow::Result<Backend<Self::Git, Self::Secrets>>;
}

/// The result of one check against one hash kind.
#[derive(Debug)]
pub struct Outcome {
    /// The check's name.
    pub check: &'static str,
    /// Which hash kind it ran against.
    pub hash_kind: gix_hash::Kind,
    /// `Ok` if it passed.
    pub result: Result<(), String>,
}

impl Outcome {
    /// `true` if this check passed.
    pub fn passed(&self) -> bool {
        self.result.is_ok()
    }
}

type Check<H> = fn(&H, gix_hash::Kind) -> anyhow::Result<()>;

fn checks<H: Harness>() -> Vec<(&'static str, Check<H>)> {
    vec![
        (
            "object_id_matches_git_ground_truth",
            object_id_matches_git_ground_truth::<H>,
        ),
        (
            "pack_roundtrip_every_entry",
            pack_roundtrip_every_entry::<H>,
        ),
        (
            "get_returns_the_verbatim_entry_bytes",
            get_returns_the_verbatim_entry_bytes::<H>,
        ),
        (
            "size_is_post_resolution_not_stored_length",
            size_is_post_resolution_not_stored_length::<H>,
        ),
        (
            "an_absent_object_is_absent",
            an_absent_object_is_absent::<H>,
        ),
        (
            "extents_are_positional_and_stable",
            extents_are_positional_and_stable::<H>,
        ),
        ("put_pack_is_idempotent", put_pack_is_idempotent::<H>),
        (
            "put_pack_observed_publishes_a_count_that_moves",
            put_pack_observed_publishes_a_count_that_moves::<H>,
        ),
        ("refs_round_trip", refs_round_trip::<H>),
        (
            "update_ref_creates_only_when_absent",
            update_ref_creates_only_when_absent::<H>,
        ),
        (
            "update_ref_is_a_real_compare_and_swap",
            update_ref_is_a_real_compare_and_swap::<H>,
        ),
        (
            "update_ref_delete_checks_its_old_value",
            update_ref_delete_checks_its_old_value::<H>,
        ),
        (
            "refs_that_cannot_coexist_are_refused",
            refs_that_cannot_coexist_are_refused::<H>,
        ),
        (
            "put_refs_cas_applies_every_edit_or_none",
            put_refs_cas_applies_every_edit_or_none::<H>,
        ),
        (
            "head_is_not_a_row_and_has_its_own_accessor",
            head_is_not_a_row_and_has_its_own_accessor::<H>,
        ),
        (
            "tx_ids_report_what_happened",
            tx_ids_report_what_happened::<H>,
        ),
        (
            "put_lands_the_pack_and_the_refs_together",
            put_lands_the_pack_and_the_refs_together::<H>,
        ),
        (
            "reachable_returns_the_whole_closure",
            reachable_returns_the_whole_closure::<H>,
        ),
        (
            "reachable_subtracts_what_the_client_has",
            reachable_subtracts_what_the_client_has::<H>,
        ),
        (
            "reachable_tolerates_a_have_it_does_not_know",
            reachable_tolerates_a_have_it_does_not_know::<H>,
        ),
        (
            "reachable_does_not_follow_a_submodule",
            reachable_does_not_follow_a_submodule::<H>,
        ),
        (
            "reachable_follows_an_annotated_tag_to_its_target",
            reachable_follows_an_annotated_tag_to_its_target::<H>,
        ),
        (
            "read_is_inflated_and_delta_resolved",
            read_is_inflated_and_delta_resolved::<H>,
        ),
        (
            "header_and_sizes_agree_with_read",
            header_and_sizes_agree_with_read::<H>,
        ),
        (
            "emit_pack_emits_the_set_it_was_given",
            emit_pack_emits_the_set_it_was_given::<H>,
        ),
        (
            "select_agrees_with_the_graph_or_declines",
            select_agrees_with_the_graph_or_declines::<H>,
        ),
        (
            "gc_keeps_everything_a_ref_points_at",
            gc_keeps_everything_a_ref_points_at::<H>,
        ),
        (
            "the_store_survives_a_restart",
            the_store_survives_a_restart::<H>,
        ),
        ("secret_put_get_delete", secret_put_get_delete::<H>),
        (
            "secret_get_or_create_survives_restart",
            secret_get_or_create_survives_restart::<H>,
        ),
        (
            "secret_get_or_create_is_atomic",
            secret_get_or_create_is_atomic::<H>,
        ),
    ]
}

/// Every check name in the suite, in order.
pub fn check_names<H: Harness>() -> Vec<&'static str> {
    checks::<H>().into_iter().map(|(name, _)| name).collect()
}

/// Run the whole suite against `harness`, for both hash kinds.
///
/// Never panics: a check that panics is reported as a failed [`Outcome`], so a
/// broken backend produces data rather than an aborted run.
pub fn run_all<H: Harness>(harness: &H) -> Vec<Outcome> {
    let mut outcomes = Vec::new();
    for (check, run) in checks::<H>() {
        for hash_kind in HASH_KINDS {
            let result =
                match std::panic::catch_unwind(AssertUnwindSafe(|| run(harness, hash_kind))) {
                    Ok(Ok(())) => Ok(()),
                    Ok(Err(err)) => Err(format!("{err:#}")),
                    Err(panic) => Err(format!("panicked: {}", panic_message(&panic))),
                };
            outcomes.push(Outcome {
                check,
                hash_kind,
                result,
            });
        }
    }
    outcomes
}

fn panic_message(panic: &Box<dyn std::any::Any + Send>) -> String {
    if let Some(s) = panic.downcast_ref::<&str>() {
        (*s).to_owned()
    } else if let Some(s) = panic.downcast_ref::<String>() {
        s.clone()
    } else {
        "<non-string panic>".to_owned()
    }
}

/// The checks that failed, as `(check, hash kind, message)`.
pub fn failures(outcomes: &[Outcome]) -> Vec<(&'static str, gix_hash::Kind, String)> {
    outcomes
        .iter()
        .filter_map(|o| {
            o.result
                .as_ref()
                .err()
                .map(|msg| (o.check, o.hash_kind, msg.clone()))
        })
        .collect()
}

/// Panic with a full report unless every outcome passed. Returns how many
/// checks ran.
pub fn assert_all_pass(outcomes: &[Outcome]) -> usize {
    let failed = failures(outcomes);
    assert!(
        !outcomes.is_empty(),
        "the conformance suite ran zero checks: a broken runner, not a green backend"
    );
    if !failed.is_empty() {
        let report = failed
            .iter()
            .map(|(check, kind, msg)| format!("  {check} [{kind}]: {msg}"))
            .collect::<Vec<_>>()
            .join("\n");
        panic!(
            "{} of {} conformance checks failed:\n{report}",
            failed.len(),
            outcomes.len()
        );
    }
    outcomes.len()
}

// ───────────────────────────── the fixture ───────────────────────────────────

/// A small repository in one packfile, with every id named.
///
/// One pack rather than several, because [`GitOps::put_pack`] is the only way
/// into a store and a suite that needed several would be asserting an ordering
/// the contract does not have.
pub struct Repo {
    /// The pack, and the ground truth about its entries.
    pub fixture: pack::Fixture,
    /// The empty blob — checked against coreutils.
    pub empty_blob: ObjectId,
    /// The empty tree — checked against coreutils.
    pub empty_tree: ObjectId,
    /// `file.txt`'s content.
    pub blob: ObjectId,
    /// `other.txt`'s content, and **the one entry stored as an `OFS_DELTA`**.
    pub delta_blob: ObjectId,
    /// The tree of the first commit.
    pub tree1: ObjectId,
    /// The root commit.
    pub commit1: ObjectId,
    /// The tree of the second commit.
    pub tree2: ObjectId,
    /// The second commit, whose parent is [`Repo::commit1`].
    pub commit2: ObjectId,
    /// An annotated tag on [`Repo::commit1`].
    pub tag: ObjectId,
    /// A tree with a gitlink entry.
    pub sub_tree: ObjectId,
    /// A commit whose tree is [`Repo::sub_tree`].
    pub sub_commit: ObjectId,
    /// The commit [`Repo::sub_tree`]'s gitlink names. **Not in the pack** — it
    /// lives in another repository, which is the whole point.
    pub gitlink: ObjectId,
}

fn oid(hash_kind: gix_hash::Kind, kind: Kind, content: &[u8]) -> anyhow::Result<ObjectId> {
    Ok(gix_object::compute_hash(hash_kind, kind, content).map_err(gix_error::Error::from)?)
}

fn tree_bytes(entries: &[(&str, u32, &gix_hash::oid)]) -> Vec<u8> {
    // Tree entries are sorted by name, and git's sort treats a tree's name as
    // though it ended in `/`. Every name here is a plain file or one gitlink, so
    // a plain sort is the same sort.
    let mut sorted: Vec<_> = entries.to_vec();
    sorted.sort_by_key(|(name, _, _)| *name);
    let mut out = Vec::new();
    for (name, mode, id) in sorted {
        out.extend_from_slice(format!("{mode:o} ").as_bytes());
        out.extend_from_slice(name.as_bytes());
        out.push(0);
        out.extend_from_slice(id.as_bytes());
    }
    out
}

fn commit_bytes(tree: &gix_hash::oid, parents: &[ObjectId], message: &str) -> Vec<u8> {
    let mut out = format!("tree {}\n", tree.to_hex());
    for parent in parents {
        out.push_str(&format!("parent {}\n", parent.to_hex()));
    }
    out.push_str("author Gunnar Test <gunnar@example.invalid> 1700000000 +0000\n");
    out.push_str("committer Gunnar Test <gunnar@example.invalid> 1700000000 +0000\n\n");
    out.push_str(message);
    out.push('\n');
    out.into_bytes()
}

fn tag_bytes(target: &gix_hash::oid, tag: &str) -> Vec<u8> {
    let mut out = format!("object {}\n", target.to_hex());
    out.push_str("type commit\n");
    out.push_str(&format!("tag {tag}\n"));
    out.push_str("tagger Gunnar Test <gunnar@example.invalid> 1700000000 +0000\n\n");
    out.push_str("a tag\n");
    out.into_bytes()
}

const BLOB: &[u8] = b"gunnar conformance blob\n";
const DELTA_BLOB: &[u8] =
    b"a second blob, stored as a delta against the first, and deliberately a \
      different length from the delta stream that encodes it\n";

/// The fixture, built for `hash_kind`.
pub fn repo(hash_kind: gix_hash::Kind) -> anyhow::Result<Repo> {
    let empty_blob = oid(hash_kind, Kind::Blob, &[])?;
    let empty_tree = oid(hash_kind, Kind::Tree, &[])?;
    let blob = oid(hash_kind, Kind::Blob, BLOB)?;
    let delta_blob = oid(hash_kind, Kind::Blob, DELTA_BLOB)?;

    let tree1_bytes = tree_bytes(&[("file.txt", 0o100644, &blob)]);
    let tree1 = oid(hash_kind, Kind::Tree, &tree1_bytes)?;
    let commit1_bytes = commit_bytes(&tree1, &[], "root");
    let commit1 = oid(hash_kind, Kind::Commit, &commit1_bytes)?;

    let tree2_bytes = tree_bytes(&[
        ("file.txt", 0o100644, &blob),
        ("other.txt", 0o100644, &delta_blob),
    ]);
    let tree2 = oid(hash_kind, Kind::Tree, &tree2_bytes)?;
    let commit2_bytes = commit_bytes(&tree2, &[commit1], "second");
    let commit2 = oid(hash_kind, Kind::Commit, &commit2_bytes)?;

    let tag_bytes = tag_bytes(&commit1, "v0.1.0");
    let tag = oid(hash_kind, Kind::Tag, &tag_bytes)?;

    // A commit in another repository. Absent by construction, so a backend that
    // follows the gitlink either fails or invents.
    let gitlink = absent_oid(hash_kind, 0xcd);
    let sub_tree_bytes = tree_bytes(&[
        ("file.txt", 0o100644, &blob),
        ("vendor", 0o160000, &gitlink),
    ]);
    let sub_tree = oid(hash_kind, Kind::Tree, &sub_tree_bytes)?;
    let sub_commit_bytes = commit_bytes(&sub_tree, &[], "with a submodule");
    let sub_commit = oid(hash_kind, Kind::Commit, &sub_commit_bytes)?;

    let mut builder = pack::PackBuilder::new(hash_kind);
    let blob_at = builder.object(Kind::Blob, BLOB);
    // The delta. Its base precedes it, it resolves to a whole blob, and its
    // stored length is not its resolved length — which is the property
    // `size_is_post_resolution_not_stored_length` exists for.
    builder.delta(blob_at, DELTA_BLOB);
    builder.object(Kind::Blob, Vec::new());
    builder.object(Kind::Tree, Vec::new());
    builder.object(Kind::Tree, tree1_bytes);
    builder.object(Kind::Commit, commit1_bytes);
    builder.object(Kind::Tree, tree2_bytes);
    builder.object(Kind::Commit, commit2_bytes);
    builder.object(Kind::Tag, tag_bytes);
    builder.object(Kind::Tree, sub_tree_bytes);
    builder.object(Kind::Commit, sub_commit_bytes);

    Ok(Repo {
        fixture: builder.build()?,
        empty_blob,
        empty_tree,
        blob,
        delta_blob,
        tree1,
        commit1,
        tree2,
        commit2,
        tag,
        sub_tree,
        sub_commit,
        gitlink,
    })
}

fn absent_oid(hash_kind: gix_hash::Kind, fill: u8) -> ObjectId {
    let hex: String = format!("{fill:02x}").repeat(hash_kind.len_in_bytes());
    ObjectId::from_hex(hex.as_bytes()).expect("constructed a valid hex oid")
}

fn hex(id: &ObjectId) -> String {
    id.to_hex().to_string()
}

fn sorted(mut ids: Vec<Vec<u8>>) -> Vec<Vec<u8>> {
    ids.sort();
    ids
}

fn as_bytes(ids: &[ObjectId]) -> Vec<Vec<u8>> {
    sorted(ids.iter().map(|id| id.as_slice().to_vec()).collect())
}

/// A harness's backend, named once so the checks below can say it in a line.
type Opened<H> = Backend<<H as Harness>::Git, <H as Harness>::Secrets>;

/// Open a slot and push the fixture pack into it.
fn loaded<H: Harness>(
    harness: &H,
    slot: &str,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<(Opened<H>, Repo)> {
    let backend = harness.open(slot, hash_kind)?;
    let repo = repo(hash_kind)?;
    backend
        .git
        .put_pack(&repo.fixture.bytes)
        .context("put_pack refused the fixture pack")?;
    Ok((backend, repo))
}

// ───────────────────────────── object checks ─────────────────────────────────

fn object_id_matches_git_ground_truth<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "ground_truth", hash_kind)?;
    let (want_blob, want_tree) = match hash_kind {
        gix_hash::Kind::Sha1 => (EMPTY_BLOB_SHA1, EMPTY_TREE_SHA1),
        gix_hash::Kind::Sha256 => (EMPTY_BLOB_SHA256, EMPTY_TREE_SHA256),
        other => bail!("no ground truth recorded for hash kind {other}"),
    };
    ensure!(
        hex(&repo.empty_blob) == want_blob,
        "the empty blob hashed to {} but git ({hash_kind}) says {want_blob}",
        hex(&repo.empty_blob)
    );
    ensure!(
        hex(&repo.empty_tree) == want_tree,
        "the empty tree hashed to {} but git ({hash_kind}) says {want_tree}",
        hex(&repo.empty_tree)
    );
    // The ids are ground truth; that the *store* answers to them is the check.
    for (what, id) in [("blob", repo.empty_blob), ("tree", repo.empty_tree)] {
        ensure!(
            b.git.has(id.as_slice())?,
            "the store does not hold the empty {what}, whose id git itself gives as {}",
            hex(&id)
        );
    }
    Ok(())
}

fn pack_roundtrip_every_entry<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "pack_roundtrip", hash_kind)?;
    for entry in &repo.fixture.entries {
        let id = entry.id;
        ensure!(
            b.git.has(id.as_slice())?,
            "{:?} {} was in the pack but has() says no",
            entry.kind,
            hex(&id)
        );
        let stored = b
            .git
            .get(id.as_slice())?
            .with_context(|| format!("get() found no {:?} {}", entry.kind, hex(&id)))?;
        ensure!(
            stored.obj_type == entry.obj_type,
            "{} was stored as {} and comes back as {}; the type is the entry's type IN THE PACK, \
             so a resolved type here means the verbatim bytes have been reinterpreted",
            hex(&id),
            entry.obj_type.as_str(),
            stored.obj_type.as_str()
        );
        ensure!(
            stored.uncompressed_size == entry.content.len() as u64,
            "{} reports {} bytes and inflates to {}",
            hex(&id),
            stored.uncompressed_size,
            entry.content.len()
        );
        ensure!(
            b.git.size(id.as_slice())? == Some(entry.content.len() as u64),
            "size() and get().uncompressed_size disagree about {}",
            hex(&id)
        );
        ensure!(
            stored.extent == b.git.extents(&[id.as_slice()])?[0].context("extents() lost it")?,
            "get() and extents() disagree about where {} is",
            hex(&id)
        );
    }
    Ok(())
}

/// **`Stored::bytes` is *"the pack entry, header and all, byte for byte as the
/// client sent them"*.**
///
/// Its own doc says why this is a struct and not a `Vec<u8>`: the verbatim bytes
/// are the truth and the resolved object a derived cache. A backend that
/// inflates and re-deflates produces a pack that passes `index-pack --strict`
/// and `fsck` while sending measured 18.4x the wire bytes — the failure nobody
/// notices. This is the check that notices.
fn get_returns_the_verbatim_entry_bytes<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "verbatim", hash_kind)?;
    for entry in &repo.fixture.entries {
        let stored = b
            .git
            .get(entry.id.as_slice())?
            .with_context(|| format!("get() found no {}", hex(&entry.id)))?;
        let want = entry.bytes(&repo.fixture.bytes);
        ensure!(
            stored.bytes.len() == want.len(),
            "{} came back as {} entry bytes; {} went in",
            hex(&entry.id),
            stored.bytes.len(),
            want.len()
        );
        ensure!(
            stored.bytes == want,
            "{} came back with different entry bytes than went in; the payload has been \
             re-encoded somewhere, and a pack-copying serve path would send it wrong",
            hex(&entry.id)
        );
    }
    Ok(())
}

/// **Post-resolution, not the stored length.**
///
/// `Stored::uncompressed_size` is *"what the object inflates to once its chain
/// is applied, which is not `bytes.len()` for a delta"*. A backend that returns
/// the stored extent here passes every wire-byte and object-count test in this
/// repository, because the number is only consulted by an object-size ceiling
/// that never fires on the test corpora — and then lets an oversized object
/// through in production.
///
/// Asserted on an entry the fixture deliberately stores as an `OFS_DELTA`, so
/// the two numbers cannot coincide.
fn size_is_post_resolution_not_stored_length<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "post_resolution", hash_kind)?;
    let entry = repo
        .fixture
        .entry(&repo.delta_blob)
        .context("the fixture lost its delta entry")?;
    ensure!(
        entry.obj_type == ObjType::OfsDelta,
        "the fixture's delta entry is stored as {}; this check measures nothing without a delta",
        entry.obj_type.as_str()
    );
    let resolved = entry.content.len() as u64;
    let on_the_wire = entry.bytes(&repo.fixture.bytes).len() as u64;
    ensure!(
        resolved != on_the_wire,
        "the fixture's delta stores exactly as many bytes as it resolves to, so this check \
         cannot tell the two answers apart"
    );

    let stored = b
        .git
        .get(repo.delta_blob.as_slice())?
        .context("get() found no delta object")?;
    ensure!(
        stored.uncompressed_size == resolved,
        "a delta object reports {} bytes; it resolves to {resolved} and its entry is {on_the_wire}",
        stored.uncompressed_size
    );
    ensure!(
        b.git.size(repo.delta_blob.as_slice())? == Some(resolved),
        "size() of a delta object is not its post-resolution size"
    );
    Ok(())
}

fn an_absent_object_is_absent<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    // Loaded first, so "the store is empty" is not what makes this pass.
    let (b, _repo) = loaded(harness, "absent", hash_kind)?;
    let missing = absent_oid(hash_kind, 0xab);
    ensure!(
        !b.git.has(missing.as_slice())?,
        "has() invented {}",
        hex(&missing)
    );
    ensure!(
        b.git.get(missing.as_slice())?.is_none(),
        "get() invented {}",
        hex(&missing)
    );
    ensure!(
        b.git.size(missing.as_slice())?.is_none(),
        "size() invented {}",
        hex(&missing)
    );
    ensure!(
        b.git.extents(&[missing.as_slice()])? == vec![None],
        "extents() invented {}",
        hex(&missing)
    );
    Ok(())
}

/// **`extents` answers positionally, and the pair is stable.**
///
/// The contract promises nothing about what an extent *means* — the unit is
/// "wherever this backend keeps the bytes", and the gix arm's is pack-local. So
/// this asserts the three things it does promise: the answer lines up with the
/// question, an absent object is `None` rather than a shifted answer, and the
/// pair does not move.
fn extents_are_positional_and_stable<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "extents", hash_kind)?;
    let missing = absent_oid(hash_kind, 0xab);
    let query: Vec<&[u8]> = vec![
        repo.commit1.as_slice(),
        missing.as_slice(),
        repo.blob.as_slice(),
        repo.delta_blob.as_slice(),
    ];

    let first = b.git.extents(&query)?;
    ensure!(
        first.len() == query.len(),
        "extents() answered {} of {} questions",
        first.len(),
        query.len()
    );
    ensure!(
        first[1].is_none(),
        "extents() answered for an object that is not there; every later answer in the \
         vector is now attributed to the wrong oid"
    );
    for (i, id) in [(0, repo.commit1), (2, repo.blob), (3, repo.delta_blob)] {
        ensure!(
            first[i].is_some(),
            "extents() has no extent for {}, which it holds",
            hex(&id)
        );
    }

    let second = b.git.extents(&query)?;
    ensure!(
        first == second,
        "two identical extents() calls answered differently: {first:?} then {second:?}"
    );
    Ok(())
}

/// The same pack, twice. Every object still readable and unchanged.
///
/// The ack path records a pack as pending and hands it to the indexer in that
/// order, and a fast worker can arrive at either end first — so absorbing a pack
/// that is already absorbed is a case that happens, not a case that is imagined.
fn put_pack_is_idempotent<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "idempotent", hash_kind)?;
    let before = b
        .git
        .get(repo.commit1.as_slice())?
        .context("the first push did not land")?;
    b.git
        .put_pack(&repo.fixture.bytes)
        .context("the same pack was refused the second time")?;
    let after = b
        .git
        .get(repo.commit1.as_slice())?
        .context("the second push lost an object the first one had")?;
    ensure!(
        before.bytes == after.bytes && before.uncompressed_size == after.uncompressed_size,
        "re-pushing a pack changed an object that was already there"
    );
    for entry in &repo.fixture.entries {
        ensure!(
            b.git.has(entry.id.as_slice())?,
            "{} went missing when its pack was pushed twice",
            hex(&entry.id)
        );
    }
    Ok(())
}

/// **A store that claims a live object count is held to it moving; one that
/// claims nothing passes by name.**
///
/// # The property, and why each clause is here
///
/// A console renders `done/total` while a fetch of somebody else's repository
/// runs for minutes. What makes that number worth rendering is not that it is
/// eventually right — `(0, N)` then `(N, N)` is eventually right and tells a
/// watcher nothing — it is that it **moves**. So:
///
/// * **more than two distinct `done` values.** Two is exactly what a store that
///   publishes nothing produces once the layers above it have added their
///   endpoints, which is the shape this check exists to make impossible to ship
///   again.
/// * **monotone non-decreasing.** A count that goes backwards is a percentage
///   that goes backwards.
/// * **`total` constant, and equal to the pack's own object count.** The
///   denominator is a fact about the packfile header; a store that renumbered it
///   mid-transfer would be publishing two different packs.
/// * **the last publish is `done == total`.** Otherwise a bar stops at 97% on a
///   push that worked, which reads exactly like one that hung.
/// * **and the pack still lands.** Publishing is not allowed to be a mode: the
///   objects are read back afterwards, from the same store, through the same
///   `has`.
///
/// # `publishes_pack_progress()` is asked FIRST, and that is the red first
///
/// A store with no entry loop of its own — `storage-git-gix`, where `gix_pack`
/// owns the split — cannot answer this question and must not pretend to. But
/// "publishes nothing" and "publishes something broken" both produce an empty
/// series, so a check that simply tolerated an empty one would be green for a
/// backend whose implementation had rotted away. The declaration separates
/// them: a `false` passes **and says so, naming the backend**, so a store that
/// quietly stopped publishing shows up as a changed declaration rather than as
/// nothing at all.
///
/// # 300 blobs, not the shared fixture
///
/// [`repo`]'s pack is seven entries and its whole point is that every *kind* of
/// entry is in it. This check needs the opposite — many entries of one boring
/// kind — because "the count moved" is not observable on a pack small enough
/// that a store could publish per pack and look identical. The blobs are
/// distinct so no two collapse to one object.
fn put_pack_observed_publishes_a_count_that_moves<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    const BLOBS: usize = 300;

    let b = harness.open("pack_progress", hash_kind)?;

    let mut builder = pack::PackBuilder::new(hash_kind);
    for n in 0..BLOBS {
        builder.object(Kind::Blob, format!("object number {n}\n").into_bytes());
    }
    let fixture = builder.build().context("building a many-object pack")?;

    /// Every publish, in order. `Mutex` because [`GitOps::put_pack_observed`]
    /// takes `&self` and the sink is `Sync`: a store is free to publish from
    /// whichever thread is doing the splitting, and the contract says so.
    #[derive(Default)]
    struct Series(std::sync::Mutex<Vec<(u64, u64)>>);

    impl git_storage_trait::PackProgress for Series {
        fn indexed(&self, done: u64, total: u64) {
            self.0
                .lock()
                .unwrap_or_else(|p| p.into_inner())
                .push((done, total));
        }
    }

    let series = Series::default();
    b.git
        .put_pack_observed(&fixture.bytes, &series)
        .context("put_pack_observed refused a pack put_pack would have taken")?;

    // The pack landed. Asserted before the series, so a store that publishes
    // beautifully and stores nothing fails on the thing that matters.
    for entry in &fixture.entries {
        ensure!(
            b.git.has(entry.id.as_slice())?,
            "{} is not in the store after put_pack_observed",
            hex(&entry.id)
        );
    }

    let seen = series.0.into_inner().unwrap_or_else(|p| p.into_inner());

    if !b.git.publishes_pack_progress() {
        ensure!(
            seen.is_empty(),
            "`{}` answers publishes_pack_progress() = false and then published \
             {} reading(s). One of the two is a lie, and a caller believes the \
             declaration.",
            harness.name(),
            seen.len()
        );
        // Not a silent pass: the report has to be able to show that this backend
        // declined rather than that the check was skipped.
        eprintln!(
            "conformance: `{}` declines a live pack-progress count \
             (publishes_pack_progress() = false); the property is vacuous for it",
            harness.name()
        );
        return Ok(());
    }

    let total = BLOBS as u64;
    ensure!(
        !seen.is_empty(),
        "`{}` answers publishes_pack_progress() = true and published nothing",
        harness.name()
    );
    for (done, declared) in &seen {
        ensure!(
            *declared == total,
            "the denominator moved: a reading declared {declared} objects for a pack \
             of {total}"
        );
        ensure!(
            *done <= total,
            "a reading claims {done} of {total} objects are in"
        );
    }
    ensure!(
        seen.windows(2).all(|w| w[0].0 <= w[1].0),
        "the count went backwards: {:?}",
        seen.iter().map(|(d, _)| *d).collect::<Vec<_>>()
    );
    ensure!(
        seen.last().map(|(done, _)| *done) == Some(total),
        "the last reading is {:?}, not ({total}, {total}) — a bar that stops short \
         of the end reads exactly like a transfer that hung",
        seen.last()
    );
    let distinct = {
        let mut values: Vec<u64> = seen.iter().map(|(done, _)| *done).collect();
        values.dedup();
        values.len()
    };
    ensure!(
        distinct > 2,
        "the count has {distinct} distinct value(s) over a pack of {total} objects. \
         Two is what a store that publishes nothing looks like once the layers above \
         it have added `(0, N)` and `(N, N)`; a count that moves is the whole claim."
    );
    Ok(())
}

// ───────────────────────────── ref checks ────────────────────────────────────

fn refs_round_trip<H: Harness>(harness: &H, hash_kind: gix_hash::Kind) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "refs_roundtrip", hash_kind)?;
    b.git.put_refs(&[
        RefUpdate::set("refs/heads/main", hex(&repo.commit1)),
        RefUpdate::set("refs/tags/v0.1.0", hex(&repo.tag)).with_peeled(hex(&repo.commit1)),
    ])?;

    let rows = b.git.refs()?;
    let names: Vec<&str> = rows.iter().map(|r| r.name.as_str()).collect();
    ensure!(
        names == vec!["refs/heads/main", "refs/tags/v0.1.0"],
        "refs() returned {names:?}; the namespace is advertised in this order"
    );
    let main = &rows[0];
    ensure!(
        main.oid.as_deref() == Some(repo.commit1.as_slice()),
        "refs/heads/main came back pointing at {:?}",
        main.oid
    );
    let tag = &rows[1];
    ensure!(
        tag.peeled.as_deref() == Some(repo.commit1.as_slice()),
        "an annotated tag came back with peeled {:?}; a client cannot resolve the tag without it",
        tag.peeled
    );

    // A delete is `target: None` and `symref_target: None`, which is the only
    // way the contract expresses one.
    b.git.put_refs(&[RefUpdate::delete("refs/tags/v0.1.0")])?;
    let after = b.git.refs()?;
    let remaining: Vec<&str> = after.iter().map(|r| r.name.as_str()).collect();
    ensure!(
        remaining == vec!["refs/heads/main"],
        "after a delete, refs() returned {remaining:?}"
    );
    Ok(())
}

/// **`old: None` means "must not exist".**
///
/// Exactly one creator wins is the property a receive-pack relies on to
/// arbitrate two racing pushes. Both the differing-value and the **same-value**
/// case are asserted, because gix's file ref store short-circuits an edit whose
/// new value equals the current one *before* it evaluates the expectation:
/// a backend that hands the expectation straight through answers `Ok`
/// to a create that must fail, and two racers proposing the same oid would both
/// be told they won.
fn update_ref_creates_only_when_absent<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "ref_create", hash_kind)?;
    let name = "refs/heads/main";
    b.git
        .update_ref(name, None, Some(repo.commit1.as_slice()))?;
    ensure!(
        b.git.refs()?.iter().any(|r| r.name == name),
        "a create was accepted and the reference is not there"
    );

    for (what, target) in [
        ("a different value", repo.commit2),
        ("ITS CURRENT VALUE", repo.commit1),
    ] {
        if b.git
            .update_ref(name, None, Some(target.as_slice()))
            .is_ok()
        {
            bail!(
                "creating an existing reference at {what} was accepted, so `old: None` is not \
                 a create guard on this backend"
            );
        }
        let row = b
            .git
            .refs()?
            .into_iter()
            .find(|r| r.name == name)
            .context("the refused create removed the reference")?;
        ensure!(
            row.oid.as_deref() == Some(repo.commit1.as_slice()),
            "a rejected create still changed the stored value"
        );
    }
    Ok(())
}

fn update_ref_is_a_real_compare_and_swap<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "ref_cas", hash_kind)?;
    let name = "refs/heads/main";
    b.git
        .update_ref(name, None, Some(repo.commit1.as_slice()))?;

    // The swap that must succeed.
    b.git
        .update_ref(
            name,
            Some(repo.commit1.as_slice()),
            Some(repo.commit2.as_slice()),
        )
        .context("a compare-and-swap at the correct old value was refused")?;
    let current = |b: &Opened<H>| -> anyhow::Result<Option<Vec<u8>>> {
        Ok(b.git
            .refs()?
            .into_iter()
            .find(|r| r.name == name)
            .and_then(|r| r.oid))
    };
    ensure!(
        current(&b)?.as_deref() == Some(repo.commit2.as_slice()),
        "a successful compare-and-swap did not move the reference"
    );

    // The swap that must fail: the reference holds commit2 and this claims
    // commit1.
    if b.git
        .update_ref(
            name,
            Some(repo.commit1.as_slice()),
            Some(repo.tag.as_slice()),
        )
        .is_ok()
    {
        bail!("a stale-old-value update was ACCEPTED");
    }
    // The assertion that matters is on applied state, not on the error: a
    // backend that rejects and then writes anyway is still broken.
    ensure!(
        current(&b)?.as_deref() == Some(repo.commit2.as_slice()),
        "a rejected compare-and-swap still changed the stored value"
    );

    // And an update of a reference that is not there at all.
    if b.git
        .update_ref(
            "refs/heads/absent",
            Some(repo.commit1.as_slice()),
            Some(repo.commit2.as_slice()),
        )
        .is_ok()
    {
        bail!("an update naming an old value for a reference that does not exist was accepted");
    }
    Ok(())
}

fn update_ref_delete_checks_its_old_value<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "ref_delete", hash_kind)?;
    let doomed = "refs/heads/doomed";
    let kept = "refs/heads/kept";
    for name in [doomed, kept] {
        b.git
            .update_ref(name, None, Some(repo.commit1.as_slice()))?;
    }

    if b.git
        .update_ref(kept, Some(repo.commit2.as_slice()), None)
        .is_ok()
    {
        bail!("a delete with a stale old value was accepted");
    }
    ensure!(
        b.git.refs()?.iter().any(|r| r.name == kept),
        "a rejected delete removed the reference anyway"
    );

    b.git
        .update_ref(doomed, Some(repo.commit1.as_slice()), None)
        .context("a delete at the correct old value was refused")?;
    let remaining: Vec<String> = b.git.refs()?.into_iter().map(|r| r.name).collect();
    ensure!(
        remaining == vec![kept.to_string()],
        "after the delete, refs() returned {remaining:?}"
    );
    Ok(())
}

/// **`refs/heads/x` and `refs/heads/x/y` cannot both exist.**
///
/// A loose-ref backend cannot hold them — one is a file where the other needs a
/// directory — and git refuses the pair on *every* backend so a repository
/// stays convertible. Measured, git 2.53.0, `git push` into a bare repository:
///
/// | server already holds | push       | answer                       |
/// |---|---|---|
/// | `refs/heads/x`       | `x/y`      | `! [remote rejected] (refname conflict)` |
/// | `refs/heads/x/y`     | `x`        | `! [remote rejected] (refname conflict)` |
/// | nothing              | both, atomic | `cannot process 'refs/heads/x' and 'refs/heads/x/y' at the same time` |
///
/// It is a conformance check and not a per-backend test because a filesystem
/// backend gets it for free from the filesystem and an in-process one does not:
/// a rule only one backend enforces is a repository that stops being portable
/// the moment it is served from another. The gix arm pre-checks it by name
/// (`storage-git-gix`'s `precheck_name_conflict`), so this is the contract both
/// arms are held to, not a rule invented here.
///
/// The last paragraph is the control: `refs/heads/x` and `refs/heads/xy` share a
/// prefix, are not a conflict, and a backend that refused them would break
/// ordinary branch naming while passing everything above.
fn refs_that_cannot_coexist_are_refused<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "ref_conflict", hash_kind)?;
    let at = hex(&repo.commit1);

    b.git.put_refs(&[RefUpdate::set("refs/heads/x", &at)])?;
    if b.git
        .put_refs(&[RefUpdate::set("refs/heads/x/y", &at)])
        .is_ok()
    {
        bail!("refs/heads/x/y was created beside refs/heads/x; git refuses the pair");
    }
    ensure!(
        !b.git.refs()?.iter().any(|r| r.name == "refs/heads/x/y"),
        "the refused create left refs/heads/x/y behind"
    );

    // The other direction, on a fresh store: a backend could hold one ordering
    // and not the other.
    let (b, repo) = loaded(harness, "ref_conflict_reverse", hash_kind)?;
    let at = hex(&repo.commit1);
    b.git.put_refs(&[RefUpdate::set("refs/heads/x/y", &at)])?;
    if b.git
        .put_refs(&[RefUpdate::set("refs/heads/x", &at)])
        .is_ok()
    {
        bail!("refs/heads/x was created beside refs/heads/x/y; git refuses the pair");
    }

    // Both in one batch, which git refuses as a batch rather than applying
    // either half.
    let (b, repo) = loaded(harness, "ref_conflict_batch", hash_kind)?;
    let at = hex(&repo.commit1);
    if b.git
        .put_refs(&[
            RefUpdate::set("refs/heads/x", &at),
            RefUpdate::set("refs/heads/x/y", &at),
        ])
        .is_ok()
    {
        bail!("one batch created both refs/heads/x and refs/heads/x/y");
    }
    ensure!(
        b.git.refs()?.is_empty(),
        "a refused batch applied half of itself: {:?}",
        b.git.refs()?
    );

    // The control.
    let (b, repo) = loaded(harness, "ref_conflict_control", hash_kind)?;
    let at = hex(&repo.commit1);
    b.git
        .put_refs(&[
            RefUpdate::set("refs/heads/x", &at),
            RefUpdate::set("refs/heads/xy", &at),
        ])
        .context(
            "refs/heads/x and refs/heads/xy are not a conflict and must coexist; a backend that \
         refuses every shared prefix passes every assertion above and breaks branch naming",
        )?;
    ensure!(
        b.git.refs()?.len() == 2,
        "the control's two non-conflicting references did not both land"
    );
    Ok(())
}

/// **`put_refs_cas`: every edit or none, and it says which one lost.**
///
/// The three properties, in the order a receive-pack depends on them:
///
/// * a batch whose expectations all hold applies **all** of it;
/// * a batch with one stale expectation applies **none** of it — asserted on the
///   stored values, because a backend that rejects and writes anyway is still
///   broken — and the rejection is recoverable as a [`RefRejection`] naming the
///   ref that lost. **Never by reading the message**: `Error::is_cas_failure()`
///   existed for exactly this and its loss is what the `GitServe` typed-CAS ruling restored;
/// * **The same-value case**: an `old: None` — *must not exist* — is refused even when the
///   value proposed is the value already there. gix's file ref store
///   short-circuits that edit and never evaluates the expectation, so a backend
///   handing the expectation through answers `Ok` to a create that must fail and
///   two racers proposing the same oid are both told they won.
fn put_refs_cas_applies_every_edit_or_none<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "refs_cas", hash_kind)?;
    let at = |name: &str| -> anyhow::Result<Option<Vec<u8>>> {
        Ok(b.git
            .refs()?
            .into_iter()
            .find(|r| r.name == name)
            .and_then(|r| r.oid))
    };
    let (one, two) = ("refs/heads/one", "refs/heads/two");

    // An empty batch is a no-op that succeeds: a deletions-free push with
    // nothing to apply calls this.
    b.git
        .put_refs_cas(&[])
        .context("an empty atomic batch was refused; it is a no-op, not an error")?;

    // Two creates, both of which must hold.
    b.git
        .put_refs_cas(&[
            RefCas {
                name: one.to_owned(),
                old: None,
                new: Some(repo.commit1.as_slice()),
                peeled: None,
            },
            RefCas {
                name: two.to_owned(),
                old: None,
                new: Some(repo.commit1.as_slice()),
                peeled: None,
            },
        ])
        .context("an atomic batch of two creates against an empty namespace was refused")?;
    ensure!(
        at(one)?.as_deref() == Some(repo.commit1.as_slice())
            && at(two)?.as_deref() == Some(repo.commit1.as_slice()),
        "an accepted atomic batch did not apply both of its edits"
    );

    // One good edit, one stale. Nothing may land.
    let err = b
        .git
        .put_refs_cas(&[
            RefCas {
                name: one.to_owned(),
                old: Some(repo.commit1.as_slice()),
                new: Some(repo.commit2.as_slice()),
                peeled: None,
            },
            RefCas {
                name: two.to_owned(),
                old: Some(repo.commit2.as_slice()),
                new: Some(repo.commit2.as_slice()),
                peeled: None,
            },
        ])
        .err()
        .context("an atomic batch with a stale expectation in it was ACCEPTED")?;
    ensure!(
        at(one)?.as_deref() == Some(repo.commit1.as_slice())
            && at(two)?.as_deref() == Some(repo.commit1.as_slice()),
        "a refused atomic batch applied part of itself, which is the one thing --atomic \
         promises never happens"
    );
    let rejection = RefRejection::of(&err).with_context(|| {
        format!(
            "the rejection carries no typed RefRejection, so receive-pack can only learn which \
             ref lost by reading the message `{err:#}`"
        )
    })?;
    ensure!(
        rejection.is_cas_failure() && rejection.name() == two,
        "the rejection named {:?} (cas_failure={}); the edit that lost was {two}",
        rejection.name(),
        rejection.is_cas_failure()
    );

    // A create against a reference that already holds THE VALUE BEING
    // PROPOSED.
    let err = b
        .git
        .put_refs_cas(&[RefCas {
            name: one.to_owned(),
            old: None,
            new: Some(repo.commit1.as_slice()),
            peeled: None,
        }])
        .err()
        .context(
            "a `must not exist` create was accepted against a reference that already exists and \
             already holds the proposed value; the expectation is being short-circuited rather \
             than evaluated, so two racing pushes would both be told they won",
        )?;
    ensure!(
        RefRejection::of(&err).is_some_and(|r| r.is_cas_failure()),
        "the same-value refusal is not a typed compare-and-swap rejection: {err:#}"
    );

    // And the batch that must succeed, so none of the above passes by refusing
    // everything.
    b.git
        .put_refs_cas(&[
            RefCas {
                name: one.to_owned(),
                old: Some(repo.commit1.as_slice()),
                new: Some(repo.commit2.as_slice()),
                peeled: None,
            },
            RefCas {
                name: two.to_owned(),
                old: Some(repo.commit1.as_slice()),
                new: None,
                peeled: None,
            },
        ])
        .context("an atomic batch whose every expectation held was refused")?;
    ensure!(
        at(one)?.as_deref() == Some(repo.commit2.as_slice()) && at(two)?.is_none(),
        "the final atomic batch did not apply its update and its delete"
    );
    Ok(())
}

/// **`HEAD` is not a row in `refs()`, and it has its own accessor.** The `GitServe` HEAD ruling.
///
/// This check exists because the suite did **not** have it: the two arms
/// disagreed about whether `HEAD` is a row for as long as there were two arms,
/// and nothing here could see it, because nothing here ever exercised `HEAD`.
/// That is precisely the class of defect a conformance suite exists for — a
/// difference between backends that no single backend's own tests can notice.
///
/// The reason it cannot be a row is a type constraint rather than taste: a name
/// type that admits `HEAD` also admits `MERGE_HEAD` and `FETCH_HEAD`, so a row
/// stream carrying pseudo-refs means widening the name type or filtering at
/// every consumer.
fn head_is_not_a_row_and_has_its_own_accessor<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "head", hash_kind)?;
    ensure!(
        b.git.head()?.is_none(),
        "a repository that has never pointed HEAD answered one; `None` is what an empty \
         repository has"
    );

    b.git
        .put_refs(&[RefUpdate::set("refs/heads/main", hex(&repo.commit1))])?;
    b.git
        .set_head("refs/heads/main")
        .context("set_head refused a symbolic target under refs/")?;

    let rows = b.git.refs()?;
    ensure!(
        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<_>>()
    );
    let names: Vec<&str> = rows.iter().map(|r| r.name.as_str()).collect();
    ensure!(
        names == vec!["refs/heads/main"],
        "filtering HEAD out of the row stream must not have filtered anything else: {names:?}"
    );

    let head = b
        .git
        .head()?
        .context("HEAD was just pointed and head() cannot find it")?;
    ensure!(
        head.name == HEAD,
        "head() answered a row named {:?}",
        head.name
    );
    ensure!(
        head.symref_target.as_deref() == Some("refs/heads/main"),
        "HEAD came back pointing at {:?}",
        head.symref_target
    );
    ensure!(
        head.oid.is_none(),
        "a symbolic HEAD has no direct target of its own, and this one reports {:?}",
        head.oid
    );

    // It moves, and it is still not a row.
    b.git
        .put_refs(&[RefUpdate::set("refs/heads/other", hex(&repo.commit2))])?;
    b.git.set_head("refs/heads/other")?;
    ensure!(
        b.git.head()?.and_then(|r| r.symref_target).as_deref() == Some("refs/heads/other"),
        "set_head did not move HEAD"
    );
    ensure!(
        b.git.refs()?.iter().all(|r| r.name != HEAD),
        "HEAD appeared in the ref stream once it had been moved"
    );

    // ── `set_head_projected` answers the same question ──────────────────────
    //
    // The 2026-08-30 addition, and the reason it is checked here rather than in
    // one arm's own tests: it exists so that a backend MAY drop the barrier for
    // a HEAD whose authority the caller keeps elsewhere, and a backend that
    // dropped the barrier and the write with it would look identical to its own
    // unit tests and to the caller — the call returns `Ok` either way. So the
    // check is on applied output: the projection is visible through `head()`
    // and is still not a row.
    b.git
        .set_head_projected("refs/heads/main")
        .context("set_head_projected refused a symbolic target under refs/")?;
    ensure!(
        b.git.head()?.and_then(|r| r.symref_target).as_deref() == Some("refs/heads/main"),
        "set_head_projected did not move HEAD; it is set_head without the barrier, not without          the write"
    );
    ensure!(
        b.git.refs()?.iter().all(|r| r.name != HEAD),
        "a projected HEAD appeared in the ref stream"
    );
    Ok(())
}

/// **A `TxId` is applied output, not an echo.**
///
/// Its own doc: *"Every field is applied output, read back from what was
/// written"*, `pack_id`/`extent` are `None` *"when the transaction carried no
/// pack"* and `push_seq` `None` *"when the transaction carried no ref update"*.
/// The negative direction is checkable on every backend, and it catches a
/// backend echoing a field it did not write.
///
/// `push_seq` is **permitted to be `None` throughout** — the gix arm keeps no ref
/// log, so it has no durable ordering authority, and that narrowing is recorded
/// as a fact rather than as a trait change. But a
/// backend that reports one must report an *advancing* one, or it is a counter
/// that orders nothing.
fn tx_ids_report_what_happened<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let b = harness.open("tx_ids", hash_kind)?;
    let repo = repo(hash_kind)?;

    let pack_tx = b.git.put_pack(&repo.fixture.bytes)?;
    ensure!(
        pack_tx.push_seq.is_none(),
        "put_pack reported a push_seq ({:?}) for a transaction that carried no ref update",
        pack_tx.push_seq
    );

    let first = b
        .git
        .put_refs(&[RefUpdate::set("refs/heads/main", hex(&repo.commit1))])?;
    ensure!(
        first.pack_id.is_none() && first.extent.is_none(),
        "put_refs reported a pack ({:?}, {:?}) for a transaction that carried none",
        first.pack_id,
        first.extent
    );

    let second = b
        .git
        .put_refs(&[RefUpdate::set("refs/heads/other", hex(&repo.commit2))])?;
    match (first.push_seq, second.push_seq) {
        (None, None) => {} // A backend with no ordering authority, said out loud.
        (Some(a), Some(b)) => ensure!(
            b > a,
            "two ref transactions reported push_seq {a} then {b}; a sequence that does not \
             advance orders nothing"
        ),
        (a, b) => bail!(
            "push_seq was {a:?} for one ref transaction and {b:?} for the next; a backend \
             either has an ordering authority or it does not"
        ),
    }
    Ok(())
}

/// **One push: the pack and the refs that point into it.**
///
/// The ordering inside `put` — bytes durable first, refs second — is not
/// observable without a crash, so what is asserted is the outcome it exists to
/// guarantee: after `put` returns, the reference resolves and every object
/// beneath it is there.
fn put_lands_the_pack_and_the_refs_together<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let b = harness.open("put_together", hash_kind)?;
    let repo = repo(hash_kind)?;

    b.git.put(
        &repo.fixture.bytes,
        &[RefUpdate::set("refs/heads/main", hex(&repo.commit2))],
    )?;

    let row = b
        .git
        .refs()?
        .into_iter()
        .find(|r| r.name == "refs/heads/main")
        .context("put() landed the pack and not the reference")?;
    ensure!(
        row.oid.as_deref() == Some(repo.commit2.as_slice()),
        "put() left refs/heads/main pointing somewhere else"
    );
    for id in [repo.commit2, repo.tree2, repo.blob, repo.delta_blob] {
        ensure!(
            b.git.has(id.as_slice())?,
            "put() advertised a reference whose closure it does not hold: {} is missing, \
             which is the one crash ordering no later pass repairs",
            hex(&id)
        );
    }
    Ok(())
}

// ───────────────────────────── graph checks ──────────────────────────────────

fn reachable_returns_the_whole_closure<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "reach_clone", hash_kind)?;
    let got = sorted(b.git.reachable(&[repo.commit2.as_slice()], &[])?);
    let want = as_bytes(&[
        repo.commit2,
        repo.commit1,
        repo.tree2,
        repo.tree1,
        repo.blob,
        repo.delta_blob,
    ]);
    ensure!(
        got == want,
        "the closure of a clone came back with {} objects, not {}. An under-send is the \
         failure no caller can detect: the clone ends short by everything beneath the tip \
         and exits zero",
        got.len(),
        want.len()
    );
    Ok(())
}

fn reachable_subtracts_what_the_client_has<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "reach_incremental", hash_kind)?;
    let got = sorted(
        b.git
            .reachable(&[repo.commit2.as_slice()], &[repo.commit1.as_slice()])?,
    );
    let want = as_bytes(&[repo.commit2, repo.tree2, repo.delta_blob]);
    ensure!(
        got == want,
        "an incremental fetch was answered with {} objects, not {}. The blob and tree the \
         client already has must not be sent again, and everything new must be",
        got.len(),
        want.len()
    );
    // The identity-value guard: a backend that returned the whole closure
    // regardless of `have` passes nothing above, and one that returned nothing
    // would pass a length check against zero.
    ensure!(
        !got.is_empty(),
        "an incremental fetch was answered with nothing, and the second commit is new"
    );
    Ok(())
}

/// A `have` naming a commit this store does not hold contributes nothing.
///
/// A client whose history diverged sends exactly that. The safe direction is to
/// send more, never less: an unknown `have` must not narrow the answer and must
/// not be an error.
fn reachable_tolerates_a_have_it_does_not_know<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "reach_unknown_have", hash_kind)?;
    let stranger = absent_oid(hash_kind, 0xef);
    let got = sorted(
        b.git
            .reachable(&[repo.commit1.as_slice()], &[stranger.as_slice()])
            .context("a `have` this store does not hold was treated as an error")?,
    );
    let want = as_bytes(&[repo.commit1, repo.tree1, repo.blob]);
    ensure!(
        got == want,
        "an unknown `have` changed the answer: {} objects, not {}",
        got.len(),
        want.len()
    );
    Ok(())
}

/// **A gitlink is not followed.**
///
/// A tree entry with git's gitlink mode names a commit in *another* repository,
/// which this one is not expected to hold. Following it would report every
/// repository with a submodule as corrupt — and a backend that instead *sends*
/// the gitlink oid produces a pack the client cannot close.
fn reachable_does_not_follow_a_submodule<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "reach_submodule", hash_kind)?;
    let got = sorted(
        b.git
            .reachable(&[repo.sub_commit.as_slice()], &[])
            .context("a repository with a submodule was reported as broken")?,
    );
    let want = as_bytes(&[repo.sub_commit, repo.sub_tree, repo.blob]);
    ensure!(
        !got.contains(&repo.gitlink.as_slice().to_vec()),
        "the closure contains the gitlink commit {}, which lives in another repository and \
         is not in this store; the resulting pack cannot be closed by any client",
        hex(&repo.gitlink)
    );
    ensure!(
        got == want,
        "the closure of a commit with a submodule came back as {} objects, not {}",
        got.len(),
        want.len()
    );
    Ok(())
}

/// **An annotated tag reaches the commit it names.**
///
/// A `want` naming a tag object must answer with the tag *and* everything under
/// it. The tag is what a `refs/tags/*` tip holds, so a backend that contributes
/// the tag's own id and stops builds a pack the receiver cannot close, and says
/// so in the one sentence that names the object nobody sent:
///
/// ```text
///   refs/tags/v0.9.10: missing necessary objects: object 74ee9228… is not in
///   this repository (peeling a tag)
/// ```
///
/// 🔴 MEASURED on the reference appliance 2026-08-31: six of 25 repositories could not
/// obtain a second copy, every one of them refused on an annotated tag, on the
/// engine the appliance ships. `git clone` of the same repositories worked
/// throughout, because the serve path peels a tip before it asks the store and
/// so never puts this question to a backend at all. The store's own contract is
/// where the answer belongs, and this is the check that asks it.
///
/// The identity-value guard is the length: a backend returning the tag alone
/// gives one object, and one is not four.
///
/// # What this check does **not** claim
///
/// The engine that failed on 2026-08-31 does not run this suite — it lives in
/// another repository and consumes only the contract — so this going green here
/// is the in-memory control conforming and nothing more. What protects the two
/// paths that matter is that both now peel before they ask a store: `gunnar
/// clone` always did, and replication does since
/// [`gunnar_wire::objects_to_send`]. This check is what makes the next backend
/// to adopt the suite answer the question rather than inherit the answer.
fn reachable_follows_an_annotated_tag_to_its_target<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "reach_tag", hash_kind)?;
    let got = sorted(b.git.reachable(&[repo.tag.as_slice()], &[])?);
    // `repo.tag` is an annotated tag on `commit1`, whose closure is the commit,
    // its tree and its blob. The tag object itself is in the answer too: it is
    // in no commit's closure — a tree never points at a tag — and the reference
    // being moved names it.
    let want = as_bytes(&[repo.tag, repo.commit1, repo.tree1, repo.blob]);
    ensure!(
        got.contains(&repo.tag.as_slice().to_vec()),
        "the tag object {} is not in its own closure; the reference being pushed names it, so \
         a pack without it moves a ref onto an object the receiver does not have",
        hex(&repo.tag)
    );
    ensure!(
        got.contains(&repo.commit1.as_slice().to_vec()),
        "the closure of the annotated tag {} does not contain the commit {} it names. This is \
         the `is not in this repository (peeling a tag)` refusal at its source: the tag goes \
         into the pack and nothing under it does",
        hex(&repo.tag),
        hex(&repo.commit1)
    );
    ensure!(
        got == want,
        "the closure of an annotated tag came back with {} objects, not {}",
        got.len(),
        want.len()
    );
    Ok(())
}

// ───────────────────────────── the reading half ──────────────────────────────

/// **`read` is the inflated object; `get` is the stored entry.**
///
/// `Stored::bytes` is *"the pack entry, header and all, byte for byte"* —
/// **possibly a delta** — and no method on the storage half returns an inflated
/// object. That gap is why `GitServe` exists at all, so this asserts the two
/// answers are genuinely different where the fixture makes them different, and
/// that the type `read` reports is the **resolved** kind rather than the pack's.
fn read_is_inflated_and_delta_resolved<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "serve_read", hash_kind)?;
    for entry in &repo.fixture.entries {
        let (obj_type, content) = b
            .git
            .read(entry.id.as_slice())?
            .with_context(|| format!("read() found no {}", hex(&entry.id)))?;
        ensure!(
            !matches!(obj_type, ObjType::OfsDelta | ObjType::RefDelta),
            "{} came back from read() typed {}; read() reports the RESOLVED kind, and a caller \
             handed a delta type would parse a delta stream as an object",
            hex(&entry.id),
            obj_type.as_str()
        );
        ensure!(
            obj_type == pack_type_of(entry.kind),
            "{} is a {:?} and read() called it {}",
            hex(&entry.id),
            entry.kind,
            obj_type.as_str()
        );
        ensure!(
            content == entry.content,
            "{} inflated to {} bytes and the fixture wrote {}",
            hex(&entry.id),
            content.len(),
            entry.content.len()
        );
    }

    // The delta entry: the whole reason this method is not `get`.
    let delta = repo
        .fixture
        .entry(&repo.delta_blob)
        .context("the fixture lost its delta entry")?;
    let stored = b
        .git
        .get(repo.delta_blob.as_slice())?
        .context("get() found no delta object")?;
    let (_, inflated) = b
        .git
        .read(repo.delta_blob.as_slice())?
        .context("read() found no delta object")?;
    ensure!(
        stored.bytes != inflated,
        "get() and read() returned the same bytes for an entry stored as {}; one of them is \
         not doing its job",
        delta.obj_type.as_str()
    );

    ensure!(
        b.git
            .read(absent_oid(hash_kind, 0xab).as_slice())?
            .is_none(),
        "read() invented an object that is not there"
    );
    Ok(())
}

/// **`header` and `sizes` answer what `read` answers, without the payload.**
///
/// `sizes` is positional and `None` at a position means *unknown — go ask
/// `header`*, never *fine*. A caller treating an unknown size as under a
/// ceiling would skip the ceiling for exactly the objects a push had just
/// introduced, so the absent case is asserted in the **middle** of the batch,
/// where a backend that compacts its answer shifts every later one onto the
/// wrong oid.
fn header_and_sizes_agree_with_read<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "serve_header", hash_kind)?;
    for entry in &repo.fixture.entries {
        let (obj_type, size) = b
            .git
            .header(entry.id.as_slice())?
            .with_context(|| format!("header() found no {}", hex(&entry.id)))?;
        ensure!(
            obj_type == pack_type_of(entry.kind) && size == entry.content.len() as u64,
            "header() says {} is a {} of {size} bytes; it is a {:?} of {}",
            hex(&entry.id),
            obj_type.as_str(),
            entry.kind,
            entry.content.len()
        );
    }
    ensure!(
        b.git
            .header(absent_oid(hash_kind, 0xab).as_slice())?
            .is_none(),
        "header() invented an object that is not there"
    );

    let missing = absent_oid(hash_kind, 0xab);
    let query: Vec<&[u8]> = vec![
        repo.commit1.as_slice(),
        missing.as_slice(),
        repo.delta_blob.as_slice(),
    ];
    let sizes = b.git.sizes(&query)?;
    ensure!(
        sizes.len() == query.len(),
        "sizes() answered {} of {} questions",
        sizes.len(),
        query.len()
    );
    ensure!(
        sizes[1].is_none(),
        "sizes() answered for an object that is not there; every later answer in the vector is \
         now attributed to the wrong oid"
    );
    let want_delta = repo
        .fixture
        .entry(&repo.delta_blob)
        .context("the fixture lost its delta entry")?
        .content
        .len() as u64;
    ensure!(
        sizes[2] == Some(want_delta),
        "sizes() reports {:?} for a delta object that resolves to {want_delta} bytes; bulk and \
         single-object sizes must be the same POST-RESOLUTION number",
        sizes[2]
    );
    Ok(())
}

/// **`emit_pack` emits the set it was given — it does not walk the graph.**
///
/// This is the shape of a wrapper that turns an optimisation off by not
/// mentioning it, and it is the reason the parameter is named
/// `objects` rather than `want`. Upload-pack holds a selection that is already
/// post-filter, post-shallow and post-`include-tag` and is *deliberately not
/// closed*; an engine that treats it as tips adds back exactly what
/// `--filter=blob:none` or `--depth=N` excluded. **The over-sent pack passes
/// `index-pack --strict` and `fsck`, the clone succeeds, and the client simply
/// receives objects it asked not to have** — so no exit code and no format
/// check can see it, and only an assertion on the emitted membership can.
///
/// What is **not** asserted is `stats.objects == objects.len()`: delta-base
/// closure is a format requirement, an `OFS_DELTA` names its base by in-pack
/// distance, and an emitter must add those. The property is that bases are the
/// **only** thing added.
fn emit_pack_emits_the_set_it_was_given<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "serve_emit", hash_kind)?;
    let caps = Caps::modern();

    // One commit, whose closure is five more objects. A graph-closing engine
    // answers this with six.
    let mut out = Vec::new();
    let stats = b
        .git
        .emit_pack(&[repo.commit2.as_slice()], &[], &caps, &mut out)
        .context("emit_pack refused a single commit")?;
    let ids = pack_ids(&out, hash_kind).context("emit_pack produced something unreadable")?;
    ensure!(
        ids == vec![repo.commit2],
        "emit_pack was given ONE object and emitted {}: {:?}. `objects` is the exact set to \
         emit, not a set of tips to close over — an engine that closes over it sends back \
         everything a --filter or a --depth had just excluded, and the client cannot tell",
        ids.len(),
        ids.iter().map(hex).collect::<Vec<_>>()
    );
    ensure!(
        stats.objects == ids.len() as u64,
        "PackStats reports {} objects and the pack carries {}",
        stats.objects,
        ids.len()
    );
    ensure!(
        stats.bytes == out.len() as u64,
        "PackStats reports {} bytes and {} were written",
        stats.bytes,
        out.len()
    );
    ensure!(
        stats.copied + stats.recompressed == stats.objects,
        "PackStats accounts for {} of {} entries as copied or recompressed; the pair is the \
         receipt that tells a pack copy from a re-deflate, and an entry in neither column \
         makes it unreadable",
        stats.copied + stats.recompressed,
        stats.objects
    );

    // ── the delta, ALONE ────────────────────────────────────────────────────
    //
    // This used to allow the base to come with it — *"that is the format"* — on
    // the reasoning that an `OFS_DELTA` names its base by in-pack distance and
    // so cannot be encoded without it. That is true of the pack format and
    // false of the clone, and allowing it here is what licensed a shipping bug
    // on both arms in turn (the Arrow one 2026-08-11, the gix one 2026-08-14):
    //
    //   * a base pulled in is an object the client never asked for, so a
    //     `--filter=blob:none` or `--depth=N` fetch gets back exactly what it
    //     asked to be left out;
    //   * and if the base is a TREE it arrives owing children the pack does not
    //     contain, so `git index-pack --check-self-contained-and-connected` —
    //     which is what `git clone` runs — refuses the whole transfer with
    //     `did not receive expected object <oid>` while the server logs a
    //     success. Any store that has ever refused a ref update holds
    //     unreachable objects, and `pack-objects` routinely stores a still-
    //     reachable object as a delta against one that is not, so this is an
    //     ordinary repository rather than an exotic one.
    //
    // The rule is stock `pack-objects`': reuse a stored delta only when its
    // base is also being packed. The base decides how an entry is ENCODED, and
    // never what the pack CONTAINS — an engine whose request cuts a delta chain
    // rebuilds the object whole and says so in `recompressed`.
    let mut out = Vec::new();
    let stats = b
        .git
        .emit_pack(&[repo.delta_blob.as_slice()], &[], &caps, &mut out)
        .context("emit_pack refused an object the fixture stores as a delta")?;
    let ids = pack_ids(&out, hash_kind)?;
    ensure!(
        ids == vec![repo.delta_blob],
        "emit_pack was asked for exactly one object — {}, which this fixture stores as a delta \
         against {} — and emitted {}: {:?}. Nothing may be added here, not by reachability and \
         not by the delta format: an out-of-set base is an object the client did not ask for, \
         and a tree one drags in children the pack does not carry, which is a clone stock git \
         refuses while the server reports success",
        hex(&repo.delta_blob),
        hex(&repo.blob),
        ids.len(),
        ids.iter().map(hex).collect::<Vec<_>>()
    );
    ensure!(
        stats.objects == ids.len() as u64,
        "PackStats reports {} objects and the pack carries {}",
        stats.objects,
        ids.len()
    );
    ensure!(
        stats.copied + stats.recompressed == stats.objects,
        "PackStats accounts for {} of {} entries as copied or recompressed; an entry whose base \
         was left behind had to be rebuilt, and a receipt that does not say so is not a receipt",
        stats.copied + stats.recompressed,
        stats.objects
    );

    // The empty emission is a pack with no entries, not an error: a fetch that
    // negotiated everything away still writes one.
    let mut out = Vec::new();
    let stats = b.git.emit_pack(&[], &[], &caps, &mut out)?;
    ensure!(
        stats.objects == 0 && pack_ids(&out, hash_kind)?.is_empty(),
        "an emission of nothing produced {} objects",
        stats.objects
    );
    Ok(())
}

/// **`select` agrees with the graph when it answers, and declines otherwise.**
///
/// `Ok(None)` is the escape hatch and it is load-bearing: the Arrow arm returns
/// it when its reachability projection is empty, which a clean restart
/// produces. So this never *requires* an answer — but it does require that an
/// answer, when given, is the same closure [`GitOps::reachable`] reports, and it
/// requires that a tip the backend does not hold produces **no** answer.
///
/// That last one is the `UnderSend` failure in its serving clothes: a backend
/// whose index does not cover a tip contributes only the tip, the pack is short
/// by everything beneath it, and the clone **exits zero**.
fn select_agrees_with_the_graph_or_declines<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "serve_select", hash_kind)?;

    if let Some(set) = b.git.select(&[repo.commit2.as_slice()], &[])? {
        let want = sorted(b.git.reachable(&[repo.commit2.as_slice()], &[])?);
        ensure!(
            sorted(set.objects.clone()) == want,
            "select() answered {} objects for a clone and reachable() says {}; two answers to \
             one question is worse than one slow answer",
            set.objects.len(),
            want.len()
        );
        ensure!(
            set.commits.iter().all(|id| set.objects.contains(id)),
            "select() reported a commit that is not in the set it is a commit of"
        );
        let commits = sorted(set.commits.clone());
        ensure!(
            commits == as_bytes(&[repo.commit1, repo.commit2]),
            "select() called {} of the objects commits; the closure of the second commit \
             contains exactly two",
            commits.len()
        );
        ensure!(
            set.client_has.is_empty(),
            "select() vouched {} objects as already held by a client that said it had none",
            set.client_has.len()
        );
    }

    // An incremental fetch, if it answers at all.
    if let Some(set) = b
        .git
        .select(&[repo.commit2.as_slice()], &[repo.commit1.as_slice()])?
    {
        let want = sorted(
            b.git
                .reachable(&[repo.commit2.as_slice()], &[repo.commit1.as_slice()])?,
        );
        ensure!(
            sorted(set.objects.clone()) == want,
            "select() and reachable() disagree about an incremental fetch: {} against {}",
            set.objects.len(),
            want.len()
        );
        ensure!(
            !set.client_has.contains(repo.commit2.as_slice()),
            "select() vouched an object THIS transfer carries as one the client already had; a \
             thin delta against it names a base the client does not have"
        );
    }

    // The tip it does not hold. `Ok(None)` or an error — never a set.
    let stranger = absent_oid(hash_kind, 0xef);
    if let Ok(Some(set)) = b.git.select(&[stranger.as_slice()], &[]) {
        bail!(
            "select() answered {} objects for a tip this store does not hold ({}). The escape \
             hatch exists for exactly this: an engine that always answers has to answer wrongly \
             somewhere, and a clone short by everything beneath a tip still exits zero",
            set.objects.len(),
            hex(&stranger)
        );
    }
    Ok(())
}

/// The [`ObjType`] a resolved git kind is reported under — never a delta code.
fn pack_type_of(kind: Kind) -> ObjType {
    match kind {
        Kind::Commit => ObjType::Commit,
        Kind::Tree => ObjType::Tree,
        Kind::Blob => ObjType::Blob,
        Kind::Tag => ObjType::Tag,
    }
}

/// Every object id in an emitted pack, deltas resolved.
///
/// The suite's own reader ([`pack::parse`]), which is honest about what it is:
/// it and [`pack::PackBuilder`] were written together, so a backend's pack being
/// readable by it proves membership rather than format. **The format oracle is
/// `gunnar_pack::index_pack`**, driven over an emitted pack by a dev-dependency
/// test in `tests/conformance.rs`, because it has no stake in either.
fn pack_ids(bytes: &[u8], hash_kind: gix_hash::Kind) -> anyhow::Result<Vec<ObjectId>> {
    let raw = pack::parse(bytes, hash_kind)?;
    let mut by_offset: std::collections::BTreeMap<u64, (Kind, Vec<u8>)> = Default::default();
    let mut ids = Vec::with_capacity(raw.len());
    for entry in &raw {
        let (kind, content) = match entry.base_offset {
            None => (
                pack::kind_of(entry.obj_type).context("a base entry with a delta type")?,
                entry.payload.clone(),
            ),
            Some(offset) => {
                let (kind, base) = by_offset.get(&offset).with_context(|| {
                    format!(
                        "the emitted pack has a delta at {} whose base at {offset} it does not \
                         carry, so no client can close it",
                        entry.extent.0
                    )
                })?;
                (*kind, pack::apply_delta(base, &entry.payload)?)
            }
        };
        by_offset.insert(entry.extent.0, (kind, content.clone()));
        ids.push(
            gix_object::compute_hash(hash_kind, kind, &content).map_err(gix_error::Error::from)?,
        );
    }
    Ok(ids)
}

// ───────────────────────────── maintenance ───────────────────────────────────

/// **`gc` keeps everything a reference points at.**
///
/// Only that direction. A backend that keeps a dead object is wasteful; a
/// backend that drops a live one has lost a repository, and no later pass
/// repairs it. Asserting removal would pin an optimisation as a contract and
/// would make an in-place arm that defers compaction non-conformant for being
/// slower rather than for being wrong.
fn gc_keeps_everything_a_ref_points_at<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let (b, repo) = loaded(harness, "gc", hash_kind)?;
    b.git
        .put_refs(&[RefUpdate::set("refs/heads/main", hex(&repo.commit2))])?;

    let report = b.git.gc().context("gc failed")?;
    ensure!(
        !report.strategy.is_empty(),
        "a GcReport with no strategy name cannot be attributed to an implementation"
    );

    for id in [
        repo.commit2,
        repo.commit1,
        repo.tree2,
        repo.tree1,
        repo.blob,
        repo.delta_blob,
    ] {
        ensure!(
            b.git.has(id.as_slice())?,
            "gc dropped {}, which refs/heads/main still reaches. A repack that loses one tree \
             deep inside history leaves every reference resolvable and every clone broken",
            hex(&id)
        );
    }
    let rows: Vec<String> = b.git.refs()?.into_iter().map(|r| r.name).collect();
    ensure!(
        rows == vec!["refs/heads/main".to_string()],
        "gc changed the ref namespace: {rows:?}"
    );
    // Twice, because an implementation that is only correct on a store it has
    // not already compacted is correct once.
    b.git.gc().context("a second gc failed")?;
    ensure!(
        b.git.has(repo.blob.as_slice())?,
        "a second gc dropped an object the first one kept"
    );
    Ok(())
}

// ───────────────────────────── restart ───────────────────────────────────────

fn the_store_survives_a_restart<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let slot = "restart";
    let repo = repo(hash_kind)?;
    let stored = {
        let b = harness.open(slot, hash_kind)?;
        b.git.put(
            &repo.fixture.bytes,
            &[RefUpdate::set("refs/heads/main", hex(&repo.commit1))],
        )?;
        b.git
            .get(repo.commit1.as_slice())?
            .context("the object is not there before the restart")?
    };

    // The handle above is dropped, so a backend holding an exclusive file lock
    // is free to take one here.
    let b = harness.open(slot, hash_kind)?;
    let after = b
        .git
        .get(repo.commit1.as_slice())?
        .context("the object did not survive a restart")?;
    ensure!(
        after.bytes == stored.bytes,
        "the object's bytes changed across a restart"
    );
    let row = b
        .git
        .refs()?
        .into_iter()
        .find(|r| r.name == "refs/heads/main")
        .context("the reference did not survive a restart")?;
    ensure!(
        row.oid.as_deref() == Some(repo.commit1.as_slice()),
        "the reference moved across a restart"
    );
    Ok(())
}

// ───────────────────────────── secret checks ─────────────────────────────────

fn secret_put_get_delete<H: Harness>(harness: &H, hash_kind: gix_hash::Kind) -> anyhow::Result<()> {
    let b = harness.open("secret_basic", hash_kind)?;
    ensure!(
        b.secrets.get("absent")?.is_none(),
        "get() invented a secret"
    );
    b.secrets.put("k", b"v1")?;
    ensure!(
        b.secrets.get("k")? == Some(b"v1".to_vec()),
        "put/get lost the value"
    );
    b.secrets.put("k", b"v2")?;
    ensure!(
        b.secrets.get("k")? == Some(b"v2".to_vec()),
        "put did not overwrite"
    );
    b.secrets.delete("k")?;
    ensure!(b.secrets.get("k")?.is_none(), "delete did not delete");
    // Deleting what is not there is not an error.
    b.secrets.delete("k")?;
    Ok(())
}

fn secret_get_or_create_survives_restart<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let slot = "secret_restart";
    let calls = AtomicUsize::new(0);
    let mint = || {
        let n = calls.fetch_add(1, Ordering::SeqCst);
        // A fresh "host key" every call, so a store that re-mints is caught by
        // the byte comparison rather than by an honour system.
        Ok(format!("ssh-ed25519-fake-key-{n}").into_bytes())
    };

    let first = {
        let b = harness.open(slot, hash_kind)?;
        b.secrets
            .get_or_create_with(crate::secret::SSH_HOST_KEY, mint)?
    };
    ensure!(
        calls.load(Ordering::SeqCst) == 1,
        "the first get_or_create called mk {} times",
        calls.load(Ordering::SeqCst)
    );

    // Simulated restart: the previous handles are dropped, the store is opened
    // afresh from whatever it persisted.
    let second = {
        let b = harness.open(slot, hash_kind)?;
        b.secrets
            .get_or_create_with(crate::secret::SSH_HOST_KEY, mint)?
    };
    ensure!(
        calls.load(Ordering::SeqCst) == 1,
        "the host key was RE-MINTED after a restart (mk ran {} times), so every client \
         would see a host-key-changed warning",
        calls.load(Ordering::SeqCst)
    );
    ensure!(
        first == second,
        "the host key changed across a restart: {first:?} then {second:?}"
    );

    let third = {
        let b = harness.open(slot, hash_kind)?;
        b.secrets.get(crate::secret::SSH_HOST_KEY)?
    };
    ensure!(
        third.as_deref() == Some(first.as_slice()),
        "get() disagrees with get_or_create after a restart"
    );
    Ok(())
}

fn secret_get_or_create_is_atomic<H: Harness>(
    harness: &H,
    hash_kind: gix_hash::Kind,
) -> anyhow::Result<()> {
    let b = harness.open("secret_race", hash_kind)?;
    let calls = AtomicUsize::new(0);
    const THREADS: usize = 8;

    // `std::thread::scope`, no pool and no runtime, and pure Rust. This is the one
    // fan-out in the crate and it is a race fixture, not a workload.
    let results = std::thread::scope(|scope| {
        let handles: Vec<_> = (0..THREADS)
            .map(|_| {
                let secrets = &b.secrets;
                let calls = &calls;
                scope.spawn(move || {
                    secrets.get_or_create_with(crate::secret::SSH_HOST_KEY, || {
                        let n = calls.fetch_add(1, Ordering::SeqCst);
                        // Sleeping widens the window a broken implementation
                        // would have to fall into.
                        std::thread::sleep(std::time::Duration::from_millis(5));
                        Ok(format!("host-key-{n}").into_bytes())
                    })
                })
            })
            .collect();
        handles
            .into_iter()
            .map(|h| h.join().expect("a get_or_create thread panicked"))
            .collect::<Vec<_>>()
    });

    let values: Vec<Vec<u8>> = results.into_iter().collect::<crate::Result<Vec<_>>>()?;
    ensure!(
        calls.load(Ordering::SeqCst) == 1,
        "{THREADS} concurrent get_or_create calls minted {} keys; the host key must \
         never be regenerated by a race",
        calls.load(Ordering::SeqCst)
    );
    let first = &values[0];
    ensure!(
        values.iter().all(|v| v == first),
        "concurrent get_or_create callers got different bytes back"
    );
    Ok(())
}

// ───────────────────────────── harnesses ─────────────────────────────────────

pub use harness::*;

mod harness {
    use super::{Backend, Harness};
    use crate::mem::{MemGitServe, MemSecretStore};
    use std::collections::HashMap;
    use std::sync::Mutex;

    /// Runs the suite against the in-memory control.
    ///
    /// A "restart" here reopens the same `Arc`-shared state rather than crossing
    /// a process boundary, an honest simulation for a store whose whole point
    /// is that it has no persistence. The real restart evidence comes from the
    /// Arrow-log harness, which closes and reopens actual files, and from the
    /// two real backends running this suite in their own repositories.
    #[derive(Default)]
    pub struct MemHarness {
        git: Mutex<HashMap<(String, gix_hash::Kind), MemGitServe>>,
        secrets: Mutex<HashMap<(String, gix_hash::Kind), MemSecretStore>>,
    }

    impl MemHarness {
        fn git_for(&self, slot: &str, hash_kind: gix_hash::Kind) -> MemGitServe {
            self.git
                .lock()
                .expect("harness lock")
                .entry((slot.to_owned(), hash_kind))
                .or_insert_with(|| MemGitServe::new(hash_kind))
                .clone()
        }

        fn secrets_for(&self, slot: &str, hash_kind: gix_hash::Kind) -> MemSecretStore {
            self.secrets
                .lock()
                .expect("harness lock")
                .entry((slot.to_owned(), hash_kind))
                .or_default()
                .clone()
        }
    }

    impl Harness for MemHarness {
        type Git = MemGitServe;
        type Secrets = MemSecretStore;

        fn name(&self) -> &str {
            "mem"
        }

        fn open(
            &self,
            slot: &str,
            hash_kind: gix_hash::Kind,
        ) -> anyhow::Result<Backend<Self::Git, Self::Secrets>> {
            Ok(Backend {
                git: self.git_for(slot, hash_kind),
                secrets: self.secrets_for(slot, hash_kind),
            })
        }
    }

    // `ArrowLogHarness` and `ArrowLogInMemoryHarness` used to be here, and they
    // are not gone: they moved to `gunnar-core` with `arrow_log` itself
    // (`crates/gunnar-core/tests/arrow_log_conformance.rs`), and they still run
    // this suite, both hash kinds, green and red. They cannot live here any
    // more because this crate no longer links Arrow — see `lib.rs`, and the
    // reason is that both real backends dev-depend on this crate from other
    // repositories.
    //
    // What that costs THIS crate's own test run is the only durable harness it
    // had: `MemHarness`'s "restart" reopens `Arc`-shared state rather than
    // replaying bytes. The replay evidence still exists, one crate over, and
    // the two real backends supply the rest in their own repositories.
}
#[cfg(test)]
mod red;

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

    #[test]
    fn the_control_conforms() {
        let ran = assert_all_pass(&run_all(&MemHarness::default()));
        assert_eq!(
            ran,
            check_names::<MemHarness>().len() * HASH_KINDS.len(),
            "the runner did not run every check against every hash kind"
        );
    }
}