alkgit/tests/doubles.rs

//! The wave-1 contract rows for the backend doubles. Row/stamp
//! discipline: `tests/common` §The row/stamp discipline — every row
//! names its property, and each `Contract stamp:` cites the pinning
//! spec §/ADR §. The doubles themselves live in
//! `tests/common/` (the recorded location decision — no public surface).

mod common;

use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::io;
use std::sync::Arc;

use common::{
    blob_fixture, commit_fixture, decode_pack, delta_stream, hostile_delta_stream, pack_bytes,
    pack_entry_base, pack_entry_delta, tag_fixture, tree_fixture, MemFind, MemObjects, MemPackGen,
    MemPackIngest, MemRefs, MemRegistry, SharedSink, ATOMIC_FAILURE_REASON, STALE_REASON,
};
use gix_hash::ObjectId;

use alkgit::{
    AccessAction, GitPackGen, GitPackIngest, GitRefs, GitRegistry, GitRegistryStore, Limits,
    RefOutcome, RefUpdate, RegistryError, RepoRecord, StorageError, Visibility,
};

fn repo(id: &str) -> RepoRecord {
    RepoRecord {
        repo_id: id.to_string(),
        visibility: Visibility::Private,
        grants: BTreeMap::new(),
    }
}

fn oid(b: u8) -> ObjectId {
    ObjectId::from([b; 20])
}

/// Seeds `(oid, loose bytes)` rows into `repo`'s namespace (test
/// support for the fixture builders).
fn seed(objects: &MemObjects, repo: &RepoRecord, rows: &[(ObjectId, Vec<u8>)]) {
    for (id, bytes) in rows {
        objects.put(repo, *id, bytes.clone());
    }
}

/// The object-level closure of `seeds` over the seeded repo, walked
/// independently of the pack pipeline (the hand-rolled cross-check
/// POC-2 established): commits queue their tree and parents, trees
/// queue their entries, tags queue their targets, blobs are leaves.
fn object_closure(
    objects: &MemObjects,
    repo: &RepoRecord,
    seeds: &[ObjectId],
) -> BTreeSet<ObjectId> {
    let mut closure = BTreeSet::new();
    let mut queue: VecDeque<ObjectId> = seeds.iter().copied().collect();
    while let Some(oid) = queue.pop_front() {
        if !closure.insert(oid) {
            continue;
        }
        let bytes = objects
            .find(repo, &oid)
            .unwrap_or_else(|| panic!("closure helper resolves the seeded repo: {oid}"));
        let (kind, _size, header_len) =
            gix_object::decode::loose_header(&bytes).expect("fixture objects decode");
        let body = &bytes[header_len..];
        match kind {
            gix_object::Kind::Blob => {}
            gix_object::Kind::Tree => {
                let tree = gix_object::TreeRef::from_bytes(body, gix_hash::Kind::Sha1)
                    .expect("fixture trees decode");
                for entry in &tree.entries {
                    queue.push_back(entry.oid.into());
                }
            }
            gix_object::Kind::Commit => {
                let commit = gix_object::CommitRef::from_bytes(body, gix_hash::Kind::Sha1)
                    .expect("fixture commits decode");
                queue.push_back(commit.tree());
                for parent in commit.parents() {
                    queue.push_back(parent);
                }
            }
            gix_object::Kind::Tag => {
                let tag = gix_object::TagRef::from_bytes(body, gix_hash::Kind::Sha1)
                    .expect("fixture tags decode");
                queue.push_back(tag.target());
            }
        }
    }
    closure
}

/// An unknown repo id resolves `NotFound` — the unknown ≡ unauthorized
/// double form; the double returns `NotFound` and nothing else.
///
/// Contract stamp: backend.md §Registry types and schemas
/// (`RegistryError`); ADR-008 (no existence oracle).
#[tokio::test]
async fn unknown_repo_resolve_is_not_found() {
    let registry = MemRegistry::default();
    let err = registry.resolve("nobody/nothing").await;
    assert_eq!(err, Err(RegistryError::NotFound));
}

/// A record round-trips: put → resolve byte-equal, update replaces the
/// row (the grant-mutation path), remove drops it, and a duplicate put
/// is `AlreadyExists` on the create side.
///
/// Contract stamp: backend.md §`RepoRecord`; ADR-011 (grants ride the
/// record — update is the grant mutation); ADR-012 §3 (the `git/repo/*`
/// create/update/remove ops are thin wrappers over
/// `Arc<dyn GitRegistryStore>`).
#[tokio::test]
async fn registry_records_round_trip_put_resolve_update_remove() {
    let registry = MemRegistry::default();

    registry.put_repo(&repo("owner/doubled")).await.unwrap();

    let mut record = repo("owner/doubled");
    record.visibility = Visibility::Public;
    record
        .grants
        .insert("u-7731".to_string(), vec![AccessAction::Manage]);
    registry.update_repo(&record).await.unwrap();
    assert_eq!(registry.resolve("owner/doubled").await.unwrap(), record);

    let duplicate = registry.put_repo(&record).await.unwrap_err();
    assert_eq!(duplicate, RegistryError::AlreadyExists);

    registry.remove_repo("owner/doubled").await.unwrap();
    assert_eq!(
        registry.resolve("owner/doubled").await,
        Err(RegistryError::NotFound)
    );

    let absent = registry.remove_repo("owner/doubled").await.unwrap_err();
    assert_eq!(absent, RegistryError::NotFound);
    let absent_update = registry.update_repo(&record).await.unwrap_err();
    assert_eq!(absent_update, RegistryError::NotFound);
}

/// A stale `expected` produces a denied [`RefOutcome`] — a business
/// result, never a `StorageError` — and the row stays untouched; this
/// covers the stale-update and the create-over-existing shape alike.
///
/// Contract stamp: ADR-013 §7 (per-ref CAS — outcomes, not errors);
/// ADR-018 §5 (`RefOutcome`, the server-chosen verbatim `ng` reason);
/// backend.md §Pinned signatures.
#[tokio::test]
async fn stale_expected_cas_is_a_denied_ref_outcome_not_an_error() {
    let refs = MemRefs::default();
    let doubled = repo("owner/doubled");
    let create = RefUpdate {
        name: "refs/heads/main".to_string(),
        expected: None,
        new: Some(oid(1)),
    };
    refs.apply_updates(&doubled, vec![create], false)
        .await
        .unwrap();

    let stale_update = RefUpdate {
        name: "refs/heads/main".to_string(),
        expected: Some(oid(2)),
        new: Some(oid(3)),
    };
    let outcomes = refs
        .apply_updates(&doubled, vec![stale_update.clone()], false)
        .await
        .unwrap();
    assert_eq!(outcomes.len(), 1);
    assert_eq!(outcomes[0].name, "refs/heads/main");
    assert_eq!(outcomes[0].result, Err(STALE_REASON.to_string()));

    let colliding_create = RefUpdate {
        name: "refs/heads/main".to_string(),
        expected: None,
        new: Some(oid(3)),
    };
    let outcomes = refs
        .apply_updates(&doubled, vec![colliding_create], false)
        .await
        .unwrap();
    assert_eq!(outcomes[0].result, Err(STALE_REASON.to_string()));

    let stale_delete = RefUpdate {
        name: "refs/heads/main".to_string(),
        expected: Some(oid(9)),
        new: None,
    };
    let outcomes = refs
        .apply_updates(&doubled, vec![stale_delete], false)
        .await
        .unwrap();
    assert_eq!(outcomes[0].result, Err(STALE_REASON.to_string()));

    let refs_now = refs.list_refs(&doubled).await.unwrap();
    assert_eq!(refs_now.len(), 1);
    assert_eq!(refs_now[0].oid, Some(oid(1)));
}

/// Non-atomic is per-ref: each update applies and reports its own
/// outcome in order — successes commit even when other refs deny.
///
/// Contract stamp: ADR-018 §6 (the `atomic` flag is the call's, not the
/// list's); ADR-013 §7–8 (per-ref results, one outcome per update).
#[tokio::test]
async fn non_atomic_updates_apply_each_and_report_per_ref_outcomes() {
    let refs = MemRefs::default();
    let doubled = repo("owner/doubled");
    let seed = vec![RefUpdate {
        name: "refs/heads/main".to_string(),
        expected: None,
        new: Some(oid(1)),
    }];
    refs.apply_updates(&doubled, seed, false).await.unwrap();

    let updates = vec![
        RefUpdate {
            name: "refs/heads/main".to_string(),
            expected: Some(oid(1)),
            new: Some(oid(2)),
        },
        RefUpdate {
            name: "refs/heads/topic".to_string(),
            expected: Some(oid(3)),
            new: Some(oid(4)),
        },
        RefUpdate {
            name: "refs/stale-away".to_string(),
            expected: None,
            new: Some(oid(5)),
        },
    ];
    let outcomes = refs.apply_updates(&doubled, updates, false).await.unwrap();
    assert_eq!(outcomes.len(), 3);
    assert_eq!(outcomes[0].result, Ok(()));
    assert_eq!(
        outcomes[1],
        RefOutcome {
            name: "refs/heads/topic".to_string(),
            result: Err(STALE_REASON.to_string()),
        }
    );
    assert_eq!(outcomes[2].result, Ok(()));

    let refs_now = refs.list_refs(&doubled).await.unwrap();
    assert_eq!(refs_now.len(), 2);
    assert_eq!(refs_now[0].name, "refs/heads/main");
    assert_eq!(refs_now[0].oid, Some(oid(2)));
    assert_eq!(refs_now[1].name, "refs/stale-away");
    assert_eq!(refs_now[1].oid, Some(oid(5)));
}

/// Atomic is all-or-nothing: any denial aborts the whole transaction —
/// nothing applied, denied refs report their own reasons, every other
/// ref of the transaction reports `atomic push failure`; an all-ok
/// atomic transaction commits.
///
/// Contract stamp: ADR-013 §10 (atomic — `ng <ref> atomic push failure`
/// on every other ref); ADR-019 §2 (one row transaction per
/// `apply_updates` call); ADR-018 §6.
#[tokio::test]
async fn atomic_denial_rolls_back_the_whole_transaction() {
    let refs = MemRefs::default();
    let doubled = repo("owner/doubled");
    let seed = vec![RefUpdate {
        name: "refs/heads/main".to_string(),
        expected: None,
        new: Some(oid(1)),
    }];
    refs.apply_updates(&doubled, seed, false).await.unwrap();

    let updates = vec![
        RefUpdate {
            name: "refs/heads/main".to_string(),
            expected: Some(oid(2)),
            new: Some(oid(3)),
        },
        RefUpdate {
            name: "refs/heads/topic".to_string(),
            expected: None,
            new: Some(oid(4)),
        },
        RefUpdate {
            name: "refs/heads/parked".to_string(),
            expected: None,
            new: Some(oid(5)),
        },
    ];
    let outcomes = refs.apply_updates(&doubled, updates, true).await.unwrap();
    assert_eq!(outcomes.len(), 3);
    assert_eq!(outcomes[0].result, Err(STALE_REASON.to_string()));
    assert_eq!(outcomes[1].result, Err(ATOMIC_FAILURE_REASON.to_string()));
    assert_eq!(outcomes[2].result, Err(ATOMIC_FAILURE_REASON.to_string()));

    let refs_now = refs.list_refs(&doubled).await.unwrap();
    assert_eq!(refs_now.len(), 1);
    assert_eq!(refs_now[0].oid, Some(oid(1)));

    let updates = vec![
        RefUpdate {
            name: "refs/heads/main".to_string(),
            expected: Some(oid(1)),
            new: None,
        },
        RefUpdate {
            name: "refs/heads/topic".to_string(),
            expected: None,
            new: Some(oid(4)),
        },
    ];
    let outcomes = refs.apply_updates(&doubled, updates, true).await.unwrap();
    assert!(outcomes.iter().all(|outcome| outcome.result.is_ok()));

    let refs_now = refs.list_refs(&doubled).await.unwrap();
    assert_eq!(refs_now.len(), 1);
    assert_eq!(refs_now[0].name, "refs/heads/topic");
}

/// A have is recognized iff it **exists** in the object store **and
/// is a commit** — decoded from the stored bytes' object type — in the
/// caller's order verbatim, duplicates preserved; unrecognized in a
/// foreign repo's namespace.
///
/// Contract stamp: ADR-014 §5 (`common_haves`, the ack/done-round seam —
/// the recognized predicate is exists + is-commit); backend.md §trait
/// family bullet 4 (the honest boundary: never ack what we cannot
/// subtract); ADR-014 §2 (the done round's boundary set is this same
/// recognized subset); core/w2-negotiation (the contract interpretation:
/// backend.md §Pinned signatures' sentence ("existence is the operative
/// predicate here") read as the §2 subtraction note — flagged for the
/// wave-2 gate to confirm or fix as docs drift; the wire layer adds no
/// refinement, ADR-014 §5).
#[tokio::test]
async fn common_haves_recognizes_a_have_iff_it_exists_and_is_a_commit() {
    let doubled = repo("owner/doubled");
    let cousin = repo("other/doubled");
    let objects = MemObjects::default();
    let (blob, blob_bytes) = blob_fixture(b"leaf");
    let (tree, tree_bytes) = tree_fixture(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob)]);
    let (commit, commit_bytes) = commit_fixture(tree, &[], "a commit");
    seed(
        &objects,
        &doubled,
        &[
            (blob, blob_bytes.clone()),
            (tree, tree_bytes),
            (commit, commit_bytes),
            (oid(1), vec![1, 2, 3]),
        ],
    );

    let gen = MemPackGen {
        objects: objects.clone(),
    };
    let unrecognized = gen.common_haves(&doubled, vec![blob, tree]).await.unwrap();
    assert!(unrecognized.is_empty(), "non-commit objects are not common");
    let malformed = gen.common_haves(&doubled, vec![oid(1)]).await.unwrap();
    assert!(
        malformed.is_empty(),
        "unrecognizable stored bytes fail closed"
    );
    let missing = gen
        .common_haves(&doubled, vec![oid(2), oid(3)])
        .await
        .unwrap();
    assert!(missing.is_empty(), "absent oids are not common");
    let foreign = gen.common_haves(&cousin, vec![commit]).await.unwrap();
    assert!(foreign.is_empty(), "a foreign namespace is not common");

    let common = gen
        .common_haves(&doubled, vec![oid(2), commit, oid(3), commit])
        .await
        .unwrap();
    assert_eq!(
        common,
        vec![commit, commit],
        "order and duplicates verbatim"
    );

    assert_eq!(
        objects.find(&doubled, &blob),
        Some(blob_bytes),
        "the Find-shaped object lookup resolves the stored bytes"
    );
    assert_eq!(objects.find(&cousin, &blob), None);
}

/// The five doubles compile against the pinned trait surface as
/// `Arc<dyn ...>`; the ingestion double consumes a real — here empty —
/// pack stream through the seam (zero objects is valid, `unpack: Ok`),
/// and the generation side aborts on the missing want.
///
/// Contract stamp: ADR-018 §4 (the pinned signatures); backend.md
/// §Pinned signatures; docs/plans/implementation.md §Wave 1 (the test
/// convention); ADR-004 (generate is the wired real pipeline — a
/// missing want aborts with the seam's missing-object error); ADR-013
/// §8 + ADR-020 §2 (the ingestion double is wired — the empty pack
/// ingests with `unpack ok`, the empty tip set walks the empty graph).
#[tokio::test]
async fn doubles_compile_against_the_pinned_trait_surface() {
    let registry: Arc<dyn GitRegistry> = Arc::new(MemRegistry::default());
    let store: Arc<dyn GitRegistryStore> = Arc::new(MemRegistry::default());
    let refs: Arc<dyn GitRefs> = Arc::new(MemRefs::default());
    let gen = MemPackGen::default();
    let gen: Arc<MemPackGen> = Arc::new(gen);
    let ingest_double = MemPackIngest::default();
    let ingest: Arc<dyn GitPackIngest> = Arc::new(ingest_double.clone());

    let doubled = repo("owner/doubled");
    assert_eq!(
        registry.resolve("owner/doubled").await.unwrap_err(),
        RegistryError::NotFound
    );
    store.put_repo(&doubled).await.unwrap();
    assert!(store.update_repo(&doubled).await.is_ok());
    assert!(store.remove_repo("owner/doubled").await.is_ok());
    assert!(refs.list_refs(&doubled).await.unwrap().is_empty());

    let generate = gen
        .generate(
            &doubled,
            vec![oid(1)],
            Vec::new(),
            Limits::default(),
            Box::new(std::io::sink()),
        )
        .await
        .unwrap_err();
    assert_eq!(generate, StorageError::ObjectMissing { oid: oid(1) });
    let empty_pack = pack_bytes(&[]);
    let prepared = ingest
        .prepare(
            &doubled,
            Vec::new(),
            Box::new(io::Cursor::new(empty_pack)),
            None,
            Limits::default(),
        )
        .await
        .unwrap();
    assert!(
        prepared.unpack.is_ok(),
        "the zero-object pack is a valid ingestion: {:?}",
        prepared.unpack
    );
    assert!(prepared.updates.is_empty());
    assert!(!ingest_double.objects().contains(&doubled, &oid(1)));
}

/// The done round's generate composes ADR-004's pipeline over the mem
/// Find: the sink re-decodes as a V2 sha1 pack whose header count,
/// whole-base entries, and verified trailer match, and the object set
/// is exactly closure(wants) — here with an empty boundary, the full
/// closure hand-walked over the seeded repo (commits, tags, and a bare
/// blob want alike).
///
/// Contract stamp: ADR-004 (the generation pipeline, the two-stage
/// closure, V2 sha1 writer, loose objects ship as compressed bases —
/// delta synthesis is the recorded backlog); ADR-014 §2 (the done round
/// generates closure(wants) − closure(haves)); ADR-018 §4 (the pinned
/// `generate` semantics over the seam's `repo`/`wants`/`boundary_haves`);
/// backend.md §Pinned signatures.
#[tokio::test]
async fn generate_produces_the_full_wants_closure_as_a_valid_pack() {
    let doubled = repo("owner/doubled");
    let (blob1, blob1_bytes) = blob_fixture(b"one");
    let (blob2, blob2_bytes) = blob_fixture(b"two");
    let (blob_lone, blob_lone_bytes) = blob_fixture(b"lone");
    let (tree1, tree1_bytes) = tree_fixture(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob1)]);
    let (tree2, tree2_bytes) = tree_fixture(&[
        (gix_object::tree::EntryKind::Blob, "a.txt", blob1),
        (gix_object::tree::EntryKind::Blob, "b.txt", blob2),
    ]);
    let (commit1, commit1_bytes) = commit_fixture(tree1, &[], "one");
    let (commit2, commit2_bytes) = commit_fixture(tree2, &[commit1], "two");
    let (tag, tag_bytes) = tag_fixture(commit2, gix_object::Kind::Commit, "v1.0");
    let objects = MemObjects::default();
    seed(
        &objects,
        &doubled,
        &[
            (blob1, blob1_bytes),
            (blob2, blob2_bytes),
            (blob_lone, blob_lone_bytes),
            (tree1, tree1_bytes),
            (tree2, tree2_bytes),
            (commit1, commit1_bytes),
            (commit2, commit2_bytes),
            (tag, tag_bytes),
        ],
    );

    let gen = MemPackGen {
        objects: objects.clone(),
    };
    let sink = SharedSink::default();
    gen.generate(
        &doubled,
        vec![commit2, tag, blob_lone],
        Vec::new(),
        Limits::default(),
        Box::new(sink.clone()),
    )
    .await
    .unwrap();

    let pack = decode_pack(&sink.snapshot());
    assert_eq!(pack.version, gix_pack::data::Version::V2);
    let ids: BTreeSet<ObjectId> = pack.objects.iter().map(|(id, _, _)| *id).collect();
    let expected = object_closure(&objects, &doubled, &[commit2, tag, blob_lone]);
    assert_eq!(ids, expected);
    assert_eq!(pack.object_count as usize, pack.objects.len());
    assert_eq!(
        pack.objects.len(),
        pack.objects
            .iter()
            .map(|(id, _, _)| *id)
            .collect::<BTreeSet<_>>()
            .len(),
        "closure members ship once"
    );
    let _ = pack.trailer;
}

/// The pack set is byte-identical across repeated generate calls over
/// the same repo and want set — the deterministic `objects_unthreaded`
/// count order.
///
/// Contract stamp: ADR-004 (deterministic order via
/// `count::objects_unthreaded`; the generation seam is
/// negotiation-agnostic — boundary sets in, pack out).
#[tokio::test]
async fn repeated_generate_calls_regenerate_byte_identical_packs() {
    let doubled = repo("owner/doubled");
    let (blob1, blob1_bytes) = blob_fixture(b"one");
    let (tree1, tree1_bytes) = tree_fixture(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob1)]);
    let (commit1, commit1_bytes) = commit_fixture(tree1, &[], "one");
    let objects = MemObjects::default();
    seed(
        &objects,
        &doubled,
        &[
            (blob1, blob1_bytes),
            (tree1, tree1_bytes),
            (commit1, commit1_bytes),
        ],
    );
    let gen = MemPackGen {
        objects: objects.clone(),
    };

    let first = SharedSink::default();
    gen.generate(
        &doubled,
        vec![commit1],
        Vec::new(),
        Limits::default(),
        Box::new(first.clone()),
    )
    .await
    .unwrap();
    let second = SharedSink::default();
    gen.generate(
        &doubled,
        vec![commit1],
        Vec::new(),
        Limits::default(),
        Box::new(second.clone()),
    )
    .await
    .unwrap();
    assert_eq!(first.snapshot(), second.snapshot());
}

/// Boundary subtraction trims the have's ancestry: a want whose history
/// contains a boundary have generates exactly closure(want) minus
/// closure(have) — the have commit, its own tree, and its ancestry are
/// absent; the surviving tip's own tree closure ships.
///
/// Contract stamp: ADR-014 §2 (the done round's boundary set — the
/// recognized subset bounds the walk; verbatim per the caller's
/// contract); ADR-004 (the generation seam is negotiation-agnostic,
/// boundary sets in → pack out).
#[tokio::test]
async fn generate_subtracts_the_boundary_haves_ancestry() {
    let doubled = repo("owner/doubled");
    let (blob_a, blob_a_bytes) = blob_fixture(b"alpha");
    let (blob_b, blob_b_bytes) = blob_fixture(b"beta");
    let (tree_a, tree_a_bytes) =
        tree_fixture(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob_a)]);
    let (tree_b, tree_b_bytes) =
        tree_fixture(&[(gix_object::tree::EntryKind::Blob, "b.txt", blob_b)]);
    let (commit_a, commit_a_bytes) = commit_fixture(tree_a, &[], "alpha");
    let (commit_b, commit_b_bytes) = commit_fixture(tree_b, &[commit_a], "beta");
    let objects = MemObjects::default();
    seed(
        &objects,
        &doubled,
        &[
            (blob_a, blob_a_bytes),
            (blob_b, blob_b_bytes),
            (tree_a, tree_a_bytes),
            (tree_b, tree_b_bytes),
            (commit_a, commit_a_bytes),
            (commit_b, commit_b_bytes),
        ],
    );
    let gen = MemPackGen {
        objects: objects.clone(),
    };

    let sink = SharedSink::default();
    gen.generate(
        &doubled,
        vec![commit_b],
        vec![commit_a],
        Limits::default(),
        Box::new(sink.clone()),
    )
    .await
    .unwrap();

    let pack = decode_pack(&sink.snapshot());
    let ids: BTreeSet<ObjectId> = pack.objects.iter().map(|(id, _, _)| *id).collect();
    let expected = object_closure(&objects, &doubled, &[commit_b]);
    let haves = object_closure(&objects, &doubled, &[commit_a]);
    let subtracted: BTreeSet<ObjectId> = expected.difference(&haves).copied().collect();
    assert_eq!(ids, subtracted);
    assert!(!ids.contains(&commit_a));
    assert!(!ids.contains(&tree_a));
}

/// An annotated-tag want whose terminal commit is a boundary have ships
/// the tag alone: the peeled chain is counted as-is against the
/// boundary's peels, so the commit the client already has never enters
/// the pack.
///
/// Contract stamp: ADR-014 §2 (the pinned closure subtraction —
/// closure(wants) − closure(haves), the amendment's honest-boundary
/// rule); ADR-004 (peeling composes before the walk).
#[tokio::test]
async fn generate_ships_a_peeled_tag_without_its_common_target_commit() {
    let doubled = repo("owner/doubled");
    let (blob_c, blob_c_bytes) = blob_fixture(b"charlie");
    let (tree_c, tree_c_bytes) =
        tree_fixture(&[(gix_object::tree::EntryKind::Blob, "c.txt", blob_c)]);
    let (commit_c, commit_c_bytes) = commit_fixture(tree_c, &[], "charlie");
    let (tag, tag_bytes) = tag_fixture(commit_c, gix_object::Kind::Commit, "v1");
    let objects = MemObjects::default();
    seed(
        &objects,
        &doubled,
        &[
            (blob_c, blob_c_bytes),
            (tree_c, tree_c_bytes),
            (commit_c, commit_c_bytes),
            (tag, tag_bytes),
        ],
    );
    let gen = MemPackGen {
        objects: objects.clone(),
    };

    let sink = SharedSink::default();
    gen.generate(
        &doubled,
        vec![tag],
        vec![commit_c],
        Limits::default(),
        Box::new(sink.clone()),
    )
    .await
    .unwrap();

    let pack = decode_pack(&sink.snapshot());
    let ids: BTreeSet<ObjectId> = pack.objects.iter().map(|(id, _, _)| *id).collect();
    assert_eq!(ids, BTreeSet::from([tag]));
}

/// An all-boundary want set — the client already has everything —
/// generates a valid zero-object packfile: header (V2, zero count),
/// no entries, verified trailer.
///
/// Contract stamp: ADR-014 §2 (an empty resulting pack is a valid
/// zero-object packfile response); ADR-004 (a valid pack is generated
/// for whatever closure remains).
#[tokio::test]
async fn all_boundary_wants_generate_a_valid_zero_object_pack() {
    let doubled = repo("owner/doubled");
    let (blob_d, blob_d_bytes) = blob_fixture(b"delta");
    let (tree_d, tree_d_bytes) =
        tree_fixture(&[(gix_object::tree::EntryKind::Blob, "d.txt", blob_d)]);
    let (commit_d, commit_d_bytes) = commit_fixture(tree_d, &[], "delta");
    let objects = MemObjects::default();
    seed(
        &objects,
        &doubled,
        &[
            (blob_d, blob_d_bytes),
            (tree_d, tree_d_bytes),
            (commit_d, commit_d_bytes),
        ],
    );
    let gen = MemPackGen {
        objects: objects.clone(),
    };

    let sink = SharedSink::default();
    gen.generate(
        &doubled,
        vec![commit_d],
        vec![commit_d],
        Limits::default(),
        Box::new(sink.clone()),
    )
    .await
    .unwrap();

    let pack = decode_pack(&sink.snapshot());
    assert_eq!(pack.version, gix_pack::data::Version::V2);
    assert_eq!(pack.object_count, 0);
    assert!(pack.objects.is_empty());
}

/// A want missing from the repo's own namespace aborts with
/// `ObjectMissing` and nothing was written to the sink — even when the
/// same object exists in another repo's namespace, the generation is
/// scoped to the seam's repo parameter.
///
/// Contract stamp: ADR-004 (missing objects abort the fetch, never a
/// broken pack); ADR-018 §7 (`ObjectMissing` is the seam variant);
/// backend.md §Pinned signatures (storage keyed on the resolved
/// `repo_id`, the namespace the Find reads).
#[tokio::test]
async fn generate_aborts_on_a_missing_want_from_the_repos_own_namespace() {
    let doubled = repo("owner/doubled");
    let cousin = repo("other/doubled");
    let (blob_x, blob_x_bytes) = blob_fixture(b"x");
    let objects = MemObjects::default();
    seed(&objects, &cousin, &[(blob_x, blob_x_bytes)]);
    let gen = MemPackGen {
        objects: objects.clone(),
    };

    let sink = SharedSink::default();
    let missing = gen
        .generate(
            &doubled,
            vec![blob_x],
            Vec::new(),
            Limits::default(),
            Box::new(sink.clone()),
        )
        .await
        .unwrap_err();
    assert_eq!(missing, StorageError::ObjectMissing { oid: blob_x });
    assert!(sink.snapshot().is_empty());
}

/// A want's commit ancestry missing from the namespace aborts with
/// `ObjectMissing` naming the missing object, before any pack byte —
/// the walk is where the ancestry's completeness is established.
///
/// Contract stamp: ADR-004 (missing objects mid-generation abort, never
/// a broken pack); ADR-018 §7 (`ObjectMissing` carries the oid the
/// caller asked for... the closure member it resolves to here).
#[tokio::test]
async fn missing_commit_ancestry_aborts_with_the_missing_oid() {
    let doubled = repo("owner/doubled");
    let (blob_e, blob_e_bytes) = blob_fixture(b"echo");
    let (tree_e, tree_e_bytes) =
        tree_fixture(&[(gix_object::tree::EntryKind::Blob, "e.txt", blob_e)]);
    let (commit_e, commit_e_bytes) = commit_fixture(tree_e, &[], "echo");
    let missing_parent = oid(9);
    let (commit_f, commit_f_bytes) = commit_fixture(tree_e, &[missing_parent], "foxtrot");
    let objects = MemObjects::default();
    seed(
        &objects,
        &doubled,
        &[
            (blob_e, blob_e_bytes),
            (tree_e, tree_e_bytes),
            (commit_e, commit_e_bytes),
            (commit_f, commit_f_bytes),
        ],
    );
    let gen = MemPackGen {
        objects: objects.clone(),
    };

    let sink = SharedSink::default();
    let missing = gen
        .generate(
            &doubled,
            vec![commit_f],
            Vec::new(),
            Limits::default(),
            Box::new(sink.clone()),
        )
        .await
        .unwrap_err();
    assert_eq!(
        missing,
        StorageError::ObjectMissing {
            oid: missing_parent
        }
    );
    assert!(sink.snapshot().is_empty());
}

/// A missing leaf in the wants' closure — a tree entry the repo's
/// namespace lacks — aborts with `ObjectMissing` before any pack byte:
/// the counted object set is existence-checked before the byte
/// pipeline opens, so no header ever reaches the sink.
///
/// Contract stamp: ADR-004 (missing objects abort — never a broken
/// pack emitted downstream, the entry statistics' `missing_objects` > 0
/// posture); ADR-018 §7.
#[tokio::test]
async fn missing_leaf_closure_object_aborts_before_any_pack_byte() {
    let doubled = repo("owner/doubled");
    let (blob_g, blob_g_bytes) = blob_fixture(b"golf");
    let missing_blob = oid(3);
    let (tree_h, tree_h_bytes) = tree_fixture(&[
        (gix_object::tree::EntryKind::Blob, "g.txt", blob_g),
        (gix_object::tree::EntryKind::Blob, "h.txt", missing_blob),
    ]);
    let (commit_h, commit_h_bytes) = commit_fixture(tree_h, &[], "hotel");
    let objects = MemObjects::default();
    seed(
        &objects,
        &doubled,
        &[
            (blob_g, blob_g_bytes),
            (tree_h, tree_h_bytes),
            (commit_h, commit_h_bytes),
        ],
    );
    let gen = MemPackGen {
        objects: objects.clone(),
    };

    let sink = SharedSink::default();
    let missing = gen
        .generate(
            &doubled,
            vec![commit_h],
            Vec::new(),
            Limits::default(),
            Box::new(sink.clone()),
        )
        .await
        .unwrap_err();
    assert_eq!(missing, StorageError::ObjectMissing { oid: missing_blob });
    assert!(sink.snapshot().is_empty());
}

/// The unborn-HEAD fixture row: a repo with an unborn HEAD lists the
/// `HEAD` symref row (`oid: None`, the symref target carried) alongside
/// its refs — and clearing the row removes it.
///
/// Contract stamp: review 001 N-5 (the unborn-HEAD fixture — a `HEAD`
/// symref row with no oid; serving validation is OQ-19 rider (a), wave
/// 5); ADR-003 (`ls-refs=unborn` advertised, the unborn line shape the
/// wire maps to — the literal `unborn` emission, rider-1-captured);
/// backend.md §Pinned signatures (`list_refs` carries the shape).
#[tokio::test]
async fn unborn_head_symref_row_appears_in_list_refs_and_clears() {
    let refs = MemRefs::default();
    let doubled = repo("owner/doubled");

    refs.set_head_symref(&doubled, Some("refs/heads/main".to_string()));
    let listed = refs.list_refs(&doubled).await.unwrap();
    assert_eq!(listed.len(), 1);
    assert_eq!(listed[0].name, "HEAD");
    assert_eq!(listed[0].oid, None);
    assert_eq!(listed[0].symref_target.as_deref(), Some("refs/heads/main"));

    refs.apply_updates(
        &doubled,
        vec![RefUpdate {
            name: "refs/heads/topic".to_string(),
            expected: None,
            new: Some(oid(6)),
        }],
        false,
    )
    .await
    .unwrap();
    let listed = refs.list_refs(&doubled).await.unwrap();
    assert_eq!(listed.len(), 2);
    assert_eq!(listed[0].name, "HEAD", "the HEAD row lists first");
    assert_eq!(listed[1].name, "refs/heads/topic");

    refs.set_head_symref(&doubled, None);
    let listed = refs.list_refs(&doubled).await.unwrap();
    assert_eq!(listed.len(), 1);
    assert_eq!(listed[0].name, "refs/heads/topic");
}

/// The Find front the pipeline consumes is the doubles' MemFind —
/// the `gix_object::Find`/`Exists` and `gix_pack::Find` impls decode the
/// stored loose-object bytes back into the objects the pipeline walks.
///
/// Contract stamp: ADR-004 (the pipeline composes against any
/// `gix_pack::Find` implementor); ADR-019 §6 (the Find-front shape —
/// loose decoding in the double's form).
#[test]
fn memfind_decodes_the_stored_loose_object_encoding() {
    let doubled = repo("owner/doubled");
    let content = b"a blob with body";
    let (blob, blob_bytes) = blob_fixture(content);
    let (tree, tree_bytes) = tree_fixture(&[(gix_object::tree::EntryKind::Blob, "x.txt", blob)]);
    let objects = MemObjects::default();
    seed(
        &objects,
        &doubled,
        &[(blob, blob_bytes), (tree, tree_bytes)],
    );
    let find = MemFind::new(&objects, &doubled);

    assert!(gix_object::Exists::exists(&find, &blob));
    assert!(gix_object::Exists::exists(&find, &tree));
    assert!(!gix_object::Exists::exists(&find, &oid(7)));

    let mut buffer = Vec::new();
    let data = gix_object::Find::try_find(&find, &blob, &mut buffer)
        .expect("the blob lookup succeeds")
        .expect("the blob resolves");
    assert_eq!(data.kind, gix_object::Kind::Blob);
    assert_eq!(data.data, content);
}

/// The object body of a fixture's loose bytes (the pack entries carry
/// bodies, the fixtures return loose encodings).
fn loose_body(bytes: &[u8]) -> (gix_object::Kind, Vec<u8>) {
    let (kind, _size, header_len) =
        gix_object::decode::loose_header(bytes).expect("fixture loose bytes decode");
    (kind, bytes[header_len..].to_vec())
}

/// A delta-product blob: the base body with a literal suffix the
/// client "applied".
fn delta_product(base_body: &[u8], suffix: &[u8]) -> (ObjectId, Vec<u8>) {
    let mut body = base_body.to_vec();
    body.extend_from_slice(suffix);
    let id = gix_object::compute_hash(gix_hash::Kind::Sha1, gix_object::Kind::Blob, &body)
        .expect("the delta product hashes");
    (id, body)
}

/// A self-contained pack ingests: every object of the pack lands in
/// the fresh ingestion namespace as a loose row keyed by its oid
/// (re-derived from the expanded body), the namespace's closure from
/// the pushed tips is exactly the pack's object set, `unpack: Ok(())`,
/// and the updates round-trip verbatim.
///
/// Contract stamp: ADR-020 §2 (the ingestion composition the double
/// mirrors — unpack-to-loose into the repo's own namespace, no
/// working area); ADR-018 §5 (`PreparedPush::updates` round-trips the
/// input verbatim); ADR-013 §5 (self-contained packs index
/// standalone); ADR-013 §8 (a clean ingestion reports `unpack ok` and
/// the session continues).
#[tokio::test]
async fn self_contained_pack_ingests_as_loose_rows_keyed_by_oid() {
    let source = repo("owner/source");
    let doubled = repo("owner/doubled");
    let (blob1, blob1_bytes) = blob_fixture(b"one");
    let (blob2, blob2_bytes) = blob_fixture(b"two");
    let (tree1, tree1_bytes) = tree_fixture(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob1)]);
    let (tree2, tree2_bytes) = tree_fixture(&[
        (gix_object::tree::EntryKind::Blob, "a.txt", blob1),
        (gix_object::tree::EntryKind::Blob, "b.txt", blob2),
    ]);
    let (commit1, commit1_bytes) = commit_fixture(tree1, &[], "one");
    let (commit2, commit2_bytes) = commit_fixture(tree2, &[commit1], "two");
    let (tag, tag_bytes) = tag_fixture(commit2, gix_object::Kind::Commit, "v1.0");
    let objects = MemObjects::default();
    seed(
        &objects,
        &source,
        &[
            (blob1, blob1_bytes),
            (blob2, blob2_bytes),
            (tree1, tree1_bytes),
            (tree2, tree2_bytes),
            (commit1, commit1_bytes),
            (commit2, commit2_bytes),
            (tag, tag_bytes),
        ],
    );
    let gen = MemPackGen {
        objects: objects.clone(),
    };
    let sink = SharedSink::default();
    gen.generate(
        &source,
        vec![commit2, tag],
        Vec::new(),
        Limits::default(),
        Box::new(sink.clone()),
    )
    .await
    .unwrap();
    let pack = sink.snapshot();

    let ingest = MemPackIngest::default();
    let updates = vec![
        RefUpdate {
            name: "refs/heads/main".to_string(),
            expected: None,
            new: Some(commit2),
        },
        RefUpdate {
            name: "refs/tags/v1".to_string(),
            expected: None,
            new: Some(tag),
        },
    ];
    let prepared = ingest
        .prepare(
            &doubled,
            updates.clone(),
            Box::new(io::Cursor::new(pack.clone())),
            None,
            Limits::default(),
        )
        .await
        .unwrap();
    assert!(
        prepared.unpack.is_ok(),
        "the self-contained pack ingests: {:?}",
        prepared.unpack
    );
    assert_eq!(prepared.updates, updates, "the updates round-trip verbatim");

    let decoded = decode_pack(&pack);
    let landed: BTreeSet<ObjectId> = decoded
        .objects
        .iter()
        .map(|(id, kind, _)| {
            let stored = ingest
                .objects()
                .find(&doubled, id)
                .unwrap_or_else(|| panic!("the pack object {id} landed"));
            let (stored_kind, _size, header_len) =
                gix_object::decode::loose_header(&stored).expect("the row decodes");
            assert_eq!(&stored_kind, kind, "the loose row keeps the object kind");
            let rehashed =
                gix_object::compute_hash(gix_hash::Kind::Sha1, stored_kind, &stored[header_len..])
                    .expect("the row re-hashes on put");
            assert_eq!(&rehashed, id, "the loose row is keyed by its own oid");
            *id
        })
        .collect();
    assert_eq!(
        landed,
        object_closure(ingest.objects(), &doubled, &[commit2, tag]),
        "the namespace closure from the pushed tips is the pack's object set"
    );
}

/// A thin pack whose delta bases exist in the repo's namespace ingests
/// — the bases resolve from the (MemFind-shaped) server odb, the delta
/// products and the rest of the pack land; a thin pack with a missing
/// base fails into the `unpack` ng leg with a server-chosen reason,
/// nothing lands, `Err` stays unreachable, and the updates still
/// round-trip verbatim on both paths.
///
/// Contract stamp: ADR-013 §5 (thin packs accepted by default, bases
/// resolved from the server odb — a base the server lacks fails
/// ingestion, the honest outcome); ADR-020 §2 (the ingestion
/// composition); ADR-013 §8 (the `unpack ng` leg continues the session
/// to the report); ADR-018 §5 (the verbatim round-trip rule holds on
/// the ng path — the wire layer needs the ref names for the
/// all-refs unpacker-error report); ADR-018 §7 (the reason may name a
/// client-computed oid, no oracle risk).
#[tokio::test]
async fn thin_pack_bases_resolve_from_the_namespace_and_missing_bases_fail_into_unpack_ng() {
    let doubled = repo("owner/doubled");
    let (blob_server, blob_server_bytes) = blob_fixture(b"server-base");
    let (tree_server, tree_server_bytes) =
        tree_fixture(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob_server)]);
    let (commit_server, commit_server_bytes) = commit_fixture(tree_server, &[], "base");
    let objects = MemObjects::default();
    let ingest = MemPackIngest {
        objects: objects.clone(),
    };
    seed(
        &objects,
        &doubled,
        &[
            (blob_server, blob_server_bytes.clone()),
            (tree_server, tree_server_bytes),
            (commit_server, commit_server_bytes),
        ],
    );

    let (_blob_base_kind, blob_base_body) = loose_body(&blob_server_bytes);
    let (blob_new, _blob_new_body) = delta_product(&blob_base_body, b"-patched");
    let (commit_new, commit_new_bytes) = commit_fixture(tree_server, &[commit_server], "thin push");
    let commit_body = loose_body(&commit_new_bytes).1;
    let base_entry = pack_entry_base(gix_object::Kind::Commit, &commit_body, 0);
    let delta_entry = pack_entry_delta(
        gix_pack::data::entry::Header::RefDelta {
            base_id: blob_server,
        },
        &delta_stream(blob_base_body.len() as u64, b"-patched"),
        base_entry.bytes_in_pack(),
    );
    let thin_pack = pack_bytes(&[delta_entry, base_entry]);

    let updates = vec![RefUpdate {
        name: "refs/heads/main".to_string(),
        expected: Some(commit_server),
        new: Some(commit_new),
    }];
    let prepared = ingest
        .prepare(
            &doubled,
            updates.clone(),
            Box::new(io::Cursor::new(thin_pack)),
            None,
            Limits::default(),
        )
        .await
        .unwrap();
    assert!(
        prepared.unpack.is_ok(),
        "the thin pack's bases resolve from the namespace: {:?}",
        prepared.unpack
    );
    assert_eq!(prepared.updates, updates);
    assert!(ingest.objects().contains(&doubled, &blob_new));
    assert!(ingest.objects().contains(&doubled, &commit_new));

    let (orphan, _orphan_body) = delta_product(&blob_base_body, b"-orphaned");
    let missing_base_pack = pack_bytes(&[pack_entry_delta(
        gix_pack::data::entry::Header::RefDelta { base_id: oid(11) },
        &delta_stream(blob_base_body.len() as u64, b"-orphaned"),
        0,
    )]);
    let prepared = ingest
        .prepare(
            &doubled,
            updates.clone(),
            Box::new(io::Cursor::new(missing_base_pack)),
            None,
            Limits::default(),
        )
        .await
        .unwrap();
    let reason = prepared
        .unpack
        .clone()
        .expect_err("a missing thin base fails into the unpack ng leg");
    assert!(!reason.is_empty(), "the ng reason is server-chosen");
    assert_eq!(prepared.updates, updates, "verbatim on the ng path too");
    assert!(
        !ingest.objects().contains(&doubled, &orphan),
        "a failed ingestion lands nothing"
    );
}

/// In-pack ofs deltas expand against their staged bases — including
/// the copy instruction — and land with them; the walk closes over the
/// pack's own objects plus the namespace's pre-existing parent
/// history.
///
/// Contract stamp: ADR-020 §2 (the ingestion composition — every pack
/// object recovered, landed as loose rows); ADR-013 §5–6 (push packs
/// carry ofs deltas against in-pack bases; ingestion indexes them); ADR-013 §8
/// (a clean ingestion reports `unpack ok`); ADR-020 §2's walk note
/// (parents included over the combined graph).
#[tokio::test]
async fn in_pack_ofs_deltas_expand_and_land_with_their_bases() {
    let doubled = repo("owner/doubled");
    let (blob_a, blob_a_bytes) = blob_fixture(b"alpha");
    let (tree_a, tree_a_bytes) =
        tree_fixture(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob_a)]);
    let (commit_a, commit_a_bytes) = commit_fixture(tree_a, &[], "alpha");
    let objects = MemObjects::default();
    let ingest = MemPackIngest { objects };
    seed(
        ingest.objects(),
        &doubled,
        &[
            (blob_a, blob_a_bytes.clone()),
            (tree_a, tree_a_bytes),
            (commit_a, commit_a_bytes),
        ],
    );

    let (_blob_a_kind, blob_a_body) = loose_body(&blob_a_bytes);
    let (blob_b, _blob_b_body) = delta_product(&blob_a_body, b"-beta");
    let (tree_b, tree_b_bytes) = tree_fixture(&[
        (gix_object::tree::EntryKind::Blob, "a.txt", blob_a),
        (gix_object::tree::EntryKind::Blob, "b.txt", blob_b),
    ]);
    let (commit_b, commit_b_bytes) = commit_fixture(tree_b, &[commit_a], "beta");
    let base_entry = pack_entry_base(gix_object::Kind::Blob, &blob_a_body, 0);
    let delta_entry = pack_entry_delta(
        gix_pack::data::entry::Header::OfsDelta {
            base_distance: base_entry.bytes_in_pack(),
        },
        &delta_stream(blob_a_body.len() as u64, b"-beta"),
        base_entry.bytes_in_pack(),
    );
    let tree_body = loose_body(&tree_b_bytes).1;
    let tree_entry = pack_entry_base(
        gix_object::Kind::Tree,
        &tree_body,
        base_entry.bytes_in_pack() + delta_entry.bytes_in_pack(),
    );
    let commit_body = loose_body(&commit_b_bytes).1;
    let commit_entry = pack_entry_base(
        gix_object::Kind::Commit,
        &commit_body,
        base_entry.bytes_in_pack() + delta_entry.bytes_in_pack() + tree_entry.bytes_in_pack(),
    );
    let pack = pack_bytes(&[base_entry, delta_entry, tree_entry, commit_entry]);

    let updates = vec![RefUpdate {
        name: "refs/heads/main".to_string(),
        expected: Some(commit_a),
        new: Some(commit_b),
    }];
    let prepared = ingest
        .prepare(
            &doubled,
            updates.clone(),
            Box::new(io::Cursor::new(pack)),
            None,
            Limits::default(),
        )
        .await
        .unwrap();
    assert!(
        prepared.unpack.is_ok(),
        "the ofs-delta pack ingests: {:?}",
        prepared.unpack
    );
    assert_eq!(prepared.updates, updates);
    let ingested = ingest.objects();
    for oid in [blob_b, tree_b, commit_b] {
        assert!(ingested.contains(&doubled, &oid), "{oid} landed");
    }
    assert_eq!(
        object_closure(ingested, &doubled, &[commit_b]),
        BTreeSet::from([blob_a, blob_b, tree_a, tree_b, commit_a, commit_b]),
        "the combined-graph walk closes over the pack and the namespace"
    );
}

/// A ref-delta whose in-pack base comes later in the pack still
/// resolves — the fixpoint pass parks the entry until its base stages —
/// and an unresolvable delta never blocks the ng reason from naming
/// its base.
///
/// Contract stamp: ADR-013 §5 (base availability is enforced by
/// ingestion — in-pack resolution including forward references, the
/// indexer's own second-pass posture); ADR-020 §2 (every pack object
/// recovered); ADR-013 §8 (failures land in the `unpack ng` leg).
#[tokio::test]
async fn forward_ref_deltas_resolve_after_their_in_pack_base() {
    let doubled = repo("owner/doubled");
    let ingest = MemPackIngest::default();

    let (blob_base, blob_base_bytes) = blob_fixture(b"forward-base");
    let (_blob_base_kind, blob_base_body) = loose_body(&blob_base_bytes);
    let (blob_new, _blob_new_body) = delta_product(&blob_base_body, b"-forwarded");
    let delta_entry = pack_entry_delta(
        gix_pack::data::entry::Header::RefDelta { base_id: blob_base },
        &delta_stream(blob_base_body.len() as u64, b"-forwarded"),
        0,
    );
    let base_entry = pack_entry_base(
        gix_object::Kind::Blob,
        &blob_base_body,
        delta_entry.bytes_in_pack(),
    );
    let pack = pack_bytes(&[delta_entry, base_entry]);

    let updates = vec![RefUpdate {
        name: "refs/heads/forward".to_string(),
        expected: None,
        new: Some(blob_new),
    }];
    let prepared = ingest
        .prepare(
            &doubled,
            updates.clone(),
            Box::new(io::Cursor::new(pack)),
            None,
            Limits::default(),
        )
        .await
        .unwrap();
    assert!(
        prepared.unpack.is_ok(),
        "the forward ref-delta resolves once its base stages: {:?}",
        prepared.unpack
    );
    assert_eq!(prepared.updates, updates);
    assert!(ingest.objects().contains(&doubled, &blob_base));
    assert!(ingest.objects().contains(&doubled, &blob_new));
}

/// The connectivity walk runs from the pushed tips with parents
/// included: a pack whose commit's parent exists in neither the pack
/// nor the namespace fails into the `unpack` ng leg naming the missing
/// parent; a push whose tip already exists server-side over an empty
/// pack passes the walk.
///
/// Contract stamp: ADR-020 §2 (the full-graph connectivity walk from
/// the pushed tips, parents included — `gix-fsck`'s
/// `Connectivity::check_commit` does not walk parents, the owning
/// implementation walks); ADR-013 §8 (the walk failure lands in the
/// `unpack ng` leg); ADR-018 §5 (the verbatim round-trip rule holds on
/// the ng path); ADR-018 §7 (the reason may name the client-computed
/// oid, no oracle risk).
#[tokio::test]
async fn connectivity_walk_from_pushed_tips_includes_parents() {
    let doubled = repo("owner/doubled");
    let (blob, blob_bytes) = blob_fixture(b"connected");
    let (tree, tree_bytes) = tree_fixture(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob)]);
    let (commit_child, commit_child_bytes) = commit_fixture(tree, &[oid(9)], "orphan child");
    let objects = MemObjects::default();
    let ingest = MemPackIngest { objects };
    seed(
        ingest.objects(),
        &doubled,
        &[
            (blob, blob_bytes),
            (tree, tree_bytes),
            (commit_child, commit_child_bytes.clone()),
        ],
    );

    let missing_parent_pack = pack_bytes(&[pack_entry_base(
        gix_object::Kind::Commit,
        &loose_body(&commit_child_bytes).1,
        0,
    )]);
    let updates = vec![RefUpdate {
        name: "refs/heads/main".to_string(),
        expected: None,
        new: Some(commit_child),
    }];
    let prepared = ingest
        .prepare(
            &doubled,
            updates.clone(),
            Box::new(io::Cursor::new(missing_parent_pack)),
            None,
            Limits::default(),
        )
        .await
        .unwrap();
    let reason = prepared
        .unpack
        .clone()
        .expect_err("a missing parent fails the walk into the unpack ng leg");
    assert!(
        reason.contains("missing object"),
        "the reason names the miss"
    );
    assert_eq!(prepared.updates, updates, "verbatim on the ng path too");

    let (commit_seed, commit_seed_bytes) = commit_fixture(tree, &[], "seeded");
    ingest
        .objects()
        .put(&doubled, commit_seed, commit_seed_bytes);
    let empty_pack = pack_bytes(&[]);
    let existing_tip = vec![RefUpdate {
        name: "refs/heads/main".to_string(),
        expected: Some(oid(9)),
        new: Some(commit_seed),
    }];
    let prepared = ingest
        .prepare(
            &doubled,
            existing_tip.clone(),
            Box::new(io::Cursor::new(empty_pack)),
            None,
            Limits::default(),
        )
        .await
        .unwrap();
    assert!(
        prepared.unpack.is_ok(),
        "a tip that already exists server-side passes over an empty pack: {:?}",
        prepared.unpack
    );
    assert_eq!(prepared.updates, existing_tip);
}

/// An empty pack (zero objects) is a valid ingestion — the trailer the
/// streaming machinery never reaches is verified by the double — and
/// reports `unpack ok`.
///
/// Contract stamp: ADR-020 §2 (the ingestion composition — zero
/// objects is valid and reported as such, mirroring upstream's
/// empty-pack note); ADR-013 §8; ADR-018 §5.
#[tokio::test]
async fn empty_pack_is_valid_and_reports_unpack_ok() {
    let doubled = repo("owner/doubled");
    let ingest = MemPackIngest::default();
    let updates = vec![RefUpdate {
        name: "refs/heads/untouched".to_string(),
        expected: None,
        new: None,
    }];
    let prepared = ingest
        .prepare(
            &doubled,
            updates.clone(),
            Box::new(io::Cursor::new(pack_bytes(&[]))),
            None,
            Limits::default(),
        )
        .await
        .unwrap();
    assert!(
        prepared.unpack.is_ok(),
        "the empty pack is valid: {:?}",
        prepared.unpack
    );
    assert_eq!(prepared.updates, updates);
}

/// A pack stream exceeding `limits.max_pack_size` fails into the
/// `unpack` ng leg through the counting reader — the impl-side defense
/// behind the wire-side relay count — never `Err`, and nothing lands.
///
/// Contract stamp: ADR-009 (the fail-closed budget, the impl's own
/// pack-size defense); ADR-013 §6 (budgets: max pack size counts the
/// streamed body); ADR-013 §8 (the ng leg continues the session);
/// backend.md §Concurrency model (limits flow into ingest); ADR-018 §5
/// (the verbatim round-trip rule holds on the ng path).
#[tokio::test]
async fn pack_stream_exceeding_max_pack_size_fails_into_unpack_ng() {
    let doubled = repo("owner/doubled");
    let (_blob, blob_bytes) = blob_fixture(b"budget-breach");
    let ingest = MemPackIngest::default();
    let oversized_pack = pack_bytes(&[pack_entry_base(
        gix_object::Kind::Blob,
        &loose_body(&blob_bytes).1,
        0,
    )]);
    let limits = Limits {
        max_pack_size: 16,
        ..Limits::default()
    };
    assert!(
        oversized_pack.len() > 16,
        "the fixture pack stream exceeds the test budget"
    );
    let updates = vec![RefUpdate {
        name: "refs/heads/main".to_string(),
        expected: None,
        new: Some(oid(4)),
    }];
    let prepared = ingest
        .prepare(
            &doubled,
            updates.clone(),
            Box::new(io::Cursor::new(oversized_pack)),
            None,
            limits,
        )
        .await
        .unwrap();
    let reason = prepared
        .unpack
        .clone()
        .expect_err("the budget breach fails into the unpack ng leg");
    assert!(!reason.is_empty());
    assert_eq!(prepared.updates, updates, "verbatim on the ng path too");
    assert!(!ingest.objects().contains(&doubled, &oid(4)));
}

/// Corrupt and truncated pack streams fail into the `unpack` ng leg
/// with server-chosen reasons — pack-level validation never produces
/// `Err` (`Err` is for infrastructure failure only) — and nothing
/// partial lands.
///
/// Contract stamp: ADR-018 §5 (pack-level validation failures land in
/// the unpack ng leg, not as `Err`); ADR-018 §7 (`Err` is substrate/
/// infrastructure failure only); ADR-020 §2 (the ingestion
/// composition's verify posture — corruption cannot enter undetected);
/// ADR-013 §8; ADR-009 (the ng leg keeps the session bounded).
#[tokio::test]
async fn corrupt_and_truncated_packs_fail_into_unpack_ng_and_err_is_unreachable() {
    let doubled = repo("owner/doubled");
    let (_blob, blob_bytes) = blob_fixture(b"truncation-fodder");
    let ingest = MemPackIngest {
        objects: MemObjects::default(),
    };
    let pack = pack_bytes(&[pack_entry_base(
        gix_object::Kind::Blob,
        &loose_body(&blob_bytes).1,
        0,
    )]);

    let truncated = pack[..pack.len() - 4].to_vec();
    let updates = vec![RefUpdate {
        name: "refs/heads/main".to_string(),
        expected: None,
        new: Some(oid(5)),
    }];
    let prepared = ingest
        .prepare(
            &doubled,
            updates.clone(),
            Box::new(io::Cursor::new(truncated)),
            None,
            Limits::default(),
        )
        .await
        .unwrap();
    let reason = prepared
        .unpack
        .clone()
        .expect_err("the truncated pack fails into the unpack ng leg");
    assert!(!reason.is_empty());
    assert_eq!(prepared.updates, updates);

    let mut corrupt = pack.clone();
    let corrupt_at = gix_pack::data::header::SIZE + 2;
    corrupt[corrupt_at] ^= 0x40;
    let prepared = ingest
        .prepare(
            &doubled,
            updates.clone(),
            Box::new(io::Cursor::new(corrupt)),
            None,
            Limits::default(),
        )
        .await
        .unwrap();
    let reason = prepared
        .unpack
        .clone()
        .expect_err("the corrupt pack fails into the unpack ng leg");
    assert!(!reason.is_empty());
    assert_eq!(prepared.updates, updates);
    assert!(!ingest.objects().contains(&doubled, &oid(5)));
}

/// A pack entry whose expansion exceeds the ingestion cap fails into
/// the `unpack` ng leg — the decompressed-size cap
/// (`Limits::max_object_size`, review 004 R-2) fires on the inflated
/// payload the pack verifies as self-consistent — never `Err`, nothing
/// lands, the updates round-trip verbatim.
///
/// The size-bomb regression: the old `inflate_entry`
/// allocate-then-fill shape trusted the pack-declared size and
/// succeeded here (landing the oversized object); the
/// incremental-read-under-the-cap guard rejects the entry on its
/// declaration instead, so a hostile declarable size can produce
/// neither an oversized object nor a declaration-sized allocation.
///
/// The cap is overridden low so the row stays a kilobyte-scale
/// fixture; the default cap's ordinary-pack tolerance is the existing
/// green ingest rows.
///
/// Contract stamp: ADR-009 (amended 2026-10-23: the
/// decompressed-size cap — `max_object_size`, bounds every object the
/// ingestion materializes); review 004 R-2 (the size-bomb class —
/// incremental inflate, no allocation sized by the pack's
/// declaration); ADR-013 §8 (the breach is the `unpack ng` leg, never
/// a session error); ADR-018 §5 (the verbatim round-trip rule holds on
/// the ng path).
#[tokio::test]
async fn a_decompressed_size_beyond_the_ingestion_cap_fails_into_unpack_ng() {
    let doubled = repo("owner/doubled");
    let body = vec![0u8; 100_000];
    let (expected, _expected_bytes) = blob_fixture(&body);
    let ingest = MemPackIngest::default();
    let pack = pack_bytes(&[pack_entry_base(gix_object::Kind::Blob, &body, 0)]);
    let limits = Limits {
        max_object_size: 65536,
        ..Limits::default()
    };
    let updates = vec![RefUpdate {
        name: "refs/heads/main".to_string(),
        expected: None,
        new: Some(expected),
    }];
    let prepared = ingest
        .prepare(
            &doubled,
            updates.clone(),
            Box::new(io::Cursor::new(pack)),
            None,
            limits,
        )
        .await
        .unwrap();
    let reason = prepared
        .unpack
        .clone()
        .expect_err("the expansion beyond the cap fails into the unpack ng leg");
    assert!(
        reason.contains("ingestion cap"),
        "the reason names the decompressed-size cap: {reason}"
    );
    assert!(!ingest.objects().contains(&doubled, &expected));
    assert_eq!(prepared.updates, updates, "verbatim on the ng path too");
}

/// A pack entry whose *declared* decompressed size is hostile (a
/// bomb-shaped declaration far above anything the compressed stream
/// can deflate to) fails into the `unpack` ng leg at the machinery's
/// own declared-size verification — the declaration mismatch never
/// reaches an allocation on the double's path and `Err` stays
/// unreachable.
///
/// Contract stamp: ADR-009 (amended 2026-10-23: the pack's declared
/// size is untrusted — nothing in the ingest path sizes an allocation
/// from it); review 004 R-2 (the size-bomb class); ADR-013 §8 (the
/// `unpack ng` leg, never a session error); ADR-018 §5 (the verbatim
/// round-trip rule holds on the ng path).
#[tokio::test]
async fn a_hostile_declared_decompressed_size_fails_into_unpack_ng() {
    let doubled = repo("owner/doubled");
    let ingest = MemPackIngest::default();
    let mut hostile = pack_entry_base(gix_object::Kind::Blob, b"", 0);
    hostile.decompressed_size = 1u64 << 40;
    let pack = pack_bytes(&[hostile]);
    let updates = vec![RefUpdate {
        name: "refs/heads/main".to_string(),
        expected: None,
        new: Some(oid(6)),
    }];
    let prepared = ingest
        .prepare(
            &doubled,
            updates.clone(),
            Box::new(io::Cursor::new(pack)),
            None,
            Limits::default(),
        )
        .await
        .unwrap();
    let reason = prepared
        .unpack
        .clone()
        .expect_err("the hostile declaration fails into the unpack ng leg");
    assert!(!reason.is_empty(), "the ng reason is server-chosen");
    assert!(!ingest.objects().contains(&doubled, &oid(6)));
    assert_eq!(prepared.updates, updates, "verbatim on the ng path too");
}

/// A delta stream whose declared materialized result exceeds the
/// ingestion cap fails into the `unpack` ng leg — the applier's cap
/// check fires before any result bytes accumulate (the old
/// declaration-sized `with_capacity` is gone) — never `Err`.
///
/// Contract stamp: ADR-009 (amended 2026-10-23: the cap bounds each
/// delta's materialized result — the object the pack produces — not
/// just the entry payloads); review 004 R-2 (the size-bomb class);
/// ADR-013 §5–8 (the delta's in-pack base resolves, then the hostile
/// result declaration takes the `unpack ng` leg, never a session
/// error).
#[tokio::test]
async fn a_delta_declaring_a_result_beyond_the_cap_fails_into_unpack_ng() {
    let doubled = repo("owner/doubled");
    let (_blob_base, blob_base_bytes) = blob_fixture(b"delta-base");
    let (_blob_base_kind, blob_base_body) = loose_body(&blob_base_bytes);
    let base_entry = pack_entry_base(gix_object::Kind::Blob, &blob_base_body, 0);
    let delta_entry = pack_entry_delta(
        gix_pack::data::entry::Header::OfsDelta {
            base_distance: base_entry.bytes_in_pack(),
        },
        &hostile_delta_stream(blob_base_body.len() as u64, 200_000),
        base_entry.bytes_in_pack(),
    );
    let ingest = MemPackIngest {
        objects: MemObjects::default(),
    };
    let pack = pack_bytes(&[base_entry, delta_entry]);
    let limits = Limits {
        max_object_size: 65536,
        ..Limits::default()
    };
    let updates = vec![RefUpdate {
        name: "refs/heads/main".to_string(),
        expected: None,
        new: Some(oid(7)),
    }];
    let prepared = ingest
        .prepare(
            &doubled,
            updates.clone(),
            Box::new(io::Cursor::new(pack)),
            None,
            limits,
        )
        .await
        .unwrap();
    let reason = prepared
        .unpack
        .clone()
        .expect_err("the hostile delta result fails into the unpack ng leg");
    assert!(
        reason.contains("ingestion cap"),
        "the reason names the decompressed-size cap: {reason}"
    );
    assert!(!ingest.objects().contains(&doubled, &oid(7)));
}