alkgit 0.1.0

//! The wave-4 GitSession push client rows (core/w4-session-push): the
//! hand-rolled send-pack client (ADR-017 §3 — gitoxide has no
//! send-pack; a mirror of ADR-013's server shapes at the same honesty
//! bar, capture-grounded per push-captures.md, POC-X finding 3) driven
//! against the in-crate receive-pack server in-process — the V0
//! advertisement parse (the sentinel included), the command section's
//! zero-id shapes with the caps on the first line only (the wire bytes
//! byte-pinned), the caller-supplied pack streamed raw after the
//! command flush (thin included), the status report parsed in both
//! framings (sideband band-1-wrapped and bare, byte-checked against
//! our server's emission both ways), the atomic denial's aborted-refs
//! report, the client riders (the packless write-side-open shape per
//! ADR-020 §3 and the report-drain RST discipline per ADR-023 §5,
//! asserted from both sides), and the budgets (wall clock, the
//! report's max_pack_size mapping, fail-closed, terminal).
//!
//! Rows are names + `Contract stamp:` versions, per `tests/common`
//! §The row/stamp discipline.

mod common;

use std::collections::{BTreeMap, HashMap};
use std::pin::Pin;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll};
use std::time::Duration;

use gix_hash::ObjectId;
use gix_pack::data::entry::Header as PackHeader;
use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt, ReadBuf};
use tokio::sync::Semaphore;

use alkcall::core::auth::Identity;
use alkcall::core::types::BiStream;
use alkgit::{
    serve_duplex, AccessAction, GitPackGen, GitPackIngest, GitRefs, GitRegistryStore, GitSession,
    Limits, PushAdvertisedRef, PushReport, PushSendOptions, RefOutcome, RefUpdate, RepoRecord,
    Service, SessionError, SubstrateError, Visibility,
};
use common::{
    blob_fixture, commit_fixture, delta_stream, loose_body, pack_bytes, pack_entry_base,
    pack_entry_delta, parse_pkt_lines, sideband_report_lines, tree_fixture, MemPackGen,
    MemPackIngest, MemRefs, MemRegistry,
};

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

fn identity(id: &str) -> Identity {
    Identity {
        id: id.to_string(),
        scopes: Vec::new(),
        resources: HashMap::new(),
    }
}

type WireRecord = Arc<Mutex<Vec<u8>>>;
type SessionJoiner = tokio::task::JoinHandle<Result<(), SubstrateError>>;

type Handles = (
    Arc<MemRegistry>,
    Arc<MemRefs>,
    Arc<MemPackGen>,
    Arc<MemPackIngest>,
);

fn handles() -> Handles {
    let objects = common::MemObjects::default();
    (
        Arc::new(MemRegistry::default()),
        Arc::new(MemRefs::default()),
        Arc::new(MemPackGen {
            objects: objects.clone(),
        }),
        Arc::new(MemPackIngest { objects }),
    )
}

/// The fixture: the doubles' registry double plus the granted repo row.
async fn fixture_fresh(holder: &str) -> (Handles, RepoRecord, Identity) {
    let handles = handles();
    let record = repo("owner/doubled");
    let holder = identity(holder);
    let mut granted = record.clone();
    granted
        .grants
        .insert(holder.id.clone(), vec![AccessAction::Write]);
    let registry = Arc::clone(&handles.0);
    registry.put_repo(&granted).await.expect("the repo row");
    (handles, granted, holder)
}

// --- the wire spies --------------------------------------------------------

struct RecordRead<S> {
    inner: S,
    record: Arc<Mutex<Vec<u8>>>,
}

impl<S: tokio::io::AsyncRead + Unpin> tokio::io::AsyncRead for RecordRead<S> {
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {
        let this = self.as_mut().get_mut();
        let before = buf.filled().len();
        match Pin::new(&mut this.inner).poll_read(cx, buf) {
            Poll::Ready(Ok(())) => {
                this.record
                    .lock()
                    .expect("spy record")
                    .extend_from_slice(&buf.filled()[before..]);
                Poll::Ready(Ok(()))
            }
            other => other,
        }
    }
}

struct RecordWrite<S> {
    inner: S,
    record: Arc<Mutex<Vec<u8>>>,
    shutdowns: Arc<AtomicUsize>,
}

impl<S: tokio::io::AsyncWrite + Unpin> tokio::io::AsyncWrite for RecordWrite<S> {
    fn poll_write(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<std::io::Result<usize>> {
        let this = self.as_mut().get_mut();
        this.record
            .lock()
            .expect("spy record")
            .extend_from_slice(buf);
        Pin::new(&mut this.inner).poll_write(cx, buf)
    }
    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        let this = self.as_mut().get_mut();
        Pin::new(&mut this.inner).poll_flush(cx)
    }
    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        let this = self.as_mut().get_mut();
        this.shutdowns.fetch_add(1, Ordering::SeqCst);
        Pin::new(&mut this.inner).poll_shutdown(cx)
    }
}

fn shutdown_counter() -> Arc<AtomicUsize> {
    Arc::new(AtomicUsize::new(0))
}

/// Spawns the in-crate duplex serving entry (`serve_duplex`) for the
/// receive-pack service over a spy-wrapped pair; returns the client
/// end plus both wire records, the write-shutdown counter, and the
/// serving task's joiner.
fn serve_twin(
    handles: &Handles,
    record: RepoRecord,
    holder: &Identity,
) -> (
    BiStream,
    WireRecord,
    WireRecord,
    Arc<AtomicUsize>,
    SessionJoiner,
) {
    let checked = alkgit::AuthorizedRepo::authorize(record, Some(holder), AccessAction::Write)
        .expect("the checked write marker");
    let refs: Arc<dyn GitRefs> = handles.1.clone();
    let packgen: Arc<dyn GitPackGen> = handles.2.clone();
    let ingest: Arc<dyn GitPackIngest> = handles.3.clone();
    let (client_end, server_end) = duplex(1 << 16);
    let received = Arc::new(Mutex::new(Vec::new()));
    let sent = Arc::new(Mutex::new(Vec::new()));
    let shutdowns = shutdown_counter();
    let (server_read, server_write) = tokio::io::split(server_end);
    let server = BiStream::from_joined(
        RecordRead {
            inner: server_read,
            record: Arc::clone(&received),
        },
        RecordWrite {
            inner: server_write,
            record: Arc::clone(&sent),
            shutdowns: Arc::clone(&shutdowns),
        },
    );
    let joiner = tokio::spawn(async move {
        serve_duplex(
            &checked,
            server,
            Limits::default(),
            Service::ReceivePack,
            refs,
            packgen,
            ingest,
            false,
            Arc::new(Semaphore::new(
                Limits::default().max_concurrent_blocking_tasks,
            )),
        )
        .await
    });
    let (client_read, client_write) = tokio::io::split(client_end);
    (
        BiStream::from_joined(client_read, client_write),
        received,
        sent,
        shutdowns,
        joiner,
    )
}

/// Opens the push session over the serving twin (the channels-mode
/// preamble rode the open op — zero preamble bytes) and runs its
/// handshake; the session carries `session_limits` (the client-side
/// budgets; the server twin runs on the defaults).
async fn open_push_session(
    handles: &Handles,
    record: RepoRecord,
    holder: &Identity,
    session_limits: Limits,
) -> (
    (WireRecord, WireRecord, Arc<AtomicUsize>, SessionJoiner),
    GitSession,
) {
    let twin = serve_twin(handles, record, holder);
    let mut session = GitSession::open_via_channels(
        twin.0,
        "owner/doubled",
        Service::ReceivePack,
        session_limits,
    )
    .expect("the construction");
    session
        .handshake()
        .await
        .expect("the push handshake drains the advertisement");
    ((twin.1, twin.2, twin.3, twin.4), session)
}

/// The client end of a raw duplex pair as the session's BiStream.
fn client_side(stream: tokio::io::DuplexStream) -> BiStream {
    let (read, write) = tokio::io::split(stream);
    BiStream::from_joined(read, write)
}

async fn seed_ref(handles: &Handles, record: &RepoRecord, name: &str, oid: ObjectId) {
    Arc::clone(&handles.1)
        .apply_updates(
            record,
            vec![RefUpdate {
                name: name.to_owned(),
                expected: None,
                new: Some(oid),
            }],
            false,
        )
        .await
        .expect("the seed");
}

fn create(name: &str, new: ObjectId) -> RefUpdate {
    RefUpdate {
        name: name.to_owned(),
        expected: None,
        new: Some(new),
    }
}

fn delete(name: &str, current: ObjectId) -> RefUpdate {
    RefUpdate {
        name: name.to_owned(),
        expected: Some(current),
        new: None,
    }
}

fn update(name: &str, expected: ObjectId, new: ObjectId) -> RefUpdate {
    RefUpdate {
        name: name.to_owned(),
        expected: Some(expected),
        new: Some(new),
    }
}

/// Frames one command line the client's text-mode writer frames it:
/// the LF-augmented pkt-line (the writer assures the trailing newline).
fn pkt_line(payload: &[u8]) -> Vec<u8> {
    let total = payload.len() + 1 + 4;
    let mut out = Vec::with_capacity(total);
    out.extend_from_slice(format!("{total:04x}").as_bytes());
    out.extend_from_slice(payload);
    out.push(b'\n');
    out
}

/// Splits the server's sent record at the advertisement's flush (the
/// server speaks first: the advertisement precedes the report) — the
/// flush is the framing boundary, not a substring of the payloads.
fn split_advertisement(body: &[u8]) -> (&[u8], &[u8]) {
    let mut cursor = body;
    while !cursor.is_empty() {
        let header = std::str::from_utf8(&cursor[..4]).expect("hex header");
        let len = usize::from_str_radix(header, 16).expect("hex length");
        if len < 4 {
            let offset = body.len() - cursor.len();
            return (&body[..offset], &body[offset + 4..]);
        }
        cursor = &cursor[len..];
    }
    (body, &[])
}

/// The first command line's caps segment, test-side (the shape the
/// wire pins): report-status always, the caller's selections,
/// delete-refs auto-ridden on deletes, the pushed object format, the
/// crate agent.
fn first_caps(options: &PushSendOptions, has_deletes: bool) -> String {
    let mut tokens = vec![if options.report_status_v2 {
        "report-status-v2"
    } else {
        "report-status"
    }];
    if options.side_band_64k {
        tokens.push("side-band-64k");
    }
    if options.atomic {
        tokens.push("atomic");
    }
    if has_deletes {
        tokens.push("delete-refs");
    }
    format!(
        "{} object-format=sha1 agent=alkgit/{}",
        tokens.join(" "),
        env!("CARGO_PKG_VERSION")
    )
}

/// A commit-chain fixture: two commits' worth of objects; the chain
/// carries the bodies (stripped loose encoding) the thin-pack row
/// re-assembles pack entries from. `seed_child` seeds the child-side
/// objects too (the ordinary row's generation source; the thin row
/// leaves them unseeded so the pack lands them).
struct Chain {
    commit_a: ObjectId,
    commit_b: ObjectId,
    blob_a: ObjectId,
    blob_b: ObjectId,
    tree_b_body: Vec<u8>,
    commit_b_body: Vec<u8>,
}

fn chain_fixture(objects: &common::MemObjects, doubled: &RepoRecord, seed_child: bool) -> Chain {
    let (blob_a, blob_a_bytes) = blob_fixture(b"root content");
    let blob_b_content: &'static [u8] = b"root content\nchild content";
    let blob_b =
        gix_object::compute_hash(gix_hash::Kind::Sha1, gix_object::Kind::Blob, blob_b_content)
            .expect("the fixture hashes its own object");
    let (tree_a, _) = tree_fixture(&[(gix_object::tree::EntryKind::Blob, "file", blob_a)]);
    let (commit_a, _) = commit_fixture(tree_a, &[], "alpha");
    let (tree_b, tree_b_bytes) = tree_fixture(&[
        (gix_object::tree::EntryKind::Blob, "file", blob_a),
        (gix_object::tree::EntryKind::Blob, "child", blob_b),
    ]);
    let (commit_b, commit_b_bytes) = commit_fixture(tree_b, &[commit_a], "beta");
    for (oid, bytes) in [
        (blob_a, blob_a_bytes),
        (
            tree_a,
            tree_fixture(&[(gix_object::tree::EntryKind::Blob, "file", blob_a)]).1,
        ),
        (commit_a, commit_fixture(tree_a, &[], "alpha").1),
    ] {
        objects.put(doubled, oid, bytes);
    }
    if seed_child {
        let blob_b_bytes = blob_fixture(blob_b_content).1;
        for (oid, bytes) in [
            (blob_b, blob_b_bytes),
            (tree_b, tree_b_bytes),
            (commit_b, commit_b_bytes),
        ] {
            objects.put(doubled, oid, bytes);
        }
    }
    Chain {
        commit_a,
        commit_b,
        blob_a,
        blob_b,
        tree_b_body: loose_body(
            &tree_fixture(&[
                (gix_object::tree::EntryKind::Blob, "file", blob_a),
                (gix_object::tree::EntryKind::Blob, "child", blob_b),
            ])
            .1,
        )
        .1,
        commit_b_body: loose_body(&commit_fixture(tree_b, &[commit_a], "beta").1).1,
    }
}

/// Contract stamp: ADR-017 §3/§4 (the ordinary push — commands in the
/// zero-id create shapes, the caller's pack raw after the command
/// flush, the report parsed to typed outcomes), ADR-013 §2/§3/§8 (the
/// caps NUL-attached on the first line only — byte-pinned; the
/// sideband framing byte-checked from the server's emission),
/// push-captures.md §Request/§Status report; core/w4-session-push.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn an_ordinary_push_sends_commands_and_pack_then_parses_the_report() {
    let (handles, record, holder) = fixture_fresh("u-write").await;
    let chain = chain_fixture(handles.3.objects(), &record, true);
    let sink = common::SharedSink::default();
    Arc::clone(&handles.2)
        .generate(
            &record,
            vec![chain.commit_b],
            Vec::new(),
            Limits::default(),
            Box::new(sink.clone()),
        )
        .await
        .expect("the caller's pack generation");
    let pack = sink.snapshot();

    let ((wire_received, wire_sent, write_shutdowns, server_joiner), mut session) =
        open_push_session(&handles, record.clone(), &holder, Limits::default()).await;

    let advertised = session
        .push_advertisement()
        .await
        .expect("the empty repo's advertisement");
    assert!(
        advertised.is_empty(),
        "the empty repo's sentinel parses as the empty set: {advertised:?}"
    );

    let options = PushSendOptions {
        side_band_64k: true,
        ..Default::default()
    };
    let report = tokio::time::timeout(
        Duration::from_secs(10),
        session.push(
            vec![create("refs/heads/main", chain.commit_b)],
            Some(&mut &pack[..]),
            &options,
        ),
    )
    .await
    .expect("the push is timely")
    .expect("the ordinary push lands");

    assert_eq!(
        report,
        PushReport {
            unpack: Ok(()),
            refs: vec![RefOutcome {
                name: "refs/heads/main".to_owned(),
                result: Ok(()),
            }],
        },
        "the report parses to the typed per-ref outcome"
    );
    let rows = Arc::clone(&handles.1)
        .list_refs(&record)
        .await
        .expect("rows");
    assert_eq!(rows.len(), 1, "the ref row landed");
    assert_eq!(rows[0].oid, Some(chain.commit_b));
    assert!(
        handles.3.objects().contains(&record, &chain.blob_b),
        "the pack's objects landed"
    );

    let zero = "0".repeat(40);
    let caps = first_caps(&options, false);
    let expected_client: Vec<u8> = [
        pkt_line(format!("{zero} {} refs/heads/main\0 {caps}", chain.commit_b).as_bytes()),
        b"0000".to_vec(),
        pack,
    ]
    .concat();
    assert_eq!(
        *wire_received.lock().expect("spy record"),
        expected_client,
        "the client's wire bytes: the command line (caps NUL-attached, zero-id create shape), \
         the command flush, then the pack raw"
    );

    let sent = wire_sent.lock().expect("spy record").clone();
    let (_advertisement, report_part) = split_advertisement(&sent);
    assert_eq!(
        sideband_report_lines(report_part),
        vec!["unpack ok", "ok refs/heads/main"],
        "the server's sideband-framed report parses to the same outcomes"
    );

    // The client drops the stream only after the report drained (the
    // ADR-023 §5 shape); the server's serving session then completes.
    drop(session);
    server_joiner
        .await
        .expect("the server join")
        .expect("the serving task completed cleanly — the push session closed after the report");
    assert_eq!(
        write_shutdowns.load(Ordering::SeqCst),
        0,
        "the client never shut the write half down mid-push (the rider's shape)"
    );
}

/// Contract stamp: ADR-013 §5 (thin packs accepted, bases from the
/// server odb — the caller's thin pack rides raw, the ref-delta base
/// resolves server-side), ADR-017 §4 (the caller-supplied pack);
/// core/w4-session-push.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_thin_push_streams_the_callers_thin_pack() {
    let (handles, record, holder) = fixture_fresh("u-write").await;
    let chain = chain_fixture(handles.3.objects(), &record, false);
    assert!(
        !handles.3.objects().contains(&record, &chain.blob_b),
        "the child-side objects are unseeded — only the pack can land them"
    );

    let mut offset = 0u64;
    // The delta's base is the server-held blob ("root content", 12
    // bytes — the delta_stream fixture form copies it whole); the
    // insertion appended reconstructs the child blob's body.
    let delta = pack_entry_delta(
        PackHeader::RefDelta {
            base_id: chain.blob_a,
        },
        delta_stream(12, b"\nchild content").as_slice(),
        offset,
    );
    offset += delta.bytes_in_pack();
    let tree_entry = pack_entry_base(gix_object::Kind::Tree, &chain.tree_b_body, offset);
    offset += tree_entry.bytes_in_pack();
    let commit_entry = pack_entry_base(gix_object::Kind::Commit, &chain.commit_b_body, offset);
    let thin_pack = pack_bytes(&[delta, tree_entry, commit_entry]);

    let (_, mut session) =
        open_push_session(&handles, record.clone(), &holder, Limits::default()).await;
    let advertised = session
        .push_advertisement()
        .await
        .expect("the empty repo's advertisement");
    assert!(advertised.is_empty());

    let options = PushSendOptions::default();
    let report = tokio::time::timeout(
        Duration::from_secs(10),
        session.push(
            vec![create("refs/heads/main", chain.commit_b)],
            Some(&mut &thin_pack[..]),
            &options,
        ),
    )
    .await
    .expect("the push is timely")
    .expect("the thin push lands");
    assert_eq!(
        report.unpack,
        Ok(()),
        "the thin pack ingests — the ref-delta base resolves from the server's own objects"
    );
    assert_eq!(
        report.refs,
        vec![RefOutcome {
            name: "refs/heads/main".to_owned(),
            result: Ok(()),
        }]
    );
    assert!(
        handles.3.objects().contains(&record, &chain.blob_b),
        "the delta'd child blob landed (reconstructed from the server-held base)"
    );
}

/// Contract stamp: ADR-013 §10 (the atomic semantics client-side: an
/// aborted transaction reports the aborted-refs shape — `atomic push
/// failure` on the would-have-succeeded refs, the denied refs keep
/// their own reasons), ADR-017 §3, ADR-013 §2 (the typed
/// advertisement rows); push-captures.md §CAS; core/w4-session-push.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn an_atomic_push_denial_parses_the_aborted_refs_report() {
    let (handles, record, holder) = fixture_fresh("u-write").await;
    let chain = chain_fixture(handles.3.objects(), &record, true);
    seed_ref(&handles, &record, "refs/heads/keep", chain.commit_a).await;

    let (_, mut session) =
        open_push_session(&handles, record.clone(), &holder, Limits::default()).await;
    let advertised = session
        .push_advertisement()
        .await
        .expect("the advertisement");
    assert_eq!(
        advertised,
        vec![PushAdvertisedRef {
            oid: chain.commit_a,
            name: "refs/heads/keep".to_owned(),
        }],
        "the seeded advertisement parses to the typed ref rows"
    );

    // One command is CAS-stale (a wrong expected id), one would have
    // succeeded: the atomic transaction aborts whole.
    let stale_expected =
        gix_object::compute_hash(gix_hash::Kind::Sha1, gix_object::Kind::Blob, b"wrong")
            .expect("a decoy oid");
    let options = PushSendOptions {
        atomic: true,
        ..Default::default()
    };
    let report = tokio::time::timeout(
        Duration::from_secs(10),
        session.push(
            vec![
                create("refs/heads/other", chain.commit_b),
                update("refs/heads/keep", stale_expected, chain.commit_b),
            ],
            None::<&mut &[u8]>,
            &options,
        ),
    )
    .await
    .expect("the push is timely")
    .expect("the atomic push reports");
    assert_eq!(report.unpack, Ok(()));
    assert_eq!(
        report.refs,
        vec![
            RefOutcome {
                name: "refs/heads/other".to_owned(),
                result: Err("atomic push failure".to_owned()),
            },
            RefOutcome {
                name: "refs/heads/keep".to_owned(),
                result: Err("stale info".to_owned()),
            },
        ],
        "the aborted-refs report: would-have-succeeded carries the atomic failure, the denied \
         ref keeps its own reason"
    );
    let rows = Arc::clone(&handles.1)
        .list_refs(&record)
        .await
        .expect("rows");
    assert_eq!(
        rows.len(),
        1,
        "nothing applied — the transaction aborted whole"
    );
    assert_eq!(
        rows[0].oid,
        Some(chain.commit_a),
        "the seeded ref untouched"
    );
}

/// The packless delete-only push, both report framings byte-checked
/// from our server's emission (sideband selected, sideband not), with
/// the ADR-020 §3 rider asserted from both sides: the client holds the
/// write side open until the report (no shutdown), and the server's
/// bounded peek takes the packless path — the report arrives and the
/// serving task completes cleanly, never a partial-header EOF error.
///
/// Contract stamp: ADR-020 §3 (the write-side-open rider, asserted
/// from both sides), ADR-020 §1 (our server's peek serves the
/// packless client's waiting shape), ADR-013 §8 (both framings), §3
/// (the zero-id delete shape, byte-pinned), ADR-023 §5 (the report is
/// read in full before the stream could drop); core/w4-session-push.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_packless_delete_push_holds_the_write_side_open_across_both_report_framings() {
    for (why, options) in [
        (
            "the sideband framing",
            PushSendOptions {
                side_band_64k: true,
                ..Default::default()
            },
        ),
        ("the bare framing", PushSendOptions::default()),
    ] {
        let (handles, record, holder) = fixture_fresh("u-write").await;
        let chain = chain_fixture(handles.3.objects(), &record, true);
        seed_ref(&handles, &record, "refs/heads/dev", chain.commit_a).await;

        let ((wire_received, wire_sent, write_shutdowns, server_joiner), mut session) =
            open_push_session(&handles, record.clone(), &holder, Limits::default()).await;
        let advertised = session
            .push_advertisement()
            .await
            .expect("the advertisement");
        assert_eq!(
            advertised,
            vec![PushAdvertisedRef {
                oid: chain.commit_a,
                name: "refs/heads/dev".to_owned(),
            }],
            "{why}: the seeded advertisement"
        );

        let report = tokio::time::timeout(
            Duration::from_secs(10),
            session.push(
                vec![delete("refs/heads/dev", chain.commit_a)],
                None::<&mut &[u8]>,
                &options,
            ),
        )
        .await
        .expect("the push is timely")
        .expect("the packless delete push lands");
        assert_eq!(
            report,
            PushReport {
                unpack: Ok(()),
                refs: vec![RefOutcome {
                    name: "refs/heads/dev".to_owned(),
                    result: Ok(()),
                }],
            },
            "{why}: the delete reports ok"
        );
        let rows = Arc::clone(&handles.1)
            .list_refs(&record)
            .await
            .expect("rows");
        assert!(
            rows.is_empty(),
            "{why}: the ref row is deleted (the packless CAS-only path ran)"
        );

        let zero = "0".repeat(40);
        let caps = first_caps(&options, true);
        let expected_client: Vec<u8> = [
            pkt_line(format!("{} {zero} refs/heads/dev\0 {caps}", chain.commit_a).as_bytes()),
            b"0000".to_vec(),
        ]
        .concat();
        assert_eq!(
            *wire_received.lock().expect("spy record"),
            expected_client,
            "{why}: the delete command (old = the current id, new = the zero id — \
             ADR-013 §3's delete shape) with the caps NUL-attached, the command flush, and \
             nothing after — the packless shape holds the write side open (no shutdown, no \
             half-close) until the report"
        );
        assert_eq!(
            write_shutdowns.load(Ordering::SeqCst),
            0,
            "{why}: the client never shut the write half down before the report"
        );

        let sent = wire_sent.lock().expect("spy record").clone();
        let (_advertisement, report_part) = split_advertisement(&sent);
        if options.side_band_64k {
            assert_eq!(
                sideband_report_lines(report_part),
                vec!["unpack ok", "ok refs/heads/dev"],
                "{why}: the band-1-wrapped report (the inner framing inside the band)"
            );
        } else {
            assert_eq!(
                parse_pkt_lines(report_part),
                vec![
                    b"unpack ok\n".to_vec(),
                    b"ok refs/heads/dev\n".to_vec(),
                    Vec::new(),
                ],
                "{why}: the bare report — pkt-lines to the single flush"
            );
        }

        drop(session);
        server_joiner.await.expect("the server join").expect(
            "{why}: the server took the packless path and closed cleanly — never the \
                     partial-header EOF error",
        );
    }
}

/// The ADR-023 §5 discipline against a scripted peer: a packless push
/// holds the write side open while waiting for the report (the peer
/// observes no EOF while it delays), and the client reads the full
/// report — the response side drained to its flush — returning before
/// the peer's scripted close (no close-with-unread-receive-data on
/// either end).
///
/// Contract stamp: ADR-020 §3 (the write-side-open shape,
/// peer-observed), ADR-023 §5 (the drain-before-drop discipline);
/// core/w4-session-push.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_packless_push_holds_the_write_side_open_and_drains_the_full_report() {
    let (client_end, server_end) = duplex(1 << 16);
    let facts: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
    let peer_facts = Arc::clone(&facts);
    tokio::spawn(async move {
        let (mut peer_read, mut peer_write) = tokio::io::split(server_end);
        peer_write
            .write_all(
                &[
                    pkt_line(
                        format!(
                            "{} capabilities^{{}}\0report-status delete-refs object-format=sha1",
                            "0".repeat(40)
                        )
                        .as_bytes(),
                    ),
                    b"0000".to_vec(),
                ]
                .concat(),
            )
            .await
            .expect("the advertisement writes");
        // The client's commands, to the command flush.
        loop {
            let mut header = [0u8; 4];
            if peer_read.read_exact(&mut header).await.is_err() {
                peer_facts
                    .lock()
                    .expect("facts")
                    .push("eof before the report".to_owned());
                return;
            }
            let len = usize::from_str_radix(std::str::from_utf8(&header).expect("hex header"), 16)
                .expect("hex length");
            if len < 4 {
                break;
            }
            let mut payload = vec![0u8; len - 4];
            peer_read
                .read_exact(&mut payload)
                .await
                .expect("the payload");
        }
        // The rider's shape: the client holds its write side open — no
        // EOF arrives while the peer delays (a half-closing client
        // would surface here as an unexpected read end).
        let eof_watcher = tokio::spawn(async move {
            let mut drained = Vec::new();
            match peer_read.read_to_end(&mut drained).await {
                Ok(_) => drained,
                Err(_) => Vec::new(),
            }
        });
        match tokio::time::timeout(Duration::from_millis(900), eof_watcher).await {
            Ok(Ok(buf)) => {
                if buf.is_empty() {
                    peer_facts
                        .lock()
                        .expect("facts")
                        .push("eof before the report".to_owned());
                } else {
                    peer_facts
                        .lock()
                        .expect("facts")
                        .push("unexpected bytes while waiting".to_owned());
                }
                return;
            }
            Err(_still_open) => {}
            Ok(Err(_)) => {
                peer_facts
                    .lock()
                    .expect("facts")
                    .push("eof before the report".to_owned());
                return;
            }
        }
        // The report, then a scripted delayed close: the client must
        // have returned after draining the full report.
        peer_facts
            .lock()
            .expect("facts")
            .push("report sent".to_owned());
        peer_write
            .write_all(
                &[
                    b"000eunpack ok\n".to_vec(),
                    b"0016ok refs/heads/dev\n".to_vec(),
                    b"0000".to_vec(),
                ]
                .concat(),
            )
            .await
            .expect("the report writes");
        tokio::time::sleep(Duration::from_millis(1500)).await;
        let _ = peer_write.shutdown().await;
    });

    let mut session = GitSession::open_via_channels(
        client_side(client_end),
        "owner/doubled",
        Service::ReceivePack,
        Limits::default(),
    )
    .expect("the construction");
    session.handshake().await.expect("the advertisement drain");

    let options = PushSendOptions::default();
    let current = gix_object::compute_hash(gix_hash::Kind::Sha1, gix_object::Kind::Blob, b"fake")
        .expect("a decoy oid");
    let report = tokio::time::timeout(
        Duration::from_secs(10),
        session.push(
            vec![delete("refs/heads/dev", current)],
            None::<&mut &[u8]>,
            &options,
        ),
    )
    .await
    .expect("the push is timely — it returned on the report, not on the peer's close")
    .expect("the report drained in full");
    assert_eq!(
        report,
        PushReport {
            unpack: Ok(()),
            refs: vec![RefOutcome {
                name: "refs/heads/dev".to_owned(),
                result: Ok(()),
            }],
        }
    );
    tokio::time::sleep(Duration::from_millis(100)).await;
    assert_eq!(
        *facts.lock().expect("facts"),
        vec!["report sent".to_owned()],
        "the client held the write side open until the report and returned before the peer's \
         scripted close (the report side was drained in full)"
    );
    tokio::time::sleep(Duration::from_millis(1600)).await;
}

/// Contract stamp: ADR-017 §6 / ADR-009 (the wall clock bounds the
/// operation client-side; the response report maps to `max_pack_size`
/// — the construction task's recorded mapping; breaches are typed and
/// fail-closed, terminal, re-reported); core/w4-session-push.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_push_budgets_fail_closed_typed() {
    // The wall clock: a peer that never sends a report stalls the
    // report read — the breach is the typed wall-clock error, terminal.
    let (client_end, server_end) = duplex(1 << 16);
    tokio::spawn(async move {
        let mut stream = server_end;
        stream
            .write_all(
                &[
                    pkt_line(
                        format!(
                            "{} capabilities^{{}}\0report-status object-format=sha1",
                            "0".repeat(40)
                        )
                        .as_bytes(),
                    ),
                    b"0000".to_vec(),
                ]
                .concat(),
            )
            .await
            .expect("the advertisement writes");
        tokio::time::sleep(Duration::from_secs(30)).await;
        let _ = stream.shutdown().await;
    });
    let limits = Limits {
        session_wall_clock: Duration::from_millis(400),
        ..Limits::default()
    };
    let mut session = GitSession::open_via_channels(
        client_side(client_end),
        "owner/doubled",
        Service::ReceivePack,
        limits,
    )
    .expect("the construction");
    session.handshake().await.expect("the advertisement drain");
    let current = gix_object::compute_hash(gix_hash::Kind::Sha1, gix_object::Kind::Blob, b"fake")
        .expect("a decoy oid");
    let error = tokio::time::timeout(
        Duration::from_secs(5),
        session.push(
            vec![delete("refs/heads/x", current)],
            None::<&mut &[u8]>,
            &Default::default(),
        ),
    )
    .await
    .expect("the breach answers promptly")
    .expect_err("the wall clock breached");
    match &error {
        SessionError::WallClock { budget } => {
            assert_eq!(*budget, Duration::from_millis(400));
        }
        other => panic!("expected the typed wall-clock breach, got {other:?}"),
    }
    let again = session
        .push(
            vec![delete("refs/heads/x", current)],
            None::<&mut &[u8]>,
            &Default::default(),
        )
        .await
        .expect_err("the session is terminal");
    assert_eq!(error, again, "the breach re-reports without re-entry");

    // The response report's budget: the report bytes map to
    // max_pack_size — a report over the budget ends the session with
    // the typed breach, fail-closed (nothing beyond the budget is
    // ever retained).
    let (handles, record, holder) = fixture_fresh("u-write").await;
    let chain = chain_fixture(handles.3.objects(), &record, true);
    seed_ref(&handles, &record, "refs/heads/dev", chain.commit_a).await;
    let client_limits = Limits {
        max_pack_size: 8,
        ..Limits::default()
    };
    let ((_, _, _, server_joiner), mut session) =
        open_push_session(&handles, record.clone(), &holder, client_limits).await;
    let error = tokio::time::timeout(
        Duration::from_secs(10),
        session.push(
            vec![delete("refs/heads/dev", chain.commit_a)],
            None::<&mut &[u8]>,
            &Default::default(),
        ),
    )
    .await
    .expect("the push is timely")
    .expect_err("the report outgrew the budget");
    match &error {
        SessionError::BudgetExceeded {
            what,
            budget_name,
            limit,
        } => {
            assert!(what.contains("report"), "{what}");
            assert_eq!(*budget_name, "max_pack_size");
            assert_eq!(*limit, 8);
        }
        other => panic!("expected the typed budget breach, got {other:?}"),
    }
    let again = session
        .push(
            vec![delete("refs/heads/dev", chain.commit_a)],
            None::<&mut &[u8]>,
            &Default::default(),
        )
        .await
        .expect_err("the session is terminal");
    assert_eq!(error, again, "the breach re-reports without re-entry");
    // The client drops (the report side was drained to the breach);
    // the server's serving session then completes.
    drop(session);
    server_joiner
        .await
        .expect("the server join")
        .expect("the server side of the budget breach completes cleanly");
}

/// Local grammar refusals fail before the wire (nothing attached, and
/// nothing would have been sent): an empty command set (the caller
/// computes the diff; a nonzero command set is the minimum shape), a
/// push on an upload-pack session. Both end the session (error-break
/// terminal).
///
/// Contract stamp: ADR-017 §6 (the local `invalid` class,
/// fail-closed), ADR-016 §3 (one service machine per stream);
/// core/w4-session-push.
#[tokio::test]
async fn local_push_refusals_fail_before_the_wire_and_terminate() {
    let (client_end, _server_end) = duplex(1 << 16);
    let mut session = GitSession::open_via_channels(
        client_side(client_end),
        "owner/doubled",
        Service::ReceivePack,
        Limits::default(),
    )
    .expect("the construction");
    let error = session
        .push(vec![], None::<&mut &[u8]>, &Default::default())
        .await
        .expect_err("an empty command set refuses locally");
    match &error {
        SessionError::InvalidInput(message) => {
            assert!(message.contains("empty"), "{message}");
        }
        other => panic!("expected the local refusal, got {other:?}"),
    }
    let again = session
        .push(vec![], None::<&mut &[u8]>, &Default::default())
        .await
        .expect_err("the session is terminal");
    assert_eq!(error, again, "the refusal re-reports without re-entry");

    let mut wrong_service = GitSession::open_via_channels(
        client_side(duplex(1 << 16).0),
        "owner/doubled",
        Service::UploadPack,
        Limits::default(),
    )
    .expect("the construction");
    let current = gix_object::compute_hash(gix_hash::Kind::Sha1, gix_object::Kind::Blob, b"fake")
        .expect("a decoy oid");
    let error = wrong_service
        .push(
            vec![delete("refs/heads/x", current)],
            None::<&mut &[u8]>,
            &Default::default(),
        )
        .await
        .expect_err("push never rides an upload-pack session");
    match &error {
        SessionError::InvalidInput(message) => {
            assert!(message.contains("receive-pack"), "{message}");
        }
        other => panic!("expected the local refusal, got {other:?}"),
    }
}