alkgit 0.1.0

//! The receive-pack machinery (ADR-013 §3–11, as amended by ADR-020
//! §1): the request grammar and the push state machine — the parse
//! layer and the orchestration of one push invocation, mirroring
//! [`parse_fetch_args`](crate::parse_fetch_args) /
//! [`serve_fetch_round`](crate::serve_fetch_round).
//!
//! The grammar ([`parse_push_request`]) is substrate-agnostic: it
//! consumes the pkt-line strings the substrates' readers produce — the
//! same [`Request`](crate::Request)-shaped input the fetch parse
//! takes, where `pre_lines` are the pkt-lines before the request's
//! terminating flush (the command section: `<old> <new> <ref>` lines,
//! caps NUL-attached on the first line only) and `args` are the
//! pkt-lines after the command flush (the push-options section,
//! present only when the push-options cap was negotiated). No wire
//! writes; the parse stops at the command flush — the pack tail is
//! the machine's (the raw-tail handoff).
//!
//! The machine ([`serve_receive_pack`]) is the ADR-013 §7 observed
//! order over one parsed request, `parsed request → prepare →
//! validate → CAS → report`, and it stays substrate-agnostic outside
//! two tail disciplines ([`PackTailDiscipline`] — one core, the
//! duplex peek and the stateless EOF twin; the stateless POST body's
//! EOF is unambiguous, ADR-020 §1). Everything on a [`Session`]
//! beyond the machine is transport-mapping shaped: a returned
//! [`SubstrateError`] error-bands + closes on duplex and http-maps on
//! stateless (ADR-005 §4, the entries' mapping).

use std::collections::{HashMap, HashSet};
use std::io;
use std::sync::Arc;

use futures::io::AsyncReadExt;
use gix_hash::ObjectId;
use tokio::sync::Semaphore;

use crate::backend::{GitPackIngest, GitRefs};
use crate::duplex::{RawTail, Request};
use crate::fetch::{storage_to_session_error, ADVERTISED_OBJECT_FORMATS};
use crate::limits::Limits;
use crate::pack_source::{bridge_pack_source, PackSourceBridge};
use crate::refname::validate_ref_name;
use crate::registry::RepoRecord;
use crate::seam::{PushOptions, RefOutcome, RefUpdate};
use crate::substrate::SubstrateError;
use crate::v2::PktLineWriter;

/// The parsed receive-pack request (ADR-013 §3, ADR-020 §1):
///
/// - `commands` — the command lines (`<old> <new> <ref>`) resolved to
///   the zero-id shapes: `old=zero` = create ([`RefUpdate::expected`]
///   `None`), `new=zero` = delete ([`RefUpdate::new`] `None`), both
///   non-zero = an ordinary update. **An empty command set is the
///   client-side nothing-to-do — a distinct shape, not an error**
///   (ADR-013 §3: the machine replies flush, no report; see
///   [`PushRequest::is_nothing_to_do`]). Duplicate ref commands fail
///   closed at parse.
/// - `capabilities` — the client's selection (the parse accepts only
///   the advertised set plus `quiet`/`agent`; unknown and duplicate
///   capabilities fail closed).
/// - `options` — the parsed push-options section ([`PushOptions`],
///   the `(key, value)` pairs verbatim, repeated keys preserved);
///   `None` when the cap was not negotiated (v1: the gate is off —
///   never negotiated, the section is never parsed, the pack follows
///   the command flush directly, ADR-013 §11).
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct PushRequest {
    /// The command-derived ref updates, request order (the prepare/CAS
    /// inputs; ADR-013 §6–7).
    pub commands: Vec<RefUpdate>,
    /// The client's capability selection (ADR-013 §2; the accepted
    /// set is the advertised set plus `quiet`/`agent`).
    pub capabilities: PushCapabilities,
    /// The negotiated push-options section (ADR-013 §11) — `None`
    /// until the gate opens.
    pub options: Option<PushOptions>,
}

impl PushRequest {
    /// The client-side nothing-to-do shape (ADR-013 §3): an immediate
    /// flush with no commands — the machine replies flush, no report.
    pub fn is_nothing_to_do(&self) -> bool {
        self.commands.is_empty()
    }
}

/// The client's capability selection (ADR-013 §2, the parse side of
/// the served set): every bool is the corresponding cap having been
/// selected on the request's first command line (NUL-attached caps).
///
/// - `quiet` — accepted, honored by omission (we never send progress;
///   upstream clients select it habitually, push-captures.md §quiet
///   semantics). Not advertised.
/// - `agent` — recorded, uninterpreted.
/// - `push_options` — the client selected `push-options` (only
///   possible when the gate advertised it; v1: off, so selecting it
///   fails closed).
/// - The object-format check is validated at parse time against the
///   pinned advertisement list (ADR-013 pins the push-side check to
///   the fetch rule) and is not recorded here.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct PushCapabilities {
    /// `report-status` — the per-ref status report is wanted.
    pub report_status: bool,
    /// `report-status-v2` — the v2 report shape (identical for plain
    /// pushes: option lines exist only when the server emits them,
    /// ADR-013 §2).
    pub report_status_v2: bool,
    /// `delete-refs` — deletes are in scope (v1 allows deletes
    /// unconditionally, ADR-013 §7).
    pub delete_refs: bool,
    /// `side-band-64k` — the report rides band-1 sideband chunks
    /// (ADR-013 §8).
    pub side_band_64k: bool,
    /// `atomic` — the one-transaction CAS home makes atomic push
    /// correct (ADR-013 §10).
    pub atomic: bool,
    /// `ofs-delta` — accepted, honored by omission (the server never
    /// sends a report leg that depends on it).
    pub ofs_delta: bool,
    /// `push-options` — the client selected the negotiated cap
    /// (v1: the gate is off, fail closed).
    pub push_options: bool,
    /// `quiet` — accepted, honored by omission.
    pub quiet: bool,
    /// `agent=<value>` — recorded, uninterpreted.
    pub agent: Option<String>,
}

/// Parses a receive-pack request (ADR-013 §3, as amended by ADR-020
/// §1; grammar-exact against push-captures.md §Request):
///
/// - **Command lines**: `<old> <new> <ref>` — two oids in the
///   negotiated object format (sha1's 40 hex by default; the sha256
///   format's 64 hex under the feature) and the ref name, one command
///   per line. Capabilities ride NUL-attached on the **first command
///   line only** (observed: a leading space after the NUL); a caps
///   string on any later line fails closed. The ref name is parsed
///   structurally only — per-ref validation is the machine's
///   (ADR-013 §7's order, `validate_ref_name` after unpack).
/// - **Shallow rejection up front** (ADR-013 §4): `shallow <sha>`
///   lines anywhere before the pack are session errors — the clear
///   pkt-line error the entries surface before any pack byte is read.
/// - **The nothing-to-do shape** (ADR-013 §3): an immediate flush —
///   [`PushRequest::is_nothing_to_do`], a distinct shape, not an error.
/// - **The object-format check** (ADR-013 pins the push-side check to
///   the fetch rule, transport.md §fetch): validated against the
///   advertised list; absent assumes the pinned sha1.
/// - **The push-options section** (ADR-013 §11): parsed only when the
///   `push-options` cap was negotiated under the gate — bare pkt-lines
///   parsed to [`PushOptions`] verbatim; when not negotiated the parse
///   never consumes post-flush lines and any stray one fails closed
///   (the flush boundary is authoritative: the section's absence must
///   not be parsed as commands).
///
/// Unknown, duplicate, or out-of-grammar request lines are session
/// errors — fail closed (a client selects only from our advertisement,
/// so an unknown capability is a malformed or hostile request).
///
/// Contract stamp: ADR-013 §2–4, §11 (the request grammar, shallow
/// policy, push-options); ADR-020 §1 (the parse stops at the command
/// flush); push-captures.md §Request, §Advertisement.
pub fn parse_push_request(
    request: &Request,
    push_options_negotiated: bool,
) -> Result<PushRequest, SubstrateError> {
    if request.pre_lines.is_empty() {
        return Ok(PushRequest::default());
    }

    let mut commands: Vec<RefUpdate> = Vec::new();
    let mut seen_refs: HashSet<String> = HashSet::new();
    let mut capabilities = PushCapabilities::default();
    let mut object_format: Option<&str> = None;

    for (index, line) in request.pre_lines.iter().enumerate() {
        let (command_text, caps_text) = match line.split_once('\0') {
            Some((command_text, caps_text)) => {
                if index != 0 {
                    return Err(SubstrateError::Protocol(format!(
                        "push: line {index} carries a caps string — caps ride the first command line only"
                    )));
                }
                (command_text, caps_text)
            }
            None => (line.as_str(), ""),
        };
        if index == 0 {
            parse_capabilities(
                caps_text,
                push_options_negotiated,
                &mut capabilities,
                &mut object_format,
            )?;
        }
        if command_text == "shallow" || command_text.starts_with("shallow ") {
            return Err(SubstrateError::Protocol(
                "push: shallow update is not served — rejected up front (ADR-013 §4)".to_owned(),
            ));
        }
        let command = parse_command(command_text, object_format_hex_size(object_format))?;
        if !seen_refs.insert(command.name.clone()) {
            return Err(SubstrateError::Protocol(format!(
                "push: duplicate ref commands for {:?} in one push — fail closed",
                command.name
            )));
        }
        commands.push(command);
    }

    let options = if capabilities.push_options {
        Some(parse_push_options(&request.args)?)
    } else if !request.args.is_empty() {
        return Err(SubstrateError::Protocol(
            "push: post-command lines where no push-options section was negotiated — fail closed"
                .to_owned(),
        ));
    } else {
        None
    };

    Ok(PushRequest {
        commands,
        capabilities,
        options,
    })
}

/// Parses line 1's NUL-attached caps string (push-captures.md
/// §Advertisement — the client's subset of the served set, observed
/// with a leading space after the NUL): only the advertised set plus
/// `quiet`/`agent` are accepted; unknown or duplicate capabilities
/// fail closed. The `object-format=` value validates against the
/// advertisement list (ADR-013 pins the push-side check to the fetch
/// rule) and rides `object_format` out so the command oids' hex size
/// follows the negotiated format.
///
/// Contract stamp: ADR-013 §2 (the served set), §4 (the object-format
/// check rides the fetch rule); push-captures.md §Advertisement.
fn parse_capabilities<'a>(
    caps_text: &'a str,
    push_options_negotiated: bool,
    parsed: &mut PushCapabilities,
    object_format: &mut Option<&'a str>,
) -> Result<(), SubstrateError> {
    for token in caps_text.split_whitespace() {
        match token {
            "report-status" => set_flag(&mut parsed.report_status, token)?,
            "report-status-v2" => set_flag(&mut parsed.report_status_v2, token)?,
            "delete-refs" => set_flag(&mut parsed.delete_refs, token)?,
            "side-band-64k" => set_flag(&mut parsed.side_band_64k, token)?,
            "atomic" => set_flag(&mut parsed.atomic, token)?,
            "ofs-delta" => set_flag(&mut parsed.ofs_delta, token)?,
            "quiet" => set_flag(&mut parsed.quiet, token)?,
            "push-options" => {
                if !push_options_negotiated {
                    return Err(SubstrateError::Protocol(
                        "push: push-options was not advertised — the selection is out of grammar"
                            .to_owned(),
                    ));
                }
                set_flag(&mut parsed.push_options, token)?;
            }
            other => {
                if let Some(agent) = other.strip_prefix("agent=") {
                    if agent.is_empty() {
                        return Err(SubstrateError::Protocol(format!(
                            "push: {other:?} is an invalid agent capability line"
                        )));
                    }
                    if parsed.agent.is_some() {
                        return Err(SubstrateError::Protocol(format!(
                            "push: duplicate capability {other:?}"
                        )));
                    }
                    parsed.agent = Some(agent.to_owned());
                } else if let Some(format) = other.strip_prefix("object-format=") {
                    if object_format.is_some() {
                        return Err(SubstrateError::Protocol(
                            "push: repeated object-format capability".to_owned(),
                        ));
                    }
                    if !ADVERTISED_OBJECT_FORMATS.contains(&format) {
                        return Err(SubstrateError::Protocol(format!(
                            "push: object-format {format:?} was not advertised — mismatch rejected (transport.md §fetch)"
                        )));
                    }
                    *object_format = Some(format);
                } else {
                    return Err(SubstrateError::Protocol(format!(
                        "push: unknown capability {other:?} — fail closed"
                    )));
                }
            }
        }
    }
    Ok(())
}

/// Sets one boolean capability; a second occurrence is a duplicate —
/// fail closed (the fetch parse's posture).
fn set_flag(flag: &mut bool, token: &str) -> Result<(), SubstrateError> {
    if *flag {
        return Err(SubstrateError::Protocol(format!(
            "push: duplicate capability {token:?}"
        )));
    }
    *flag = true;
    Ok(())
}

/// The oid hex size the request's negotiated object format carries:
/// sha1's 40 (the pinned default when the cap is absent) or the
/// sha256 format's 64 (reachable only when the feature is enabled —
/// the advertisement list decides admission, not this size).
fn object_format_hex_size(object_format: Option<&str>) -> usize {
    if object_format == Some("sha256") {
        64
    } else {
        40
    }
}

/// Parses one command line `<old> <new> <ref>` (the upstream-faithful
/// split: two oids, then the ref name is the rest of the line) into
/// the zero-id shapes — `old=zero` = create
/// ([`RefUpdate::expected`] `None`), `new=zero` = delete
/// ([`RefUpdate::new`] `None`), both non-zero = an ordinary update.
/// Bad hex, a missing field, an empty ref name, and the both-zero
/// shape are session errors — fail closed.
///
/// Contract stamp: ADR-013 §3 (the command grammar, the zero-id
/// create/delete shapes); push-captures.md §Request (deletes:
/// old=current new=zero; creates: old=zero).
fn parse_command(line: &str, hex_size: usize) -> Result<RefUpdate, SubstrateError> {
    let mut fields = line.splitn(3, ' ');
    let old_hex = fields.next().unwrap_or_default();
    let new_hex = fields.next().unwrap_or_default();
    let name = fields.next().unwrap_or_default();
    if old_hex.is_empty() || new_hex.is_empty() || name.is_empty() {
        return Err(SubstrateError::Protocol(format!(
            "push: {line:?} is a malformed command line — <old> <new> <ref> expected"
        )));
    }
    let old = parse_oid(old_hex, hex_size, "old", line)?;
    let new = parse_oid(new_hex, hex_size, "new", line)?;
    let update = match (old.is_null(), new.is_null()) {
        (true, true) => {
            return Err(SubstrateError::Protocol(format!(
                "push: {line:?} carries the zero id on both sides — out of grammar"
            )))
        }
        (true, false) => RefUpdate {
            name: name.to_owned(),
            expected: None,
            new: Some(new),
        },
        (false, true) => RefUpdate {
            name: name.to_owned(),
            expected: Some(old),
            new: None,
        },
        (false, false) => RefUpdate {
            name: name.to_owned(),
            expected: Some(old),
            new: Some(new),
        },
    };
    Ok(update)
}

/// Parses one oid of the negotiated hex size; a size mismatch or bad
/// hex is a malformed command line — fail closed.
fn parse_oid(hex: &str, size: usize, what: &str, line: &str) -> Result<ObjectId, SubstrateError> {
    if hex.len() != size {
        return Err(SubstrateError::Protocol(format!(
            "push: {line:?} carries a malformed {what} oid {hex:?} — {size} hex characters expected"
        )));
    }
    ObjectId::from_hex(hex.as_bytes()).map_err(|error| {
        SubstrateError::Protocol(format!(
            "push: {line:?} carries a malformed {what} oid {hex:?}: {error}"
        ))
    })
}

/// Parses the negotiated push-options section (ADR-013 §11): bare
/// pkt-lines between the command flush and the pack, surfaced as the
/// `(key, value)` pairs verbatim and un-interpreted (the upstream
/// convention is `<key>=<value>`; a line without `=` is a key with an
/// empty value), repeated keys preserved. An empty line is malformed —
/// fail closed.
///
/// Contract stamp: ADR-013 §11 (the section's grammar; verbatim,
/// un-interpreted); push-captures.md §Request.
fn parse_push_options(lines: &[String]) -> Result<PushOptions, SubstrateError> {
    let mut pairs = Vec::with_capacity(lines.len());
    for line in lines {
        if line.is_empty() {
            return Err(SubstrateError::Protocol(
                "push: an empty push-option line is out of grammar".to_owned(),
            ));
        }
        let pair = match line.split_once('=') {
            Some((key, value)) => (key.to_owned(), value.to_owned()),
            None => (line.to_owned(), String::new()),
        };
        pairs.push(pair);
    }
    Ok(PushOptions { pairs })
}

/// The all-refs denial of an `unpack ng` leg (ADR-013 §8, the observed
/// raw-stdio shape): a pack-level failure reports `ng <ref> unpacker
/// error` for every ref — no per-ref reason survives a failed unpack.
/// Constant because the reasons clients display must be stable.
///
/// Contract stamp: ADR-013 §8 (per-ref `unpacker error` on pack-level
/// failure); push-captures.md §Raw stdio (corrupt pack), §Pack-read
/// ordering (definitive unpack-first order).
pub const UNPACKER_ERROR_REASON: &str = "unpacker error";

/// The aborted-transaction denial of an atomic push (ADR-013 §10,
/// observed live): on any denial of the whole-transaction CAS, the
/// refs that would otherwise have succeeded report
/// `ng <ref> atomic push failure` while the denied refs **keep their
/// own reasons** (captured: `ng refs/heads/dev deletion prohibited` +
/// `ng refs/heads/main atomic push failure` in one report). It maps
/// twice — inside the CAS (the outcome data) and on the machine's
/// name-validation abort branch — where the same upstream shape holds:
/// a name the validation rejects keeps its `funny refname`,
/// the valid updates report this.
///
/// Contract stamp: ADR-013 §10 (the atomic-abort mapping, verbatim);
/// push-captures.md §Status report (the captured pair); ADR-013 §7 (a
/// rejected refname keeps `funny refname`, the whole push aborts).
pub const ATOMIC_PUSH_FAILURE_REASON: &str = "atomic push failure";

const PACK_HEADER_BYTES: usize = 4;

/// The pack-tail discipline the machine serves (ADR-020 §1 — one
/// state-machine core, two tail disciplines): the two substrates
/// differ only in how the pack's absence is knowable below the
/// command flush.
///
/// - On a **duplex** transport the stream has no client-controlled
///   EOF below the flush (the client keeps its write side open until
///   the report — ADR-020 §3): the machine runs the **bounded peek**
///   within [`Limits::packless_peek`], timeout-classifying it (the
///   four ordering rules; rule (i) is already done by the grammar —
///   the options section was consumed from the request before the
///   tail ever surrenders), and after a packless report the tail
///   drains to EOF so the session's close is clean, never a
///   mid-stream RST (rule (iv)).
/// - On the **stateless** POST body EOF is unambiguous: the same
///   classification runs EOF-driven instead of timeout-driven (EOF at
///   the tail's start = the zero-byte packless case; EOF after a
///   partial header = rule (iii)'s malformed-push error), and no
///   drain runs — the body is either consumed or ended.
///
/// Contract stamp: ADR-020 §1 (the four ordering rules verbatim), §3
/// (the client's write-side-open shape); ADR-013 §3 (the tail's
/// unframed raw handoff); ADR-009 (the peek is a named budget kind —
/// `Limits::packless_peek` — never ambient behavior).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum PackTailDiscipline {
    /// A duplex transport's tail: the bounded peek window
    /// ([`Limits::packless_peek`]), drained after a packless report.
    Duplex,
    /// A stateless POST body's tail: EOF-driven, no drain.
    Stateless,
}

impl PackTailDiscipline {
    /// The peek's wall-clock bound for this discipline: the named
    /// [`Limits::packless_peek`] window on duplex; unbounded (EOF
    /// decides) on stateless.
    fn peek_window(self, limits: &Limits) -> Option<std::time::Duration> {
        match self {
            PackTailDiscipline::Duplex => Some(limits.packless_peek),
            PackTailDiscipline::Stateless => None,
        }
    }

    /// Whether the packless path drains the tail after the report
    /// (ADR-020 §1 rule (iv) — the duplex close's clean-close
    /// discipline; the stateless body needs nothing).
    fn drains_packless(self) -> bool {
        matches!(self, PackTailDiscipline::Duplex)
    }

    /// The peek's classification name for the machine's error/trace
    /// surface.
    fn discipline_name(self) -> &'static str {
        match self {
            PackTailDiscipline::Duplex => "duplex",
            PackTailDiscipline::Stateless => "stateless",
        }
    }
}

/// What the bounded peek observed below the command flush (ADR-020
/// §1's classification, the machine's branch point).
enum PackObservation {
    /// A full [`PACK_HEADER_BYTES`]-byte header: ingestion follows,
    /// the observed bytes prepend the bridged tail (the POC's
    /// reassembly shape — the peeked bytes are the stream's first
    /// bytes, never re-read or lost).
    Header(Vec<u8>),
    /// Zero pack bytes at the peek's boundary — the window timed out
    /// (duplex; rule (ii)) or the stream ended (the EOF twin, both
    /// disciplines): the packless, CAS-only path.
    Packless,
    /// A partial header byte string at the boundary: the malformed-
    /// push session error — never a silent packless downgrade
    /// (rule (iii)).
    Malformed(usize),
}

/// The peek's boundary classification shared by the two twin
/// disciplines (ADR-020 §1 rules (ii)/(iii)). Zero observed bytes is
/// the packless path; anything partial is malformed — a half-sent
/// pack cannot be re-interpreted as absence.
fn packless_or_malformed(observed: usize) -> PackObservation {
    if observed == 0 {
        PackObservation::Packless
    } else {
        PackObservation::Malformed(observed)
    }
}

/// Reads up to [`PACK_HEADER_BYTES`] bytes off the raw tail, bounded by
/// `window` when the discipline carries one (the duplex peek's window
/// bounds the whole gather, bytes observed so far survive the timeout
/// into [`PackObservation::Malformed`]); an early stream end shortens
/// the gather the same way on both disciplines — EOF with zero observed
/// bytes is packless, EOF after a partial header is malformed (the
/// client half-closing after a packless push is served without waiting
/// the window out — ADR-020 §3's client shape). A tail io error is the
/// session's terminal error.
///
/// The observed bytes are gathered byte-exactly and returned with the
/// [`PackObservation::Header`] outcome, so the bridge prepends them
/// with zero re-reads.
///
/// Contract stamp: ADR-020 §1 (the bounded peek, rules (ii)/(iii)),
/// §3 (the client's waiting shape); ADR-009 (the named window — the
/// only new budget kind of wave 3); push-captures.md (an ordinary
/// push's pack header is immediate, a packless client already
/// waiting).
async fn peek_pack_header<R>(
    tail: &mut RawTail<R>,
    window: Option<std::time::Duration>,
) -> Result<PackObservation, SubstrateError>
where
    R: futures::io::AsyncRead + Unpin,
{
    let deadline = window.map(|window| tokio::time::Instant::now() + window);
    let mut observed = Vec::with_capacity(PACK_HEADER_BYTES);
    let mut scratch = [0u8; 1];
    loop {
        if observed.len() == PACK_HEADER_BYTES {
            return Ok(PackObservation::Header(observed));
        }
        let read = tail.read(&mut scratch);
        let read = match deadline {
            Some(deadline) => match tokio::time::timeout_at(deadline, read).await {
                Ok(outcome) => outcome,
                Err(_elapsed) => return Ok(packless_or_malformed(observed.len())),
            },
            None => read.await,
        };
        match read {
            Ok(0) => return Ok(packless_or_malformed(observed.len())),
            Ok(n) => observed.extend_from_slice(&scratch[..n]),
            Err(error) => return Err(SubstrateError::Io(error)),
        }
    }
}

/// Drains the duplex tail to EOF after a packless report (ADR-020 §1
/// rule (iv)): the client keeps its write side open until the report
/// (§3) and closes after it, so the drain ends promptly — with the
/// stream read dry the session's close is clean, never a mid-stream
/// RST. A drain error is terminal (the transport broke closing). A
/// client that never closes is the entries' wall clock's victim, not
/// the machine's (the drain hangs on their session budget — the
/// machine has no wall clock of its own, ADR-009's enforcement split).
///
/// Contract stamp: ADR-020 §1 rule (iv), §3; ADR-005 §4 (io terminal);
/// ADR-009 (the wall clock is the entries').
async fn drain_pack_tail<R>(tail: &mut RawTail<R>) -> Result<(), SubstrateError>
where
    R: futures::io::AsyncRead + Unpin,
{
    let mut scratch = vec![0u8; 8192];
    loop {
        match tail.read(&mut scratch).await {
            Ok(0) => return Ok(()),
            Ok(_) => continue,
            Err(error) => return Err(SubstrateError::Io(error)),
        }
    }
}

/// Writes the status report when the client selected one (ADR-013 §8 —
/// `report-status`/`report-status-v2`; the shape is identical for both,
/// v2 adds option lines only when the server emits them, and v1 emits
/// none). Without the selection: nothing — no report, no flush (real
/// receive-pack's `report()` returns before any byte; the session ends
/// quietly). The framing (band-1 wrapped or bare) is the report
/// writer's, selected by the client's `side-band-64k`.
async fn write_report_maybe(
    writer: &mut impl PktLineWriter,
    unpack: &Result<(), String>,
    outcomes: &[RefOutcome],
    sideband: bool,
    wants_report: bool,
) -> Result<(), SubstrateError> {
    if !wants_report {
        return Ok(());
    }
    crate::v0::write_status_report(writer, unpack, outcomes, sideband).await
}

/// Places the CAS's per-ref [`RefOutcome`]s over the request-order
/// outcome slots the name validation left open (ADR-013 §8: the report
/// is in request order; the reasons frame verbatim — a stale ref's
/// server-chosen text and the atomic-abort mapping are the CAS's data,
/// never rewritten). A valid slot the CAS's return carries no outcome
/// for is fail-closed: the slot reports an error, never a silent ok.
fn merge_cas_outcomes(outcomes: &mut [RefOutcome], cas: Vec<RefOutcome>) {
    let mut by_name: HashMap<String, RefOutcome> = cas
        .into_iter()
        .map(|outcome| (outcome.name.clone(), outcome))
        .collect();
    for outcome in outcomes.iter_mut() {
        if outcome.result.is_ok() {
            *outcome = by_name.remove(&outcome.name).unwrap_or_else(|| RefOutcome {
                name: outcome.name.clone(),
                result: Err(format!(
                    "the ref CAS returned no outcome for {}; fail closed",
                    outcome.name
                )),
            });
        }
    }
}

/// The name-validate → CAS → report leg (ADR-013 §7–8, the observed
/// unpack-first-then-per-ref order — this leg runs only after the
/// unpack leg succeeded; ADR-020 §1's packless path runs it on the
/// parsed updates with no prepare call):
///
/// 1. **Per-ref name validation** ([`validate_ref_name`], ADR-013
///    §9): a rejected ref reports `ng <ref> funny refname` and is
///    **excluded from the CAS call** — the backend never sees an
///    invalid name (ADR-013 §9; the reason is the pinned observed
///    text).
/// 2. **The single CAS home**
///    ([`GitRefs::apply_updates`](crate::GitRefs::apply_updates),
///    ADR-013 §7): the valid updates in request order under one
///    transaction; `atomic` is the client's selected cap (ADR-013
///    §10). A `StorageError` is the session error (the
///    storage-to-session mapping); denials are outcomes.
/// 3. **The atomic abort's machine-side branch** (ADR-013 §10, the
///    live-captured shape): when `atomic` was negotiated and the name
///    validation rejected any ref, the transaction aborts whole — no
///    CAS call runs at all (a partial apply is precisely what atomic
///    forbids); the rejected refs keep `funny refname`, the valid ones
///    report [`ATOMIC_PUSH_FAILURE_REASON`].
///
/// The v1 policy leg is CAS-only: deletes are allowed and non-FF
/// updates are allowed (ADR-013 §7 — force-push is the client's
/// explicit choice, CAS exact-match decides).
///
/// Contract stamp: ADR-013 §7 (the order, the CAS-only policy), §8
/// (the report's request order + verbatim reasons), §9 (the name
/// check's home, exclusion from CAS), §10 (the abort mapping);
/// push-captures.md §CAS / race semantics, §Status report; ADR-020 §1
/// (the packless path runs this on the parsed updates).
async fn apply_and_report(
    writer: &mut impl PktLineWriter,
    updates: &[RefUpdate],
    refs: Arc<dyn GitRefs>,
    repo: &RepoRecord,
    atomic: bool,
    sideband: bool,
    wants_report: bool,
) -> Result<(), SubstrateError> {
    let mut outcomes: Vec<RefOutcome> = Vec::with_capacity(updates.len());
    let mut valid: Vec<RefUpdate> = Vec::with_capacity(updates.len());
    let mut rejected = false;
    for update in updates {
        match validate_ref_name(&update.name) {
            Ok(()) => {
                outcomes.push(RefOutcome {
                    name: update.name.clone(),
                    result: Ok(()),
                });
                valid.push(update.clone());
            }
            Err(reason) => {
                rejected = true;
                outcomes.push(RefOutcome {
                    name: update.name.clone(),
                    result: Err(reason),
                });
            }
        }
    }
    if atomic && rejected {
        for outcome in &mut outcomes {
            if outcome.result.is_ok() {
                outcome.result = Err(ATOMIC_PUSH_FAILURE_REASON.to_owned());
            }
        }
    } else if !valid.is_empty() {
        let cas = refs
            .apply_updates(repo, valid, atomic)
            .await
            .map_err(storage_to_session_error)?;
        merge_cas_outcomes(&mut outcomes, cas);
    }
    write_report_maybe(writer, &Ok(()), &outcomes, sideband, wants_report).await
}

/// Serves one receive-pack invocation (ADR-013 §3–8, as amended by
/// ADR-020 §1) — the push state machine, the substrate-agnostic core
/// the serving entries call with the parsed request's raw tail.
///
/// The tail is the raw-tail handoff the substrate surrenders after the
/// command flush
/// ([`Session::take_raw_tail`](crate::Session::take_raw_tail) on
/// duplex, [`StatelessRequest::take_raw_tail`](crate::StatelessRequest::take_raw_tail)
/// on stateless) — the pack streams there unframed, byte-exactly
/// continuing the flush (ADR-013 §3). `request` is the substrates'
/// parsed request shape: the command lines in `pre_lines`, the
/// push-options section (only when negotiated) in `args`.
///
/// The orchestration, in ADR-013 §7's observed order:
///
/// - **Up-front rejections**: the grammar runs first ([`parse_push_request`])
///   — a shallow line (ADR-013 §4), an object-format mismatch (§2/§4),
///   a duplicate or out-of-grammar line, or a bad command are session
///   errors before any pack byte is read or any ref is heard of. The
///   **nothing-to-do shape** (an immediate flush, ADR-013 §3) replies
///   flush with no report — no peek, no CAS.
/// - **The pack-arrival branch** ([`PackTailDiscipline`]): the bounded
///   peek classifies the tail — a full header proceeds to ingestion
///   (the peeked bytes prepend the bridged tail, zero loss); zero
///   bytes at the boundary (window timeout, or EOF) is the **packless,
///   CAS-only path** — the parsed updates skip prepare, go straight to
///   the validate-→-CAS leg (ADR-020 §1 rule (ii), applied per ADR-013
///   §7; the client stays open until the report); a partial header is
///   the **malformed-push session error** (rule (iii) — a half-sent
///   pack is never a silently-downgraded absence). The duplex packless
///   path drains the tail after the report (rule (iv)).
/// - **The ordinary path**: under the **pipeline-concurrency permit**
///   (ADR-009's enforcement point — the shared semaphore the fetch
///   machine's `generate` wraps too; exhaustion delays the call, a
///   closed semaphore is the 503-mapped session error class), the
///   bridge's blocking end carries the pack into
///   [`GitPackIngest::prepare`](crate::GitPackIngest::prepare) on
///   `spawn_blocking` under `Handle::block_on` (the machine owns the
///   blocking-thread posture, backend.md §concurrency model; the
///   wave-2 relay pattern) — a `StorageError` from prepare is the
///   session error, an `unpack` ng leg continues to the report.
///   The relay is joined after the ingestion finished with the end
///   (the break-before-join discipline), then the wire-side fault is
///   consulted: a budget breach or a transport failure on the pack
///   stream fails the session closed (ADR-009 — the client-fault
///   class), never a report over a broken stream.
/// - **The unpack-ng leg**: `unpack ng <reason>` + **all refs
///   `ng <ref> unpacker error`** — the observed raw-stdio shape —
///   with no CAS call (ADR-013 §8).
/// - **The CAS leg** (the `apply_and_report` leg): per-ref name validation,
///   then the single `apply_updates` call, then the report —
///   `unpack ok` + the per-ref outcomes framed verbatim in request
///   order through the report writer (sideband-selected framing,
///   ADR-013 §8). Report-free when the client selected neither
///   `report-status` cap (real receive-pack's shape).
///
/// Substrate-agnostic per convention 15: the writer is the
/// [`PktLineWriter`](crate::v2::PktLineWriter) surface, the tail is the
/// surrender state's raw reader — no duplex- or http-specific type
/// appears below the two disciplines; no cross-request state is held.
///
/// Contract stamp: ADR-013 §3–§11 (the state machine this implements),
/// ADR-020 §1 (the bounded peek's four rules, the packless CAS path),
/// §2 (the ingestion composition lives impl-side; the machine calls
/// the trait), §3 (the client's write-side-open shape), ADR-009 (the
/// permit, the pack-size relay budget, the peek window, fail-closed
/// breaches), ADR-005 (the substrate owns the report framing);
/// push-captures.md §Pack-read ordering (the definitive order),
/// §Status report, §CAS / race semantics; core/w3-push-machine (this
/// wave).
#[allow(clippy::too_many_arguments)]
pub async fn serve_receive_pack<R>(
    writer: &mut impl PktLineWriter,
    request: &Request,
    tail: RawTail<R>,
    push_options_negotiated: bool,
    refs: Arc<dyn GitRefs>,
    ingest: Arc<dyn GitPackIngest>,
    repo: &RepoRecord,
    limits: &Limits,
    tail_discipline: PackTailDiscipline,
    pipeline: Arc<Semaphore>,
) -> Result<(), SubstrateError>
where
    R: futures::io::AsyncRead + Unpin + Send + 'static,
{
    let parsed = parse_push_request(request, push_options_negotiated)?;

    let wants_report = parsed.capabilities.report_status || parsed.capabilities.report_status_v2;
    let sideband = parsed.capabilities.side_band_64k;
    let atomic = parsed.capabilities.atomic;

    if parsed.is_nothing_to_do() {
        return writer.write_flush().await;
    }

    let mut tail = tail;
    let observation = peek_pack_header(&mut tail, tail_discipline.peek_window(limits)).await?;
    match observation {
        PackObservation::Packless => {
            apply_and_report(
                writer,
                &parsed.commands,
                refs,
                repo,
                atomic,
                sideband,
                wants_report,
            )
            .await?;
            if tail_discipline.drains_packless() {
                drain_pack_tail(&mut tail).await?;
            }
            Ok(())
        }
        PackObservation::Malformed(observed) => Err(SubstrateError::Protocol(format!(
            "push: only {observed} of {PACK_HEADER_BYTES} pack-header bytes reached the peek window ({}) — a half-sent pack cannot be re-interpreted as absence: malformed push, session error (ADR-020 §1 rule (iii))",
            tail_discipline.discipline_name()
        ))),
        PackObservation::Header(header) => {
            use futures::io::AsyncReadExt as _;

            let permit = pipeline.clone().acquire_owned().await.map_err(|_| {
                SubstrateError::Protocol("the pipeline-concurrency semaphore is closed".to_owned())
            })?;
            let PackSourceBridge { end, fault, relay } = bridge_pack_source(
                futures::io::Cursor::new(header).chain(tail),
                limits,
            );
            let handle = tokio::runtime::Handle::current();
            let owned_repo = repo.clone();
            let owned_limits = *limits;
            let updates = parsed.commands;
            let push_options = parsed.options;
            let prepared = tokio::task::spawn_blocking(move || {
                let _permit = permit;
                handle.block_on(ingest.prepare(
                    &owned_repo,
                    updates,
                    Box::new(end),
                    push_options,
                    owned_limits,
                ))
            })
            .await
            .map_err(|join| {
                SubstrateError::Io(io::Error::other(format!(
                    "the pack ingestion task failed: {join}"
                )))
            })?
            .map_err(storage_to_session_error)?;
            if let Some(failure) = fault.take_failure() {
                // A wire-side pack failure (a budget breach — the
                // client-fault class; a transport failure — terminal)
                // fails the session closed, never a report over the
                // broken stream (ADR-009's fail-closed behavior).
                return Err(failure);
            }

            match &prepared.unpack {
                Ok(()) => {
                    apply_and_report(
                        writer,
                        &prepared.updates,
                        refs,
                        repo,
                        atomic,
                        sideband,
                        wants_report,
                    )
                    .await?
                }
                Err(_) => {
                    let outcomes: Vec<RefOutcome> = prepared
                        .updates
                        .iter()
                        .map(|update| RefOutcome {
                            name: update.name.clone(),
                            result: Err(UNPACKER_ERROR_REASON.to_owned()),
                        })
                        .collect();
                    write_report_maybe(writer, &prepared.unpack, &outcomes, sideband, wants_report)
                        .await?
                }
            }

            // The report is out, the blocking end is gone with its
            // task: the relay ends on its next send or at the tail's
            // EOF — which on both disciplines comes only after the
            // report (a healthy client's tail EOFs when the client
            // closes past it, ADR-020 §3) — so the join is the
            // machine's last act; the pre-report join of the wave-2
            // pattern's join-half would deadlock exactly that healthy
            // shape. A client that never closes parks the relay here:
            // the entries' wall clock is its bound (ADR-009).
            relay.await.map_err(|join| {
                SubstrateError::Io(io::Error::other(format!(
                    "the pack-source relay task failed: {join}"
                )))
            })?;
            Ok(())
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::duplex::Request;

    fn zero() -> String {
        "0".repeat(40)
    }

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

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

    fn parse(
        request: &Request,
        push_options_negotiated: bool,
    ) -> Result<PushRequest, SubstrateError> {
        parse_push_request(request, push_options_negotiated)
    }

    fn protocol_error(result: Result<PushRequest, SubstrateError>) -> String {
        match result {
            Err(SubstrateError::Protocol(message)) => message,
            Ok(_) => panic!("expected a protocol error"),
            Err(other) => panic!("expected a protocol error, got {other:?}"),
        }
    }

    #[test]
    fn ordinary_commands_parse_into_the_zero_id_shapes() {
        let parsed = parse(
            &request(
                &[
                    &format!(
                        "{} {} refs/heads/create\0report-status ofs-delta",
                        zero(),
                        hex_oid(0x0a)
                    ),
                    &format!("{} {} refs/heads/update", hex_oid(0x1b), hex_oid(0x2c)),
                    &format!("{} {} refs/heads/delete", hex_oid(0x3d), zero()),
                ],
                &[],
            ),
            false,
        )
        .unwrap();
        assert_eq!(
            parsed.commands,
            vec![
                RefUpdate {
                    name: "refs/heads/create".to_owned(),
                    expected: None,
                    new: Some(ObjectId::from([0x0a; 20]))
                },
                RefUpdate {
                    name: "refs/heads/update".to_owned(),
                    expected: Some(ObjectId::from([0x1b; 20])),
                    new: Some(ObjectId::from([0x2c; 20]))
                },
                RefUpdate {
                    name: "refs/heads/delete".to_owned(),
                    expected: Some(ObjectId::from([0x3d; 20])),
                    new: None,
                }
            ]
        );
        assert!(parsed.capabilities.report_status);
        assert!(parsed.capabilities.ofs_delta);
        assert!(!parsed.capabilities.report_status_v2);
        assert_eq!(parsed.options, None);
        assert!(!parsed.is_nothing_to_do());
    }

    #[test]
    fn the_capture_shape_caps_selection_parses() {
        // push-captures.md §Advertisement: a leading space after the NUL.
        let parsed = parse(
            &request(
                &[&format!(
                    "{} {} refs/heads/dev\0 report-status-v2 side-band-64k quiet object-format=sha1 agent=git/2.43.0",
                    zero(),
                    hex_oid(0x0a)
                )],
                &[],
            ),
            false,
        )
        .unwrap();
        assert!(parsed.capabilities.report_status_v2);
        assert!(parsed.capabilities.side_band_64k);
        assert!(parsed.capabilities.quiet);
        assert!(!parsed.capabilities.ofs_delta);
        assert_eq!(parsed.capabilities.agent.as_deref(), Some("git/2.43.0"));
    }

    #[test]
    fn an_immediate_flush_is_the_nothing_to_do_shape() {
        let parsed = parse(&request(&[], &[]), false).unwrap();
        assert!(parsed.is_nothing_to_do());
        assert!(parsed.commands.is_empty());
        assert_eq!(parsed.capabilities, PushCapabilities::default());
        assert_eq!(parsed.options, None);
    }

    #[test]
    fn caps_ride_the_first_command_line_only() {
        let message = protocol_error(parse(
            &request(
                &[
                    &format!(
                        "{} {} refs/heads/main\0report-status",
                        zero(),
                        hex_oid(0x0a)
                    ),
                    &format!(
                        "{} {} refs/heads/other\0quiet",
                        hex_oid(0x1b),
                        hex_oid(0x2c)
                    ),
                ],
                &[],
            ),
            false,
        ));
        assert!(message.contains("first command line"), "{message}");
    }

    #[test]
    fn unknown_and_duplicate_caps_and_empty_agent_fail_closed() {
        let message = protocol_error(parse(
            &request(
                &[&format!(
                    "{} {} refs/heads/main\0push-cert",
                    zero(),
                    hex_oid(0x0a)
                )],
                &[],
            ),
            false,
        ));
        assert!(message.contains("unknown capability"), "{message}");
        let message = protocol_error(parse(
            &request(
                &[&format!(
                    "{} {} refs/heads/main\0report-status report-status",
                    zero(),
                    hex_oid(0x0a)
                )],
                &[],
            ),
            false,
        ));
        assert!(message.contains("duplicate capability"), "{message}");
        let message = protocol_error(parse(
            &request(
                &[&format!(
                    "{} {} refs/heads/main\0agent=",
                    zero(),
                    hex_oid(0x0a)
                )],
                &[],
            ),
            false,
        ));
        assert!(message.contains("agent"), "{message}");
    }

    #[test]
    fn the_object_format_mismatch_is_rejected_and_drives_the_oid_size() {
        let message = protocol_error(parse(
            &request(
                &[&format!(
                    "{} {} refs/heads/main\0object-format=sha1024",
                    zero(),
                    hex_oid(0x0a)
                )],
                &[],
            ),
            false,
        ));
        assert!(message.contains("object-format"), "{message}");
        // A sha1 request carrying 64-hex oids is out of grammar, even
        // where from_hex would parse it (the sha256 feature's shape).
        let long = hex_oid(0x0a).repeat(2)[..64].to_owned();
        let message = protocol_error(parse(
            &request(
                &[&format!(
                    "{} {} refs/heads/main\0object-format=sha1",
                    zero(),
                    long
                )],
                &[],
            ),
            false,
        ));
        assert!(message.contains("40 hex"), "{message}");
    }

    #[cfg(feature = "sha256")]
    #[test]
    fn the_advertised_sha256_format_is_accepted_under_the_feature() {
        let long = "0".repeat(64);
        let new = hex_oid(0x0a).repeat(2)[..64].to_owned();
        let parsed = parse(
            &request(
                &[&format!(
                    "{long} {new} refs/heads/main\0object-format=sha256"
                )],
                &[],
            ),
            false,
        )
        .unwrap();
        match &parsed.commands[0] {
            RefUpdate {
                expected: None,
                new: Some(new),
                ..
            } => assert_eq!(new.as_bytes().len(), 32),
            other => panic!("expected a create, got {other:?}"),
        }
    }

    #[test]
    fn shallow_lines_are_rejected_up_front() {
        let shallow = String::from("shallow ") + &hex_oid(0x0a);
        for lines in [
            vec![
                shallow.clone(),
                format!("{} {} refs/heads/main", zero(), hex_oid(0x0a)),
            ],
            vec![
                format!("{} {} refs/heads/main", zero(), hex_oid(0x0a)),
                shallow,
            ],
            vec![String::from("shallow")],
        ] {
            let borrowed: Vec<&str> = lines.iter().map(String::as_str).collect();
            let message = protocol_error(parse(&request(&borrowed, &[]), false));
            assert!(message.contains("shallow"), "{message}");
        }
    }

    #[test]
    fn malformed_command_lines_fail_closed() {
        for line in [
            format!("{} {} ", zero(), hex_oid(0x0a)),
            format!("{} refs/heads/short", zero()),
            "zz zz refs/heads/badhex".to_owned(),
            "\0report-status".to_owned(),
            format!("{} {} refs/heads/main", zero(), String::from("0")),
        ] {
            let message = protocol_error(parse(&request(&[&line], &[]), false));
            assert!(
                message.contains("malformed") || message.contains("carries the zero id"),
                "{message}"
            );
        }
    }

    #[test]
    fn duplicate_ref_commands_fail_closed() {
        let message = protocol_error(parse(
            &request(
                &[
                    &format!("{} {} refs/heads/main", zero(), hex_oid(0x0a)),
                    &format!("{} {} refs/heads/main", hex_oid(0x1b), hex_oid(0x2c)),
                ],
                &[],
            ),
            false,
        ));
        assert!(message.contains("duplicate ref commands"), "{message}");
    }

    #[test]
    fn the_push_options_section_parses_only_when_negotiated() {
        let command = format!(
            "{} {} refs/heads/main\0report-status push-options",
            zero(),
            hex_oid(0x0a)
        );
        let commands = [command.as_str()];

        // Not advertised (the gate is off): selecting the cap fails closed.
        let message = protocol_error(parse(&request(&commands, &[]), false));
        assert!(message.contains("not advertised"), "{message}");

        // Negotiated: the section parses to the verbatim pairs,
        // repeated keys preserved.
        let parsed = parse(
            &request(&commands, &["private=yes", "bare-option", "private=no"]),
            true,
        )
        .unwrap();
        assert_eq!(
            parsed.options,
            Some(PushOptions {
                pairs: vec![
                    ("private".to_owned(), "yes".to_owned()),
                    ("bare-option".to_owned(), String::new()),
                    ("private".to_owned(), "no".to_owned()),
                ],
            })
        );

        // Negotiated: the section may be empty (the flush-only shape).
        let parsed = parse(&request(&commands, &[]), true).unwrap();
        assert_eq!(parsed.options, Some(PushOptions::default()));

        // Not negotiated: the parse never consumes post-flush lines —
        // a stray one fails closed (never parsed as commands).
        let message = protocol_error(parse(
            &request(
                &[&format!(
                    "{} {} refs/heads/main\0report-status",
                    zero(),
                    hex_oid(0x0a)
                )],
                &["private=yes"],
            ),
            false,
        ));
        assert!(message.contains("fail closed"), "{message}");

        // An empty push-option line is out of grammar.
        let message = protocol_error(parse(&request(&commands, &[""]), true));
        assert!(message.contains("empty push-option"), "{message}");
    }
}