alkgit 0.1.0

//! The V2 fetch negotiation machine (ADR-014): the ack loop and the
//! done round, substrate-agnostic per ADR-005 §3 (both serving entries
//! feed the same machine through [`PktLineWriter`](crate::v2::PktLineWriter)).
//!
//! One call serves one fetch command round (negotiation-captures.md §
//! Round grammar: `command=fetch`, capability lines, `0001` delim, args,
//! `0000` flush — the substrate's parsed [`Request`]):
//!
//! - **Round grammar served** (ADR-014 §1): a no-`done` round is answered
//!   with the exact `acknowledgments` section — `ACK <oid>` per
//!   recognized have, `NAK` only when no ACKs in the round, flush
//!   terminator. `ready` is never sent (the DELIM variant exists only
//!   for `ready` responses — ours always ends at FLUSH), keeping
//!   `fetch=wait-for-done` honest (ADR-014 §4; the advertisement text
//!   is unchanged).
//! - **The ack rule is the seam's, verbatim** (ADR-014 §5): a have is
//!   recognized iff it exists in the object store and is a commit —
//!   [`GitPackGen::common_haves`](crate::GitPackGen::common_haves)'s
//!   predicate — and the wire layer acks exactly its output, adding no
//!   refinement (the protocol core is odb-free, ADR-010).
//! - **Want-less rounds are answered empty** (ADR-014 §3):
//!   `acknowledgments` + `NAK` + flush, no pack — the degenerate no-work
//!   shape, not an error.
//! - **The done round** (ADR-014 §2 as amended by review 001 A-4): the
//!   boundary set is the *recognized* subset — request haves filtered
//!   through `common_haves`, never an unverified have — then
//!   `generate(wants, boundary_haves, limits, sink)` streamed as the
//!   `packfile` section (`packfile` text line, band-1 sideband chunks,
//!   the section flush — the POC-1/POC-2 observed shape;
//!   negotiation-captures.md: no acknowledgments section, sideband +
//!   flush). An empty resulting pack is a valid zero-object packfile.
//!   A `generate` abort maps to the session's protocol error — the
//!   error-band write is the serving entries' (ADR-005 §4; the stateless
//!   substrate has no error band, ADR-005 §2) — and nothing of the
//!   section is emitted when the impl honored ADR-004's abort-first
//!   contract: never a broken pack downstream.
//! - **Budgets fail closed** (ADR-009, ADR-014 §6): every fetch round
//!   consumes the session's [`FetchRounds`] counter against
//!   [`Limits::max_negotiation_rounds`], and a round carrying more than
//!   [`Limits::max_haves_per_round`] haves errors before any backend
//!   call or wire byte. Breach returns the session error the entries
//!   map to error band + close (duplex) — no new budget kinds.
//! - **The pipeline-concurrency permit** (ADR-009's enforcement point,
//!   backend.md §concurrency model): the shared
//!   `Arc<tokio::sync::Semaphore>` — constructed from
//!   [`Limits::max_concurrent_blocking_tasks`] by the serving entries —
//!   is acquired around each `generate` call; exhaustion is the
//!   503-mapped session error class (ADR-009). Wave 3's push machine
//!   reuses the machinery around `prepare`.
//! - **Statelessness preserved** (ADR-014 §7): the machine holds no
//!   negotiation state — the client re-sends wants + commons every
//!   round. [`FetchRounds`] is budget accounting, not negotiation state.

use std::collections::HashSet;
use std::io;
use std::sync::Arc;

use gix_hash::ObjectId;
use tokio::sync::{mpsc, Semaphore};

use crate::backend::GitPackGen;
use crate::duplex::{Request, SIDEBAND_CHANNEL_CAPACITY};
use crate::error::StorageError;
use crate::limits::{Limits, SIDEBAND_CHUNK_SIZE};
use crate::registry::RepoRecord;
use crate::substrate::SubstrateError;
use crate::v2::PktLineWriter;

/// The object-format values the advertisement carries (ADR-003's honest
/// list: `object-format=sha1` pinned; the sha256 line rides the feature
/// flag per OQ-05's passthrough policy). The fetch command's
/// `object-format` check validates against exactly this set
/// (transport.md §fetch; ADR-013 pins the push-side check to the same
/// rule).
#[cfg(not(feature = "sha256"))]
pub(crate) const ADVERTISED_OBJECT_FORMATS: &[&str] = &["sha1"];
#[cfg(feature = "sha256")]
pub(crate) const ADVERTISED_OBJECT_FORMATS: &[&str] = &["sha1", "sha256"];

/// The parsed `command=fetch` request (transport.md §fetch; the
/// negotiation-captures.md round grammar): wants/haves/done and the
/// honored args, with the request's capability lines validated.
///
/// - `thin-pack`, `no-progress`, `ofs-delta`, and `include-tag` are
///   **accepted and honored by omission**: the generated packfile carries
///   no deltas and no progress bands (ADR-004 — loose objects ship as
///   compressed bases), and tag carrying is never synthesized — the want
///   closure is served as generated (tag ref tips ride as wants; a
///   configured client's tag backfill re-fetches tag refs directly), so
///   the args cost nothing to serve. Shallow/filter arguments
///   (`filter`, `shallow`, `deepen`, …) are unimplemented
///   surface and rejected — unknown arguments are session errors, fail
///   closed (the ls-refs machine's posture). The accepted set is the
///   OQ-19 rider-3 decisions (`core/w5-oq19-real-client-tokens`):
///   `include-tag` is un-gated in the real V2 client's send path
///   (every configured fetch sends it; there is no fetch= feature slot
///   to advertise or retract it in V2's grammar), so declining it would
///   break the real client the advertisement cannot inform —
///   accept-and-honor-by-omission, the thin-pack shape, validated
///   end-to-end (`tests/real_client.rs`).
/// - Wants and haves are deduplicated in first-occurrence order (the
///   client re-sends its common set every round; ADR-014 §7).
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct FetchArgs {
    /// The `want <oid>` lines — at least one on every working round (the
    /// want-less round is answered empty before this matters, ADR-014 §3).
    pub wants: Vec<ObjectId>,
    /// The `have <oid>` lines — the complete common + negotiator view for
    /// this round (negotiation-captures.md §Statelessness; ADR-014 §7).
    pub haves: Vec<ObjectId>,
    /// `done` — the client ended negotiation; the response is the
    /// generated packfile section, never an acknowledgments section
    /// (ADR-014 §2).
    pub done: bool,
    /// `thin-pack` — accepted, honored by omission (no deltas are
    /// synthesized, ADR-004).
    pub thin_pack: bool,
    /// `no-progress` — accepted, honored by omission (no progress bands
    /// are ever emitted).
    pub no_progress: bool,
    /// `ofs-delta` — accepted, honored by omission (no deltas are
    /// synthesized, ADR-004).
    pub ofs_delta: bool,
    /// `include-tag` — accepted, honored by omission: tag carrying is
    /// never synthesized; the served pack is the want closure as
    /// generated (the OQ-19 rider-3 decision — the real client's V2
    /// send path is un-gated on this argument, and there is no fetch=
    /// feature slot to advertise or retract it in V2's grammar;
    /// `tests/real_client.rs` validates the token survivable against
    /// real git on both substrate doors).
    pub include_tag: bool,
}

/// Parses `command=fetch` request lines (transport.md §fetch): the
/// capability lines (agent / object-format) and the argument lines.
///
/// The **object-format check** validates the request's `object-format`
/// line against the advertisement's set
/// (`ADVERTISED_OBJECT_FORMATS`, the ADR-003 honest list) on every
/// fetch command — absent
/// `object-format` assumes the pinned sha1 (the advertisement's
/// default); a mismatch rejects the command as a session error
/// (transport.md §fetch; ADR-013 pins the push-side check to the same
/// rule). Duplicate capability lines, unknown capability lines
/// (`session-id=` among them — git's V2 client sends it only when the
/// server advertises the `session-id` capability, and this
/// advertisement never declares it: the honest retraction, the OQ-19
/// rider-3 decision), an
/// empty `agent=`, repeated `done`, malformed oids, and unknown
/// arguments are session errors — fail closed.
pub fn parse_fetch_args(request: &Request) -> Result<FetchArgs, SubstrateError> {
    let mut agent_seen = false;
    let mut object_format: Option<&str> = None;
    for line in &request.pre_lines {
        if let Some(agent) = line.strip_prefix("agent=") {
            if agent_seen || agent.is_empty() {
                return Err(SubstrateError::Protocol(format!(
                    "fetch: {line:?} is an invalid agent capability line"
                )));
            }
            agent_seen = true;
        } else if let Some(format) = line.strip_prefix("object-format=") {
            if object_format.is_some() {
                return Err(SubstrateError::Protocol(
                    "fetch: repeated object-format capability line".to_owned(),
                ));
            }
            object_format = Some(format);
        } else {
            return Err(SubstrateError::Protocol(format!(
                "fetch: unexpected capability line {line:?}"
            )));
        }
    }
    if let Some(format) = object_format {
        if !ADVERTISED_OBJECT_FORMATS.contains(&format) {
            return Err(SubstrateError::Protocol(format!(
                "fetch: object-format {format:?} was not advertised — mismatch rejected (transport.md §fetch)"
            )));
        }
    }

    let mut parsed = FetchArgs::default();
    // Set-backed dedup, first-occurrence order (the client re-sends its
    // common set every round, ADR-014 §7): a `Vec::contains` scan would
    // be O(n²) over attacker-controlled request length — the
    // `max_haves_per_round` cap is applied only after the parse, so the
    // dedup must not amplify the unbounded parse input quadratically
    // (ADR-009's bounded-resources posture).
    let mut wants_seen: HashSet<ObjectId> = HashSet::new();
    let mut haves_seen: HashSet<ObjectId> = HashSet::new();
    for arg in &request.args {
        match arg.as_str() {
            "thin-pack" => parsed.thin_pack = true,
            "no-progress" => parsed.no_progress = true,
            "ofs-delta" => parsed.ofs_delta = true,
            "include-tag" => parsed.include_tag = true,
            "done" => {
                if parsed.done {
                    return Err(SubstrateError::Protocol(
                        "fetch: repeated done argument".to_owned(),
                    ));
                }
                parsed.done = true;
            }
            "want" | "have" => {
                return Err(SubstrateError::Protocol(format!(
                    "fetch: {arg:?} rides the <oid> on one line — the bare token is out of grammar"
                )));
            }
            other => {
                let (token, value) = match other.split_once(' ') {
                    Some(pair) => pair,
                    None => {
                        return Err(SubstrateError::Protocol(format!(
                            "fetch: unexpected argument {other:?}"
                        )));
                    }
                };
                let oid = ObjectId::from_hex(value.as_bytes()).map_err(|_| {
                    SubstrateError::Protocol(format!(
                        "fetch: {token:?} carries a malformed oid {value:?}"
                    ))
                })?;
                let slot = match token {
                    "want" => &mut parsed.wants,
                    "have" => &mut parsed.haves,
                    _ => {
                        return Err(SubstrateError::Protocol(format!(
                            "fetch: unexpected argument {other:?}"
                        )));
                    }
                };
                let seen = match token {
                    "want" => &mut wants_seen,
                    _ => &mut haves_seen,
                };
                if seen.insert(oid) {
                    slot.push(oid);
                }
            }
        }
    }
    Ok(parsed)
}

/// The negotiation-rounds budget counter (ADR-009, ADR-014 §6): one per
/// session (duplex) or per connection (stateless), held by the serving
/// entries and handed to [`serve_fetch_round`] — **this is budget
/// accounting, not negotiation state**: the client re-sends wants +
/// commons every round (ADR-014 §7), and the counter only bounds a
/// client that keeps sending rounds without progressing.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct FetchRounds {
    rounds_used: usize,
}

impl FetchRounds {
    /// Rounds consumed so far — observability for the serving entries.
    pub fn rounds_used(&self) -> usize {
        self.rounds_used
    }

    /// Consumes one fetch round against the cap; breach is a session
    /// error (fail-closed, ADR-009 + ADR-014 §6).
    fn consume(&mut self, cap: usize) -> Result<(), SubstrateError> {
        self.rounds_used = self.rounds_used.saturating_add(1);
        if self.rounds_used > cap {
            return Err(SubstrateError::Protocol(format!(
                "fetch: negotiation rounds budget of {cap} exceeded — fail closed (ADR-009, ADR-014 §6)"
            )));
        }
        Ok(())
    }
}

/// Writes the `acknowledgments` section (ADR-014 §1, the exact grammar):
/// the section header, `ACK <oid>` per recognized have, `NAK` only when
/// no ACKs in the round, flush terminator — never `ready` (ours always
/// ends at FLUSH, ADR-014 §4).
async fn write_acknowledgments(
    writer: &mut impl PktLineWriter,
    common: &[ObjectId],
) -> Result<(), SubstrateError> {
    writer.write_text_line(b"acknowledgments").await?;
    if common.is_empty() {
        writer.write_text_line(b"NAK").await?;
    } else {
        for oid in common {
            writer
                .write_text_line(format!("ACK {oid}").as_bytes())
                .await?;
        }
    }
    writer.write_flush().await
}

/// Maps a backend `StorageError` to the session's protocol error — the
/// entries band-map the result (ADR-005 §4). [`StorageError::ObjectMissing`]
/// Display carries the oid; the client computed it, so including it in
/// the reason is no oracle (backend.md §`StorageError`).
pub(crate) fn storage_to_session_error(error: StorageError) -> SubstrateError {
    SubstrateError::Protocol(error.to_string())
}

/// Serves one `command=fetch` round (ADR-014): budgets, the request
/// parse with the object-format check, the no-done acknowledgments
/// shape, and the done round's packfile section.
///
/// - `rounds` — the session's [`FetchRounds`] counter (the entries own
///   its state; this machine enforces the cap).
/// - `packgen` — the backend handle
///   [`GitPackGen::common_haves`] and [`GitPackGen::generate`] are
///   called through.
/// - `pipeline` — the shared pipeline-concurrency semaphore (ADR-009's
///   enforcement point): a permit is acquired around the done round's
///   `generate` call; exhaustion delays the call (the 503-mapped class).
///
/// Every response path is flush-terminated (acknowledgments / the
/// packfile section); an error path emits nothing beyond what was
/// already relayed and returns the terminal error the entries map.
pub async fn serve_fetch_round(
    writer: &mut impl PktLineWriter,
    request: &Request,
    rounds: &mut FetchRounds,
    packgen: Arc<dyn GitPackGen>,
    repo: &RepoRecord,
    limits: &Limits,
    pipeline: Arc<Semaphore>,
) -> Result<(), SubstrateError> {
    rounds.consume(limits.max_negotiation_rounds)?;
    let args = parse_fetch_args(request)?;
    if args.haves.len() > limits.max_haves_per_round {
        return Err(SubstrateError::Protocol(format!(
            "fetch: haves budget of {} exceeded by {} haves — fail closed (ADR-009, ADR-014 §6)",
            limits.max_haves_per_round,
            args.haves.len()
        )));
    }
    if args.wants.is_empty() {
        return write_acknowledgments(writer, &[]).await;
    }
    let common = packgen
        .common_haves(repo, args.haves.clone())
        .await
        .map_err(storage_to_session_error)?;
    if !args.done {
        return write_acknowledgments(writer, &common).await;
    }
    generate_packfile_section(writer, packgen, repo, args.wants, common, limits, pipeline).await
}

/// The done round's packfile section (ADR-014 §2): the `packfile` text
/// line, band-1 sideband chunks of the generated pack, the section flush
/// — no acknowledgments section (negotiation-captures.md's done-path
/// shape; POC-1/POC-2 observed).
///
/// The `packfile` header is written **lazily, on the first relayed
/// chunk**: a `generate` that aborts before any pack byte (ADR-004's
/// abort-first contract — the missing-object case) leaves the section
/// unopened, so the machine never emits a broken pack downstream — the
/// error surfaces as the machine's return value instead (the entries'
/// error-band mapping; this is the ADR-004 sideband-error-band shape).
///
/// Mechanics: the seam's blocking sink reaches `generate` through a
/// bounded bridge ([`BridgeSink`] — 65000-byte band chunks over a
/// bounded channel, the ADR-005 §1 bounded-sink posture) and the machine
/// relays chunks through [`PktLineWriter::write_data_band`] while the
/// pipeline runs on a blocking thread (the machine owns the
/// blocking-thread posture, ADR-009; backend.md §concurrency model —
/// the permit stays wire-side around the call, the impls' internal
/// `spawn_blocking` is their own detail).
async fn generate_packfile_section(
    writer: &mut impl PktLineWriter,
    packgen: Arc<dyn GitPackGen>,
    repo: &RepoRecord,
    wants: Vec<ObjectId>,
    boundary: Vec<ObjectId>,
    limits: &Limits,
    pipeline: Arc<Semaphore>,
) -> Result<(), SubstrateError> {
    use tokio::runtime::Handle;

    let (tx, mut rx) = mpsc::channel::<BridgeMsg>(SIDEBAND_CHANNEL_CAPACITY);
    let handle = Handle::current();
    let permit = pipeline.clone().acquire_owned().await.map_err(|_| {
        SubstrateError::Protocol("the pipeline-concurrency semaphore is closed".to_owned())
    })?;
    let owned_repo = repo.clone();
    let owned_limits = *limits;
    let generate = tokio::task::spawn_blocking(move || {
        let _permit = permit;
        let sink = BridgeSink {
            tx,
            buf: Vec::with_capacity(SIDEBAND_CHUNK_SIZE),
            flushed: false,
        };
        handle.block_on(packgen.generate(
            &owned_repo,
            wants,
            boundary,
            owned_limits,
            Box::new(sink),
        ))
    });

    enum RelayExit {
        /// The bridge closed after a normal (flush-terminated) relay.
        Complete,
        /// A writer failure mid-relay — the session error to return.
        Writer(SubstrateError),
    }

    let mut header_written = false;
    let mut section_flushed = false;
    let mut relay_exit = RelayExit::Complete;
    loop {
        let Some(message) = rx.recv().await else {
            break;
        };
        match message {
            BridgeMsg::Data(chunk) => {
                if section_flushed {
                    relay_exit = RelayExit::Writer(SubstrateError::Protocol(
                        "packfile section data after the section flush".to_owned(),
                    ));
                    break;
                }
                if !header_written {
                    writer.write_text_line(b"packfile").await?;
                    header_written = true;
                }
                writer.write_data_band(&chunk).await?;
            }
            BridgeMsg::Flush => {
                writer.write_flush().await?;
                section_flushed = true;
            }
        }
    }
    // Break the bridge before joining: with the receiver dropped, a
    // still-generating pipeline fails its next send instead of blocking
    // the permit holder forever.
    drop(rx);
    let exit = match relay_exit {
        RelayExit::Writer(error) => RelayExit::Writer(error),
        RelayExit::Complete => {
            match generate
                .await
                .map_err(|join| {
                    SubstrateError::Io(io::Error::other(format!(
                        "the pack generation task failed: {join}"
                    )))
                })
                .and_then(|generation| generation.map_err(storage_to_session_error))
            {
                Err(error) => RelayExit::Writer(error),
                Ok(()) => {
                    // Defensive: the generation seam's flush closed the
                    // section through the bridge; a seam that finished
                    // without it still leaves the section terminated.
                    if header_written && !section_flushed {
                        writer.write_flush().await?;
                    }
                    RelayExit::Complete
                }
            }
        }
    };
    match exit {
        RelayExit::Complete => Ok(()),
        RelayExit::Writer(error) => Err(error),
    }
}

/// The bridge's relayed messages (bounded channel — the ADR-005 §1
/// bounded-sink posture).
enum BridgeMsg {
    /// One ≤[`SIDEBAND_CHUNK_SIZE`] band-1 chunk.
    Data(Vec<u8>),
    /// The one-shot section flush.
    Flush,
}

/// The blocking-side bridge sink [`generate`](crate::GitPackGen::generate)
/// streams the pack into (the seam's `Box<dyn io::Write + Send>`): a
/// 65000-byte-chunked, one-shot-flush sink whose chunks ride the bounded
/// channel to the machine's relay loop — the ADR-005 §1 bounded-sink
/// posture (no unbounded buffer, backpressure when the relay falls
/// behind).
///
/// Constructed inside the machine's `spawn_blocking` pipeline task and
/// used only there: its sends park the pipeline thread on the bounded
/// channel (the blocking-posture requirement), and the send runs under
/// the task's `Handle::block_on` context — see [`BridgeSink::send`].
struct BridgeSink {
    tx: mpsc::Sender<BridgeMsg>,
    buf: Vec<u8>,
    flushed: bool,
}

impl BridgeSink {
    fn send(&mut self, message: BridgeMsg) -> io::Result<()> {
        // futures' thread-park block_on — not tokio's blocking_send:
        // the sink runs inside the pipeline task's Handle::block_on
        // runtime context, where blocking_send refuses nested blocking
        // (the wave-1 Sideband makes the same choice).
        futures::executor::block_on(self.tx.send(message))
            .map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "the bridge relay is gone"))
    }

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

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

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

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

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

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

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

    fn protocol_error<T>(result: Result<T, 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 parse_reads_wants_haves_done_and_flags() {
        let args = parse_fetch_args(&request(
            &["agent=git/2.43.0", "object-format=sha1"],
            &[
                "thin-pack",
                "no-progress",
                "ofs-delta",
                "include-tag",
                &format!("want {}", hex_oid(0x0a)),
                &format!("have {}", hex_oid(0x1b)),
                "done",
            ],
        ))
        .unwrap();
        assert!(args.thin_pack);
        assert!(args.no_progress);
        assert!(args.ofs_delta);
        assert!(args.include_tag);
        assert!(args.done);
        assert_eq!(args.wants, vec![oid(0x0a)]);
        assert_eq!(args.haves, vec![oid(0x1b)]);
    }

    #[test]
    fn parse_deduplicates_wants_and_haves_in_first_occurrence_order() {
        let args = parse_fetch_args(&request(
            &[],
            &[
                &format!("want {}", hex_oid(0x0a)),
                &format!("want {}", hex_oid(0x0a)),
                &format!("have {}", hex_oid(0x1b)),
                &format!("have {}", hex_oid(0x1b)),
                &format!("have {}", hex_oid(0x2c)),
                &format!("want {}", hex_oid(0x0a)),
            ],
        ))
        .unwrap();
        assert_eq!(args.wants, vec![oid(0x0a)]);
        assert_eq!(args.haves, vec![oid(0x1b), oid(0x2c)]);
    }

    #[test]
    fn an_absent_object_format_assumes_the_pinned_sha1() {
        let want = format!("want {}", hex_oid(0x0a));
        let args = parse_fetch_args(&request(&[], &[&want])).unwrap();
        assert_eq!(args.wants, vec![oid(0x0a)]);
        let args = parse_fetch_args(&request(&["object-format=sha1"], &[])).unwrap();
        assert_eq!(args, FetchArgs::default());
    }

    #[test]
    fn an_unadvertised_object_format_mismatch_is_rejected() {
        // sha1024 is unadvertised under either hash feature policy.
        let message = protocol_error(parse_fetch_args(&request(&["object-format=sha1024"], &[])));
        assert!(message.contains("object-format"), "{message}");
        assert!(message.contains("sha1024"), "{message}");
    }

    #[cfg(feature = "sha256")]
    #[test]
    fn the_advertised_sha256_format_is_accepted_under_the_feature() {
        let args = parse_fetch_args(&request(&["object-format=sha256"], &[])).unwrap();
        assert_eq!(args, FetchArgs::default());
    }

    #[test]
    fn a_repeated_done_argument_is_a_session_error() {
        let message = protocol_error(parse_fetch_args(&request(&[], &["done", "done"])));
        assert!(message.contains("done"), "{message}");
    }

    #[test]
    fn include_tag_is_accepted_and_honored_by_omission() {
        // The OQ-19 rider-3 decision: the real V2 client's send path for
        // `include-tag` is un-gated on the advertisement (every configured
        // fetch sends it), and there is no fetch= feature slot in V2's
        // grammar to advertise or retract it — the decline cannot inform a
        // conforming client, so the honest non-breaking posture is the same
        // honor-by-omission shape as thin-pack (tag carrying is never
        // synthesized; tests/real_client.rs validates the token against
        // real git on both substrate doors).
        let args = parse_fetch_args(&request(&[], &["include-tag"])).unwrap();
        assert!(args.include_tag);
    }

    #[test]
    fn unknown_fetch_arguments_are_session_errors() {
        for arg in ["shallow 0a", "deepen 1", "filter blob:none"] {
            let message = protocol_error(parse_fetch_args(&request(&[], &[arg])));
            assert!(
                message.contains(arg.split(' ').next().expect("token")),
                "{message}"
            );
        }
    }

    #[test]
    fn malformed_and_bare_oids_are_session_errors() {
        let message = protocol_error(parse_fetch_args(&request(&[], &["want zz"])));
        assert!(message.contains("malformed"), "{message}");
        let message = protocol_error(parse_fetch_args(&request(&[], &["have"])));
        assert!(message.contains("have"), "{message}");
    }

    #[test]
    fn unknown_and_duplicate_capability_lines_are_session_errors() {
        let message = protocol_error(parse_fetch_args(&request(&["shallow"], &[])));
        assert!(message.contains("capability"), "{message}");
        let message = protocol_error(parse_fetch_args(&request(&["agent=a", "agent=b"], &[])));
        assert!(message.contains("agent"), "{message}");
        let message = protocol_error(parse_fetch_args(&request(
            &["object-format=sha1", "object-format=sha1"],
            &[],
        )));
        assert!(message.contains("object-format"), "{message}");
        let message = protocol_error(parse_fetch_args(&request(&["agent="], &[])));
        assert!(message.contains("agent"), "{message}");
        // The session-id pin (the OQ-19 rider-3 decision — the honest
        // retraction): the advertisement never declares the capability, so
        // a conforming client never sends this pre-line; a pre-line that
        // reaches the machine anyway is a session error, fail closed.
        let message = protocol_error(parse_fetch_args(&request(
            &["session-id=20261023T000000Z-test"],
            &[],
        )));
        assert!(message.contains("session-id"), "{message}");
    }

    #[test]
    fn the_rounds_budget_fails_closed_at_the_cap() {
        let mut rounds = FetchRounds::default();
        rounds.consume(2).unwrap();
        rounds.consume(2).unwrap();
        let message = protocol_error(rounds.consume(2));
        assert!(message.contains("rounds budget"), "{message}");
        assert_eq!(rounds.rounds_used(), 3);
    }
}