alkgit 0.1.0

//! The wave-3 receive-pack state machine rows (core/w3-push-machine):
//! parsed request → prepare → validate → CAS → report per ADR-013
//! §3–11 as amended by ADR-020 §1 — the duplex peek's four ordering
//! rules, the stateless EOF twin, the packless CAS-only path, the
//! report's shapes byte-against push-captures.md, and the budgets
//! (the permit around prepare, the pack-size relay, fail-closed
//! breaches) — over the in-memory backend doubles.
//!
//! Rows are names + `Contract stamp:` versions, per
//! `tests/common` §The row/stamp discipline.

mod common;

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

use async_trait::async_trait;
use common::{
    blob_fixture, commit_fixture, loose_body, pack_bytes, pack_entry_base, tree_fixture,
    MemPackIngest, MemRefs,
};
use futures::io::AsyncWrite as FuturesAsyncWrite;
use futures::stream;
use gix_hash::ObjectId;
use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt};
use tokio::sync::Semaphore;

use alkgit::{
    serve_receive_pack, BodyReader, GitPackIngest, GitRefs, Limits, PackTailDiscipline,
    PreparedPush, PushOptions, RawTail, RefUpdate, RepoRecord, Request, Service, Session,
    StatelessPost, StatelessRequest, SubstrateError, Visibility, ATOMIC_PUSH_FAILURE_REASON,
    UNPACKER_ERROR_REASON,
};

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

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

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

/// 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
}

/// A create command line (`old = zero`, ADR-013 §3's shapes).
fn create_command(name: &str, new: ObjectId) -> String {
    format!("{} {new} {name}", oid(0))
}

/// A delete command line (`new = zero`).
fn delete_command(name: &str, current: ObjectId) -> String {
    format!("{current} {} {name}", oid(0))
}

/// An update command line (`old = current`, `new = target`).
fn update_command(name: &str, current: ObjectId, new: ObjectId) -> String {
    format!("{current} {new} {name}")
}

/// A fixture object: its oid and the (kind, body) its loose bytes
/// carry — the pack-entry source.
struct Fixture {
    oid: ObjectId,
    kind: gix_object::Kind,
    body: Vec<u8>,
}

fn blob_fix(content: &[u8]) -> Fixture {
    let (id, bytes) = blob_fixture(content);
    let (kind, body) = loose_body(&bytes);
    Fixture {
        oid: id,
        kind,
        body,
    }
}

fn tree_fix(entries: &[(gix_object::tree::EntryKind, &str, ObjectId)]) -> Fixture {
    let (id, bytes) = tree_fixture(entries);
    let (kind, body) = loose_body(&bytes);
    Fixture {
        oid: id,
        kind,
        body,
    }
}

fn commit_fix(tree: ObjectId, parents: &[ObjectId], message: &str) -> Fixture {
    let (id, bytes) = commit_fixture(tree, parents, message);
    let (kind, body) = loose_body(&bytes);
    Fixture {
        oid: id,
        kind,
        body,
    }
}

/// Assembles a self-contained V2 sha1 pack from the fixtures
/// (cumulative pack offsets make `pack_bytes` whole).
fn pack_from_fixtures(fixtures: &[&Fixture]) -> Vec<u8> {
    let mut offset = 0u64;
    let entries: Vec<_> = fixtures
        .iter()
        .map(|fixture| {
            let entry = pack_entry_base(fixture.kind, &fixture.body, offset);
            offset += entry.bytes_in_pack();
            entry
        })
        .collect();
    pack_bytes(&entries)
}

/// A two-commit chain fixture + its pack — the rows' ordinary push.
struct ChainPack {
    pack: Vec<u8>,
    commit_a: ObjectId,
    commit_b: ObjectId,
}

fn chain_pack() -> ChainPack {
    let blob_a = blob_fix(b"alpha");
    let blob_b = blob_fix(b"beta");
    let tree_a = tree_fix(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob_a.oid)]);
    let tree_b = tree_fix(&[
        (gix_object::tree::EntryKind::Blob, "a.txt", blob_a.oid),
        (gix_object::tree::EntryKind::Blob, "b.txt", blob_b.oid),
    ]);
    let commit_a = commit_fix(tree_a.oid, &[], "alpha");
    let commit_b = commit_fix(tree_b.oid, &[commit_a.oid], "beta");
    let pack = pack_from_fixtures(&[&blob_a, &blob_b, &tree_a, &tree_b, &commit_a, &commit_b]);
    ChainPack {
        pack,
        commit_a: commit_a.oid,
        commit_b: commit_b.oid,
    }
}

/// The duplex machine-level request bytes: the V2-framed request the
/// duplex reader consumes until `Session` gains its V0 read shape
/// (core/w3-duplex-serving re-baselines this onto the real V0 stream
/// shape) — the command section with the caps NUL-attached on the first
/// command line, the command-section flush, and — when args are present
/// (the negotiated push-options gate) — the args section behind the
/// delimiter, the mapping the push grammar consumes.
fn request_bytes(caps: &str, commands: &[String], args: &[String]) -> Vec<u8> {
    let mut out = pkt(b"command=git-receive-pack\n");
    let mut commands = commands.iter();
    if let Some(first) = commands.next() {
        let line = if caps.is_empty() {
            first.clone()
        } else {
            format!("{first}\0{caps}")
        };
        out.extend_from_slice(&pkt(line.as_bytes()));
    }
    for command in commands {
        out.extend_from_slice(&pkt(command.as_bytes()));
    }
    if args.is_empty() {
        out.extend_from_slice(FLUSH);
    } else {
        out.extend_from_slice(DELIMITER);
        for arg in args {
            out.extend_from_slice(&pkt(arg.as_bytes()));
        }
        out.extend_from_slice(FLUSH);
    }
    out
}

/// The stateless POST body chunks — the real V0 shape
/// (push-captures.md §Request; ADR-013 §3): the command section with
/// the caps NUL-attached on the first command line, the
/// command-section flush, and the pack tail's chunks after it (raw,
/// ADR-013 §3). Zero commands on this wire shape is the immediate
/// flush — the nothing-to-do/probe body.
fn body_chunks(caps: &str, commands: &[String], pack_chunks: Vec<Vec<u8>>) -> Vec<Vec<u8>> {
    let mut chunks = Vec::new();
    let mut commands = commands.iter();
    if let Some(first) = commands.next() {
        let line = if caps.is_empty() {
            first.clone()
        } else {
            format!("{first}\0{caps}")
        };
        chunks.push(pkt(line.as_bytes()));
    }
    for command in commands {
        chunks.push(pkt(command.as_bytes()));
    }
    chunks.push(FLUSH.to_vec());
    chunks.extend(pack_chunks);
    chunks
}

type ChunkStream = stream::Iter<std::vec::IntoIter<io::Result<bytes::Bytes>>>;
type PostBody = BodyReader<ChunkStream, io::Error>;

fn body_reader(chunks: Vec<Vec<u8>>) -> PostBody {
    let items: Vec<io::Result<bytes::Bytes>> = chunks
        .into_iter()
        .map(|chunk| io::Result::Ok(bytes::Bytes::from(chunk)))
        .collect();
    BodyReader::new(stream::iter(items))
}

/// An async sink whose framed bytes the stateless rows inspect after
/// the fact — the http door's response-body 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: Pin<&mut Self>,
        _cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<io::Result<usize>> {
        self.0
            .lock()
            .expect("sink is not poisoned")
            .extend_from_slice(buf);
        Poll::Ready(Ok(buf.len()))
    }

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

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

/// Pkt-line payloads of a framed byte string.
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()
}

fn payloads_as_text(frames: &[Vec<u8>]) -> Vec<String> {
    frames.iter().map(|frame| text(frame)).collect()
}

/// The sideband-wrapped report's inner pkt-line lines (ADR-013 §8):
/// asserts the framing itself — every data packet is band-1, the
/// response ends at exactly one outer flush, the band payload's own
/// framing ends at its inner flush — and returns the report's lines.
fn sideband_report_lines(packets: &[Vec<u8>]) -> Vec<String> {
    let mut payload = Vec::new();
    let mut outer_flushes = 0;
    for packet in packets {
        if packet.is_empty() {
            outer_flushes += 1;
            continue;
        }
        assert_eq!(packet[0], 1, "band-1 framing carries the report");
        payload.extend_from_slice(&packet[1..]);
    }
    assert_eq!(outer_flushes, 1, "exactly one outer flush");
    let mut inner = parse_pkt_lines(&payload);
    let last = inner.pop().expect("the inner framing is non-empty");
    assert!(
        last.is_empty(),
        "the inner flush terminates the report: {last:?}"
    );
    payloads_as_text(&inner)
}

/// The bare report's lines (no sideband was selected): the report's
/// lines, the terminating flush asserted to be the last packet.
fn bare_report_lines(packets: &[Vec<u8>]) -> Vec<String> {
    let (last, rest) = packets.split_last().expect("the report is non-empty");
    assert!(last.is_empty(), "the report ends at the flush: {last:?}");
    payloads_as_text(rest)
}

/// 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());
            return packets;
        }
        let mut payload = vec![0u8; len - 4];
        reader
            .read_exact(&mut payload)
            .await
            .expect("pkt-line payload");
        packets.push(payload);
    }
}

/// Limits with a short peek window (rows run fast; the 750 ms default
/// is asserted in `limits.rs`).
fn fast_limits() -> Limits {
    Limits {
        packless_peek: Duration::from_millis(100),
        ..Limits::default()
    }
}

/// Seeds one ref row into the refs double (row support for deletes and
/// stale-old denials).
async fn seed_ref(refs: &MemRefs, record: &RepoRecord, name: &str, value: ObjectId) {
    refs.apply_updates(
        record,
        vec![RefUpdate {
            name: name.to_owned(),
            expected: None,
            new: Some(value),
        }],
        false,
    )
    .await
    .unwrap();
}

async fn list_ref(refs: &MemRefs, record: &RepoRecord, name: &str) -> Option<ObjectId> {
    refs.list_refs(record)
        .await
        .unwrap()
        .into_iter()
        .find(|line| line.name == name)
        .and_then(|line| line.oid)
}

/// Spawns the duplex machine — the session machinery built inside the
/// task over the client's stream. Rows where the client waits for the
/// report before closing (the packless drain's shape) join after
/// reading + dropping; every other row writes the request (+ pack)
/// first and joins immediately.
fn spawn_duplex_machine(
    stream: tokio::io::DuplexStream,
    refs: Arc<dyn GitRefs>,
    ingest: Arc<dyn GitPackIngest>,
    record: RepoRecord,
    limits: Limits,
    push_options_negotiated: bool,
    pipeline: Arc<Semaphore>,
) -> tokio::task::JoinHandle<Result<(), SubstrateError>> {
    tokio::spawn(async move {
        let mut session = Session::new(stream, limits, Service::ReceivePack);
        let request = session
            .read_request()
            .await
            .unwrap()
            .expect("the invocation's request");
        let tail = session.take_raw_tail().unwrap();
        serve_receive_pack(
            &mut session,
            &request,
            tail,
            push_options_negotiated,
            refs,
            ingest,
            &record,
            &limits,
            PackTailDiscipline::Duplex,
            pipeline,
        )
        .await
    })
}

/// Drives the stateless machine inline over one POST body (no packless
/// drain exists on this discipline — no client cooperation needed);
/// returns the outcome and the response sink.
async fn stateless_machine(
    chunks: Vec<Vec<u8>>,
    refs: Arc<dyn GitRefs>,
    ingest: Arc<dyn GitPackIngest>,
    record: &RepoRecord,
    limits: Limits,
    push_options_negotiated: bool,
) -> (Result<(), SubstrateError>, VecSink) {
    let sink = VecSink::default();
    let mut post = StatelessRequest::new(
        body_reader(chunks),
        sink.clone(),
        limits,
        Service::ReceivePack,
    );
    // The stateless POST body parses V0-framed (ADR-013 §3); the
    // nothing-to-do body (an immediate flush) resolves to the probe at
    // the substrate (ADR-005 §2's service-agnostic probe rule), so the
    // machine-level nothing-to-do shape composes here directly: the
    // empty request, no tail.
    let (command, tail) = match post.read_post().await.expect("the POST parses") {
        StatelessPost::Command(command) => (command, post.take_raw_tail().unwrap()),
        StatelessPost::Probe => (
            Request {
                command: "git-receive-pack".to_owned(),
                pre_lines: Vec::new(),
                args: Vec::new(),
            },
            RawTail::new(body_reader(Vec::new()), Arc::new(AtomicU64::new(0))),
        ),
    };
    let outcome = serve_receive_pack(
        post.response(),
        &command,
        tail,
        push_options_negotiated,
        refs,
        ingest,
        record,
        &limits,
        PackTailDiscipline::Stateless,
        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
    )
    .await;
    (outcome, sink)
}

/// A prepare double that counts its invocations (rows asserting the
/// machine's call discipline: prepare runs under the permit; never on
/// the packless path).
struct PrepareSignal {
    inner: MemPackIngest,
    runs: Arc<AtomicUsize>,
}

impl PrepareSignal {
    fn new(run_count: Arc<AtomicUsize>) -> Self {
        PrepareSignal {
            inner: MemPackIngest::default(),
            runs: run_count,
        }
    }
}

#[async_trait]
impl GitPackIngest for PrepareSignal {
    async fn prepare(
        &self,
        record: &RepoRecord,
        updates: Vec<RefUpdate>,
        pack: Box<dyn io::Read + Send>,
        push_options: Option<PushOptions>,
        limits: Limits,
    ) -> Result<PreparedPush, alkgit::StorageError> {
        self.runs.fetch_add(1, Ordering::SeqCst);
        self.inner
            .prepare(record, updates, pack, push_options, limits)
            .await
    }
}

fn session_error(result: Result<(), SubstrateError>) -> SubstrateError {
    match result {
        Err(error) => error,
        Ok(()) => panic!("expected a session error"),
    }
}

fn protocol_message(error: SubstrateError) -> String {
    let SubstrateError::Protocol(message) = error else {
        panic!("expected a protocol-class session error: {error:?}");
    };
    message
}

/// An ordinary push (commands + pack, dribbled three ways) ingests via
/// `prepare` under the permit, validates the names, applies via the
/// one CAS call, and reports `unpack ok` + per-ref `ok` lines in
/// request order — sideband-framed (band-1 payload self-framed, the
/// inner flush inside the band, one outer flush last) against
/// push-captures.md §Status report; the pack's objects land in the
/// backend namespace and the refs row-table reflects the CAS.
///
/// Contract stamp: ADR-013 §3 (the request shape), §6–7 (prepare →
/// name → CAS — the one CAS home), §8 (the sideband report, request
/// order); ADR-020 §1 (the peek's full-header branch — the peeked
/// bytes prepend the bridged tail; the first pack byte rides its own
/// chunk, reassembled with zero loss); ADR-009 (the permit wraps the
/// prepare call); push-captures.md §Status report (the byte-level
/// basis); tests/common §MemPackIngest (the ingestion double).
#[tokio::test]
async fn ordinary_push_ingests_applies_and_reports_in_request_order() {
    let chain = chain_pack();
    let record = repo("owner/doubled");
    let mem_refs = MemRefs::default();
    let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
    let mem_ingest = MemPackIngest::default();
    let ingest = Arc::new(mem_ingest.clone()) as Arc<dyn GitPackIngest>;
    let commands = [
        create_command("refs/heads/main", chain.commit_b),
        create_command("refs/heads/dev", chain.commit_a),
    ];

    let (mut client, stream) = duplex(1 << 20);
    client
        .write_all(&request_bytes(
            "report-status side-band-64k",
            &commands,
            &[],
        ))
        .await
        .unwrap();
    client.write_all(&chain.pack[..1]).await.unwrap();
    client.write_all(&chain.pack[1..24]).await.unwrap();
    client.write_all(&chain.pack[24..]).await.unwrap();

    let limits = fast_limits();
    let machine = spawn_duplex_machine(
        stream,
        Arc::clone(&refs),
        Arc::clone(&ingest),
        record.clone(),
        limits,
        false,
        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
    );

    let packets = read_pkts(&mut client).await;
    assert_eq!(
        sideband_report_lines(&packets),
        vec![
            "unpack ok".to_owned(),
            "ok refs/heads/main".to_owned(),
            "ok refs/heads/dev".to_owned(),
        ],
        "the report's lines in request order, band-1 framed with the inner flush"
    );
    drop(client);
    let outcome = tokio::time::timeout(Duration::from_secs(5), machine)
        .await
        .expect("the machine completes")
        .unwrap();
    assert!(outcome.is_ok(), "the machine ends cleanly: {outcome:?}");
    assert_eq!(
        list_ref(&mem_refs, &record, "refs/heads/main").await,
        Some(chain.commit_b),
        "the first command applied through the CAS home"
    );
    assert_eq!(
        list_ref(&mem_refs, &record, "refs/heads/dev").await,
        Some(chain.commit_a),
        "the second command applied"
    );
    assert!(
        mem_ingest.objects().contains(&record, &chain.commit_b),
        "the pushed objects landed in the backend namespace"
    );
}

/// A packless delete-only push (duplex) takes the peek's packless
/// path: no prepare call runs (CAS-only), the CAS deletes through the
/// same ref home, the report is the same shape, and after the report
/// the tail drains to EOF — the client stays open until the report and
/// closing afterward ends the session cleanly, never a mid-stream RST
/// (the drain's Ok completion is the observable).
///
/// Contract stamp: ADR-020 §1 rule (ii) (a timeout with zero pack
/// bytes takes the packless, CAS-only path) and rule (iv) (drain after
/// the report, never a mid-stream RST — the client stays open until
/// the report); ADR-020 §3 (the client's write-side-open shape);
/// ADR-013 §7 (the same one-transaction CAS home, applied exactly as a
/// pushed pack's updates would be; the same report); ADR-009 (the peek
/// window is `Limits::packless_peek`); poco-findings.md (the ~40 ms
/// datum — the report shape is the capture-level acceptance);
/// push-captures.md §Status report.
#[tokio::test]
async fn packless_delete_push_takes_the_cas_only_path_with_a_clean_drain() {
    let record = repo("owner/doubled");
    let current = oid(0x1b);
    let mem_refs = MemRefs::default();
    seed_ref(&mem_refs, &record, "refs/heads/gone", current).await;
    let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
    let runs = Arc::new(AtomicUsize::new(0));
    let ingest = Arc::new(PrepareSignal::new(Arc::clone(&runs))) as Arc<dyn GitPackIngest>;

    let (mut client, stream) = duplex(1 << 20);
    client
        .write_all(&request_bytes(
            "report-status side-band-64k",
            &[delete_command("refs/heads/gone", current)],
            &[],
        ))
        .await
        .unwrap();

    let limits = fast_limits();
    let machine = spawn_duplex_machine(
        stream,
        Arc::clone(&refs),
        Arc::clone(&ingest),
        record.clone(),
        limits,
        false,
        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
    );

    // The client waits for the report (ADR-020 §3), then closes.
    let packets = read_pkts(&mut client).await;
    assert_eq!(
        sideband_report_lines(&packets),
        vec!["unpack ok".to_owned(), "ok refs/heads/gone".to_owned()],
        "the packless report shape, no pack involved"
    );
    drop(client);
    let outcome = tokio::time::timeout(Duration::from_secs(5), machine)
        .await
        .expect("the drain ends when the client closes")
        .unwrap();
    assert!(outcome.is_ok(), "the session drains cleanly: {outcome:?}");
    assert_eq!(
        runs.load(Ordering::SeqCst),
        0,
        "no prepare call runs on the packless path"
    );
    assert_eq!(
        list_ref(&mem_refs, &record, "refs/heads/gone").await,
        None,
        "the CAS deleted the ref"
    );
}

/// A packless create-from-nothing command applies through the same one
/// CAS home (ADR-013 §7 exactly as a pushed pack's updates would — the
/// v1 policy leg is CAS-old-value only, non-FF allowed, ADR-013 §7).
///
/// Contract stamp: ADR-020 §1 rule (ii) (delete/create-from-nothing
/// are the packless path's in-scope commands); ADR-013 §7 (CAS-only,
/// the packless create reaches the same CAS home); push-captures.md
/// §CAS / race semantics.
#[tokio::test]
async fn packless_create_applies_through_the_cas_home() {
    let record = repo("owner/doubled");
    let target = oid(0x2c);
    let mem_refs = MemRefs::default();
    let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
    let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;

    let (mut client, stream) = duplex(1 << 20);
    client
        .write_all(&request_bytes(
            "report-status",
            &[create_command("refs/heads/born", target)],
            &[],
        ))
        .await
        .unwrap();

    let limits = fast_limits();
    let machine = spawn_duplex_machine(
        stream,
        Arc::clone(&refs),
        Arc::clone(&ingest),
        record.clone(),
        limits,
        false,
        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
    );

    let packets = read_pkts(&mut client).await;
    assert_eq!(
        bare_report_lines(&packets),
        vec!["unpack ok".to_owned(), "ok refs/heads/born".to_owned()],
        "the bare report shape without sideband selection"
    );
    drop(client);
    assert!(tokio::time::timeout(Duration::from_secs(5), machine)
        .await
        .unwrap()
        .unwrap()
        .is_ok());
    assert_eq!(
        list_ref(&mem_refs, &record, "refs/heads/born").await,
        Some(target),
        "the create applied through the CAS home"
    );
}

/// A timeout with a partial pack-header byte string is the malformed-
/// push session error — never a silent packless downgrade (a half-sent
/// pack cannot be re-interpreted as absence): the session errors, no
/// ref applies, no report runs.
///
/// Contract stamp: ADR-020 §1 rule (iii); ADR-005 §4 (the session
/// error's wire mapping is the entries'); ADR-009 (fail-closed);
/// push-captures.md (vs. the stdio shape — the duplex amendment is the
/// pinned machine behavior).
#[tokio::test]
async fn a_partial_header_at_peek_timeout_is_a_malformed_push_session_error() {
    let record = repo("owner/doubled");
    let mem_refs = MemRefs::default();
    let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
    let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;

    let (mut client, stream) = duplex(1 << 20);
    client
        .write_all(&request_bytes(
            "report-status",
            &[create_command("refs/heads/main", oid(0x2c))],
            &[],
        ))
        .await
        .unwrap();
    client.write_all(b"PA").await.unwrap();

    let limits = fast_limits();
    let machine = spawn_duplex_machine(
        stream,
        Arc::clone(&refs),
        Arc::clone(&ingest),
        record.clone(),
        limits,
        false,
        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
    );
    let outcome = tokio::time::timeout(Duration::from_secs(5), machine)
        .await
        .expect("the peek's window times out — no park forever")
        .unwrap();
    let message = protocol_message(session_error(outcome));
    assert!(message.contains("pack-header"), "{message}");
    assert!(message.contains("malformed push"), "{message}");
    assert_eq!(
        list_ref(&mem_refs, &record, "refs/heads/main").await,
        None,
        "no ref applied through the malformed push"
    );
}

/// An atomic push whose CAS denies one ref reports the CAS's outcomes
/// verbatim in request order: the denied ref keeps its own server-
/// chosen reason, the otherwise-succeeding refs report `atomic push
/// failure`, and nothing applies (the one-transaction home rolled the
/// whole push back).
///
/// Contract stamp: ADR-013 §10 (the atomic-abort mapping, verbatim);
/// ADR-013 §7 (the single CAS home + per-ref denials as outcomes);
/// push-captures.md §Status report (the captured pair), §CAS / race
/// semantics; tests/common §CAS semantics (the double's transaction
/// shape).
#[tokio::test]
async fn an_atomic_cas_denial_frames_the_outcomes_verbatim() {
    let record = repo("owner/doubled");
    let chain = chain_pack();
    let mem_refs = MemRefs::default();
    let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
    let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;
    let commands = [
        create_command("refs/heads/main", chain.commit_b),
        update_command("refs/heads/dev", oid(0xaa), chain.commit_a),
    ];

    let (mut client, stream) = duplex(1 << 20);
    client
        .write_all(&request_bytes(
            "report-status side-band-64k atomic",
            &commands,
            &[],
        ))
        .await
        .unwrap();
    client.write_all(&chain.pack).await.unwrap();

    let limits = fast_limits();
    let machine = spawn_duplex_machine(
        stream,
        Arc::clone(&refs),
        Arc::clone(&ingest),
        record.clone(),
        limits,
        false,
        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
    );

    let packets = read_pkts(&mut client).await;
    assert_eq!(
        sideband_report_lines(&packets),
        vec![
            "unpack ok".to_owned(),
            format!("ng refs/heads/main {ATOMIC_PUSH_FAILURE_REASON}"),
            "ng refs/heads/dev stale info".to_owned(),
        ],
        "the aborted ref reports atomic push failure; the denied ref keeps its own reason"
    );
    drop(client);
    assert!(tokio::time::timeout(Duration::from_secs(5), machine)
        .await
        .unwrap()
        .unwrap()
        .is_ok());
    assert_eq!(
        list_ref(&mem_refs, &record, "refs/heads/main").await,
        None,
        "the atomic rollback applied nothing"
    );
    assert_eq!(
        list_ref(&mem_refs, &record, "refs/heads/dev").await,
        None,
        "the denied ref did not apply"
    );
}

/// With `atomic` negotiated and the name validation rejecting a ref,
/// the transaction aborts whole — no CAS call runs at all (a partial
/// apply is precisely what atomic forbids): the rejected ref keeps
/// `funny refname` (its own pinned reason, the denied-ref rule) and
/// the valid refs report `atomic push failure` — the captured upstream
/// shape.
///
/// Contract stamp: ADR-013 §10 (the atomic abort on any failure), §7
/// (a rejected refname keeps `funny refname`), §9 (the rejected name
/// never reaches the CAS), §8 (reasons verbatim); push-captures.md
/// §Status report (the captured pair's shape); poco-findings.md (the
/// live-captured atomic-abort-on-a-name-check).
#[tokio::test]
async fn an_atomic_push_with_a_rejected_name_aborts_the_whole_push() {
    let record = repo("owner/doubled");
    let chain = chain_pack();
    let mem_refs = MemRefs::default();
    let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
    let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;
    let commands = [
        create_command("refs/heads/../evil", chain.commit_b),
        create_command("refs/heads/main", chain.commit_b),
    ];

    let (mut client, stream) = duplex(1 << 20);
    client
        .write_all(&request_bytes(
            "report-status side-band-64k atomic",
            &commands,
            &[],
        ))
        .await
        .unwrap();
    client.write_all(&chain.pack).await.unwrap();

    let limits = fast_limits();
    let machine = spawn_duplex_machine(
        stream,
        Arc::clone(&refs),
        Arc::clone(&ingest),
        record.clone(),
        limits,
        false,
        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
    );

    let packets = read_pkts(&mut client).await;
    assert_eq!(
        sideband_report_lines(&packets),
        vec![
            "unpack ok".to_owned(),
            "ng refs/heads/../evil funny refname".to_owned(),
            format!("ng refs/heads/main {ATOMIC_PUSH_FAILURE_REASON}"),
        ],
        "the rejected name keeps its pinned reason; the aborted refs report the abort"
    );
    drop(client);
    assert!(tokio::time::timeout(Duration::from_secs(5), machine)
        .await
        .unwrap()
        .unwrap()
        .is_ok());
    assert_eq!(
        list_ref(&mem_refs, &record, "refs/heads/main").await,
        None,
        "no ref applied — the atomic push aborted whole"
    );
    assert_eq!(
        mem_refs.list_refs(&record).await.unwrap().len(),
        0,
        "the backend's ref table never saw any name"
    );
}

/// A rejected ref name reports `ng <ref> funny refname` (the pinned
/// observed text) and is excluded from the CAS call — the backend's
/// ref table never sees an invalid name — while the remaining refs
/// apply in their request order.
///
/// Contract stamp: ADR-013 §7 (per-ref validation: name first), §9
/// (the check's home + the exclusion from CAS), §8 (the report's
/// per-ref reasons verbatim, in request order); push-captures.md
/// §Raw stdio (the `funny refname` ground truth).
#[tokio::test]
async fn a_rejected_refname_reports_funny_refname_and_is_excluded_from_the_cas() {
    let record = repo("owner/doubled");
    let chain = chain_pack();
    let mem_refs = MemRefs::default();
    let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
    let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;
    let commands = [
        create_command("refs/heads/../evil", chain.commit_b),
        create_command("refs/heads/main", chain.commit_a),
    ];

    let (mut client, stream) = duplex(1 << 20);
    client
        .write_all(&request_bytes("report-status", &commands, &[]))
        .await
        .unwrap();
    client.write_all(&chain.pack).await.unwrap();

    let limits = fast_limits();
    let machine = spawn_duplex_machine(
        stream,
        Arc::clone(&refs),
        Arc::clone(&ingest),
        record.clone(),
        limits,
        false,
        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
    );

    let packets = read_pkts(&mut client).await;
    assert_eq!(
        bare_report_lines(&packets),
        vec![
            "unpack ok".to_owned(),
            "ng refs/heads/../evil funny refname".to_owned(),
            "ok refs/heads/main".to_owned(),
        ],
        "report in request order — the rejected slot keeps its place"
    );
    drop(client);
    assert!(tokio::time::timeout(Duration::from_secs(5), machine)
        .await
        .unwrap()
        .unwrap()
        .is_ok());
    assert_eq!(
        list_ref(&mem_refs, &record, "refs/heads/main").await,
        Some(chain.commit_a),
        "the remaining refs applied"
    );
    assert_eq!(
        mem_refs.list_refs(&record).await.unwrap().len(),
        1,
        "the invalid name never reached the CAS home"
    );
}

/// A pack-level failure (`unpack ng <reason>`) reports the observed
/// raw-stdio shape — all refs `ng <ref> unpacker error` — with no CAS
/// call at all: nothing applies through a failed unpack.
///
/// Contract stamp: ADR-013 §8 (the unpack ng leg + all-refs
/// `unpacker error`, the pack-level reason forwarded verbatim); ADR-018
/// §5 (the updates round-trip verbatim on the ng path — the all-refs
/// names survive); ADR-013 §7 (per-ref checks run AFTER unpack);
/// push-captures.md §Raw stdio (corrupt pack), §Pack-read ordering
/// (definitive unpack-first order).
#[tokio::test]
async fn an_unpack_ng_leg_reports_all_refs_unpacker_error_without_a_cas_call() {
    let record = repo("owner/doubled");
    let mem_refs = MemRefs::default();
    let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
    let runs = Arc::new(AtomicUsize::new(0));
    let ingest = Arc::new(PrepareSignal::new(Arc::clone(&runs))) as Arc<dyn GitPackIngest>;
    let commands = [
        create_command("refs/heads/main", oid(0x2c)),
        create_command("refs/heads/other", oid(0x3d)),
    ];

    let (mut client, stream) = duplex(1 << 20);
    client
        .write_all(&request_bytes(
            "report-status side-band-64k",
            &commands,
            &[],
        ))
        .await
        .unwrap();
    // A corrupt pack header (a full 12-byte header with a bad
    // signature): the pack-level failure is detected without waiting
    // for more bytes — the stdio capture's shape (`unpack protocol
    // error (pack signature mismatch detected)`), not a short stall
    // (a client stalling mid-header parks the blocking ingestion on
    // bytes only its EOF resolves — the entries' wall clock's victim,
    // never an in-machine error path).
    client.write_all(b"NOPE-NOPE-NO").await.unwrap();

    let limits = fast_limits();
    let machine = spawn_duplex_machine(
        stream,
        Arc::clone(&refs),
        Arc::clone(&ingest),
        record.clone(),
        limits,
        false,
        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
    );

    let packets = read_pkts(&mut client).await;
    let lines = sideband_report_lines(&packets);
    assert!(
        lines[0].starts_with("unpack ng "),
        "pack-level reason verbatim: {lines:?}"
    );
    assert!(!lines[0].contains(UNPACKER_ERROR_REASON), "{lines:?}");
    assert_eq!(
        lines[1..],
        vec![
            format!("ng refs/heads/main {UNPACKER_ERROR_REASON}"),
            format!("ng refs/heads/other {UNPACKER_ERROR_REASON}"),
        ],
        "all refs report the unpacker-error shape"
    );
    drop(client);
    assert!(tokio::time::timeout(Duration::from_secs(5), machine)
        .await
        .unwrap()
        .unwrap()
        .is_ok());
    assert_eq!(
        runs.load(Ordering::SeqCst),
        1,
        "the prepare call ran once (the pack failed inside it)"
    );
    assert_eq!(
        mem_refs.list_refs(&record).await.unwrap().len(),
        0,
        "no ref applied through the failed unpack"
    );
}

/// The nothing-to-do shape (an immediate flush, no commands) replies
/// flush with no report and no other byte — no peek, no CAS, no report
/// section, no prepare call.
///
/// Contract stamp: ADR-013 §3 (the nothing-to-do shape: reply flush,
/// no report); push-captures.md §CAS / race semantics (the empty
/// command list's client-side shape).
#[tokio::test]
async fn the_nothing_to_do_shape_replies_flush_with_no_report() {
    let record = repo("owner/doubled");
    let mem_refs = MemRefs::default();
    let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
    let runs = Arc::new(AtomicUsize::new(0));
    let ingest = Arc::new(PrepareSignal::new(Arc::clone(&runs))) as Arc<dyn GitPackIngest>;

    let (mut client, stream) = duplex(1 << 20);
    client
        .write_all(&request_bytes("report-status side-band-64k", &[], &[]))
        .await
        .unwrap();

    let limits = fast_limits();
    let machine = spawn_duplex_machine(
        stream,
        Arc::clone(&refs),
        Arc::clone(&ingest),
        record.clone(),
        limits,
        false,
        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
    );

    let packets = read_pkts(&mut client).await;
    assert_eq!(packets.len(), 1, "exactly the reply flush");
    assert!(packets[0].is_empty(), "the reply flush has no payload");
    drop(client);
    assert!(tokio::time::timeout(Duration::from_secs(5), machine)
        .await
        .unwrap()
        .unwrap()
        .is_ok());
    assert_eq!(
        runs.load(Ordering::SeqCst),
        0,
        "no prepare call ran on the nothing-to-do shape"
    );
    assert_eq!(
        mem_refs.list_refs(&record).await.unwrap().len(),
        0,
        "no CAS ran"
    );
}

/// The grammar's up-front rejections surface as session errors before
/// any pack byte is read: shallow lines (ADR-013 §4, rejected up
/// front) and an unadvertised object-format (the push-side check, the
/// fetch rule). Nothing applies, nothing reports.
///
/// Contract stamp: ADR-013 §4 (shallow's up-front rejection), §2/§4
/// (the object-format check rides the fetch rule); ADR-005 §4 (the
/// session error's wire mapping is the entries'); ADR-013 §7 (the
/// per-ref checks run after unpack — none of this ever reaches them).
#[tokio::test]
async fn shallow_and_object_format_reject_up_front_as_session_errors() {
    let record = repo("owner/doubled");
    let mem_refs = MemRefs::default();
    let refs = Arc::new(MemRefs::default()) as Arc<dyn GitRefs>;
    let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;

    let commands = [
        format!("shallow {}", oid(0x1b)),
        create_command("refs/heads/main", oid(0x2c)),
    ];
    let (mut client, stream) = duplex(1 << 20);
    client
        .write_all(&request_bytes("", &commands, &[]))
        .await
        .unwrap();
    let machine = spawn_duplex_machine(
        stream,
        Arc::clone(&refs),
        Arc::clone(&ingest),
        record.clone(),
        fast_limits(),
        false,
        Arc::new(Semaphore::new(1)),
    );
    let message = protocol_message(session_error(machine.await.unwrap()));
    assert!(message.contains("shallow"), "{message}");
    assert_eq!(mem_refs.list_refs(&record).await.unwrap().len(), 0);

    let commands = [create_command("refs/heads/main", oid(0x2c))];
    let (mut client, stream) = duplex(1 << 20);
    client
        .write_all(&request_bytes("object-format=sha1024", &commands, &[]))
        .await
        .unwrap();
    let machine = spawn_duplex_machine(
        stream,
        Arc::clone(&refs),
        Arc::clone(&ingest),
        record.clone(),
        fast_limits(),
        false,
        Arc::new(Semaphore::new(1)),
    );
    let message = protocol_message(session_error(machine.await.unwrap()));
    assert!(message.contains("object-format"), "{message}");
    assert_eq!(mem_refs.list_refs(&record).await.unwrap().len(), 0);
}

/// The permit wraps prepare (ADR-009's enforcement point): a saturated
/// pipeline semaphore delays the prepare call — nothing runs until a
/// permit frees — and once it frees the push completes with its
/// report (the pre-written pack parks inside the bounded bridge the
/// whole time).
///
/// Contract stamp: ADR-009 (the pipeline-concurrency budget's
/// enforcement point around the backend pipeline calls — the same
/// semaphore the fetch machine's `generate` runs under); backend.md
/// §concurrency model (the permit stays wire-side, the blocking thread
/// impl-side); ADR-013 §6's surviving budgets paragraph.
#[tokio::test]
async fn the_pipeline_concurrency_permit_wraps_prepare() {
    let record = repo("owner/doubled");
    let chain = chain_pack();
    let refs = Arc::new(MemRefs::default()) as Arc<dyn GitRefs>;
    let runs = Arc::new(AtomicUsize::new(0));
    let ingest = Arc::new(PrepareSignal::new(Arc::clone(&runs))) as Arc<dyn GitPackIngest>;

    let (mut client, stream) = duplex(1 << 20);
    client
        .write_all(&request_bytes(
            "report-status",
            &[create_command("refs/heads/main", chain.commit_b)],
            &[],
        ))
        .await
        .unwrap();
    client.write_all(&chain.pack).await.unwrap();

    let limits = fast_limits();
    let pipeline = Arc::new(Semaphore::new(1));
    let held = pipeline
        .acquire()
        .await
        .expect("the rows hold the permit first");
    let machine = spawn_duplex_machine(
        stream,
        Arc::clone(&refs),
        Arc::clone(&ingest),
        record.clone(),
        limits,
        false,
        Arc::clone(&pipeline),
    );
    tokio::time::sleep(Duration::from_millis(200)).await;
    assert_eq!(
        runs.load(Ordering::SeqCst),
        0,
        "a saturated semaphore delays the prepare call"
    );
    drop(held);
    // The report is read while the client end stays alive: the relay
    // (joined by the machine after the report) ends at the tail's EOF
    // — the client's close past the report.
    let packets = read_pkts(&mut client).await;
    drop(client);
    let outcome = tokio::time::timeout(Duration::from_secs(5), machine)
        .await
        .expect("the machine completes once the permit frees")
        .unwrap();
    assert!(outcome.is_ok());
    assert_eq!(
        runs.load(Ordering::SeqCst),
        1,
        "prepare ran exactly once, under the permit"
    );
    let lines = bare_report_lines(&packets);
    assert_eq!(
        lines,
        vec!["unpack ok".to_owned(), "ok refs/heads/main".to_owned()]
    );
}

/// The wire-side pack-size budget breaches fail closed (ADR-009): the
/// relay stops before the breaching bytes reach the ingestion, the
/// session returns the budget failure (the client-fault class), and no
/// ref applies, no report runs.
///
/// Contract stamp: ADR-009 (the `max_pack_size` budget, fail-closed,
/// its client-fault class — 413-mapped at the door on stateless);
/// pack_source.rs (the relay's wire-side enforcement point); ADR-020 §2
/// (the impl-side counting-reader is the impl's own defense).
#[tokio::test]
async fn a_pack_size_budget_breach_fails_the_session_closed() {
    let record = repo("owner/doubled");
    let chain = chain_pack();
    let mem_refs = MemRefs::default();
    let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
    let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;

    let (mut client, stream) = duplex(1 << 20);
    client
        .write_all(&request_bytes(
            "report-status",
            &[create_command("refs/heads/main", chain.commit_b)],
            &[],
        ))
        .await
        .unwrap();
    client.write_all(&chain.pack).await.unwrap();

    let limits = Limits {
        max_pack_size: 2,
        packless_peek: Duration::from_millis(100),
        ..Limits::default()
    };
    let machine = spawn_duplex_machine(
        stream,
        Arc::clone(&refs),
        Arc::clone(&ingest),
        record.clone(),
        limits,
        false,
        Arc::new(Semaphore::new(4)),
    );
    let outcome = tokio::time::timeout(Duration::from_secs(5), machine)
        .await
        .expect("the breach fails the session promptly")
        .unwrap();
    let error = session_error(outcome);
    let SubstrateError::Protocol(message) = error else {
        panic!("the budget breach is the client-fault class: {error:?}");
    };
    assert!(message.contains("max_pack_size"), "{message}");
    assert_eq!(
        mem_refs.list_refs(&record).await.unwrap().len(),
        0,
        "no ref applied through the breached push"
    );
}

/// The negotiated push-options section (ADR-013 §11, the args section
/// when the gate is on) is consumed by the grammar **before** the peek
/// (rule (i)): the section parses, the peek watches only for the pack's
/// first bytes, and the push proceeds ordinarily.
///
/// Contract stamp: ADR-013 §11 (the section's grammar; verbatim,
/// un-interpreted); ADR-020 §1 rule (i); push-captures.md §Request.
#[tokio::test]
async fn the_negotiated_push_options_section_is_consumed_before_the_peek() {
    let record = repo("owner/doubled");
    let chain = chain_pack();
    let mem_refs = MemRefs::default();
    let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
    let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;
    let commands = [create_command("refs/heads/main", chain.commit_b)];

    let (mut client, stream) = duplex(1 << 20);
    client
        .write_all(&request_bytes(
            "report-status push-options",
            &commands,
            &["private=yes".to_owned()],
        ))
        .await
        .unwrap();
    client.write_all(&chain.pack).await.unwrap();

    let limits = fast_limits();
    let machine = spawn_duplex_machine(
        stream,
        Arc::clone(&refs),
        Arc::clone(&ingest),
        record.clone(),
        limits,
        true,
        Arc::new(Semaphore::new(4)),
    );

    let packets = read_pkts(&mut client).await;
    assert_eq!(
        bare_report_lines(&packets),
        vec!["unpack ok".to_owned(), "ok refs/heads/main".to_owned()],
        "the push flowed through, the options section consumed ahead of the peek"
    );
    drop(client);
    assert!(tokio::time::timeout(Duration::from_secs(5), machine)
        .await
        .unwrap()
        .unwrap()
        .is_ok());
    assert_eq!(
        list_ref(&mem_refs, &record, "refs/heads/main").await,
        Some(chain.commit_b)
    );
}

/// Selecting `push-options` when the gate is off fails closed (the
/// grammar's rule): the session errors before the peek, no ref
/// applies.
///
/// Contract stamp: ADR-013 §11 (the gate's default-off posture — the
/// selection is out of grammar), §2 (only advertised caps select);
/// push.rs (the grammar's fail-closed rule).
#[tokio::test]
async fn selecting_push_options_with_the_gate_off_fails_closed() {
    let record = repo("owner/doubled");
    let mem_refs = MemRefs::default();
    let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
    let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;

    let (mut client, stream) = duplex(1 << 20);
    client
        .write_all(&request_bytes(
            "report-status push-options",
            &[create_command("refs/heads/main", oid(0x2c))],
            &["private=yes".to_owned()],
        ))
        .await
        .unwrap();
    let machine = spawn_duplex_machine(
        stream,
        Arc::clone(&refs),
        Arc::clone(&ingest),
        record.clone(),
        fast_limits(),
        false,
        Arc::new(Semaphore::new(1)),
    );
    let message = protocol_message(session_error(machine.await.unwrap()));
    assert!(message.contains("not advertised"), "{message}");
    assert_eq!(mem_refs.list_refs(&record).await.unwrap().len(), 0);
}

/// The stateless tail ingests dribbled pack bytes — including a pack
/// header split across body chunks (the EOF-driven twin of the duplex
/// peek's reassembly) — and reports the same shapes: `unpack ok` +
/// per-ref `ok` lines in request order, sideband-framed; the objects
/// land.
///
/// Contract stamp: ADR-020 §1 (the EOF-driven twin — the stateless POST
/// body's EOF is unambiguous); ADR-013 §3 (one POST body is one
/// invocation, the pack raw to the body's end), §7–8 (the CAS + the
/// report, request order); push-captures.md §Status report.
#[tokio::test]
async fn the_stateless_tail_ingests_dribbled_pack_bytes() {
    let record = repo("owner/doubled");
    let chain = chain_pack();
    let mem_refs = MemRefs::default();
    let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
    let mem_ingest = MemPackIngest::default();
    let ingest = Arc::new(mem_ingest.clone()) as Arc<dyn GitPackIngest>;
    let commands = [
        create_command("refs/heads/main", chain.commit_b),
        create_command("refs/heads/dev", chain.commit_a),
    ];

    let (outcome, sink) = stateless_machine(
        body_chunks(
            "report-status side-band-64k",
            &commands,
            vec![
                chain.pack[..1].to_vec(),
                chain.pack[1..24].to_vec(),
                chain.pack[24..].to_vec(),
            ],
        ),
        Arc::clone(&refs),
        Arc::clone(&ingest),
        &record,
        fast_limits(),
        false,
    )
    .await;
    assert!(outcome.is_ok(), "the machine ends cleanly: {outcome:?}");
    let packets = parse_pkt_lines(&sink.body());
    assert_eq!(
        sideband_report_lines(&packets),
        vec![
            "unpack ok".to_owned(),
            "ok refs/heads/main".to_owned(),
            "ok refs/heads/dev".to_owned(),
        ]
    );
    assert_eq!(
        list_ref(&mem_refs, &record, "refs/heads/main").await,
        Some(chain.commit_b)
    );
    assert_eq!(
        list_ref(&mem_refs, &record, "refs/heads/dev").await,
        Some(chain.commit_a)
    );
    assert!(
        mem_ingest.objects().contains(&record, &chain.commit_b),
        "the pushed objects landed"
    );
}

/// EOF at the stateless tail's start (zero pack bytes) is the packless
/// path: CAS-only (no prepare call), the same report, the delete-only
/// push served.
///
/// Contract stamp: ADR-020 §1 (the EOF twin's zero-bytes mapping — the
/// same packless, CAS-only path; EOF is instead of a timeout); ADR-013
/// §7 (the same CAS home); core/w3-push-machine (the EOF twin's rule).
#[tokio::test]
async fn eof_at_the_tails_start_takes_the_stateless_packless_path() {
    let record = repo("owner/doubled");
    let current = oid(0x1b);
    let mem_refs = MemRefs::default();
    seed_ref(&mem_refs, &record, "refs/heads/gone", current).await;
    let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
    let runs = Arc::new(AtomicUsize::new(0));
    let ingest = Arc::new(PrepareSignal::new(Arc::clone(&runs))) as Arc<dyn GitPackIngest>;

    let (outcome, sink) = stateless_machine(
        body_chunks(
            "report-status side-band-64k",
            &[delete_command("refs/heads/gone", current)],
            Vec::new(),
        ),
        Arc::clone(&refs),
        Arc::clone(&ingest),
        &record,
        fast_limits(),
        false,
    )
    .await;
    assert!(outcome.is_ok());
    let packets = parse_pkt_lines(&sink.body());
    assert_eq!(
        sideband_report_lines(&packets),
        vec!["unpack ok".to_owned(), "ok refs/heads/gone".to_owned()],
    );
    assert_eq!(
        runs.load(Ordering::SeqCst),
        0,
        "no prepare call runs on the packless path"
    );
    assert_eq!(
        list_ref(&mem_refs, &record, "refs/heads/gone").await,
        None,
        "the CAS deleted the ref"
    );
}

/// EOF after a partial pack header is the stateless malformed-push
/// error — the timeout rule's EOF-driven twin — never a silent
/// packless downgrade; no ref applies, no report writes.
///
/// Contract stamp: ADR-020 §1 rule (iii) (the same rule, EOF-driven
/// instead of timeout-driven); ADR-009 (fail-closed).
#[tokio::test]
async fn eof_after_a_partial_header_is_the_stateless_malformed_push_error() {
    let record = repo("owner/doubled");
    let mem_refs = MemRefs::default();
    let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
    let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;

    let (outcome, sink) = stateless_machine(
        body_chunks(
            "report-status side-band-64k",
            &[create_command("refs/heads/main", oid(0x2c))],
            vec![b"PA".to_vec()],
        ),
        Arc::clone(&refs),
        Arc::clone(&ingest),
        &record,
        fast_limits(),
        false,
    )
    .await;
    let message = protocol_message(session_error(outcome));
    assert!(message.contains("pack-header"), "{message}");
    assert!(message.contains("malformed push"), "{message}");
    assert!(sink.body().is_empty(), "no report bytes were written");
    assert_eq!(mem_refs.list_refs(&record).await.unwrap().len(), 0);
}

/// The stateless nothing-to-do shape replies flush with no report.
///
/// Contract stamp: ADR-013 §3 (both substrates).
#[tokio::test]
async fn the_stateless_nothing_to_do_shape_replies_flush_with_no_report() {
    let record = repo("owner/doubled");
    let mem_refs = MemRefs::default();
    let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
    let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;

    let (outcome, sink) = stateless_machine(
        body_chunks("report-status side-band-64k", &[], Vec::new()),
        Arc::clone(&refs),
        Arc::clone(&ingest),
        &record,
        fast_limits(),
        false,
    )
    .await;
    assert!(outcome.is_ok());
    let packets = parse_pkt_lines(&sink.body());
    assert_eq!(packets.len(), 1, "exactly the reply flush");
    assert!(packets[0].is_empty(), "the reply flush has no payload");
    assert_eq!(mem_refs.list_refs(&record).await.unwrap().len(), 0);
}

/// A client that selected neither report cap still gets its CAS
/// applied — the machine reports nothing (no report, no flush; real
/// receive-pack's `report()` shape) — with the CAS's outcomes intact.
///
/// Contract stamp: ADR-013 §8 (the report exists for a report-status
/// selection; without it nothing is written — the mirror of
/// receive-pack.c's report guard); push-captures.md §Status report
/// (every captured client selected the cap; the no-selection shape is
/// the guard's own).
#[tokio::test]
async fn a_client_not_selecting_report_caps_gets_no_report_bytes() {
    let record = repo("owner/doubled");
    let chain = chain_pack();
    let mem_refs = MemRefs::default();
    let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
    let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;

    let (outcome, sink) = stateless_machine(
        body_chunks(
            "",
            &[create_command("refs/heads/main", chain.commit_b)],
            vec![chain.pack.clone()],
        ),
        Arc::clone(&refs),
        Arc::clone(&ingest),
        &record,
        fast_limits(),
        false,
    )
    .await;
    assert!(outcome.is_ok());
    assert!(sink.body().is_empty(), "no report bytes without the cap");
    assert_eq!(
        list_ref(&mem_refs, &record, "refs/heads/main").await,
        Some(chain.commit_b),
        "the CAS applied regardless of the report selection"
    );
}

/// The duplex machine's report framing follows the client's
/// `side-band-64k` selection; without the selection the report's
/// pkt-lines are bare to the single flush (the file:// capture's
/// shape).
///
/// Contract stamp: ADR-013 §8 (the framing selects on the client's
/// side-band-64k — an unwrapped report under a sideband selection
/// aborts real clients, observed live); ADR-005 (the substrate owns
/// the framing); push-captures.md §Status report.
#[tokio::test]
async fn the_reports_framing_follows_the_sideband_selection() {
    let record = repo("owner/doubled");
    let chain = chain_pack();
    let mem_refs = MemRefs::default();
    let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
    let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;
    let commands = [create_command("refs/heads/main", chain.commit_b)];

    let (mut client, stream) = duplex(1 << 20);
    client
        .write_all(&request_bytes(
            "report-status side-band-64k",
            &commands,
            &[],
        ))
        .await
        .unwrap();
    client.write_all(&chain.pack).await.unwrap();

    let limits = fast_limits();
    let machine = spawn_duplex_machine(
        stream,
        Arc::clone(&refs),
        Arc::clone(&ingest),
        record.clone(),
        limits,
        false,
        Arc::new(Semaphore::new(4)),
    );

    let packets = read_pkts(&mut client).await;
    let lines = sideband_report_lines(&packets);
    assert_eq!(
        lines,
        vec!["unpack ok".to_owned(), "ok refs/heads/main".to_owned()],
        "the sideband-selected client rides band-1 with the self-framed report"
    );
    drop(client);
    assert!(tokio::time::timeout(Duration::from_secs(5), machine)
        .await
        .unwrap()
        .unwrap()
        .is_ok());
}

/// The stateless machine upholds the same up-front rejections (the
/// grammar's session errors are substrate-blind).
///
/// Contract stamp: ADR-013 §4 (the rejections are substrate-blind);
/// core/w3-push-machine (the EOF twin's parity).
#[tokio::test]
async fn the_stateless_machine_rejects_up_front_the_same_way() {
    let record = repo("owner/doubled");
    let mem_refs = MemRefs::default();
    let refs = Arc::new(mem_refs.clone()) as Arc<dyn GitRefs>;
    let ingest = Arc::new(MemPackIngest::default()) as Arc<dyn GitPackIngest>;

    let (outcome, sink) = stateless_machine(
        body_chunks(
            "report-status",
            &[
                format!("shallow {}", oid(0x1b)),
                create_command("refs/heads/main", oid(0x2c)),
            ],
            Vec::new(),
        ),
        Arc::clone(&refs),
        Arc::clone(&ingest),
        &record,
        fast_limits(),
        false,
    )
    .await;
    let message = protocol_message(session_error(outcome));
    assert!(message.contains("shallow"), "{message}");
    assert!(sink.body().is_empty(), "no report on the rejected push");

    let (outcome, sink) = stateless_machine(
        body_chunks(
            "object-format=sha1024",
            &[create_command("refs/heads/main", oid(0x2c))],
            Vec::new(),
        ),
        Arc::clone(&refs),
        Arc::clone(&ingest),
        &record,
        fast_limits(),
        false,
    )
    .await;
    let message = protocol_message(session_error(outcome));
    assert!(message.contains("object-format"), "{message}");
    assert!(sink.body().is_empty(), "no report on the rejected push");
    assert_eq!(mem_refs.list_refs(&record).await.unwrap().len(), 0);
}

/// The seam the machine composes: `prepare`'s `PreparedPush.updates`
/// round-trips verbatim and becomes the CAS call's input — the
/// machine's leg boundary, pinned on the doubles.
///
/// Contract stamp: ADR-018 §5 (`PreparedPush::updates` = the prepare/
/// CAS inputs, verbatim); ADR-013 §6–7 (prepare validates and returns,
/// refs commits).
#[tokio::test]
async fn prepared_updates_are_the_cas_inputs() {
    let record = repo("owner/doubled");
    let chain = chain_pack();
    let ingest = MemPackIngest::default();
    let updates = vec![RefUpdate {
        name: "refs/heads/main".to_owned(),
        expected: None,
        new: Some(chain.commit_b),
    }];
    let prepared = ingest
        .prepare(
            &record,
            updates.clone(),
            Box::new(io::Cursor::new(chain.pack)),
            None,
            Limits::default(),
        )
        .await
        .unwrap();
    assert!(prepared.unpack.is_ok());
    assert_eq!(prepared.updates, updates, "verbatim into the CAS call");
}