alkgit 0.1.0

//! The stateless request/response substrate (ADR-005 §2): the
//! smart-http framing facts as machinery every http door's route runs
//! under.
//!
//! One POST body is one command invocation ([`StatelessRequest`]): the
//! machinery performs the **capability-dump skip** (the first POST
//! carries the cached `info/refs` response — the smart prefix and the
//! capability advertisement, each flush-delimited — ahead of the
//! command; the parse skips any number of flush-delimited sections
//! until the section carrying the `command=` line), the
//! **delim-aware arg split** (capability lines — `command=`, `agent=`,
//! `object-format=…` — before the `0001` delimiter, command args
//! after), and a **flush-only probe** resolution (remote-curl's
//! `probe_rpc` on the large-request path POSTs a bare flush before the
//! real request; answering an error status breaks the whole clone —
//! the probe reads as the empty-OK shape).
//!
//! The response end ([`StatelessResponse`]) embeds the response rules:
//! a response ends at a flush and **never carries `0002`** (remote-curl
//! synthesizes the stateless response-end packet itself — an explicit
//! `0002` in an http response body kills the client), data after the
//! flush is refused, and the flush is **one-shot per response** (the
//! idempotent-flush guard is mandatory: a pack stream's own trailer
//! flush followed by the handler's explicit flush would otherwise emit
//! a second flush and break the client — POC-3's recorded bug).
//!
//! Front-door-blind (convention 15): [`StatelessRequest`] is generic
//! over the request-body reader (`futures::io::AsyncRead`; the door
//! adapts its body stream with [`BodyReader`]) and the response writer
//! (`futures::io::AsyncWrite`), and no http/door type appears anywhere
//! below the stream — this layer sees bytes in, bytes out. The service
//! is *supplied by the door's route* (ADR-016: nothing in a POST body
//! distinguishes the services) and rides the inputs as the
//! state-machine selector; [`serve_stateless_post`] is the per-POST
//! entry that dispatches on it (one command invocation per POST: the
//! V2 fetch commands, or — receive-pack's service — the V0-framed push
//! request, [`StatelessRequest::read_post`]'s service-shaped parse),
//! and [`serve_stateless_advertisement`] is the door's `info/refs` GET
//! emission.
//!
//! The transport-level error taxonomy is [`SubstrateError`] with the
//! duplex machinery's semantics: terminal errors
//! ([`SubstrateError::Io`], [`SubstrateError::Protocol`]) are cached
//! and re-reported without re-entering the transport (the unconditional
//! error-break rule — an error-shaped read that resets and retries
//! spins a worker at 100%, POC-3's recorded bug class), and
//! protocol-shaped failures map to http statuses at the door (ADR-005
//! §4; client-fault budgets like the request-body limit map to 413 per
//! ADR-009's breach behavior).
//!
//! The receive-pack pack tail is the raw-tail escape's (ADR-013 §3): on
//! the push path the pack streams *after* the request's terminating
//! flush inside the same POST body, unframed —
//! [`StatelessRequest::read_post`] stops the parse at the last framed
//! flush and [`StatelessRequest::take_raw_tail`] surrenders the body's
//! remaining bytes as a raw async reader ([`RawTail`], the duplex
//! escape's twin), with the body-budget counter riding it for
//! observability. Reads after the surrender are refused machinery
//! misuse ([`SubstrateError::RawTailUnavailable`]) — one command
//! invocation per POST, the session ends after the response.

use std::io;
use std::marker::PhantomData;
use std::pin::Pin;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::task::{Context, Poll};

use bytes::Bytes;
use futures::io::AsyncWriteExt as FuturesAsyncWriteExt;
use futures::{AsyncRead as FuturesAsyncRead, AsyncWrite as FuturesAsyncWrite, Stream};
use gix_packetline::async_io::{encode, StreamingPeekableIter};
use gix_packetline::{Channel, PacketLineRef};
use tokio::sync::Semaphore;

use crate::authorize::AuthorizedRepo;
use crate::backend::{GitPackGen, GitPackIngest, GitRefs};
use crate::duplex::{CountingRead, RawTail, Request, STOP_AT_FLUSH, UNKNOWN_COMMAND_DECLINE};
use crate::fetch::{serve_fetch_round, storage_to_session_error, FetchRounds};
use crate::limits::{Limits, SIDEBAND_CHUNK_SIZE};
use crate::push::{serve_receive_pack, PackTailDiscipline};
use crate::substrate::{Service, SubstrateError};
use crate::v0::advertise_v0;
use crate::v2::{advertise_v2, parse_ls_refs_args, serve_ls_refs};

/// The generic request-body adapter (POC-3's `BodyReader`): bridges a
/// byte-chunk stream into the `futures::io::AsyncRead` end the pkt-line
/// codec consumes.
///
/// Streaming by construction: reads serve from the current chunk's
/// cursor and the next chunk is pulled only when the cursor drains —
/// no accumulation beyond one chunk. The door's http body stream (any
/// `Stream<Item = Result<Bytes, E>>`) plugs in directly; stream
/// failures surface as io errors with the source attached.
pub struct BodyReader<S, E> {
    inner: S,
    pending: Bytes,
    cursor: usize,
    error: PhantomData<fn() -> E>,
}

impl<S, E> BodyReader<S, E>
where
    S: Stream<Item = Result<Bytes, E>> + Unpin,
    E: std::error::Error + Send + Sync + 'static,
{
    /// Adapts a byte-chunk stream into the reader end.
    pub fn new(inner: S) -> Self {
        BodyReader {
            inner,
            pending: Bytes::new(),
            cursor: 0,
            error: PhantomData,
        }
    }
}

impl<S, E> FuturesAsyncRead for BodyReader<S, E>
where
    S: Stream<Item = Result<Bytes, E>> + Unpin,
    E: std::error::Error + Send + Sync + 'static,
{
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut [u8],
    ) -> Poll<io::Result<usize>> {
        let this = self.get_mut();
        loop {
            if this.cursor < this.pending.len() {
                let n = buf.len().min(this.pending.len() - this.cursor);
                buf[..n].copy_from_slice(&this.pending[this.cursor..this.cursor + n]);
                this.cursor += n;
                return Poll::Ready(Ok(n));
            }
            this.pending = Bytes::new();
            this.cursor = 0;
            match Pin::new(&mut this.inner).poll_next(cx) {
                Poll::Ready(Some(Ok(chunk))) => {
                    if chunk.is_empty() {
                        continue;
                    }
                    this.pending = chunk;
                }
                Poll::Ready(Some(Err(e))) => return Poll::Ready(Err(io::Error::other(e))),
                Poll::Ready(None) => return Poll::Ready(Ok(0)),
                Poll::Pending => return Poll::Pending,
            }
        }
    }
}

/// One parsed POST body (ADR-005 §2): a command invocation, or a
/// flush-only probe.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum StatelessPost {
    /// A flush-only (or empty) POST body — remote-curl's `probe_rpc`
    /// on the large-request path, answered with the **empty-OK shape**:
    /// git's own http-backend answers 200 with an empty body and
    /// upload-pack exits without output; answering an error status
    /// breaks the whole clone (`RPC failed; HTTP 400`). The machinery
    /// resolves the shape and writes nothing; the 200 status mapping is
    /// the door's (ADR-005 §4), and a flush-terminated empty body via
    /// [`StatelessResponse::write_flush`] is the POC-3-verified
    /// alternative form.
    Probe,
    /// A command invocation: the capability-dump sections skipped, the
    /// command line and capability lines before the `0001` delimiter
    /// split from the args after it, terminated by the client's flush.
    Command(Request),
}

/// The stateless response end (ADR-005 §2): the http response body's
/// pkt-line writer, structurally confined to the response rules.
///
/// Only data framing ([`StatelessResponse::write_text_line`],
/// [`StatelessResponse::write_data_band`]) and the **one-shot flush**
/// ([`StatelessResponse::write_flush`]) exist here: `0002` is never
/// emitted (remote-curl synthesizes it; an explicit one in an http
/// response body kills the client), data after the flush is refused
/// (a second flush-delimited section inside one response is the POC-3
/// recorded client-killer), and the flush is idempotent — the guard is
/// mandatory because a pack stream's own trailer flush and the
/// handler's explicit flush both arrive through this surface.
///
/// Error mapping is the door's, not a wire shape: protocol failures map
/// to http statuses (ADR-005 §4), so this end has no error-band writer.
pub struct StatelessResponse<W> {
    writer: W,
    flushed: bool,
}

impl<W> StatelessResponse<W> {
    /// Wraps the response writer into the response-rules end — the
    /// serving entries' construction (the one-shot-flush guard starts
    /// unflushed).
    pub(crate) fn new(writer: W) -> Self {
        StatelessResponse {
            writer,
            flushed: false,
        }
    }
}

impl<W: FuturesAsyncWrite + Unpin> StatelessResponse<W> {
    /// Whether the one-shot section flush has already been emitted
    /// (every later [`StatelessResponse::write_flush`] is a no-op).
    pub fn is_flushed(&self) -> bool {
        self.flushed
    }

    fn after_flush_error(&self) -> SubstrateError {
        SubstrateError::Protocol(
            "data written after the response flush (the flush is one-shot per response)".to_owned(),
        )
    }

    /// Writes one pkt-line text line (newline-augmented by the codec).
    pub async fn write_text_line(&mut self, line: &[u8]) -> Result<(), SubstrateError> {
        if self.flushed {
            return Err(self.after_flush_error());
        }
        encode::text_to_write(line, &mut self.writer).await?;
        Ok(())
    }

    /// Writes band-1 sideband data, chunked at [`SIDEBAND_CHUNK_SIZE`]
    /// (65000 — the protocol's fixed cap, with headroom under the
    /// pkt-line ceiling). The packfile section streams through this.
    pub async fn write_data_band(&mut self, data: &[u8]) -> Result<(), SubstrateError> {
        if self.flushed {
            return Err(self.after_flush_error());
        }
        for chunk in data.chunks(SIDEBAND_CHUNK_SIZE) {
            encode::band_to_write(Channel::Data, chunk, &mut self.writer).await?;
        }
        Ok(())
    }

    /// Writes the terminating section flush — **once per response**
    /// (the idempotent-flush guard; every later call is a no-op), and
    /// pushes the framed bytes through to the underlying sink (the
    /// flush frame alone does not guarantee it).
    pub async fn write_flush(&mut self) -> Result<(), SubstrateError> {
        if self.flushed {
            return Ok(());
        }
        encode::flush_to_write(&mut self.writer).await?;
        FuturesAsyncWriteExt::flush(&mut self.writer).await?;
        self.flushed = true;
        Ok(())
    }
}

/// The stateless request/response substrate per POST (ADR-005 §2):
/// wraps (request-reader, response-writer, [`Limits`],
/// [`Service`](crate::Service)) into the machinery that performs the
/// capability-dump skip, the delim-aware arg split, and the flush-only
/// probe rule on the read side, and hands the response rules to the
/// machine on the write side.
///
/// The transport-level error taxonomy is [`SubstrateError`] with the
/// duplex machinery's semantics: terminal errors are cached and
/// re-reported without re-entering the transport (the unconditional
/// error-break rule, observable via
/// [`StatelessRequest::transport_reads`]).
pub struct StatelessRequest<R, W> {
    limits: Limits,
    service: Service,
    reader: Option<StreamingPeekableIter<CountingRead<R>>>,
    response: StatelessResponse<W>,
    consumed: Arc<AtomicU64>,
    reads: Arc<AtomicU64>,
    read_complete: bool,
    command_parsed: bool,
    terminal: Option<SubstrateError>,
}

impl<R, W> StatelessRequest<R, W>
where
    R: FuturesAsyncRead + Unpin,
    W: FuturesAsyncWrite + Unpin,
{
    /// Wraps one POST's (request-reader, response-writer, limits,
    /// service) per ADR-005 §2. The service rides the inputs per
    /// ADR-016 (the door's route supplies it — nothing in the POST
    /// body distinguishes the services); dispatch on it is wave 2.
    pub fn new(reader: R, writer: W, limits: Limits, service: Service) -> Self {
        let consumed = Arc::new(AtomicU64::new(0));
        let reads = Arc::new(AtomicU64::new(0));
        StatelessRequest {
            limits,
            service,
            reader: Some(StreamingPeekableIter::new(
                CountingRead {
                    inner: reader,
                    consumed: Arc::clone(&consumed),
                    reads: Arc::clone(&reads),
                },
                STOP_AT_FLUSH,
                false,
            )),
            response: StatelessResponse {
                writer,
                flushed: false,
            },
            consumed,
            reads,
            read_complete: false,
            command_parsed: false,
            terminal: None,
        }
    }

    /// The service this POST was accepted for (ADR-016 §3).
    pub fn service(&self) -> Service {
        self.service
    }

    /// The session's budget bundle (ADR-009; the request-body budget
    /// is enforced by [`StatelessRequest::read_post`], the rest is the
    /// later waves' machines' to reach).
    pub fn limits(&self) -> &Limits {
        &self.limits
    }

    /// Request-body bytes consumed by the read side so far.
    pub fn consumed_bytes(&self) -> u64 {
        self.consumed.load(Ordering::Relaxed)
    }

    /// Transport read calls issued so far — the observable behind the
    /// unconditional error-break rule: after a terminal read error this
    /// counter must stop advancing.
    pub fn transport_reads(&self) -> u64 {
        self.reads.load(Ordering::Relaxed)
    }

    /// The response end (ADR-005 §2's response rules: one-shot flush,
    /// no `0002`, data-after-flush refused).
    pub fn response(&mut self) -> &mut StatelessResponse<W> {
        &mut self.response
    }

    /// The raw-tail escape (ADR-013 §3; ADR-018 §4): after
    /// [`StatelessRequest::read_post`] resolved a command invocation,
    /// the pkt-line machinery surrenders the POST body's remaining
    /// bytes as a raw async reader ([`RawTail`]) — the pack follows the
    /// command flush **unframed** inside the same body. The no-byte-loss
    /// argument mirrors the duplex escape: the demand-driven iterator
    /// reads with exact framing and clears its buffer at the stop, so
    /// at surrender the body reader is positioned byte-exactly after
    /// the terminating flush; the body-budget consumed counter rides
    /// the tail for observability (the POST's total reconciles).
    ///
    /// Machinery misuse is refused, never silently re-entered:
    ///
    /// - before a command-shaped parse (no [`StatelessRequest::read_post`]
    ///   yet, or the probe shape — a probe body *has* no tail) —
    ///   [`SubstrateError::RawTailUnavailable`], the reader stays in
    ///   place;
    /// - after a terminal error — the cached terminal error;
    /// - a second escape — [`SubstrateError::RawTailUnavailable`] (the
    ///   escape is one-shot); every later
    ///   [`StatelessRequest::read_post`] stays refused (one command
    ///   invocation per POST).
    ///
    /// Contract stamp: ADR-013 §3, ADR-020 §1, ADR-018 §4, ADR-005 §2,
    /// ADR-009 (the body-budget observability).
    pub fn take_raw_tail(&mut self) -> Result<RawTail<R>, SubstrateError> {
        if let Some(cached) = self.terminal.clone() {
            return Err(cached);
        }
        let Some(reader) = self.reader.take() else {
            return Err(SubstrateError::RawTailUnavailable(
                "the raw tail was already surrendered — the escape is one-shot",
            ));
        };
        if !self.command_parsed {
            self.reader = Some(reader);
            return Err(SubstrateError::RawTailUnavailable(
                "the POST parse has not produced a command — the tail escapes only after the command invocation's terminating flush",
            ));
        }
        let counting = reader.into_inner();
        Ok(RawTail::new(counting.inner, counting.consumed))
    }

    /// Reads the POST body as one command invocation (ADR-005 §2). The
    /// request shape is the service's ([`Service::UploadPack`] — the V2
    /// command invocation; [`Service::ReceivePack`] — the V0-framed push
    /// request, ADR-013 §3), and the parse rules — terminal failures,
    /// the probe shape, the per-POST refusal — are service-shared.
    pub async fn read_post(&mut self) -> Result<StatelessPost, SubstrateError> {
        match self.service {
            Service::UploadPack => self.read_post_upload_pack().await,
            Service::ReceivePack => self.read_post_receive_pack().await,
        }
    }

    /// Reads the POST body as one V0-framed push request (ADR-013 §3 —
    /// one POST / one stream): the **command lines** (`<old> <new> <ref>`,
    /// caps NUL-attached on the first line) terminated by the command
    /// section's flush, then — when the client selected the
    /// `push-options` capability — the push-options section's bare
    /// pkt-lines terminated by its flush (ADR-020 §1 rule (i): the
    /// options section is consumed *before* the pack tail is ever
    /// looked at), and everything past the last flush is the raw pack
    /// tail ([`StatelessRequest::take_raw_tail`]) — exactly the
    /// body real clients send (push-captures.md §Request; no
    /// `command=` line, no delimiter — the receive-pack POST body has
    /// no V2 wrapper).
    ///
    /// The section-2 decision is the client's capability **selection**
    /// (upstream's shape: the section exists only when the client
    /// selected `push-options`); the negotiated-gate check rides the
    /// grammar's fail-closed enforcement in
    /// [`parse_push_request`](crate::parse_push_request) — a selection
    /// where the gate was off is a session error there, and a section
    /// the grammar never sent fails closed the same way (misparsed
    /// pack bytes cannot form valid options). With no selection the
    /// parse stops at the command flush and the tail reads unframed —
    /// an EOF right there is the packless path's zero bytes (the
    /// machine's EOF twin, ADR-020 §1).
    ///
    /// The probe shape stays service-agnostic (ADR-005 §2, ADR-013 §12):
    /// a flush-only or empty body — the probe POST any large push or
    /// fetch sends, and on the push path byte-identical to nothing-to-do
    /// (which real clients answer client-side; upstream's http-backend
    /// answers it 200-empty all the same) — resolves to
    /// [`StatelessPost::Probe`] before any service dispatch. Truncated
    /// bodies (an EOF before the section's flush, mid- or post-content)
    /// are terminal, never reset-and-retried; the request-body budget
    /// fails closed on the framed sections (the raw tail counts against
    /// [`Limits::max_pack_size`] in the pack relay instead); `0001`
    /// delimiters and `0002` response-end packets are out of the V0
    /// grammar; a second parse is refused; terminal errors re-report
    /// cached without re-entering the transport.
    ///
    /// Contract stamp: ADR-013 §3 (one POST / one stream, the V0 body
    /// shape), §12 (the http framing facts); ADR-020 §1 rule (i) (the
    /// options section is consumed before the peek); ADR-005 §2 (the
    /// probe shape, the per-POST rules); ADR-009 (the body budget, the
    /// unconditional error break); push-captures.md §Request;
    /// core/w3-stateless-serving (the entry's reader shape).
    async fn read_post_receive_pack(&mut self) -> Result<StatelessPost, SubstrateError> {
        let Self {
            limits,
            reader,
            consumed,
            read_complete,
            command_parsed,
            terminal,
            ..
        } = self;
        if let Some(cached) = terminal.clone() {
            return Err(cached);
        }
        if *read_complete {
            return Err(SubstrateError::Protocol(
                "one command invocation per POST; this body is already parsed".to_owned(),
            ));
        }
        let Some(reader) = reader else {
            return Err(SubstrateError::RawTailUnavailable(
                "the pkt-line reader is gone — the raw tail was surrendered; one command invocation per POST",
            ));
        };
        let consumed_entry = consumed.load(Ordering::Relaxed);

        enum Event {
            Stop,
            Text(String),
            StopPacket(String),
            ErrLine(String),
        }

        // The two framed sections of the V0 body (ADR-020 §1 rule (i)):
        // the command lines first, then — only when the client selected
        // `push-options` — the options section. `in_options` selects
        // which side collects the section's lines.
        let mut commands: Vec<String> = Vec::new();
        let mut args: Vec<String> = Vec::new();
        let mut in_options = false;
        let mut at_section_start = true;

        loop {
            let event: Event = match reader.read_line().await {
                None => Event::Stop,
                Some(Ok(Err(decode_error))) => {
                    let error = SubstrateError::Protocol(format!(
                        "malformed pkt-line in the POST body: {decode_error}"
                    ));
                    *terminal = Some(error.clone());
                    return Err(error);
                }
                Some(Err(io_error)) => {
                    // The body ends where the transport does. A clean
                    // end sits at a section boundary
                    // (`at_section_start`): with the body untouched, or
                    // the section having carried only its flush, it is
                    // the probe; with the command section read but
                    // nothing after its flush (and no options selection
                    // pending) the body simply ended — the zero-byte
                    // tail, the packless path's EOF twin. A body ending
                    // mid-section, or an unfinished options section, is
                    // a truncation — terminal, never reset-and-retried
                    // (POC-3's spin bug class).
                    if io_error.kind() == io::ErrorKind::UnexpectedEof && at_section_start {
                        let untouched = consumed.load(Ordering::Relaxed) == consumed_entry
                            && commands.is_empty()
                            && args.is_empty();
                        if untouched || commands.is_empty() {
                            // An absolutely empty body, or one carrying
                            // only a flush: the service-agnostic probe.
                            *read_complete = true;
                            return Ok(StatelessPost::Probe);
                        }
                        if !in_options {
                            let request = Request {
                                command: "git-receive-pack".to_owned(),
                                pre_lines: std::mem::take(&mut commands),
                                args: Vec::new(),
                            };
                            *read_complete = true;
                            *command_parsed = true;
                            return Ok(StatelessPost::Command(request));
                        }
                    }
                    let error = SubstrateError::Io(io_error);
                    *terminal = Some(error.clone());
                    return Err(error);
                }
                Some(Ok(Ok(line))) => {
                    if let Some(message) = line
                        .check_error()
                        .map(|e| String::from_utf8_lossy(e.0).into_owned())
                    {
                        Event::ErrLine(message)
                    } else {
                        match line {
                            PacketLineRef::Data(d) => {
                                Event::Text(String::from_utf8_lossy(d).into_owned())
                            }
                            PacketLineRef::Delimiter => Event::StopPacket(
                                "a 0001 delimiter in a receive-pack POST body is not a push shape"
                                    .to_owned(),
                            ),
                            // Defensive: the reader is configured to
                            // stop at flush (a flush surfaces as None),
                            // so an inline flush is handled as the
                            // section stop it means.
                            PacketLineRef::Flush => Event::Stop,
                            PacketLineRef::ResponseEnd => Event::StopPacket(
                                "a response-end packet (0002) in a POST body is not a request shape"
                                    .to_owned(),
                            ),
                        }
                    }
                }
            };

            match event {
                Event::Stop => {
                    // The section's flush stopped the iteration — the
                    // stream sits byte-exact past it (the raw-tail
                    // escape's no-loss argument). An empty command
                    // section is the flush-only probe body
                    // (service-agnostic, ADR-005 §2 — on the push path
                    // this is also the nothing-to-do shape's bytes, and
                    // real clients answer it client-side; upstream
                    // http-backend answers it 200-empty all the same).
                    // After a non-empty command section's flush, the
                    // options section is decided from the first command
                    // line's caps (upstream's shape: the section exists
                    // only when the client selected `push-options`; the
                    // negotiated-gate check is the grammar's
                    // fail-closed enforcement); with no selection the
                    // parse stops dead — the tail beyond the flush is
                    // the pack stream's. The options section's own
                    // flush ends the parse — the tail reads raw past
                    // it.
                    reader.reset();
                    at_section_start = true;
                    if commands.is_empty() && !in_options {
                        *read_complete = true;
                        return Ok(StatelessPost::Probe);
                    }
                    if !in_options
                        && commands
                            .first()
                            .and_then(|line| line.split_once('\0'))
                            .map(|(_, caps)| caps.split_whitespace().any(|t| t == "push-options"))
                            .unwrap_or(false)
                    {
                        in_options = true;
                        continue;
                    }
                    let (pre_lines, args) = if in_options {
                        (std::mem::take(&mut commands), std::mem::take(&mut args))
                    } else {
                        (std::mem::take(&mut commands), Vec::new())
                    };
                    let request = Request {
                        command: "git-receive-pack".to_owned(),
                        pre_lines,
                        args,
                    };
                    *read_complete = true;
                    *command_parsed = true;
                    return Ok(StatelessPost::Command(request));
                }
                Event::StopPacket(message) => {
                    let error = SubstrateError::Protocol(message);
                    *terminal = Some(error.clone());
                    return Err(error);
                }
                Event::ErrLine(message) => {
                    let error =
                        SubstrateError::Protocol(format!("peer sent an ERR line: {message}"));
                    *terminal = Some(error.clone());
                    return Err(error);
                }
                Event::Text(text) => {
                    let text = text.trim_end_matches(['\n', '\r']).to_owned();
                    if in_options {
                        args.push(text);
                    } else {
                        commands.push(text);
                    }
                    at_section_start = false;
                }
            }

            if consumed.load(Ordering::Relaxed) > limits.max_request_body {
                let error = SubstrateError::Protocol(format!(
                    "the request body exceeds the max_request_body budget ({})",
                    limits.max_request_body
                ));
                *terminal = Some(error.clone());
                return Err(error);
            }
        }
    }

    /// Reads the POST body as one V2 command invocation (ADR-005 §2, the
    /// upload-pack request shape).
    ///
    /// - `Ok(StatelessPost::Command(request))` — the capability-dump
    ///   sections (any number; only the *first* POST carries them) were
    ///   skipped, the command line and capability lines before the
    ///   `0001` delimiter were split from the args after it, and the
    ///   client's flush terminated the parse.
    /// - `Ok(StatelessPost::Probe)` — the body was flush-only (or
    ///   empty): the empty-OK probe shape; the door answers 200-empty.
    /// - `Err(SubstrateError::Io | SubstrateError::Protocol)` — terminal
    ///   ([`SubstrateError::is_terminal`]): malformed pkt-lines, client
    ///   `ERR` lines, `0002` in the request body, repeated delimiters
    ///   or command lines, a body ending mid-section (truncation), or a
    ///   body that ends at a section boundary without ever carrying a
    ///   `command=` line. The [`Limits::max_request_body`] budget
    ///   breach is a client-fault protocol failure (mapped to 413 at the
    ///   door per ADR-009's breach behavior). Later calls re-report the
    ///   cached terminal error without touching the transport — the
    ///   unconditional error-break rule, POC-3's spin bug class.
    /// - A second [`StatelessRequest::read_post`] on an already-parsed
    ///   POST is refused with [`SubstrateError::Protocol`]: one command
    ///   invocation per POST.
    ///
    /// The parse leaves any bytes after the command invocation's
    /// terminating flush (receive-pack's pack tail, ADR-013 §3) for the
    /// raw-tail escape ([`StatelessRequest::take_raw_tail`]) — the pack
    /// is unframed there, so the pkt-line machinery bows out.
    async fn read_post_upload_pack(&mut self) -> Result<StatelessPost, SubstrateError> {
        let Self {
            limits,
            reader,
            consumed,
            read_complete,
            command_parsed,
            terminal,
            ..
        } = self;
        if let Some(cached) = terminal.clone() {
            return Err(cached);
        }
        if *read_complete {
            return Err(SubstrateError::Protocol(
                "one command invocation per POST; this body is already parsed".to_owned(),
            ));
        }
        let Some(reader) = reader else {
            return Err(SubstrateError::RawTailUnavailable(
                "the pkt-line reader is gone — the raw tail was surrendered; one command invocation per POST",
            ));
        };
        let consumed_entry = consumed.load(Ordering::Relaxed);

        enum Event {
            Stop,
            Text(String),
            Delimiter,
            StopPacket(String),
            ErrLine(String),
        }

        let mut command: Option<String> = None;
        let mut pre_lines: Vec<String> = Vec::new();
        let mut args: Vec<String> = Vec::new();
        let mut delim_seen = false;
        let mut content_seen = false;
        let mut at_section_start = true;

        loop {
            let event: Event = match reader.read_line().await {
                None => Event::Stop,
                Some(Ok(Err(decode_error))) => {
                    let error = SubstrateError::Protocol(format!(
                        "malformed pkt-line in the POST body: {decode_error}"
                    ));
                    *terminal = Some(error.clone());
                    return Err(error);
                }
                Some(Err(io_error)) => {
                    // The body ends where the transport does. A clean
                    // end sits at a section boundary (`at_section_start`
                    // — after a flush, or before anything was read); a
                    // body ending mid-section is truncation. Truncation
                    // and other transport failures are terminal and
                    // never reset-and-retried (the POC-3 spin bug class):
                    // the error is cached and every later call
                    // re-reports it without re-entering the transport.
                    if io_error.kind() == io::ErrorKind::UnexpectedEof && at_section_start {
                        let untouched = consumed.load(Ordering::Relaxed) == consumed_entry
                            && command.is_none()
                            && pre_lines.is_empty()
                            && args.is_empty()
                            && !delim_seen
                            && !content_seen;
                        if untouched {
                            // An absolutely empty body is the
                            // body-less probe.
                            *read_complete = true;
                            return Ok(StatelessPost::Probe);
                        }
                        let error = if content_seen {
                            SubstrateError::Protocol(
                                "the POST body ended without a command= line".to_owned(),
                            )
                        } else {
                            // Flush-only body: the remote-curl probe.
                            *read_complete = true;
                            return Ok(StatelessPost::Probe);
                        };
                        *terminal = Some(error.clone());
                        return Err(error);
                    }
                    let error = SubstrateError::Io(io_error);
                    *terminal = Some(error.clone());
                    return Err(error);
                }
                Some(Ok(Ok(line))) => {
                    if let Some(message) = line
                        .check_error()
                        .map(|e| String::from_utf8_lossy(e.0).into_owned())
                    {
                        Event::ErrLine(message)
                    } else {
                        match line {
                            PacketLineRef::Data(d) => Event::Text(
                                String::from_utf8_lossy(d)
                                    .trim_end_matches(['\n', '\r'])
                                    .to_owned(),
                            ),
                            PacketLineRef::Delimiter => Event::Delimiter,
                            // Defensive: the reader is configured to
                            // stop at flush (a flush surfaces as None),
                            // so an inline flush is handled as the
                            // section stop it means.
                            PacketLineRef::Flush => Event::Stop,
                            PacketLineRef::ResponseEnd => Event::StopPacket(
                                "a response-end packet (0002) in a POST body is not a request shape"
                                    .to_owned(),
                            ),
                        }
                    }
                }
            };

            match event {
                Event::Stop => {
                    // The section's (or the probe's lone) flush stopped
                    // the iteration. A section without the command line
                    // is capability-dump content: discard its lines and
                    // keep skipping.
                    reader.reset();
                    at_section_start = true;
                    if let Some(command) = command.take() {
                        *read_complete = true;
                        *command_parsed = true;
                        return Ok(StatelessPost::Command(Request {
                            command,
                            pre_lines: std::mem::take(&mut pre_lines),
                            args: std::mem::take(&mut args),
                        }));
                    }
                    pre_lines.clear();
                    args.clear();
                    delim_seen = false;
                }
                Event::StopPacket(message) => {
                    let error = SubstrateError::Protocol(message);
                    *terminal = Some(error.clone());
                    return Err(error);
                }
                Event::ErrLine(message) => {
                    let error =
                        SubstrateError::Protocol(format!("peer sent an ERR line: {message}"));
                    *terminal = Some(error.clone());
                    return Err(error);
                }
                Event::Text(text) => {
                    content_seen = true;
                    at_section_start = false;
                    if delim_seen {
                        args.push(text);
                    } else if let Some(value) = text.strip_prefix("command=") {
                        if command.is_some() {
                            let error = SubstrateError::Protocol(
                                "the POST body carries a second command line".to_owned(),
                            );
                            *terminal = Some(error.clone());
                            return Err(error);
                        }
                        if value.is_empty() {
                            let error = SubstrateError::Protocol("empty command= line".to_owned());
                            *terminal = Some(error.clone());
                            return Err(error);
                        }
                        command = Some(value.to_owned());
                    } else {
                        pre_lines.push(text);
                    }
                }
                Event::Delimiter => {
                    if delim_seen {
                        let error = SubstrateError::Protocol("repeated 0001 delimiter".to_owned());
                        *terminal = Some(error.clone());
                        return Err(error);
                    }
                    delim_seen = true;
                    content_seen = true;
                    at_section_start = false;
                }
            }

            if consumed.load(Ordering::Relaxed) > limits.max_request_body {
                let error = SubstrateError::Protocol(format!(
                    "the request body exceeds the max_request_body budget ({})",
                    limits.max_request_body
                ));
                *terminal = Some(error.clone());
                return Err(error);
            }
        }
    }
}

/// The stateless serving entry (ADR-002's second session entry point,
/// the stateless tuple of ADR-016 §3): one POST body is one command
/// invocation, the same machines as the duplex session run under it
/// with **no cross-request state** (ADR-014 §7 — each POST re-derives
/// cleanly, the client re-sends wants + commons every round, and this
/// entry holds nothing between invocations).
///
/// The inputs are the transport.md §substrate stateless tuple: the
/// [`AuthorizedRepo`] marker (the door constructs it after its
/// resolve → authorize sequence, ADR-007 — the entry never resolves or
/// authorizes, it asserts the marker's presence by type), the
/// request-response pair, the [`Limits`], the [`Service`] (the
/// state-machine selector, ADR-016), the backend handles the machines
/// are called through, the `push-options` negotiated-gate flag (the
/// same assembler config value the advertisement GET takes, ADR-013
/// §11 — the machine's grammar enforces the gate against it), and the
/// *shared* pipeline-concurrency semaphore the door constructs once
/// from `Limits::max_concurrent_blocking_tasks` (the permit wraps the
/// `generate`/`prepare` calls as on the duplex side, the ADR-009
/// enforcement point — backend.md §concurrency model).
///
/// Per-POST behavior:
///
/// - [`StatelessPost::Probe`] resolves to the **empty-OK shape** with
///   nothing written — the probe POST any large push or fetch sends
///   (ADR-013 §12) is answered the way the clone survives; the 200
///   mapping is the door's (ADR-005 §4). The probe is answered before
///   any service dispatch: the probe shape is service-agnostic
///   (breaking it breaks the large-request path of *any* client).
/// - [`StatelessPost::Command`] dispatches on the service first (ADR-016
///   §3: the service is the state-machine selector) and then on the
///   request shape: UploadPack's `ls-refs` → [`serve_ls_refs`],
///   `fetch` → [`serve_fetch_round`]; an unknown command is a typed
///   protocol failure — the stateless side has no error band, so the
///   door maps the returned [`SubstrateError`] to the http status
///   (ADR-005 §4). ReceivePack's POST is the V0-framed push request
///   ([`StatelessRequest::read_post`]'s V0 parse): the parse stops at
///   the last framed flush, the body's remaining bytes escape the raw
///   tail ([`StatelessRequest::take_raw_tail`]) and the invocation
///   rides [`serve_receive_pack`] under the **stateless tail
///   discipline** (`PackTailDiscipline::Stateless` — EOF at the tail's
///   start = packless; EOF after a partial header = malformed; bytes →
///   ingestion, ADR-020 §1): the report rides the response through the
///   report writer (sideband-selected framing), and the response ends
///   at the flush — the one-shot-flush guard holds. No cross-request
///   state: one push per POST, the session ends after the response.
///
/// Per-request budgets (ADR-009; the stateless substrate has no session
/// to amortize them, ADR-002's cost note): the request-body budget is
/// the machinery's ([`StatelessRequest::read_post`]); the push's pack
/// tail counts against [`Limits::max_pack_size`] in the pack relay (the
/// client-fault class, 413-mapped at the door); the wall clock applies
/// **per request** — the whole invocation runs under
/// [`Limits::session_wall_clock`], breach is a typed failure of the
/// server/session class (503-mapped at the door). A breach abandons the
/// in-flight POST future; a blocking pipeline already launched for the
/// done round or the pack ingestion runs to completion under its own
/// permit — the permit accounting stays exact.
///
/// Negotiation rounds are budget-accounted per POST (a fresh
/// [`FetchRounds`] per invocation, so the fetch machine's grammar holds
/// for the one round the POST carries); cross-request round counting
/// would be cross-request state, which this entry by definition holds
/// none of — the per-request wall clock and the door's rate limiting
/// bound POST volume instead.
///
/// (The ten inputs are the pinned stateless tuple itself — ADR-016 §3's
/// eight plus the wave-3 push threading — not a shape to flatten.)
#[allow(clippy::too_many_arguments)]
pub async fn serve_stateless_post<R, W>(
    repo: &AuthorizedRepo,
    reader: R,
    writer: W,
    limits: Limits,
    service: Service,
    refs: Arc<dyn GitRefs>,
    packgen: Arc<dyn GitPackGen>,
    ingest: Arc<dyn GitPackIngest>,
    push_options_negotiated: bool,
    pipeline: Arc<Semaphore>,
) -> Result<(), SubstrateError>
where
    R: FuturesAsyncRead + Unpin + Send + 'static,
    W: FuturesAsyncWrite + Unpin + Send,
{
    let mut post = StatelessRequest::new(reader, writer, limits, service);
    match tokio::time::timeout(
        limits.session_wall_clock,
        drive_one_post(
            repo,
            &mut post,
            service,
            refs,
            packgen,
            ingest,
            push_options_negotiated,
            limits,
            pipeline,
        ),
    )
    .await
    {
        Ok(result) => result,
        Err(_elapsed) => Err(wall_clock_breach(&limits)),
    }
}

#[allow(clippy::too_many_arguments)]
async fn drive_one_post<R, W>(
    repo: &AuthorizedRepo,
    post: &mut StatelessRequest<R, W>,
    service: Service,
    refs: Arc<dyn GitRefs>,
    packgen: Arc<dyn GitPackGen>,
    ingest: Arc<dyn GitPackIngest>,
    push_options_negotiated: bool,
    limits: Limits,
    pipeline: Arc<Semaphore>,
) -> Result<(), SubstrateError>
where
    R: FuturesAsyncRead + Unpin + Send + 'static,
    W: FuturesAsyncWrite + Unpin + Send,
{
    match post.read_post().await? {
        StatelessPost::Probe => Ok(()),
        StatelessPost::Command(request) => match service {
            Service::UploadPack => match request.command.as_str() {
                "ls-refs" => {
                    let args = parse_ls_refs_args(&request.args)?;
                    let listed = refs
                        .list_refs(repo.record())
                        .await
                        .map_err(storage_to_session_error)?;
                    serve_ls_refs(post.response(), listed, &args, &limits).await
                }
                "fetch" => {
                    // Fresh per POST: budget accounting, not negotiation
                    // state — one command invocation re-derives per POST
                    // (ADR-014 §7; no cross-request state is held).
                    let mut rounds = FetchRounds::default();
                    serve_fetch_round(
                        post.response(),
                        &request,
                        &mut rounds,
                        packgen,
                        repo.record(),
                        &limits,
                        pipeline,
                    )
                    .await
                }
                other => Err(SubstrateError::Protocol(format!(
                    "unknown command {other:?}{UNKNOWN_COMMAND_DECLINE}"
                ))),
            },
            Service::ReceivePack => {
                // One POST body is one push invocation (ADR-013 §3): the
                // parse stopped at the last framed flush, the pack tail
                // escapes raw, the machine runs under the stateless tail
                // discipline, and the report rides the response — the
                // session ends after the response (no cross-request
                // state).
                let tail = post.take_raw_tail()?;
                serve_receive_pack(
                    post.response(),
                    &request,
                    tail,
                    push_options_negotiated,
                    refs,
                    ingest,
                    repo.record(),
                    &limits,
                    PackTailDiscipline::Stateless,
                    pipeline,
                )
                .await
            }
        },
    }
}

fn wall_clock_breach(limits: &Limits) -> SubstrateError {
    SubstrateError::Protocol(format!(
        "the request exceeded the session_wall_clock budget ({:?}) — server/session budget class, 503-mapped at the door (ADR-009)",
        limits.session_wall_clock
    ))
}

/// The door's `info/refs` GET emission (doors.md's advertisement route):
/// the **smart prefix** (`# service=git-{upload,receive}-pack`, its
/// flush) followed by the capability advertisement as a stateless
/// response ending at flush — the shape the client caches and re-sends
/// inside its first POST, and the door serves this entry without
/// touching pkt-line APIs. The Content-Type is the door's.
///
/// The V2 advertisement is per-request-set on the http substrate (the
/// client re-sends the dump), so the emission runs on every GET (no
/// once-per-session discipline applies here — that's the duplex
/// serving entry's). The capability dump itself is the static honest
/// list ([`advertise_v2`]).
///
/// The receive-pack advertisement is the same wrapper with the **V0
/// push ref advertisement** ([`advertise_v0`], ADR-013 §1–2) in place
/// of the V2 dump: the listed refs come from the backend's
/// [`list_refs`](crate::GitRefs::list_refs) — the emission's one
/// backend call, pure wire over it (`advertise_v0` holds no backend
/// logic). A listing failure surfaces as the typed session error,
/// never a partial ref list (review 001 N-1); the ref cap
/// ([`Limits::max_advertisement_refs`]) fails closed inside the
/// machine. The `push-options` gate flag (ADR-013 §11; the assembler
/// enables it, default off in v1) rides the emission additively —
/// opening the gate later is a value change.
///
/// The entry takes the [`AuthorizedRepo`] marker by type as its
/// authorization assertion — the door's ADR-007 check precedes any
/// response byte (ACL before advertisement): a private repo never
/// reaches either emission.
pub async fn serve_stateless_advertisement<W>(
    repo: &AuthorizedRepo,
    writer: W,
    service: Service,
    refs: Arc<dyn GitRefs>,
    limits: &Limits,
    push_options_negotiated: bool,
) -> Result<(), SubstrateError>
where
    W: FuturesAsyncWrite + Unpin + Send,
{
    match service {
        Service::UploadPack => {
            let mut writer = writer;
            encode::text_to_write(
                format!("# service={}", service.as_str()).as_bytes(),
                &mut writer,
            )
            .await?;
            encode::flush_to_write(&mut writer).await?;
            advertise_v2(&mut StatelessResponse::new(writer)).await
        }
        Service::ReceivePack => {
            // The listing is the emission's one backend call: bounded by
            // the same per-request wall clock the POST entry enforces —
            // without it the GET emission is the one unbounded session
            // surface (a hung backend never returns to the door's
            // route).
            let listed = match tokio::time::timeout(
                limits.session_wall_clock,
                refs.list_refs(repo.record()),
            )
            .await
            {
                Ok(Ok(listed)) => listed,
                Ok(Err(error)) => return Err(storage_to_session_error(error)),
                Err(_elapsed) => return Err(wall_clock_breach(limits)),
            };
            let mut writer = writer;
            encode::text_to_write(
                format!("# service={}", service.as_str()).as_bytes(),
                &mut writer,
            )
            .await?;
            encode::flush_to_write(&mut writer).await?;
            advertise_v0(
                &mut StatelessResponse::new(writer),
                listed,
                push_options_negotiated,
                limits,
            )
            .await
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use futures::io::AsyncReadExt as FuturesAsyncReadExt;
    use futures::stream;
    use std::sync::Mutex as StdMutex;

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

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

    /// A shared-buffer writer end (the door-side response sink's role).
    #[derive(Clone, Default)]
    struct TestSink(Arc<StdMutex<Vec<u8>>>);

    impl futures::AsyncWrite for TestSink {
        fn poll_write(
            self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            buf: &[u8],
        ) -> Poll<io::Result<usize>> {
            self.0.lock().unwrap().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<()>> {
            Poll::Ready(Ok(()))
        }
    }

    type BodyChunks = stream::Iter<std::vec::IntoIter<io::Result<Bytes>>>;
    type TestPost = StatelessRequest<BodyReader<BodyChunks, io::Error>, TestSink>;

    fn post_with(chunks: Vec<Vec<u8>>, limits: Limits) -> (TestPost, Arc<StdMutex<Vec<u8>>>) {
        let sink = TestSink::default();
        let items: Vec<io::Result<Bytes>> = chunks
            .into_iter()
            .map(|c| io::Result::Ok(Bytes::from(c)))
            .collect();
        let post = StatelessRequest::new(
            BodyReader::new(stream::iter(items)),
            sink.clone(),
            limits,
            Service::UploadPack,
        );
        (post, Arc::clone(&sink.0))
    }

    fn post(chunks: Vec<Vec<u8>>) -> (TestPost, Arc<StdMutex<Vec<u8>>>) {
        post_with(chunks, Limits::default())
    }

    /// The first POST's capability dump: the smart prefix section, its
    /// flush, the V2 capability advertisement section, its flush —
    /// then the command invocation.
    fn first_post_body(command: &[u8], pre: &[&[u8]], args: &[&[u8]]) -> 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)).collect::<Vec<u8>>(),
            DELIMITER.to_vec(),
            args.iter().flat_map(|l| pkt(l)).collect::<Vec<u8>>(),
            FLUSH.to_vec(),
        ]
        .concat()
    }

    /// Subsequent POSTs carry only the command invocation.
    fn command_post_body(command: &[u8], pre: &[&[u8]], args: &[&[u8]]) -> Vec<u8> {
        [
            pkt(command),
            pre.iter().flat_map(|l| pkt(l)).collect::<Vec<u8>>(),
            DELIMITER.to_vec(),
            args.iter().flat_map(|l| pkt(l)).collect::<Vec<u8>>(),
            FLUSH.to_vec(),
        ]
        .concat()
    }

    /// Every pkt-line payload of a framed byte stream (the flush as a
    /// trailing empty payload).
    fn parse_pkts(buffer: &[u8]) -> Vec<Vec<u8>> {
        let mut out = Vec::new();
        let mut i = 0;
        while i < buffer.len() {
            assert!(i + 4 <= buffer.len(), "trailing partial header");
            let len =
                usize::from_str_radix(std::str::from_utf8(&buffer[i..i + 4]).unwrap(), 16).unwrap();
            if len == 0 {
                out.push(Vec::new());
                i += 4;
            } else {
                assert!(len >= 4 && i + len <= buffer.len(), "partial pkt-line");
                out.push(buffer[i + 4..i + len].to_vec());
                i += len;
            }
        }
        out
    }

    #[tokio::test]
    async fn stateless_constructs_from_reader_writer_limits_service() {
        let (mut post, _) = post(vec![command_post_body(b"command=ls-refs\n", &[], &[])]);
        assert_eq!(post.service(), Service::UploadPack);
        assert_eq!(*post.limits(), Limits::default());
        assert_eq!(post.consumed_bytes(), 0);
        assert_eq!(post.transport_reads(), 0);
        assert!(!post.response().is_flushed());
    }

    /// The first POST's capability dump — both flush-delimited sections
    /// — is skipped; the command line and capabilities ride before the
    /// `0001` delimiter and the args after it. Parsing the pre-delim
    /// lines as args would shadow the real args on `fetch` (POC-3's
    /// recorded framing fact) — the split is delim-aware.
    #[tokio::test]
    async fn capability_dump_is_skipped_and_the_arg_split_is_delim_aware() {
        let (mut post, _) = post(vec![first_post_body(
            b"command=ls-refs\n",
            &[b"object-format=sha1"],
            &[b"peel", b"ref-prefix refs/heads/"],
        )]);

        let parsed = post.read_post().await.unwrap();
        let StatelessPost::Command(request) = parsed else {
            panic!("the first POST parses as a command: {parsed:?}")
        };
        assert_eq!(request.command, "ls-refs");
        assert_eq!(request.pre_lines, vec!["object-format=sha1"]);
        assert_eq!(request.args, vec!["peel", "ref-prefix refs/heads/"]);
        assert!(post.consumed_bytes() > 0);
    }

    /// Subsequent POSTs (negotiation rounds) carry only the command
    /// request; the same machinery parses them.
    #[tokio::test]
    async fn subsequent_posts_parse_the_command_directly() {
        let (mut post, _) = post(vec![command_post_body(
            b"command=fetch\n",
            &[b"agent=git/2.43"],
            &[b"have deadbeef", b"done"],
        )]);

        let parsed = post.read_post().await.unwrap();
        let StatelessPost::Command(request) = parsed else {
            panic!("a negotiation-round POST parses as a command: {parsed:?}")
        };
        assert_eq!(request.command, "fetch");
        assert_eq!(request.pre_lines, vec!["agent=git/2.43"]);
        assert_eq!(request.args, vec!["have deadbeef", "done"]);
    }

    /// The POST body arrives in dribbled chunks: the parse reads across
    /// chunk boundaries without buffering the body whole (POC-3's
    /// streaming measurement, encoded as a test).
    #[tokio::test]
    async fn the_body_across_dribbled_chunks_parses_streaming() {
        let body = first_post_body(b"command=ls-refs\n", &[], &[b"peel"]);
        let (mut post, _) = post(vec![
            body[..40].to_vec(),
            body[40..90].to_vec(),
            body[90..].to_vec(),
        ]);

        let parsed = post.read_post().await.unwrap();
        let StatelessPost::Command(request) = parsed else {
            panic!("chunked bodies parse like a whole one: {parsed:?}")
        };
        assert_eq!(request.command, "ls-refs");
        assert_eq!(request.args, vec!["peel"]);
    }

    /// remote-curl's `probe_rpc` POSTs a flush-only body before the
    /// real request on the large-request path: the machinery resolves
    /// it to the empty-OK probe shape, never an error (answering an
    /// error status breaks the whole clone).
    #[tokio::test]
    async fn flush_only_post_resolves_to_the_probe_shape() {
        let (mut post, _) = post(vec![FLUSH.to_vec()]);

        let parsed = post.read_post().await.unwrap();
        assert_eq!(parsed, StatelessPost::Probe);
        // The probe consumes the POST: one command invocation (or probe
        // shape) per POST.
        let error = post.read_post().await.unwrap_err();
        assert!(error
            .to_string()
            .contains("one command invocation per POST"));
    }

    /// A body-less POST resolves to the same empty-OK probe shape
    /// (proxies do drop bodies; the answer stays the safe one).
    #[tokio::test]
    async fn empty_post_body_resolves_to_the_probe_shape() {
        let (mut post, _) = post(vec![Vec::new()]);
        let parsed = post.read_post().await.unwrap();
        assert_eq!(parsed, StatelessPost::Probe);
    }

    /// The probe answer takes the response rules: written nothing, it
    /// is the empty body (the door maps Probe to the empty 200); with a
    /// flush it is the POC-3-verified flush-only form — one `0000`,
    /// exactly once (idempotent).
    #[tokio::test]
    async fn the_probe_answer_takes_the_response_rules() {
        let (mut post, buffer) = post(vec![FLUSH.to_vec()]);
        post.read_post().await.unwrap();

        post.response().write_flush().await.unwrap();
        post.response().write_flush().await.unwrap();
        assert_eq!(&*buffer.lock().unwrap(), FLUSH, "a bare flush, once");
        assert!(post.response().is_flushed());
    }

    /// A body that ends at a section boundary without ever carrying a
    /// `command=` line is a protocol failure (truncated dump), not a
    /// probe — a flush-only or empty body is the probe.
    #[tokio::test]
    async fn body_ending_without_a_command_line_is_a_protocol_error() {
        let (mut post, _) = post(vec![[
            pkt(b"# service=git-upload-pack\n"),
            FLUSH.to_vec(),
            pkt(b"version 2\n"),
            FLUSH.to_vec(),
        ]
        .concat()]);
        let error = post.read_post().await.unwrap_err();
        assert!(
            error.is_terminal(),
            "a truncated dump is terminal: {error:?}"
        );
        match error {
            SubstrateError::Protocol(message) => {
                assert!(message.contains("command"), "{message}");
            }
            other => panic!("a command-less body is out of grammar: {other:?}"),
        }
        assert!(post.read_post().await.is_err(), "terminal stays terminal");
    }

    /// The unconditional error-break rule: a body truncated mid-section
    /// (the transport ends before the section's flush) is a terminal io
    /// error; every later call re-reports the cached error without
    /// re-entering the transport (read-call counter frozen) — no
    /// reset-and-retry loop is reachable (POC-3's spin bug class).
    #[tokio::test]
    async fn truncated_bodies_break_unconditionally() {
        let full = command_post_body(b"command=ls-refs\n", &[], &[b"peel"]);
        let (mut post, _) = post(vec![full[..full.len() - 6].to_vec()]);

        let error = post.read_post().await.unwrap_err();
        let SubstrateError::Io(e) = &error else {
            panic!("a mid-section truncation is terminal io: {error:?}");
        };
        assert_eq!(e.kind(), io::ErrorKind::UnexpectedEof);
        assert!(error.is_terminal());

        let reads_frozen = post.transport_reads();
        for _ in 0..5 {
            assert!(post.read_post().await.is_err(), "terminal stays terminal");
        }
        assert_eq!(
            post.transport_reads(),
            reads_frozen,
            "no retry may re-enter the transport"
        );
    }

    #[tokio::test]
    async fn malformed_pkt_lines_are_protocol_errors() {
        let (mut post, _) = post(vec![b"zzzz".to_vec()]);
        let error = post.read_post().await.unwrap_err();
        assert!(
            error.is_terminal(),
            "decode failures are terminal: {error:?}"
        );
        match error {
            SubstrateError::Protocol(message) => {
                assert!(message.contains("malformed pkt-line"), "{message}");
            }
            other => panic!("decode errors are protocol-shaped: {other:?}"),
        }
    }

    #[tokio::test]
    async fn an_err_line_in_the_body_is_a_protocol_error() {
        let (mut post, _) = post(vec![pkt(b"ERR malformed client")]);
        let error = post.read_post().await.unwrap_err();
        match error {
            SubstrateError::Protocol(message) => {
                assert!(message.contains("ERR"), "{message}");
                assert!(message.contains("malformed client"), "{message}");
            }
            other => panic!("client ERR lines are protocol-shaped: {other:?}"),
        }
    }

    /// `0002` never appears in request bodies; one appearing is a
    /// protocol failure.
    #[tokio::test]
    async fn a_response_end_packet_in_the_body_is_a_protocol_error() {
        let (mut post, _) = post(vec![b"0002".to_vec()]);
        let error = post.read_post().await.unwrap_err();
        match error {
            SubstrateError::Protocol(message) => {
                assert!(message.contains("response-end"), "{message}");
            }
            other => panic!("0002 in a POST body is out of grammar: {other:?}"),
        }
    }

    #[tokio::test]
    async fn repeated_delimiter_is_a_protocol_error() {
        let (mut post, _) = post(vec![[
            pkt(b"command=ls-refs\n"),
            DELIMITER.to_vec(),
            DELIMITER.to_vec(),
            pkt(b"peel\n"),
            FLUSH.to_vec(),
        ]
        .concat()]);
        let error = post.read_post().await.unwrap_err();
        match error {
            SubstrateError::Protocol(message) => {
                assert!(message.contains("repeated"), "{message}");
            }
            other => panic!("a repeated delimiter is out of grammar: {other:?}"),
        }
    }

    #[tokio::test]
    async fn a_second_command_line_is_a_protocol_error() {
        let (mut post, _) = post(vec![command_post_body(
            b"command=ls-refs\n",
            &[b"command=fetch"],
            &[],
        )]);
        let error = post.read_post().await.unwrap_err();
        match error {
            SubstrateError::Protocol(message) => {
                assert!(message.contains("second command"), "{message}");
            }
            other => panic!("a second command line is out of grammar: {other:?}"),
        }
    }

    /// The request-body budget (ADR-009, client-fault) fails closed: a
    /// body exceeding `max_request_body` is a protocol failure mapped
    /// to 413 at the door.
    #[tokio::test]
    async fn the_request_body_budget_fails_closed() {
        let body = first_post_body(b"command=ls-refs\n", &[], &[b"peel"]);
        let (mut post, _) = post_with(
            vec![body],
            Limits {
                max_request_body: 10,
                ..Limits::default()
            },
        );
        let error = post.read_post().await.unwrap_err();
        assert!(
            error.is_terminal(),
            "a budget breach fails closed: {error:?}"
        );
        match error {
            SubstrateError::Protocol(message) => {
                assert!(message.contains("max_request_body"), "{message}");
            }
            other => panic!("a budget breach is protocol-shaped: {other:?}"),
        }
    }

    /// The `BodyReader` adapter serves chunk-at-a-time: reads drain the
    /// current chunk before pulling the next, stream failures surface
    /// as io errors, and EOF reads as clean end-of-input.
    #[tokio::test]
    async fn body_reader_serves_chunks_as_reads() {
        let stream = stream::iter([
            Ok::<Bytes, io::Error>(Bytes::from_static(b"hello ")),
            Ok(Bytes::from_static(b"world")),
            Err(io::Error::other("body failed")),
        ]);
        let mut reader = BodyReader::new(stream);

        let mut got = vec![0u8; 3];
        reader.read_exact(&mut got).await.unwrap();
        assert_eq!(got, b"hel".to_vec());
        let n = reader.read(&mut got).await.unwrap();
        assert_eq!(&got[..n], b"lo ", "short reads stop at the cursor");
        let mut rest = Vec::new();
        reader
            .read_to_end(&mut rest)
            .await
            .expect_err("the failing chunk surfaces as an io error");
    }

    #[tokio::test]
    async fn body_reader_reads_to_clean_eof() {
        let stream = stream::iter([Ok::<Bytes, io::Error>(Bytes::from_static(b"abcd"))]);
        let mut reader = BodyReader::new(stream);
        let mut got = Vec::new();
        reader.read_to_end(&mut got).await.unwrap();
        assert_eq!(got, b"abcd".to_vec());
    }

    /// Responses end at flush and never carry `0002`: the written bytes
    /// are a text pkt-line, band-1 chunks, and exactly one `0000`, with
    /// no `0002` anywhere.
    #[tokio::test]
    async fn responses_end_at_flush_and_never_0002() {
        let (mut post, buffer) = post(vec![command_post_body(b"command=fetch\n", &[], &[b"done"])]);
        post.read_post().await.unwrap();

        post.response().write_text_line(b"packfile").await.unwrap();
        post.response()
            .write_data_band(&vec![b'x'; SIDEBAND_CHUNK_SIZE + 10])
            .await
            .unwrap();
        post.response().write_flush().await.unwrap();

        let body = buffer.lock().unwrap().clone();
        let pkts = parse_pkts(&body);
        assert_eq!(pkts[0], b"packfile\n".to_vec());
        assert_eq!(pkts[1][0], 1, "band 1");
        assert_eq!(pkts[1].len() - 1, SIDEBAND_CHUNK_SIZE);
        assert_eq!(pkts[2][0], 1, "band 1");
        assert_eq!(pkts[2].len() - 1, 10);
        assert!(
            pkts.last().unwrap().is_empty(),
            "the response ends at flush"
        );
        assert!(
            !body.windows(4).any(|w| w == b"0002"),
            "no 0002 anywhere in a response body"
        );
    }

    /// The flush is one-shot per response (the idempotent-flush guard
    /// is mandatory — the POC-3 recorded bug is the second flush): a
    /// second write_flush emits nothing, and data after the flush is
    /// refused.
    #[tokio::test]
    async fn the_response_flush_is_one_shot_and_data_after_it_is_refused() {
        let (mut post, buffer) = post(vec![command_post_body(b"command=ls-refs\n", &[], &[])]);
        post.read_post().await.unwrap();

        post.response().write_text_line(b"version 2").await.unwrap();
        post.response().write_flush().await.unwrap();
        assert!(post.response().is_flushed());

        let snapshot = buffer.lock().unwrap().clone();
        post.response().write_flush().await.unwrap();
        assert_eq!(
            &*buffer.lock().unwrap(),
            &snapshot,
            "the second flush is a no-op"
        );

        let error = post.response().write_text_line(b"after").await.unwrap_err();
        match error {
            SubstrateError::Protocol(message) => {
                assert!(message.contains("one-shot"), "{message}");
            }
            other => panic!("data after the flush is refused: {other:?}"),
        }
        let error = post.response().write_data_band(b"after").await.unwrap_err();
        match error {
            SubstrateError::Protocol(_) => {}
            other => panic!("band data after the flush is refused: {other:?}"),
        }
    }

    /// Data-band writes chunk at the fixed 65000 cap
    /// ([`SIDEBAND_CHUNK_SIZE`]): one write call, n band-1 pkt-lines.
    #[tokio::test]
    async fn data_band_writes_chunk_at_the_cap() {
        let (mut post, buffer) = post(vec![command_post_body(b"command=fetch\n", &[], &[b"done"])]);
        post.read_post().await.unwrap();

        let payload = vec![b'x'; SIDEBAND_CHUNK_SIZE * 2 + 20000];
        post.response().write_data_band(&payload).await.unwrap();
        post.response().write_flush().await.unwrap();

        let body = buffer.lock().unwrap().clone();
        let pkts = parse_pkts(&body);
        let sizes: Vec<usize> = pkts.iter().map(|p| p.len()).collect();
        assert_eq!(
            sizes,
            vec![1 + 65000, 1 + 65000, 1 + 20000, 0],
            "band-1 chunk sizes then the flush"
        );
        assert!(
            pkts.iter().all(|p| p.is_empty() || p[0] == 1),
            "band 1 only"
        );
    }

    /// The machinery is generic over both ends: a tokio read end goes
    /// through compat (the door's channel/body stream shape), a futures
    /// write end plugs in directly — no http type anywhere below the
    /// stream.
    #[tokio::test]
    async fn the_machinery_runs_over_arbitrary_reader_writer_ends() {
        use tokio::io::AsyncWriteExt as TokioAsyncWriteExt;

        let (mut peer, stream) = tokio::io::duplex(1 << 15);
        TokioAsyncWriteExt::write_all(
            &mut peer,
            &first_post_body(b"command=ls-refs\n", &[], &[b"peel"]),
        )
        .await
        .unwrap();

        let sink = TestSink::default();
        let mut post = StatelessRequest::new(
            tokio_util::compat::TokioAsyncReadCompatExt::compat(stream),
            sink.clone(),
            Limits::default(),
            Service::UploadPack,
        );
        let parsed = post.read_post().await.unwrap();
        let StatelessPost::Command(request) = parsed else {
            panic!("tokio-compat ends parse identically")
        };
        assert_eq!(request.command, "ls-refs");
        assert_eq!(request.args, vec!["peel"]);
        post.response().write_flush().await.unwrap();
        assert_eq!(&*sink.0.lock().unwrap(), FLUSH);
        drop(peer);
    }

    /// The stateless raw-tail escape (ADR-013 §3) — the POST body's
    /// pack tail continues byte-exactly from the command invocation's
    /// terminating flush (dribbled chunks split inside the flush and
    /// inside the tail), and the body-budget consumed counter
    /// reconciles: framed bytes + tail bytes = the body's length.
    #[tokio::test]
    async fn raw_tail_escapes_after_the_command_flush_without_byte_loss() {
        let tail: Vec<u8> = (0..6000u32).map(|i| (i % 251) as u8).collect();
        let command = command_post_body(b"command=ls-refs\n", &[], &[b"peel"]);
        let body = [&command[..], &tail].concat();
        let command_len = command.len();

        let (mut post, _) = post_with(
            vec![
                body[..12].to_vec(),
                body[12..command_len - 2].to_vec(),
                body[command_len - 2..command_len + 2].to_vec(),
                body[command_len + 2..command_len + 40].to_vec(),
                body[command_len + 40..].to_vec(),
            ],
            Limits::default(),
        );

        let parsed = post.read_post().await.unwrap();
        let StatelessPost::Command(request) = parsed else {
            panic!("the command invocation parses: {parsed:?}")
        };
        assert_eq!(request.command, "ls-refs");
        assert_eq!(
            post.consumed_bytes(),
            command_len as u64,
            "the parse consumed exactly the framed body"
        );

        let mut tail_reader = post.take_raw_tail().unwrap();
        let mut got = Vec::new();
        tail_reader.read_to_end(&mut got).await.unwrap();
        assert_eq!(got, tail, "the tail continues byte-exactly from the flush");
        assert_eq!(
            tail_reader.consumed_bytes(),
            body.len() as u64,
            "the tail rides the body counter: framed + tail"
        );
        assert_eq!(
            tail_reader.consumed_bytes(),
            post.consumed_bytes(),
            "one shared counter, reconciled"
        );
    }

    /// The escape is machinery misuse without a command-shaped parse —
    /// before any read_post, and after the probe shape (a probe body
    /// has no tail) — and the refusal leaves the reader in place.
    #[tokio::test]
    async fn the_stateless_tail_escape_refuses_without_a_command_parse() {
        let (mut post, _) = post(vec![FLUSH.to_vec()]);
        let error = post.take_raw_tail().unwrap_err();
        match error {
            SubstrateError::RawTailUnavailable(message) => {
                assert!(message.contains("command"), "{message}");
            }
            other => panic!("the escape before a parse is misuse: {other:?}"),
        }

        // Refused, not destroyed: the reader stays usable — the probe
        // parses after the refusal.
        let parsed = post.read_post().await.unwrap();
        assert_eq!(parsed, StatelessPost::Probe);

        // The probe shape has no tail either — the escape stays refused,
        // and the reader is inert after the parse ("already parsed").
        let error = post.take_raw_tail().unwrap_err();
        assert!(
            matches!(error, SubstrateError::RawTailUnavailable(_)),
            "a probe body surrenders no tail: {error:?}"
        );
        assert!(post.read_post().await.is_err());
    }

    /// The escape is one-shot; after it the pkt-line reader is gone and
    /// every later read_post stays refused — never a silent re-entry.
    #[tokio::test]
    async fn the_stateless_tail_escape_is_one_shot_and_the_reader_is_gone() {
        let (mut post, _) = post(vec![command_post_body(b"command=ls-refs\n", &[], &[])]);
        let StatelessPost::Command(_) = post.read_post().await.unwrap() else {
            panic!("the command invocation parses")
        };

        let _tail = post.take_raw_tail().unwrap();
        let error = post.take_raw_tail().unwrap_err();
        match error {
            SubstrateError::RawTailUnavailable(message) => {
                assert!(message.contains("one-shot"), "{message}");
            }
            other => panic!("a second escape is misuse: {other:?}"),
        }
        let error = post.read_post().await.unwrap_err();
        match error {
            SubstrateError::Protocol(message) => {
                assert!(
                    message.contains("one command invocation per POST"),
                    "{message}"
                );
            }
            other => panic!("post-escape reads are refused: {other:?}"),
        }
    }

    /// The escape after a terminal error reports the cached terminal —
    /// the escape dies with the session.
    #[tokio::test]
    async fn the_stateless_tail_escape_refuses_after_a_terminal_error() {
        let full = command_post_body(b"command=ls-refs\n", &[], &[b"peel"]);
        let (mut post, _) = post(vec![full[..full.len() - 6].to_vec()]);

        let terminal = post.read_post().await.unwrap_err();
        assert!(terminal.is_terminal());
        let error = post.take_raw_tail().unwrap_err();
        match (terminal, error) {
            (SubstrateError::Io(first), SubstrateError::Io(second)) => {
                assert_eq!(first.kind(), io::ErrorKind::UnexpectedEof);
                assert_eq!(second.kind(), io::ErrorKind::UnexpectedEof);
            }
            other => panic!("the terminal error stays terminal: {other:?}"),
        }
    }
}