alkgit/substrate/tests/replication.rs

//! The w8 replication loop end-to-end — ADR-023's v1 posture on the
//! real code: two substrate-backed nodes (each its own composition
//! entry, its own stores) with the replicator built in test code over
//! the published pieces — a `GitSession` client plus the registry
//! stream consumer (ADR-023 §1's embedder shape), the source's real
//! machinery underneath. The POC-X `e2e_mirror.ts`-class gates
//! re-landed: real `git push` into node A → the receive legs (the
//! marker watermark + the stream event) → the replicator pass (diff =
//! peer refs via `ls_refs` on B vs the source's ref rows; fetch from A
//! with the have-aware trim; apply = the receive-pack CAS into B) →
//! real `git clone` from B, `fsck --strict` clean, the pushed commit
//! checked out. The recipe's idempotence/tolerance riders (ADR-023 §3):
//! the zero-diff noop, delete propagation via the packless push
//! commands, the suppressed-announce crash window closed by the
//! reconcile tick, and CAS-idempotent re-transfer.
//!
//! The v1 HEAD posture (ADR-023 §4) is asserted as recorded: the loop
//! never carries a HEAD row (the diff is ref rows; the push grammar has
//! no symref command) — each replica's HEAD stays its own seeded unborn
//! symref, and the replica's seed row serves the real clone's
//! default-branch resolution.
//!
//! `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. The replicator itself never spawns git.
//!
//! Rows are names + `Contract stamp:` versions per
//! `substrate/tests/common` §The row/stamp discipline.

mod common;

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

use alkcall::core::BiStream;
use alkgit::{
    FetchRound, GitSession, Limits, LsRefsArgs, PushReport, PushSendOptions, RefLine, RefUpdate,
    Service,
};
use alkgit_substrate::{MaintenanceOpts, RegistryEvent};
use common::{
    extract_and_walk_client_pack, git, parse_client_pack, rev_list_oids, DoorOptions, Harness,
};
use gix_hash::ObjectId;
use tokio::net::TcpStream;

const OP_TIMEOUT: Duration = Duration::from_secs(60);

/// The fixture file's shape: the body plus the commit's marker line
/// (every marker a distinct blob).
fn file_content(marker: &str) -> String {
    format!("the replication fixture body\n{marker}")
}

/// The rows' local git client repo (the w8 real-git harness's client
/// shape: main, repo-local identity).
struct ClientDir(PathBuf);

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

async fn init_client(base: &Path, name: &str) -> 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
    .expect("the row rides a resolved git binary")
    .expect_ok();
    git(
        Some(&dir),
        &["config", "user.name", "alkgit replication e2e"],
    )
    .await
    .expect("the run rides a resolved git binary")
    .expect_ok();
    git(Some(&dir), &["config", "user.email", "e2e@alk.dev"])
        .await
        .expect("the run rides a resolved git binary")
        .expect_ok();
    ClientDir(dir)
}

/// Writes the fixture file at `marker` and commits; returns the new
/// tip's hex oid.
async fn commit_file(client: &ClientDir, marker: &str, message: &str) -> 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
        .expect("the run rides a resolved git binary")
        .expect_ok();
    git(Some(client.path()), &["commit", "-q", "-m", message])
        .await
        .expect("the run rides a resolved git binary")
        .expect_ok();
    rev_parse(client, "HEAD").await
}

/// A plain `rev-parse` (hex, trimmed).
async fn rev_parse(client: &ClientDir, what: &str) -> String {
    git(Some(client.path()), &["rev-parse", what])
        .await
        .expect("the run rides a resolved git binary")
        .expect_ok()
        .stdout
        .trim()
        .to_owned()
}

/// `rev-list --objects` over a range (`"tip ^base"` split into args),
/// as an oid set.
async fn rev_list_objects(client: &ClientDir, range: &[&str]) -> BTreeSet<ObjectId> {
    let mut args = vec!["rev-list", "--objects"];
    args.extend_from_slice(range);
    let run = git(Some(client.path()), &args)
        .await
        .expect("the run rides a resolved git binary")
        .expect_ok();
    rev_list_oids(&run.stdout)
}

async fn git_simple(client: &ClientDir, args: &[&str]) {
    git(Some(client.path()), args)
        .await
        .expect("the run rides a resolved git binary")
        .expect_ok();
}

/// Real `git push <url>` with extra args prefix (delete refspecs,
/// force flags — the client computes; the rows' pushes are ordinary or
/// delete-only).
async fn real_push(client: &ClientDir, url: &str, args: &[&str], refspecs: &[&str]) {
    let mut all = vec!["push"];
    all.extend_from_slice(args);
    all.push(url);
    for refspec in refspecs {
        all.push(refspec);
    }
    git(Some(client.path()), &all)
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
}

/// Real `git clone <url> <dst>` (V2).
async fn real_clone(url: &str, dst: &Path) {
    git(
        None,
        &[
            "-c",
            "protocol.version=2",
            "clone",
            "-q",
            url,
            dst.to_str().expect("utf8 dst"),
        ],
    )
    .await
    .expect("the row rides a resolved git binary")
    .expect_ok();
}

/// Real `git fsck --strict` inside a checkout.
async fn fsck_strict(dst: &Path) {
    git(Some(dst), &["fsck", "--strict"])
        .await
        .expect("the run rides a resolved git binary")
        .expect_ok();
}

/// The clone's post-state: HEAD resolves to the expected tip, the
/// worktree file's bytes are the expected content, and `git fsck
/// --strict` is clean.
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;
    assert_eq!(head, expected_tip, "the clone checked out the expected tip");
    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 replicated content"
    );
    fsck_strict(dst).await;
}

fn door_url(addr: SocketAddr, repo_id: &str) -> String {
    format!("git://{addr}/{repo_id}")
}

/// A row's base directory (unique per process; removed at the row's
/// end).
fn row_base(tag: &str) -> PathBuf {
    let dir = std::env::temp_dir().join(format!(
        "alkgit-substrate-replication-{tag}-{}",
        std::process::id()
    ));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).expect("the row base creates");
    dir
}

/// A node's ref rows as an oid map. The unborn HEAD seed row carries
/// no oid and never enters anything derived from this (the diff is
/// over ref rows — ADR-023 §4's no-HEAD-propagation posture).
async fn refs_map(harness: &Harness) -> BTreeMap<String, ObjectId> {
    harness
        .assembly
        .refs()
        .list_refs(&harness.record)
        .await
        .expect("the ref rows list")
        .into_iter()
        .filter_map(|line: RefLine| line.oid.map(|oid| (line.name, oid)))
        .collect()
}

async fn node_watermark(harness: &Harness) -> i64 {
    harness
        .assembly
        .reactor()
        .registry()
        .watermark()
        .await
        .expect("the watermark reads")
}

// ——— the replicator (test-level; ADR-023 §1's embedder shape) ———

/// One replicator pass's outcome (the rows' observation data).
#[derive(Debug, Default)]
struct ReplicateOutcome {
    /// The zero-diff noop (no fetch, no push — the recipe's
    /// idempotence).
    noop: bool,
    /// The rounds' positive acks (the commons the peer's known closure
    /// produced — the have-aware trim's recognition leg), deduped and
    /// ordered.
    fetch_acks: Vec<ObjectId>,
    /// The done round's fetched pack bytes (sideband-unwrapped);
    /// `None` when the diff introduced nothing.
    fetched_pack: Option<Vec<u8>>,
    /// The status report the replica's receive-pack machine emitted.
    push_report: Option<PushReport>,
}

/// Dials one daemon door as a direct-mode `GitSession` (the request
/// line rides the handshake — the replicator's transport is the same
/// shape the doors serve; one connection = one service machine).
async fn dial(addr: SocketAddr, repo_id: &str, service: Service) -> GitSession {
    let conn = TcpStream::connect(addr)
        .await
        .expect("the replicator's dial connects");
    let (read, write) = tokio::io::split(conn);
    GitSession::connect_direct(
        BiStream::from_joined(read, write),
        repo_id,
        service,
        Some("replicator"),
        Limits::default(),
    )
    .expect("the session constructs")
}

/// The peer's ref tips via `ls_refs` over the replica's door (the
/// diff's peer-ref leg, ADR-023 §1). The unborn HEAD row is not
/// requested (no `unborn` argument, `refs/` prefixes only) — the
/// symmetric no-HEAD shape of the diff.
async fn peer_tips(addr: SocketAddr, repo_id: &str) -> BTreeMap<String, ObjectId> {
    let mut session = dial(addr, repo_id, Service::UploadPack).await;
    let listed = tokio::time::timeout(
        OP_TIMEOUT,
        session.ls_refs(&LsRefsArgs {
            ref_prefixes: vec!["refs/".to_owned()],
            ..LsRefsArgs::default()
        }),
    )
    .await
    .expect("the ls-refs answers inside the deadline")
    .expect("the peer's ref list parses");
    listed
        .into_iter()
        .filter_map(|line| line.oid.map(|oid| (line.name, oid)))
        .collect()
}

/// **One replication pass** (the test's replicator, ADR-023 §1's
/// recipe, driven as a diff of the two nodes' ref rows):
///
/// 1. diff = the source's ref rows vs the peer's `ls_refs` tips
///    (creates / updates / deletes); the zero diff is a noop by
///    construction (no fetch, no push);
/// 2. transfer = a `GitSession` fetch from the source — the peer's
///    known tips ride the rounds' haves (the have-aware trim through
///    the source's Find/common_haves: the boundary closure the peer
///    holds is never re-sent);
/// 3. apply = a `GitSession` push into the replica — the CAS old
///    values from the replica's own V0 advertisement, creates/updates
///    carrying the fetched pack, deletes riding the packless commands
///    (ADR-020 §3's client-side rule).
///
/// A stale CAS between the diff and the apply is business-as-usual
/// (the recipe): the report carries the denial and the next pass
/// re-diffs — the CAS rider constructs that window deliberately.
async fn replicate_pass(
    source: &Harness,
    source_door: SocketAddr,
    replica: &Harness,
    replica_door: SocketAddr,
) -> ReplicateOutcome {
    let source_map = refs_map(source).await;
    let peer = peer_tips(replica_door, &replica.record.repo_id).await;

    let creates: Vec<(String, ObjectId)> = source_map
        .iter()
        .filter(|(name, _)| !peer.contains_key(*name))
        .map(|(name, oid)| (name.clone(), *oid))
        .collect();
    let updates: Vec<(String, ObjectId, ObjectId)> = source_map
        .iter()
        .filter_map(|(name, source_oid)| {
            let peer_oid = peer.get(name).copied()?;
            (peer_oid != *source_oid).then(|| (name.clone(), peer_oid, *source_oid))
        })
        .collect();
    let deletes: Vec<String> = peer
        .keys()
        .filter(|name| !source_map.contains_key(*name))
        .cloned()
        .collect();

    if creates.is_empty() && updates.is_empty() && deletes.is_empty() {
        return ReplicateOutcome {
            noop: true,
            ..ReplicateOutcome::default()
        };
    }

    let mut outcome = ReplicateOutcome::default();

    // ——— the transfer leg: fetch the source (an object-introducing
    // diff only) ———
    if !creates.is_empty() || !updates.is_empty() {
        let wants: Vec<ObjectId> = creates
            .iter()
            .map(|(_, oid)| *oid)
            .chain(updates.iter().map(|(_, _, oid)| *oid))
            .collect();
        let haves: Vec<ObjectId> = peer.values().copied().collect();
        let mut session = dial(source_door, &source.record.repo_id, Service::UploadPack).await;
        let rounds = if haves.is_empty() {
            vec![FetchRound {
                wants,
                haves: Vec::new(),
                done: true,
            }]
        } else {
            // The no-done round probes the commons; the done round
            // re-sends the same complete have view (stateless rounds —
            // the caller re-sends its commons every round; the
            // source's acks are a subset of the haves it carries).
            vec![
                FetchRound {
                    wants: wants.clone(),
                    haves: haves.clone(),
                    done: false,
                },
                FetchRound {
                    wants,
                    haves,
                    done: true,
                },
            ]
        };
        let mut pack = Vec::new();
        let reports = tokio::time::timeout(OP_TIMEOUT, session.fetch(rounds, &mut pack))
            .await
            .expect("the fetch answers inside the deadline")
            .expect("the fetch round plan runs");
        outcome.fetch_acks = reports
            .iter()
            .flat_map(|report| report.acks.iter().copied())
            .collect::<BTreeSet<_>>()
            .into_iter()
            .collect();
        assert!(
            !pack.is_empty(),
            "an object-introducing diff's fetch delivered its pack"
        );
        outcome.fetched_pack = Some(pack);
    }

    // ——— the apply leg: the CAS into the replica ———
    let mut session = dial(replica_door, &replica.record.repo_id, Service::ReceivePack).await;
    let advertised: BTreeMap<String, ObjectId> =
        tokio::time::timeout(OP_TIMEOUT, session.push_advertisement())
            .await
            .expect("the push advertisement answers inside the deadline")
            .expect("the replica's V0 advertisement parses")
            .into_iter()
            .map(|row| (row.name, row.oid))
            .collect();

    let mut commands: Vec<RefUpdate> = Vec::new();
    for (name, new) in &creates {
        commands.push(RefUpdate {
            name: name.clone(),
            expected: None,
            new: Some(*new),
        });
    }
    for (name, _old_view, new) in &updates {
        commands.push(RefUpdate {
            name: name.clone(),
            expected: advertised.get(name).copied(),
            new: Some(*new),
        });
    }
    for name in &deletes {
        commands.push(RefUpdate {
            name: name.clone(),
            expected: advertised.get(name).copied(),
            new: None,
        });
    }

    let report = tokio::time::timeout(OP_TIMEOUT, async {
        match outcome.fetched_pack.as_ref() {
            Some(pack) => {
                session
                    .push(commands, Some(&mut &pack[..]), &PushSendOptions::default())
                    .await
            }
            None => {
                session
                    .push(commands, None::<&mut &[u8]>, &PushSendOptions::default())
                    .await
            }
        }
    })
    .await
    .expect("the push answers inside the deadline")
    .expect("the push's report parses");
    assert_eq!(
        report.unpack,
        Ok(()),
        "the replicator's push unpacked — the fetched trim's pack ingested, or the \
         packless shape's no-pack leg"
    );
    outcome.push_report = Some(report);
    outcome
}

/// The registry stream consumer (ADR-023 §1's second half): observed
/// events accumulate in the row's book, the composition's applied
/// watermark advancing with each observation.
fn spawn_stream_consumer(
    harness: &Harness,
    stop: alkstore::StopToken,
) -> (
    Arc<Mutex<Vec<RegistryEvent>>>,
    tokio::task::JoinHandle<alkstore::Result<()>>,
) {
    let seen: Arc<Mutex<Vec<RegistryEvent>>> = Arc::new(Mutex::new(Vec::new()));
    let applied = harness.assembly.applied_watermark().clone();
    let reactor = harness.assembly.reactor().clone();
    let seen_probe = Arc::clone(&seen);
    let joiner = tokio::spawn(async move {
        reactor
            .run_stream_consumer("replicator-test", &applied, stop, move |event, _raw| {
                seen_probe.lock().expect("the consumer's book").push(event)
            })
            .await
    });
    (seen, joiner)
}

/// Waits until the stream consumer observed `count` Mutation markers.
async fn wait_mutations(seen: &Arc<Mutex<Vec<RegistryEvent>>>, count: usize) {
    let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
    loop {
        let mutations = seen
            .lock()
            .expect("the consumer's book")
            .iter()
            .filter(|event| matches!(event, RegistryEvent::Mutation { .. }))
            .count();
        if mutations >= count {
            return;
        }
        assert!(
            tokio::time::Instant::now() < deadline,
            "the stream consumer observed {count} mutation markers inside the deadline \
             (observed {mutations})"
        );
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
}

/// Decodes a fetched pack's shape (the rows' pack-decode helper).
struct FetchedPack {
    object_count: usize,
    full_oids: BTreeSet<ObjectId>,
}

fn decode_fetched_pack(pack: &[u8]) -> FetchedPack {
    let decoded = parse_client_pack(pack).expect("the fetched pack decodes");
    FetchedPack {
        object_count: decoded.object_count as usize,
        full_oids: decoded.full_oids.iter().copied().collect(),
    }
}

// ——— the rows ———

/// The two-node loop converges on real code: real `git push` into node
/// A (the receive-pack legs fire — the marker watermark advances and
/// the registry stream consumer observes the Mutation event), the
/// replicator pass replicates onto node B (each its own composition
/// entry, its own stores), a real `git clone` from B is `fsck
/// --strict`-clean with the pushed commit checked out; a second real
/// push into A re-runs the pass with B's tip as the have — the fetch
/// acks the peer's known closure and the fetched pack is exactly the
/// introduced set (the have-aware trim through the source's
/// Find/common_haves, POC-X's 16-object → 3-object shape), the fresh
/// clone from B checks out the new tip; the v1 HEAD posture holds —
/// the loop never carries a HEAD row and B's own seeded unborn symref
/// serves the clone's default-branch resolution; a zero-diff re-run of
/// the pass is the recipe's noop (no fetch, no push — neither node's
/// watermark moves).
///
/// Contract stamp: ADR-023 §1 (the v1 posture: GitSession client +
/// registry stream consumer; single-direction, source authoritative)
/// §3 (the recipe: diff-driven passes, the zero diff ⇒ zero work) §4
/// (the HEAD posture — the recorded limitation, test-pinned); ADR-017
/// §1/§3 (the fetch/ls-refs client, the push client with the CAS old
/// values from the advertisement); ADR-014 §2/§7 (the done round's
/// haves are the complete boundary view — the trim's payoff;
/// stateless rounds); ADR-013 §7/§8 (the CAS receives at the replica,
/// the per-ref outcomes); ADR-019 §7 (the source's marker legs the
/// consumer observes); substrate/w8-e2e-replication.
#[tokio::test(flavor = "multi_thread")]
async fn the_two_node_loop_converges_real_push_into_a_replicate_pass_and_real_clone_from_b() {
    if common::git_command().is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return;
    }
    let source = Harness::mem("rep-source", "owner/rep-source").await;
    let replica = Harness::mem("rep-replica", "owner/rep-replica").await;
    let source_door = common::spawn_daemon_door(&source, DoorOptions::default()).await;
    let replica_door = common::spawn_daemon_door(&replica, DoorOptions::default()).await;
    let base = row_base("gate");
    let source_url = door_url(source_door.addr, &source.record.repo_id);
    let replica_url = door_url(replica_door.addr, &replica.record.repo_id);

    let stop = alkstore::StopToken::new();
    let (consumer, consumer_task) = spawn_stream_consumer(&source, stop.clone());

    let client = init_client(&base, "writer").await;
    let tip_c1 = commit_file(&client, "", "the first replication commit").await;
    let tip_c1_oid = ObjectId::from_hex(tip_c1.as_bytes()).expect("the tip oid parses");
    real_push(&client, &source_url, &[], &["main"]).await;

    wait_mutations(&consumer, 1).await;
    let watermark_after_first = node_watermark(&source).await;
    assert_eq!(
        watermark_after_first, 2,
        "the source's durable watermark carried the put (1) + the push's Mutation marker (2)"
    );
    assert_eq!(
        source.assembly.applied_watermark().get(),
        watermark_after_first,
        "the stream consumer applied the event leg (the watermark advanced with it)"
    );

    let pass = replicate_pass(&source, source_door.addr, &replica, replica_door.addr).await;
    assert!(!pass.noop, "the first pass had a diff to carry");
    let fetched = decode_fetched_pack(&pass.fetched_pack.expect("the first pass fetched"));
    let introduced = rev_list_objects(&client, &[&tip_c1]).await;
    assert!(
        fetched.full_oids.iter().all(|oid| introduced.contains(oid)),
        "the first pass's pack introduces only its closure's objects (B had nothing)"
    );
    assert_eq!(
        fetched.object_count,
        introduced.len(),
        "the first pass's pack is the introduced closure"
    );

    let replica_rows = refs_map(&replica).await;
    assert_eq!(
        replica_rows.get("refs/heads/main"),
        Some(&tip_c1_oid),
        "the replica's main carried the source's tip (the CAS applied)"
    );
    assert_eq!(
        node_watermark(&replica).await,
        2,
        "the replica's own reactive legs fired for the applied CAS (put 1 + its marker 2)"
    );
    assert_eq!(
        pass.push_report.expect("the pass pushed").refs,
        vec![alkgit::RefOutcome {
            name: "refs/heads/main".to_owned(),
            result: Ok(()),
        }],
    );

    let clone_a = base.join("clone-first");
    real_clone(&replica_url, &clone_a).await;
    assert_clone_checked_out(&clone_a, &tip_c1, &file_content("")).await;

    // The second push: the pass re-runs with the peer's known tip as
    // the have — the fetch acks it and the pack is the trimmed delta.
    let tip_c2 = commit_file(
        &client,
        "the second commit's line\n",
        "the replicated advance",
    )
    .await;
    let tip_c2_oid = ObjectId::from_hex(tip_c2.as_bytes()).expect("the tip oid parses");
    real_push(&client, &source_url, &[], &["main"]).await;
    wait_mutations(&consumer, 2).await;

    let pass = replicate_pass(&source, source_door.addr, &replica, replica_door.addr).await;
    assert_eq!(
        pass.fetch_acks,
        vec![tip_c1_oid],
        "the peer's known closure acked — the have-aware trim's recognition leg"
    );
    let trimmed = decode_fetched_pack(&pass.fetched_pack.expect("the second pass fetched"));
    let introduced = rev_list_objects(&client, &[&tip_c2, &format!("^{}", tip_c1)]).await;
    let full_closure = rev_list_objects(&client, &[&tip_c2]).await;
    assert!(
        full_closure.len() > introduced.len(),
        "the untrimmed closure is larger — the trim is observable"
    );
    assert!(
        trimmed.full_oids.iter().all(|oid| introduced.contains(oid)),
        "the fetched pack's full entries are the introduced set (closure(c2) − closure(c1))"
    );
    assert_eq!(
        trimmed.object_count,
        introduced.len(),
        "the pack's object count is the trimmed delta — the peer's closure is never \
         re-sent (the POC-X measured trim)"
    );
    assert_eq!(
        refs_map(&replica).await.get("refs/heads/main"),
        Some(&tip_c2_oid),
        "the replica's main advanced to the second tip"
    );

    let clone_b = base.join("clone-second");
    real_clone(&replica_url, &clone_b).await;
    assert_clone_checked_out(
        &clone_b,
        &tip_c2,
        &file_content("the second commit's line\n"),
    )
    .await;

    // The v1 HEAD posture: neither node's HEAD row ever rides the loop
    // (the diff is ref rows; the push grammar has no symref command) —
    // each side's HEAD stays its own put_repo-seeded unborn symref
    // (→ refs/heads/main), and B's seed row served the clone's
    // default-branch resolution above.
    for (why, harness) in [("the source", &source), ("the replica", &replica)] {
        let head = harness
            .assembly
            .refs()
            .list_refs(&harness.record)
            .await
            .expect("the ref rows list")
            .into_iter()
            .find(|line| line.name == "HEAD")
            .expect("the seeded HEAD row persists");
        assert!(
            head.oid.is_none() && head.symref_target.as_deref() == Some("refs/heads/main"),
            "{why}'s HEAD row is its own seeded unborn symref — never carried by the loop: \
             {head:?}"
        );
    }

    // The zero-diff noop: a re-run with nothing changed does nothing.
    let replica_watermark_before = node_watermark(&replica).await;
    let source_watermark_before = node_watermark(&source).await;
    let pass = replicate_pass(&source, source_door.addr, &replica, replica_door.addr).await;
    assert!(
        pass.noop,
        "the re-run with a zero diff is a noop by construction"
    );
    assert!(
        pass.fetched_pack.is_none() && pass.push_report.is_none(),
        "the noop fetched and pushed nothing"
    );
    assert_eq!(
        node_watermark(&source).await,
        source_watermark_before,
        "the noop's reads left the source's watermark untouched"
    );
    assert_eq!(
        node_watermark(&replica).await,
        replica_watermark_before,
        "the noop applied no CAS — the replica's marker legs never fired"
    );

    stop.cancel();
    consumer_task
        .await
        .expect("the consumer joins")
        .expect("the consumer ran clean");
    source_door.accept.abort();
    replica_door.accept.abort();
    std::fs::remove_dir_all(&base).ok();
}

/// The suppressed-announce crash window (the recipe's tolerance rider,
/// POC-R's recipe with the wave-7 chain): the source's durable commit
/// moved (the real push landed and its marker legs ran) while the
/// consumer's bookmark — the composition's applied watermark — never
/// moved (no stream consumer attached: the crash-window shape where
/// the event leg's delivery is lost); the maintenance chain's
/// reconcile tick drives `rows_since(bookmark)` and anchors the
/// applied watermark at the durable watermark — the loss closure —
/// and the replicator pass then converges the replica.
///
/// Contract stamp: ADR-023 §3 (the event leg as latency only and the
/// reconcile tick as the loss closure — the watermark movement drives,
/// no event application); ADR-019 §7 (the reactive halves' fail-soft
/// posture: a suppressed announce carries no error path, the sweep
/// closes it); substrate.md §maintenance (the chain's reconcile tick);
/// ADR-023 §1 (the pass's convergence); substrate/w8-e2e-replication.
#[tokio::test(flavor = "multi_thread")]
async fn the_reconcile_tick_closes_the_suppressed_announce_window_and_the_pass_converges() {
    if common::git_command().is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return;
    }
    let source = Harness::mem_tuned(
        "rep-reconcile",
        "owner/rep-source",
        MaintenanceOpts {
            reconcile_tick: Duration::from_millis(100),
            ..MaintenanceOpts::default()
        },
    )
    .await;
    let replica = Harness::mem("rep-replica", "owner/rep-replica").await;
    let source_door = common::spawn_daemon_door(&source, DoorOptions::default()).await;
    let replica_door = common::spawn_daemon_door(&replica, DoorOptions::default()).await;
    let base = row_base("reconcile");

    let client = init_client(&base, "writer").await;
    let tip_c1 = commit_file(&client, "", "the crash-window commit").await;
    real_push(
        &client,
        &door_url(source_door.addr, &source.record.repo_id),
        &[],
        &["main"],
    )
    .await;

    // The crash window: the durable side moved, the consumer's
    // bookmark did not (no stream consumer ever attached; nothing
    // called the reconcile pass).
    let durable = node_watermark(&source).await;
    assert_eq!(
        durable, 2,
        "the put + the push's marker moved the durable side"
    );
    assert_eq!(
        source.assembly.applied_watermark().get(),
        0,
        "the applied watermark never moved — the window is open"
    );

    // The chain's reconcile tick closes it (the wave-7 chain driving
    // the composition's applied watermark — the recipe's loss
    // closure).
    let stop = alkstore::StopToken::new();
    let applied = source.assembly.applied_watermark().clone();
    let chain_run = source
        .assembly
        .chain()
        .run("rep-reconcile-worker", stop.clone(), |_| {});
    tokio::pin!(chain_run);
    let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
    let mut chain_done: Option<alkstore::Result<()>> = None;
    while applied.get() < durable {
        assert!(
            tokio::time::Instant::now() < deadline,
            "the reconcile tick anchored the applied watermark at the durable watermark"
        );
        tokio::select! {
            outcome = &mut chain_run => {
                chain_done = Some(outcome);
                break;
            }
            _ = tokio::time::sleep(Duration::from_millis(10)) => {}
        }
    }
    assert_eq!(
        applied.get(),
        durable,
        "the tick's sweep closed the crash window"
    );

    // The replicator pass — driven by the reconciled watermark —
    // converges the replica.
    let pass = replicate_pass(&source, source_door.addr, &replica, replica_door.addr).await;
    assert!(!pass.noop, "the window's diff still rode the loop");
    let tip_c1_oid = ObjectId::from_hex(tip_c1.as_bytes()).expect("the tip oid parses");
    assert_eq!(
        refs_map(&replica).await.get("refs/heads/main"),
        Some(&tip_c1_oid),
        "the replica caught up through the tick-closed window"
    );
    let clone = base.join("clone");
    real_clone(
        &door_url(replica_door.addr, &replica.record.repo_id),
        &clone,
    )
    .await;
    assert_clone_checked_out(&clone, &tip_c1, &file_content("")).await;

    stop.cancel();
    let outcome = match chain_done {
        Some(outcome) => outcome,
        None => chain_run.await,
    };
    outcome.expect("the chain stopped clean");
    source_door.accept.abort();
    replica_door.accept.abort();
    std::fs::remove_dir_all(&base).ok();
}

/// Delete propagation through the packless push commands (the recipe's
/// rider; ADR-020 §3's client-side rule): a branch deleted at the
/// source (a real delete-only push) re-diffs the pass to a
/// delete-only command set — no fetch runs, the apply rides the
/// packless push (captured at the replica's door: no pack bytes ever
/// ride the wire), the replica's ref row is gone, its own marker legs
/// fire for the delete, and the deleted branch's objects lose their
/// coverage — the registered closure releases them and the sweep
/// reclaims them while the surviving branch's objects stay covered.
///
/// Contract stamp: ADR-023 §3 (deletes propagate via packless push
/// commands); ADR-020 §3 (the write-side-open packless client rule —
/// the delete waits for the report) + §1 (the bounded peek's packless
/// classification); ADR-013 §3/§7/§8 (the zero-id delete shape, the
/// CAS home, the report); ADR-018 §6 (the liveness closure releasing
/// with the ref row); substrate/w8-e2e-replication.
#[tokio::test(flavor = "multi_thread")]
async fn a_deleted_refs_delete_propagates_through_the_replicators_packless_push() {
    if common::git_command().is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return;
    }
    let source = Harness::mem("rep-source", "owner/rep-source").await;
    let replica = Harness::mem("rep-replica", "owner/rep-replica").await;
    let source_door = common::spawn_daemon_door(&source, DoorOptions::default()).await;
    let replica_door = common::spawn_daemon_door(&replica, DoorOptions::default()).await;
    let base = row_base("delete");
    let source_url = door_url(source_door.addr, &source.record.repo_id);

    let client = init_client(&base, "writer").await;
    let tip_c1 = commit_file(&client, "", "main's commit").await;
    real_push(&client, &source_url, &[], &["main"]).await;
    git_simple(&client, &["checkout", "-q", "-b", "dev"]).await;
    let tip_d1 = commit_file(&client, "the dev branch's own content\n", "dev's commit").await;
    real_push(&client, &source_url, &[], &["dev"]).await;

    let pass = replicate_pass(&source, source_door.addr, &replica, replica_door.addr).await;
    assert!(!pass.noop, "the converge pass had a diff to carry");
    let tip_c1_oid = ObjectId::from_hex(tip_c1.as_bytes()).expect("the tip oid parses");
    let tip_d1_oid = ObjectId::from_hex(tip_d1.as_bytes()).expect("the tip oid parses");
    let replica_rows = refs_map(&replica).await;
    assert_eq!(replica_rows.get("refs/heads/main"), Some(&tip_c1_oid));
    assert_eq!(
        replica_rows.get("refs/heads/dev"),
        Some(&tip_d1_oid),
        "the branch converged before the delete rider"
    );
    assert_eq!(
        node_watermark(&replica).await,
        2,
        "the converge's one push (both creates rode one CAS) fired the replica's marker"
    );

    // The captured replica door rides only the delete pass.
    let delete_capture = common::StreamCapture::new();
    let replica_delete_door = common::spawn_daemon_door(
        &replica,
        DoorOptions {
            capture: Some(delete_capture.clone()),
            ..DoorOptions::default()
        },
    )
    .await;

    real_push(&client, &source_url, &[], &[":refs/heads/dev"]).await;
    let pass = replicate_pass(
        &source,
        source_door.addr,
        &replica,
        replica_delete_door.addr,
    )
    .await;
    assert!(
        pass.fetched_pack.is_none(),
        "a delete-only diff fetches nothing — the transfer leg is empty"
    );
    let report = pass.push_report.expect("the delete pass applied");
    assert_eq!(
        report.refs,
        vec![alkgit::RefOutcome {
            name: "refs/heads/dev".to_owned(),
            result: Ok(()),
        }],
        "the replica's CAS applied the delete"
    );
    assert!(
        extract_and_walk_client_pack(&delete_capture.to_server()).is_none(),
        "the captured replica-side stream carries no pack bytes — the delete rode the \
         packless push commands (the wire shape ADR-020 §3's rule serves)"
    );

    let replica_rows = refs_map(&replica).await;
    assert!(
        !replica_rows.contains_key("refs/heads/dev"),
        "the replica's ref row is gone"
    );
    assert_eq!(
        replica_rows.get("refs/heads/main"),
        Some(&tip_c1_oid),
        "the surviving branch held"
    );
    assert_eq!(
        node_watermark(&replica).await,
        3,
        "the delete's CAS rode the replica's marker legs like any push (watermark 3)"
    );

    // The deleted branch's own objects lost their coverage — the
    // sweep reclaims them; the surviving branch stays covered.
    let swept_tip = tip_d1_oid;
    let swept_blob_oid = {
        let hex = rev_parse(&client, &format!("{tip_d1}:a.txt")).await;
        ObjectId::from_hex(hex.as_bytes()).expect("the blob oid parses")
    };
    let report = replica
        .assembly
        .blobs()
        .sweep()
        .await
        .expect("the sweep runs");
    assert!(
        report.aborted.is_none(),
        "the sweep ran: {:?}",
        report.aborted
    );
    assert!(
        report.deleted >= 1,
        "the deleted ref's objects lost their coverage — the sweep reclaimed them"
    );
    for oid in [swept_tip, swept_blob_oid] {
        assert!(
            replica
                .assembly
                .front()
                .stat(&oid)
                .await
                .expect("the front stats")
                .is_none(),
            "the released branch's object was reclaimed: {oid}"
        );
    }
    assert!(
        replica
            .assembly
            .front()
            .stat(&tip_c1_oid)
            .await
            .expect("the front stats")
            .is_some(),
        "the surviving branch's objects remain covered"
    );

    source_door.accept.abort();
    replica_door.accept.abort();
    replica_delete_door.accept.abort();
    std::fs::remove_dir_all(&base).ok();
}

/// The partial-transfer window closes airtight (the recipe's rider):
/// a CAS'd application denied (`stale info`) after the pack already
/// ingested — the objects pre-landed (the custody's pending pins ride
/// the CAS home, the denied ref's guards dropped with the abort) —
/// and the next pass's re-transfer is a no-op at the object store: the
/// same trimmed need set re-fetches (the peer's held base still the
/// have), the re-unpack rides the idempotent CAS puts, the CAS now
/// applies with the advertisement's fresh old value, and the replica
/// converges with its objects covered across a sweep.
///
/// Contract stamp: ADR-023 §3 (partial-transfer windows close airtight
/// — pre-landed objects shrink the next need set, CAS puts make
/// re-transfer a no-op); ADR-013 §5 (the trimmed pack's bases held at
/// the server odb) §7 (the CAS — the stale denial is a business
/// result, the pack's fate rides it) §8 (the report's verbatim
/// reason); ADR-020 §2 (the ingestion's custody seam — the pins' drop
/// on the denied CAS); ADR-023 §Consequences (replication and GC
/// compose when refs apply within the pass);
/// substrate/w8-e2e-replication.
#[tokio::test(flavor = "multi_thread")]
async fn a_partial_transfer_window_closes_airtight_the_cas_retry_is_the_object_noop() {
    if common::git_command().is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return;
    }
    let source = Harness::mem("rep-source", "owner/rep-source").await;
    let replica = Harness::mem("rep-replica", "owner/rep-replica").await;
    let source_door = common::spawn_daemon_door(&source, DoorOptions::default()).await;
    let replica_door = common::spawn_daemon_door(&replica, DoorOptions::default()).await;
    let base = row_base("cas");

    let client = init_client(&base, "writer").await;
    let tip_c1 = commit_file(&client, "", "the converge commit").await;
    real_push(
        &client,
        &door_url(source_door.addr, &source.record.repo_id),
        &[],
        &["main"],
    )
    .await;
    let pass = replicate_pass(&source, source_door.addr, &replica, replica_door.addr).await;
    assert!(!pass.noop, "the converge pass had a diff to carry");
    let tip_c1_oid = ObjectId::from_hex(tip_c1.as_bytes()).expect("the tip oid parses");
    assert_eq!(
        refs_map(&replica).await.get("refs/heads/main"),
        Some(&tip_c1_oid),
        "the converge landed"
    );

    let tip_c2 = commit_file(&client, "the window commit's line\n", "the window commit").await;
    real_push(
        &client,
        &door_url(source_door.addr, &source.record.repo_id),
        &[],
        &["main"],
    )
    .await;
    let tip_c2_oid = ObjectId::from_hex(tip_c2.as_bytes()).expect("the tip oid parses");

    // The window, constructed: the transfer leg fetches the trimmed
    // delta, the apply leg's CAS carries a stale old value — the pack
    // ingested (its objects landed) but the ref never applied.
    let mut stale_session = dial(
        source_door.addr,
        &source.record.repo_id,
        Service::UploadPack,
    )
    .await;
    let mut pack = Vec::new();
    let reports = tokio::time::timeout(
        OP_TIMEOUT,
        stale_session.fetch(
            vec![
                FetchRound {
                    wants: vec![tip_c2_oid],
                    haves: vec![tip_c1_oid],
                    done: false,
                },
                FetchRound {
                    wants: vec![tip_c2_oid],
                    haves: vec![tip_c1_oid],
                    done: true,
                },
            ],
            &mut pack,
        ),
    )
    .await
    .expect("the fetch answers")
    .expect("the fetch runs");
    assert!(
        reports
            .iter()
            .flat_map(|report| report.acks.iter())
            .any(|oid| *oid == tip_c1_oid),
        "the fetched delta was trimmed against the peer's held base (the no-done round's ack)"
    );
    let base_tree = rev_parse(&client, &format!("{tip_c1}^{{tree}}")).await;
    let stale_expected = ObjectId::from_hex(base_tree.as_bytes()).expect("the decoy oid parses");
    let mut window_session = dial(
        replica_door.addr,
        &replica.record.repo_id,
        Service::ReceivePack,
    )
    .await;
    let report = tokio::time::timeout(
        OP_TIMEOUT,
        window_session.push(
            vec![RefUpdate {
                name: "refs/heads/main".to_owned(),
                expected: Some(stale_expected),
                new: Some(tip_c2_oid),
            }],
            Some(&mut &pack[..]),
            &PushSendOptions::default(),
        ),
    )
    .await
    .expect("the window push answers")
    .expect("the window push's report parses");
    assert_eq!(report.unpack, Ok(()), "the window's pack ingested");
    assert_eq!(
        report.refs,
        vec![alkgit::RefOutcome {
            name: "refs/heads/main".to_owned(),
            result: Err("stale info".to_owned()),
        }],
        "the stale CAS denies the ref application"
    );
    assert_eq!(
        refs_map(&replica).await.get("refs/heads/main"),
        Some(&tip_c1_oid),
        "the replica's ref never moved — the window is a real partial-transfer shape"
    );
    let pre_meta = replica
        .assembly
        .front()
        .stat(&tip_c2_oid)
        .await
        .expect("the front stats");
    assert!(
        pre_meta.is_some(),
        "the window's pack objects pre-landed (unpack-before-CAS — the shrinkable need set)"
    );

    // The next pass closes the window: the fetch re-trims against the
    // same held base; the re-transfer's re-unpack is the CAS put's
    // no-op; the CAS applies with the fresh advertisement's old value.
    let pass = replicate_pass(&source, source_door.addr, &replica, replica_door.addr).await;
    assert!(!pass.noop, "the window's diff re-diffed to a real pass");
    assert_eq!(
        pass.fetch_acks,
        vec![tip_c1_oid],
        "the pre-landed shape's base is still the peer's held closure"
    );
    let retry = decode_fetched_pack(&pass.fetched_pack.expect("the retry fetched its pack"));
    let introduced = rev_list_objects(&client, &[&tip_c2, &format!("^{}", tip_c1)]).await;
    assert_eq!(
        retry.object_count,
        introduced.len(),
        "the retry's need set is exactly the introduced delta — the pre-landed base \
         shrank it to the boundary's subtraction"
    );
    let report = pass.push_report.expect("the retry applied");
    assert_eq!(
        report.refs,
        vec![alkgit::RefOutcome {
            name: "refs/heads/main".to_owned(),
            result: Ok(()),
        }],
        "the CAS applied this time (the recipe: a denied application is business-as-usual \
         for the next pass)"
    );
    assert_eq!(
        refs_map(&replica).await.get("refs/heads/main"),
        Some(&tip_c2_oid),
        "the replica converged through the airtight closure"
    );
    let post_meta = replica
        .assembly
        .front()
        .stat(&tip_c2_oid)
        .await
        .expect("the front stats");
    assert_eq!(
        post_meta, pre_meta,
        "the re-transfer was the object store's no-op — the CAS puts are idempotent (the \
         entry's stat identity held across the re-put)"
    );

    let sweep = replica
        .assembly
        .blobs()
        .sweep()
        .await
        .expect("the sweep runs");
    assert!(
        sweep.aborted.is_none() && sweep.deleted == 0,
        "the ref applied within the pass — the replica's custody covers the converged \
         graph across the sweep (replication and GC compose)"
    );

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