alkgit 0.1.0

//! The consumer half's push workstream (ADR-017 §3 — gitoxide has no
//! send-pack; a mirror of ADR-013's server shapes at the same honesty
//! bar, POC-X finding 3): the V0 advertisement parsed client-side
//! ([`GitSession::push_advertisement`]), the hand-rolled send-pack
//! ([`GitSession::push`]) — command lines in the zero-id create/delete
//! shapes with the capabilities NUL-attached on the first line only,
//! the caller-supplied pack streamed **raw after the command flush**
//! (ADR-017 §4 — pack generation is not re-implemented; a packless
//! push is commands + flush + nothing) — and the status report parsed
//! in both framings (sideband band-1-wrapped and bare — ADR-013 §8)
//! to typed per-ref outcomes plus the `unpack ok|ng` leg.
//!
//! The push helpers are a deliberate mirror of the server's grammar
//! (ADR-023's honest-duplication posture; review 005 R-4's recorded
//! judgment: the two sides are direction-opposed — write vs parse —
//! over the same wire shapes, pinned independently by
//! push-captures.md). The push-options section is sent never
//! (ADR-013 §11's gate-off posture mirrored; the additive path
//! intact). The client riders: a packless push holds its write side
//! open until the report (ADR-020 §3 — a half-close turns the
//! server's bounded pack-header peek into an EOF error instead of the
//! packless path), and every push reads the full report — the
//! response side drained to its flush — before the session could drop
//! (ADR-023 §5's read-your-request RST discipline, client side). Push
//! rides its own stream (one service machine per connection,
//! ADR-023 §2): one push operation per session.
//!
//! One workstream one file (the construction+handshake workstream and
//! the shared drill helpers live in the `session` module — the flat
//! sibling shape, convention 14); the session's budget kinds and the
//! terminal error-break discipline are the `session` module's story.

use bstr::{BString, ByteSlice as _};
use gix_hash::ObjectId;
use gix_transport::client::{Capabilities, MessageKind, WriteMode};

use crate::seam::{RefOutcome, RefUpdate};
use crate::session::{
    drain_banded_payload, map_client_error, GitSession, SessionError, CLIENT_AGENT,
};
use crate::substrate::Service;

impl GitSession {
    /// The push service's V0 ref advertisement, parsed client-side
    /// (ADR-017 §3's first leg): the drained
    /// [`HandshakeResponse`](crate::HandshakeResponse)'s `refs` lines
    /// are the server-speaks-first ref list — the first line's
    /// NUL-attached caps moved to
    /// [`HandshakeResponse::capabilities`](crate::HandshakeResponse::capabilities)
    /// by gix's reader normalization — so each line parses `<oid>
    /// <ref>` to a [`PushAdvertisedRef`] row. The `capabilities^{}`
    /// zero-oid sentinel (an empty repo's only line) parses as an
    /// empty set; a sentinel beside ref lines, a zero-oid ref row, or
    /// a line the grammar cannot explain is a session error,
    /// fail-closed. The advertisement runs once per session (the
    /// handshake's cache serves the re-parse without wire re-entry).
    ///
    /// The session must be the receive-pack service; anything else is
    /// a local refusal (`push` never rides an upload-pack stream).
    ///
    /// Contract stamp: ADR-017 §3 (the V0 advertisement parse — the
    /// client mirror of ADR-013 §2's server shapes),
    /// push-captures.md §Advertisement (the normative shapes); ADR-020
    /// §3 (the caller holds the session open); core/w4-session-push.
    pub async fn push_advertisement(&mut self) -> Result<Vec<PushAdvertisedRef>, SessionError> {
        if let Some(cached) = &self.terminal {
            return Err(cached.clone());
        }
        if self.service != Service::ReceivePack {
            let error = SessionError::InvalidInput(format!(
                "push rides the receive-pack service; this session is {:?}",
                self.service
            ));
            self.terminal = Some(error.clone());
            return Err(error);
        }
        let handshake = self.handshake().await?;
        match parse_v0_advertisement(&handshake.refs) {
            Ok(refs) => Ok(refs),
            Err(error) => {
                self.terminal = Some(error.clone());
                Err(error)
            }
        }
    }

    /// The hand-rolled send-pack operation (ADR-017 §3 — gitoxide has
    /// no send-pack; a mirror of ADR-013's server shapes at the same
    /// honesty bar): the `commands` are the caller-computed ref
    /// updates ([`RefUpdate`] — `expected: None` = the zero-id create
    /// shape, `new: None` = the zero-id delete shape, both sides =
    /// an ordinary update; the caller computes the CAS old values from
    /// [`GitSession::push_advertisement`]'s rows), written as `<old> <new> <ref>`
    /// lines with the capabilities NUL-attached on the first line only
    /// ([`PushSendOptions`] names the caller-supplied selections within
    /// the ADR-013 §2 grammar: `report-status` always rides — the
    /// machine's typed outcome return is built on it — and `atomic`,
    /// `report-status-v2`, `side-band-64k`, `quiet`, and `agent` ride
    /// as the options carry them; `delete-refs` rides automatically
    /// when any command deletes). The push-options section is sent
    /// never (v1's gate-off posture mirrored — ADR-013 §11; the
    /// additive path intact).
    ///
    /// `pack` is the **caller-supplied pack stream** (ADR-017 §4 — pack
    /// generation is not re-implemented), streamed raw after the
    /// command flush; [`None`] is the packless push (commands + flush +
    /// nothing — the delete/create-only shape). The caller's stream is
    /// read to EOF; nothing re-frames its bytes.
    ///
    /// The status report then parses in the framing the client's own
    /// `side-band-64k` selection decides (ADR-013 §8's distinction,
    /// client-side: band-1-wrapped inner report, or bare pkt-lines) to
    /// the typed [`PushReport`] — the `unpack ok|ng` leg plus per-ref
    /// `ok|ng` outcomes. A report line the grammar cannot explain
    /// terminates the session, fail-closed; the peer's error band
    /// (band 3) carries its text into the typed error.
    ///
    /// **The client riders**: the write side is never shut down before
    /// the report — a packless push especially waits with the write
    /// side open until the report arrives (ADR-020 §3: a half-close
    /// turns the server's bounded pack-header peek into an EOF error
    /// instead of the packless path) — and the report is read **in
    /// full** to its flush before the operation returns, on every
    /// shape (ADR-023 §5's RST discipline — no close-with-unread-
    /// receive-data).
    ///
    /// Local grammar refusals ([`SessionError::InvalidInput`], fail
    /// before the wire): an empty command set, duplicate ref names, a
    /// ref name carrying the command grammar's framing bytes (space,
    /// NUL, LF), the both-zero command shape, a command oid outside
    /// the peer's advertised object format (the format the caller's
    /// oids carry is deduced from the oids themselves; sha256 rides
    /// only where the peer advertised it), and an off-grammar agent
    /// token.
    ///
    /// Budgets (ADR-017 §6, ADR-009 — no new kinds): the whole
    /// operation rides under the session's remaining wall clock; the
    /// response report is bounded by
    /// [`max_pack_size`](crate::Limits::max_pack_size) (the
    /// construction task's recorded mapping; the advertisement already
    /// rode the handshake's bounds). Breaches are typed and
    /// fail-closed; any session error is terminal (error-break) —
    /// later calls re-report without re-entry. The handshake runs
    /// implicitly first when absent.
    ///
    /// Push rides its own stream (one service machine per connection,
    /// ADR-023 §2): one push operation per session; the caller dials a
    /// fresh stream for the next push.
    ///
    /// Contract stamp: ADR-017 §3/§4/§6 (the hand-rolled client, the
    /// caller-supplied pack, the bounds), ADR-013 §2/§3/§8/§10 (the
    /// grammar the client mirrors — the caps line, the zero-id shapes,
    /// the report framings, the atomic cap), ADR-013 §11 (the
    /// push-options section sent never), ADR-020 §3 (the write-side-
    /// open rider), ADR-023 §2/§5 (push rides its own stream; the RST
    /// discipline), ADR-009 (the budget kinds, client-side),
    /// push-captures.md (the capture basis); POC-X finding 3 (the
    /// proven composition); core/w4-session-push.
    pub async fn push<S>(
        &mut self,
        commands: Vec<RefUpdate>,
        pack: Option<&mut S>,
        options: &PushSendOptions,
    ) -> Result<PushReport, SessionError>
    where
        S: tokio::io::AsyncRead + Unpin,
    {
        if let Some(cached) = &self.terminal {
            return Err(cached.clone());
        }
        if self.service != Service::ReceivePack {
            let error = SessionError::InvalidInput(format!(
                "push rides the receive-pack service; this session is {:?}",
                self.service
            ));
            self.terminal = Some(error.clone());
            return Err(error);
        }
        if commands.is_empty() {
            let error = SessionError::InvalidInput(
                "push: the command set is empty — the caller computes the diff and pushes \
                 only a nonzero one (nothing-to-do is the caller's decision)"
                    .to_owned(),
            );
            self.terminal = Some(error.clone());
            return Err(error);
        }
        if let Some(agent) = &options.agent {
            if agent.is_empty()
                || agent
                    .bytes()
                    .any(|byte| byte.is_ascii_whitespace() || byte == 0)
            {
                let error = SessionError::InvalidInput(format!(
                    "push: the agent token {agent:?} carries grammar-breaking bytes (whitespace/NUL)"
                ));
                self.terminal = Some(error.clone());
                return Err(error);
            }
        }
        let handshake = self.handshake().await?;
        let format = push_object_format(&commands, &handshake.capabilities)?;
        let remaining = self.remaining_wall_clock();
        match tokio::time::timeout(remaining, self.push_once(commands, pack, options, format)).await
        {
            Ok(inner) => {
                if let Err(error) = &inner {
                    self.terminal = Some(error.clone());
                }
                inner
            }
            Err(_elapsed) => Err(self.typed_wall_clock_breach()),
        }
    }

    async fn push_once<S>(
        &mut self,
        commands: Vec<RefUpdate>,
        mut pack: Option<&mut S>,
        options: &PushSendOptions,
        format: PushObjectFormat,
    ) -> Result<PushReport, SessionError>
    where
        S: tokio::io::AsyncRead + Unpin,
    {
        use futures::io::AsyncWriteExt as FuturesAsyncWriteExt;
        use gix_transport::client::async_io::Transport as _;

        let max_pack_size = self.limits.max_pack_size;
        let caps =
            push_capabilities_text(options, format, commands.iter().any(|c| c.new.is_none()));
        let zero = "0".repeat(format.hex_size());

        let mut writer = self
            .transport_mut()
            .request(
                WriteMode::OneLfTerminatedLinePerWriteCall,
                MessageKind::Flush,
                false,
            )
            .map_err(map_client_error)?;
        for (index, command) in commands.iter().enumerate() {
            let old = command
                .expected
                .map(|oid| oid.to_string())
                .unwrap_or_else(|| zero.clone());
            let new = command
                .new
                .map(|oid| oid.to_string())
                .unwrap_or_else(|| zero.clone());
            let mut line = format!("{old} {new} {}", command.name);
            if index == 0 {
                line.push('\0');
                line.push(' ');
                line.push_str(&caps);
            }
            FuturesAsyncWriteExt::write_all(&mut writer, line.as_bytes())
                .await
                .map_err(|error| {
                    SessionError::Transport(format!("the command section failed to write: {error}"))
                })?;
        }
        writer
            .write_message(MessageKind::Flush)
            .await
            .map_err(|error| {
                SessionError::Transport(format!("the command flush failed to write: {error}"))
            })?;
        FuturesAsyncWriteExt::flush(&mut writer)
            .await
            .map_err(|error| {
                SessionError::Transport(format!("the command section failed to flush: {error}"))
            })?;
        let (mut raw_writer, mut reader) = writer.into_parts();

        if let Some(pack) = pack.as_mut() {
            stream_caller_pack(pack, &mut raw_writer).await?;
        }
        drop(raw_writer);

        read_push_report(&mut reader, options.side_band_64k, max_pack_size).await
    }
}

/// One advertised ref row of the push service's V0 advertisement
/// ([`GitSession::push_advertisement`]): the advertisement's `<oid>
/// <ref>` line parsed (the caps NUL-attachment rode
/// [`HandshakeResponse::capabilities`](crate::HandshakeResponse::capabilities)).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct PushAdvertisedRef {
    /// The ref's current oid — the caller's CAS `expected` input.
    pub oid: ObjectId,
    /// The ref name.
    pub name: String,
}

/// The caller-supplied capability selections for [`GitSession::push`]
/// (ADR-017 §3: options **within the ADR-013 §2 grammar**). The
/// defaults are the plainest honest grammar: `report-status` only
/// (always sent — the machine's typed return is built on it), bare
/// report framing, non-atomic, the crate's own agent token.
///
/// The push-options cap is deliberately absent: the section is gated
/// off v1 and the client sends none (ADR-013 §11's gate-off posture
/// mirrored client-side; the additive path intact — a later gate-open
/// is an option field, not a protocol change).
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct PushSendOptions {
    /// `atomic` — the whole-transaction CAS (ADR-013 §10); a denial
    /// aborts every update (`ng <ref> atomic push failure` on the
    /// would-have-succeeded refs).
    pub atomic: bool,
    /// `report-status-v2` — request the v2 report shape (identical for
    /// plain pushes: v2 adds option lines only when the server emits
    /// them, and v1 emits none — ADR-013 §2). The default
    /// (`false`) requests plain `report-status`.
    pub report_status_v2: bool,
    /// `side-band-64k` — the report rides band-1 chunks whose payload
    /// is itself pkt-line-framed (ADR-013 §8's framing distinction;
    /// the machine parses both framings, selected by what it sent).
    pub side_band_64k: bool,
    /// `quiet` — suppress progress chatter (the peer decides; the
    /// server-side accepted-and-honored-by-omission shape mirrored).
    pub quiet: bool,
    /// The `agent=<value>` token (upstream's convention). [`None`]
    /// sends the crate agent (`agent=alkgit/<version>`). A value
    /// carrying whitespace or NUL is a local refusal.
    pub agent: Option<String>,
}

/// The push status report ([`GitSession::push`], ADR-013 §8's shapes
/// parsed): the `unpack ok|ng <reason>` leg plus the per-ref
/// `ok|ng` outcomes. The reasons are the peer's text, verbatim
/// (client-displayed; never rewritten).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct PushReport {
    /// The pack leg — [`Ok`] for `unpack ok`, the ng reason for
    /// `unpack ng <reason>` (a pack-level failure; the per-ref
    /// outcomes then carry the peer's mapped text, typically
    /// `unpacker error`).
    pub unpack: Result<(), String>,
    /// The per-ref outcomes, report order (`ok <ref>` / `ng <ref>
    /// <reason>`).
    pub refs: Vec<RefOutcome>,
}

/// The object format the pushed commands carry — deduced from the
/// command oids' own byte length (the caller's oids speak for
/// themselves; sha256 rides only where the peer advertised it,
/// OQ-05's passthrough policy client-side).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum PushObjectFormat {
    Sha1,
    Sha256,
}

impl PushObjectFormat {
    /// The oid hex size this format's wire lines carry.
    fn hex_size(self) -> usize {
        match self {
            PushObjectFormat::Sha1 => 40,
            PushObjectFormat::Sha256 => 64,
        }
    }

    /// The capability token the first command line advertises (the
    /// upstream send-pack echo discipline, client-side).
    fn token(self) -> &'static str {
        match self {
            PushObjectFormat::Sha1 => "sha1",
            PushObjectFormat::Sha256 => "sha256",
        }
    }
}

/// The object-format values the peer advertised (the capability's
/// echo check runs against this list).
fn advertised_object_formats(capabilities: &Capabilities) -> Vec<String> {
    capabilities
        .iter()
        .filter(|capability| capability.name() == b"object-format")
        .filter_map(|capability| {
            capability
                .value()
                .and_then(|value| value.to_str().ok().map(ToOwned::to_owned))
        })
        .collect()
}

/// Deduces the pushed commands' object format and validates the
/// command set's local grammar (fail before the wire): non-empty ref
/// names without the command grammar's framing bytes (space, NUL, LF),
/// no both-zero command shape, no duplicate ref names, one oid byte
/// size across the whole set (one push carries one format), and a
/// sha256 set only where the peer advertised the format.
fn push_object_format(
    commands: &[RefUpdate],
    advertised: &Capabilities,
) -> Result<PushObjectFormat, SessionError> {
    let mut size: Option<usize> = None;
    let mut seen = std::collections::HashSet::new();
    for command in commands {
        if command.name.is_empty()
            || command
                .name
                .bytes()
                .any(|byte| byte == b' ' || byte == 0 || byte == b'\n')
        {
            return Err(SessionError::InvalidInput(format!(
                "push: the ref name {:?} is off the command grammar (empty, or carries space/NUL/LF)",
                command.name
            )));
        }
        if command.expected.is_none() && command.new.is_none() {
            return Err(SessionError::InvalidInput(format!(
                "push: the command for {} carries the zero id on both sides — the \
                 create/delete shapes carry exactly one update side (ADR-013 §3)",
                command.name
            )));
        }
        if !seen.insert(command.name.clone()) {
            return Err(SessionError::InvalidInput(format!(
                "push: duplicate ref commands for {:?} in one push — fail closed before the wire",
                command.name
            )));
        }
        for oid in [command.expected, command.new].into_iter().flatten() {
            let oid_size = oid.as_bytes().len();
            match size {
                Some(existing) if existing != oid_size => {
                    return Err(SessionError::InvalidInput(
                        "push: the command set mixes object formats (20- and 32-byte oids) — \
                         one push carries one format (ADR-013 §2's object-format capability \
                         is a single value), fail closed"
                            .to_owned(),
                    ));
                }
                None => size = Some(oid_size),
                _ => {}
            }
        }
    }
    match size {
        Some(20) => Ok(PushObjectFormat::Sha1),
        Some(32) => {
            if !advertised_object_formats(advertised)
                .iter()
                .any(|value| value == "sha256")
            {
                return Err(SessionError::InvalidInput(
                    "push: the commands carry sha256 (32-byte) oids but the peer advertised \
                     no object-format=sha256 — fail closed before the wire"
                        .to_owned(),
                ));
            }
            Ok(PushObjectFormat::Sha256)
        }
        other => Err(SessionError::InvalidInput(format!(
            "push: the command oids carry {other:?}-byte ids — outside the sha1 (20) / \
             sha256 (32) grammar, fail closed"
        ))),
    }
}

/// Builds the first command line's NUL-attached capability string (the
/// caller's selections within the ADR-013 §2 grammar, the object
/// format the commands carry, `delete-refs` auto-riding on delete
/// commands, agent last): `report-status` always rides — the typed
/// report return is built on it.
fn push_capabilities_text(
    options: &PushSendOptions,
    format: PushObjectFormat,
    has_deletes: bool,
) -> String {
    let mut tokens = Vec::new();
    tokens.push(
        if options.report_status_v2 {
            "report-status-v2"
        } else {
            "report-status"
        }
        .to_owned(),
    );
    if options.side_band_64k {
        tokens.push("side-band-64k".to_owned());
    }
    if options.atomic {
        tokens.push("atomic".to_owned());
    }
    if has_deletes {
        tokens.push("delete-refs".to_owned());
    }
    tokens.push(format!("object-format={}", format.token()));
    if options.quiet {
        tokens.push("quiet".to_owned());
    }
    tokens.push(match &options.agent {
        Some(agent) => format!("agent={agent}"),
        None => format!("agent={CLIENT_AGENT}"),
    });
    tokens.join(" ")
}

/// Parses the drained V0 advertisement lines (the handshake's
/// normalization moved the first line's caps) into typed rows: `<oid>
/// <ref>` per line; the `capabilities^{}` zero-oid sentinel is the
/// empty repo's only line and parses as an empty set; a sentinel
/// beside ref lines, a zero-oid ref row, or a line the grammar cannot
/// explain is a fail-closed error.
fn parse_v0_advertisement(refs: &[BString]) -> Result<Vec<PushAdvertisedRef>, SessionError> {
    let invalid = |line: &[u8]| {
        SessionError::Protocol(format!(
            "the push advertisement line {:?} is not a V0 ref row ('<oid> <ref>') — fail closed",
            String::from_utf8_lossy(line)
        ))
    };
    let mut advertised = Vec::with_capacity(refs.len());
    let mut saw_sentinel = false;
    for line in refs {
        let text = String::from_utf8_lossy(line);
        let text = text.trim_end();
        let Some((oid_hex, name)) = text.split_once(' ') else {
            return Err(invalid(line));
        };
        let oid = match ObjectId::from_hex(oid_hex.as_bytes()) {
            Ok(oid) => oid,
            Err(_) => return Err(invalid(line)),
        };
        if name.is_empty() {
            return Err(invalid(line));
        }
        if name == "capabilities^{}" {
            if !oid.is_null() {
                return Err(SessionError::Protocol(format!(
                    "the push advertisement's capabilities^{{}} sentinel carries a non-zero \
                     oid {oid} — fail closed"
                )));
            }
            saw_sentinel = true;
            continue;
        }
        if saw_sentinel {
            return Err(SessionError::Protocol(
                "the push advertisement carries ref lines beside the capabilities^{} \
                 sentinel — one shape or the other, never both, fail closed"
                    .to_owned(),
            ));
        }
        if oid.is_null() {
            return Err(SessionError::Protocol(format!(
                "the push advertisement line {text:?} carries a zero oid — the zero id is the \
                 sentinel's shape alone, fail closed"
            )));
        }
        advertised.push(PushAdvertisedRef {
            oid,
            name: name.to_owned(),
        });
    }
    Ok(advertised)
}

/// Streams the caller's pack bytes raw into the request writer's
/// verbatim handle (ADR-017 §4 — the pack generation is the caller's;
/// the bytes ride re-framed, read to EOF).
async fn stream_caller_pack<S>(
    pack: &mut S,
    raw: &mut (impl futures::io::AsyncWrite + Unpin),
) -> Result<(), SessionError>
where
    S: tokio::io::AsyncRead + Unpin,
{
    use futures::io::AsyncWriteExt as _;
    use tokio::io::AsyncReadExt as _;

    let mut buffer = vec![0u8; 64 * 1024];
    loop {
        let read = pack.read(&mut buffer).await.map_err(|error| {
            SessionError::Transport(format!("the caller's pack stream failed to read: {error}"))
        })?;
        if read == 0 {
            return raw.flush().await.map_err(|error| {
                SessionError::Transport(format!("the pack stream failed to flush: {error}"))
            });
        }
        raw.write_all(&buffer[..read]).await.map_err(|error| {
            SessionError::Transport(format!("the pack stream failed to write: {error}"))
        })?;
    }
}

/// Reads and parses the push status report (ADR-013 §8, both
/// framings): the reads run to the framing's flush — the response side
/// fully drained before the operation returns (ADR-023 §5's RST
/// discipline) — with the report bytes accumulated under
/// `max_pack_size` (no byte past the budget is ever retained). In the
/// sideband framing the reader's progress handler unwraps band 1 (the
/// inner report is the band payload, itself pkt-line-framed) and the
/// peer's error band (band 3) carries its text into the typed error.
///
/// The drain loop is the `session` module's shared band-drain pump
/// (review 005 R-1's fold-in — identical shape as the fetch client's
/// done-round pack pump).
async fn read_push_report(
    reader: &mut Box<dyn gix_transport::client::async_io::ExtendedBufRead<'_> + Unpin + '_>,
    sideband: bool,
    max_pack_size: u64,
) -> Result<PushReport, SessionError> {
    let mut inner = Vec::new();
    drain_banded_payload(
        reader,
        &mut inner,
        sideband,
        max_pack_size,
        "the push report",
        "the push report",
    )
    .await?;
    parse_push_report(&inner, sideband)
}

/// Decodes the sideband framing's inner report (band-1 payloads are
/// themselves pkt-line-framed — the report lines and the inner flush
/// ride inside the band): one text line per inner frame, terminated by
/// the inner flush. Malformed inner framing is fail-closed.
fn framed_report_lines(payload: &[u8]) -> Result<Vec<String>, SessionError> {
    let invalid = |why: String| {
        SessionError::Protocol(format!(
            "the push report's inner sideband framing is malformed: {why} — fail closed"
        ))
    };
    let mut lines = Vec::new();
    let mut cursor = payload;
    while !cursor.is_empty() {
        if cursor.len() < 4 {
            return Err(invalid(format!(
                "truncated at a {}-byte pkt-line header",
                cursor.len()
            )));
        }
        let hex = std::str::from_utf8(&cursor[..4])
            .map_err(|_| invalid("the pkt-line header is not hex".to_owned()))?;
        let len = usize::from_str_radix(hex, 16)
            .map_err(|_| invalid(format!("the pkt-line header {hex:?} is not a hex length")))?;
        cursor = &cursor[4..];
        if len < 4 {
            break;
        }
        let take = len - 4;
        if cursor.len() < take {
            return Err(invalid(format!("truncated at a {take}-byte line body")));
        }
        let (line, rest) = cursor.split_at(take);
        cursor = rest;
        let text = line.strip_suffix(b"\n").unwrap_or(line);
        lines.push(String::from_utf8_lossy(text).into_owned());
    }
    Ok(lines)
}

/// Parses the report's lines into the typed [`PushReport`] (ADR-013
/// §8): the `unpack ok|ng <reason>` leg opens the report, then per-ref
/// `ok <ref>` / `ng <ref> <reason>` lines; reasons are the peer's
/// verbatim text. Any other shape is a session error, fail-closed.
fn parse_push_report(inner: &[u8], sideband: bool) -> Result<PushReport, SessionError> {
    let lines: Vec<String> = if sideband {
        framed_report_lines(inner)?
    } else {
        String::from_utf8_lossy(inner)
            .lines()
            .map(ToOwned::to_owned)
            .collect()
    };
    let mut lines = lines.into_iter().filter(|line| !line.is_empty());
    let unpack = match lines.next() {
        Some(line) if line == "unpack ok" => Ok(()),
        Some(line) if line.starts_with("unpack ng") => {
            let reason = line
                .strip_prefix("unpack ng")
                .and_then(|rest| rest.strip_prefix(' ').or(Some(rest)))
                .unwrap_or_default();
            Err(reason.to_owned())
        }
        Some(line) => {
            return Err(SessionError::Protocol(format!(
                "the push report opens with {line:?} — 'unpack ok|ng <reason>' expected, \
                 fail closed"
            )));
        }
        None => {
            return Err(SessionError::Protocol(
                "the push report carries no lines — fail closed".to_owned(),
            ));
        }
    };
    let mut refs = Vec::new();
    for line in lines {
        if let Some(name) = line.strip_prefix("ok ") {
            refs.push(RefOutcome {
                name: name.to_owned(),
                result: Ok(()),
            });
        } else if let Some(rest) = line.strip_prefix("ng ") {
            let (name, reason) = match rest.split_once(' ') {
                Some((name, reason)) => (name, reason),
                None => (rest, ""),
            };
            refs.push(RefOutcome {
                name: name.to_owned(),
                result: Err(reason.to_owned()),
            });
        } else {
            return Err(SessionError::Protocol(format!(
                "the push report line {line:?} is not a report line ('ok <ref>' / \
                 'ng <ref> <reason>') — fail closed"
            )));
        }
    }
    Ok(PushReport { unpack, refs })
}

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

    /// The push client's first-line capability string: `report-status`
    /// always rides, the caller's options ride within the ADR-013 §2
    /// grammar, `delete-refs` auto-rides on delete commands, the
    /// pushed format's object-format token, the agent token last.
    ///
    /// Contract stamp: ADR-013 §2 (the grammar the client speaks
    /// client-side), ADR-017 §3 (options within the grammar);
    /// core/w4-session-push.
    #[test]
    fn the_push_capabilities_line_is_the_grammar_the_client_speaks() {
        assert_eq!(
            push_capabilities_text(&PushSendOptions::default(), PushObjectFormat::Sha1, false),
            format!("report-status object-format=sha1 agent={CLIENT_AGENT}"),
            "the default: report-status only, the crate agent, bare framing"
        );
        let options = PushSendOptions {
            atomic: true,
            report_status_v2: true,
            side_band_64k: true,
            quiet: true,
            agent: Some("replicator/7".to_owned()),
        };
        assert_eq!(
            push_capabilities_text(&options, PushObjectFormat::Sha256, true),
            "report-status-v2 side-band-64k atomic delete-refs object-format=sha256 quiet \
             agent=replicator/7",
            "every option rides, delete-refs auto-ridden on deletes, format follows the oids"
        );
        assert!(
            !push_capabilities_text(&PushSendOptions::default(), PushObjectFormat::Sha1, false)
                .contains("push-options"),
            "the push-options section is sent never — the gate-off posture mirrored (ADR-013 §11)"
        );
    }

    /// The command set's local grammar (fail before the wire): one
    /// object format, legal ref names, the zero-id create/delete
    /// shapes carrying exactly one update side, no duplicates, and
    /// sha256 only where the peer advertised it.
    ///
    /// Contract stamp: ADR-013 §2/§3 (the grammar the client mirrors),
    /// ADR-017 §6 (fail-closed); core/w4-session-push.
    #[test]
    fn the_push_command_set_validates_before_the_wire() {
        let oid20 = |byte: u8| ObjectId::from([byte; 20]);
        let advertised_sha1 = Capabilities::from_lines("version 2\nobject-format=sha1\n".into())
            .expect("the fixture capability list");
        assert_eq!(
            push_object_format(
                &[RefUpdate {
                    name: "refs/heads/main".to_owned(),
                    expected: None,
                    new: Some(oid20(7)),
                }],
                &advertised_sha1,
            )
            .expect("the create shape"),
            PushObjectFormat::Sha1
        );
        assert_eq!(
            push_object_format(
                &[
                    RefUpdate {
                        name: "refs/heads/main".to_owned(),
                        expected: None,
                        new: Some(oid20(1)),
                    },
                    RefUpdate {
                        name: "refs/heads/dev".to_owned(),
                        expected: Some(oid20(1)),
                        new: Some(oid20(2)),
                    },
                    RefUpdate {
                        name: "refs/heads/old".to_owned(),
                        expected: Some(oid20(3)),
                        new: None,
                    },
                ],
                &advertised_sha1,
            )
            .expect("create + update + delete shapes"),
            PushObjectFormat::Sha1
        );
        for (why, commands) in [
            (
                "the both-zero command shape",
                vec![RefUpdate {
                    name: "refs/heads/void".to_owned(),
                    expected: None,
                    new: None,
                }],
            ),
            (
                "duplicate ref commands",
                vec![
                    RefUpdate {
                        name: "refs/heads/main".to_owned(),
                        expected: None,
                        new: Some(oid20(1)),
                    },
                    RefUpdate {
                        name: "refs/heads/main".to_owned(),
                        expected: Some(oid20(1)),
                        new: Some(oid20(2)),
                    },
                ],
            ),
            (
                "a name with grammar-breaking bytes",
                vec![RefUpdate {
                    name: "refs/heads/bad name".to_owned(),
                    expected: None,
                    new: Some(oid20(1)),
                }],
            ),
        ] {
            match push_object_format(&commands, &advertised_sha1) {
                Err(SessionError::InvalidInput(message)) => {
                    assert!(
                        message.contains("push:"),
                        "{why}: the refusal names the rule: {message}"
                    );
                }
                other => panic!("{why}: expected the local refusal, got {other:?}"),
            }
        }
        #[cfg(feature = "sha256")]
        {
            let oid32 = ObjectId::null(gix_hash::Kind::Sha256);
            let advertised_both = Capabilities::from_lines(
                "version 2\nobject-format=sha1\nobject-format=sha256\n".into(),
            )
            .expect("the fixture capability list");
            assert_eq!(
                push_object_format(
                    &[RefUpdate {
                        name: "refs/heads/main".to_owned(),
                        expected: None,
                        new: Some(oid32),
                    }],
                    &advertised_both,
                )
                .expect("sha256 rides where advertised"),
                PushObjectFormat::Sha256
            );
            match push_object_format(
                &[RefUpdate {
                    name: "refs/heads/main".to_owned(),
                    expected: None,
                    new: Some(oid32),
                }],
                &advertised_sha1,
            ) {
                Err(SessionError::InvalidInput(message)) => {
                    assert!(message.contains("sha256"), "{message}");
                }
                other => panic!("sha256 without the advertisement refuses, got {other:?}"),
            }
            match push_object_format(
                &[
                    RefUpdate {
                        name: "refs/heads/a".to_owned(),
                        expected: None,
                        new: Some(oid20(1)),
                    },
                    RefUpdate {
                        name: "refs/heads/b".to_owned(),
                        expected: None,
                        new: Some(oid32),
                    },
                ],
                &advertised_both,
            ) {
                Err(SessionError::InvalidInput(message)) => {
                    assert!(message.contains("mixes"), "{message}");
                }
                other => panic!("mixed formats refuse, got {other:?}"),
            }
        }
    }

    /// The V0 advertisement parse client-side: ref rows, the
    /// `capabilities^{}` sentinel as the empty repo, off-grammar rows
    /// fail closed.
    ///
    /// Contract stamp: ADR-013 §2 (the advertisement shapes the client
    /// mirrors), ADR-017 §3; push-captures.md §Advertisement;
    /// core/w4-session-push.
    #[test]
    fn the_v0_advertisement_parses_its_rows_and_the_sentinel() {
        let oid = |byte: u8| ObjectId::from([byte; 20]);
        assert_eq!(
            parse_v0_advertisement(&[]).expect("the empty drain is the empty repo"),
            Vec::new()
        );
        assert_eq!(
            parse_v0_advertisement(&[BString::from(format!(
                "{} capabilities^{{}}\n",
                "0".repeat(40)
            ))])
            .expect("the sentinel"),
            Vec::new(),
            "the empty repo's advertisement is the empty set"
        );
        assert_eq!(
            parse_v0_advertisement(&[
                BString::from(format!("{} HEAD\n", oid(1))),
                BString::from(format!("{} refs/heads/main\n", oid(2))),
            ])
            .expect("the ref rows"),
            vec![
                PushAdvertisedRef {
                    oid: oid(1),
                    name: "HEAD".to_owned(),
                },
                PushAdvertisedRef {
                    oid: oid(2),
                    name: "refs/heads/main".to_owned(),
                },
            ]
        );
        #[cfg(feature = "sha256")]
        {
            let sha256_zero_sentinel = format!("{} capabilities^{{}}\n", "0".repeat(64));
            assert_eq!(
                parse_v0_advertisement(&[BString::from(sha256_zero_sentinel)])
                    .expect("the sha256-sized sentinel oid"),
                Vec::new()
            );
        }
        for (why, lines) in [
            (
                "refs beside the sentinel",
                vec![
                    BString::from(format!("{} capabilities^{{}}\n", "0".repeat(40))),
                    BString::from(format!("{} refs/heads/x\n", oid(1))),
                ],
            ),
            (
                "a zero-oid ref row",
                vec![BString::from(format!("{} refs/heads/x\n", "0".repeat(40)))],
            ),
            ("a line without <oid> <ref>", vec![BString::from("HEAD\n")]),
            (
                "an undecodable oid",
                vec![BString::from("zz refs/heads/x\n")],
            ),
        ] {
            match parse_v0_advertisement(&lines) {
                Err(SessionError::Protocol(message)) => {
                    assert!(message.contains("fail closed"), "{why}: {message}");
                }
                other => panic!("{why}: expected the protocol error, got {other:?}"),
            }
        }
    }

    /// The status report parse, both framings (ADR-013 §8 client-side):
    /// bare report lines and the sideband framing's inner
    /// pkt-line-framed report (inner flush terminated) parse to the
    /// same typed outcomes; reasons ride verbatim; off-grammar lines
    /// and malformed inner framing fail closed.
    ///
    /// Contract stamp: ADR-013 §8 (the report shapes, verbatim
    /// reasons), ADR-017 §3 (parsed client-side);
    /// push-captures.md §Status report; core/w4-session-push.
    #[test]
    fn the_push_report_parses_both_framings() {
        let bare: &[u8] = b"unpack ok\nok refs/heads/main\nng refs/heads/dev funny refname\n";
        assert_eq!(
            parse_push_report(bare, false).expect("the bare report"),
            PushReport {
                unpack: Ok(()),
                refs: vec![
                    RefOutcome {
                        name: "refs/heads/main".to_owned(),
                        result: Ok(()),
                    },
                    RefOutcome {
                        name: "refs/heads/dev".to_owned(),
                        result: Err("funny refname".to_owned()),
                    },
                ],
            }
        );
        // The sideband framing: the inner report is itself pkt-line
        // framed (each line one frame, the inner flush terminates).
        let framed_inner = |texts: &[&str]| {
            let mut out = Vec::new();
            for text in texts {
                let payload = format!("{text}\n");
                out.extend_from_slice(format!("{:04x}", payload.len() + 4).as_bytes());
                out.extend_from_slice(payload.as_bytes());
            }
            out.extend_from_slice(b"0000");
            out
        };
        let sideband = framed_inner(&[
            "unpack ok",
            "ok refs/heads/main",
            "ng refs/heads/dev funny refname",
        ]);
        assert_eq!(
            parse_push_report(&sideband, true).expect("the band-wrapped report"),
            parse_push_report(bare, false).expect("the bare report"),
            "both framations carry the same report"
        );
        // The unpack ng leg's reason is verbatim; a reason-less ng line
        // carries the empty reason.
        let unpack_ng: &[u8] =
            b"unpack ng eof before pack header was fully read\nng refs/heads/x\n";
        assert_eq!(
            parse_push_report(unpack_ng, false).expect("the ng leg"),
            PushReport {
                unpack: Err("eof before pack header was fully read".to_owned()),
                refs: vec![RefOutcome {
                    name: "refs/heads/x".to_owned(),
                    result: Err(String::new()),
                }],
            }
        );
        for (why, payload, sideband) in [
            (
                "opens without the unpack leg",
                b"ok refs/heads/x\n" as &[u8],
                false,
            ),
            (
                "an off-grammar ref line",
                b"unpack ok\nhuh refs/heads/x\n" as &[u8],
                false,
            ),
            ("an entirely empty report", b"" as &[u8], false),
            ("malformed inner framing", b"0zz0unpack ok\n" as &[u8], true),
            ("truncated inner framing", b"0010unpack" as &[u8], true),
        ] {
            match parse_push_report(payload, sideband) {
                Err(SessionError::Protocol(message)) => {
                    assert!(message.contains("fail closed"), "{why}: {message}");
                }
                other => panic!("{why}: expected the protocol error, got {other:?}"),
            }
        }
    }
}