alkgit 0.1.0

//! The duplex session substrate (ADR-005 §1): the framing layer every
//! duplex-path machine runs under.
//!
//! [`Session`] wraps (stream, [`Limits`], [`Service`](crate::Service))
//! into the pkt-line session — the pkt-line session machinery shapes
//! the transport refuses to run without (ADR-009, the session tuple of
//! ADR-002/ADR-016). It owns:
//!
//! - **the split/compat bridge** — the tokio stream halves bridged into
//!   gix-packetline's futures-io codec via tokio-util compat (POC-1's
//!   validated shape; no hand-written adapter).
//! - **the request reader** — the V2 command grammar: the `command=`
//!   line (plus any additional pre-delim lines), the `0001` delimiter,
//!   the argument lines, and the terminating flush (POC-1 §2). The
//!   reader implements the ADR-005 §1 request shape; receive-pack's V0
//!   command-less request is
//!   [`Session::read_push_request`](crate::Session::read_push_request)
//!   (ADR-013 §3, one request per stream — the pack tail then escapes
//!   raw).
//! - **the raw-tail escape** — after a request's terminating flush,
//!   [`Session::take_raw_tail`] surrenders the read side's remaining
//!   bytes as a raw async reader ([`RawTail`]): on the push path the
//!   pack follows the command flush **unframed** (ADR-013 §3, as
//!   amended by ADR-020 §1), so the pkt-line machinery bows out with
//!   zero byte loss — the demand-driven iterator read exactly the
//!   flushed request's bytes (read-exact framing, never a peek past
//!   the stop) — and the consumed-bytes counter rides the tail for
//!   budget observability. Every later pkt-line read after the
//!   surrender is refused ([`SubstrateError::RawTailUnavailable`], the
//!   machinery-misuse taxonomy); the session ends after the push
//!   report.
//! - **the writer** — text/error-band/flush framing, the band-1 data
//!   framing the machines' packfile-section relay uses
//!   ([`Session::write_data_band`]), plus the sideband sink
//!   ([`Sideband`]), the blocking-side `std::io::Write` end with
//!   65000-byte band-1 chunks ([`SIDEBAND_CHUNK_SIZE`], ADR-004/009),
//!   idempotent one-shot-per-response flush, and exactly one driver
//!   task per sink, prevented by construction: the sink takes the
//!   session's response writer out, and the drain task parks on a
//!   bounded channel ([`SIDEBAND_CHANNEL_CAPACITY`], ADR-005 §1's
//!   bounded-sink rule) — so no second sink can be constructed and no
//!   unbounded buffer can grow.
//!
//! Two failure modes are structurally prevented (the recorded POC hang
//! and spin bug classes):
//!
//! 1. **The stop-delimiter discipline.** The request reader stops at
//!    the client's terminating flush and the packetline iterator is
//!    inert until reset — forgetting the reset silently swallows the
//!    next command (POC-1's second hang). The substrate encodes that:
//!    after a request completes, [`Session::read_request`] refuses with
//!    [`SubstrateError::ResetRequired`] until
//!    [`Session::reset_request`] re-arms it — the next command can be
//!    heard only after the explicit reset, never swallowed.
//! 2. **The unconditional error-break rule.** An error-shaped read
//!    breaks the session's read side unconditionally: io errors and
//!    decode errors are terminal ([`SubstrateError::is_terminal`]), the
//!    transport is never re-entered (re-reads return the cached
//!    terminal error), and no reset-and-retry is reachable — retrying
//!    an error-shaped read spins a worker at 100% (POC-3's recorded bug
//!    class).
//!
//! Raw packetline types (`PacketLineRef`, `decode::Error`, the three
//! `Option<io::Result<Result<…>>>` shape) never surface past this
//! module: the reader normalizes peer lines into owned strings and the
//! taxonomy of [`SubstrateError`] (ADR-005 §4).
//!
//! Front-door-blind (convention 15): the stream is a plain
//! `tokio::io::AsyncRead + AsyncWrite` — doors hand it their
//! channel-specific type; no channel/http types appear here. Tests run
//! over in-memory `tokio::io::duplex` pairs — no backend (wave-1
//! scope).
//!
//! The serving entry lands here too (wave 2, the duplex half of
//! ADR-002's session boundary): [`serve_duplex`] turns
//! ([`AuthorizedRepo`], [`Service`], stream, [`Limits`], backend
//! handles) into the running session — advertisement-once, then the
//! command loop (fetch) or the one-request push (receive-pack, wave 3:
//! the V0 ref advertisement server-speaks-first, one V0 request, the
//! push machine under the duplex tail discipline, the session ends
//! after the report), the whole thing under [`Limits::session_wall_clock`]
//! with the ADR-005 §4 error mapping; [`serve_duplex_resolved`] is the
//! ADR-007 front over [`GitRegistry`](crate::GitRegistry) whose
//! resolve→authorize gate collapses to the generic denial before any
//! protocol byte (ADR-008/N-2).

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

use futures::executor::block_on;
use futures::io::AsyncRead as FuturesAsyncRead;
use futures::io::AsyncWriteExt as FuturesAsyncWriteExt;
use gix_packetline::async_io::{encode, StreamingPeekableIter};
use gix_packetline::{Channel, PacketLineRef};
use tokio::io::{AsyncRead as TokioAsyncRead, AsyncWrite as TokioAsyncWrite, ReadHalf, WriteHalf};
use tokio::sync::{mpsc, oneshot};
use tokio_util::compat::{Compat, TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};

use crate::authorize::AuthorizedRepo;
use crate::backend::{GitPackGen, GitPackIngest, GitRefs, GitRegistry};
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::registry::AccessAction;
use crate::substrate::{Service, SubstrateError};
use crate::v0::advertise_v0;
use crate::v2::{advertise_v2, parse_ls_refs_args, serve_ls_refs};

use alkcall::core::Identity;
use tokio::sync::Semaphore;

/// The stop-delimiter configuration of both substrates' request
/// readers (POC-1 §3.1): the client's terminating flush stops the
/// iteration, not a yielded line — one home serving the duplex reader
/// and the stateless reader alike (review 005 R-3's const dedup).
pub(crate) const STOP_AT_FLUSH: &[PacketLineRef<'static>] = &[PacketLineRef::Flush];

/// The unknown-command decline's text (ADR-003's honest surface), one
/// home for both substrates' dispatch sites (review 005 R-3's const
/// dedup) — the dispatch's own `{other:?}` command name is the prefix.
pub(crate) const UNKNOWN_COMMAND_DECLINE: &str =
    " — declined, only the served commands dispatch (ADR-003's honest surface)";

/// Bounded capacity of the sideband sink's channel (ADR-005 §1: the
/// sink parks on a bounded channel — no unbounded buffer). Small by
/// construction: the driver parks when the drain falls behind, which is
/// the intended backpressure.
pub const SIDEBAND_CHANNEL_CAPACITY: usize = 4;

type SessionReader<T> = StreamingPeekableIter<CountingRead<Compat<ReadHalf<T>>>>;
type SessionWriter<T> = Compat<WriteHalf<T>>;

/// Counts transport reads so terminal reads can prove they never
/// re-enter the stream, and so consumption is observable (the
/// bounded-resources enforcement later waves' machines build on).
#[derive(Debug)]
pub(crate) struct CountingRead<T> {
    pub(crate) inner: T,
    pub(crate) consumed: Arc<AtomicU64>,
    pub(crate) reads: Arc<AtomicU64>,
}

impl<T> futures::io::AsyncRead for CountingRead<T>
where
    T: futures::io::AsyncRead + Unpin,
{
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut [u8],
    ) -> Poll<io::Result<usize>> {
        let this = self.get_mut();
        this.reads.fetch_add(1, Ordering::Relaxed);
        match Pin::new(&mut this.inner).poll_read(cx, buf) {
            Poll::Ready(Ok(n)) => {
                this.consumed.fetch_add(n as u64, Ordering::Relaxed);
                Poll::Ready(Ok(n))
            }
            other => other,
        }
    }
}

/// The raw read side the pkt-line machinery surrenders to the push
/// machine (the raw-tail escape, ADR-013 §3 as amended by ADR-020 §1):
/// an async reader over the bytes remaining after the request's
/// terminating flush — on the duplex substrate the transport's read
/// half past the flush, on the stateless substrate the POST body's
/// remaining bytes. Reads count into the substrate's consumed-bytes
/// counter, so the tail's consumption reconciles against the session's
/// (the request-side counter) — the budget observability the
/// wire-side pack-size relay rides
/// ([`crate::pack_source::bridge_pack_source`]).
///
/// Contract stamp: ADR-013 §3 (the pack is raw, to end of body/stream),
/// ADR-020 §1 (the bounded peek watches this tail — the escape is its
/// precondition), ADR-005 §1 (the flush boundary the tail continues
/// from), ADR-009 (the budget observability).
pub struct RawTail<R> {
    read: R,
    consumed: Arc<AtomicU64>,
}

impl<R> RawTail<R> {
    /// Composes a tail over (raw reader, shared consumed-bytes
    /// counter) — the substrates' escape constructor
    /// ([`Session::take_raw_tail`] /
    /// [`StatelessRequest::take_raw_tail`](crate::StatelessRequest::take_raw_tail));
    /// pub for the same pre-publish posture as the serving entries (the
    /// machine-level test compositions ride it too — one wire shape,
    /// one constructor).
    pub fn new(read: R, consumed: Arc<AtomicU64>) -> Self {
        RawTail { read, consumed }
    }

    /// The substrate's consumed-bytes counter (the read side's total —
    /// framed request bytes *plus* tail bytes, one shared counter for
    /// budget observability; reconciling is tail + framed = this).
    pub fn consumed_bytes(&self) -> u64 {
        self.consumed.load(Ordering::Relaxed)
    }
}

impl<R> std::fmt::Debug for RawTail<R> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RawTail")
            .field("consumed_bytes", &self.consumed_bytes())
            .finish_non_exhaustive()
    }
}

impl<R> FuturesAsyncRead for RawTail<R>
where
    R: FuturesAsyncRead + Unpin,
{
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut [u8],
    ) -> Poll<io::Result<usize>> {
        let this = self.get_mut();
        match Pin::new(&mut this.read).poll_read(cx, buf) {
            Poll::Ready(Ok(n)) => {
                this.consumed.fetch_add(n as u64, Ordering::Relaxed);
                Poll::Ready(Ok(n))
            }
            other => other,
        }
    }
}

/// One parsed V2 command request (ADR-005 §1's request reader shape):
/// the `command=` line, any additional pre-delim lines, the argument
/// lines after the `0001` delimiter, terminated by the client's flush.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Request {
    /// The request's command value — the text after `command=` on the
    /// first pre-delim line (e.g. `ls-refs`).
    pub command: String,
    /// Pre-delim data lines other than the command line, in order.
    pub pre_lines: Vec<String>,
    /// Data lines after the `0001` delimiter, in order.
    pub args: Vec<String>,
}

/// The duplex session substrate (ADR-005 §1).
///
/// Constructed from (stream, [`Limits`], [`Service`]) — the service
/// dimension rides the inputs per ADR-016 (the session tuple's
/// selector); *using* it to select the advertisement and state machine
/// at session start is wave 2. Exposes the request-reader machinery
/// ([`Session::read_request`] / [`Session::reset_request`]), the
/// raw-tail escape ([`Session::take_raw_tail`], the push path's
/// pack-tail handoff), and the writer machinery
/// ([`Session::write_text_line`], [`Session::write_flush`],
/// [`Session::write_error_band`], and the sideband sink).
///
/// The transport-level error taxonomy is [`SubstrateError`]: terminal
/// errors ([`SubstrateError::Io`], [`SubstrateError::Protocol`]) end
/// the session — the read side stays inert afterward, re-reporting the
/// cached terminal error without re-entering the transport.
pub struct Session<T> {
    limits: Limits,
    service: Service,
    reader: Option<SessionReader<T>>,
    writer: Option<SessionWriter<T>>,
    consumed: Arc<AtomicU64>,
    reads: Arc<AtomicU64>,
    stopped: bool,
    terminal: Option<SubstrateError>,
}

impl<T> Session<T>
where
    T: TokioAsyncRead + TokioAsyncWrite + Unpin,
{
    /// Wraps (stream, limits, service) into the pkt-line session
    /// (ADR-005 §1, the session tuple's substrate inputs per ADR-016).
    pub fn new(stream: T, limits: Limits, service: Service) -> Self {
        let (read_half, write_half) = tokio::io::split(stream);
        let consumed = Arc::new(AtomicU64::new(0));
        let reads = Arc::new(AtomicU64::new(0));
        Session {
            limits,
            service,
            reader: Some(StreamingPeekableIter::new(
                CountingRead {
                    inner: read_half.compat(),
                    consumed: Arc::clone(&consumed),
                    reads: Arc::clone(&reads),
                },
                STOP_AT_FLUSH,
                false,
            )),
            writer: Some(write_half.compat_write()),
            consumed,
            reads,
            stopped: false,
            terminal: None,
        }
    }

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

    /// The session's budget bundle (ADR-009; enforcement is the later
    /// waves' machines, the substrate carries the budgets into reach).
    pub fn limits(&self) -> &Limits {
        &self.limits
    }

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

    /// Reads the next V2 command request (ADR-005 §1): the `command=`
    /// line, any additional pre-delim lines, the `0001` delimiter, the
    /// argument lines, and the terminating flush.
    ///
    /// - `Ok(Some(request))` — a complete request was parsed.
    /// - `Ok(None)` — the session is over before any request byte:
    ///   either the clean close (the peer closed the stream) or the
    ///   **flush-only request at the command position** — the client's
    ///   trailing flush before its close (real clients send `0000`
    ///   after the last round's response; upstream's serve.c ends the
    ///   session on it). Subsequent calls report the cached terminal
    ///   error.
    /// - `Err(SubstrateError::ResetRequired)` — the previous request
    ///   completed (its flush was seen) and
    ///   [`Session::reset_request`] has not re-armed the reader; the
    ///   next command is *refused* rather than silently swallowed
    ///   (POC-1's second hang, structurally prevented).
    /// - `Err(SubstrateError::Io | SubstrateError::Protocol)` — terminal
    ///   ([`SubstrateError::is_terminal`]): the read side ends, and all
    ///   later calls re-report the cached error without touching the
    ///   transport (the unconditional error-break rule, POC-3's spin
    ///   bug class).
    ///
    /// After [`Session::take_raw_tail`] surrendered the reader, every
    /// call is machinery misuse
    /// ([`SubstrateError::RawTailUnavailable`], never a wire shape): the
    /// pack flows raw on the tail, the session ends after the report.
    ///
    /// A protocol violation terminates (`SubstrateError::Protocol`):
    /// malformed pkt-lines, client `ERR` lines, unexpected stop
    /// packets, repeated delimiters or command lines, and a request
    /// that carried input but no `command=` line. The machine maps it
    /// to a pkt-line error band
    /// via [`Session::write_error_band`] before ending the session.
    ///
    /// The per-request budget (`Limits::max_request_body`, ADR-009's
    /// stateless counterpart enforced by
    /// [`StatelessRequest::read_post`](crate::StatelessRequest::read_post))
    /// bounds the accumulated request bytes here too: a request whose
    /// transport consumption exceeds the budget since the request began
    /// is terminal (fail-closed, the client-fault class) — an
    /// authenticated duplex client may otherwise fill the accumulated
    /// `pre_lines`/`args` for the whole session wall clock.
    pub async fn read_request(&mut self) -> Result<Option<Request>, SubstrateError> {
        let budget = self.limits.max_request_body;
        // Field-split borrows: the reader may stay borrowed across the
        // awaits while the state fields are updated freely.
        let Self {
            reader,
            consumed,
            stopped,
            terminal,
            ..
        } = self;
        if let Some(cached) = terminal.clone() {
            return Err(cached);
        }
        let Some(reader) = reader else {
            return Err(SubstrateError::RawTailUnavailable(
                "the pkt-line reader is gone — the raw tail was surrendered; push reads no more commands",
            ));
        };
        if *stopped {
            return Err(SubstrateError::ResetRequired);
        }
        let consumed_before = consumed.load(Ordering::Relaxed);

        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;

        loop {
            // Classified into owned payloads so the packetline buffer
            // borrow ends before any state mutation.
            enum Payload {
                Text(String),
                Delimiter,
                StopPacket(String),
                ErrLine(String),
            }
            let payload: Option<Payload> = match reader.read_line().await {
                None => {
                    // The client's terminating flush stopped the
                    // iteration — end of request.
                    None
                }
                Some(Ok(Ok(line))) => Some(match line {
                    PacketLineRef::Data(d) => {
                        if let Some(message) = line
                            .check_error()
                            .map(|e| String::from_utf8_lossy(e.0).into_owned())
                        {
                            Payload::ErrLine(message)
                        } else {
                            let text = String::from_utf8_lossy(d)
                                .trim_end_matches(['\n', '\r'])
                                .to_owned();
                            Payload::Text(text)
                        }
                    }
                    PacketLineRef::Delimiter => Payload::Delimiter,
                    PacketLineRef::Flush | PacketLineRef::ResponseEnd => {
                        Payload::StopPacket("unexpected stop packet mid-request".to_owned())
                    }
                }),
                Some(Ok(Err(decode_error))) => {
                    let error = SubstrateError::Protocol(format!(
                        "malformed pkt-line from peer: {decode_error}"
                    ));
                    *terminal = Some(error.clone());
                    return Err(error);
                }
                Some(Err(io_error)) => {
                    let consumed_total = consumed.load(Ordering::Relaxed);
                    let clean_close = io_error.kind() == io::ErrorKind::UnexpectedEof
                        && consumed_total == consumed_before
                        && command.is_none()
                        && pre_lines.is_empty()
                        && args.is_empty()
                        && !delim_seen;
                    if clean_close {
                        *terminal = Some(SubstrateError::Io(io::Error::new(
                            io::ErrorKind::UnexpectedEof,
                            "the peer closed the stream",
                        )));
                        return Ok(None);
                    }
                    let error = SubstrateError::Io(io_error);
                    *terminal = Some(error.clone());
                    return Err(error);
                }
            };

            match payload {
                None => break,
                Some(Payload::ErrLine(message)) => {
                    let error =
                        SubstrateError::Protocol(format!("peer sent an ERR line: {message}"));
                    *terminal = Some(error.clone());
                    return Err(error);
                }
                Some(Payload::StopPacket(message)) => {
                    let error = SubstrateError::Protocol(message);
                    *terminal = Some(error.clone());
                    return Err(error);
                }
                Some(Payload::Delimiter) => {
                    if delim_seen {
                        let error = SubstrateError::Protocol("repeated 0001 delimiter".to_owned());
                        *terminal = Some(error.clone());
                        return Err(error);
                    }
                    delim_seen = true;
                }
                Some(Payload::Text(text)) => {
                    if delim_seen {
                        args.push(text);
                    } else if let Some(value) = text.strip_prefix("command=") {
                        if command.is_some() {
                            let error = SubstrateError::Protocol(
                                "request 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);
                    }
                }
            }

            let consumed_total = consumed.load(Ordering::Relaxed);
            if consumed_total.saturating_sub(consumed_before) > budget {
                let error = SubstrateError::Protocol(format!(
                    "the request exceeds the max_request_body budget ({budget}) — fail closed (ADR-009)"
                ));
                *terminal = Some(error.clone());
                return Err(error);
            }
        }

        let Some(command) = command else {
            if pre_lines.is_empty() && args.is_empty() && !delim_seen {
                // The flush-only request at the command position: the
                // client's trailing flush before its close (real
                // clients send `0000` after the last round's response) —
                // the session end, not a fault (upstream serve.c's
                // process_request; the clean-close read of the flush).
                *terminal = Some(SubstrateError::Io(io::Error::new(
                    io::ErrorKind::UnexpectedEof,
                    "the peer closed the stream",
                )));
                return Ok(None);
            }
            let error = SubstrateError::Protocol("request contained no command= line".to_owned());
            *terminal = Some(error.clone());
            return Err(error);
        };
        // The flush was observed: the reader is inert until reset. Record
        // it before returning so no call can silently skip the discipline.
        *stopped = true;
        Ok(Some(Request {
            command,
            pre_lines,
            args,
        }))
    }

    /// Re-arms the request reader after a completed request (the
    /// stop-delimiter discipline, ADR-005 §1). Required after every
    /// [`Session::read_request`] that returned a request; inert while
    /// already re-armed. Never revives a terminated session: terminal
    /// errors stay terminal. Inert after
    /// [`Session::take_raw_tail`] — the reader is gone, and every later
    /// pkt-line read refuses
    /// ([`SubstrateError::RawTailUnavailable`]).
    pub fn reset_request(&mut self) {
        if let Some(reader) = &mut self.reader {
            reader.reset();
        }
        self.stopped = false;
    }

    /// The raw-tail escape (ADR-013 §3, as amended by ADR-020 §1;
    /// ADR-018 §4): after [`Session::read_request`] observed the
    /// command request's terminating flush, the pkt-line machinery
    /// surrenders the read side's remaining bytes as a raw async reader
    /// ([`RawTail`]) — the push pack follows the flush **unframed**, so
    /// the reader bows out. The no-byte-loss argument: the demand-driven
    /// iterator reads with exact framing (never past the stop line) and
    /// clears its buffer at the flush, so at surrender the transport is
    /// positioned byte-exactly after the terminating flush — the tail
    /// reader's first bytes continue the stream exactly there, and the
    /// consumed-bytes counter rides the tail for observability (the
    /// session's total reconciles).
    ///
    /// Machinery misuse is refused, never silently re-entered:
    ///
    /// - before a completed parse ([`Session::read_request`] has not
    ///   returned a request) — [`SubstrateError::RawTailUnavailable`]
    ///   and the reader stays in place;
    /// - after a terminal error — the cached terminal error (the escape
    ///   is dead with the session);
    /// - a second escape — [`SubstrateError::RawTailUnavailable`] (the
    ///   escape is one-shot), and every later
    ///   [`Session::read_request`] is refused the same way (the reader
    ///   is gone — push reads no more pkt-lines; the session ends after
    ///   the report).
    ///
    /// Contract stamp: ADR-013 §3, ADR-020 §1, ADR-018 §4, ADR-005 §1,
    /// ADR-009 (the consumed-counter observability).
    pub fn take_raw_tail(&mut self) -> Result<RawTail<Compat<ReadHalf<T>>>, 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.stopped {
            self.reader = Some(reader);
            return Err(SubstrateError::RawTailUnavailable(
                "the request parse has not completed — the tail escapes only after the terminating flush",
            ));
        }
        let counting = reader.into_inner();
        Ok(RawTail::new(counting.inner, counting.consumed))
    }

    /// Reads one V0-framed push request (ADR-013 §3, one request per
    /// stream, as amended by ADR-020 §1): the command lines
    /// (`<old> <new> <ref>`, caps NUL-attached on the first line —
    /// there is no `command=` line and no `0001` delimiter on the push
    /// path), the command section's terminating flush, then — only when
    /// the client selected the `push-options` cap — the options
    /// section's bare lines to its own flush (ADR-020 §1 rule (i): the
    /// section is consumed *before* the pack tail is ever looked at).
    /// The pack follows the last flush **unframed**, so the parse stops
    /// dead at it and
    /// [`Session::take_raw_tail`] surrenders the read side.
    ///
    /// Returns:
    ///
    /// - `Ok(Some(request))` — the parsed request ([`Request`] with the
    ///   receive-pack command marker; the command lines in `pre_lines`,
    ///   the options section in `args`). An **empty command set is the
    ///   nothing-to-do shape, not an error** (ADR-013 §3 — the machine
    ///   replies flush, no report).
    /// - `Ok(None)` — the peer closed the stream cleanly before any
    ///   request byte; the session is over (the same shape
    ///   [`Session::read_request`] reports for the fetch loop's clean
    ///   close; a client that never sent an advertisement answer is no
    ///   error).
    /// - `Err(SubstrateError::Io | SubstrateError::Protocol)` — terminal
    ///   ([`SubstrateError::is_terminal`]), exactly as
    ///   [`Session::read_request`]: malformed pkt-lines, peer `ERR`
    ///   lines, truncation (the stream ended mid-section — the client
    ///   died before the flush), and `0001`/`0002` packets (not push
    ///   shapes — the V0 body has no V2 wrapper). The entries band-map
    ///   protocol failures; io errors end the session with no band.
    ///
    /// The stop-delimiter discipline applies one-sided: the reader is
    /// left inert after the terminating flush (one request per stream —
    /// the machinery's refusals
    /// ([`SubstrateError::ResetRequired`], [`SubstrateError::RawTailUnavailable`])
    /// hold even against a misbehaving caller), and
    /// [`Session::reset_request`] would never legitimately run after it.
    ///
    /// 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.
    ///
    /// Contract stamp: ADR-013 §3 (one POST / one stream, the V0
    /// request grammar, the nothing-to-do shape), §11 (the
    /// push-options section); ADR-020 §1 rule (i) (consumed before the
    /// peek); push-captures.md §Request; core/w3-duplex-serving.
    ///
    /// The per-request budget
    /// ([`Limits::max_request_body`](crate::Limits::max_request_body))
    /// bounds the accumulated command/options-section bytes, mirroring
    /// `read_post`'s stateless enforcement (ADR-009's fail-closed
    /// client-fault class) — see [`Session::read_request`]'s budget
    /// note.
    pub async fn read_push_request(&mut self) -> Result<Option<Request>, SubstrateError> {
        let budget = self.limits.max_request_body;
        let Self {
            reader,
            consumed,
            stopped,
            terminal,
            ..
        } = self;
        if let Some(cached) = terminal.clone() {
            return Err(cached);
        }
        let Some(reader) = reader else {
            return Err(SubstrateError::RawTailUnavailable(
                "the pkt-line reader is gone — the raw tail was surrendered; push reads no more commands",
            ));
        };
        if *stopped {
            return Err(SubstrateError::ResetRequired);
        }
        let consumed_before = consumed.load(Ordering::Relaxed);

        // Classified into owned payloads so the packetline buffer borrow
        // ends before any state mutation (the read_request's shape).
        enum Payload {
            Text(String),
            StopPacket(String),
            ErrLine(String),
        }

        let mut commands: Vec<String> = Vec::new();
        let mut args: Vec<String> = Vec::new();
        let mut in_options = false;

        loop {
            let payload: Option<Payload> = match reader.read_line().await {
                None => {
                    // The section's terminating flush stopped the
                    // iteration — the stream sits byte-exact past it,
                    // the escape's no-loss precondition.
                    None
                }
                Some(Ok(Ok(line))) => Some(match line {
                    PacketLineRef::Data(d) => {
                        if let Some(message) = line
                            .check_error()
                            .map(|e| String::from_utf8_lossy(e.0).into_owned())
                        {
                            Payload::ErrLine(message)
                        } else {
                            let text = String::from_utf8_lossy(d)
                                .trim_end_matches(['\n', '\r'])
                                .to_owned();
                            Payload::Text(text)
                        }
                    }
                    PacketLineRef::Delimiter => Payload::StopPacket(
                        "a 0001 delimiter in a push request is not a push shape".to_owned(),
                    ),
                    PacketLineRef::Flush | PacketLineRef::ResponseEnd => {
                        Payload::StopPacket("unexpected stop packet mid-request".to_owned())
                    }
                }),
                Some(Ok(Err(decode_error))) => {
                    let error = SubstrateError::Protocol(format!(
                        "malformed pkt-line in the push request: {decode_error}"
                    ));
                    *terminal = Some(error.clone());
                    return Err(error);
                }
                Some(Err(io_error)) => {
                    // The client died before any request byte: the quiet
                    // session-end shape. Past a heard byte the
                    // truncation is terminal (the client's flush never
                    // came — the pack tail's state is unknowable).
                    let clean_close = io_error.kind() == io::ErrorKind::UnexpectedEof
                        && consumed.load(Ordering::Relaxed) == consumed_before
                        && commands.is_empty()
                        && args.is_empty();
                    if clean_close {
                        *terminal = Some(SubstrateError::Io(io::Error::new(
                            io::ErrorKind::UnexpectedEof,
                            "the peer closed the stream",
                        )));
                        return Ok(None);
                    }
                    let error = SubstrateError::Io(io_error);
                    *terminal = Some(error.clone());
                    return Err(error);
                }
            };

            match payload {
                None => {
                    reader.reset();
                    if commands.is_empty() && !in_options {
                        // The nothing-to-do shape's bytes (an immediate
                        // flush) — a parsed request with no commands,
                        // the machine replies flush, no report.
                        *stopped = true;
                        return Ok(Some(Request {
                            command: "git-receive-pack".to_owned(),
                            pre_lines: Vec::new(),
                            args: Vec::new(),
                        }));
                    }
                    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)
                    {
                        // The client selected the cap: the options
                        // section rides before the pack (rule (i)).
                        in_options = true;
                        continue;
                    }
                    *stopped = true;
                    return Ok(Some(Request {
                        command: "git-receive-pack".to_owned(),
                        pre_lines: commands,
                        args,
                    }));
                }
                Some(Payload::ErrLine(message)) => {
                    let error =
                        SubstrateError::Protocol(format!("peer sent an ERR line: {message}"));
                    *terminal = Some(error.clone());
                    return Err(error);
                }
                Some(Payload::StopPacket(message)) => {
                    let error = SubstrateError::Protocol(message);
                    *terminal = Some(error.clone());
                    return Err(error);
                }
                Some(Payload::Text(text)) => {
                    if in_options {
                        args.push(text);
                    } else {
                        commands.push(text);
                    }
                }
            }

            let consumed_total = consumed.load(Ordering::Relaxed);
            if consumed_total.saturating_sub(consumed_before) > budget {
                let error = SubstrateError::Protocol(format!(
                    "the request exceeds the max_request_body budget ({budget}) — fail closed (ADR-009)"
                ));
                *terminal = Some(error.clone());
                return Err(error);
            }
        }
    }

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

    /// Writes a terminating flush.
    pub async fn write_flush(&mut self) -> Result<(), SubstrateError> {
        let writer = self.writer.as_mut().ok_or(SubstrateError::WriterTaken)?;
        encode::flush_to_write(writer).await?;
        Ok(())
    }

    /// Writes band-1 sideband data, chunked at [`SIDEBAND_CHUNK_SIZE`]
    /// (defensively re-chunked here; the machines' bridges deliver
    /// pre-chunked data) — the done-round packfile section's data band
    /// (ADR-014 §2). The seam's blocking-side producers use the
    /// [`Sideband`] sink instead; this is the machines' async relay end.
    pub async fn write_data_band(&mut self, data: &[u8]) -> Result<(), SubstrateError> {
        let writer = self.writer.as_mut().ok_or(SubstrateError::WriterTaken)?;
        for chunk in data.chunks(SIDEBAND_CHUNK_SIZE) {
            encode::band_to_write(Channel::Data, chunk, &mut *writer).await?;
        }
        Ok(())
    }

    /// Writes the pkt-line error band (`ERR <message>`): the duplex
    /// mapping ADR-005 §4 prescribes for protocol errors before the
    /// session ends.
    pub async fn write_error_band(&mut self, message: &str) -> Result<(), SubstrateError> {
        let writer = self.writer.as_mut().ok_or(SubstrateError::WriterTaken)?;
        encode::error_to_write(message.as_bytes(), writer).await?;
        Ok(())
    }
}

impl<T> Session<T>
where
    T: TokioAsyncRead + TokioAsyncWrite + Unpin + Send + 'static,
{
    /// Starts a sideband response: takes the session's response writer
    /// out and returns the blocking-side sink ([`Sideband`],
    /// ADR-004/005 §1) whose drain task emits ≤65000-byte band-1 chunks
    /// over a bounded channel.
    ///
    /// Exactly one task drives a given sink (POC-2's contract), enforced
    /// by construction: the writer is moved into the drain task, so
    /// while the sink lives no second sink can be constructed and no
    /// text/error-band write can run — both fail with
    /// [`SubstrateError::WriterTaken`]. Finish with
    /// [`Session::finish_sideband`], which sends the one-shot section
    /// flush (idempotent) and restores the writer.
    ///
    /// The sink's `std::io::Write` end must be driven from a blocking
    /// thread (`tokio::task::spawn_blocking`): writes park on the
    /// bounded channel — never drive it from a runtime worker thread.
    pub fn sideband(&mut self) -> Result<Sideband<T>, SubstrateError> {
        let Some(writer) = self.writer.take() else {
            return Err(SubstrateError::WriterTaken);
        };
        let (tx, rx) = mpsc::channel(SIDEBAND_CHANNEL_CAPACITY);
        let (restore, restore_rx) = oneshot::channel();
        tokio::spawn(drive_sideband(rx, writer, restore));
        Ok(Sideband {
            buf: Vec::with_capacity(SIDEBAND_CHUNK_SIZE),
            tx,
            restore_rx,
            flushed: false,
            bytes_pushed: 0,
        })
    }

    /// Ends the sideband response: ensures the one-shot section flush
    /// (idempotent — a sink already flushed is not flushed again),
    /// waits for the drain task, and restores the response writer onto
    /// the session. The write failures the drain task hit surface here
    /// (terminal, [`SubstrateError::is_terminal`]).
    pub async fn finish_sideband(&mut self, sink: Sideband<T>) -> Result<(), SubstrateError> {
        let writer = sink.finish().await?;
        self.writer = Some(writer);
        Ok(())
    }
}

enum SidebandMsg {
    /// One ≤[`SIDEBAND_CHUNK_SIZE`] band-1 chunk.
    Chunk(Vec<u8>),
    /// The one-shot section flush.
    Flush,
}

/// The sideband sink's drain task: emits each bounded chunk as a band-1
/// pkt-line, the flush once, then restores the writer through the
/// oneshot. Writer transport failures terminate the response.
async fn drive_sideband<T>(
    mut rx: mpsc::Receiver<SidebandMsg>,
    mut writer: SessionWriter<T>,
    restore: oneshot::Sender<Result<SessionWriter<T>, SubstrateError>>,
) where
    T: TokioAsyncWrite + Unpin,
{
    let mut flushed = false;
    let mut failure: Option<SubstrateError> = None;
    while let Some(message) = rx.recv().await {
        match message {
            SidebandMsg::Chunk(chunk) => {
                if flushed {
                    failure = Some(SubstrateError::Protocol(
                        "sideband data written after the section flush".to_owned(),
                    ));
                    break;
                }
                if let Err(e) = encode::band_to_write(Channel::Data, &chunk, &mut writer).await {
                    failure = Some(SubstrateError::Io(e));
                    break;
                }
            }
            SidebandMsg::Flush => {
                if !flushed {
                    if let Err(e) = encode::flush_to_write(&mut writer).await {
                        failure = Some(SubstrateError::Io(e));
                        break;
                    }
                    flushed = true;
                }
            }
        }
    }
    // Push the framed bytes onto the underlying stream, best-effort —
    // the packetline flush frame alone does not guarantee a downstream
    // flush.
    let _ = FuturesAsyncWriteExt::flush(&mut writer).await;
    let _ = restore.send(match failure {
        Some(e) => Err(e),
        None => Ok(writer),
    });
}

/// The blocking-side sideband sink (ADR-004/ADR-005 §1, POC-2's
/// `SidebandSink`): a `std::io::Write` end for pack generation that
/// caps band-1 chunks at [`SIDEBAND_CHUNK_SIZE`] (65000; underneath the
/// protocol's own 65515 ceiling — headroom for the band byte and length
/// prefix) and parks on a bounded channel ([`SIDEBAND_CHANNEL_CAPACITY`])
/// when the drain task falls behind.
///
/// Contract (POC-2's, structurally enforced here):
///
/// - **exactly one driver task**, on a blocking thread
///   (`tokio::task::spawn_blocking`) — construction takes the
///   session's response writer, so no second sink or concurrent write
///   path can exist; write/flush park the blocking thread on the
///   bounded channel, never a runtime worker;
/// - **idempotent, one-shot flush** — the first
///   [`io::Write::flush`] emits the buffered remainder and the section
///   flush pkt-line; every later flush is a no-op, and writes after it
///   fail with [`io::ErrorKind::BrokenPipe`] (data after the section
///   flush would be a second section — a protocol error);
/// - **drop without [`Session::finish_sideband`]** is loud later: the
///   writer stays gone and every write path reports
///   [`SubstrateError::WriterTaken`].
pub struct Sideband<T> {
    buf: Vec<u8>,
    tx: mpsc::Sender<SidebandMsg>,
    restore_rx: oneshot::Receiver<Result<SessionWriter<T>, SubstrateError>>,
    flushed: bool,
    bytes_pushed: u64,
}

impl<T> std::fmt::Debug for Sideband<T> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Sideband")
            .field("flushed", &self.flushed)
            .field("bytes_pushed", &self.bytes_pushed)
            .finish_non_exhaustive()
    }
}

impl<T> Sideband<T> {
    /// Bytes handed to the sink so far (POC-2's counter — server
    /// metrics feed on it).
    pub fn bytes_pushed(&self) -> u64 {
        self.bytes_pushed
    }

    fn send(&mut self, message: SidebandMsg) -> io::Result<()> {
        block_on(self.tx.send(message)).map_err(|_| {
            io::Error::new(io::ErrorKind::BrokenPipe, "the sideband drain task is gone")
        })
    }

    fn push_buffered_chunk(&mut self) -> io::Result<()> {
        if self.buf.is_empty() {
            return Ok(());
        }
        let chunk = std::mem::take(&mut self.buf);
        self.send(SidebandMsg::Chunk(chunk))
    }

    async fn finish(mut self) -> Result<SessionWriter<T>, SubstrateError> {
        std::io::Write::flush(&mut self).map_err(SubstrateError::Io)?;
        // The drain task ends only once every sender is gone — drop the
        // channel front before awaiting the restore, or this await would
        // deadlock with the drain's own wait for the senders to drop.
        let Sideband {
            buf,
            tx,
            restore_rx,
            ..
        } = self;
        drop(buf);
        drop(tx);
        match restore_rx.await {
            Ok(Ok(writer)) => Ok(writer),
            Ok(Err(e)) => Err(e),
            Err(_) => Err(SubstrateError::Io(io::Error::other(
                "the sideband drain task ended without restoring the response writer",
            ))),
        }
    }
}

impl<T> io::Write for Sideband<T> {
    fn write(&mut self, data: &[u8]) -> io::Result<usize> {
        if self.flushed {
            return Err(io::Error::new(
                io::ErrorKind::BrokenPipe,
                "the sideband section was already flushed",
            ));
        }
        let mut consumed = 0;
        while consumed < data.len() {
            let space = SIDEBAND_CHUNK_SIZE - self.buf.len();
            let take = space.min(data.len() - consumed);
            self.buf.extend_from_slice(&data[consumed..consumed + take]);
            consumed += take;
            if self.buf.len() == SIDEBAND_CHUNK_SIZE {
                self.push_buffered_chunk()?;
            }
        }
        self.bytes_pushed += consumed as u64;
        Ok(consumed)
    }

    fn flush(&mut self) -> io::Result<()> {
        if self.flushed {
            return Ok(());
        }
        self.push_buffered_chunk()?;
        self.send(SidebandMsg::Flush)?;
        self.flushed = true;
        Ok(())
    }
}

/// The wall-clock breach error (ADR-009's server/session budget class —
/// 503-mapped at the door on http; on duplex the breach maps to the
/// error band + close).
fn wall_clock_breach(limits: &Limits) -> SubstrateError {
    SubstrateError::Protocol(format!(
        "the session exceeded the session_wall_clock budget ({:?}) — the session closes, server/session budget class (ADR-009)",
        limits.session_wall_clock
    ))
}

/// Maps a terminal protocol error to its ADR-005 §4 duplex wire shape:
/// the pkt-line error band, then the session ends. `io` errors get no
/// band (the transport may be dead — the unconditional error-break rule
/// is already the machinery's); machinery misuse never reaches the wire.
/// Best-effort: a failing band write (a peer already gone) leaves the
/// original error as the session's return value.
async fn write_protocol_band<T>(
    session: &mut Session<T>,
    error: &SubstrateError,
) -> Result<(), SubstrateError>
where
    T: TokioAsyncRead + TokioAsyncWrite + Unpin,
{
    match error {
        SubstrateError::Protocol(message) => session.write_error_band(message).await,
        _ => Ok(()),
    }
}

/// Drives one parsed `ls-refs` command through the machine.
async fn drive_ls_refs<T>(
    repo: &AuthorizedRepo,
    session: &mut Session<T>,
    request: &Request,
    refs: Arc<dyn GitRefs>,
    limits: &Limits,
) -> Result<(), SubstrateError>
where
    T: TokioAsyncRead + TokioAsyncWrite + Unpin + Send,
{
    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(session, listed, &args, limits).await
}

/// Drives one parsed `fetch` command through the machine.
async fn drive_fetch<T>(
    repo: &AuthorizedRepo,
    session: &mut Session<T>,
    request: &Request,
    rounds: &mut FetchRounds,
    packgen: Arc<dyn GitPackGen>,
    limits: &Limits,
    pipeline: Arc<Semaphore>,
) -> Result<(), SubstrateError>
where
    T: TokioAsyncRead + TokioAsyncWrite + Unpin + Send,
{
    serve_fetch_round(
        session,
        request,
        rounds,
        packgen,
        repo.record(),
        limits,
        pipeline,
    )
    .await
}

/// Drives one parsed request through its command's machine — the
/// dispatch on `Request.command` (ADR-002's duplex command loop body):
/// `ls-refs` → the ls-refs machine, `fetch` → the negotiation machine,
/// an unknown command → the typed decline (banded by the loop). The
/// seven inputs are the dispatch surface both machines share, not a
/// shape to flatten.
#[allow(clippy::too_many_arguments)]
async fn drive_command<T>(
    repo: &AuthorizedRepo,
    session: &mut Session<T>,
    request: &Request,
    rounds: &mut FetchRounds,
    refs: Arc<dyn GitRefs>,
    packgen: Arc<dyn GitPackGen>,
    limits: &Limits,
    pipeline: Arc<Semaphore>,
) -> Result<(), SubstrateError>
where
    T: TokioAsyncRead + TokioAsyncWrite + Unpin + Send,
{
    match request.command.as_str() {
        "ls-refs" => drive_ls_refs(repo, session, request, refs, limits).await,
        "fetch" => drive_fetch(repo, session, request, rounds, packgen, limits, pipeline).await,
        other => Err(SubstrateError::Protocol(format!(
            "unknown command {other:?}{UNKNOWN_COMMAND_DECLINE}"
        ))),
    }
}

/// The duplex serving entry (ADR-002's first session entry point, as
/// amended by ADR-016 §3): turns (authorization inputs, service, duplex
/// stream, [`Limits`], backend handles) into a running protocol
/// session. This is the entry the `GitAdapter` and the channels open op
/// drive (the wave-2 native transport), and the shape an embedder's
/// stream door drives directly.
///
/// The inputs are the ADR-016 §3 duplex tuple realized: the
/// [`AuthorizedRepo`] marker is the authorization's outcome — the
/// ADR-007 resolve → authorize sequence (steps 2–3) runs upstream (the
/// adapter establishes it), so the entry never resolves or authorizes;
/// taking the marker by type is the enforcement that skipping the check
/// cannot compile ([`crate::authorize`]).
///
/// The service selects the session (ADR-016 §3, the state-machine
/// selector):
///
/// - **[`Service::UploadPack`]** (wave 2): the V2 fetch session —
///   advertisement-once, then the command loop.
/// - **[`Service::ReceivePack`]** (wave 3, ADR-013 §3 "one POST / one
///   stream"): the push session is **one request per stream** — the V0
///   ref advertisement ([`advertise_v0`], server speaks first, no
///   service prefix — ADR-013 §1–2), then one V0 request read
///   ([`Session::read_push_request`], the command grammar; a clean
///   peer close is the quiet session end), then the raw tail surrenders
///   and the push machine ([`serve_receive_pack`]) runs under the
///   **duplex tail discipline** (the bounded peek's rules — ADR-020
///   §1), and the session ends cleanly after the report
///   (advertisement-once is trivially held — one advertisement, never
///   between commands; a second command read is never attempted).
/// - **The command loop** (fetch): commands dispatch on `Request.command`
///   — `ls-refs` → [`serve_ls_refs`], `fetch` → [`serve_fetch_round`]
///   (the negotiation-rounds [`FetchRounds`] budget counter lives for
///   the whole session); an unknown command declines via the error
///   band; between commands the reset discipline applies
///   ([`Session::reset_request`] re-arms; [`SubstrateError::ResetRequired`]
///   is machinery misuse and never a wire shape).
///
/// Behavior per ADR-005 §4 and ADR-009 (shared by both services):
///
/// - **The wall clock**: the whole session runs under
///   [`Limits::session_wall_clock`]; breach is the server/session
///   budget class — on duplex, the error band + close (ADR-009's breach
///   behavior). Every phase bounds itself by the remaining session
///   budget (on the push session: the advertisement, the request read,
///   and the machine — the peek window ([`Limits::packless_peek`])
///   runs inside the machine's); a breach abandons the in-flight step —
///   a blocking pipeline already launched for a done round or a pack
///   ingestion still frees its permit when its relay breaks (accounting
///   stays exact).
/// - **The error mapping**: `SubstrateError::Protocol` →
///   [`Session::write_error_band`] then the session ends (the typed
///   failure returns to the caller for telemetry); io errors are
///   terminal with no band (the transport may be dead); a clean close
///   (`Ok(None)`) ends the session quietly. Backend failures band-map
///   through the negotiation module's storage-to-session mapping (the
///   push advertisement's listing failure included). The malformed-push
///   peek error and the shallow/object-format rejections are protocol
///   failures (banded). Denials are upstream: the resolve→authorize
///   front's generic band precedes any protocol byte.
/// - **The pipeline-concurrency semaphore** (ADR-009's enforcement
///   point, backend.md §concurrency model): **provided by the caller**
///   — the server-wide shared instance the assembled construction owns
///   (assembler config origin; ADR-009's server-wide wording) — and
///   shared with the machines; the permit wraps each `generate`/`prepare`
///   call.
/// - **The `push-options` negotiated-gate flag** (ADR-013 §11; the
///   assembler enables it, default off in v1): the same assembler
///   config value the stateless advertisement GET carries — the V0
///   advertisement gates the `push-options` token on it and the
///   machine's grammar enforces the fail-closed selection check.
///
/// Front-door-blind (convention 15): the stream is the plain duplex
/// byte stream (`AsyncRead + AsyncWrite + Unpin`, the alkcall
/// `BiStream` contract, + `Send + 'static` for the machines' spawned
/// relay tasks — `Sync` is deliberately absent so the alkcall
/// `BiStream` satisfies the bound); no channel/http types appear below
/// it.
#[allow(clippy::too_many_arguments)]
pub async fn serve_duplex<T>(
    repo: &AuthorizedRepo,
    stream: T,
    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
    T: TokioAsyncRead + TokioAsyncWrite + Unpin + Send + 'static,
{
    let mut session = Session::new(stream, limits, service);
    let deadline = tokio::time::Instant::now() + limits.session_wall_clock;

    match service {
        Service::ReceivePack => {
            let listed =
                match tokio::time::timeout_at(deadline, refs.list_refs(repo.record())).await {
                    Ok(Ok(listed)) => listed,
                    Ok(Err(error)) => {
                        let error = storage_to_session_error(error);
                        let _ = write_protocol_band(&mut session, &error).await;
                        return Err(error);
                    }
                    Err(_elapsed) => return Err(end_wall_clock_breach(&mut session, &limits).await),
                };
            match tokio::time::timeout_at(
                deadline,
                advertise_v0(&mut session, listed, push_options_negotiated, &limits),
            )
            .await
            {
                Ok(Ok(())) => {}
                Ok(Err(e)) => return Err(e),
                Err(_elapsed) => return Err(end_wall_clock_breach(&mut session, &limits).await),
            }

            let read = tokio::time::timeout_at(deadline, session.read_push_request()).await;
            let request = match read {
                Ok(Ok(Some(request))) => request,
                Ok(Ok(None)) => return Ok(()),
                Ok(Err(e)) => {
                    let _ = write_protocol_band(&mut session, &e).await;
                    return Err(e);
                }
                Err(_elapsed) => return Err(end_wall_clock_breach(&mut session, &limits).await),
            };

            let tail = session.take_raw_tail()?;
            let outcome = serve_receive_pack(
                &mut session,
                &request,
                tail,
                push_options_negotiated,
                Arc::clone(&refs),
                Arc::clone(&ingest),
                repo.record(),
                &limits,
                PackTailDiscipline::Duplex,
                Arc::clone(&pipeline),
            );
            match tokio::time::timeout_at(deadline, outcome).await {
                Ok(Ok(())) => Ok(()),
                Ok(Err(e)) => {
                    let _ = write_protocol_band(&mut session, &e).await;
                    Err(e)
                }
                Err(_elapsed) => Err(end_wall_clock_breach(&mut session, &limits).await),
            }
        }
        Service::UploadPack => {
            match tokio::time::timeout_at(deadline, advertise_v2(&mut session)).await {
                Ok(Ok(())) => {}
                Ok(Err(e)) => return Err(e),
                Err(_elapsed) => return Err(wall_clock_breach(&limits)),
            }

            let mut rounds = FetchRounds::default();
            loop {
                let read = tokio::time::timeout_at(deadline, session.read_request()).await;
                let request = match read {
                    Ok(Ok(Some(request))) => request,
                    Ok(Ok(None)) => return Ok(()),
                    Ok(Err(e)) => {
                        let _ = write_protocol_band(&mut session, &e).await;
                        return Err(e);
                    }
                    Err(_elapsed) => {
                        return Err(end_wall_clock_breach(&mut session, &limits).await);
                    }
                };

                let outcome = drive_command(
                    repo,
                    &mut session,
                    &request,
                    &mut rounds,
                    Arc::clone(&refs),
                    Arc::clone(&packgen),
                    &limits,
                    Arc::clone(&pipeline),
                );
                let served = tokio::time::timeout_at(deadline, outcome).await;

                match served {
                    Ok(Ok(())) => session.reset_request(),
                    Ok(Err(e)) => {
                        let _ = write_protocol_band(&mut session, &e).await;
                        return Err(e);
                    }
                    Err(_elapsed) => {
                        return Err(end_wall_clock_breach(&mut session, &limits).await);
                    }
                }
            }
        }
    }
}

/// Ends a session at the wall-clock breach (ADR-009's duplex mapping):
/// the band + close. Best-effort — a failing band write (the peer is
/// gone, or stops reading) still returns the breach error.
async fn end_wall_clock_breach<T>(session: &mut Session<T>, limits: &Limits) -> SubstrateError
where
    T: TokioAsyncRead + TokioAsyncWrite + Unpin,
{
    let breach = wall_clock_breach(limits);
    if let SubstrateError::Protocol(message) = &breach {
        let _ = session.write_error_band(message).await;
    }
    breach
}

/// The ADR-007 front over the registry (steps 2–3, the duplex entry's
/// pre-protocol gate): resolves the wire-repo id against the
/// [`GitRegistry`], evaluates [`authorize`](crate::authorize) with the
/// action the service selects (read for fetch, write for push — ADR-011/015),
/// constructs the [`AuthorizedRepo`] marker, and hands the established
/// session to [`serve_duplex`].
///
/// Every failure collapses to **one** generic denial before any
/// protocol byte (ADR-008/N-2 — the no-existence-oracle rule, transport.md
/// §error taxonomy): a resolution failure of any kind (unknown repo,
/// registry-side denial, backing-store failure — all are not-served
/// from the caller's view) and a failed policy evaluation both produce
/// the byte-identical denial: the generic error band + close, and
/// [`SubstrateError::AccessDenied`] as the entry's return value.
/// The band's message carries only the client-requested repo id
/// (nothing the response itself reveals).
///
/// The wire repo id is an opaque registry key — validated for rejection
/// by the registry, never decomposed or path-joined (ADR-008,
/// backend.md §repo-id grammar).
#[allow(clippy::too_many_arguments)]
pub async fn serve_duplex_resolved<T>(
    repo_id: &str,
    identity: Option<&Identity>,
    service: Service,
    stream: T,
    limits: Limits,
    registry: Arc<dyn GitRegistry>,
    refs: Arc<dyn GitRefs>,
    packgen: Arc<dyn GitPackGen>,
    ingest: Arc<dyn GitPackIngest>,
    push_options_negotiated: bool,
    pipeline: Arc<Semaphore>,
) -> Result<(), SubstrateError>
where
    T: TokioAsyncRead + TokioAsyncWrite + Unpin + Send + 'static,
{
    let action = if service.is_write() {
        AccessAction::Write
    } else {
        AccessAction::Read
    };
    let marker = match registry.resolve(repo_id).await {
        Ok(record) => AuthorizedRepo::authorize(record, identity, action),
        Err(_) => None,
    };
    match marker {
        Some(marker) => {
            serve_duplex(
                &marker,
                stream,
                limits,
                service,
                refs,
                packgen,
                ingest,
                push_options_negotiated,
                pipeline,
            )
            .await
        }
        None => Err(deny_generic(repo_id, stream).await),
    }
}

/// The one pre-session wire error shape, shared by the collapsed generic
/// denial and the native preamble's declines: a single `ERR <message>`
/// pkt-line, flushed, then the stream closes — best-effort, the message
/// *is* the wire (a failing write leaves the caller to close quietly).
async fn write_session_error<T>(stream: T, message: &str) -> Result<(), io::Error>
where
    T: TokioAsyncRead + TokioAsyncWrite + Unpin,
{
    use tokio::io::AsyncWriteExt;

    let mut compat = stream.compat_write();
    encode::error_to_write(message.as_bytes(), &mut compat).await?;
    FuturesAsyncWriteExt::flush(&mut compat).await?;
    let shutdown = compat.into_inner().shutdown().await;
    shutdown
}

/// The generic pre-session denial (ADR-008/N-2's collapsed wire shape):
/// the error band naming only the client-sent repo id + close, with
/// [`SubstrateError::AccessDenied`] as the typed return — no cause
/// distinguishes the failures that produce it (unknown repo, a failed
/// authorize evaluation, a grammar-invalid repo id on the serving
/// paths). Shared by the resolve→authorize front and the native
/// transport.
pub(crate) async fn deny_generic<T>(repo_id: &str, stream: T) -> SubstrateError
where
    T: TokioAsyncRead + TokioAsyncWrite + Unpin,
{
    let _ = write_session_error(stream, &format!("access denied: {repo_id}")).await;
    SubstrateError::AccessDenied
}

/// Writes the native preamble's clear decline (a malformed line, an
/// unknown service, the V0/V1 fetch version-decline path — all repo
///blind, no existence oracle) as the same pre-session error shape.
pub(crate) async fn deny_decline<T>(stream: T, message: &str)
where
    T: TokioAsyncRead + TokioAsyncWrite + Unpin,
{
    let _ = write_session_error(stream, message).await;
}

#[cfg(test)]
mod tests {
    use super::*;
    use futures::io::AsyncReadExt as FuturesAsyncReadExt;
    use std::io::Write as _;
    use std::time::Duration;
    use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt};
    use tokio::time::timeout;

    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
    }

    async fn read_pkts(reader: &mut tokio::io::DuplexStream, count: usize) -> Vec<Vec<u8>> {
        let mut packets = Vec::with_capacity(count);
        for _ in 0..count {
            let mut header = [0u8; 4];
            reader
                .read_exact(&mut header)
                .await
                .expect("pkt-line header");
            let len = usize::from_str_radix(std::str::from_utf8(&header).expect("hex header"), 16)
                .expect("hex length");
            let mut payload = vec![0u8; len.saturating_sub(4)];
            reader
                .read_exact(&mut payload)
                .await
                .expect("pkt-line payload");
            packets.push(payload);
        }
        packets
    }

    fn session() -> (tokio::io::DuplexStream, Session<tokio::io::DuplexStream>) {
        let (peer, stream) = duplex(1 << 16);
        (
            peer,
            Session::new(stream, Limits::default(), Service::UploadPack),
        )
    }

    #[tokio::test]
    async fn session_constructs_from_stream_limits_service() {
        let (_peer, session) = session();
        assert_eq!(session.service(), Service::UploadPack);
        assert_eq!(*session.limits(), Limits::default());
        assert_eq!(session.consumed_bytes(), 0);
        assert_eq!(session.transport_reads(), 0);
    }

    #[tokio::test]
    async fn request_reader_parses_command_pre_lines_args_and_flush() {
        let (mut peer, mut session) = session();

        let request = [
            pkt(b"command=ls-refs\n"),
            pkt(b"agent=git/2.43\n"),
            DELIMITER.to_vec(),
            pkt(b"peel\n"),
            pkt(b"ref-prefix refs/heads/\n"),
            FLUSH.to_vec(),
        ]
        .concat();
        peer.write_all(&request).await.unwrap();

        let parsed = session.read_request().await.unwrap().expect("a request");
        assert_eq!(parsed.command, "ls-refs");
        assert_eq!(parsed.pre_lines, vec!["agent=git/2.43"]);
        assert_eq!(parsed.args, vec!["peel", "ref-prefix refs/heads/"]);
        assert!(session.consumed_bytes() > 0);
    }

    /// The raw-tail escape (ADR-013 §3): after the command request's
    /// terminating flush the read side surrenders raw — the tail's
    /// first bytes continue the stream byte-exactly from the flush
    /// (dribbled chunks split inside the flush and inside the tail's
    /// own framing), the session's consumed counter reconciles with the
    /// read totals, and EOF ends the tail cleanly.
    #[tokio::test]
    async fn raw_tail_escapes_after_the_flush_without_byte_loss() {
        let (mut peer, mut session) = session();
        let framed = [
            pkt(b"command=git-receive-pack\n"),
            pkt(b"agent=git/2.43\n"),
            DELIMITER.to_vec(),
            pkt(b"push-options"),
        ]
        .concat();
        let tail: Vec<u8> = (0..9000u32).map(|i| (i % 251) as u8).collect();
        // Dribble: the flush and the tail's first bytes ride one chunk,
        // later tail bytes split inside the magic bytes.
        peer.write_all(&framed[..14]).await.unwrap();
        peer.write_all(&framed[14..]).await.unwrap();
        peer.write_all(&FLUSH[..2]).await.unwrap();
        peer.write_all(&[&FLUSH[2..], &tail[..5]].concat())
            .await
            .unwrap();
        peer.write_all(&tail[5..4000]).await.unwrap();
        peer.flush().await.unwrap();
        peer.write_all(&tail[4000..]).await.unwrap();

        let parsed = session.read_request().await.unwrap().expect("a request");
        assert_eq!(parsed.command, "git-receive-pack");
        assert_eq!(
            session.consumed_bytes(),
            (framed.len() + FLUSH.len()) as u64,
            "the parse consumed exactly the framed request (flush included)"
        );

        let mut tail_reader = session.take_raw_tail().unwrap();
        drop(peer);
        let mut got = Vec::new();
        tail_reader.read_to_end(&mut got).await.unwrap();
        assert_eq!(got, tail, "the tail continues byte-exactly");
        assert_eq!(
            tail_reader.consumed_bytes(),
            (framed.len() + FLUSH.len()) as u64 + tail.len() as u64,
            "the tail rides the substrate counter: framed + tail"
        );
        assert_eq!(
            tail_reader.consumed_bytes(),
            session.consumed_bytes(),
            "one shared counter"
        );
    }

    /// The escape is machinery misuse before a completed parse — and
    /// the refusal leaves the reader in place (the session keeps
    /// working normally).
    #[tokio::test]
    async fn the_raw_tail_escape_refuses_without_a_completed_parse() {
        let (mut peer, mut session) = session();
        peer.write_all(&pkt(b"command=ls-refs\n")).await.unwrap();
        peer.write_all(FLUSH).await.unwrap();

        let error = session.take_raw_tail().unwrap_err();
        match error {
            SubstrateError::RawTailUnavailable(message) => {
                assert!(message.contains("parse"), "{message}");
            }
            other => panic!("the escape before a parse is misuse: {other:?}"),
        }
        // Refused, not destroyed: the reader stays usable.
        let parsed = session.read_request().await.unwrap().expect("a request");
        assert_eq!(parsed.command, "ls-refs");
    }

    /// The escape is one-shot; after it the pkt-line reader is gone and
    /// every later read_request is the refused misuse — never a silent
    /// re-entry.
    #[tokio::test]
    async fn the_raw_tail_escape_is_one_shot_and_the_reader_is_gone() {
        let (mut peer, mut session) = session();
        peer.write_all(&pkt(b"command=ls-refs\n")).await.unwrap();
        peer.write_all(FLUSH).await.unwrap();
        session.read_request().await.unwrap().expect("a request");

        let _tail = session.take_raw_tail().unwrap();
        let error = session.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 = session.read_request().await.unwrap_err();
        match error {
            SubstrateError::RawTailUnavailable(message) => {
                assert!(message.contains("gone"), "{message}");
                assert!(!error.is_terminal(), "misuse, not a wire shape");
            }
            other => panic!("post-escape reads are misuse: {other:?}"),
        }
    }

    /// The escape after a terminal error reports the cached terminal —
    /// the escape dies with the session.
    #[tokio::test]
    async fn the_raw_tail_escape_refuses_after_a_terminal_error() {
        let (mut peer, mut session) = session();
        peer.write_all(b"zzzz").await.unwrap();

        let terminal = session.read_request().await.unwrap_err();
        assert!(terminal.is_terminal());
        let error = session.take_raw_tail().unwrap_err();
        match (terminal, error) {
            (SubstrateError::Protocol(first), SubstrateError::Protocol(message)) => {
                assert!(first.contains("malformed pkt-line"), "{first}");
                assert!(message.contains("malformed pkt-line"), "{message}");
            }
            other => panic!("the terminal error stays terminal: {other:?}"),
        }
    }

    /// The stop-delimiter discipline: after a flush the reader refuses
    /// loudly until `reset_request` — and after the reset the next
    /// command is still heard in full. The swallowed-command hang
    /// (POC-1's second hang) is structurally prevented.
    #[tokio::test]
    async fn reset_is_required_after_flush_and_the_next_command_is_not_swallowed() {
        let (mut peer, mut session) = session();
        let requests = [
            pkt(b"command=ls-refs\n"),
            FLUSH.to_vec(),
            pkt(b"command=fetch\n"),
            FLUSH.to_vec(),
            pkt(b"command=ls-refs\n"),
            FLUSH.to_vec(),
        ]
        .concat();
        peer.write_all(&requests).await.unwrap();

        for expected in ["ls-refs", "fetch", "ls-refs"] {
            let parsed = session.read_request().await.unwrap().expect("a request");
            assert_eq!(parsed.command, expected);
            // Without a reset the reader refuses instead of inertly
            // swallowing what follows the flush.
            match session.read_request().await.unwrap_err() {
                SubstrateError::ResetRequired => {}
                other => panic!("expected the reset discipline: {other:?}"),
            }
            session.reset_request();
        }
    }

    /// The unconditional error-break rule: an error-shaped read (EOF
    /// mid-header, the POC-3 spin bug's shape) terminates the read side
    /// and every retry re-reports the cached error — the transport is
    /// never re-entered (read-call counter frozen), so no
    /// reset-and-retry loop is reachable.
    #[tokio::test]
    async fn error_shaped_reads_break_unconditionally() {
        let (peer, mut session) = session();
        let driver = tokio::spawn(async move {
            let mut peer = peer;
            // Three header bytes, then the stream ends mid-pkt-line.
            peer.write_all(b"000").await.ok();
            drop(peer);
        });
        driver.await.unwrap();

        let error = session.read_request().await.unwrap_err();
        let SubstrateError::Io(e) = &error else {
            panic!("a mid-header EOF is terminal io: {error:?}");
        };
        assert_eq!(e.kind(), io::ErrorKind::UnexpectedEof);
        assert!(error.is_terminal());
        // The partial header bytes were transported (not a clean close).
        assert_eq!(session.consumed_bytes(), 3);

        let reads_frozen = session.transport_reads();
        for _ in 0..10 {
            let retry = session.read_request().await.unwrap_err();
            let SubstrateError::Io(e) = &retry else {
                panic!("terminal errors stay terminal: {retry:?}");
            };
            assert_eq!(e.kind(), io::ErrorKind::UnexpectedEof);
        }
        assert_eq!(
            session.transport_reads(),
            reads_frozen,
            "no retry may re-enter the transport"
        );

        // Even an explicit reset cannot revive a terminated session.
        session.reset_request();
        assert!(session.read_request().await.is_err());
    }

    /// A clean peer close before any request byte reads as `Ok(None)`
    /// and ends the session — afterward the read side is terminal (no
    /// re-entry, no silent restart).
    #[tokio::test]
    async fn clean_peer_close_reads_as_none_and_then_is_terminal() {
        let (peer, mut session) = session();
        let driver = tokio::spawn(async move {
            let mut peer = peer;
            peer.write_all(&pkt(b"command=ls-refs\n")).await.unwrap();
            peer.write_all(FLUSH).await.unwrap();
            drop(peer);
        });
        driver.await.unwrap();

        let parsed = session.read_request().await.unwrap().expect("a request");
        assert_eq!(parsed.command, "ls-refs");
        session.reset_request();

        let closed = session.read_request().await.unwrap();
        assert!(closed.is_none(), "a clean close is not an error");
        let reads_frozen = session.transport_reads();
        assert!(session.read_request().await.is_err());
        assert_eq!(
            session.transport_reads(),
            reads_frozen,
            "after the clean close the transport is never re-entered"
        );
    }

    /// The flush-only request at the command position reads as the
    /// quiet session end (`Ok(None)`), not a fault: a real client
    /// sends the trailing flush before its close after the last
    /// round's response (captured live — upstream serve.c ends the
    /// session on it); afterward the read side is terminal.
    #[tokio::test]
    async fn a_flush_only_request_at_the_command_position_ends_the_session_quietly() {
        let (peer, mut session) = session();
        let driver = tokio::spawn(async move {
            let mut peer = peer;
            peer.write_all(&pkt(b"command=ls-refs\n")).await.unwrap();
            peer.write_all(FLUSH).await.unwrap();
            peer.write_all(FLUSH).await.unwrap();
            peer.shutdown().await.unwrap();
        });
        driver.await.unwrap();

        let parsed = session.read_request().await.unwrap().expect("a request");
        assert_eq!(parsed.command, "ls-refs");
        session.reset_request();

        let closed = session.read_request().await.unwrap();
        assert!(closed.is_none(), "the trailing flush ends the session");
        assert!(session.read_request().await.is_err(), "terminal afterward");
    }

    #[tokio::test]
    async fn malformed_pkt_lines_are_protocol_errors() {
        let (mut peer, mut session) = session();
        peer.write_all(b"zzzz").await.unwrap();

        let error = session.read_request().await.unwrap_err();
        match error {
            SubstrateError::Protocol(message) => {
                assert!(message.contains("malformed pkt-line"), "{message}");
            }
            other => panic!("decode errors are protocol-shaped: {other:?}"),
        }
        assert!(
            session.read_request().await.is_err(),
            "terminal protocol errors stay terminal"
        );
    }

    #[tokio::test]
    async fn client_err_lines_map_to_protocol_errors() {
        let (mut peer, mut session) = session();
        peer.write_all(&pkt(b"ERR unauthorized client"))
            .await
            .unwrap();
        peer.write_all(FLUSH).await.unwrap();

        let error = session.read_request().await.unwrap_err();
        match error {
            SubstrateError::Protocol(message) => {
                assert!(message.contains("ERR"), "{message}");
                assert!(message.contains("unauthorized client"), "{message}");
            }
            other => panic!("client ERR lines are band-mappable: {other:?}"),
        }
    }

    #[tokio::test]
    async fn a_request_without_a_command_line_is_a_protocol_error() {
        let (mut peer, mut session) = session();
        peer.write_all(&pkt(b"peel\n")).await.unwrap();
        peer.write_all(FLUSH).await.unwrap();

        let error = session.read_request().await.unwrap_err();
        match error {
            SubstrateError::Protocol(message) => {
                assert!(message.contains("no command"), "{message}");
            }
            other => panic!("a command-less request is out of grammar: {other:?}"),
        }
    }

    /// The per-request budget (`max_request_body`, the stateless
    /// enforcement's duplex twin): a request whose transport bytes pass
    /// the budget is terminal — fail-closed, never unbounded
    /// accumulation over the session wall clock.
    #[tokio::test]
    async fn a_request_over_the_max_request_body_budget_fails_closed() {
        let (peer, stream) = duplex(1 << 16);
        let mut session = Session::new(
            stream,
            Limits {
                max_request_body: 200,
                ..Limits::default()
            },
            Service::UploadPack,
        );
        let driver = tokio::spawn(async move {
            let mut peer = peer;
            for _ in 0..100 {
                peer.write_all(&pkt(b"peel\n")).await.unwrap();
            }
            peer.write_all(FLUSH).await.unwrap();
        });

        let error = session.read_request().await.unwrap_err();
        driver.await.unwrap();
        match &error {
            SubstrateError::Protocol(message) => {
                assert!(message.contains("max_request_body"), "{message}");
            }
            other => panic!("a budget breach is protocol-shaped: {other:?}"),
        }
        assert!(error.is_terminal());

        // The push request reader carries the same budget.
        let (peer, stream) = duplex(1 << 16);
        let mut session = Session::new(
            stream,
            Limits {
                max_request_body: 200,
                ..Limits::default()
            },
            Service::ReceivePack,
        );
        let driver = tokio::spawn(async move {
            let mut peer = peer;
            for _ in 0..100 {
                peer.write_all(&pkt(format!(
                    "{} {} refs/heads/x",
                    "0".repeat(40),
                    "0a".repeat(20)
                )
                .as_bytes()))
                    .await
                    .unwrap();
            }
            peer.write_all(FLUSH).await.unwrap();
        });

        let error = session.read_push_request().await.unwrap_err();
        driver.await.unwrap();
        match &error {
            SubstrateError::Protocol(message) => {
                assert!(message.contains("max_request_body"), "{message}");
            }
            other => panic!("a budget breach is protocol-shaped: {other:?}"),
        }
        assert!(error.is_terminal());
    }

    /// Under budget, a request parses in full — the budget is the
    /// rejection surface, not a request feature.
    #[tokio::test]
    async fn a_request_under_the_budget_still_parses() {
        let (mut peer, stream) = duplex(1 << 16);
        let mut session = Session::new(
            stream,
            Limits {
                max_request_body: 4000,
                ..Limits::default()
            },
            Service::UploadPack,
        );
        let mut request = Vec::new();
        request.extend_from_slice(&pkt(b"command=ls-refs\n"));
        for i in 0..20u32 {
            request.extend_from_slice(&pkt(format!("peel-{i}\n").as_bytes()));
        }
        request.extend_from_slice(FLUSH);
        peer.write_all(&request).await.unwrap();

        let parsed = session.read_request().await.unwrap().expect("a request");
        assert_eq!(parsed.command, "ls-refs");
        assert_eq!(parsed.args.len(), 0);
        assert_eq!(
            parsed.pre_lines.len(),
            20,
            "command line separate, 20 pre-lines"
        );
    }

    #[tokio::test]
    async fn writes_frame_text_error_and_flush_lines() {
        let (mut peer, mut session) = session();
        session.write_text_line(b"version 2").await.unwrap();
        session
            .write_text_line(b"object-format=sha1")
            .await
            .unwrap();
        session
            .write_error_band("repository not found")
            .await
            .unwrap();
        session.write_flush().await.unwrap();

        let packets = read_pkts(&mut peer, 4).await;
        assert_eq!(packets[0], b"version 2\n".to_vec());
        assert_eq!(packets[1], b"object-format=sha1\n".to_vec());
        assert_eq!(packets[2], b"ERR repository not found".to_vec());
        assert_eq!(packets[3], Vec::<u8>::new(), "the flush pkt-line");
    }

    /// The sideband sink caps band-1 chunks at 65000 bytes, emits the
    /// section flush exactly once (idempotent flush), and refuses data
    /// after the flush.
    #[tokio::test]
    async fn sideband_chunks_at_the_cap_and_flush_is_one_shot() {
        let (mut peer, mut session) = session();
        let sink = session.sideband().unwrap();

        let payload = vec![b'x'; 150_000];
        let driver = tokio::task::spawn_blocking(move || {
            let mut sink = sink;
            sink.write_all(&payload)
                .expect("the sink accepts any byte count");
            sink.flush().unwrap();
            sink.flush().unwrap();
            let refused = sink
                .write(b"after-flush")
                .expect_err("no data after the flush");
            assert_eq!(refused.kind(), io::ErrorKind::BrokenPipe);
            (sink.bytes_pushed(), sink)
        });

        let packets = read_pkts(&mut peer, 4).await;
        for (i, size) in [65000usize, 65000, 20000].into_iter().enumerate() {
            assert_eq!(packets[i][0], 1, "band 1");
            assert_eq!(packets[i].len() - 1, size, "chunk {i}");
        }
        assert!(packets[3].is_empty(), "the one-shot section flush");

        let (pushed, sink) = driver.await.unwrap();
        assert_eq!(pushed, 150_000);

        session.finish_sideband(sink).await.unwrap();
        // Exactly one flush ever reached the wire: nothing more is sent
        // after finish_sideband.
        let mut extra = [0u8; 1];
        let nothing = timeout(Duration::from_millis(150), peer.read_exact(&mut extra)).await;
        assert!(
            nothing.is_err(),
            "no bytes may follow the one-shot section flush"
        );
    }

    /// Exactly one driver per sink, prevented by construction: while
    /// the sink holds the response writer, a second sink and every
    /// write path fail loudly; `finish_sideband` restores the writer.
    #[tokio::test]
    async fn the_sideband_sink_exclusively_holds_the_writer() {
        let (mut peer, mut session) = session();
        let sink = session.sideband().unwrap();

        match session.sideband().unwrap_err() {
            SubstrateError::WriterTaken => {}
            other => panic!("no second sink while one lives: {other:?}"),
        }
        match session.write_text_line(b"blocked").await.unwrap_err() {
            SubstrateError::WriterTaken => {}
            other => panic!("the writer is exclusively the sink's: {other:?}"),
        }
        session.finish_sideband(sink).await.unwrap();

        session.write_text_line(b"restored").await.unwrap();
        session.write_flush().await.unwrap();
        let packets = read_pkts(&mut peer, 3).await;
        assert!(packets[0].is_empty(), "the response flush from the sink");
        assert_eq!(packets[1], b"restored\n".to_vec());
        assert!(packets[2].is_empty(), "the session flush");
    }

    /// A transport failure inside the drain task surfaces through
    /// `finish_sideband`, and the sink refuses further writes once the
    /// bounded channel's consumer is gone. The sink is driven from a
    /// blocking task — the machinery contract.
    #[tokio::test]
    async fn sideband_drain_failures_surface_through_finish() {
        let (peer, mut session) = session();
        let sink = session.sideband().unwrap();
        drop(peer);

        let driver = tokio::task::spawn_blocking(move || {
            let mut sink = sink;
            let chunk = vec![b'y'; SIDEBAND_CHUNK_SIZE];
            let mut errored = false;
            for _ in 0..32 {
                match sink.write(&chunk) {
                    Err(_) => {
                        errored = true;
                        break;
                    }
                    Ok(written) => {
                        assert!(written > 0, "a successful write accepts bytes");
                        std::thread::sleep(Duration::from_millis(5));
                    }
                }
            }
            assert!(errored, "the sink must surface the drain failure");
            sink
        });
        let sink = driver.await.unwrap();

        let error = session.finish_sideband(sink).await.unwrap_err();
        assert!(
            error.is_terminal(),
            "the drain failure is terminal: {error:?}"
        );
    }
}