alkgit 0.1.0

//! The wave-2 stateless serving entry (core/w2-stateless-serving): the
//! per-POST dispatch the http door mounts, the info/refs advertisement
//! emission, the per-request budgets, and the no-cross-request-state
//! property — the same V2 machines the duplex session feeds, driven
//! over in-memory reader/writer ends with the wave-2 backend doubles.
//!
//! Rows are names + `Contract stamp:` versions, per
//! `tests/common` §The row/stamp discipline.

mod common;

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

use async_trait::async_trait;
use futures::io::AsyncWrite as FuturesAsyncWrite;
use futures::stream;
use gix_hash::ObjectId;
use tokio::sync::{oneshot, Semaphore};

use alkgit::{
    serve_stateless_advertisement, serve_stateless_post, AccessAction, AuthorizedRepo, BodyReader,
    GitPackGen, GitPackIngest, GitRefs, Limits, PreparedPush, PushOptions, RefLine, RefOutcome,
    RefUpdate, RepoRecord, Service, StorageError, SubstrateError, Visibility,
};
use common::{
    blob_fixture, chain_pack, commit_fixture, create_command, decode_pack, delete_command,
    parse_pkt_lines, push_commands, sideband_report_lines, tree_fixture, wired_text as text,
    MemObjects, MemPackGen, MemPackIngest, MemRefs,
};

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: std::collections::BTreeMap::new(),
    }
}

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

fn hex_oid(b: u8) -> String {
    let mut hex = String::with_capacity(40);
    for _ in 0..20 {
        hex.push_str(&format!("{b:02x}"));
    }
    hex
}

/// The marker the door hands the entries — constructible only through
/// the checked path; here for a private repo with a read-granted identity.
fn marker(record: RepoRecord, identity: Option<&alkcall::core::Identity>) -> AuthorizedRepo {
    AuthorizedRepo::authorize(record, identity, AccessAction::Read)
        .expect("the door's check passed")
}

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

/// The first POST body: the smart prefix section, its flush, the V2
/// capability dump section, its flush, then the command invocation.
fn first_post_body(command: &[u8], pre: &[&str], args: &[&str]) -> Vec<u8> {
    [
        pkt(b"# service=git-upload-pack\n"),
        FLUSH.to_vec(),
        pkt(b"version 2\n"),
        pkt(b"agent=git/2.43\n"),
        pkt(b"object-format=sha1\n"),
        FLUSH.to_vec(),
        pkt(command),
        pre.iter()
            .flat_map(|l| pkt(l.as_bytes()))
            .collect::<Vec<u8>>(),
        DELIMITER.to_vec(),
        args.iter()
            .flat_map(|l| pkt(l.as_bytes()))
            .collect::<Vec<u8>>(),
        FLUSH.to_vec(),
    ]
    .concat()
}

/// A subsequent POST body (negotiation rounds): only the command.
fn command_post_body(command: &[u8], pre: &[&str], args: &[&str]) -> Vec<u8> {
    [
        pkt(command),
        pre.iter()
            .flat_map(|l| pkt(l.as_bytes()))
            .collect::<Vec<u8>>(),
        DELIMITER.to_vec(),
        args.iter()
            .flat_map(|l| pkt(l.as_bytes()))
            .collect::<Vec<u8>>(),
        FLUSH.to_vec(),
    ]
    .concat()
}

/// An async sink whose framed bytes the 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<std::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<std::io::Result<()>> {
        Poll::Ready(Ok(()))
    }

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

/// A request-body end that never yields and never ends — the wall-clock
/// budget's victim (a stalled client or a stalled door body).
struct PendingReader;

impl futures::AsyncRead for PendingReader {
    fn poll_read(
        self: Pin<&mut Self>,
        _cx: &mut Context<'_>,
        _buf: &mut [u8],
    ) -> Poll<std::io::Result<usize>> {
        Poll::Pending
    }
}

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

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

/// The door's per-POST assembly: the marker, the backend handles keyed
/// on the resolved repo, the shared semaphore — the shape
/// serve_stateless_post consumes.
struct Fixture {
    marker: AuthorizedRepo,
    refs: Arc<MemRefs>,
    gen: Arc<dyn GitPackGen>,
    ingest: Arc<MemPackIngest>,
    objects: MemObjects,
    repo_id: &'static str,
}

fn fixture() -> Fixture {
    let repo_id = "owner/doubled";
    let mut record = repo(repo_id);
    record
        .grants
        .insert("u-7731".to_string(), vec![AccessAction::Read]);
    let holder = alkcall::core::Identity {
        id: "u-7731".to_string(),
        scopes: Vec::new(),
        resources: std::collections::HashMap::new(),
    };
    let objects = MemObjects::default();
    Fixture {
        marker: marker(record, Some(&holder)),
        refs: Arc::new(MemRefs::default()),
        gen: Arc::new(MemPackGen {
            objects: objects.clone(),
        }),
        ingest: Arc::new(MemPackIngest {
            objects: objects.clone(),
        }),
        objects,
        repo_id,
    }
}

impl Fixture {
    fn record(&self) -> RepoRecord {
        repo(self.repo_id)
    }

    fn with_gen(&self, gen: Arc<dyn GitPackGen>) -> Fixture {
        Fixture {
            marker: self.marker.clone(),
            refs: Arc::clone(&self.refs),
            gen,
            ingest: Arc::clone(&self.ingest),
            objects: self.objects.clone(),
            repo_id: self.repo_id,
        }
    }

    async fn serve(
        &self,
        body: PostBody,
        limits: Limits,
        service: Service,
        pipeline: Arc<Semaphore>,
    ) -> (Result<(), SubstrateError>, VecSink) {
        let sink = VecSink::default();
        let refs: Arc<dyn GitRefs> = self.refs.clone();
        let result = serve_stateless_post(
            &self.marker,
            body,
            sink.clone(),
            limits,
            service,
            refs,
            Arc::clone(&self.gen),
            self.ingest.clone() as Arc<dyn GitPackIngest>,
            false,
            pipeline,
        )
        .await;
        (result, sink)
    }

    /// The push POST's serving call: `Service::ReceivePack`, the
    /// `push-options` negotiated-gate flag alongside (the same
    /// assembler config the GET carries).
    async fn serve_push(
        &self,
        body: PostBody,
        limits: Limits,
        push_options_negotiated: bool,
        refs: Arc<dyn GitRefs>,
        ingest: Arc<dyn GitPackIngest>,
        pipeline: Arc<Semaphore>,
    ) -> (Result<(), SubstrateError>, VecSink) {
        let sink = VecSink::default();
        let result = serve_stateless_post(
            &self.marker,
            body,
            sink.clone(),
            limits,
            Service::ReceivePack,
            refs,
            Arc::clone(&self.gen),
            ingest,
            push_options_negotiated,
            pipeline,
        )
        .await;
        (result, sink)
    }
}

fn default_pipeline() -> Arc<Semaphore> {
    Arc::new(Semaphore::new(
        Limits::default().max_concurrent_blocking_tasks,
    ))
}

async fn assert_protocol_error(result: Result<(), SubstrateError>, needle: &str) {
    let error = result.expect_err("expected a typed protocol failure");
    assert!(
        error.is_terminal(),
        "the typed failure is terminal: {error:?}"
    );
    match &error {
        SubstrateError::Protocol(message) => {
            assert!(message.contains(needle), "{message}");
        }
        other => panic!("expected a protocol-shaped failure, got {other:?}"),
    }
}

/// The first POST carries the cached capability dump — the entry skips
/// both flush-delimited sections and serves the ls-refs command: ref
/// lines from the backend listing, attributes on the same line, the
/// unborn HEAD row emitted as the literal-unborn line, the response
/// ending at flush.
///
/// Contract stamp: ADR-005 §2 (one POST body is one command invocation,
/// the capability-dump skip is the machinery's); doors.md §alkhttp git
/// feature (POST /{repo}/git-upload-pack, one V2 command per POST);
/// ADR-003 (the honest advertisement the client re-sends); review 001
/// N-5 (the unborn-HEAD line shape, wire form captured literal-unborn —
/// the rider-1 real-client sessions, wave 5, corrected the zero-oid
/// form); core/w2-stateless-serving (the entry dispatches the same
/// machines).
#[tokio::test]
async fn the_first_posts_capability_dump_is_skipped_and_ls_refs_serves() {
    let fx = fixture();
    let doubled = fx.record();
    fx.refs
        .apply_updates(
            &doubled,
            vec![
                RefUpdate {
                    name: "refs/heads/main".to_string(),
                    expected: None,
                    new: Some(oid(1)),
                },
                RefUpdate {
                    name: "refs/tags/v1".to_string(),
                    expected: None,
                    new: Some(oid(2)),
                },
            ],
            false,
        )
        .await
        .unwrap();
    fx.refs
        .set_head_symref(&doubled, Some("refs/heads/main".to_string()));

    let body = body_parts(vec![first_post_body(
        b"command=ls-refs\n",
        &["agent=git/2.43"],
        &["symrefs", "unborn"],
    )]);
    let (result, sink) = fx
        .serve(
            body,
            Limits::default(),
            Service::UploadPack,
            default_pipeline(),
        )
        .await;
    result.unwrap();

    let lines = parse_pkt_lines(&sink.body());
    assert_eq!(text(&lines[0]), "unborn HEAD symref-target:refs/heads/main");
    assert_eq!(text(&lines[1]), format!("{} refs/heads/main", oid(1)));
    assert_eq!(text(&lines[2]), format!("{} refs/tags/v1", oid(2)));
    assert!(lines[3].is_empty(), "the response ends at flush");
    assert_eq!(lines.len(), 4);
}

/// No cross-request state (ADR-014 §7): two consecutive POSTs ride two
/// separate entry invocations and re-derive identically — byte-equal
/// response bodies, no session to restore anywhere.
///
/// Contract stamp: ADR-002 (the stateless session entry, per-request
/// state); ADR-014 §7 (statelessness is a substrate property — the
/// client re-sends wants + commons every round);
/// core/w2-stateless-serving (the entry holds nothing between
/// invocations).
#[tokio::test]
async fn two_consecutive_posts_re_derive_identically() {
    let fx = fixture();
    let body = || {
        body_parts(vec![first_post_body(
            b"command=ls-refs\n",
            &[],
            &["peel", "ref-prefix refs/heads/"],
        )])
    };

    let (first, first_sink) = fx
        .serve(
            body(),
            Limits::default(),
            Service::UploadPack,
            default_pipeline(),
        )
        .await;
    first.unwrap();

    let (second, second_sink) = fx
        .serve(
            body(),
            Limits::default(),
            Service::UploadPack,
            default_pipeline(),
        )
        .await;
    second.unwrap();

    assert_eq!(
        first_sink.body(),
        second_sink.body(),
        "two POSTs, one deterministic response"
    );
}

/// A full first-POST fetch round (want + done, capability dump ahead of
/// the command) serves the done round through the entry: the packfile
/// section streams as band-1 over the response rules and decodes as
/// closure(wants) − closure(recognized haves).
///
/// Contract stamp: ADR-014 §2 (the done round, the packfile section —
/// `packfile` header, band-1 sideband, section flush, no
/// acknowledgments); ADR-005 §2 (the first POST's dump skip composes
/// with the machines); ADR-014 §5 (the ack seam — the recognized have
/// is subtracted); core/w2-stateless-serving (the entry composes the
/// negotiation machine per POST).
#[tokio::test]
async fn a_full_first_post_fetch_round_serves_the_done_round_pack() {
    let fx = fixture();
    let doubled = fx.record();
    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),
    ] {
        fx.objects.put(&doubled, id, bytes);
    }
    let fx = fx.with_gen(Arc::new(MemPackGen {
        objects: fx.objects.clone(),
    }));

    let body = body_parts(vec![first_post_body(
        b"command=fetch\n",
        &["object-format=sha1"],
        &[
            &format!("want {commit_b}"),
            &format!("have {commit_a}"),
            "thin-pack",
            "no-progress",
            "done",
        ],
    )]);
    let (result, sink) = fx
        .serve(
            body,
            Limits::default(),
            Service::UploadPack,
            default_pipeline(),
        )
        .await;
    result.unwrap();

    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([commit_b, tree_b, blob_b]),
        "closure(commit_b) − closure(commit_a) over the stateless entry"
    );
    let _ = decoded.trailer;
    assert!(
        lines.last().unwrap().is_empty(),
        "the response ends at flush"
    );
}

/// The OQ-19 rider-2 positive-ack capture over the stateless entry (the
/// machinery-side run): a no-`done` POST with multiple real haves acks
/// **each** recognized have — per-have `ACK <oid>` lines in the
/// request's order, only the seam-recognized subset (the missing and the
/// exists-but-not-a-commit haves stay unacked), `NAK` absent — and the
/// done round POST's boundary set is the recognized subset: the
/// unrecognized haves subtract nothing, both commons bound the walk.
///
/// Contract stamp: OQ-19 rider 2 (the positive-ACK capture,
/// tasks/architecture/oq-19-wire-validations.md §2 — the
/// machinery-side run; the substrate-backed re-run is track B wave 7);
/// ADR-014 §1 (ACK per recognized have, NAK only when no ACKs);
/// ADR-014 §2 as amended by review 001 A-4 (the boundary set is the
/// recognized subset); ADR-014 §5 (exists + is-commit — the blob have
/// drives the negative leg); ADR-014 §7 (each POST is one round, no
/// cross-request state).
#[tokio::test]
async fn a_multi_have_post_acks_each_recognized_have_and_the_done_boundary_is_the_recognized_subset(
) {
    let fx = fixture();
    let doubled = fx.record();
    let (blob_a, blob_a_bytes) = blob_fixture(b"alpha");
    let (blob_b, blob_b_bytes) = blob_fixture(b"beta");
    let (blob_c, blob_c_bytes) = blob_fixture(b"gamma");
    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 (tree_c, tree_c_bytes) =
        tree_fixture(&[(gix_object::tree::EntryKind::Blob, "c.txt", blob_c)]);
    let (commit_a, commit_a_bytes) = commit_fixture(tree_a, &[], "alpha");
    let (commit_b, commit_b_bytes) = commit_fixture(tree_b, &[commit_a], "beta");
    let (commit_c, commit_c_bytes) = commit_fixture(tree_c, &[commit_b], "gamma");
    for (id, bytes) in [
        (blob_a, blob_a_bytes),
        (blob_b, blob_b_bytes),
        (blob_c, blob_c_bytes),
        (tree_a, tree_a_bytes),
        (tree_b, tree_b_bytes),
        (tree_c, tree_c_bytes),
        (commit_a, commit_a_bytes),
        (commit_b, commit_b_bytes),
        (commit_c, commit_c_bytes),
    ] {
        fx.objects.put(&doubled, id, bytes);
    }
    let fx = fx.with_gen(Arc::new(MemPackGen {
        objects: fx.objects.clone(),
    }));

    let missing = oid(9u8);
    let body = body_parts(vec![command_post_body(
        b"command=fetch\n",
        &["object-format=sha1", "agent=git/2.43"],
        &[
            &format!("want {}", commit_c),
            &format!("have {}", commit_a),
            &format!("have {}", missing),
            &format!("have {}", commit_b),
        ],
    )]);
    let (result, sink) = fx
        .serve(
            body,
            Limits::default(),
            Service::UploadPack,
            default_pipeline(),
        )
        .await;
    result.unwrap();
    let ack_round: Vec<String> = parse_pkt_lines(&sink.body())
        .iter()
        .map(|p| text(p))
        .collect();
    assert_eq!(
        ack_round,
        vec![
            "acknowledgments".to_owned(),
            format!("ACK {}", commit_a),
            format!("ACK {}", commit_b),
            String::new(),
        ],
        "per-have ACKs in request order, unrecognized haves silent, flush-terminated"
    );

    let body = body_parts(vec![command_post_body(
        b"command=fetch\n",
        &["object-format=sha1", "agent=git/2.43"],
        &[
            &format!("want {}", commit_c),
            &format!("have {}", commit_a),
            &format!("have {}", missing),
            &format!("have {}", blob_a),
            &format!("have {}", commit_b),
            "done",
        ],
    )]);
    let (result, sink) = fx
        .serve(
            body,
            Limits::default(),
            Service::UploadPack,
            default_pipeline(),
        )
        .await;
    result.unwrap();
    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([commit_c, tree_c, blob_c]),
        "the boundary set is the recognized subset: both commons bounded the \
         walk, the missing and the non-commit have subtracted nothing"
    );
    let _ = decoded.trailer;
}

/// A no-`done` fetch POST re-derives the acknowledgments section per
/// POST — the client re-sent its wants + haves, the entry held nothing.
///
/// Contract stamp: ADR-014 §1 (the exact acknowledgments grammar,
/// `ready` never sent); ADR-014 §7 (the stateless rounds re-derive);
/// core/w2-stateless-serving.
#[tokio::test]
async fn a_no_done_fetch_post_re_derives_the_acknowledgments_per_post() {
    let fx = fixture();
    let doubled = fx.record();
    let (blob, blob_bytes) = blob_fixture(b"paid");
    let (tree, tree_bytes) = tree_fixture(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob)]);
    let (commit_a, commit_a_bytes) = commit_fixture(tree, &[], "alpha");
    let (commit_b, commit_b_bytes) = commit_fixture(tree, &[commit_a], "beta");
    for (id, bytes) in [
        (blob, blob_bytes),
        (tree, tree_bytes),
        (commit_a, commit_a_bytes),
        (commit_b, commit_b_bytes),
    ] {
        fx.objects.put(&doubled, id, bytes);
    }
    let fx = fx.with_gen(Arc::new(MemPackGen {
        objects: fx.objects.clone(),
    }));

    let body = body_parts(vec![command_post_body(
        b"command=fetch\n",
        &[],
        &[
            &format!("want {}", hex_oid(0x0d)),
            &format!("have {commit_a}"),
        ],
    )]);
    let (result, sink) = fx
        .serve(
            body,
            Limits::default(),
            Service::UploadPack,
            default_pipeline(),
        )
        .await;
    result.unwrap();

    let lines = parse_pkt_lines(&sink.body());
    assert_eq!(text(&lines[0]), "acknowledgments");
    assert_eq!(text(&lines[1]), format!("ACK {commit_a}"));
    assert!(lines.last().unwrap().is_empty());
}

/// The flush-only probe resolves to the empty-OK shape: `Ok` with
/// nothing written — the 200 mapping is the door's (the machinery
/// resolves; the entry surfaces nothing).
///
/// Contract stamp: ADR-005 §2 (the probe rule — answering an error
/// status breaks the whole clone, poc3-findings §probe_rpc);
/// core/w2-stateless-serving (`StatelessPost::Probe` → the empty-OK
/// shape, the response rules honored with nothing written).
#[tokio::test]
async fn the_probe_resolves_to_the_empty_ok_shape_with_nothing_written() {
    let fx = fixture();
    let (result, sink) = fx
        .serve(
            body_parts(vec![FLUSH.to_vec()]),
            Limits::default(),
            Service::UploadPack,
            default_pipeline(),
        )
        .await;
    result.unwrap();
    assert!(sink.body().is_empty(), "nothing written on the probe");

    let (result, sink) = fx
        .serve(
            body_parts(vec![Vec::new()]),
            Limits::default(),
            Service::UploadPack,
            default_pipeline(),
        )
        .await;
    result.unwrap();
    assert!(sink.body().is_empty(), "the body-less probe as well");
}

/// An unknown command declines as a typed protocol failure — the
/// stateless side has no error band; the entry surfaces the failure the
/// door maps to an http status, writing nothing.
///
/// Contract stamp: ADR-005 §4 (the error taxonomy: protocol-shaped
/// failures map to http statuses at the door); ADR-003 (honest surface,
/// unimplemented features declined); core/w2-stateless-serving
/// (unknown command → the door-mapped protocol failure).
#[tokio::test]
async fn an_unknown_command_declines_as_the_typed_failure() {
    let fx = fixture();
    let (result, sink) = fx
        .serve(
            body_parts(vec![command_post_body(b"command=object-info\n", &[], &[])]),
            Limits::default(),
            Service::UploadPack,
            default_pipeline(),
        )
        .await;
    assert_protocol_error(result, "unknown command").await;
    assert!(sink.body().is_empty(), "the decline writes nothing");
}

/// A `git-receive-pack` probe POST resolves to the empty-OK shape
/// before any service dispatch — the probe shape is service-agnostic
/// (breaking it breaks the large-request path of *any* client), and on
/// the push path the large-request probe is ADR-013 §12's fact. The
/// same flush-only body is the V0 nothing-to-do shape over http
/// (upstream http-backend answers it 200-empty all the same) — the
/// reply-flush report shape of that body is the machine-level
/// discipline, pinned by core/w3-push-machine rows; nothing reaches
/// any backend handle here.
///
/// Contract stamp: ADR-005 §2 (the service-agnostic probe rule — the
/// empty-OK answer precedes any service dispatch); ADR-013 §12 (the
/// probe POST on the push path); ADR-013 §3 (the immediate-flush
/// nothing-to-do shape; over http it shares the probe's bytes);
/// ADR-020 §1; core/w3-stateless-serving.
#[tokio::test]
async fn the_flush_only_receive_pack_post_is_the_probe_shape_before_any_dispatch() {
    let fx = fixture();
    let runs = Arc::new(AtomicUsize::new(0));
    let ingest = counting_ingest(fx.ingest.clone(), Arc::clone(&runs));
    let refs: Arc<dyn GitRefs> = fx.refs.clone();

    let (result, sink) = fx
        .serve_push(
            body_parts(vec![FLUSH.to_vec()]),
            Limits::default(),
            false,
            Arc::clone(&refs),
            ingest,
            default_pipeline(),
        )
        .await;
    result.unwrap();
    assert!(
        sink.body().is_empty(),
        "the probe answer writes nothing (the empty-OK shape)"
    );
    assert_eq!(
        runs.load(AtomicOrdering::SeqCst),
        0,
        "nothing was dispatched to any backend handle"
    );
}

/// The wall-clock budget applies per request (the stateless substrate
/// has no session to amortize it): a stalled request body breaches the
/// budget and the entry surfaces the typed server/session failure the
/// door maps to 503.
///
/// Contract stamp: ADR-009 (the wall clock, per-request on http —
/// ADR-002's cost note; server/session budgets map to 503);
/// core/w2-stateless-serving (the wall clock applies per request).
#[tokio::test]
async fn the_wall_clock_budget_fails_closed_per_request() {
    let fx = fixture();
    let limits = Limits {
        session_wall_clock: Duration::from_millis(20),
        ..Limits::default()
    };
    let sink = VecSink::default();
    let refs: Arc<dyn GitRefs> = fx.refs.clone();
    let result = serve_stateless_post(
        &fx.marker,
        PendingReader,
        sink.clone(),
        limits,
        Service::UploadPack,
        refs,
        Arc::clone(&fx.gen),
        fx.ingest.clone() as Arc<dyn GitPackIngest>,
        false,
        default_pipeline(),
    )
    .await;
    match result {
        Err(SubstrateError::Protocol(message)) => {
            assert!(error_contains(&message, "session_wall_clock"), "{message}");
        }
        other => panic!("a wall-clock breach is a typed failure: {other:?}"),
    }
    assert!(sink.body().is_empty(), "the breach wrote nothing");
}

fn error_contains(message: &str, needle: &str) -> bool {
    message.contains(needle)
}

/// A saturated pipeline-concurrency semaphore delays the entry's
/// generate call — the door's shared semaphore is handed through and
/// the permit wraps the done round; releasing a permit lets the POST
/// complete.
///
/// Contract stamp: ADR-009 (the blocking-pool budget, the permit
/// acquired around each `GitPackGen::generate` call — backend.md
/// §concurrency model); core/w2-stateless-serving (the shared
/// concurrency semaphore is the entry's input).
#[tokio::test]
async fn a_saturated_concurrency_permit_delays_the_entry_generate() {
    let fx = fixture();
    let doubled = fx.record();
    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),
    ] {
        fx.objects.put(&doubled, id, bytes);
    }
    let gen: Arc<dyn GitPackGen> = Arc::new(MemPackGen {
        objects: fx.objects.clone(),
    });

    let pipeline = Arc::new(Semaphore::new(1));
    let held = pipeline.clone().acquire_owned().await.unwrap();
    let body = body_parts(vec![command_post_body(
        b"command=fetch\n",
        &[],
        &[&format!("want {commit}"), "done"],
    )]);
    let sink = VecSink::default();
    let (done_tx, mut done_rx) = oneshot::channel();
    let refs: Arc<dyn GitRefs> = fx.refs.clone();
    let marker = fx.marker.clone();
    let ingest = fx.ingest.clone() as Arc<dyn GitPackIngest>;
    tokio::spawn(async move {
        let result = serve_stateless_post(
            &marker,
            body,
            sink.clone(),
            Limits::default(),
            Service::UploadPack,
            refs,
            gen,
            ingest,
            false,
            pipeline,
        )
        .await;
        done_tx.send((result, sink)).ok();
    });

    tokio::time::sleep(Duration::from_millis(50)).await;
    assert!(
        done_rx.try_recv().is_err(),
        "the saturated permit delays the done round's generate"
    );

    drop(held);
    let (result, sink) = done_rx
        .await
        .expect("the entry completes once a permit releases");
    result.unwrap();
    let lines = parse_pkt_lines(&sink.body());
    assert_eq!(text(&lines[0]), "packfile");
}

/// The entry is IO-abstract (convention 15): a tokio read end adapted
/// through compat plugs in like any door stream — no http type anywhere
/// below the stream.
///
/// Contract stamp: ADR-002 (front-door-blind core — the entry consumes
/// reader/writer ends); ADR-005 §2 (the machinery is generic over both
/// ends); convention 15 (no http/channel types below the stream);
/// core/w2-stateless-serving.
#[tokio::test]
async fn the_entry_runs_over_arbitrary_reader_writer_ends() {
    use tokio::io::AsyncWriteExt as TokioAsyncWriteExt;
    use tokio_util::compat::TokioAsyncReadCompatExt;

    let fx = fixture();
    let (mut peer, stream) = tokio::io::duplex(1 << 15);
    TokioAsyncWriteExt::write_all(
        &mut peer,
        &command_post_body(b"command=ls-refs\n", &[], &["ref-prefix refs/heads/"]),
    )
    .await
    .unwrap();
    drop(peer);

    let sink = VecSink::default();
    let refs: Arc<dyn GitRefs> = fx.refs.clone();
    let result = serve_stateless_post(
        &fx.marker,
        stream.compat(),
        sink.clone(),
        Limits::default(),
        Service::UploadPack,
        refs,
        Arc::clone(&fx.gen),
        fx.ingest.clone() as Arc<dyn GitPackIngest>,
        false,
        default_pipeline(),
    )
    .await;
    result.unwrap();
    let lines = parse_pkt_lines(&sink.body());
    assert!(
        lines.last().unwrap().is_empty(),
        "the response ends at flush"
    );
}

/// The info/refs advertisement emission: the smart prefix line, its
/// flush, then the V2 capability advertisement (the honest list) ending
/// at flush — the per-request-set shape the door's GET route serves,
/// the response body remote-curl caches and re-sends in its first POST.
///
/// Contract stamp: doors.md §alkhttp git feature (GET info/refs → smart
/// prefix + capability dump + flush); transport.md §Advertisement
/// (per-request-set on http); ADR-003 (the honest capability list);
/// ADR-016 §3 (the service rides the route); poc3-findings §1 (the
/// two-flush info/refs body is the shape the client caches);
/// core/w2-stateless-serving (the emission exists for the door's
/// route).
#[tokio::test]
async fn the_info_refs_advertisement_emits_the_smart_prefix_and_the_capability_dump() {
    let fx = fixture();
    let refs: Arc<dyn GitRefs> = fx.refs.clone();
    let sink = VecSink::default();
    serve_stateless_advertisement(
        &fx.marker,
        sink.clone(),
        Service::UploadPack,
        refs,
        &Limits::default(),
        false,
    )
    .await
    .unwrap();

    let lines = parse_pkt_lines(&sink.body());
    assert_eq!(text(&lines[0]), "# service=git-upload-pack");
    assert!(lines[1].is_empty(), "the prefix section ends at flush");
    assert_eq!(text(&lines[2]), "version 2");
    // The honest capability list, in emit order — the sha256 rider
    // rides the feature per OQ-05's passthrough policy.
    #[cfg(not(feature = "sha256"))]
    {
        assert_eq!(text(&lines[3]), "ls-refs=unborn");
        assert_eq!(text(&lines[4]), "fetch=wait-for-done");
        assert_eq!(text(&lines[5]), "object-format=sha1");
        assert!(lines[6].is_empty(), "the dump ends at flush");
        assert_eq!(lines.len(), 7);
    }
    #[cfg(feature = "sha256")]
    {
        assert_eq!(text(&lines[3]), "ls-refs=unborn");
        assert_eq!(text(&lines[4]), "fetch=wait-for-done");
        assert_eq!(text(&lines[5]), "object-format=sha1");
        assert_eq!(text(&lines[6]), "object-format=sha256");
        assert!(lines[7].is_empty(), "the dump ends at flush");
        assert_eq!(lines.len(), 8);
    }
    assert!(
        !sink.body().windows(4).any(|w| w == b"0002"),
        "no 0002 anywhere in the response body"
    );
}

/// The receive-pack advertisement GET: the smart prefix + flush, then
/// the V0 ref advertisement over the backend's listed refs — caps
/// NUL-attached on the first ref line, bare ref lines, flush-terminated
/// — the shape the push client caches and re-sends inside its first
/// POST. A listed ref rides the emission; the ACL-first marker precedes
/// any byte.
///
/// Contract stamp: ADR-013 §1–2 (the V0 framing, the served set);
/// push-captures.md §Advertisement (the normative shapes); ADR-007
/// (ACL first via the marker); ADR-005 §2 (the two-flush info/refs
/// body is the shape the client caches); core/w3-v0-advertisement.
#[tokio::test]
async fn the_receive_pack_advertisement_emits_the_prefix_and_the_v0_ref_list() {
    let fx = fixture();
    let record = fx.record();
    let outcomes = fx
        .refs
        .apply_updates(
            &record,
            vec![RefUpdate {
                name: "refs/heads/main".to_owned(),
                expected: None,
                new: Some(oid(1)),
            }],
            false,
        )
        .await
        .unwrap();
    assert_eq!(outcomes.len(), 1, "the fixture's ref was created");

    let refs: Arc<dyn GitRefs> = fx.refs.clone();
    let sink = VecSink::default();
    serve_stateless_advertisement(
        &fx.marker,
        sink.clone(),
        Service::ReceivePack,
        refs,
        &Limits::default(),
        false,
    )
    .await
    .unwrap();

    let lines = parse_pkt_lines(&sink.body());
    assert_eq!(text(&lines[0]), "# service=git-receive-pack");
    assert!(lines[1].is_empty(), "the prefix section ends at flush");
    // The sha256 passthrough rider rides the feature (OQ-05), exactly
    // as on the V2 advertisement.
    #[allow(unused_variables)]
    let rider = if cfg!(feature = "sha256") {
        " object-format=sha256"
    } else {
        ""
    };
    assert_eq!(
        text(&lines[2]),
        format!("{} refs/heads/main\0report-status report-status-v2 delete-refs side-band-64k atomic ofs-delta object-format=sha1{rider} agent=alkgit/{}", hex_oid(1), env!("CARGO_PKG_VERSION")),
        "caps NUL-attached on the FIRST ref line (the served set, agent=alkgit/<version>)"
    );
    assert!(lines[3].is_empty(), "the advertisement ends at flush");
    assert_eq!(lines.len(), 4);
    assert!(
        !sink.body().windows(4).any(|w| w == b"0002"),
        "no 0002 anywhere in the response body"
    );
}

/// An empty repo (no refs, no unborn-HEAD row) advertises exactly the
/// `capabilities^{}` zero-id sentinel line — never a bare capability
/// dump.
///
/// Contract stamp: ADR-013 §2 (the sentinel rule);
/// push-captures.md §Advertisement; core/w3-v0-advertisement.
#[tokio::test]
async fn the_receive_pack_advertisement_for_an_empty_repo_emits_the_sentinel() {
    let fx = fixture();
    let refs: Arc<dyn GitRefs> = fx.refs.clone();
    let sink = VecSink::default();
    serve_stateless_advertisement(
        &fx.marker,
        sink.clone(),
        Service::ReceivePack,
        refs,
        &Limits::default(),
        false,
    )
    .await
    .unwrap();

    let lines = parse_pkt_lines(&sink.body());
    assert_eq!(text(&lines[0]), "# service=git-receive-pack");
    assert!(lines[1].is_empty(), "the prefix section ends at flush");
    #[allow(unused_variables)]
    let rider = if cfg!(feature = "sha256") {
        " object-format=sha256"
    } else {
        ""
    };
    assert_eq!(
        text(&lines[2]),
        format!(
            "{} capabilities^{{}}\0report-status report-status-v2 delete-refs side-band-64k atomic ofs-delta object-format=sha1{rider} agent=alkgit/{}",
            "0".repeat(40),
            env!("CARGO_PKG_VERSION"),
        ),
        "the sentinel line carries the caps"
    );
    assert!(lines[3].is_empty(), "the advertisement ends at flush");
    assert_eq!(lines.len(), 4);
}

/// The GET emission's `push-options` gate (ADR-013 §11, the assembler's
/// config; default off in v1): the token appears in the advertised text
/// only under the gate flag — additive to open later as a value change.
///
/// Contract stamp: ADR-013 §2, §11 (the gate-gated token);
/// core/w3-v0-advertisement.
#[tokio::test]
async fn the_push_options_token_rides_the_gate_only() {
    let fx = fixture();
    let refs: Arc<dyn GitRefs> = fx.refs.clone();

    let gated = {
        let sink = VecSink::default();
        serve_stateless_advertisement(
            &fx.marker,
            sink.clone(),
            Service::ReceivePack,
            Arc::clone(&refs),
            &Limits::default(),
            true,
        )
        .await
        .unwrap();
        String::from_utf8_lossy(&sink.body()).into_owned()
    };
    assert!(
        gated.contains("\0report-status"),
        "caps ride the first ref line"
    );
    assert!(
        gated.contains(" push-options "),
        "the gate adds the push-options token: {gated}"
    );

    let ungated = {
        let sink = VecSink::default();
        serve_stateless_advertisement(
            &fx.marker,
            sink.clone(),
            Service::ReceivePack,
            refs,
            &Limits::default(),
            false,
        )
        .await
        .unwrap();
        String::from_utf8_lossy(&sink.body()).into_owned()
    };
    assert!(
        !ungated.contains("push-options"),
        "push-options is absent by default (v1's gate is off): {ungated}"
    );
}

/// The V0 emission projects the seam rows to oid+name: the unborn-HEAD
/// symref row (oid: None) is filtered — a V0 advertisement line has no
/// symref shape.
///
/// Contract stamp: ADR-013 §2 (the V0 ref-line shape);
/// push-captures.md §Advertisement; core/w3-v0-advertisement (the
/// filtered_head posture; the OQ-19 rider (a) covers the ls-refs side).
#[tokio::test]
async fn the_unborn_head_row_is_filtered_from_the_push_advertisement() {
    let fx = fixture();
    let record = fx.record();
    fx.refs
        .set_head_symref(&record, Some("refs/heads/main".to_owned()));
    let outcomes = fx
        .refs
        .apply_updates(
            &record,
            vec![RefUpdate {
                name: "refs/heads/main".to_owned(),
                expected: None,
                new: Some(oid(1)),
            }],
            false,
        )
        .await
        .unwrap();
    assert_eq!(outcomes.len(), 1, "the fixture's ref was created");

    let refs: Arc<dyn GitRefs> = fx.refs.clone();
    let sink = VecSink::default();
    serve_stateless_advertisement(
        &fx.marker,
        sink.clone(),
        Service::ReceivePack,
        refs,
        &Limits::default(),
        false,
    )
    .await
    .unwrap();

    let lines = parse_pkt_lines(&sink.body());
    let body = String::from_utf8_lossy(&sink.body()).into_owned();
    assert!(
        !body.contains("symref-target"),
        "a V0 advertisement line has no symref shape — the unborn row is filtered: {body}"
    );
    assert!(
        !lines
            .iter()
            .any(|line| line.len() >= 4 && String::from_utf8_lossy(&line[4..]).starts_with("HEAD")),
        "the no-oid HEAD row cannot be advertised: {body}"
    );
}

/// The entries' marker is the type-level ACL ordering: a denied check
/// yields the `None` the door collapses to its generic denial before
/// any entry call — no unchecked session value can exist to feed the
/// entries.
///
/// Contract stamp: ADR-007 step 3 (the marker is the only path in —
/// resolve → authorize precedes any protocol byte); ADR-011 §3 (the
/// policy evaluation producing it); ADR-008/N-2 (the collapse rule is
/// the door's, before the marker exists);
/// core/w2-stateless-serving (the entries never resolve or authorize).
#[test]
fn only_the_checked_marker_path_feeds_the_entries() {
    let mut private = repo("owner/doubled");
    private
        .grants
        .insert("u-7731".to_string(), vec![AccessAction::Read]);
    let holder = alkcall::core::Identity {
        id: "u-7731".to_string(),
        scopes: Vec::new(),
        resources: std::collections::HashMap::new(),
    };
    assert_eq!(
        AuthorizedRepo::authorize(private.clone(), None, AccessAction::Read),
        None,
        "anonymous on a private repo is no marker"
    );
    assert!(
        AuthorizedRepo::authorize(private, Some(&holder), AccessAction::Read).is_some(),
        "the granted identity is the entry's input"
    );
}

/// A backend listing failure surfaces as the typed session error —
/// never a partial ref list.
///
/// Contract stamp: ADR-005 §4 (backend failures band-map through the
/// entries); review 001 N-1 (fail closed — a partial ref list is the
/// worst advertisement failure mode); core/w2-stateless-serving.
#[tokio::test]
async fn a_backend_listing_failure_serves_nothing() {
    struct ListingFailure;

    #[async_trait::async_trait]
    impl GitRefs for ListingFailure {
        async fn list_refs(&self, _repo: &RepoRecord) -> Result<Vec<RefLine>, StorageError> {
            Err(StorageError::Invalid("listing unavailable".to_string()))
        }

        async fn apply_updates(
            &self,
            _repo: &RepoRecord,
            _updates: Vec<RefUpdate>,
            _atomic: bool,
        ) -> Result<Vec<RefOutcome>, StorageError> {
            Err(StorageError::Invalid("listing unavailable".to_string()))
        }
    }

    let fx = fixture();
    let refs: Arc<dyn GitRefs> = Arc::new(ListingFailure);
    let sink = VecSink::default();
    let result = serve_stateless_post(
        &fx.marker,
        body_parts(vec![command_post_body(b"command=ls-refs\n", &[], &[])]),
        sink.clone(),
        Limits::default(),
        Service::UploadPack,
        refs,
        Arc::clone(&fx.gen),
        fx.ingest.clone() as Arc<dyn GitPackIngest>,
        false,
        default_pipeline(),
    )
    .await;
    let error = result.expect_err("a listing failure must not serve");
    assert!(
        error.is_terminal(),
        "backend failures band-map as terminal: {error:?}"
    );
    assert!(sink.body().is_empty(), "a failed listing wrote nothing");
}

/// The GET emission's one backend call is under the per-request wall
/// clock: a hung listing times out into the typed breach error — the
/// advertisement route is never the session surface without a budget
/// (the POST entry's wall-clock discipline, mirrored).
///
/// Contract stamp: ADR-009 (the wall clock is a budget of every
/// session/request surface; fail-closed breach behavior); ADR-005 §4
/// (the typed failure maps at the door).
#[tokio::test]
async fn the_advertisement_listing_hangs_until_the_wall_clock_not_beyond() {
    struct HangingListing;

    #[async_trait::async_trait]
    impl GitRefs for HangingListing {
        async fn list_refs(&self, _repo: &RepoRecord) -> Result<Vec<RefLine>, StorageError> {
            futures::future::pending().await
        }

        async fn apply_updates(
            &self,
            _repo: &RepoRecord,
            _updates: Vec<RefUpdate>,
            _atomic: bool,
        ) -> Result<Vec<RefOutcome>, StorageError> {
            futures::future::pending().await
        }
    }

    let fx = fixture();
    let refs: Arc<dyn GitRefs> = Arc::new(HangingListing);
    let sink = VecSink::default();
    let limits = Limits {
        session_wall_clock: std::time::Duration::from_millis(100),
        ..Limits::default()
    };
    let started = std::time::Instant::now();
    let result = serve_stateless_advertisement(
        &fx.marker,
        sink.clone(),
        Service::ReceivePack,
        refs,
        &limits,
        false,
    )
    .await;
    assert!(
        started.elapsed() < std::time::Duration::from_secs(5),
        "the listing did not hang past the wall clock"
    );
    let error = result.expect_err("a hung listing breaches the wall clock");
    match &error {
        SubstrateError::Protocol(message) => {
            assert!(message.contains("session_wall_clock"), "{message}");
        }
        other => panic!("a wall-clock breach is protocol-shaped: {other:?}"),
    }
    assert!(sink.body().is_empty(), "a breached emission wrote nothing");
}

/// A prepare double that counts its invocations — the observable for
/// the packless path (no prepare call) and the probe (no dispatch).
struct CountedIngest {
    inner: Arc<MemPackIngest>,
    runs: Arc<AtomicUsize>,
}

fn counting_ingest(inner: Arc<MemPackIngest>, runs: Arc<AtomicUsize>) -> Arc<dyn GitPackIngest> {
    Arc::new(CountedIngest { inner, runs })
}

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

/// A prepare double that records the push-options it received — the
/// observable for the V0 POST body's options-section routing.
struct OptionsRecorder {
    inner: Arc<MemPackIngest>,
    seen: Arc<StdMutex<Option<PushOptions>>>,
}

#[async_trait]
impl GitPackIngest for OptionsRecorder {
    async fn prepare(
        &self,
        record: &RepoRecord,
        updates: Vec<RefUpdate>,
        pack: Box<dyn std::io::Read + Send>,
        push_options: Option<PushOptions>,
        limits: Limits,
    ) -> Result<PreparedPush, StorageError> {
        *self.seen.lock().expect("seen is not poisoned") = push_options.clone();
        self.inner
            .prepare(record, updates, pack, push_options, limits)
            .await
    }
}

/// The one POST / one stream push flow over the entry (ADR-013 §3): an
/// ordinary push POST — two create commands with the caps
/// NUL-attached on the first, the command-section flush, then the pack
/// tail raw — dribbled in chunks across the body. The commands + tail
/// ride the push machine under the stateless tail discipline; the
/// report streams back sideband-framed (exactly one outer flush — the
/// one-shot-flush guard holds), the refs row-table reflects the CAS,
/// and the pack's objects land via the ingest handle.
///
/// Contract stamp: ADR-013 §3 (one POST / one stream, the raw tail),
/// §7 (the one-transaction CAS home), §8 (the sideband report,
/// request order), §12 (the http framing — the substrate sees bytes
/// in/out); ADR-020 §1 (the stateless EOF twin); ADR-005 §2 (the
/// response ends at the flush, the one-shot-flush guard, no 0002);
/// ADR-009 (the pack tail counts against max_pack_size in the relay);
/// core/w3-stateless-serving (the entry's ReceivePack branch serves
/// the machine).
#[tokio::test]
async fn an_ordinary_push_post_applies_and_reports_sideband_framed() {
    let fx = fixture();
    let record = fx.record();
    let chain = chain_pack();
    let commands = [
        create_command("refs/heads/main", chain.commit_b),
        create_command("refs/heads/dev", chain.commit_a),
    ];

    let body = body_parts(
        push_commands("report-status side-band-64k", &commands)
            .into_iter()
            .chain([
                chain.pack[..1].to_vec(),
                chain.pack[1..24].to_vec(),
                chain.pack[24..].to_vec(),
            ])
            .collect::<Vec<Vec<u8>>>(),
    );
    let (result, sink) = fx
        .serve(
            body,
            Limits::default(),
            Service::ReceivePack,
            default_pipeline(),
        )
        .await;
    result.unwrap();

    let packets = parse_pkt_lines(&sink.body());
    assert_eq!(
        sideband_report_lines(&sink.body()),
        vec![
            "unpack ok".to_owned(),
            "ok refs/heads/main".to_owned(),
            "ok refs/heads/dev".to_owned(),
        ]
    );
    assert!(
        !sink.body().windows(4).any(|w| w == b"0002"),
        "no 0002 anywhere in the response body"
    );
    assert!(packets.last().unwrap().is_empty(), "one flush ends it");
    assert_eq!(
        fx.refs
            .list_refs(&record)
            .await
            .unwrap()
            .iter()
            .map(|line| (line.name.clone(), line.oid))
            .collect::<BTreeSet<_>>(),
        BTreeSet::from([
            ("refs/heads/main".to_owned(), Some(chain.commit_b)),
            ("refs/heads/dev".to_owned(), Some(chain.commit_a)),
        ]),
        "the CAS applied the pushed updates"
    );
    assert!(
        fx.ingest.objects().contains(&record, &chain.commit_b),
        "the pushed objects landed via the ingest handle"
    );
}

/// The packless delete POST (ADR-020 §1's EOF-driven twin): the
/// command section's flush ends the body — the tail's EOF at its start
/// takes the packless, CAS-only path (no prepare call), and the same
/// sideband report shape the duplex capture pinned streams back.
///
/// Contract stamp: ADR-020 §1 (rule (ii)'s EOF mapping — zero tail
/// bytes = the packless CAS-only path; the report shape equals the
/// duplex capture); ADR-013 §3, §7–8; core/w3-stateless-serving.
#[tokio::test]
async fn a_packless_delete_post_takes_the_eof_driven_packless_path() {
    let fx = fixture();
    let record = fx.record();
    let current = oid(0x1b);
    fx.refs
        .apply_updates(
            &record,
            vec![RefUpdate {
                name: "refs/heads/gone".to_owned(),
                expected: None,
                new: Some(current),
            }],
            false,
        )
        .await
        .unwrap();

    let runs = Arc::new(AtomicUsize::new(0));
    let ingest = counting_ingest(Arc::clone(&fx.ingest), Arc::clone(&runs));
    let refs: Arc<dyn GitRefs> = fx.refs.clone();
    let body = body_parts(vec![push_commands(
        "report-status side-band-64k",
        &[delete_command("refs/heads/gone", current)],
    )
    .concat()]);
    let (result, sink) = fx
        .serve_push(
            body,
            Limits::default(),
            false,
            refs,
            ingest,
            default_pipeline(),
        )
        .await;
    result.unwrap();

    assert_eq!(
        sideband_report_lines(&sink.body()),
        vec!["unpack ok".to_owned(), "ok refs/heads/gone".to_owned()],
        "the same report shape the duplex capture pinned"
    );
    assert_eq!(
        runs.load(AtomicOrdering::SeqCst),
        0,
        "the packless path runs no prepare call"
    );
    assert!(
        !fx.refs
            .list_refs(&record)
            .await
            .unwrap()
            .iter()
            .any(|line| line.name == "refs/heads/gone"),
        "the CAS deleted the ref"
    );
}

/// The gate-on push-options section is consumed before the pack tail
/// (ADR-020 §1 rule (i)): a push POST whose body carries commands +
/// flush + the options section + flush + the pack parses the options
/// through to `prepare` and the pack still ingests byte-exact — the
/// escape sits byte-exact past the *second* flush.
///
/// Contract stamp: ADR-013 §11 (the section's grammar — bare
/// pkt-lines, its flush, the negotiated gate); ADR-020 §1 rule (i)
/// (consumed before the peek); ADR-005 §2 (the body's framing);
/// core/w3-stateless-serving (the reader's section-two decision rides
/// the client's capability selection).
#[tokio::test]
async fn the_negotiated_push_options_section_is_consumed_before_the_pack_tail() {
    let fx = fixture();
    let record = fx.record();
    let chain = chain_pack();

    let seen = Arc::new(StdMutex::new(None));
    let ingest = Arc::new(OptionsRecorder {
        inner: Arc::clone(&fx.ingest),
        seen: Arc::clone(&seen),
    }) as Arc<dyn GitPackIngest>;
    let refs: Arc<dyn GitRefs> = fx.refs.clone();

    let body = body_parts(
        push_commands(
            "report-status push-options side-band-64k",
            &[create_command("refs/heads/main", chain.commit_b)],
        )
        .into_iter()
        .chain(vec![
            pkt(b"deploy-key=abc"),
            FLUSH.to_vec(),
            chain.pack.clone(),
        ])
        .collect::<Vec<Vec<u8>>>(),
    );
    let (result, sink) = fx
        .serve_push(
            body,
            Limits::default(),
            true,
            refs,
            ingest,
            default_pipeline(),
        )
        .await;
    result.unwrap();

    assert_eq!(
        *seen.lock().expect("seen is not poisoned"),
        Some(PushOptions {
            pairs: vec![("deploy-key".to_owned(), "abc".to_owned())]
        }),
        "the negotiated options section routed into prepare"
    );
    assert_eq!(
        sideband_report_lines(&sink.body()),
        vec!["unpack ok".to_owned(), "ok refs/heads/main".to_owned()],
    );
    assert!(
        fx.ingest.objects().contains(&record, &chain.commit_b),
        "the pack after the options flush ingested byte-exact"
    );
}

/// The request-body budget fails closed on the push POST's framed
/// section (the client-fault class, 413-mapped at the door): a command
/// list beyond the budget errors before any backend call, nothing
/// written.
///
/// Contract stamp: ADR-009 (max_request_body's breach behavior — the
/// client-fault budget class); ADR-013 §3 (the framed command
/// sections are the body budget's scope; the pack tail counts against
/// max_pack_size in the relay); core/w3-stateless-serving.
#[tokio::test]
async fn a_push_post_body_budget_breach_fails_closed() {
    let fx = fixture();
    let runs = Arc::new(AtomicUsize::new(0));
    let ingest = counting_ingest(Arc::clone(&fx.ingest), Arc::clone(&runs));
    let refs: Arc<dyn GitRefs> = fx.refs.clone();
    let limits = Limits {
        max_request_body: 5,
        ..Limits::default()
    };
    let (result, sink) = fx
        .serve_push(
            body_parts(vec![push_commands(
                "report-status",
                &[create_command("refs/heads/main", oid(0x2c))],
            )
            .concat()]),
            limits,
            false,
            refs,
            ingest,
            default_pipeline(),
        )
        .await;
    assert_protocol_error(result, "max_request_body").await;
    assert!(sink.body().is_empty(), "the breach wrote nothing");
    assert_eq!(
        runs.load(AtomicOrdering::SeqCst),
        0,
        "the breach failed before any backend call"
    );
}

/// The push tail's pack-size budget fails closed (ADR-009's
/// max_pack_size in the relay — the client-fault class, 413-mapped at
/// the door): a pack beyond the budget errors the session before any
/// report is written, no ref applies.
///
/// Contract stamp: ADR-009 (the pack-size breach — the client-fault
/// budget class, fail-closed); ADR-013 §12; ADR-005 §2 (the pack is
/// raw in the same body — the relay counts its bytes);
/// core/w3-stateless-serving.
#[tokio::test]
async fn a_push_post_pack_size_breach_fails_closed() {
    let fx = fixture();
    let record = fx.record();
    let limits = Limits {
        max_pack_size: 1000,
        ..Limits::default()
    };
    let oversized: Vec<u8> = (0..10_000u32).map(|i| (i % 251) as u8).collect();
    let body = body_parts(
        push_commands(
            "report-status side-band-64k",
            &[create_command("refs/heads/main", oid(0x2c))],
        )
        .into_iter()
        .chain(oversized.chunks(300).map(|c| c.to_vec()))
        .collect::<Vec<Vec<u8>>>(),
    );
    let (result, sink) = fx
        .serve(body, limits, Service::ReceivePack, default_pipeline())
        .await;
    assert_protocol_error(result, "max_pack_size").await;
    assert!(sink.body().is_empty(), "no report over the broken stream");
    assert!(
        fx.refs.list_refs(&record).await.unwrap().is_empty(),
        "the breach applied no ref"
    );
}

/// The door-facing push pair over the entry: the advertisement GET
/// serves the V0 emission (smart prefix + flush + the ref
/// advertisement + flush), the POST — one body, commands + flush +
/// pack — applies and reports, and the next advertisement GET lists
/// the pushed ref: the stateless composition real clients cache and
/// re-derive per request, with no cross-request state.
///
/// Contract stamp: ADR-013 §1–2 (the V0 advertisement), §3, §7–8 (the
/// request + CAS + report), §12 (the http framing facts — the
/// Content-Types are the door's mapping, the substrate sees bytes);
/// ADR-005 §2 (the stateless response rules); ADR-007 (the marker
/// gates both emissions); core/w3-stateless-serving (the entry-level
/// composition the task verifies).
#[tokio::test]
async fn the_advertisement_get_and_push_post_pair_serve_the_v0_flow() {
    let fx = fixture();
    let record = fx.record();
    let chain = chain_pack();

    async fn advert(fx: &Fixture) -> Vec<String> {
        let sink = VecSink::default();
        let refs: Arc<dyn GitRefs> = fx.refs.clone();
        serve_stateless_advertisement(
            &fx.marker,
            sink.clone(),
            Service::ReceivePack,
            refs,
            &Limits::default(),
            false,
        )
        .await
        .unwrap();
        parse_pkt_lines(&sink.body())
            .iter()
            .map(|p| text(p))
            .collect()
    }

    let before = advert(&fx).await;
    assert_eq!(
        *before.last().expect("the advertisement is non-empty"),
        String::new(),
        "the advertisement ends at flush"
    );
    assert!(
        before[2].contains("capabilities^{"),
        "the empty repo's sentinel line: {:?}",
        before[2]
    );

    let body = body_parts(
        push_commands(
            "report-status side-band-64k",
            &[create_command("refs/heads/main", chain.commit_b)],
        )
        .into_iter()
        .chain([chain.pack.clone()])
        .collect::<Vec<Vec<u8>>>(),
    );
    let (result, sink) = fx
        .serve(
            body,
            Limits::default(),
            Service::ReceivePack,
            default_pipeline(),
        )
        .await;
    result.unwrap();
    assert_eq!(
        sideband_report_lines(&sink.body()),
        vec!["unpack ok".to_owned(), "ok refs/heads/main".to_owned()],
    );
    assert_eq!(
        fx.refs
            .list_refs(&record)
            .await
            .unwrap()
            .iter()
            .find(|line| line.name == "refs/heads/main")
            .and_then(|line| line.oid),
        Some(chain.commit_b),
        "the CAS applied the pushed update"
    );

    let mut after = advert(&fx).await;
    assert_eq!(
        after.pop(),
        Some(String::new()),
        "the advertisement ends at flush"
    );
    assert_eq!(
        after.len(),
        3,
        "the prefix flush + one ref line follow the prefix section"
    );
    let pushed_line = format!("{} refs/heads/main", chain.commit_b);
    assert!(
        after[2]
            .split('\0')
            .next()
            .expect("the ref line has its caps section")
            .starts_with(&pushed_line),
        "the pushed ref appears in the next advertisement: {:?}",
        after[2]
    );
}