grit-lib 0.5.18

//! Wire-protocol fetch orchestration over a live transport connection.
//!
//! [`fetch_remote`] drives `git-upload-pack` negotiation: match refspecs against
//! advertised refs, run `want`/`have`/`done`, demultiplex the side-band pack,
//! ingest it, and classify ref updates into a shared fetch outcome. Protocol v2
//! over streaming transports runs `ls-refs` before `fetch` when the connection
//! advertises no refs on connect.

use std::collections::HashSet;
use std::io::{Read, Write};
use std::path::Path;

use crate::environment::RepositoryOptions;
use crate::error::{Error, Result};
use crate::fetch_negotiator::SkippingNegotiator;
use crate::objects::ObjectId;
use crate::pkt_line;
use crate::protocol_v2;
use crate::refspec::RefspecItem;

type LsRefsAdvertised = (
    Vec<(String, ObjectId)>,
    Option<String>,
    std::collections::HashMap<String, ObjectId>,
    Option<ObjectId>,
);
use crate::progress::ProgressSink;
use crate::transfer::{
    classify_update, open_odb, prune_tracking_refs, ref_excluded, validate_fetch_depth_local,
    validate_fetch_depth_server, FetchOptions, FetchOutcome, RefUpdate, UpdateMode,
};
use crate::transport::Connection;

/// Sink for wire transport progress: typed phases via [`ProgressSink`] plus
/// raw side-band channel 2 bytes from the remote.
///
/// Implementations receive the server's progress bytes (typically `\r`-delimited
/// counter lines) through [`Progress::message`]. Typed
/// [`progress::ProgressPhase`](crate::progress::ProgressPhase) updates use the
/// [`crate::progress::ProgressSink`] methods inherited from this supertrait.
pub trait Progress: ProgressSink {
    /// Receive a chunk of progress bytes from side-band channel 2.
    fn message(&mut self, _bytes: &[u8]) {}

    /// When true, fetch omits `no-progress` so the remote may emit channel-2 progress.
    fn remote_sideband_progress(&self) -> bool {
        false
    }

    /// When true, a pack being received is also counted in objects and
    /// reported through [`Progress::received_objects`].
    ///
    /// Counting inflates every entry once more on a background thread (the
    /// receive itself never waits on it), so leave this off unless the counts
    /// are shown.
    fn wants_received_objects(&self) -> bool {
        false
    }

    /// Objects received so far in the pack being streamed in, reported during
    /// the [`ReceivingObjects`](crate::progress::ProgressPhase::ReceivingObjects)
    /// phase when [`Progress::wants_received_objects`] is true. Counts may lag
    /// the bytes reported through [`ProgressSink::inc`] slightly.
    fn received_objects(&mut self, _counts: crate::progress::ReceivedObjects) {}
}

/// A [`Progress`] / [`ProgressSink`] that discards everything.
pub struct NoProgress;

impl Progress for NoProgress {}

impl ProgressSink for NoProgress {}

// --- Negotiation flush schedule (mirrors fetch-pack.c) --------------------

const INITIAL_FLUSH: usize = 16;
const PIPESAFE_FLUSH: usize = 32;

fn next_flush_count(count: usize) -> usize {
    if count < PIPESAFE_FLUSH {
        count * 2
    } else {
        count + PIPESAFE_FLUSH
    }
}

#[derive(Clone, Copy, PartialEq, Eq)]
enum AckKind {
    /// `ACK <oid>` with no status suffix (post-`done` or legacy).
    Bare,
    Common,
    Continue,
    Ready,
}

fn parse_ack(line: &str) -> Option<(ObjectId, AckKind)> {
    if line == "NAK" {
        return None;
    }
    let rest = line.strip_prefix("ACK ")?;
    let hex = rest.split_whitespace().next()?;
    let oid = ObjectId::from_hex(hex).ok()?;
    let tail = rest.strip_prefix(hex).unwrap_or("").trim();
    let kind = if tail.contains("continue") {
        AckKind::Continue
    } else if tail.contains("common") {
        AckKind::Common
    } else if tail.contains("ready") {
        AckKind::Ready
    } else {
        AckKind::Bare
    };
    Some((oid, kind))
}

/// Read one ACK round, feeding `common`/`continue`/`ready` acks to the
/// negotiator. Lifted from `read_ack_round_with_negotiator`.
fn read_ack_round(reader: &mut dyn Read, negotiator: &mut SkippingNegotiator) -> Result<()> {
    let mut reader = reader;
    while let Some(pkt) = pkt_line::read_packet(&mut reader)? {
        match pkt {
            pkt_line::Packet::Flush => break,
            pkt_line::Packet::Data(ln) => {
                let ln = ln.trim_end();
                if ln == "NAK" {
                    // `upload-pack` sends `NAK` as the last line of a round with no trailing
                    // flush; waiting for another packet would block forever.
                    break;
                }
                let Some((ack_oid, kind)) = parse_ack(ln) else {
                    break;
                };
                if kind == AckKind::Bare {
                    break;
                }
                let _ = negotiator.ack(ack_oid)?;
            }
            _ => {}
        }
    }
    Ok(())
}

/// Demultiplex the side-band-64k stream after `done`: collect channel-1 pack
/// bytes into `out` (scanning for the `PACK` magic, which may span chunk
/// boundaries), and forward channel-2 progress to `progress`. Channel 3 is a
/// fatal error. Lifted from `read_sideband_pack_until_done`.
fn read_sideband_pack(
    r: &mut dyn Read,
    target: &mut crate::pack_receive::PackReceiveTarget<'_>,
    progress: &mut dyn Progress,
) -> Result<()> {
    crate::pack_receive::read_sideband_pack_to(r, target, progress)
}

/// New shallow boundaries the server reported during a fetch, captured from the
/// `shallow-info` section so [`fetch_remote`] (and the HTTP fetch paths) can
/// update the local `shallow` file and surface them in [`FetchOutcome`].
#[derive(Default)]
pub(crate) struct ShallowUpdate {
    pub(crate) shallow: Vec<ObjectId>,
    pub(crate) unshallow: Vec<ObjectId>,
}

/// Append the v0/v1 shallow/deepen request lines (after the `want`s, before the
/// terminating flush): the client's current `shallow <oid>` grafts and any
/// `deepen` / `deepen-since` / `deepen-not` the caller requested. Gated on the
/// matching server capability where one exists. Mirrors the CLI's
/// `append_fetch_request_extensions_v0_v1`.
fn append_filter_request_v0(req: &mut Vec<u8>, wire_filter: &str) -> Result<()> {
    pkt_line::write_line_to_vec(req, &format!("filter {wire_filter}"))?;
    Ok(())
}

fn append_shallow_request_v0(
    req: &mut Vec<u8>,
    server_caps: &str,
    local_shallow: &[ObjectId],
    opts: &FetchOptions,
) -> Result<()> {
    let sends_shallow = !local_shallow.is_empty()
        || opts.wire_deepen().is_some()
        || opts
            .deepen_since
            .as_deref()
            .is_some_and(|s| !s.trim().is_empty())
        || opts.deepen_not.iter().any(|s| !s.trim().is_empty());
    if sends_shallow && !server_caps.contains("shallow") {
        return Err(Error::FetchShallowUnsupported);
    }
    for oid in local_shallow {
        pkt_line::write_line_to_vec(req, &format!("shallow {}", oid.to_hex()))?;
    }
    if let Some((depth, relative)) = opts.wire_deepen() {
        if relative && !server_caps.contains("deepen-relative") {
            return Err(Error::FetchDeepenRelativeUnsupported);
        }
        pkt_line::write_line_to_vec(req, &format!("deepen {depth}"))?;
    }
    if let Some(since) = opts
        .deepen_since
        .as_deref()
        .filter(|s| !s.trim().is_empty())
    {
        if server_caps.contains("deepen-since") {
            let value = crate::shallow::deepen_since_wire_value(since);
            pkt_line::write_line_to_vec(req, &format!("deepen-since {value}"))?;
        }
    }
    if server_caps.contains("deepen-not") {
        for excl in &opts.deepen_not {
            let excl = excl.trim();
            if !excl.is_empty() {
                pkt_line::write_line_to_vec(req, &format!("deepen-not {excl}"))?;
            }
        }
    }
    Ok(())
}

/// Negotiate with `git-upload-pack` over the connection and return the raw
/// packfile bytes for the requested `wants`, plus any shallow-boundary updates
/// the server reported (`shallow`/`unshallow`).
///
/// Drives the [`SkippingNegotiator`] over the connection: sends `want` lines
/// (with v0/v1 capabilities) and the advertised refs as `known_common`, batches
/// local `have`s with flushes (reading interleaved ACK rounds), sends `done`,
/// consumes the final ACK/NAK, then demuxes the side-band pack.
///
/// When `opts` requests a deepen (or the repo is already shallow), the `want`
/// block carries the client's `shallow <oid>` grafts and the `deepen*` args, and
/// the server precedes the pack with a `shallow-info` section that this reads
/// into the returned [`ShallowUpdate`].
fn negotiate_pack(
    local_git_dir: &Path,
    conn: &mut dyn Connection,
    wants: &[ObjectId],
    opts: &mut FetchOptions,
    local_shallow: &[ObjectId],
    wire_filter: Option<&str>,
    progress: &mut dyn Progress,
) -> Result<(Option<std::path::PathBuf>, ShallowUpdate)> {
    crate::trace::trace_span!("grit.fetch.negotiate");
    let local_repo = crate::repo::Repository::open(local_git_dir, &RepositoryOptions::empty())?;
    let want_set: HashSet<ObjectId> = wants.iter().copied().collect();

    let Some(first_want) = wants.first().copied() else {
        return Ok((None, ShallowUpdate::default()));
    };

    // A deepen/shallow request changes the negotiation: the server precedes the
    // pack with a `shallow-info` section, and the client's local history is not a
    // usable negotiation base (its objects bottom out at grafts), so we skip
    // offering `have`s. Mirrors `fetch-pack.c`'s shallow handling.
    let shallow_request = opts.has_deepen_request() || !local_shallow.is_empty();

    // Capture the advertised refs before borrowing the writer (avoids aliasing
    // the connection's reader/writer with its accessors). v0/v1 shallow servers
    // append `shallow <oid>` trailer lines to the advertisement; the capability
    // string we read from the advertisement drives `deepen-since`/`deepen-not`.
    let advertised: Vec<(String, ObjectId)> = conn.advertised_refs().to_vec();
    let server_caps: String = conn.capabilities().join(" ");
    validate_fetch_depth_server(opts, Some(&server_caps), None)?;

    // Capability set matching `git fetch-pack`'s first `want` line for v0/v1.
    let mut caps = if opts.remote_progress {
        String::from(" multi_ack_detailed side-band-64k thin-pack include-tag ofs-delta agent=grit")
    } else {
        String::from(
            " multi_ack_detailed side-band-64k thin-pack no-progress include-tag ofs-delta agent=grit",
        )
    };
    if opts.wire_deepen().is_some_and(|(_, relative)| relative) {
        if !server_caps.contains("deepen-relative") {
            return Err(Error::FetchDeepenRelativeUnsupported);
        }
        caps.push_str(" deepen-relative");
    }
    if wire_filter.is_some() {
        caps.push_str(" filter");
    }

    let mut req: Vec<u8> = Vec::new();
    let w0 = format!("want {}{}", first_want.to_hex(), caps);
    pkt_line::write_line_to_vec(&mut req, &w0)?;
    for w in wants.iter().skip(1) {
        pkt_line::write_line_to_vec(&mut req, &format!("want {}", w.to_hex()))?;
    }
    // Match `git fetch-pack`: with a single unique OID, repeat the bare want.
    // git-daemon expects this. (Not done for shallow requests, which append
    // shallow/deepen lines instead.)
    if wants.len() == 1 && !shallow_request {
        pkt_line::write_line_to_vec(&mut req, &format!("want {}", first_want.to_hex()))?;
    }
    if let Some(spec) = wire_filter {
        append_filter_request_v0(&mut req, spec)?;
    }
    append_shallow_request_v0(&mut req, &server_caps, local_shallow, opts)?;
    req.extend_from_slice(b"0000");
    conn.writer().write_all(&req)?;
    conn.writer().flush()?;

    // Build the negotiator from local ref tips (heads, tags, HEAD), peeled to
    // commits, excluding the wants. Advertised tips we already have become
    // `known_common`.
    let mut negotiator = SkippingNegotiator::new(local_repo);
    let mut tips: Vec<ObjectId> = Vec::new();
    let mut seen_tip: HashSet<ObjectId> = HashSet::new();
    for prefix in ["refs/heads/", "refs/tags/"] {
        if let Ok(entries) = crate::refs::list_refs(local_git_dir, prefix) {
            for (_, oid) in entries {
                if let Ok(Some(c)) = negotiator.repo().odb().peel_to_commit(oid) {
                    if !want_set.contains(&c) && seen_tip.insert(c) {
                        tips.push(c);
                    }
                }
            }
        }
    }
    if let Ok(h) = crate::refs::resolve_ref(local_git_dir, "HEAD") {
        if let Ok(Some(c)) = negotiator.repo().odb().peel_to_commit(h) {
            if !want_set.contains(&c) && seen_tip.insert(c) {
                tips.push(c);
            }
        }
    }
    tips.sort_by_key(ObjectId::to_hex);
    if !shallow_request {
        for t in tips {
            negotiator.add_tip(t)?;
        }
        // Seed `known_common` only for advertised refs we already track at the same
        // peeled commit. Unchanged refs avoid walking parent chains for every server
        // tip during incremental fetch; moved refs are discovered via `have`s.
        let mut local_by_name: std::collections::HashMap<String, ObjectId> =
            std::collections::HashMap::new();
        for prefix in ["refs/heads/", "refs/tags/"] {
            if let Ok(entries) = crate::refs::list_refs(local_git_dir, prefix) {
                for (name, oid) in entries {
                    local_by_name.insert(name, oid);
                }
            }
        }
        for (name, oid) in &advertised {
            if want_set.contains(oid) {
                continue;
            }
            let Some(local_oid) = local_by_name.get(name) else {
                continue;
            };
            if local_oid != oid {
                continue;
            }
            if let Ok(Some(c)) = negotiator.repo().odb().peel_to_commit(*oid) {
                negotiator.known_common(c)?;
            }
        }
    }

    // Shallow-info section: for a deepen/shallow request the v0/v1 server emits
    // its `shallow`/`unshallow` lines (flush-terminated) immediately after the
    // wants block, before any ACK round. Read it now so the subsequent ACK/NAK
    // and pack reads line up.
    let mut shallow_update = ShallowUpdate::default();
    if shallow_request {
        let (sh, unsh) = crate::shallow::read_shallow_info_section(&mut conn.reader())?;
        shallow_update.shallow = sh;
        shallow_update.unshallow = unsh;
    } else if conn.stateless_rpc() {
        // Stateless HTTP completes a POST for the want block; consume its response
        // before the first have round posts again (otherwise the next POST drains it).
        read_ack_round(conn.reader(), &mut negotiator)?;
    }

    // Have/ACK exchange: batch haves, flush, read interleaved ACK rounds.
    let mut count: usize = 0;
    let mut flush_at: usize = INITIAL_FLUSH;
    let mut pending: Vec<u8> = Vec::new();
    let mut flushes: i32 = 0;
    while let Some(oid) = negotiator.next_have()? {
        pkt_line::write_line_to_vec(&mut pending, &format!("have {}", oid.to_hex()))?;
        count += 1;
        if flush_at <= count {
            pending.extend_from_slice(b"0000");
            conn.writer().write_all(&pending)?;
            conn.writer().flush()?;
            pending.clear();
            flush_at = next_flush_count(count);
            flushes += 1;
            // Duplex: keep one window ahead. Stateless: every POST must be read
            // before the next POST replaces the response.
            if count == INITIAL_FLUSH && !conn.stateless_rpc() {
                continue;
            }
            read_ack_round(conn.reader(), &mut negotiator)?;
            flushes -= 1;
        }
    }
    if !pending.is_empty() {
        pending.extend_from_slice(b"0000");
        conn.writer().write_all(&pending)?;
        conn.writer().flush()?;
        flushes += 1;
    }
    while flushes > 0 {
        read_ack_round(conn.reader(), &mut negotiator)?;
        flushes -= 1;
    }

    // Send `done` (single pkt-line, no trailing flush) and read the ACK/NAK.
    let mut tail = Vec::new();
    pkt_line::write_line_to_vec(&mut tail, "done")?;
    conn.writer().write_all(&tail)?;
    conn.writer().flush()?;

    match pkt_line::read_packet(&mut conn.reader())? {
        None => return Err(Error::Message("unexpected EOF after done".to_owned())),
        Some(pkt_line::Packet::Flush) => {
            return Err(Error::Message("unexpected flush after done".to_owned()))
        }
        Some(pkt_line::Packet::Data(ln)) => {
            let ln = ln.trim_end();
            if ln != "NAK" {
                if let Some((ack_oid, kind)) = parse_ack(ln) {
                    if kind != AckKind::Bare {
                        let _ = negotiator.ack(ack_oid)?;
                    }
                } else if let Some(msg) = ln.strip_prefix("ERR ") {
                    return Err(Error::Message(format!("remote error: {}", msg.trim_end())));
                }
            }
        }
        Some(_) => {}
    }

    let local_odb = open_odb(local_git_dir);
    let mut pack_receive = crate::pack_receive::TempPackReceive::create(local_odb.objects_dir())?;
    pack_receive.read_sideband(conn.reader(), progress)?;
    let pack_path = pack_receive.finish()?;
    Ok((pack_path, shallow_update))
}

// ===========================================================================
// Protocol v2 (streaming transports: git://, ssh)
// ===========================================================================
//
// A v2 connection advertises no refs on connect (only the capability block).
// `v2_ls_refs` recovers the ref map with a `command=ls-refs`; `negotiate_pack_v2`
// runs the `command=fetch` negotiation and returns the demuxed pack. Both lift
// the exact pkt-line shapes from the CLI's `file_upload_pack_v2` /
// `fetch_transport` v2 paths and reuse `protocol_v2` cap helpers, the shared
// `SkippingNegotiator`, and `read_sideband_pack`.

/// The `object-format=` value to put on the wire for a v2 request: echo the
/// server's advertised object-format when present, else fall back to the local
/// odb's hash algorithm (sha1/sha256). Keeps the negotiation hash-algo-aware.
pub(crate) fn v2_object_format(server_caps: &[String], local_odb: &crate::odb::Odb) -> String {
    for c in server_caps {
        if let Some(fmt) = c.strip_prefix("object-format=") {
            let f = fmt.trim();
            if !f.is_empty() {
                return f.to_ascii_lowercase();
            }
        }
    }
    local_odb.hash_algo().name().to_owned()
}

fn push_unique_prefix(out: &mut Vec<String>, value: &str) {
    if !out.iter().any(|v| v == value) {
        out.push(value.to_owned());
    }
}

/// Ref namespace prefixes to enumerate when advertising refs for fetch (local dir,
/// v2 `ls-refs`, and opportunistic remote-tracking updates).
///
/// Always includes `refs/heads/` and `refs/tags/`. Explicit refspec sources also
/// add literal prefixes (e.g. `refs/heads/main`, `refs/pull/123/head`) so DWIM can
/// resolve `pull/123/head` under `refs/`.
pub(crate) fn fetch_ref_list_prefixes(refspecs: &[String]) -> Vec<String> {
    let mut out = vec!["refs/heads/".to_owned(), "refs/tags/".to_owned()];
    for spec in refspecs {
        if spec.starts_with('^') {
            continue;
        }
        let raw = spec.strip_prefix('+').unwrap_or(spec.as_str());
        let src = raw.split_once(':').map(|(s, _)| s).unwrap_or(raw).trim();
        if src.is_empty() {
            continue;
        }
        if src == "HEAD" {
            push_unique_prefix(&mut out, "HEAD");
            continue;
        }
        if let Some(star) = src.find('*') {
            let prefix = &src[..star];
            if prefix.is_empty() {
                continue;
            }
            if prefix.starts_with("refs/") {
                push_unique_prefix(&mut out, prefix);
            } else {
                push_unique_prefix(&mut out, &format!("refs/heads/{prefix}"));
            }
            continue;
        }
        if src.starts_with("refs/") {
            push_unique_prefix(&mut out, src);
        } else {
            push_unique_prefix(&mut out, &format!("refs/heads/{src}"));
            // Git DWIM: bare names may live directly under `refs/` (pull refs, etc.).
            push_unique_prefix(&mut out, &format!("refs/{src}"));
        }
    }
    out
}

/// Parse an `unborn HEAD symref-target:…` line from a v2 `ls-refs` response.
#[must_use]
pub(crate) fn parse_v2_ls_refs_unborn_head_symref(line: &str) -> Option<String> {
    const PREFIX: &str = "unborn HEAD symref-target:";
    let target = line.strip_prefix(PREFIX)?;
    let target = target.split_whitespace().next()?.to_owned();
    crate::refs::is_valid_advertised_symref_target(&target).then_some(target)
}

/// Parse one v2 `ls-refs` advertisement line into `(refname, oid, symref_target)`.
///
/// Lines look like `<oid> <refname>[ symref-target:<t>][ peeled:<oid>]`. Lib-side
/// port of the CLI's `parse_ls_refs_v2_line` (the order of the optional suffixes
/// is whichever the server emits; we scan for both tokens). Returns `None` for a
/// malformed line.
pub(crate) fn parse_ls_refs_v2_line(
    line: &str,
) -> Option<(String, ObjectId, Option<String>, Option<ObjectId>)> {
    const SYM: &str = " symref-target:";
    const PEEL: &str = " peeled:";
    let (oid_hex, after_oid) = line.split_once(' ')?;
    let oid = ObjectId::from_hex(oid_hex).ok()?;

    // The refname ends at the first ` symref-target:` or ` peeled:` token.
    let sym_at = after_oid.find(SYM);
    let peel_at = after_oid.find(PEEL);
    let name_end = match (sym_at, peel_at) {
        (Some(a), Some(b)) => a.min(b),
        (Some(a), None) => a,
        (None, Some(b)) => b,
        (None, None) => after_oid.len(),
    };
    let name = after_oid[..name_end].trim().to_owned();
    if name.is_empty() {
        return None;
    }
    let symref_target = sym_at.map(|pos| {
        let tail = &after_oid[pos + SYM.len()..];
        let end = tail.find(' ').unwrap_or(tail.len());
        tail[..end].to_owned()
    });
    let advertised_peel = peel_at.and_then(|pos| {
        let tail = &after_oid[pos + PEEL.len()..];
        let end = tail.find(' ').unwrap_or(tail.len());
        ObjectId::from_hex(tail[..end].trim()).ok()
    });
    Some((name, oid, symref_target, advertised_peel))
}

/// Issue `command=ls-refs` over a v2 connection and parse the ref map.
///
/// Sends the capability echo (agent/object-format via
/// [`protocol_v2::cap_lines_for_command_request`]), the `0001` delimiter, then
/// `symrefs`, `peel`, and `ref-prefix <p>` lines derived from `refspecs` (plus
/// `refs/tags/` when `tags != None`), then flush. Returns the advertised
/// `refs/heads/*` and `refs/tags/*` refs (peeled `^{}` carrier lines dropped) and
/// the `HEAD` symref target. Lifted from the CLI's `v2_ls_refs_for_fetch`.
fn v2_ls_refs(
    conn: &mut dyn Connection,
    server_caps: &[String],
    local_odb: &crate::odb::Odb,
    tags: crate::transfer::TagMode,
    refspecs: &[String],
) -> Result<LsRefsAdvertised> {
    let req = build_v2_ls_refs_request(server_caps, local_odb, tags, refspecs)?;
    conn.writer().write_all(&req)?;
    conn.writer().flush()?;
    parse_v2_ls_refs_response(conn.reader())
}

/// Build the `command=ls-refs` request body (capability echo + `0001` + the
/// `symrefs`/`peel`/`ref-prefix` argument lines + flush) for a v2 fetch.
///
/// Factored out of [`v2_ls_refs`] so the streaming transports (which write it to
/// a duplex socket) and the stateless smart-HTTP transport (which POSTs it as a
/// request body) share one request builder. `HEAD` is always requested so the
/// server advertises its `symref-target`; `refs/tags/` is added under `--tags` /
/// tag-following even when the refspecs name only heads.
pub(crate) fn build_v2_ls_refs_request(
    server_caps: &[String],
    local_odb: &crate::odb::Odb,
    tags: crate::transfer::TagMode,
    refspecs: &[String],
) -> Result<Vec<u8>> {
    let object_format = v2_object_format(server_caps, local_odb);
    let cap_echo = protocol_v2::cap_lines_for_command_request(server_caps);

    let mut req: Vec<u8> = Vec::new();
    pkt_line::write_line(&mut req, "command=ls-refs")?;
    // Echo agent/object-format; if the server advertised neither (rare), still
    // pin the object-format so a sha256 server agrees on hash width.
    if cap_echo.iter().any(|c| c.starts_with("object-format=")) {
        for line in &cap_echo {
            pkt_line::write_line(&mut req, line)?;
        }
    } else {
        for line in &cap_echo {
            pkt_line::write_line(&mut req, line)?;
        }
        pkt_line::write_line(&mut req, &format!("object-format={object_format}"))?;
    }
    pkt_line::write_delim(&mut req)?;
    pkt_line::write_line(&mut req, "symrefs")?;
    if crate::protocol_v2::server_advertises_ls_refs_unborn(server_caps) {
        pkt_line::write_line(&mut req, "unborn")?;
    }
    pkt_line::write_line(&mut req, "peel")?;

    // Always request `HEAD` so the server advertises its `symref-target`, which
    // drives `FetchOutcome::default_branch` (the wire equivalent of the v0/v1
    // `symref=HEAD:` capability). `HEAD` is dropped from the fetchable ref set.
    pkt_line::write_line(&mut req, "ref-prefix HEAD")?;
    let mut prefixes = fetch_ref_list_prefixes(refspecs);
    if tags != crate::transfer::TagMode::None && !prefixes.iter().any(|p| p == "refs/tags/") {
        // Tag-following / `--tags` wants the tag namespace advertised so we can
        // add tags from the ls-refs result, even if the refspecs only name heads.
        prefixes.push("refs/tags/".to_owned());
    }
    for p in &prefixes {
        pkt_line::write_line(&mut req, &format!("ref-prefix {p}"))?;
    }
    pkt_line::write_flush(&mut req)?;
    Ok(req)
}

/// Parse a `command=ls-refs` response into `(advertised refs, HEAD symref)`.
///
/// Reads `<oid> <refname>[ symref-target:…][ peeled:…]` lines up to the
/// terminating flush, dropping peeled `^{}` carriers and recording the `HEAD`
/// symref target. Shared by the streaming and stateless-HTTP v2 paths.
pub(crate) fn parse_v2_ls_refs_response(reader: &mut dyn Read) -> Result<LsRefsAdvertised> {
    // Response: `<oid> <refname>[ symref-target:…][ peeled:…]` lines, flush-terminated.
    let mut advertised: Vec<(String, ObjectId)> = Vec::new();
    let mut head_symref: Option<String> = None;
    let mut head_oid: Option<ObjectId> = None;
    let mut advertised_peel: std::collections::HashMap<String, ObjectId> =
        std::collections::HashMap::new();
    let mut reader = reader;
    loop {
        match pkt_line::read_packet(&mut reader)? {
            None | Some(pkt_line::Packet::Flush) | Some(pkt_line::Packet::Delim) => break,
            Some(pkt_line::Packet::ResponseEnd) => break,
            Some(pkt_line::Packet::Data(line)) => {
                let line = line.trim_end_matches('\n');
                if let Some(msg) = line.strip_prefix("ERR ") {
                    return Err(Error::Message(format!("remote error: {}", msg.trim_end())));
                }
                if let Some(t) = parse_v2_ls_refs_unborn_head_symref(line) {
                    head_symref = Some(t);
                    continue;
                }
                let Some((name, oid, symref_target, peel)) = parse_ls_refs_v2_line(line) else {
                    continue;
                };
                if name.contains("^{") || name.ends_with("^{}") {
                    continue;
                }
                if name == "HEAD" {
                    head_oid = Some(oid);
                    if let Some(t) =
                        symref_target.filter(|t| crate::refs::is_valid_advertised_symref_target(t))
                    {
                        head_symref = Some(t);
                    }
                    // HEAD itself is not a fetchable ref here; refspecs target heads/tags.
                    continue;
                }
                if crate::refs::is_valid_fetch_advertised_ref(&name) {
                    if let Some(p) = peel {
                        advertised_peel.insert(name.clone(), p);
                    }
                    advertised.push((name, oid));
                }
            }
        }
    }
    Ok((advertised, head_symref, advertised_peel, head_oid))
}

/// Build the ordered `have` candidate list for a v2 fetch from the local ref
/// tips (heads, tags, HEAD), peeled to commits and excluding the wants, driven
/// through the [`SkippingNegotiator`]'s skipping schedule.
///
/// Shared by the streaming (`negotiate_pack_v2`) and stateless-HTTP v2 fetch
/// paths so both offer the server the same `have`s in the same order. The wire
/// rounds (how many haves per request, when to send `done`) are batched by the
/// caller, which differs between a duplex socket and stateless POSTs.
pub(crate) fn v2_local_haves(local_git_dir: &Path, wants: &[ObjectId]) -> Result<Vec<ObjectId>> {
    let want_set: HashSet<ObjectId> = wants.iter().copied().collect();
    let local_repo = crate::repo::Repository::open(local_git_dir, &RepositoryOptions::empty())?;
    let mut negotiator = SkippingNegotiator::new(local_repo);
    let mut tips: Vec<ObjectId> = Vec::new();
    let mut seen_tip: HashSet<ObjectId> = HashSet::new();
    for prefix in ["refs/heads/", "refs/tags/"] {
        if let Ok(entries) = crate::refs::list_refs(local_git_dir, prefix) {
            for (_, oid) in entries {
                if let Ok(Some(c)) = negotiator.repo().odb().peel_to_commit(oid) {
                    if !want_set.contains(&c) && seen_tip.insert(c) {
                        tips.push(c);
                    }
                }
            }
        }
    }
    if let Ok(h) = crate::refs::resolve_ref(local_git_dir, "HEAD") {
        if let Ok(Some(c)) = negotiator.repo().odb().peel_to_commit(h) {
            if !want_set.contains(&c) && seen_tip.insert(c) {
                tips.push(c);
            }
        }
    }
    tips.sort_by_key(ObjectId::to_hex);
    for t in tips {
        negotiator.add_tip(t)?;
    }
    // Drain the negotiator into an ordered have list (it already applies the
    // skipping schedule); the caller batches the wire rounds.
    let mut haves: Vec<ObjectId> = Vec::new();
    while let Some(oid) = negotiator.next_have()? {
        haves.push(oid);
    }
    Ok(haves)
}

/// Run a v2 `command=fetch` negotiation over the connection and return the raw
/// pack bytes for `wants`.
///
/// Drives the [`SkippingNegotiator`] exactly like the v0/v1 path, but frames the
/// request as v2 (`command=fetch`, capability echo, `0001`, then
/// `thin-pack`/`no-progress`/`ofs-delta`, `want <oid>` lines, `have <oid>` lines,
/// and `done`). Multi-round: round 1 sends the first batch of haves *without*
/// `done`, reads the `acknowledgments` section (looking for `ready`); if not yet
/// ready it sends the remaining haves + `done`. Then reads the response sections
/// (`acknowledgments`, optional `shallow-info`, then `wanted-refs`, then `packfile`) and
/// demuxes the side-band-64k pack. Lifted from `write_v2_fetch_request` +
/// `read_v2_acknowledgments` / `read_v2_fetch_pack_response`.
#[allow(clippy::too_many_arguments)]
fn negotiate_pack_v2(
    local_git_dir: &Path,
    conn: &mut dyn Connection,
    server_caps: &[String],
    local_odb: &crate::odb::Odb,
    wants: &[ObjectId],
    deepen: &V2DeepenArgs,
    wire_filter: Option<&str>,
    progress: &mut dyn Progress,
) -> Result<(Option<std::path::PathBuf>, ShallowUpdate)> {
    if wants.is_empty() {
        return Ok((None, ShallowUpdate::default()));
    }
    let object_format = v2_object_format(server_caps, local_odb);
    let cap_echo = protocol_v2::cap_lines_for_command_request(server_caps);
    let sideband_all = protocol_v2::fetch_supports_sideband_all(server_caps);

    // A deepen/shallow request does not offer local haves (the local objects
    // bottom out at grafts and are not a usable negotiation base), forcing the
    // single-round path so the server sends a `shallow-info` section + pack.
    let shallow_request = deepen.is_shallow_request();

    // The ordered `have` list, built from the local ref tips with the skipping
    // negotiator (shared with the stateless-HTTP v2 path). Empty for a shallow
    // request.
    let haves = if shallow_request {
        Vec::new()
    } else {
        v2_local_haves(local_git_dir, wants)?
    };

    let mut pack_receive = crate::pack_receive::TempPackReceive::create(local_odb.objects_dir())?;
    let mut shallow_update = ShallowUpdate::default();
    if haves.is_empty() {
        // No local history to offer: single round, wants + done, read the pack.
        write_v2_fetch_request(
            conn.writer(),
            &object_format,
            &cap_echo,
            wants,
            &[],
            sideband_all,
            deepen,
            wire_filter,
            true,
        )?;
        read_v2_fetch_pack_response(
            conn.reader(),
            &mut Vec::new(),
            Some(&mut pack_receive),
            &mut shallow_update,
            progress,
            sideband_all,
        )?;
        return Ok((pack_receive.finish()?, shallow_update));
    }

    // Multi-round: round 1 sends the first batch of haves WITHOUT done.
    let first_batch = haves.len().min(INITIAL_FLUSH);
    write_v2_fetch_request(
        conn.writer(),
        &object_format,
        &cap_echo,
        wants,
        &haves[..first_batch],
        sideband_all,
        deepen,
        wire_filter,
        false,
    )?;

    let ack = read_v2_acknowledgments(conn.reader(), sideband_all)?;
    match ack {
        // Server is `ready`: the pack follows in the SAME response after a delim.
        Some(round) if round.ready => {
            read_v2_fetch_pack_response(
                conn.reader(),
                &mut Vec::new(),
                Some(&mut pack_receive),
                &mut shallow_update,
                progress,
                sideband_all,
            )?;
        }
        // Server skipped acknowledgments and went straight to the pack header
        // (consumed inside the reader); read the pack now.
        None => {
            read_v2_fetch_pack_response(
                conn.reader(),
                &mut Vec::new(),
                Some(&mut pack_receive),
                &mut shallow_update,
                progress,
                sideband_all,
            )?;
        }
        // Not ready yet: round 2 sends the remaining haves + `done`, then pack.
        Some(_) => {
            let haves_round2: &[ObjectId] = if conn.stateless_rpc() {
                &haves[..]
            } else {
                &haves[first_batch..]
            };
            write_v2_fetch_request(
                conn.writer(),
                &object_format,
                &cap_echo,
                wants,
                haves_round2,
                sideband_all,
                deepen,
                wire_filter,
                true,
            )?;
            read_v2_fetch_pack_response(
                conn.reader(),
                &mut Vec::new(),
                Some(&mut pack_receive),
                &mut shallow_update,
                progress,
                sideband_all,
            )?;
        }
    }
    Ok((pack_receive.finish()?, shallow_update))
}

/// The shallow/deepen arguments for a v2 `command=fetch` request, derived from
/// [`FetchOptions`] plus the local `shallow` file. Built once by the fetch driver
/// and passed to [`write_v2_fetch_request`] on each round (every stateless POST
/// must resend them).
#[derive(Clone, Default)]
pub(crate) struct V2DeepenArgs {
    /// The client's current shallow grafts (`shallow <oid>` lines).
    pub(crate) local_shallow: Vec<ObjectId>,
    /// `deepen <n>` (absolute depth, or `INFINITE_DEPTH` for `--unshallow`).
    pub(crate) depth: Option<u32>,
    /// When true, send v0/v1/v2 `deepen-relative` after `deepen`.
    pub(crate) deepen_relative: bool,
    /// `deepen-since <unix-ts>`.
    pub(crate) deepen_since: Option<String>,
    /// `deepen-not <ref>` exclusions.
    pub(crate) deepen_not: Vec<String>,
    /// When true, omit the v2 `no-progress` fetch argument.
    pub(crate) remote_progress: bool,
}

impl V2DeepenArgs {
    /// Build the v2 deepen args from the fetch options and the local shallow file,
    /// translating `--unshallow` into the `INFINITE_DEPTH` deepen Git uses.
    pub(crate) fn from_opts(opts: &FetchOptions, local_shallow: &[ObjectId]) -> Self {
        let (depth, deepen_relative) = opts
            .wire_deepen()
            .map(|(d, r)| (Some(d), r))
            .unwrap_or((None, false));
        Self {
            local_shallow: local_shallow.to_vec(),
            depth,
            deepen_relative,
            deepen_since: opts
                .deepen_since
                .as_deref()
                .filter(|s| !s.trim().is_empty())
                .map(crate::shallow::deepen_since_wire_value),
            deepen_not: opts
                .deepen_not
                .iter()
                .map(|s| s.trim().to_owned())
                .filter(|s| !s.is_empty())
                .collect(),
            remote_progress: opts.remote_progress,
        }
    }

    /// Whether any deepen/shallow argument is present (drives `shallow-info`
    /// handling and the "skip offering haves" decision).
    pub(crate) fn is_shallow_request(&self) -> bool {
        self.depth.is_some()
            || self.deepen_relative
            || self.deepen_since.is_some()
            || !self.deepen_not.is_empty()
            || !self.local_shallow.is_empty()
    }
}
#[expect(clippy::too_many_arguments)]
/// Write a v2 `command=fetch` request: capability echo, `0001`, the standard
/// `thin-pack`/`no-progress`/`ofs-delta` (+ `sideband-all`/`include-tag`)
/// arguments, the shallow/deepen arguments, the `want <oid>` lines, the
/// `have <oid>` lines, and `done` when `send_done`, terminated by flush. Lifted
/// from the CLI's `write_v2_fetch_request` (streaming-fetch subset).
pub(crate) fn write_v2_fetch_request(
    w: &mut dyn Write,
    object_format: &str,
    cap_echo: &[String],
    wants: &[ObjectId],
    haves: &[ObjectId],
    sideband_all: bool,
    deepen: &V2DeepenArgs,
    wire_filter: Option<&str>,
    send_done: bool,
) -> Result<()> {
    let mut req: Vec<u8> = Vec::new();
    pkt_line::write_line(&mut req, "command=fetch")?;
    if cap_echo.iter().any(|c| c.starts_with("object-format=")) {
        for line in cap_echo {
            pkt_line::write_line(&mut req, line)?;
        }
    } else {
        for line in cap_echo {
            pkt_line::write_line(&mut req, line)?;
        }
        pkt_line::write_line(&mut req, &format!("object-format={object_format}"))?;
    }
    pkt_line::write_delim(&mut req)?;

    pkt_line::write_line(&mut req, "thin-pack")?;
    if !deepen.remote_progress {
        pkt_line::write_line(&mut req, "no-progress")?;
    }
    pkt_line::write_line(&mut req, "ofs-delta")?;
    if sideband_all {
        pkt_line::write_line(&mut req, "sideband-all")?;
    }
    // Ask the server to bundle tag objects pointing at fetched history; the
    // TagMode plumbing in `fetch_remote` decides which tag refs to write.
    pkt_line::write_line(&mut req, "include-tag")?;

    // Shallow/deepen arguments (the `fetch` v2 command's `shallow`/`deepen*` args).
    for oid in &deepen.local_shallow {
        pkt_line::write_line(&mut req, &format!("shallow {}", oid.to_hex()))?;
    }
    if let Some(depth) = deepen.depth {
        pkt_line::write_line(&mut req, &format!("deepen {depth}"))?;
    }
    if deepen.deepen_relative {
        pkt_line::write_line(&mut req, "deepen-relative")?;
    }
    if let Some(since) = &deepen.deepen_since {
        pkt_line::write_line(&mut req, &format!("deepen-since {since}"))?;
    }
    for excl in &deepen.deepen_not {
        pkt_line::write_line(&mut req, &format!("deepen-not {excl}"))?;
    }
    if let Some(spec) = wire_filter {
        pkt_line::write_line(&mut req, &format!("filter {spec}"))?;
    }

    for want in wants {
        pkt_line::write_line(&mut req, &format!("want {}", want.to_hex()))?;
    }
    for have in haves {
        pkt_line::write_line(&mut req, &format!("have {}", have.to_hex()))?;
    }
    if send_done {
        pkt_line::write_line(&mut req, "done")?;
    }
    pkt_line::write_flush(&mut req)?;
    w.write_all(&req)?;
    w.flush()?;
    Ok(())
}

/// Outcome of reading one v2 `acknowledgments` section.
pub(crate) struct V2AckRound {
    /// Server emitted `ready`: the packfile follows in the same response after a
    /// delimiter — the caller reads the pack now without sending more.
    pub(crate) ready: bool,
}

/// Read a v2 `acknowledgments` section header and its `ACK`/`NAK`/`ready` lines.
///
/// Returns `Some(round)` for an `acknowledgments` section (with `ready` set when
/// the server is ready to send the pack), or `None` when the server skipped the
/// section and started a different one (e.g. went straight to `packfile`) — in
/// which case the header has been consumed and the caller proceeds to read the
/// pack response directly. Lifted from the CLI's `read_v2_acknowledgments`.
pub(crate) fn read_v2_acknowledgments(
    reader: &mut dyn Read,
    sideband_all: bool,
) -> Result<Option<V2AckRound>> {
    let mut reader = reader;
    let hdr = match crate::shallow::read_v2_fetch_primary_line(&mut reader, sideband_all)? {
        Some(s) => s,
        None => return Ok(Some(V2AckRound { ready: false })),
    };
    let hdr = hdr.as_str();
    if let Some(msg) = hdr.strip_prefix("ERR ") {
        return Err(Error::Message(format!("remote error: {}", msg.trim_end())));
    }
    if hdr != "acknowledgments" {
        // The server started a non-acknowledgments section; the pack reader,
        // called next, re-dispatches on this header. We cannot push it back, so
        // signal `None` only when we know the pack reader will see the same
        // header — which it will, because the next read picks up where we left
        // off. To make that work, the caller treats `None` as "read the pack".
        // The header we just consumed (`shallow-info`/`wanted-refs`/`packfile`)
        // would be lost; for the streaming fetch we only reach here after a
        // first round of haves, where servers always emit `acknowledgments`
        // first. Reaching a different header is therefore unexpected.
        return Err(Error::Message(format!(
            "unexpected v2 fetch section before acknowledgments: {hdr}"
        )));
    }
    let mut ready = false;
    loop {
        match pkt_line::read_packet(&mut reader)? {
            Some(pkt_line::Packet::Delim) | Some(pkt_line::Packet::Flush) | None => break,
            Some(pkt_line::Packet::ResponseEnd) => break,
            Some(pkt_line::Packet::Data(payload)) => {
                let Some(ln) = crate::shallow::v2_primary_line_from_pkt(payload, sideband_all)?
                else {
                    continue;
                };
                if ln == "NAK" || ln.starts_with("ACK ") {
                    continue;
                }
                if ln == "ready" {
                    ready = true;
                    continue;
                }
                return Err(Error::Message(format!(
                    "unexpected acknowledgment line: '{ln}'"
                )));
            }
        }
    }
    Ok(Some(V2AckRound { ready }))
}

/// Whether `line` is a v2 `wanted-refs` (or ls-refs) advertisement row:
/// `<oid> <refname>[ …attributes…]`.
fn v2_wanted_ref_ad_line(line: &str) -> bool {
    let line = line.trim_end();
    let Some((oid_hex, rest)) = line.split_once(' ') else {
        return false;
    };
    if ObjectId::from_hex(oid_hex.trim()).is_err() {
        return false;
    }
    !rest.trim().is_empty()
}

/// Read a v2 `command=fetch` response: capture the `shallow-info` section's
/// `shallow`/`unshallow` lines into `shallow_out`, skip the other non-pack
/// sections (`acknowledgments`/`shallow-info`/`wanted-refs`/`packfile-uris`), and demux the
/// side-band-64k pack from the `packfile` section into `out`. Lifted from the
/// CLI's `read_v2_fetch_pack_response`, extended to surface shallow updates.
pub(crate) fn read_v2_fetch_pack_response(
    reader: &mut dyn Read,
    out: &mut Vec<u8>,
    pack_sink: Option<&mut crate::pack_receive::TempPackReceive>,
    shallow_out: &mut ShallowUpdate,
    progress: &mut dyn Progress,
    sideband_all: bool,
) -> Result<()> {
    loop {
        let hdr = loop {
            match pkt_line::read_packet(&mut &mut *reader)? {
                Some(pkt_line::Packet::Data(payload)) => {
                    if let Some(line) =
                        crate::shallow::v2_primary_line_from_pkt(payload, sideband_all)?
                    {
                        break line;
                    }
                }
                Some(pkt_line::Packet::Flush) | None => return Ok(()),
                Some(pkt_line::Packet::Delim) => continue,
                Some(other) => {
                    return Err(Error::Message(format!(
                        "unexpected v2 fetch response: {other:?}"
                    )))
                }
            }
        };
        let hdr = hdr.as_str();
        if let Some(msg) = hdr.strip_prefix("ERR ") {
            return Err(Error::Message(format!("remote error: {}", msg.trim_end())));
        }
        match hdr {
            h if h == crate::protocol_v2::FetchResponseSection::ShallowInfo.header() => {
                // Capture the shallow/unshallow boundary updates. The section is
                // delim-terminated (before the `packfile` header), which
                // `read_shallow_info_section` stops at, leaving the header intact.
                let (sh, unsh) =
                    crate::shallow::read_shallow_info_section_sideband(&mut *reader, sideband_all)?;
                shallow_out.shallow.extend(sh);
                shallow_out.unshallow.extend(unsh);
            }
            h if h == crate::protocol_v2::FetchResponseSection::Acknowledgments.header()
                || h == crate::protocol_v2::FetchResponseSection::WantedRefs.header()
                || h == crate::protocol_v2::FetchResponseSection::PackfileUris.header() =>
            {
                skip_v2_section_until_boundary(&mut *reader, sideband_all)?;
            }
            h if h == crate::protocol_v2::FetchResponseSection::Packfile.header() => {
                // The `packfile` section body is side-band-64k framed; reuse the
                // shared demuxer (channel 1 = pack, channel 2 = progress, 3 = err).
                if let Some(sink) = pack_sink {
                    sink.read_sideband(&mut *reader, progress)?;
                } else {
                    let mut target = crate::pack_receive::PackReceiveTarget::Memory(out);
                    read_sideband_pack(&mut *reader, &mut target, progress)?;
                }
                return Ok(());
            }
            other if v2_wanted_ref_ad_line(other) => {
                // `wanted-refs` payload line (some servers omit the section header).
            }
            other => {
                return Err(Error::Message(format!(
                    "unexpected v2 fetch section: {other}"
                )))
            }
        }
    }
}

/// Skip a v2 response section up to its terminating flush/delim.
fn skip_v2_section_until_boundary(reader: &mut dyn Read, sideband_all: bool) -> Result<()> {
    loop {
        match pkt_line::read_packet(&mut &mut *reader)? {
            None | Some(pkt_line::Packet::Flush) | Some(pkt_line::Packet::Delim) => return Ok(()),
            Some(pkt_line::Packet::ResponseEnd) => return Ok(()),
            Some(pkt_line::Packet::Data(payload)) => {
                let _ = crate::shallow::v2_primary_line_from_pkt(payload, sideband_all)?;
            }
        }
    }
}

/// Fetch from a remote over a live [`Connection`], driving the upload-pack
/// negotiation and writing the resulting tracking-ref updates into
/// `local_git_dir`.
///
/// The flow mirrors local fetch, but the remote ref list comes from the
/// connection's advertisement, objects arrive over the wire as a negotiated pack,
/// and the local repo is opened to classify ancestry. Reuses refspec matching,
/// tag-mode, prune, and classification helpers shared with other fetch paths.
///
/// Handles protocol v0, v1, and v2. For a v2 connection the ref map is recovered
/// via a `command=ls-refs` round (no refs are advertised on connect) and the
/// pack is negotiated with `command=fetch`; v0/v1 use the connect-time
/// advertisement and the classic `want`/`have`/`done` exchange.
///
/// # Errors
///
/// Returns an error if a refspec is invalid, if the negotiation or pack ingest
/// fails, or on ref/odb I/O failure.
pub fn fetch_remote(
    local_git_dir: &Path,
    conn: &mut dyn Connection,
    opts: &mut FetchOptions,
    progress: &mut dyn Progress,
) -> Result<FetchOutcome> {
    use crate::net_trace::net_trace;
    // Wall-clock since fetch_remote entry, included in phase-boundary traces so
    // `GRIT_NET_DEBUG=1` shows where time goes (see the "matched"/"done" lines).
    let t_begin = std::time::Instant::now();
    net_trace!(
        opts.network_trace,
        opts.diagnostics.as_ref(),
        "fetch_remote: begin — protocol v{}, {} refspec(s), tags={:?}, depth={:?}",
        conn.protocol_version(),
        opts.refspecs.len(),
        opts.tags,
        opts.fetch_depth
    );
    validate_fetch_depth_local(local_git_dir, opts)?;
    let local_cfg = crate::config::ConfigSet::load(&opts.environment, Some(local_git_dir), true)?;
    if opts.filter.is_none() {
        if let Some(remote) = opts.remote_name.as_deref() {
            opts.filter = crate::partial_clone::stored_remote_filter(&local_cfg, remote);
        }
    }
    let local_odb = open_odb(local_git_dir);

    // 1. Remote refs + default branch.
    //
    // For protocol v2 the connect-time advertisement carries no refs (only the
    // capability block); we obtain them now with an `ls-refs` command, derived
    // from the fetch refspecs. For v0/v1 they come from the connect-time
    // advertisement directly.
    let (remote_refs, default_branch, v2_caps, advertised_peel, remote_head_oid) = if conn
        .protocol_version()
        >= 2
    {
        let caps: Vec<String> = conn.capabilities().to_vec();
        let (refs, head_symref, peel, head_oid) =
            v2_ls_refs(conn, &caps, &local_odb, opts.tags, &opts.refspecs)?;
        let default_branch =
            head_symref.map(|t| t.strip_prefix("refs/heads/").unwrap_or(&t).to_owned());
        (refs, default_branch, Some(caps), peel, head_oid)
    } else {
        let default_branch = conn
            .head_symref()
            .map(|t| t.strip_prefix("refs/heads/").unwrap_or(t).to_owned());
        let advertised_peel = conn.advertised_tag_peel().cloned().unwrap_or_default();
        let remote_head_oid =
            crate::transfer::remote_head_oid_from_advertised(conn.advertised_refs());
        let remote_refs: Vec<(String, ObjectId)> = conn
            .advertised_refs()
            .iter()
            .filter(|(n, _)| {
                n != "HEAD" && !n.ends_with("^{}") && crate::refs::is_valid_fetch_advertised_ref(n)
            })
            .cloned()
            .collect();
        (
            remote_refs,
            default_branch,
            None,
            advertised_peel,
            remote_head_oid,
        )
    };
    net_trace!(
        opts.network_trace,
        opts.diagnostics.as_ref(),
        "fetch_remote: remote advertised {} ref(s){}",
        remote_refs.len(),
        v2_caps
            .as_ref()
            .map(|_| " (via v2 ls-refs)")
            .unwrap_or(" (v0/v1 advertisement)")
    );

    if opts.initial_remote_fetch
        && remote_refs.is_empty()
        && !crate::transport::connection_handshake_complete(conn)
    {
        return Err(Error::RemoteReadFailed);
    }

    let crate::transfer::FetchMatchSet {
        mut matched,
        positive,
        negatives,
        explicit_refspec_fetch,
    } = crate::transfer::build_fetch_matched_set(opts, &remote_refs, remote_head_oid)?;
    let mut matched_oids: HashSet<ObjectId> = matched.iter().map(|m| m.oid).collect();
    let mut seen_remote_ref: HashSet<String> =
        matched.iter().map(|m| m.remote_ref.clone()).collect();

    // TagMode: add tags. Tag-following needs the closure of fetched objects,
    // which we cannot compute remotely; the wire `include-tag` capability makes
    // the server send tag objects with the pack, so we add advertised tags by
    // mode here and let classification proceed once the pack lands. For
    // `Following` we approximate using the advertised remote odb if present
    // (it is not, over the wire), so we add following tags whose oid is among
    // the matched set after the fact — handled below using the local odb.
    //
    // `following_only` collects the oids of tags added *provisionally* under
    // `Following`. These must NOT be `want`ed up front: git's tag-following only
    // keeps a tag whose target is already reachable from the fetched heads, so
    // wanting the tag itself would drag down its (otherwise unreachable) target
    // and incorrectly keep the tag. They are pruned by `retain_following_tags`.
    let tags_present = if opts.tags == crate::transfer::TagMode::Following {
        tags_already_local(local_git_dir, &remote_refs)
    } else {
        HashSet::new()
    };
    let following_only = add_wire_tags(
        opts.tags,
        &remote_refs,
        &advertised_peel,
        &negatives,
        &tags_present,
        &mut matched,
        &mut matched_oids,
        &mut seen_remote_ref,
    );
    let fetch_commit_oids: HashSet<ObjectId> = matched
        .iter()
        .filter(|m| !m.is_tag)
        .map(|m| m.oid)
        .collect();
    let mut following_tag_wants =
        following_tag_wants_with_local_peel(&matched, &following_only, &local_odb);

    // The client's current shallow grafts (drives the wire `shallow <oid>` lines
    // and the "this is a shallow request" decisions in the negotiators).
    let local_shallow = crate::shallow::load_shallow_oids(local_git_dir)?;
    let shallow_request = opts.has_deepen_request() || !local_shallow.is_empty();
    if shallow_request {
        let local_shallow_set: HashSet<ObjectId> = local_shallow.iter().copied().collect();
        let peel_set = following_peel_commit_set_for_shallow(
            &local_odb,
            &fetch_commit_oids,
            opts.absolute_depth_for_tags(),
            &local_shallow_set,
        );
        following_tag_wants.extend(following_tag_wants_peel_among_fetched_commits(
            &matched,
            &following_only,
            &peel_set,
        ));
    }

    // 4. Wants. Normally the matched oids that are absent locally. For a
    // deepen/`--unshallow` request the wanted tips may already be present (a prior
    // shallow fetch landed them); we must still `want` them so the server fills in
    // the now-reachable ancestors past the old boundary.
    let wants: Vec<ObjectId> = if shallow_request {
        matched_oids
            .iter()
            .copied()
            .filter(|oid| !following_only.contains(oid) || following_tag_wants.contains(oid))
            .collect()
    } else {
        matched_oids
            .iter()
            .copied()
            .filter(|oid| {
                if following_only.contains(oid) {
                    return following_tag_wants.contains(oid);
                }
                !local_odb.exists(oid)
            })
            .collect()
    };

    // Shallow-boundary updates the server reports (`shallow`/`unshallow`), applied
    // to the local `shallow` file and surfaced in the outcome.
    let mut shallow_update = ShallowUpdate::default();
    let mut pack_oids: HashSet<ObjectId> = HashSet::new();
    let mut transfer = crate::transfer::TransferStats::default();

    net_trace!(
        opts.network_trace,
        opts.diagnostics.as_ref(),
        "fetch_remote: {} matched ref(s), want {} object(s){} [+{:?} since begin]",
        matched.len(),
        wants.len(),
        if shallow_request {
            " (shallow request)"
        } else {
            ""
        },
        t_begin.elapsed()
    );

    let mut dry_run_quarantine: Option<crate::receive_quarantine::ReceiveQuarantine> = None;

    let v0_caps = if v2_caps.is_none() {
        Some(conn.capabilities().join(" "))
    } else {
        None
    };
    let (wire_filter, _filter_sent) = crate::partial_clone::wire_filter_for_capabilities(
        opts.filter.as_ref(),
        v2_caps.as_deref(),
        v0_caps.as_deref(),
    );
    // Match Git: a requested/stored partial-clone filter marks the received pack
    // promisor even when the server lacks `filter` and fetches unfiltered.
    opts.mark_received_pack_promisor = opts.filter.is_some();
    let wire_filter = wire_filter.as_deref();

    if wants.is_empty() {
        crate::transport::send_client_flush(conn)?;
    } else {
        net_trace!(
            opts.network_trace,
            opts.diagnostics.as_ref(),
            "fetch_remote: negotiating + fetching pack…"
        );
        let (pack_path, su) = if let Some(caps) = v2_caps.as_ref() {
            validate_fetch_depth_server(opts, None, Some(caps))?;
            let deepen = V2DeepenArgs::from_opts(opts, &local_shallow);
            negotiate_pack_v2(
                local_git_dir,
                conn,
                caps,
                &local_odb,
                &wants,
                &deepen,
                wire_filter,
                progress,
            )?
        } else {
            negotiate_pack(
                local_git_dir,
                conn,
                &wants,
                opts,
                &local_shallow,
                wire_filter,
                progress,
            )?
        };
        shallow_update = su;
        net_trace!(
            opts.network_trace,
            opts.diagnostics.as_ref(),
            "fetch_remote: received pack, ingesting via index-pack…"
        );
        if let Some(path) = pack_path {
            if opts.pack_spool_only {
                net_trace!(
                    opts.network_trace,
                    opts.diagnostics.as_ref(),
                    "fetch_remote: pack spooled to {} (pack_spool_only)",
                    path.display()
                );
                return Ok(FetchOutcome {
                    updates: Vec::new(),
                    default_branch: None,
                    new_shallow: crate::shallow::dedupe_shallow_list(shallow_update.shallow),
                    new_unshallow: shallow_update.unshallow,
                    transfer: crate::transfer::TransferStats::default(),
                });
            }
            let pack_config =
                crate::config::ConfigSet::load(&opts.environment, Some(local_git_dir), true)?;
            let index_threads = crate::parallel::index_pack_threads_for_ingest(
                opts.thread_policy,
                &pack_config,
                opts.index_pack_threads,
            )?;
            if opts.dry_run && dry_run_quarantine.is_none() {
                dry_run_quarantine = Some(crate::receive_quarantine::ReceiveQuarantine::create(
                    &local_odb,
                    local_git_dir,
                )?);
            }
            let ingested = if let Some(q) = dry_run_quarantine.as_ref() {
                let q_odb = q.odb();
                crate::index_pack::ingest_received_pack_path_with_progress(
                    path,
                    &q_odb,
                    &crate::index_pack::IngestPackOptions {
                        fix_thin: true,
                        skip_post_index_verify: true,
                        threads: index_threads,
                        merged_config: Some(pack_config),
                        environment: opts.environment.clone(),
                        mark_promisor_pack: opts.mark_received_pack_promisor,
                    },
                    Some(progress),
                )?
            } else {
                crate::index_pack::ingest_received_pack_path_with_progress(
                    path,
                    &local_odb,
                    &crate::index_pack::IngestPackOptions {
                        fix_thin: true,
                        skip_post_index_verify: true,
                        threads: index_threads,
                        merged_config: Some(pack_config),
                        environment: opts.environment.clone(),
                        mark_promisor_pack: opts.mark_received_pack_promisor,
                    },
                    Some(progress),
                )?
            };
            pack_oids = ingested.object_ids;
            let bytes = std::fs::metadata(&ingested.pack_path)
                .map_err(crate::error::Error::Io)?
                .len();
            transfer = crate::transfer::TransferStats {
                objects: u64::try_from(pack_oids.len()).unwrap_or(u64::MAX),
                bytes,
                kinds: Some(ingested.object_counts),
            };
        }
    }

    // Apply the shallow/unshallow boundary updates to the on-disk `shallow` file
    // before classifying refs (so connectivity reflects the new graft set).
    if !opts.dry_run {
        crate::shallow::apply_shallow_updates(
            local_git_dir,
            &shallow_update.shallow,
            &shallow_update.unshallow,
        )?;
    }

    // Close the write side once the v2 conversation is done so the server's
    // persistent `serve_loop` sees EOF and exits — even when we sent only an
    // `ls-refs` (no wants) and skipped `command=fetch`. Without this a streaming
    // transport (ssh subprocess, daemon socket) hangs at teardown. No-op for
    // v0/v1, where the server closes after its single response.
    if v2_caps.is_some() && !wants.is_empty() {
        conn.finish_send();
    }

    // For TagMode::Following, prune tags whose target did not arrive in the
    // pack (now resolvable against the local odb, which holds the fetched
    // objects). All/None already handled; Following kept only when reachable.
    if opts.tags == crate::transfer::TagMode::Following {
        let t = std::time::Instant::now();
        let tag_odb = dry_run_quarantine
            .as_ref()
            .map(|q| q.odb())
            .unwrap_or_else(|| local_odb.clone());
        retain_following_tags(
            &tag_odb,
            &mut matched,
            &pack_oids,
            &crate::shallow::load_shallow_boundaries(local_git_dir),
        )?;
        net_trace!(
            opts.network_trace,
            opts.diagnostics.as_ref(),
            "fetch_remote: retain_following_tags {:?}",
            t.elapsed()
        );
    }

    // 5. Classify + apply ref updates (ancestry via the now-populated local repo).
    let t_apply = std::time::Instant::now();
    let classify_odb = dry_run_quarantine.as_ref().map(|q| q.odb());
    let classify_repo = if opts.dry_run {
        None
    } else {
        crate::repo::Repository::open(local_git_dir, &RepositoryOptions::empty()).ok()
    };
    let d_repo_open = t_apply.elapsed();

    let mut updates: Vec<RefUpdate> = Vec::new();

    let mut prune_batch: Vec<crate::refs::store::RefUpdate> = Vec::new();
    let t_prune = std::time::Instant::now();
    if opts.prune {
        prune_tracking_refs(
            local_git_dir,
            &positive,
            &remote_refs,
            opts.dry_run,
            &mut updates,
            if opts.dry_run {
                None
            } else {
                Some(&mut prune_batch)
            },
        )?;
    }
    if !opts.dry_run && !prune_batch.is_empty() {
        crate::refs::commit_ref_store_batch(local_git_dir, &prune_batch)?;
    }
    let d_prune = t_prune.elapsed();

    let mut store_batch: Vec<crate::refs::store::RefUpdate> = Vec::new();

    // Load `packed-refs` once for the whole apply. Resolving each ref's previous
    // value (and the per-write namespace-conflict check) would otherwise re-read
    // and re-scan the entire file per ref — O(refs × packed-refs size), which
    // dominates fetches on repositories with tens of thousands of packed refs
    // even when no objects are transferred.
    let t_packed = std::time::Instant::now();
    let packed = crate::refs::PackedRefs::load(local_git_dir)?;
    let d_packed = t_packed.elapsed();

    // Per-sub-step timing accumulators for the apply loop, surfaced via the
    // env-gated "done" trace so the dominant cost is visible without a profiler.
    let mut d_resolve = std::time::Duration::ZERO;
    let mut d_classify = std::time::Duration::ZERO;
    let mut d_write = std::time::Duration::ZERO;
    let mut n_written = 0usize;
    let mut n_ancestry = 0usize; // refs whose old != new → classify walks ancestry

    for m in &matched {
        let Some(local_ref) = &m.local_ref else {
            updates.push(RefUpdate {
                remote_ref: m.remote_ref.clone(),
                local_ref: None,
                old_oid: None,
                new_oid: Some(m.oid),
                mode: UpdateMode::NoChangeNeeded,
                note: Some("not stored (empty destination)".to_owned()),
            });
            continue;
        };

        let t = std::time::Instant::now();
        let old = crate::refs::resolve_ref_cached(local_git_dir, local_ref, &packed)
            .ok()
            .flatten();
        d_resolve += t.elapsed();

        if old.as_ref().is_some_and(|o| o != &m.oid) {
            n_ancestry += 1;
        }

        let t = std::time::Instant::now();
        let mode = classify_update(
            old.as_ref(),
            &m.oid,
            m.force,
            m.dest_is_tag,
            classify_repo.as_ref(),
            classify_odb.as_ref(),
        );
        d_classify += t.elapsed();

        let write = matches!(
            mode,
            UpdateMode::New | UpdateMode::FastForward | UpdateMode::Forced
        );
        crate::transfer::enforce_fetch_ref_update(
            local_ref,
            old,
            m.oid,
            mode,
            opts,
            explicit_refspec_fetch,
            classify_repo.as_ref(),
        )?;
        if write && !opts.dry_run {
            if !crate::refs::is_valid_storable_ref_name(local_ref) {
                updates.push(RefUpdate {
                    remote_ref: m.remote_ref.clone(),
                    local_ref: Some(local_ref.clone()),
                    old_oid: old,
                    new_oid: Some(m.oid),
                    mode,
                    note: Some("skipped (invalid ref name)".to_owned()),
                });
                continue;
            }
            store_batch.push(crate::refs::store::RefUpdate {
                name: local_ref.clone(),
                new_value: Some(crate::refs::store::RawRef::Direct(m.oid)),
                expected: crate::refs::store::Expected::Any,
                reflog: None,
                flags: Default::default(),
            });
            n_written += 1;
        }

        updates.push(RefUpdate {
            remote_ref: m.remote_ref.clone(),
            local_ref: Some(local_ref.clone()),
            old_oid: old,
            new_oid: Some(m.oid),
            mode,
            note: None,
        });
    }

    if !opts.dry_run && !store_batch.is_empty() {
        let t = std::time::Instant::now();
        crate::refs::commit_ref_store_batch(local_git_dir, &store_batch)?;
        d_write += t.elapsed();
    }

    net_trace!(
        opts.network_trace,
        opts.diagnostics.as_ref(),
        "fetch_remote: apply {:?} — repo_open {:?}, prune {:?}, packed_load {:?}, \
         resolve {:?}, classify {:?} ({} ancestry-checked), write {:?} ({} written)",
        t_apply.elapsed(),
        d_repo_open,
        d_prune,
        d_packed,
        d_resolve,
        d_classify,
        n_ancestry,
        d_write,
        n_written
    );

    net_trace!(
        opts.network_trace,
        opts.diagnostics.as_ref(),
        "fetch_remote: done — {} ref update(s){} [+{:?} total]",
        updates.len(),
        default_branch
            .as_deref()
            .map(|b| format!(", default branch '{b}'"))
            .unwrap_or_default(),
        t_begin.elapsed()
    );
    finish_initial_remote_fetch_layout(local_git_dir, opts, default_branch.as_deref())?;
    if let Some(remote) = opts.remote_name.as_deref() {
        ensure_remote_tracking_head_symref(
            local_git_dir,
            remote,
            default_branch.as_deref(),
            opts.dry_run,
        )?;
    }
    crate::transfer::finish_anonymous_url_fetch(
        local_git_dir,
        opts,
        explicit_refspec_fetch,
        default_branch.as_deref(),
        &remote_refs,
        &matched,
    )?;
    Ok(FetchOutcome {
        updates,
        default_branch,
        new_shallow: crate::shallow::dedupe_shallow_list(shallow_update.shallow),
        new_unshallow: shallow_update.unshallow,
        transfer,
    })
}

/// Identity string for fetch/clone reflog entries (`Name <email> epoch tz`).
pub fn fetch_operation_identity(
    env: &crate::environment::Environment,
    git_dir: &Path,
) -> Result<String> {
    use crate::ident_config::ident_default_name;
    let cfg = crate::config::ConfigSet::load(env, Some(git_dir), true)?;
    let name = ident_default_name(&cfg, env);
    let email = cfg
        .get("user.email")
        .filter(|e| !e.trim().is_empty())
        .unwrap_or_else(|| "unknown".to_owned());
    let now = time::OffsetDateTime::now_utc();
    Ok(format!("{name} <{email}> {} +0000", now.unix_timestamp()))
}

/// Ensure `refs/remotes/<remote>/HEAD` exists when the remote default branch is known.
pub(crate) fn ensure_remote_tracking_head_symref(
    git_dir: &Path,
    remote: &str,
    default_branch: Option<&str>,
    dry_run: bool,
) -> Result<()> {
    if dry_run {
        return Ok(());
    }
    let Some(branch) = default_branch else {
        return Ok(());
    };
    let head_ref = format!("refs/remotes/{remote}/HEAD");
    if crate::refs::resolve_ref(git_dir, &head_ref).is_ok() {
        return Ok(());
    }
    let target = format!("refs/remotes/{remote}/{branch}");
    if crate::refs::resolve_ref(git_dir, &target).is_err() {
        return Ok(());
    }
    crate::refs::write_symbolic_ref(git_dir, &head_ref, &target)
}

pub(crate) fn finish_initial_remote_fetch_layout(
    git_dir: &Path,
    opts: &FetchOptions,
    default_branch: Option<&str>,
) -> Result<()> {
    if opts.dry_run || !opts.initial_remote_fetch {
        return Ok(());
    }
    let Some(remote) = opts.remote_name.as_deref() else {
        return Ok(());
    };
    crate::refs::pack_remote_tracking_refs_for_clone(git_dir, remote)?;
    if let Some(branch) = default_branch {
        let target = format!("refs/remotes/{remote}/{branch}");
        let head_ref = format!("refs/remotes/{remote}/HEAD");
        if crate::refs::resolve_ref(git_dir, &target).is_ok() {
            crate::refs::write_symbolic_ref(git_dir, &head_ref, &target)?;
            if let Some(ref log) = opts.clone_reflog {
                let zero = ObjectId::zero();
                let tip = crate::refs::resolve_ref(git_dir, &target).unwrap_or(zero);
                crate::refs::append_reflog(
                    git_dir,
                    &head_ref,
                    &zero,
                    &tip,
                    &log.identity,
                    &log.message,
                    true,
                )?;
            }
        }
    }
    Ok(())
}

/// Add advertised tags to the matched set per [`crate::transfer::TagMode`].
///
/// Over the wire we cannot peel remote tags before the pack arrives, so:
/// * `All` adds every advertised tag (and `want`s it unconditionally).
/// * `Following` provisionally adds every advertised tag here; unreachable ones
///   are dropped by [`retain_following_tags`] after the pack is ingested.
/// * `None` adds nothing.
///
/// Returns the oids of tags added under `Following` — the caller must keep these
/// out of the default `want` list so an unreachable tag does not drag its target
/// into the pack (which would make it look reachable and survive the prune).
/// Annotated tags whose peeled commit is already local are still wanted via
/// [`following_tag_wants_with_local_peel`].
///
/// Under `Following`, tags named in `already_local` (see [`tags_already_local`])
/// are skipped: the local repository already has them at the advertised value.
#[allow(clippy::too_many_arguments)]
pub(crate) fn add_wire_tags(
    mode: crate::transfer::TagMode,
    remote_refs: &[(String, ObjectId)],
    advertised_peel: &std::collections::HashMap<String, ObjectId>,
    negatives: &[RefspecItem],
    already_local: &HashSet<String>,
    matched: &mut Vec<crate::transfer::MatchedRef>,
    matched_oids: &mut HashSet<ObjectId>,
    seen_remote_ref: &mut HashSet<String>,
) -> HashSet<ObjectId> {
    let mut following_only: HashSet<ObjectId> = HashSet::new();
    if mode == crate::transfer::TagMode::None {
        return following_only;
    }
    for (name, oid) in remote_refs {
        if !name.starts_with("refs/tags/") {
            continue;
        }
        if seen_remote_ref.contains(name) || ref_excluded(name, negatives) {
            continue;
        }
        if mode == crate::transfer::TagMode::Following && already_local.contains(name) {
            continue;
        }
        seen_remote_ref.insert(name.clone());
        // Only tag oids newly added here belong in `following_only`. A lightweight
        // tag at the same commit as an already-matched branch tip must not suppress
        // the branch's `want` when building the pack request.
        if matched_oids.insert(*oid) && mode == crate::transfer::TagMode::Following {
            following_only.insert(*oid);
        }
        matched.push(crate::transfer::MatchedRef {
            remote_ref: name.clone(),
            local_ref: Some(name.clone()),
            oid: *oid,
            force: false,
            is_tag: true,
            dest_is_tag: true,
            advertised_peel: advertised_peel.get(name).copied(),
            fetch_head: crate::fetch_request::FetchHeadPolicy::ForMerge,
        });
    }
    following_only
}

/// Under `TagMode::Following`, tag OIDs in `following_only` that should still
/// be `want`ed: annotated tags pointing at a commit already present locally.
pub(crate) fn following_tag_wants_with_local_peel(
    matched: &[crate::transfer::MatchedRef],
    following_only: &HashSet<ObjectId>,
    local_odb: &crate::odb::Odb,
) -> HashSet<ObjectId> {
    let mut wants = HashSet::new();
    for m in matched {
        if !m.is_tag || !following_only.contains(&m.oid) {
            continue;
        }
        let Some(peel) = m.advertised_peel else {
            continue;
        };
        if local_odb.exists(&peel) && !local_odb.exists(&m.oid) {
            wants.insert(m.oid);
        }
    }
    wants
}

/// Commit OIDs reachable within an absolute shallow `depth` from each root,
/// walking all parents (upload-pack `deepen` shape), stopping at
/// `shallow_boundaries`.
pub(crate) fn shallow_fetched_commit_oids(
    odb: &crate::odb::Odb,
    roots: &[ObjectId],
    depth: u32,
    shallow_boundaries: &HashSet<ObjectId>,
) -> HashSet<ObjectId> {
    use crate::objects::{parse_commit, ObjectKind};
    use std::collections::{HashMap, VecDeque};

    if depth == 0 {
        return HashSet::new();
    }
    let max_depth = usize::try_from(depth).unwrap_or(usize::MAX);
    let mut best_depth: HashMap<ObjectId, usize> = HashMap::new();
    let mut q: VecDeque<(ObjectId, usize)> = VecDeque::new();
    for &root in roots {
        best_depth.insert(root, 1);
        q.push_back((root, 1));
    }
    let mut out = HashSet::new();
    while let Some((oid, d)) = q.pop_front() {
        if d > max_depth {
            continue;
        }
        if best_depth.get(&oid).copied() != Some(d) {
            continue;
        }
        out.insert(oid);
        if d == max_depth {
            continue;
        }
        if shallow_boundaries.contains(&oid) {
            continue;
        }
        if !odb.exists(&oid) {
            continue;
        }
        let Ok(obj) = odb.read(&oid) else {
            continue;
        };
        if obj.kind != ObjectKind::Commit {
            continue;
        }
        let Ok(c) = parse_commit(&obj.data) else {
            continue;
        };
        for p in c.parents {
            let nd = d.saturating_add(1);
            if nd > max_depth {
                continue;
            }
            let prev = best_depth.get(&p).copied().unwrap_or(usize::MAX);
            if nd < prev {
                best_depth.insert(p, nd);
                q.push_back((p, nd));
            }
        }
    }
    out
}

/// Commits whose peels count as "fetched" for shallow tag-following: branch tips
/// when there is no absolute depth, otherwise the depth-limited closure.
pub(crate) fn following_peel_commit_set_for_shallow(
    odb: &crate::odb::Odb,
    commit_tips: &HashSet<ObjectId>,
    depth: Option<u32>,
    shallow_boundaries: &HashSet<ObjectId>,
) -> HashSet<ObjectId> {
    let roots: Vec<ObjectId> = commit_tips.iter().copied().collect();
    match depth.filter(|d| *d > 0) {
        Some(d) => shallow_fetched_commit_oids(odb, &roots, d, shallow_boundaries),
        None => commit_tips.clone(),
    }
}

/// On a shallow fetch, annotated tags whose peel is a branch tip being fetched
/// must be copied even when the peel is not local yet (matches upload-pack
/// include-tag under depth).
pub(crate) fn following_tag_wants_peel_among_fetched_commits(
    matched: &[crate::transfer::MatchedRef],
    following_only: &HashSet<ObjectId>,
    fetch_commit_oids: &HashSet<ObjectId>,
) -> HashSet<ObjectId> {
    let mut wants = HashSet::new();
    for m in matched {
        if !m.is_tag || !following_only.contains(&m.oid) {
            continue;
        }
        let Some(peel) = m.advertised_peel else {
            continue;
        };
        if peel != m.oid && fetch_commit_oids.contains(&peel) {
            wants.insert(m.oid);
        }
    }
    wants
}

/// Drop provisional `Following` tags whose object (or peeled target) did not
/// arrive in the fetched pack — i.e. is not reachable from the other matched,
/// non-tag refs we fetched. Matches `git fetch`'s default tag-following: a tag
/// is kept when it points into the fetched history.
pub(crate) fn retain_following_tags(
    local_odb: &crate::odb::Odb,
    matched: &mut Vec<crate::transfer::MatchedRef>,
    pack_oids: &HashSet<ObjectId>,
    shallow_boundaries: &HashSet<ObjectId>,
) -> Result<()> {
    if !matched.iter().any(|m| m.is_tag) {
        return Ok(());
    }

    let roots: Vec<ObjectId> = matched
        .iter()
        .filter(|m| !m.is_tag)
        .map(|m| m.oid)
        .collect();
    let peeled_by_tag: std::collections::HashMap<ObjectId, ObjectId> = matched
        .iter()
        .filter(|m| m.is_tag)
        .map(|m| {
            let peeled = m
                .advertised_peel
                .unwrap_or_else(|| local_odb.peel_tags(m.oid).map_or(m.oid, |(oid, _)| oid));
            (m.oid, peeled)
        })
        .collect();
    let targets: HashSet<ObjectId> = peeled_by_tag.values().copied().collect();
    let commit_reach = reachable_commits(local_odb, &roots, &targets, shallow_boundaries);
    let pack_reach = pack_delivered_reachable(local_odb, &roots, pack_oids)?;
    matched.retain(|m| {
        if !m.is_tag {
            return true;
        }
        let peeled = peeled_by_tag.get(&m.oid).copied().unwrap_or(m.oid);
        let reachable = commit_reach.contains(&peeled) || pack_reach.contains(&peeled);
        if !reachable {
            return false;
        }
        // Lightweight tags store the peeled commit directly; no tag object is required.
        if m.oid == peeled {
            return true;
        }
        // Annotated tags need their tag object (new in this fetch or already local).
        pack_oids.contains(&m.oid) || local_odb.exists(&m.oid)
    });
    Ok(())
}

/// Objects reachable from `roots` by following only links whose target oid is in `pack_oids`.
fn pack_delivered_reachable(
    odb: &crate::odb::Odb,
    roots: &[ObjectId],
    pack_oids: &HashSet<ObjectId>,
) -> Result<HashSet<ObjectId>> {
    use crate::objects::{parse_commit, parse_tag, parse_tree, ObjectKind};

    let mut seen: HashSet<ObjectId> = HashSet::new();
    let mut stack: Vec<ObjectId> = roots
        .iter()
        .copied()
        .filter(|oid| pack_oids.contains(oid))
        .collect();
    while let Some(oid) = stack.pop() {
        if !seen.insert(oid) {
            continue;
        }
        let obj = odb.read(&oid)?;
        match obj.kind {
            ObjectKind::Commit => {
                let c = parse_commit(&obj.data)?;
                if pack_oids.contains(&c.tree) {
                    stack.push(c.tree);
                }
                for p in c.parents {
                    if pack_oids.contains(&p) {
                        stack.push(p);
                    }
                }
            }
            ObjectKind::Tag => {
                let t = parse_tag(&obj.data)?;
                if pack_oids.contains(&t.object) {
                    stack.push(t.object);
                }
            }
            ObjectKind::Tree => {
                let entries = parse_tree(&obj.data)?;
                for entry in entries {
                    if pack_oids.contains(&entry.oid) {
                        stack.push(entry.oid);
                    }
                }
            }
            ObjectKind::Blob => {}
        }
    }
    Ok(seen)
}

/// Names of the remote `refs/tags/*` entries in `remote_refs` that the local
/// repository already stores under the same name at the same object id.
///
/// Tag-following never needs to consider these: Git's auto-follow skips tags
/// that already exist locally, and there is nothing to fetch or write for them.
/// Excluding them up front keeps the reachability walk off the whole history
/// when the client already has every advertised tag (the common incremental
/// fetch).
///
/// # Parameters
///
/// * `local_git_dir` — the fetching repository's git directory.
/// * `remote_refs` — advertised `(name, oid)` pairs (peeled `^{}` entries excluded).
///
/// Unreadable local refs count as absent, so they still go through the regular
/// tag-following path.
pub(crate) fn tags_already_local(
    local_git_dir: &Path,
    remote_refs: &[(String, ObjectId)],
) -> HashSet<String> {
    let mut remote_tags = remote_refs
        .iter()
        .filter(|(name, _)| name.starts_with("refs/tags/"))
        .peekable();
    if remote_tags.peek().is_none() {
        return HashSet::new();
    }
    let Ok(resolver) = crate::refs::CachedRefResolver::open(local_git_dir) else {
        return HashSet::new();
    };
    remote_tags
        .filter(|(name, oid)| resolver.resolve(name).ok().flatten() == Some(*oid))
        .map(|(name, _)| name.clone())
        .collect()
}

/// Commit-level reachability from `roots`: the set of commit OIDs reachable by
/// following commit parents (peeling annotated tags to their target). Unlike
/// [`reachable_closure`] it never descends into trees or blobs, so it stays cheap
/// across thousands of heads. Sufficient for tag-following retention, whose only
/// question is whether a tag's target commit is reachable from the fetched heads.
///
/// The walk stops as soon as every oid in `targets` has been reached, so the
/// returned set is only guaranteed complete for membership tests of `targets`
/// (and is the full closure whenever some target is unreachable). An empty
/// `targets` set returns immediately with an empty result.
pub(crate) fn reachable_commits(
    odb: &crate::odb::Odb,
    roots: &[ObjectId],
    targets: &HashSet<ObjectId>,
    shallow_boundaries: &HashSet<ObjectId>,
) -> HashSet<ObjectId> {
    use crate::objects::{parse_commit, parse_tag, ObjectKind};

    let mut seen: HashSet<ObjectId> = HashSet::new();
    let mut remaining: HashSet<ObjectId> = targets.clone();
    if remaining.is_empty() {
        return seen;
    }

    // Read commit parents from the commit-graph file when present — an mmap-style
    // lookup that avoids decompressing/parsing every commit object (the dominant
    // cost on large repos). Commits absent from the graph (recently fetched) and
    // octopus merges fall back to the object read below.
    let graph = crate::commit_graph_file::CommitGraphChain::load(odb.objects_dir());

    let mut stack: Vec<ObjectId> = roots.to_vec();
    while let Some(oid) = stack.pop() {
        if remaining.is_empty() {
            break;
        }
        if seen.contains(&oid) {
            continue;
        }
        if shallow_boundaries.contains(&oid) {
            if odb.exists(&oid) {
                seen.insert(oid);
                remaining.remove(&oid);
            }
            continue;
        }
        if let Some(graph) = graph.as_ref() {
            if let Some((parents, _time)) = graph.graph_commit(&oid) {
                seen.insert(oid);
                remaining.remove(&oid);
                stack.extend(parents);
                continue;
            }
        }
        let Ok(obj) = odb.read(&oid) else {
            continue;
        };
        seen.insert(oid);
        remaining.remove(&oid);
        match obj.kind {
            ObjectKind::Commit => {
                if let Ok(c) = parse_commit(&obj.data) {
                    for p in c.parents {
                        stack.push(p);
                    }
                }
            }
            ObjectKind::Tag => {
                if let Ok(t) = parse_tag(&obj.data) {
                    stack.push(t.object);
                }
            }
            // Tree/blob reachability is irrelevant to tag-following; don't descend.
            ObjectKind::Tree | ObjectKind::Blob => {}
        }
    }
    seen
}

#[cfg(test)]
mod fetch_advertised_ref_tests {
    use super::*;
    use crate::objects::ObjectId;
    use std::io::Cursor;

    #[test]
    fn fetch_ref_list_prefixes_includes_refs_dwim_for_pull() {
        let specs = vec!["pull/123/head:pr-123".to_owned()];
        let prefixes = fetch_ref_list_prefixes(&specs);
        assert!(prefixes.iter().any(|p| p == "refs/pull/123/head"));
        assert!(prefixes.iter().any(|p| p == "refs/heads/"));
    }

    #[test]
    fn build_v2_ls_refs_request_includes_unborn_when_advertised() {
        use crate::odb::Odb;
        use crate::transfer::TagMode;

        let caps = vec!["ls-refs=unborn".to_owned()];
        let odb = Odb::new(std::path::Path::new("/dev/null"));
        let req = build_v2_ls_refs_request(&caps, &odb, TagMode::None, &[]).unwrap();
        let text = String::from_utf8_lossy(&req);
        assert!(
            text.contains("unborn"),
            "request must ask for unborn HEAD:\n{text}"
        );
    }

    #[test]
    fn parse_v2_ls_refs_unborn_head_symref_line() {
        let mut body = Vec::new();
        crate::pkt_line::write_line(&mut body, "unborn HEAD symref-target:refs/heads/trunk")
            .unwrap();
        crate::pkt_line::write_flush(&mut body).unwrap();

        let (_, head, _, _) = parse_v2_ls_refs_response(&mut Cursor::new(body)).unwrap();
        assert_eq!(head.as_deref(), Some("refs/heads/trunk"));
    }

    #[test]
    fn parse_v2_ls_refs_drops_traversal_ref() {
        let oid = ObjectId::from_hex("aabbccddeeff00112233445566778899aabbccdd").unwrap();
        let line = format!("{oid} refs/heads/../../../config");
        let mut body = Vec::new();
        crate::pkt_line::write_line(&mut body, &line).unwrap();
        crate::pkt_line::write_flush(&mut body).unwrap();

        let (refs, _, _, _) = parse_v2_ls_refs_response(&mut Cursor::new(body)).unwrap();
        assert!(refs.is_empty());
    }

    #[test]
    fn refspec_maps_traversal_to_non_storable_local_name() {
        use crate::refs::is_valid_storable_ref_name;
        use crate::refspec::parse_fetch_refspec;
        use crate::transfer::match_positive;

        let spec = parse_fetch_refspec("+refs/heads/*:refs/remotes/origin/*").unwrap();
        let positive = vec![spec];
        let local = match_positive("refs/heads/../../../config", &positive)
            .flatten()
            .expect("wildcard matches malicious remote name");
        assert!(!is_valid_storable_ref_name(&local));
    }

    #[test]
    fn add_wire_tags_following_skips_following_only_when_oid_already_matched() {
        use crate::transfer::TagMode;
        use std::collections::{HashMap, HashSet};

        let shared = ObjectId::from_hex("4eda3c62a0f0a0a0a0a0a0a0a0a0a0a0a0a0a0a0").unwrap();
        let unreachable_tag =
            ObjectId::from_hex("1111111111111111111111111111111111111111").unwrap();
        let remote_refs = vec![
            ("refs/heads/rel".to_owned(), shared),
            ("refs/tags/light".to_owned(), shared),
            ("refs/tags/orphan".to_owned(), unreachable_tag),
        ];
        let advertised_peel = HashMap::new();
        let negatives: Vec<crate::refspec::RefspecItem> = Vec::new();
        let mut matched = Vec::new();
        let mut matched_oids = HashSet::from([shared]);
        let mut seen = HashSet::from(["refs/heads/rel".to_owned()]);
        matched.push(crate::transfer::MatchedRef {
            remote_ref: "refs/heads/rel".to_owned(),
            local_ref: Some("refs/remotes/origin/rel".to_owned()),
            oid: shared,
            force: false,
            is_tag: false,
            dest_is_tag: false,
            advertised_peel: None,
            fetch_head: crate::fetch_request::FetchHeadPolicy::ForMerge,
        });

        let following_only = add_wire_tags(
            TagMode::Following,
            &remote_refs,
            &advertised_peel,
            &negatives,
            &HashSet::new(),
            &mut matched,
            &mut matched_oids,
            &mut seen,
        );

        assert!(
            !following_only.contains(&shared),
            "shared branch/tag tip must stay wanted"
        );
        assert!(
            following_only.contains(&unreachable_tag),
            "tag-only oid stays following-only (no upfront want)"
        );
        assert_eq!(matched.iter().filter(|m| m.is_tag).count(), 2);
    }

    #[test]
    fn following_tag_wants_includes_annotated_tag_when_peel_is_local() {
        use std::collections::HashSet;

        let tag_obj = ObjectId::from_hex("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb").unwrap();
        let tmp = tempfile::tempdir().expect("tempdir");
        let git_dir = tmp.path().join(".git");
        std::fs::create_dir_all(git_dir.join("objects")).expect("objects");
        let odb = crate::odb::Odb::new(git_dir.join("objects").as_path());
        // Seed only the peeled commit (tag object absent).
        let commit_bytes = b"tree 0000000000000000000000000000000000000000\nauthor T <t@example.com> 0 +0000\ncommitter T <t@example.com> 0 +0000\n\nx\n";
        let commit = odb
            .write_loose_materialize(crate::objects::ObjectKind::Commit, commit_bytes)
            .expect("write commit");

        let following_only = HashSet::from([tag_obj]);
        let matched = vec![crate::transfer::MatchedRef {
            remote_ref: "refs/tags/v".to_owned(),
            local_ref: Some("refs/tags/v".to_owned()),
            oid: tag_obj,
            force: false,
            is_tag: true,
            dest_is_tag: true,
            advertised_peel: Some(commit),
            fetch_head: crate::fetch_request::FetchHeadPolicy::ForMerge,
        }];
        let wants = following_tag_wants_with_local_peel(&matched, &following_only, &odb);
        assert!(wants.contains(&tag_obj));
    }

    #[test]
    fn following_tag_wants_peel_among_fetched_commits_before_peel_is_local() {
        use std::collections::HashSet;

        let tag_obj = ObjectId::from_hex("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb").unwrap();
        let tip = ObjectId::from_hex("cccccccccccccccccccccccccccccccccccccccc").unwrap();
        let following_only = HashSet::from([tag_obj]);
        let fetch_commits = HashSet::from([tip]);
        let matched = vec![crate::transfer::MatchedRef {
            remote_ref: "refs/tags/v".to_owned(),
            local_ref: Some("refs/tags/v".to_owned()),
            oid: tag_obj,
            force: false,
            is_tag: true,
            dest_is_tag: true,
            advertised_peel: Some(tip),
            fetch_head: crate::fetch_request::FetchHeadPolicy::ForMerge,
        }];
        let wants = following_tag_wants_peel_among_fetched_commits(
            &matched,
            &following_only,
            &fetch_commits,
        );
        assert!(wants.contains(&tag_obj));
    }

    #[test]
    fn shallow_peel_set_includes_parent_at_depth_two_not_tip_only() {
        use std::collections::HashSet;

        let tmp = tempfile::tempdir().expect("tempdir");
        let objects = tmp.path().join("objects");
        std::fs::create_dir_all(&objects).expect("objects");
        let odb = crate::odb::Odb::new(objects.as_path());
        let chain = write_commit_chain(&odb, 3);
        let tip = *chain.last().expect("tip");
        let parent = chain[1];
        let tips = HashSet::from([tip]);
        let tips_only =
            following_peel_commit_set_for_shallow(&odb, &tips, Some(1), &HashSet::new());
        assert!(!tips_only.contains(&parent));
        let depth_two =
            following_peel_commit_set_for_shallow(&odb, &tips, Some(2), &HashSet::new());
        assert!(depth_two.contains(&parent));
    }

    /// Writes a linear chain of `n` commits (oldest first) into `odb`.
    fn write_commit_chain(odb: &crate::odb::Odb, n: usize) -> Vec<ObjectId> {
        let mut chain: Vec<ObjectId> = Vec::new();
        for i in 0..n {
            let parent = chain
                .last()
                .map(|p| format!("parent {}\n", p.to_hex()))
                .unwrap_or_default();
            let body = format!(
                "tree 4b825dc642cb6eb9a060e54bf8d69288fbee4904\n{parent}\
                 author T <t@example.com> {i} +0000\ncommitter T <t@example.com> {i} +0000\n\nc{i}\n"
            );
            let oid = odb
                .write_loose_materialize(crate::objects::ObjectKind::Commit, body.as_bytes())
                .expect("write commit");
            chain.push(oid);
        }
        chain
    }

    #[test]
    fn reachable_commits_stops_once_every_target_is_found() {
        use std::collections::HashSet;

        let tmp = tempfile::tempdir().expect("tempdir");
        let objects = tmp.path().join("objects");
        std::fs::create_dir_all(&objects).expect("objects");
        let odb = crate::odb::Odb::new(&objects);
        let chain = write_commit_chain(&odb, 5);
        let tip = chain[4];
        let none = HashSet::new();

        let near = HashSet::from([chain[3]]);
        let seen = reachable_commits(&odb, &[tip], &near, &none);
        assert!(seen.contains(&chain[3]), "reachable target is reported");
        assert!(
            !seen.contains(&chain[0]),
            "walk must stop before the root once every target is found"
        );

        let unreachable =
            HashSet::from([
                ObjectId::from_hex("1111111111111111111111111111111111111111").unwrap(),
            ]);
        let seen = reachable_commits(&odb, &[tip], &unreachable, &none);
        assert_eq!(
            seen,
            chain.iter().copied().collect::<HashSet<_>>(),
            "an unreachable target forces the full closure"
        );

        assert!(
            reachable_commits(&odb, &[tip], &HashSet::new(), &none).is_empty(),
            "no targets means no walk"
        );
    }

    #[test]
    fn tags_already_local_matches_name_and_oid_across_loose_and_packed() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let git_dir = tmp.path().join(".git");
        std::fs::create_dir_all(git_dir.join("objects")).expect("objects");
        std::fs::create_dir_all(git_dir.join("refs/tags")).expect("refs/tags");
        std::fs::write(git_dir.join("HEAD"), "ref: refs/heads/main\n").expect("HEAD");

        let a = ObjectId::from_hex("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa").unwrap();
        let b = ObjectId::from_hex("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb").unwrap();
        let c = ObjectId::from_hex("cccccccccccccccccccccccccccccccccccccccc").unwrap();
        std::fs::write(git_dir.join("refs/tags/loose"), format!("{}\n", a.to_hex()))
            .expect("loose tag");
        std::fs::write(
            crate::refs::store::paths::packed_refs_path(&git_dir),
            format!(
                "# pack-refs with: peeled fully-peeled sorted \n{} refs/tags/moved\n{} refs/tags/packed\n",
                a.to_hex(),
                b.to_hex()
            ),
        )
        .expect("packed-refs");

        let remote_refs = vec![
            ("refs/heads/main".to_owned(), a),
            ("refs/tags/loose".to_owned(), a),
            ("refs/tags/packed".to_owned(), b),
            ("refs/tags/moved".to_owned(), c),
            ("refs/tags/new".to_owned(), c),
        ];
        let present = tags_already_local(&git_dir, &remote_refs);
        assert_eq!(
            present,
            HashSet::from(["refs/tags/loose".to_owned(), "refs/tags/packed".to_owned()]),
            "only tags stored locally at the advertised oid count as present"
        );
        assert!(tags_already_local(&git_dir, &remote_refs[..1]).is_empty());
    }

    #[test]
    fn add_wire_tags_following_skips_tags_already_local() {
        use crate::transfer::TagMode;
        use std::collections::HashMap;

        let tip = ObjectId::from_hex("2222222222222222222222222222222222222222").unwrap();
        let old_tag = ObjectId::from_hex("3333333333333333333333333333333333333333").unwrap();
        let new_tag = ObjectId::from_hex("4444444444444444444444444444444444444444").unwrap();
        let remote_refs = vec![
            ("refs/heads/main".to_owned(), tip),
            ("refs/tags/old".to_owned(), old_tag),
            ("refs/tags/new".to_owned(), new_tag),
        ];
        let already_local = HashSet::from(["refs/tags/old".to_owned()]);

        for (mode, expect) in [
            (TagMode::Following, vec!["refs/tags/new"]),
            (TagMode::All, vec!["refs/tags/old", "refs/tags/new"]),
        ] {
            let mut matched = Vec::new();
            let mut matched_oids = HashSet::new();
            let mut seen = HashSet::new();
            add_wire_tags(
                mode,
                &remote_refs,
                &HashMap::new(),
                &[],
                &already_local,
                &mut matched,
                &mut matched_oids,
                &mut seen,
            );
            let names: Vec<&str> = matched.iter().map(|m| m.remote_ref.as_str()).collect();
            assert_eq!(names, expect, "{mode:?}");
        }
    }
}