alkgit-substrate 0.1.0

//! The w8 end-to-end rows: real `git` (2.43.0, the capture-session
//! client) driven against the substrate-backed server — the composition
//! entry's complete backend — over the core's real-client door shapes
//! (the git-daemon-framed duplex door; the smart-http-framed stateless
//! door). The POC-O `tests/e2e.rs` gates re-landed here (the task row:
//! "the rows re-land its patterns over the substrate backend"); every
//! machine consumes the real substrate (registry pair, refs rows,
//! packgen, the w8 ingest), not doubles.
//!
//! The doors and the git-client helpers are the shared harness
//! (`substrate/tests/common`; the core's `tests/common` is not
//! importable across the crate seam — the w7-packgen byte-identity
//! note's posture).
//!
//! `git` is a test client only (the serving path spawns nothing — the
//! crate convention); the rows skip cleanly (with a note) when the
//! `git` binary is absent.
//!
//! Rows are names + `Contract stamp:` versions per
//! `substrate/tests/common` §The row/stamp discipline.

mod common;

use std::collections::BTreeSet;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

#[cfg(feature = "sqlite")]
use alkgit_substrate::PackServing;
use gix_hash::ObjectId;

use alkgit_substrate::{MaintenanceOpts, WorkerEvent};
use common::{
    extract_and_walk_client_pack, git, git_run, rev_list_oids, sideband_report_lines, DoorOptions,
    GateRendezvous, Harness, DEFAULT_REPO_ID,
};
use parking_lot::Mutex;

/// The multi-KB fixture file one real client commits (large enough that
/// the real client's thin-pack computation deltas it: the second push's
/// blob change rides a ref-delta whose base is the pre-push blob, packed
/// wire-shape-verified against real git 2.43.0).
const FIXTURE_LINES: usize = 300;

fn fixture_line(i: usize) -> String {
    format!("alpha line {i} of the thin-push fixture file\n")
}

fn file_content(marker: &str) -> String {
    let mut content: String = (0..FIXTURE_LINES).map(fixture_line).collect();
    content.push_str(marker);
    content
}

/// The rows' local git client repo: the seed commit (the fixture file's
/// baseline) already made; the repo's identity is its own config (the
/// isolated env carries none).
struct ClientDir(PathBuf);

impl ClientDir {
    fn path(&self) -> &Path {
        &self.0
    }
}

/// A local repo initialized at `base/name` (main, repo-local identity).
async fn init_client(base: &Path, name: &str) -> Option<ClientDir> {
    let dir = base.join(name);
    std::fs::create_dir_all(&dir).expect("the client dir creates");
    git(
        Some(base),
        &["init", "-q", "-b", "main", dir.to_str().expect("utf8 path")],
    )
    .await?;
    git(Some(&dir), &["config", "user.name", "alkgit e2e"]).await?;
    git(Some(&dir), &["config", "user.email", "e2e@alk.dev"]).await?;
    Some(ClientDir(dir))
}

/// Writes the fixture file at `marker` and commits; returns the new
/// tip's hex oid.
async fn commit_fixture_file(client: &ClientDir, marker: &str, message: &str) -> Option<String> {
    std::fs::write(client.path().join("a.txt"), file_content(marker))
        .expect("the fixture file writes");
    git(Some(client.path()), &["add", "a.txt"]).await?;
    git(Some(client.path()), &["commit", "-q", "-m", message]).await?;
    rev_parse(client, "HEAD").await
}

/// A plain `rev-parse` (hex, trimmed).
async fn rev_parse(client: &ClientDir, what: &str) -> Option<String> {
    let run = git(Some(client.path()), &["rev-parse", what]).await?;
    Some(run.stdout.trim().to_owned())
}

/// `git rev-list --objects` from the client repo, as an oid set.
async fn introduced_oids(client: &ClientDir, new: &str, base: &str) -> Option<BTreeSet<ObjectId>> {
    let run = git(
        Some(client.path()),
        &["rev-list", "--objects", new, &format!("^{}", base)],
    )
    .await?;
    Some(rev_list_oids(&run.stdout))
}

/// Real `git push <url>` with `args` prefix (thin/force/atomic options).
async fn real_push(
    client: &ClientDir,
    url: &str,
    mut args: Vec<&str>,
    refspecs: &[&str],
) -> Option<common::GitRun> {
    let mut all = vec!["push"];
    all.append(&mut args);
    all.push(url);
    for refspec in refspecs {
        all.push(refspec);
    }
    git(Some(client.path()), &all).await
}

/// Real `git clone <url> <dst>` (V2; the fetch sessions are V2-only
/// client-side too).
async fn real_clone(url: &str, dst: &Path) -> Option<common::GitRun> {
    git(
        None,
        &[
            "-c",
            "protocol.version=2",
            "clone",
            "-q",
            url,
            dst.to_str().expect("utf8 dst"),
        ],
    )
    .await
}

/// Real `git fetch origin`.
async fn real_fetch(dst: &Path) -> Option<common::GitRun> {
    git(Some(dst), &["fetch", "origin"]).await
}

/// Real `git fsck --strict` inside a checkout.
async fn fsck_strict(dst: &Path) -> Option<common::GitRun> {
    git(Some(dst), &["fsck", "--strict"]).await
}

/// The clone's post-state: `HEAD` resolves to the expected tip, the
/// worktree file's bytes are the fixture content, and `git fsck
/// --strict` is clean (the clone's own pack ingestion validated the
/// served stream).
async fn assert_clone_checked_out(dst: &Path, expected_tip: &str, expected_content: &str) {
    let client = ClientDir(dst.to_path_buf());
    let head = rev_parse(&client, "HEAD")
        .await
        .expect("the clone's HEAD resolves");
    assert_eq!(
        head, expected_tip,
        "the clone checked out the pushed commit"
    );
    let bytes = std::fs::read(dst.join("a.txt")).expect("the fixture file checked out");
    assert_eq!(
        String::from_utf8_lossy(&bytes),
        expected_content,
        "the checked-out file is the pushed content"
    );
    fsck_strict(dst)
        .await
        .expect("the run rode a resolved git binary")
        .expect_ok();
}

/// A door URL helper for the row's push invocations.
fn door_url(kind: &str, addr: std::net::SocketAddr, repo_id: &str) -> String {
    match kind {
        "git" => format!("git://{addr}/{repo_id}"),
        _ => format!("http://{addr}/{repo_id}"),
    }
}

/// The durable feed's decoded events up to the current watermark.
async fn feed_events(harness: &Harness, since: i64) -> Vec<alkgit_substrate::RegistryEvent> {
    let store = harness.assembly.reactor().store().clone();
    let feed = store
        .stream(alkgit_substrate::EVENTS_STREAM)
        .await
        .expect("the feed opens");
    feed.read_since(since, 100)
        .await
        .expect("the feed reads")
        .iter()
        .filter_map(alkgit_substrate::RegistryEvent::decode_event)
        .collect()
}

/// The refs side's listed tips of a record, by name.
async fn listed_tips(harness: &Harness) -> Vec<(String, Option<ObjectId>)> {
    harness
        .assembly
        .refs()
        .list_refs(&harness.record)
        .await
        .expect("the refs side lists")
        .into_iter()
        .map(|line| (line.name, line.oid))
        .collect()
}

/// The seeded HEAD row's identity: the unborn symref line persists
/// through pushes and the row never lists an oid.
async fn assert_unborn_head_row_persists(harness: &Harness) {
    let tips = listed_tips(harness).await;
    let head = tips
        .iter()
        .find(|(name, _)| name == "HEAD")
        .expect("the seeded HEAD row persists");
    assert!(
        head.1.is_none(),
        "the seeded HEAD stays the unborn symref row (a push never touches it)"
    );
}

/// The pushed-graph liveness coverage at the sweep, with the working
/// area's residue assertion folded in (the working-area discard on
/// every path): the pushes' work directories are gone.
async fn assert_work_area_clean(harness: &Harness) {
    let root = harness.work_root();
    let mut read = tokio::fs::read_dir(&root)
        .await
        .expect("the working-area root opens");
    let mut names = Vec::new();
    while let Some(entry) = read.next_entry().await.expect("the work dir reads") {
        names.push(entry.file_name());
    }
    assert!(
        names.is_empty(),
        "no per-push working directory remains after the pushes: {names:?}"
    );
}

/// An ordinary real `git push` over the duplex git-daemon door on the
/// substrate-backed server: objects land in the store under their git
/// oids (the client's commit, tree, and blob each stat present at the
/// front), the ref CAS applies (`refs/heads/main` from the unborn-HEAD
/// seed; the seeded HEAD row persists unborn), the status report frames
/// per ADR-013 §8 (band-1 wrapped when side-band-64k was selected:
/// `unpack ok` + per-ref `ok <ref>`, inner flush, outer flush), the
/// reactive legs fire at the end-to-end level (the marker watermark
/// advances past the record's put; the durable `Mutation` event lands
/// in the feed; the wake channel delivers), and the pins commit after
/// the CAS — the registered closure holds the pushed graph across a
/// sweep (sweep safety).
///
/// Contract stamp: ADR-020 §2 (the ingestion composition, unpack-to-loose
/// under the git oids' family keys, the pin bridge); ADR-013 §7–8 (the
/// CAS home on the unborn-HEAD seed, the status report framing);
/// ADR-016 §2 (the git-daemon request line the real client sends for
/// push — no `version=2`); ADR-019 §7 (the marker + announce legs, the
/// watermark); poco-findings (the wire-backed gate this row re-lands);
/// substrate/w8-e2e-real-git.
#[tokio::test(flavor = "multi_thread")]
async fn a_real_git_push_lands_objects_the_cas_and_the_reactive_legs_over_the_duplex_door() {
    if common::git_command().is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return;
    }
    let harness = Harness::mem("ordinary-push", DEFAULT_REPO_ID).await;
    let capture = common::StreamCapture::new();
    let base =
        std::env::temp_dir().join(format!("alkgit-real-git-ordinary-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&base);
    std::fs::create_dir_all(&base).expect("the row's base creates");

    let door = common::spawn_daemon_door(
        &harness,
        DoorOptions {
            capture: Some(capture.clone()),
            ..DoorOptions::default()
        },
    )
    .await;
    let client = init_client(&base, "seed").await.expect("the client inits");
    commit_fixture_file(&client, "", "the ordinary push's commit")
        .await
        .expect("the seed commit lands");

    let mut wake = harness
        .assembly
        .reactor()
        .store()
        .listen(alkgit_substrate::WAKE_CHANNEL)
        .await
        .expect("the wake source opens");

    let url = door_url("git", door.addr, &harness.record.repo_id);
    let push = real_push(&client, &url, Vec::new(), &["main"])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();

    let wake_event = tokio::time::timeout(
        Duration::from_secs(5),
        wake.recv_timeout(Duration::from_secs(5)),
    )
    .await
    .expect("the announce's wake leg delivers inside the deadline")
    .expect("the wake source stayed open")
    .expect("the wake event delivered");
    assert_eq!(wake_event.channel, alkgit_substrate::WAKE_CHANNEL);
    assert!(
        push.stderr().contains("main") && (push.stderr().contains("new branch")),
        "the real client reported the new branch: {}",
        push.stderr()
    );

    let report = sideband_report_lines(&capture.to_client());
    assert_eq!(
        report,
        vec!["unpack ok".to_owned(), "ok refs/heads/main".to_owned(),],
        "the captured status report frames per ADR-013 §8 (band-1 wrapped, inner flush)"
    );
    assert!(
        extract_and_walk_client_pack(&capture.to_server()).is_some(),
        "the ordinary push's pack rode the wire (the peek saw the header)"
    );

    let commit_tip = rev_parse(&client, "HEAD").await.expect("tip");
    let tree = rev_parse(&client, "HEAD^{tree}")
        .await
        .expect("tree resolves");
    let blob = rev_parse(&client, "HEAD:a.txt")
        .await
        .expect("blob resolves");
    let tip_oid = gix_hash::ObjectId::from_hex(commit_tip.as_bytes()).expect("the tip oid parses");
    let tree_oid = gix_hash::ObjectId::from_hex(tree.as_bytes()).expect("the tree oid parses");
    let blob_oid = gix_hash::ObjectId::from_hex(blob.as_bytes()).expect("the blob oid parses");
    for (what, oid) in [
        ("the commit (tip)", tip_oid),
        ("the tree", tree_oid),
        ("the blob", blob_oid),
    ] {
        let stat = harness
            .assembly
            .front()
            .stat(&oid)
            .await
            .expect("the front stats");
        assert!(
            stat.is_some(),
            "the pushed {what} landed in the store under its git oid"
        );
    }
    assert_unborn_head_row_persists(&harness).await;
    let tips = listed_tips(&harness).await;
    assert_eq!(
        tips.iter()
            .find(|(name, _)| name == "refs/heads/main")
            .and_then(|(_, oid)| *oid),
        Some(tip_oid),
        "the ref CAS applied the pushed main over the unborn-HEAD seed"
    );
    assert!(
        harness.assembly.refs_version() >= 1,
        "the CAS bumped the write counter"
    );

    assert_eq!(
        harness
            .assembly
            .reactor()
            .registry()
            .watermark()
            .await
            .expect("the watermark reads"),
        2,
        "the marker watermark advanced past the record put (1) to the push's marker (2)"
    );
    assert_eq!(
        feed_events(&harness, 0).await,
        vec![
            alkgit_substrate::RegistryEvent::Put {
                seq: 1,
                record: harness.record.clone(),
            },
            alkgit_substrate::RegistryEvent::Mutation {
                seq: 2,
                repo_id: harness.record.repo_id.clone(),
            },
        ],
        "the reactive legs' durable feed shows the put and the push's Mutation marker"
    );

    let report = harness
        .assembly
        .blobs()
        .sweep()
        .await
        .expect("the sweep runs");
    assert!(
        report.aborted.is_none(),
        "the sweep ran with the registered closure"
    );
    assert_eq!(
        report.deleted, 0,
        "the pushed objects are liveness-covered — the pins committed after the CAS"
    );
    assert!(harness
        .assembly
        .front()
        .stat(&tip_oid)
        .await
        .expect("the front stats")
        .is_some());

    assert_work_area_clean(&harness).await;
    door.accept.abort();
    std::fs::remove_dir_all(&base).ok();
    drop(harness);
}

/// A thin real `git push`: a client that cloned the substrate-backed
/// repo pushes a small change, and the real client's pack — captured at
/// the door — carries the thin shape (a ref-delta entry whose base oid
/// is not among the pack's own entries: the base objects are never
/// re-sent), the pushed pack's full entries are exactly clientside
/// introduced objects (`rev-list --objects new ^base`), the thin bases
/// resolve from the substrate Find (the push commits — `git
/// index-pack --strict`-class acceptance, the machine's ingestion
/// resolving bases server-side), and a fresh clone of the substrate
/// checked out at the new tip reads the exact new file content.
///
/// Contract stamp: ADR-020 §2 (thin-pack bases resolving from the
/// substrate odb — default client behavior, no capability); ADR-013 §5
/// (the server-odb thin-base resolution); ADR-019 §6 (the Find front);
/// push-captures.md §Thin pack (git sends thin packs by default —
/// push.thin=true); poco-findings DQ-4 (the ingest composition's
/// thin-base leg); substrate/w8-e2e-real-git.
#[tokio::test(flavor = "multi_thread")]
async fn a_real_thin_push_pushes_no_base_objects_and_resolves_them_from_the_substrate() {
    if common::git_command().is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return;
    }
    let harness = Harness::mem("thin-push", DEFAULT_REPO_ID).await;
    let capture = common::StreamCapture::new();
    let base = std::env::temp_dir().join(format!("alkgit-real-git-thin-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&base);
    std::fs::create_dir_all(&base).expect("the row's base creates");

    let door = common::spawn_daemon_door(&harness, DoorOptions::default()).await;
    let door_thin = common::spawn_daemon_door(
        &harness,
        DoorOptions {
            capture: Some(capture.clone()),
            ..DoorOptions::default()
        },
    )
    .await;
    let client = init_client(&base, "origin")
        .await
        .expect("the client inits");
    let tip_c1 = commit_fixture_file(&client, "", "the first push's commit")
        .await
        .expect("the first commit lands");
    let url = door_url("git", door.addr, &harness.record.repo_id);
    real_push(&client, &url, Vec::new(), &["main"])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();

    let url_thin = door_url("git", door_thin.addr, &harness.record.repo_id);

    let tip_c2 = commit_fixture_file(
        &client,
        "this second push appends a small marker line to the fixture file\n",
        "the thin push's commit",
    )
    .await
    .expect("the second commit lands");
    let introduced = introduced_oids(&client, &tip_c2, &tip_c1)
        .await
        .expect("the introduced set resolves");
    assert!(
        !introduced.is_empty(),
        "the thin push has something to introduce"
    );
    let old_blob = rev_parse(&client, &format!("{tip_c1}:a.txt"))
        .await
        .expect("the base blob resolves");
    let old_blob_oid =
        gix_hash::ObjectId::from_hex(old_blob.as_bytes()).expect("the base blob oid parses");
    assert!(
        harness
            .assembly
            .front()
            .stat(&old_blob_oid)
            .await
            .expect("the front stats")
            .is_some(),
        "the thin base exists server-side before the push"
    );

    let pack = {
        real_push(&client, &url_thin, Vec::new(), &["main"])
            .await
            .expect("the row rides a resolved git binary")
            .expect_ok();
        extract_and_walk_client_pack(&capture.to_server())
            .expect("the thin push's pack rode the wire")
    };

    assert_eq!(
        pack.object_count as usize, pack.entry_count,
        "the captured pack's entries all parsed"
    );
    let thin_bases: Vec<ObjectId> = pack
        .deltas
        .iter()
        .filter(|(_, kind)| *kind == common::DeltaKind::ThinBase)
        .map(|(base, _)| *base)
        .collect();
    assert!(
        !thin_bases.is_empty(),
        "the real client's pack is thin: it carries delta entries whose bases are \
         not among the pack's own objects (push-captures.md §Thin pack)"
    );
    for base in &thin_bases {
        assert!(
            harness
                .assembly
                .front()
                .stat(base)
                .await
                .expect("the front stats")
                .is_some(),
            "every thin base resolves from the substrate Find"
        );
        assert!(
            !introduced.contains(base),
            "the thin bases are the ancestry the client excluded — never re-sent"
        );
    }
    assert!(
        pack.full_oids.iter().all(|oid| introduced.contains(oid)),
        "the pack's full entries are exactly the introduced objects"
    );

    let tips = listed_tips(&harness).await;
    assert_eq!(
        tips.iter()
            .find(|(name, _)| name == "refs/heads/main")
            .and_then(|(_, oid)| *oid),
        Some(gix_hash::ObjectId::from_hex(tip_c2.as_bytes()).expect("the tip oid parses")),
        "the thin push's CAS applied the pushed main"
    );

    let verify_dir = base.join("verify");
    real_clone(&url, &verify_dir)
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    let content =
        std::fs::read_to_string(verify_dir.join("a.txt")).expect("the fresh clone has the file");
    assert_eq!(
        content,
        file_content("this second push appends a small marker line to the fixture file\n"),
        "the fresh clone reads the exact pushed content — the thin base materialized server-side"
    );
    fsck_strict(&verify_dir)
        .await
        .expect("the run rode a resolved git binary")
        .expect_ok();

    door.accept.abort();
    door_thin.accept.abort();
    std::fs::remove_dir_all(&base).ok();
    drop(harness);
}
/// A packless real push: the delete-only row over the duplex
/// git-daemon door. A real `git push` that only deletes sends the
/// commands + flush and then *no pack bytes at all* (captured: the
/// client's byte stream ends at the command flush) and waits for the
/// report — the machine's bounded peek takes the packless, CAS-only
/// path (`delete-refs` was advertised; the same one-transaction CAS
/// home, the same status report), the committed delete rides the
/// mutation marker (the watermark advances), and the deleted ref's
/// objects lose their coverage: the sweep reclaims them (`old`'s own
/// commit and blob are gone from the store; `main`'s objects remain).
///
/// Contract stamp: ADR-020 §1 (the bounded peek's packless path, the
/// real client's waiting shape — §3) + §3; ADR-013 §7 (the CAS home the
/// packless path rides, deletes allowed), §8 (the report framing);
/// ADR-016 §2 (the push request line); ADR-019 §7 (the marker legs);
/// poco-findings §the packless protocol fact (the gate this row
/// re-lands); ADR-018 §6 (the liveness closure releasing with the ref
/// row); substrate/w8-e2e-real-git.
#[tokio::test(flavor = "multi_thread")]
async fn a_real_packless_delete_push_takes_the_bounded_peek_and_the_sweep_reclaims_the_released_refs_objects(
) {
    if common::git_command().is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return;
    }
    let harness = Harness::mem("packless-delete", DEFAULT_REPO_ID).await;
    let base = std::env::temp_dir().join(format!("alkgit-real-git-delete-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&base);
    std::fs::create_dir_all(&base).expect("the row's base creates");

    let door_setup = common::spawn_daemon_door(&harness, DoorOptions::default()).await;
    let delete_capture = common::StreamCapture::new();
    let door_delete = common::spawn_daemon_door(
        &harness,
        DoorOptions {
            capture: Some(delete_capture.clone()),
            ..DoorOptions::default()
        },
    )
    .await;
    let client = init_client(&base, "origin")
        .await
        .expect("the client inits");
    let url_setup = door_url("git", door_setup.addr, &harness.record.repo_id);
    let url_delete = door_url("git", door_delete.addr, &harness.record.repo_id);

    let tip_main = commit_fixture_file(&client, "", "main's commit")
        .await
        .expect("main's commit lands");
    real_push(&client, &url_setup, Vec::new(), &["main"])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();

    git(Some(client.path()), &["checkout", "-q", "-b", "old"])
        .await
        .expect("the run rides a resolved git binary")
        .expect_ok();
    let tip_old = commit_fixture_file(&client, "the old branch's own content\n", "old's commit")
        .await
        .expect("old's commit lands");
    let old_blob = rev_parse(&client, "HEAD:a.txt")
        .await
        .expect("old's blob resolves");
    real_push(&client, &url_setup, Vec::new(), &["old"])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    git(Some(client.path()), &["checkout", "-q", "main"])
        .await
        .expect("the run rides a resolved git binary")
        .expect_ok();

    let delete = real_push(&client, &url_delete, Vec::new(), &[":refs/heads/old"])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    assert!(
        delete.stderr().contains("[deleted]"),
        "the real client reported the deletion: {}",
        delete.stderr()
    );

    assert!(
        extract_and_walk_client_pack(&delete_capture.to_server()).is_none(),
        "the delete-only push sent no pack bytes — the wire shape the \
         bounded peek classifies (captured live: the POC's packless fact)"
    );
    let report = sideband_report_lines(&delete_capture.to_client());
    assert_eq!(
        report,
        vec!["unpack ok".to_owned(), "ok refs/heads/old".to_owned()],
        "the packless push's report frames per ADR-013 §8"
    );

    assert_unborn_head_row_persists(&harness).await;
    let tips = listed_tips(&harness).await;
    assert!(
        tips.iter().all(|(name, _)| name != "refs/heads/old"),
        "the delete's CAS removed the ref row"
    );
    assert_eq!(
        tips.iter()
            .find(|(name, _)| name == "refs/heads/main")
            .and_then(|(_, oid)| *oid),
        Some(gix_hash::ObjectId::from_hex(tip_main.as_bytes()).expect("the tip oid parses")),
        "main held"
    );
    assert_eq!(
        harness
            .assembly
            .reactor()
            .registry()
            .watermark()
            .await
            .expect("the watermark reads"),
        4,
        "the deletion ride's the marker legs like a pushed pack's commit \
         (1 = the put, 2/3 = the pushes, 4 = the delete)"
    );

    let report = harness
        .assembly
        .blobs()
        .sweep()
        .await
        .expect("the sweep runs");
    assert!(report.aborted.is_none());
    assert!(
        report.deleted >= 1,
        "the deleted ref's objects lost their coverage — the sweep reclaimed them"
    );
    let old_tip_oid = gix_hash::ObjectId::from_hex(tip_old.as_bytes()).expect("the tip oid parses");
    let old_blob_oid =
        gix_hash::ObjectId::from_hex(old_blob.as_bytes()).expect("the blob oid parses");
    for oid in [old_tip_oid, old_blob_oid] {
        assert!(
            harness
                .assembly
                .front()
                .stat(&oid)
                .await
                .expect("the front stats")
                .is_none(),
            "the released refs' object reclaimed"
        );
    }
    let main_commit = rev_parse(&client, "main")
        .await
        .expect("main's tip resolves");
    let main_oid = gix_hash::ObjectId::from_hex(main_commit.as_bytes()).expect("the oid parses");
    assert!(
        harness
            .assembly
            .front()
            .stat(&main_oid)
            .await
            .expect("the front stats")
            .is_some(),
        "main's objects remain covered"
    );
    assert_work_area_clean(&harness).await;

    door_setup.accept.abort();
    door_delete.accept.abort();
    std::fs::remove_dir_all(&base).ok();
    drop(harness);
}

/// A real atomic push clashing at the CAS: the deterministic clashing-
/// client orchestration — the clashing client's session is held at the
/// door (its advertisement was served, its request transmitted — the
/// pack rides the hold), a plain client's push commits `main`'s advance
/// meanwhile, and the hold then releases the clashing request: the
/// one-transaction CAS takes it whole, the stale `main` update is
/// denied `stale info`, the valid `feature` create reports `atomic push
/// failure` (the captured upstream pair, ADR-013 §10), nothing applied
/// (the push rolled back), the aborted CAS rides no legs (the watermark
/// is exactly the committed pushes'), and the pushed pack's objects
/// recover GC-decreed — the following sweep reclaims them.
///
/// Contract stamp: ADR-013 §10 (the atomic abort's outcome matrix —
/// denied refs keep their reasons, the rest report `atomic push
/// failure`; the rollback); ADR-013 §8 (the report's verbatim reasons;
/// the framing); ADR-020 §2 + ADR-019 §7 (the pins' drop-on-abort, the
/// no-legs rule); ADR-002 (the duplex session boundary the hold freezes
/// mid-race); push-captures.md §Status report (the live-captured pair);
/// substrate/w8-e2e-real-git.
#[tokio::test(flavor = "multi_thread")]
async fn a_real_atomic_push_clashed_at_the_cas_rolls_back_whole_and_reports_the_capture_pair() {
    if common::git_command().is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return;
    }
    let harness = Harness::mem("atomic-clash", "owner/atomic").await;
    let base = std::env::temp_dir().join(format!("alkgit-real-git-atomic-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&base);
    std::fs::create_dir_all(&base).expect("the row's base creates");

    let door = common::spawn_daemon_door(&harness, DoorOptions::default()).await;
    let url = door_url("git", door.addr, &harness.record.repo_id);

    let seed = init_client(&base, "seed").await.expect("the seed inits");
    let tip_c1 = commit_fixture_file(&seed, "", "the seed commit")
        .await
        .expect("the seed commit lands");
    real_push(&seed, &url, Vec::new(), &["main"])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();

    let clash_dir = base.join("clash");
    real_clone(&url, &clash_dir)
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    let clash = ClientDir(clash_dir);
    let tip_a = commit_fixture_file(&clash, "the clashing client's first commit\n", "clash-a")
        .await
        .expect("the clash client's commits land");
    let tip_b = commit_fixture_file(&clash, "the clashing client's second commit\n", "clash-b")
        .await
        .expect("the clash client's commits land");
    git(Some(clash.path()), &["branch", "feature", "HEAD^"])
        .await
        .expect("the run rides a resolved git binary")
        .expect_ok();

    let winner_dir = base.join("winner");
    real_clone(&url, &winner_dir)
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    let winner = ClientDir(winner_dir);
    let tip_z = commit_fixture_file(&winner, "the winner's commit\n", "winner")
        .await
        .expect("the winner's commit lands");

    let gate = GateRendezvous::new();
    let clash_capture = common::StreamCapture::new();
    let door_gated = common::spawn_daemon_door(
        &harness,
        DoorOptions {
            capture: Some(clash_capture.clone()),
            gate: Some(gate.clone()),
        },
    )
    .await;
    let url_gated = door_url("git", door_gated.addr, &harness.record.repo_id);
    let clashing = tokio::spawn({
        let dir = clash.path().to_path_buf();
        async move {
            let mut all = vec!["push", "--atomic"];
            all.push(&url_gated);
            all.push("main");
            all.push("feature");
            git_run(Some(&dir), &all).await
        }
    });

    tokio::time::timeout(Duration::from_secs(30), gate.held.notified())
        .await
        .expect("the clashing client's request reached the hold inside the deadline");

    real_push(&winner, &url, Vec::new(), &["main"])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    assert_eq!(
        listed_tips(&harness)
            .await
            .iter()
            .find(|(name, _)| name == "refs/heads/main")
            .and_then(|(_, oid)| *oid),
        Some(gix_hash::ObjectId::from_hex(tip_z.as_bytes()).expect("the winner's tip parses")),
        "the plain client's main advance committed first"
    );

    gate.release();
    let clashing = clashing
        .await
        .expect("the clashing client's task joins")
        .expect("the row rides a resolved git binary");
    assert_ne!(
        clashing.code, 0,
        "the clashing push was denied by the CAS: {}{}",
        clashing.stdout, clashing.stderr
    );
    assert!(
        clashing.stderr.contains("stale info"),
        "the stale ref reports the server-chosen reason: {}",
        clashing.stderr
    );
    assert!(
        clashing.stderr.contains("atomic push failure"),
        "the would-be-successful ref reports the atomic-abort reason: {}",
        clashing.stderr
    );
    let report = sideband_report_lines(&clash_capture.to_client());
    assert_eq!(
        report
            .iter()
            .filter(|line| **line != "unpack ok")
            .map(String::as_str)
            .collect::<std::collections::BTreeSet<&str>>(),
        [
            "ng refs/heads/main stale info",
            "ng refs/heads/feature atomic push failure"
        ]
        .into_iter()
        .collect(),
        "the report carries the capture pair verbatim: {:?}",
        report
    );

    let tips = listed_tips(&harness).await;
    assert!(
        tips.iter().all(|(name, _)| name != "refs/heads/feature"),
        "the atomic rollback applied nothing — feature's create never landed"
    );
    assert_eq!(
        harness
            .assembly
            .reactor()
            .registry()
            .watermark()
            .await
            .expect("the watermark reads"),
        3,
        "the aborted CAS rides no legs: 1 = the put, 2/3 = the two committed \
         pushes' markers"
    );

    let introduced = introduced_oids(&clash, &tip_b, &tip_c1)
        .await
        .expect("the clashing client's introduced set resolves");
    let report = harness
        .assembly
        .blobs()
        .sweep()
        .await
        .expect("the sweep runs");
    assert!(report.aborted.is_none());
    assert!(
        report.deleted >= 1,
        "the dropped guards' objects are orphans — the sweep reclaims them"
    );
    let tip_a_oid =
        gix_hash::ObjectId::from_hex(tip_a.as_bytes()).expect("the clash tip oid parses");
    for oid in introduced.iter().chain(std::iter::once(&tip_a_oid)) {
        assert!(
            harness
                .assembly
                .front()
                .stat(oid)
                .await
                .expect("the front stats")
                .is_none(),
            "the rollback's pack objects recovered GC-decreed: {oid}"
        );
    }
    let winner_oid =
        gix_hash::ObjectId::from_hex(tip_z.as_bytes()).expect("the winner's tip parses");
    assert!(
        harness
            .assembly
            .front()
            .stat(&winner_oid)
            .await
            .expect("the front stats")
            .is_some(),
        "the winner's objects stay covered"
    );
    assert_work_area_clean(&harness).await;

    door.accept.abort();
    door_gated.accept.abort();
    std::fs::remove_dir_all(&base).ok();
    drop(harness);
}

/// A real `git clone` then `git fetch` complete over the duplex
/// git-daemon door on the substrate-backed server: the clone's V2
/// handshake (the daemon request line's `version=2`), the ls-refs
/// round's default-branch resolution (the unborn-HEAD symref row
/// resolves the client's HEAD onto `refs/heads/main`), the done-round
/// packfile section riding the substrate packgen's generation, and the
/// client's own pack ingestion validating the served stream — `git
/// fsck --strict` clean, the pushed commit checked out, the fixture
/// file's exact bytes at the working tree. The follow-up configured
/// fetch negotiates against the substrate Find's `common_haves` (the
/// new tip's pack is the closure minus the recognized haves) and
/// advances the tracking ref.
///
/// Contract stamp: ADR-005 §1 (the duplex session the framed transport
/// drives); ADR-003 (the V2-only fetch); ADR-014 (the negotiation
/// loop over the real Find); ADR-019 (the Find front the generation +
/// `common_haves` ride); ADR-018 §5 (the unborn-HEAD symref row the
/// default-branch resolution consumes); doors.md §git:// (the daemon
/// framing); poco-findings (the clone gate re-landed);
/// substrate/w8-e2e-real-git.
#[tokio::test(flavor = "multi_thread")]
async fn a_real_git_clone_and_fetch_complete_over_the_duplex_door_with_the_pushed_commit_checked_out(
) {
    if common::git_command().is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return;
    }
    let harness = Harness::mem("clone-fetch-duplex", DEFAULT_REPO_ID).await;
    let base = std::env::temp_dir().join(format!(
        "alkgit-real-git-clone-duplex-{}",
        std::process::id()
    ));
    let _ = std::fs::remove_dir_all(&base);
    std::fs::create_dir_all(&base).expect("the row's base creates");

    let door = common::spawn_daemon_door(&harness, DoorOptions::default()).await;
    let url = door_url("git", door.addr, &harness.record.repo_id);

    let seed = init_client(&base, "seed").await.expect("the seed inits");
    let tip_c1 = commit_fixture_file(&seed, "", "the seed commit")
        .await
        .expect("the seed commit lands");
    real_push(&seed, &url, Vec::new(), &["main"])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();

    let dst = base.join("clone");
    real_clone(&url, &dst)
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    assert_clone_checked_out(&dst, &tip_c1, &file_content("")).await;

    let tip_c2 = commit_fixture_file(&seed, "the fetch's new tip\n", "the advanced commit")
        .await
        .expect("the advanced commit lands");
    real_push(&seed, &url, Vec::new(), &["main"])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();

    real_fetch(&dst)
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    let fetched = rev_parse(&ClientDir(dst.clone()), "refs/remotes/origin/main")
        .await
        .expect("the tracking ref resolves");
    assert_eq!(
        fetched, tip_c2,
        "the real fetch negotiated and advanced the tracking ref"
    );
    assert_work_area_clean(&harness).await;

    door.accept.abort();
    std::fs::remove_dir_all(&base).ok();
    drop(harness);
}

/// The same real clone + fetch sessions over the smart-http-framed
/// stateless door (the core's service-stub shape): the GET
/// advertisement (the smart prefix, the anonymous public read), the
/// one-POST-per-round invocations riding the substrate packgen, the
/// pushed commit checked out with `git fsck --strict` clean, and the
/// follow-up configured fetch advancing the tracking ref.
///
/// Contract stamp: ADR-005 §2 (the stateless request/response rules the
/// real client drives); ADR-005 §3 (one machine core over two
/// substrates); ADR-003; ADR-014; ADR-011 §3 (the anonymous public
/// read the http door carries); doors.md §smart-http (the
/// service-stub shape); poco-findings; substrate/w8-e2e-real-git.
#[tokio::test(flavor = "multi_thread")]
async fn the_real_clone_and_fetch_sessions_complete_over_the_stateless_http_door() {
    if common::git_command().is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return;
    }
    let harness = Harness::mem("clone-fetch-http", DEFAULT_REPO_ID).await;
    let base =
        std::env::temp_dir().join(format!("alkgit-real-git-clone-http-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&base);
    std::fs::create_dir_all(&base).expect("the row's base creates");

    let door = common::spawn_daemon_door(&harness, DoorOptions::default()).await;
    let door_http = common::spawn_http_door(&harness).await;
    let url_git = door_url("git", door.addr, &harness.record.repo_id);
    let url_http = door_url("http", door_http.addr, &harness.record.repo_id);

    let seed = init_client(&base, "seed").await.expect("the seed inits");
    let tip_c1 = commit_fixture_file(&seed, "", "the seed commit")
        .await
        .expect("the seed commit lands");
    real_push(&seed, &url_git, Vec::new(), &["main"])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();

    let dst = base.join("clone");
    real_clone(&url_http, &dst)
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    assert_clone_checked_out(&dst, &tip_c1, &file_content("")).await;

    let tip_c2 = commit_fixture_file(&seed, "the http fetch's new tip\n", "the advanced commit")
        .await
        .expect("the advanced commit lands");
    real_push(&seed, &url_git, Vec::new(), &["main"])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();

    real_fetch(&dst)
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    let fetched = rev_parse(&ClientDir(dst.clone()), "refs/remotes/origin/main")
        .await
        .expect("the tracking ref resolves");
    assert_eq!(
        fetched, tip_c2,
        "the real fetch advanced the tracking ref over the stateless door"
    );
    assert_work_area_clean(&harness).await;

    door.accept.abort();
    door_http.accept.abort();
    std::fs::remove_dir_all(&base).ok();
    drop(harness);
}

/// The default engine's durable shape: the sqlite assembly (the default
/// single-host shape — one co-tenant file for the registry/ref tables,
/// the alkstore engine and the object store) wired with the
/// through-store pack serving, serving a real push (the duplex door;
/// the refs CAS rides the co-tenant rows) and a real clone + fetch over
/// the http door — the fetch's pack served in the through-store shape
/// (put once, ranged windows) behind the real client's standard
/// handshake, `git fsck --strict` clean, the pushed commit checked out.
///
/// Contract stamp: substrate.md §assembly shapes (the sqlite row; the
/// through-store column's engine requires the large tier — the
/// co-tenant file's object store carries it); ADR-019 §2 (the refs
/// rows' durable CAS home); ADR-020 §2 (the ingestion's unpack-to-loose
/// over the durable object store); substrate.md §object storage "Packs"
/// (both serving shapes validated — the through-store shape's row rides
/// a real client here; the wave-10 assembly exercises are the
/// full-matrix round); ADR-009 (the pinned budgets unchanged);
/// substrate/w8-e2e-real-git (the in-task pinned through-store
/// decision).
#[cfg(feature = "sqlite")]
#[tokio::test(flavor = "multi_thread")]
async fn the_sqlite_assembly_serves_a_real_push_and_a_real_clone_through_store() {
    if common::git_command().is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return;
    }
    let harness = Harness::sqlite(
        "sqlite-through-store",
        DEFAULT_REPO_ID,
        PackServing::ThroughStore,
    )
    .await;
    let base = std::env::temp_dir().join(format!("alkgit-real-git-sqlite-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&base);
    std::fs::create_dir_all(&base).expect("the row's base creates");

    let door = common::spawn_daemon_door(&harness, DoorOptions::default()).await;
    let door_http = common::spawn_http_door(&harness).await;
    let url_git = door_url("git", door.addr, &harness.record.repo_id);
    let url_http = door_url("http", door_http.addr, &harness.record.repo_id);

    let seed = init_client(&base, "seed").await.expect("the seed inits");
    let tip_c1 = commit_fixture_file(&seed, "", "the seed commit")
        .await
        .expect("the seed commit lands");
    real_push(&seed, &url_git, Vec::new(), &["main"])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    assert_unborn_head_row_persists(&harness).await;
    let tips = listed_tips(&harness).await;
    assert_eq!(
        tips.iter()
            .find(|(name, _)| name == "refs/heads/main")
            .and_then(|(_, oid)| *oid),
        Some(gix_hash::ObjectId::from_hex(tip_c1.as_bytes()).expect("the tip oid parses")),
        "the refs CAS applied over the co-tenant rows"
    );

    let dst = base.join("clone");
    real_clone(&url_http, &dst)
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    assert_clone_checked_out(&dst, &tip_c1, &file_content("")).await;

    let tip_c2 = commit_fixture_file(
        &seed,
        "the through-store fetch's new tip\n",
        "the advanced commit",
    )
    .await
    .expect("the advanced commit lands");
    real_push(&seed, &url_git, Vec::new(), &["main"])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    real_fetch(&dst)
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    let fetched = rev_parse(&ClientDir(dst.clone()), "refs/remotes/origin/main")
        .await
        .expect("the tracking ref resolves");
    assert_eq!(
        fetched, tip_c2,
        "the http fetch advanced through the through-store serving shape"
    );
    assert_work_area_clean(&harness).await;

    door.accept.abort();
    door_http.accept.abort();
    std::fs::remove_dir_all(&base).ok();
    drop(harness);
}

/// The chain backstop at the end-to-end level: after a real push (over
/// the duplex door), the composed reactive state closes the loop — the
/// on-demand reconcile pass catches the record row up and anchors the
/// applied watermark at the durable watermark; the maintenance chain's
/// worker drains the push's enqueued pack-housekeeping job; the chain's
/// scheduled sweep runs and its verdict holds the pushed graph covered
/// (liveness closure, deleted = 0).
///
/// Contract stamp: ADR-019 §7 (the reactive halves the push's marker
/// legs feed — the reconcile backstop's crash-window closure);
/// substrate.md §maintenance (the chain's legs: the sweep schedule +
/// supervised runner, the worker's claim loop, the reconcile tick);
/// ADR-018 §6 (the registered closure covering the pushed refs);
/// MaintenanceChain's pins (the sweep verdict as data); substrate/
/// w8-e2e-real-git (the acceptance's chain-smoke row).
#[tokio::test(flavor = "multi_thread")]
async fn the_maintenance_chain_backstops_a_real_pushs_composed_legs() {
    if common::git_command().is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return;
    }
    let harness = Harness::mem_tuned(
        "chain-backstop",
        DEFAULT_REPO_ID,
        MaintenanceOpts {
            sweep_spec: "@every 1s".to_owned(),
            reconcile_tick: Duration::from_secs(10),
            worker_poll: Duration::from_millis(20),
            ..MaintenanceOpts::default()
        },
    )
    .await;
    let base = std::env::temp_dir().join(format!("alkgit-real-git-chain-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&base);
    std::fs::create_dir_all(&base).expect("the row's base creates");

    let door = common::spawn_daemon_door(&harness, DoorOptions::default()).await;
    let url = door_url("git", door.addr, &harness.record.repo_id);
    let seed = init_client(&base, "seed").await.expect("the seed inits");
    let _tip = commit_fixture_file(&seed, "", "the chain row's commit")
        .await
        .expect("the seed commit lands");
    real_push(&seed, &url, Vec::new(), &["main"])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();

    let watermark = harness
        .assembly
        .reactor()
        .registry()
        .watermark()
        .await
        .expect("the watermark reads");

    let applied = harness.assembly.applied_watermark().clone();
    let missed = harness
        .assembly
        .reactor()
        .reconcile(&applied)
        .await
        .expect("the reconcile pass closes the record row");
    assert!(
        !missed.is_empty(),
        "the reconcile caught the record row up (the markers never ride rows_since)"
    );
    assert_eq!(
        applied.get(),
        watermark,
        "the applied watermark anchored at the durable watermark"
    );

    let seen: Arc<Mutex<Vec<WorkerEvent>>> = Arc::new(Mutex::new(Vec::new()));
    let mut sweep_verdict: Option<alkgit_substrate::SweepVerdict> = None;
    let stop = alkstore::StopToken::new();
    let chain_run = harness.assembly.chain().run("e2e-worker", stop.clone(), {
        let seen = seen.clone();
        move |event| seen.lock().push(event)
    });
    tokio::pin!(chain_run);

    let started = tokio::time::Instant::now();
    let mut result: Option<alkstore::Result<()>> = None;
    while result.is_none() {
        let housekeeping = seen
            .lock()
            .iter()
            .any(|event| matches!(event, WorkerEvent::PackHousekeeping(_)));
        sweep_verdict = seen.lock().iter().find_map(|event| match event {
            WorkerEvent::Sweep(verdict) => Some(verdict.clone()),
            _ => None,
        });
        if (housekeeping && sweep_verdict.is_some() || started.elapsed() > Duration::from_secs(10))
            && !stop.is_cancelled()
        {
            stop.cancel();
        }
        tokio::select! {
            outcome = &mut chain_run => result = Some(outcome),
            _ = tokio::time::sleep(Duration::from_millis(50)) => {}
        }
    }
    let done = result.expect("the chain's run reached its end");
    assert!(
        done.is_ok(),
        "the chain ran clean under the cancelled stop token: {done:?}"
    );
    assert!(
        seen.lock()
            .iter()
            .any(|event| matches!(event, WorkerEvent::PackHousekeeping(_))),
        "the real push's enqueued housekeeping job drained: {:?}",
        seen.lock()
    );
    let verdict = sweep_verdict.expect("the chain's scheduled sweep ran");
    match verdict {
        alkgit_substrate::SweepVerdict::Report(report) => {
            assert!(report.aborted.is_none(), "the sweep ran: {report:?}");
            assert_eq!(
                report.deleted, 0,
                "the pushed graph is liveness-covered across the chain's sweep"
            );
        }
        other => panic!("the chain's sweep verdict ran: {other:?}"),
    }

    door.accept.abort();
    std::fs::remove_dir_all(&base).ok();
}