alkgit/tests/fetch_machines.rs

//! The wave-2 fetch negotiation machine over the duplex session
//! substrate and the stateless response end, with the wave-2 backend
//! doubles (`core/w2-negotiation`): the ADR-014 ack loop, the done
//! round's packfile section, and the budgets — wire shapes asserted
//! byte-by-byte against negotiation-captures.md's captures.
//!
//! Rows are names + `Contract stamp:` versions, per
//! `tests/common` §The row/stamp discipline.

mod common;

use std::collections::BTreeSet;
use std::sync::{Arc, Mutex as StdMutex};
use std::time::Duration;

use common::{blob_fixture, commit_fixture, decode_pack, tree_fixture, MemObjects, MemPackGen};
use futures::io::AsyncWrite as FuturesAsyncWrite;
use gix_hash::ObjectId;
use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt};
use tokio::sync::Semaphore;

use alkgit::{
    parse_fetch_args, serve_fetch_round, FetchRounds, Limits, RepoRecord, Request, Service,
    Session, StatelessRequest, SubstrateError, Visibility,
};

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

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

fn request(pre_lines: &[&str], args: &[&str]) -> Request {
    Request {
        command: "fetch".to_owned(),
        pre_lines: pre_lines.iter().map(|l| l.to_string()).collect(),
        args: args.iter().map(|l| l.to_string()).collect(),
    }
}

fn want_oid(hex: &str) -> String {
    format!("want {hex}")
}

fn have_oid(hex: &str) -> String {
    format!("have {hex}")
}

const FLUSH: &[u8] = b"0000";
const DELIMITER: &[u8] = b"0001";

/// Encodes one pkt-line, client-side framing style.
fn pkt(payload: &[u8]) -> Vec<u8> {
    let total = payload.len() + 4;
    let mut out = Vec::with_capacity(total);
    out.extend_from_slice(format!("{total:04x}").as_bytes());
    out.extend_from_slice(payload);
    out
}

/// Encodes a fetch round request's wire bytes.
fn fetch_request_bytes(pre_lines: &[&str], args: &[&str]) -> Vec<u8> {
    let mut out = pkt(b"command=fetch\n");
    for line in pre_lines {
        out.extend_from_slice(&pkt(line.as_bytes()));
    }
    out.extend_from_slice(DELIMITER);
    for arg in args {
        out.extend_from_slice(&pkt(arg.as_bytes()));
    }
    out.extend_from_slice(FLUSH);
    out
}

/// Reads pkt-lines from the client end until (and including) a flush.
async fn read_pkts(reader: &mut tokio::io::DuplexStream) -> Vec<Vec<u8>> {
    let mut packets = Vec::new();
    loop {
        let mut header = [0u8; 4];
        reader
            .read_exact(&mut header)
            .await
            .expect("pkt-line header");
        let len = usize::from_str_radix(std::str::from_utf8(&header).expect("hex header"), 16)
            .expect("hex length");
        if len < 4 {
            packets.push(Vec::new());
            break;
        }
        let mut payload = vec![0u8; len - 4];
        reader
            .read_exact(&mut payload)
            .await
            .expect("pkt-line payload");
        packets.push(payload);
    }
    packets
}

/// A duplex session + its client end, over a buffer large enough to hold
/// a full packfile section (the machine streams before the rows read).
fn session(
    limits: Limits,
) -> (
    tokio::io::DuplexStream,
    Session<tokio::io::DuplexStream>,
    Limits,
) {
    let (peer, stream) = duplex(1 << 20);
    (
        peer,
        Session::new(stream, limits, Service::UploadPack),
        limits,
    )
}

/// Pkt-line payloads of a framed response body (the stateless rows).
fn parse_pkt_lines(body: &[u8]) -> Vec<Vec<u8>> {
    let mut lines = Vec::new();
    let mut cursor = body;
    while !cursor.is_empty() {
        let (header, rest) = cursor.split_at(4);
        let len = usize::from_str_radix(std::str::from_utf8(header).expect("hex header"), 16)
            .expect("hex length");
        cursor = rest;
        if len < 4 {
            lines.push(Vec::new());
            continue;
        }
        let (payload, rest) = cursor.split_at(len - 4);
        cursor = rest;
        lines.push(payload.to_vec());
    }
    lines
}

fn text(payload: &[u8]) -> String {
    String::from_utf8_lossy(payload)
        .trim_end_matches('\n')
        .to_owned()
}

/// An async sink whose framed bytes the rows inspect after the fact —
/// the stateless response end's writer (the http door's surface shape).
#[derive(Clone, Default)]
struct VecSink(Arc<StdMutex<Vec<u8>>>);

impl VecSink {
    fn body(&self) -> Vec<u8> {
        self.0.lock().expect("sink is not poisoned").clone()
    }
}

impl FuturesAsyncWrite for VecSink {
    fn poll_write(
        self: std::pin::Pin<&mut Self>,
        _cx: &mut std::task::Context<'_>,
        buf: &[u8],
    ) -> std::task::Poll<std::io::Result<usize>> {
        self.0
            .lock()
            .expect("sink is not poisoned")
            .extend_from_slice(buf);
        std::task::Poll::Ready(Ok(buf.len()))
    }

    fn poll_flush(
        self: std::pin::Pin<&mut Self>,
        _cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        std::task::Poll::Ready(Ok(()))
    }

    fn poll_close(
        self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        self.poll_flush(cx)
    }
}

/// Incompressible pseudo-random bytes (LCG) — forces multi-chunk bands.
fn noisy_bytes(len: usize) -> Vec<u8> {
    let mut state = 0x12345678u64;
    (0..len)
        .map(|_| {
            state = state
                .wrapping_mul(6364136223846793005)
                .wrapping_add(1442695040888963407);
            (state >> 33) as u8
        })
        .collect()
}

/// A two-commit chain fixture: commit_b(child) → commit_a(root), each
/// with its own tree and blob. Returns the objects and the loose bytes
/// to seed.
struct Chain {
    blob_a: ObjectId,
    blob_b: ObjectId,
    tree_a: ObjectId,
    tree_b: ObjectId,
    commit_a: ObjectId,
    commit_b: ObjectId,
}

fn two_commit_chain(objects: &MemObjects, doubled: &RepoRecord) -> Chain {
    let (blob_a, blob_a_bytes) = blob_fixture(b"alpha");
    let (blob_b, blob_b_bytes) = blob_fixture(b"beta");
    let (tree_a, tree_a_bytes) =
        tree_fixture(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob_a)]);
    let (tree_b, tree_b_bytes) =
        tree_fixture(&[(gix_object::tree::EntryKind::Blob, "b.txt", blob_b)]);
    let (commit_a, commit_a_bytes) = commit_fixture(tree_a, &[], "alpha");
    let (commit_b, commit_b_bytes) = commit_fixture(tree_b, &[commit_a], "beta");
    for (id, bytes) in [
        (blob_a, blob_a_bytes),
        (blob_b, blob_b_bytes),
        (tree_a, tree_a_bytes),
        (tree_b, tree_b_bytes),
        (commit_a, commit_a_bytes),
        (commit_b, commit_b_bytes),
    ] {
        objects.put(doubled, id, bytes);
    }
    Chain {
        blob_a,
        blob_b,
        tree_a,
        tree_b,
        commit_a,
        commit_b,
    }
}

/// A no-`done` round is answered with the exact `acknowledgments`
/// grammar of negotiation-captures.md's captures: `acknowledgments`,
/// `ACK <oid>` per recognized have — the seam's output, bare 40-hex
/// oids, no suffixes — `NAK` absent when ACKs were sent, flush
/// terminator, never `ready`, never a packfile line.
///
/// Contract stamp: ADR-014 §1 (the round grammar + the ACK rule); ADR-014
/// §5 (the wire layer acks exactly `common_haves`' output, no
/// refinement); negotiation-captures.md §Round grammar (NAK placement,
/// FLUSH terminator, bare-oid ACK line format); ADR-014 §4 (`ready` is
/// never sent, `fetch=wait-for-done` honest).
#[tokio::test]
async fn no_done_round_answers_the_exact_acknowledgments_grammar() {
    let doubled = repo("owner/doubled");
    let objects = MemObjects::default();
    let chain = two_commit_chain(&objects, &doubled);
    let gen = Arc::new(MemPackGen {
        objects: objects.clone(),
    });

    let (mut peer, mut session, limits) = session(Limits::default());
    peer.write_all(&fetch_request_bytes(
        &["agent=git/2.43.0", "object-format=sha1"],
        &[
            &want_oid(&chain.commit_b.to_string()),
            &have_oid(&chain.commit_a.to_string()),
            &have_oid(&oid(9).to_string()),
        ],
    ))
    .await
    .unwrap();

    let raw = session.read_request().await.unwrap().expect("a request");
    let args = parse_fetch_args(&raw).unwrap();
    assert!(!args.done, "the no-done round parses without done");
    let mut rounds = FetchRounds::default();
    serve_fetch_round(
        &mut session,
        &raw,
        &mut rounds,
        gen.clone(),
        &doubled,
        &limits,
        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
    )
    .await
    .unwrap();

    let packets = read_pkts(&mut peer).await;
    let shapes: Vec<String> = packets.iter().map(|p| text(p)).collect();
    assert_eq!(shapes[0], "acknowledgments");
    assert_eq!(shapes[1], format!("ACK {}", chain.commit_a));
    assert_eq!(packets.len(), 3, "the section ends at flush");
    assert!(packets[2].is_empty(), "the flush terminator");
    for forbidden in ["NAK", "ready"] {
        assert!(
            shapes.iter().all(|s| !s.contains(forbidden)),
            "{forbidden} is absent when ACKs were sent (negotiation-captures.md)"
        );
    }
    assert!(shapes.iter().all(|s| !s.contains("packfile")));
    assert_eq!(rounds.rounds_used(), 1, "the round consumed the budget");
}

/// A round whose haves the seam recognizes none of — missing or
/// non-commit alike — is `acknowledgments` + `NAK` + flush: NAK only
/// when no ACKs in the round.
///
/// Contract stamp: ADR-014 §1 (NAK only when no ACKs); negotiation-captures.md
/// §Round grammar (the observed NAK-alone shape); ADR-014 §5 (the seam's
/// exists + is-commit predicate decides).
#[tokio::test]
async fn a_round_with_no_recognized_haves_yields_the_nak_only_shape() {
    let doubled = repo("owner/doubled");
    let objects = MemObjects::default();
    let (blob, blob_bytes) = blob_fixture(b"leaf");
    let (tree, tree_bytes) = tree_fixture(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob)]);
    let (commit, commit_bytes) = commit_fixture(tree, &[], "root");
    for (id, bytes) in [
        (blob, blob_bytes),
        (tree, tree_bytes),
        (commit, commit_bytes),
    ] {
        objects.put(&doubled, id, bytes);
    }
    let gen = Arc::new(MemPackGen {
        objects: objects.clone(),
    });

    let (mut peer, mut session, limits) = session(Limits::default());
    peer.write_all(&fetch_request_bytes(
        &[],
        &[
            &want_oid(&commit.to_string()),
            &have_oid(&oid(9).to_string()),
            &have_oid(&blob.to_string()),
        ],
    ))
    .await
    .unwrap();

    let raw = session.read_request().await.unwrap().expect("a request");
    let mut rounds = FetchRounds::default();
    serve_fetch_round(
        &mut session,
        &raw,
        &mut rounds,
        gen.clone(),
        &doubled,
        &limits,
        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
    )
    .await
    .unwrap();

    let packets = read_pkts(&mut peer).await;
    let shapes: Vec<String> = packets.iter().map(|p| text(p)).collect();
    assert_eq!(
        shapes[..2],
        ["acknowledgments", "NAK"],
        "the missing and the non-commit have are both unrecognized, NAK alone"
    );
    assert!(packets[2].is_empty(), "the flush terminator");
}

/// A want-less round is answered empty — `acknowledgments` + `NAK`
/// + flush, no pack: the degenerate no-work shape (an up-to-date
/// client's exit round), not an error.
///
/// Contract stamp: ADR-014 §3 (want-less rounds answered empty); negotiation-captures.md
/// §Client round behavior (the captured up-to-date client's round).
#[tokio::test]
async fn want_less_round_gets_the_empty_acknowledgments_shape() {
    let doubled = repo("owner/doubled");
    let objects = MemObjects::default();
    let gen = Arc::new(MemPackGen {
        objects: objects.clone(),
    });

    let (mut peer, mut session, limits) = session(Limits::default());
    peer.write_all(&fetch_request_bytes(
        &[],
        &["thin-pack", "no-progress", &have_oid(&oid(9).to_string())],
    ))
    .await
    .unwrap();

    let raw = session.read_request().await.unwrap().expect("a request");
    let args = parse_fetch_args(&raw).unwrap();
    assert!(args.wants.is_empty() && args.no_progress);
    let mut rounds = FetchRounds::default();
    serve_fetch_round(
        &mut session,
        &raw,
        &mut rounds,
        gen.clone(),
        &doubled,
        &limits,
        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
    )
    .await
    .unwrap();

    let packets = read_pkts(&mut peer).await;
    let shapes: Vec<String> = packets.iter().map(|p| text(p)).collect();
    assert_eq!(shapes[0], "acknowledgments");
    assert_eq!(shapes[1], "NAK");
    assert_eq!(packets.len(), 3, "flush-terminated, no packfile section");
    assert!(shapes.iter().all(|s| !s.contains("packfile")));
}

/// The done round generates via `GitPackGen` with the recognized-subset
/// boundary, streams the `packfile` section — header line, band-1
/// sideband chunks (the 65000 cap forces several), the section flush —
/// and no acknowledgments section appears. The reassembled band data
/// re-decodes as the V2 sha1 pack of closure(wants) − closure(haves).
///
/// Contract stamp: ADR-014 §2 (the done round, no acknowledgments
/// section); negotiation-captures.md (the done-path response shape —
/// `packfile` + sideband + flush, POC-1 re-confirmed); ADR-004 (the
/// generation pipeline, V2 sha1 packs); ADR-009 §sideband chunk size
/// (65000 fixed per protocol).
#[tokio::test]
async fn done_round_subtracts_the_recognized_boundary_and_streams_the_packfile_section() {
    let doubled = repo("owner/doubled");
    let objects = MemObjects::default();
    let (blob_a, blob_a_bytes) = blob_fixture(b"alpha");
    let (blob_b, blob_b_bytes) = blob_fixture(b"beta");
    let (big, big_bytes) = blob_fixture(&noisy_bytes(150_000));
    let (tree_a, tree_a_bytes) =
        tree_fixture(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob_a)]);
    let (tree_b, tree_b_bytes) = tree_fixture(&[
        (gix_object::tree::EntryKind::Blob, "b.txt", blob_b),
        (gix_object::tree::EntryKind::Blob, "big.bin", big),
    ]);
    let (commit_a, commit_a_bytes) = commit_fixture(tree_a, &[], "alpha");
    let (commit_b, commit_b_bytes) = commit_fixture(tree_b, &[commit_a], "beta");
    for (id, bytes) in [
        (blob_a, blob_a_bytes),
        (blob_b, blob_b_bytes),
        (big, big_bytes),
        (tree_a, tree_a_bytes),
        (tree_b, tree_b_bytes),
        (commit_a, commit_a_bytes),
        (commit_b, commit_b_bytes),
    ] {
        objects.put(&doubled, id, bytes);
    }
    let gen = Arc::new(MemPackGen {
        objects: objects.clone(),
    });

    let (mut peer, mut session, limits) = session(Limits::default());
    peer.write_all(&fetch_request_bytes(
        &[],
        &[
            "thin-pack",
            "ofs-delta",
            "no-progress",
            &want_oid(&commit_b.to_string()),
            &have_oid(&commit_a.to_string()),
            "done",
        ],
    ))
    .await
    .unwrap();

    let raw = session.read_request().await.unwrap().expect("a request");
    let mut rounds = FetchRounds::default();
    serve_fetch_round(
        &mut session,
        &raw,
        &mut rounds,
        gen.clone(),
        &doubled,
        &limits,
        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
    )
    .await
    .unwrap();

    let packets = read_pkts(&mut peer).await;
    let shapes: Vec<String> = packets.iter().map(|p| text(p)).collect();
    assert_eq!(
        shapes[0], "packfile",
        "the section opens with the header line"
    );
    // Text frames before the band data: only the header line — no
    // acknowledgments section opens the done round (the band chunks are
    // binary; the substring scan only covers framed text lines).
    let text_lines: Vec<&String> = shapes[1..packets.len() - 1]
        .iter()
        .filter(|s| !s.is_empty() && s.chars().all(|c| c.is_ascii_graphic() || c == ' '))
        .collect();
    assert!(
        text_lines.is_empty(),
        "no text frames inside the band: {text_lines:?}"
    );
    // band-1 chunks, then the flush; the 150 KB blob forces >2 chunks.
    let band_chunks: Vec<&[u8]> = packets[1..packets.len() - 1]
        .iter()
        .map(|p| {
            assert_eq!(p[0], 1, "band 1");
            &p[1..]
        })
        .collect();
    assert!(
        band_chunks.len() > 2,
        "the 65000 cap forces several chunks: {}",
        band_chunks.len()
    );
    assert!(
        packets.last().expect("flush").is_empty(),
        "the section flush"
    );

    let pack: Vec<u8> = band_chunks.concat();
    let decoded = decode_pack(&pack);
    assert_eq!(decoded.version, gix_pack::data::Version::V2);
    let ids: BTreeSet<ObjectId> = decoded.objects.iter().map(|(id, _, _)| *id).collect();
    assert_eq!(
        ids,
        BTreeSet::from([commit_b, tree_b, blob_b, big]),
        "closure(commit_b) − closure(commit_a)"
    );
    assert_eq!(
        decoded.object_count as usize,
        ids.len(),
        "the header count matches"
    );
    let _ = decoded.trailer;
}

/// The boundary set is the *recognized* subset: a have the store lacks
/// is never honored — the done round ships the full closure, and
/// nothing else changed vs the subtraction path.
///
/// Contract stamp: ADR-014 §2 as amended by review 001 A-4 (the
/// boundary set is request haves filtered through `common_haves` —
/// never an unverified have); backend.md §trait family bullet 4 (the
/// honest boundary).
#[tokio::test]
async fn an_unverified_have_is_never_subtracted() {
    let doubled = repo("owner/doubled");
    let objects = MemObjects::default();
    let chain = two_commit_chain(&objects, &doubled);
    let gen = Arc::new(MemPackGen {
        objects: objects.clone(),
    });

    let (mut peer, mut session, limits) = session(Limits::default());
    peer.write_all(&fetch_request_bytes(
        &[],
        &[
            &want_oid(&chain.commit_b.to_string()),
            &have_oid(&oid(9).to_string()),
            "done",
        ],
    ))
    .await
    .unwrap();

    let raw = session.read_request().await.unwrap().expect("a request");
    let mut rounds = FetchRounds::default();
    serve_fetch_round(
        &mut session,
        &raw,
        &mut rounds,
        gen.clone(),
        &doubled,
        &limits,
        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
    )
    .await
    .unwrap();

    let packets = read_pkts(&mut peer).await;
    let band_chunks: Vec<&[u8]> = packets[1..packets.len() - 1]
        .iter()
        .map(|p| &p[1..])
        .collect();
    let pack: Vec<u8> = band_chunks.concat();
    let decoded = decode_pack(&pack);
    let ids: BTreeSet<ObjectId> = decoded.objects.iter().map(|(id, _, _)| *id).collect();
    assert_eq!(
        ids,
        BTreeSet::from([
            chain.commit_a,
            chain.blob_a,
            chain.tree_a,
            chain.commit_b,
            chain.blob_b,
            chain.tree_b,
        ]),
        "the missing have subtracts nothing — the full closure ships"
    );
}

/// A done round where the client already has everything (the want is
/// itself the recognized boundary) is a valid zero-object packfile —
/// header, no entries, verified trailer — a valid response, not an
/// error.
///
/// Contract stamp: ADR-014 §2 (an empty resulting pack is a valid
/// zero-object packfile response); ADR-004 (a valid pack for whatever
/// closure remains).
#[tokio::test]
async fn zero_object_done_round_is_a_valid_packfile_section() {
    let doubled = repo("owner/doubled");
    let objects = MemObjects::default();
    let (blob, blob_bytes) = blob_fixture(b"solo");
    let (tree, tree_bytes) = tree_fixture(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob)]);
    let (commit, commit_bytes) = commit_fixture(tree, &[], "solo");
    for (id, bytes) in [
        (blob, blob_bytes),
        (tree, tree_bytes),
        (commit, commit_bytes),
    ] {
        objects.put(&doubled, id, bytes);
    }
    let gen = Arc::new(MemPackGen {
        objects: objects.clone(),
    });

    let (mut peer, mut session, limits) = session(Limits::default());
    peer.write_all(&fetch_request_bytes(
        &[],
        &[
            &want_oid(&commit.to_string()),
            &have_oid(&commit.to_string()),
            "done",
        ],
    ))
    .await
    .unwrap();

    let raw = session.read_request().await.unwrap().expect("a request");
    let mut rounds = FetchRounds::default();
    serve_fetch_round(
        &mut session,
        &raw,
        &mut rounds,
        gen.clone(),
        &doubled,
        &limits,
        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
    )
    .await
    .unwrap();

    let packets = read_pkts(&mut peer).await;
    assert_eq!(text(&packets[0]), "packfile");
    let band_chunks: Vec<&[u8]> = packets[1..packets.len() - 1]
        .iter()
        .map(|p| &p[1..])
        .collect();
    let pack: Vec<u8> = band_chunks.concat();
    let decoded = decode_pack(&pack);
    assert_eq!(decoded.version, gix_pack::data::Version::V2);
    assert_eq!(decoded.object_count, 0, "the zero-object pack ships");
    assert!(decoded.objects.is_empty());
}

/// A `generate` `ObjectMissing` abort maps to the session's protocol
/// error (the entries' error-band mapping — this machine's shape) with
/// **nothing emitted**: no `packfile` header, no partial band chunks —
/// never a broken pack downstream.
///
/// Contract stamp: ADR-004 (missing objects abort the fetch — sideband
/// error band, never a broken pack); ADR-018 §7 (`ObjectMissing`); ADR-005
/// §4 (protocol errors band-map at the entries; the stateless side maps
/// at the door).
#[tokio::test]
async fn object_missing_done_round_aborts_without_emitting_a_broken_pack() {
    let doubled = repo("owner/doubled");
    let objects = MemObjects::default();
    let gen = Arc::new(MemPackGen {
        objects: objects.clone(),
    });

    let (mut peer, mut session, limits) = session(Limits::default());
    peer.write_all(&fetch_request_bytes(
        &[],
        &[&want_oid(&oid(9).to_string()), "done"],
    ))
    .await
    .unwrap();

    let raw = session.read_request().await.unwrap().expect("a request");
    let mut rounds = FetchRounds::default();
    let error = serve_fetch_round(
        &mut session,
        &raw,
        &mut rounds,
        gen.clone(),
        &doubled,
        &limits,
        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
    )
    .await
    .unwrap_err();
    match error {
        SubstrateError::Protocol(message) => {
            assert!(message.contains("does not exist"), "{message}");
            assert!(message.contains(&oid(9).to_string()), "{message}");
        }
        other => panic!("the abort is a session error: {other:?}"),
    }
    drop(session);
    let mut buf = Vec::new();
    peer.read_to_end(&mut buf).await.unwrap();
    assert_eq!(buf, Vec::<u8>::new(), "no pack byte, no header was emitted");
}

/// The budgets fail closed (ADR-009, ADR-014 §6, no new budget kinds):
/// the second fetch round of a capped session errors with nothing
/// emitted, and a round carrying more haves than the budget errors
/// before any backend call.
///
/// Contract stamp: ADR-009 (rounds/haves budgets, breach ends the
/// session); ADR-014 §6 (the loop consumes the rounds cap).
#[tokio::test]
async fn rounds_and_haves_budgets_fail_closed() {
    let doubled = repo("owner/doubled");
    let objects = MemObjects::default();
    let gen = Arc::new(MemPackGen {
        objects: objects.clone(),
    });

    // (a) rounds: a session capped at one round serves round 1 and
    // fails round 2 before anything is written.
    let limits = Limits {
        max_negotiation_rounds: 1,
        ..Limits::default()
    };
    let (mut peer, mut session, limits) = session(limits);
    peer.write_all(&fetch_request_bytes(&[], &[&want_oid(&oid(1).to_string())]))
        .await
        .unwrap();
    let raw = session.read_request().await.unwrap().expect("a request");
    let mut rounds = FetchRounds::default();
    serve_fetch_round(
        &mut session,
        &raw,
        &mut rounds,
        gen.clone(),
        &doubled,
        &limits,
        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
    )
    .await
    .unwrap();
    read_pkts(&mut peer).await;

    session.reset_request();
    peer.write_all(&fetch_request_bytes(&[], &[&want_oid(&oid(1).to_string())]))
        .await
        .unwrap();
    let raw = session.read_request().await.unwrap().expect("a request");
    let error = serve_fetch_round(
        &mut session,
        &raw,
        &mut rounds,
        gen.clone(),
        &doubled,
        &limits,
        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
    )
    .await
    .unwrap_err();
    match error {
        SubstrateError::Protocol(message) => {
            assert!(message.contains("rounds budget"), "{message}");
        }
        other => panic!("a rounds breach is a session error: {other:?}"),
    }
    drop(session);
    let mut extra = Vec::new();
    peer.read_to_end(&mut extra).await.unwrap();
    assert_eq!(extra, Vec::<u8>::new(), "the breaching round wrote nothing");

    // (b) haves: a round over the haves budget errors before any
    // backend call or wire byte.
    let limits = Limits {
        max_haves_per_round: 2,
        ..Limits::default()
    };
    let want = want_oid(&oid(1).to_string());
    let haves: Vec<String> = (2..6u8).map(|b| have_oid(&oid(b).to_string())).collect();
    let mut args = vec![want];
    args.extend(haves);
    let arg_refs: Vec<&str> = args.iter().map(String::as_str).collect();
    let raw = request(&[], &arg_refs);
    let mut rounds = FetchRounds::default();
    let sink = VecSink::default();
    let mut post = StatelessRequest::new(
        futures::io::empty(),
        sink.clone(),
        limits,
        Service::UploadPack,
    );
    let error = serve_fetch_round(
        post.response(),
        &raw,
        &mut rounds,
        gen.clone(),
        &doubled,
        &limits,
        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
    )
    .await
    .unwrap_err();
    match error {
        SubstrateError::Protocol(message) => {
            assert!(message.contains("haves budget"), "{message}");
        }
        other => panic!("a haves breach is a session error: {other:?}"),
    }
    assert_eq!(
        sink.body(),
        Vec::<u8>::new(),
        "the breaching round wrote nothing"
    );
}

/// A saturated pipeline-concurrency semaphore delays the done round's
/// `generate` call — the permit wraps the call (ADR-009's enforcement
/// point); releasing a permit lets the round complete.
///
/// Contract stamp: ADR-009 (the blocking-pool budget, the permit
/// acquired around each `GitPackGen::generate` call); backend.md
/// §concurrency model (the wire layer enforces; exhaustion delays —
/// the 503-mapped class).
#[tokio::test]
async fn a_saturated_concurrency_permit_delays_the_done_round_generate() {
    let doubled = repo("owner/doubled");
    let objects = MemObjects::default();
    let (blob, blob_bytes) = blob_fixture(b"paid");
    let (tree, tree_bytes) = tree_fixture(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob)]);
    let (commit, commit_bytes) = commit_fixture(tree, &[], "paid");
    for (id, bytes) in [
        (blob, blob_bytes),
        (tree, tree_bytes),
        (commit, commit_bytes),
    ] {
        objects.put(&doubled, id, bytes);
    }
    let gen = Arc::new(MemPackGen {
        objects: objects.clone(),
    });

    let limits = Limits::default();
    let pipeline = Arc::new(Semaphore::new(1));
    let held = pipeline.clone().acquire_owned().await.unwrap();
    let raw = request(
        &[],
        &[
            &want_oid(&commit.to_string()),
            &have_oid(&oid(9).to_string()),
            "done",
        ],
    );

    let sink = VecSink::default();
    let task = {
        let gen = gen.clone();
        let doubled = doubled.clone();
        let pipeline = pipeline.clone();
        let raw = raw.clone();
        let task_sink = sink.clone();
        tokio::spawn(async move {
            let mut post =
                StatelessRequest::new(futures::io::empty(), task_sink, limits, Service::UploadPack);
            let mut rounds = FetchRounds::default();
            serve_fetch_round(
                post.response(),
                &raw,
                &mut rounds,
                gen,
                &doubled,
                &limits,
                pipeline,
            )
            .await
            .expect("the done round completes once a permit frees");
        })
    };

    // The machine is parked on the permit: nothing finished, nothing emitted.
    tokio::time::sleep(Duration::from_millis(200)).await;
    assert!(!task.is_finished(), "the round must wait for the permit");
    assert_eq!(
        sink.body(),
        Vec::<u8>::new(),
        "nothing emitted while parked"
    );

    drop(held);
    task.await.unwrap();
    assert!(
        !sink.body().is_empty(),
        "releasing the permit lets it stream"
    );
    let lines = parse_pkt_lines(&sink.body());
    assert_eq!(text(&lines[0]), "packfile");
    let band_chunks: Vec<&[u8]> = lines[1..lines.len() - 1].iter().map(|p| &p[1..]).collect();
    let decoded = decode_pack(&band_chunks.concat());
    assert_eq!(
        decoded.object_count, 3,
        "the delayed generate ran to the pack"
    );
}

/// The machines run unchanged through the stateless response end (the
/// http door's surface): the acknowledgments shape and the packfile
/// section with identical wire text, each response one-shot
/// flush-terminated.
///
/// Contract stamp: ADR-005 §3 (two session substrates feeding one state
/// machine core); ADR-005 §2 (the stateless response rules — one-shot
/// flush, no `0002`); negotiation-captures.md §Statelessness (each POST
/// is one round, no cross-request state).
#[tokio::test]
async fn the_machine_runs_unchanged_through_the_stateless_response_end() {
    let doubled = repo("owner/doubled");
    let objects = MemObjects::default();
    let chain = two_commit_chain(&objects, &doubled);
    let gen = Arc::new(MemPackGen {
        objects: objects.clone(),
    });
    let limits = Limits::default();
    let pipeline = Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks));

    // The no-done round: acknowledgments + ACK + flush.
    let raw = request(
        &["object-format=sha1"],
        &[
            &want_oid(&chain.commit_b.to_string()),
            &have_oid(&chain.commit_a.to_string()),
        ],
    );
    let sink = VecSink::default();
    let mut post = StatelessRequest::new(
        futures::io::empty(),
        sink.clone(),
        limits,
        Service::UploadPack,
    );
    let mut rounds = FetchRounds::default();
    serve_fetch_round(
        post.response(),
        &raw,
        &mut rounds,
        gen.clone(),
        &doubled,
        &limits,
        pipeline.clone(),
    )
    .await
    .unwrap();
    assert!(post.response().is_flushed(), "the response ends at flush");
    let shapes: Vec<String> = parse_pkt_lines(&sink.body())
        .iter()
        .map(|p| text(p))
        .collect();
    assert_eq!(shapes[0], "acknowledgments");
    assert_eq!(shapes[1], format!("ACK {}", chain.commit_a));

    // The done round: packfile + band-1 chunks + flush.
    let raw = request(
        &[],
        &[
            &want_oid(&chain.commit_b.to_string()),
            &have_oid(&chain.commit_a.to_string()),
            "done",
        ],
    );
    let sink = VecSink::default();
    let mut post = StatelessRequest::new(
        futures::io::empty(),
        sink.clone(),
        limits,
        Service::UploadPack,
    );
    let mut rounds = FetchRounds::default();
    serve_fetch_round(
        post.response(),
        &raw,
        &mut rounds,
        gen.clone(),
        &doubled,
        &limits,
        pipeline.clone(),
    )
    .await
    .unwrap();
    assert!(post.response().is_flushed(), "the response ends at flush");
    let lines = parse_pkt_lines(&sink.body());
    assert_eq!(text(&lines[0]), "packfile");
    let band_chunks: Vec<&[u8]> = lines[1..lines.len() - 1].iter().map(|p| &p[1..]).collect();
    let pack: Vec<u8> = band_chunks.concat();
    let decoded = decode_pack(&pack);
    let ids: BTreeSet<ObjectId> = decoded.objects.iter().map(|(id, _, _)| *id).collect();
    assert_eq!(
        ids,
        BTreeSet::from([chain.commit_b, chain.blob_b, chain.tree_b]),
        "closure(commit_b) − closure(commit_a) over the stateless surface"
    );
    let _ = decoded.trailer;
}

/// `include-tag` is an accepted argument honored by omission (the OQ-19
/// rider-3 decision): a done round carrying it serves the identical wire
/// response — the same `packfile` section, the same closure, the same
/// pack bytes — as the identical round without it. Tag carrying is never
/// synthesized; the pack is the want closure as generated.
///
/// Contract stamp: OQ-19 rider 3 (accept-and-honor-by-omission — the
/// `thin-pack` shape, tasks/architecture/oq-19-wire-validations.md §3);
/// ADR-004 (the generation pipeline over the want closure — no tag
/// synthesis); transport.c v2.43.0 (`args.include_tag =
/// data->options.followtags`, un-gated in V2's send path —
/// `core/w5-oq19-real-client-tokens`'s capture basis);
/// tests/real_client.rs (the end-to-end validation).
#[tokio::test]
async fn include_tag_is_honored_by_omission_serving_the_identical_pack() {
    let doubled = repo("owner/doubled");
    let objects = MemObjects::default();
    let chain = two_commit_chain(&objects, &doubled);
    let gen = Arc::new(MemPackGen {
        objects: objects.clone(),
    });

    let (mut peer, mut server, limits) = session(Limits::default());
    peer.write_all(&fetch_request_bytes(
        &[],
        &[
            &want_oid(&chain.commit_b.to_string()),
            &have_oid(&chain.commit_a.to_string()),
            "done",
        ],
    ))
    .await
    .unwrap();
    let raw = server.read_request().await.unwrap().expect("a request");
    let mut rounds = FetchRounds::default();
    serve_fetch_round(
        &mut server,
        &raw,
        &mut rounds,
        gen.clone(),
        &doubled,
        &limits,
        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
    )
    .await
    .unwrap();
    let without = read_pkts(&mut peer).await;

    let (mut peer2, mut session2, limits2) = session(Limits::default());
    peer2
        .write_all(&fetch_request_bytes(
            &[],
            &[
                "include-tag",
                &want_oid(&chain.commit_b.to_string()),
                &have_oid(&chain.commit_a.to_string()),
                "done",
            ],
        ))
        .await
        .unwrap();
    let raw = session2.read_request().await.unwrap().expect("a request");
    assert!(
        parse_fetch_args(&raw).unwrap().include_tag,
        "the accepted set parsed the token"
    );
    let mut rounds = FetchRounds::default();
    serve_fetch_round(
        &mut session2,
        &raw,
        &mut rounds,
        gen.clone(),
        &doubled,
        &limits2,
        Arc::new(Semaphore::new(limits2.max_concurrent_blocking_tasks)),
    )
    .await
    .unwrap();
    let with = read_pkts(&mut peer2).await;

    assert_eq!(
        without, with,
        "the identical wire response with and without include-tag"
    );
    let band_chunks: Vec<&[u8]> = with[1..with.len() - 1].iter().map(|p| &p[1..]).collect();
    let decoded = decode_pack(&band_chunks.concat());
    let ids: BTreeSet<ObjectId> = decoded.objects.iter().map(|(id, _, _)| *id).collect();
    assert_eq!(
        ids,
        BTreeSet::from([chain.commit_b, chain.blob_b, chain.tree_b]),
        "the pack is the want closure as generated"
    );
    let _ = decoded.trailer;
}