grit_lib/

transfer.rs

//! Embedder-facing transfer (fetch / push) result & option types, plus the
//! negotiation-driven pack builder.
//!
//! This module is the foundation for the in-process fetch/push APIs that
//! embedders such as `jj` and GitButler consume in place of `gix` transport.
//! It defines the structured input/output types those APIs use and implements
//! the single most important primitive — [`build_pack`] — which packs **only**
//! the objects reachable from a negotiated set of `wants` and not already
//! reachable from the remote's `haves`.
//!
//! Scope note (phase 1): only the local / `file://` object+ref copy path is in
//! scope. `git://`, `http(s)`, and `ssh` transports plus credential-helper
//! execution are out of scope and are left as TODOs in later phases.
//!
//! Push *result* reporting reuses [`crate::push_report::PushRefResult`] /
//! [`crate::push_report::PushRefStatus`] rather than redefining it.

use crate::environment::RepositoryOptions;
use crate::partial_clone::ObjectFilterSpec;

use std::collections::{HashMap, HashSet, VecDeque};
use std::fs;
use std::path::{Path, PathBuf};

use crate::error::{Error, RemoteUnreadableReason, Result};
use crate::fetch::{NoProgress, Progress};
use crate::objects::{parse_commit, parse_tag, parse_tree, HashAlgo, ObjectId, ObjectKind};
use crate::odb::Odb;
use crate::progress::{ProgressPhase, ProgressSink};
use crate::push_report::{PushRefResult, PushRefStatus};
use crate::refspec::{parse_fetch_refspec, RefspecItem};
use std::num::NonZeroU32;

/// How a single reference resolved during a fetch (or would resolve in a push).
///
/// Mirrors the shapes of `gix::remote::fetch::refs::update::Mode` that `jj`
/// already consumes, so the embedder's translation layer stays a thin adapter.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum UpdateMode {
    /// The local tracking ref did not exist and was created.
    New,
    /// The update advanced the ref along its existing history.
    FastForward,
    /// A non-fast-forward update that was applied because force was requested.
    Forced,
    /// The local ref already matched the remote value; nothing to do.
    UpToDate,
    /// No change was required (e.g. a no-op refspec).
    NoChangeNeeded,
    /// A non-fast-forward update that was rejected (force not requested).
    NonFastForwardRejected,
    /// A tag update was rejected (tags are not overwritten without force).
    TagUpdateRejected,
    /// The source object named by the refspec was not found on the remote.
    SourceObjectNotFound,
    /// The remote ref is unborn (points at nothing yet).
    Unborn,
    /// A prune/delete was requested but the local ref was already missing.
    DeletedMissing,
    /// The local tracking ref could not be created or updated (namespace conflict, lock, …).
    RefLockRejected,
}

impl UpdateMode {
    /// Returns whether fetch applied this update to the local repository.
    ///
    /// Rejected or no-op outcomes (for example an existing tag that moved on the
    /// remote) are excluded so callers can report only refs that actually changed.
    #[must_use]
    pub fn modified_repo(self) -> bool {
        matches!(
            self,
            Self::New | Self::FastForward | Self::Forced | Self::DeletedMissing
        )
    }

    /// Returns whether fetch refused this update (non-fast-forward without force, a moved
    /// tag, or a source missing on the remote).
    #[must_use]
    pub fn is_rejected(self) -> bool {
        matches!(
            self,
            Self::NonFastForwardRejected
                | Self::TagUpdateRejected
                | Self::SourceObjectNotFound
                | Self::RefLockRejected
        )
    }

    /// Returns whether this update belongs in a fetch report: the ref changed
    /// ([`Self::modified_repo`]) or the update was refused ([`Self::is_rejected`]).
    ///
    /// Up-to-date, no-op and unborn refs are left out.
    #[must_use]
    pub fn is_reported(self) -> bool {
        self.modified_repo() || self.is_rejected()
    }
}

/// The resolved outcome of one reference during a fetch.
#[derive(Clone, Debug)]
pub struct RefUpdate {
    /// The remote-side ref name (e.g. `refs/heads/main`).
    pub remote_ref: String,
    /// The local-side ref name written, if any (e.g. `refs/remotes/origin/main`).
    pub local_ref: Option<String>,
    /// Previous value of the local ref (`None` when newly created).
    pub old_oid: Option<ObjectId>,
    /// New value written to the local ref (`None` for deletions / unborn).
    pub new_oid: Option<ObjectId>,
    /// How the update resolved.
    pub mode: UpdateMode,
    /// Optional human-readable note (reason text), for embedder display.
    pub note: Option<String>,
}

/// Which tags to fetch alongside the requested refs.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Default)]
pub enum TagMode {
    /// Do not fetch any tags automatically.
    None,
    /// Fetch tags that point into the history being fetched (like `git fetch`'s
    /// default tag-following; not what `git clone` does).
    ///
    /// A tag is kept when the object it peels to arrives with this fetch, when
    /// it is a commit already present locally and reachable from the fetched
    /// refs, or when it is a tree or blob already present locally. Annotated
    /// tags also need their tag object. Deepening or unshallowing follows tags
    /// into the newly available history in the same fetch, without letting
    /// automatic tags expand the requested branch depth.
    ///
    /// Git also keeps a tag on a local commit
    /// that the fetched refs do not reach; grit does not. Deciding this reads
    /// no objects for delivered targets, walks only commits for local ones, and
    /// never lazy-fetches in a partial clone, so it stays cheap on large
    /// histories.
    ///
    /// Tags the local repository already has at the advertised value are not
    /// considered again, so they do not appear in [`FetchOutcome::updates`].
    #[default]
    Following,
    /// Fetch every advertised `refs/tags/*` ref (same idea as `git clone` and
    /// `git fetch --tags`).
    All,
}

/// Shallow-history control for fetch and clone (`--depth`, `--deepen`, `--unshallow`).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum FetchDepth {
    /// Truncate or establish history at the given absolute depth from each tip
    /// (`git fetch --depth N`).
    Depth(NonZeroU32),
    /// Deepen an existing shallow repository by the given number of commits
    /// (`git fetch --deepen N`, wire `deepen N` + `deepen-relative`).
    Deepen(NonZeroU32),
    /// Convert a shallow repository into a complete one (`git fetch --unshallow`).
    Unshallow,
}

impl FetchDepth {
    /// Build a [`FetchDepth::Depth`] when `n > 0`.
    #[must_use]
    pub fn depth(n: u32) -> Option<Self> {
        NonZeroU32::new(n).map(Self::Depth)
    }

    /// Build a [`FetchDepth::Deepen`] when `n > 0`.
    #[must_use]
    pub fn deepen(n: u32) -> Option<Self> {
        NonZeroU32::new(n).map(Self::Deepen)
    }
}

/// Options controlling a fetch.
pub struct FetchOptions {
    /// Positive refspecs selecting what to fetch.
    pub refspecs: Vec<String>,
    /// Negative refspecs excluding refs from the positive set.
    pub negative_refspecs: Vec<String>,
    /// Tag-following policy.
    pub tags: TagMode,
    /// Whether to prune local tracking refs that vanished on the remote.
    pub prune: bool,
    /// Compute and report updates without writing any refs or objects.
    pub dry_run: bool,
    /// Shallow fetch control (`--depth`, `--deepen`, `--unshallow`). `None` leaves
    /// history depth unchanged (aside from an existing on-disk `shallow` file).
    pub fetch_depth: Option<FetchDepth>,
    /// Deepen history to include commits no older than this cutoff
    /// (`git fetch --shallow-since <date>`). The value is sent verbatim as the
    /// wire `deepen-since <value>`; callers should pass the Unix timestamp Git's
    /// `upload-pack` expects (a bare integer), not a human date string.
    pub deepen_since: Option<String>,
    /// Deepen history but stop at (exclude) commits reachable from these refs/oids
    /// (`git fetch --shallow-exclude <ref>`). Each entry is sent as a wire
    /// `deepen-not <ref>`.
    pub deepen_not: Vec<String>,
    /// First fetch populating a remote-tracking namespace (clone): write `packed-refs`
    /// and `refs/remotes/{remote}/HEAD`.
    pub initial_remote_fetch: bool,
    /// Remote name paired with [`Self::initial_remote_fetch`] (e.g. `"origin"`).
    pub remote_name: Option<String>,
    /// Source label written into [`FETCH_HEAD`](https://git-scm.com/docs/git-fetch#_discussion).
    /// [`crate::remote::Remote::fetch`] sets this from the fetch URL when unset.
    pub fetch_source: Option<String>,
    /// Configured `remote.<name>.fetch` refspecs (for opportunistic tracking when [`Self::refspecs`]
    /// overrides them). [`crate::remote::Remote::fetch`] fills this for named remotes.
    pub configured_refspecs: Vec<String>,
    /// Clone-only reflog entries for `refs/remotes/{remote}/HEAD` (not remote branches).
    pub clone_reflog: Option<CloneReflog>,
    /// Diagnostic sink for [`crate::diagnostics::Trace::Network`] events.
    pub diagnostics: Option<crate::diagnostics::DiagnosticsHandle>,
    /// When true with [`Self::diagnostics`], fetch paths may emit network trace events.
    pub network_trace: bool,
    /// Override worker threads for index-pack after a received pack (`None` = config/default).
    pub index_pack_threads: Option<usize>,
    /// Global / embedder thread policy (`--threads`); applied when [`Self::index_pack_threads`] is unset.
    pub thread_policy: crate::parallel::ThreadPolicy,
    /// Discovery/config environment for pack ingest (`GIT_CONFIG_*`, `-c`); defaults to empty.
    pub environment: crate::environment::Environment,
    /// When true, spool the negotiated pack to disk and return without index-pack or ref updates.
    pub pack_spool_only: bool,
    /// Override the HTTP `Git-Protocol` header for this fetch (`None` → client default, usually v2).
    pub http_protocol_version: Option<u8>,
    /// When true, omit the `no-progress` capability/argument so the remote may
    /// send side-band channel-2 progress. Default false preserves today's wire bytes.
    pub remote_progress: bool,
    /// Object filter for partial clone (`blob:none`, `blob:limit=<n>`). Omitted when
    /// unset; promisor remotes may reapply `remote.*.partialclonefilter` automatically.
    pub filter: Option<ObjectFilterSpec>,
    /// When true, the received pack is written with a sibling `.promisor` marker.
    ///
    /// Set by the fetch driver when a filter was sent on the wire; callers should
    /// leave this `false`.
    pub mark_received_pack_promisor: bool,
    /// Explicit object ids to fetch (promisor on-demand); skips refspec matching and ref updates.
    pub object_wants: Vec<ObjectId>,
    /// On-demand promisor object fetch: always mark the received pack promisor.
    pub promisor_object_fetch: bool,
}

impl Default for FetchOptions {
    fn default() -> Self {
        Self {
            refspecs: Vec::new(),
            negative_refspecs: Vec::new(),
            tags: TagMode::Following,
            prune: false,
            dry_run: false,
            fetch_depth: None,
            deepen_since: None,
            deepen_not: Vec::new(),
            initial_remote_fetch: false,
            remote_name: None,
            fetch_source: None,
            configured_refspecs: Vec::new(),
            clone_reflog: None,
            diagnostics: None,
            network_trace: false,
            index_pack_threads: None,
            thread_policy: crate::parallel::ThreadPolicy::default(),
            environment: crate::environment::Environment::empty(),
            pack_spool_only: false,
            http_protocol_version: None,
            remote_progress: false,
            filter: None,
            mark_received_pack_promisor: false,
            object_wants: Vec::new(),
            promisor_object_fetch: false,
        }
    }
}

/// Identity and message for clone reflog entries written by the library.
#[derive(Clone, Debug)]
pub struct CloneReflog {
    /// Committer identity for the clone reflog line.
    pub identity: crate::ident::Signature,
    /// Reflog message (e.g. `clone: from <url>`).
    pub message: String,
}

impl Clone for FetchOptions {
    fn clone(&self) -> Self {
        Self {
            refspecs: self.refspecs.clone(),
            negative_refspecs: self.negative_refspecs.clone(),
            tags: self.tags,
            prune: self.prune,
            dry_run: self.dry_run,
            fetch_depth: self.fetch_depth,
            deepen_since: self.deepen_since.clone(),
            deepen_not: self.deepen_not.clone(),
            initial_remote_fetch: self.initial_remote_fetch,
            remote_name: self.remote_name.clone(),
            fetch_source: self.fetch_source.clone(),
            configured_refspecs: self.configured_refspecs.clone(),
            clone_reflog: self.clone_reflog.clone(),
            diagnostics: self.diagnostics.clone(),
            network_trace: self.network_trace,
            index_pack_threads: self.index_pack_threads,
            thread_policy: self.thread_policy,
            environment: self.environment.clone(),
            pack_spool_only: self.pack_spool_only,
            http_protocol_version: self.http_protocol_version,
            remote_progress: self.remote_progress,
            filter: self.filter,
            mark_received_pack_promisor: self.mark_received_pack_promisor,
            object_wants: self.object_wants.clone(),
            promisor_object_fetch: self.promisor_object_fetch,
        }
    }
}

impl FetchOptions {
    /// Whether this fetch carries any shallow/deepen request (an explicit
    /// `fetch_depth`/`deepen-since`/`deepen-not`). Note this does NOT cover the
    /// "already shallow, fetching more of the same boundary" case — that is
    /// driven by the on-disk `shallow` file, checked separately by the fetch
    /// paths via [`crate::shallow::load_shallow_oids`].
    #[must_use]
    pub fn has_deepen_request(&self) -> bool {
        self.fetch_depth.is_some()
            || self
                .deepen_since
                .as_deref()
                .is_some_and(|v| !v.trim().is_empty())
            || self.deepen_not.iter().any(|v| !v.trim().is_empty())
    }

    /// Wire `deepen N` value and whether `deepen-relative` must follow.
    #[must_use]
    pub(crate) fn wire_deepen(&self) -> Option<(u32, bool)> {
        match self.fetch_depth {
            Some(FetchDepth::Depth(n)) => Some((n.get(), false)),
            Some(FetchDepth::Deepen(n)) => Some((n.get(), true)),
            Some(FetchDepth::Unshallow) => Some((crate::shallow::INFINITE_DEPTH, false)),
            None => None,
        }
    }

    /// Absolute depth hint for tag-following on a shallow fetch (`--depth` only).
    #[must_use]
    pub(crate) fn absolute_depth_for_tags(&self) -> Option<u32> {
        match self.fetch_depth {
            Some(FetchDepth::Depth(n)) => Some(n.get()),
            Some(FetchDepth::Deepen(_)) | Some(FetchDepth::Unshallow) | None => None,
        }
    }
}

/// Validate shallow fetch options against the local repository before negotiation.
///
/// # Errors
///
/// Returns [`Error::FetchUnshallowNotShallow`] when [`FetchDepth::Unshallow`] is
/// requested but the repository has no shallow boundaries.
pub fn validate_fetch_depth_local(local_git_dir: &Path, opts: &FetchOptions) -> Result<()> {
    if matches!(opts.fetch_depth, Some(FetchDepth::Unshallow)) {
        let local = crate::shallow::load_shallow_oids(local_git_dir)?;
        if local.is_empty() {
            return Err(Error::FetchUnshallowNotShallow);
        }
    }
    Ok(())
}

/// Require server shallow/deepen capabilities for the requested fetch options.
pub(crate) fn validate_fetch_depth_server(
    opts: &FetchOptions,
    v0_caps: Option<&str>,
    v2_caps: Option<&[String]>,
) -> Result<()> {
    let needs_shallow = opts.has_deepen_request()
        || 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 !needs_shallow {
        return Ok(());
    }
    if let Some(caps) = v0_caps {
        if !caps.contains("shallow") {
            return Err(Error::FetchShallowUnsupported);
        }
        if opts.wire_deepen().is_some_and(|(_, rel)| rel) && !caps.contains("deepen-relative") {
            return Err(Error::FetchDeepenRelativeUnsupported);
        }
    } else if let Some(caps) = v2_caps {
        if !crate::protocol_v2::fetch_supports_feature(caps, "shallow") {
            return Err(Error::FetchShallowUnsupported);
        }
    }
    Ok(())
}

/// Resolve whether a fetch should prune stale remote-tracking refs.
///
/// When `cli_prune` is [`Some`], that value wins (matching `--prune` /
/// `--no-prune`). Otherwise reads `remote.<remote>.prune`, then `fetch.prune`,
/// in Git precedence order. Unset keys default to `false`.
///
/// # Errors
///
/// Returns [`Error::Config`] when a configured prune key is not a valid boolean.
pub fn resolve_fetch_prune(
    config: &crate::config::ConfigSet,
    remote: &str,
    cli_prune: Option<bool>,
) -> Result<bool> {
    if let Some(enabled) = cli_prune {
        return Ok(enabled);
    }
    let remote_key = format!("remote.{remote}.prune");
    if let Some(value) = optional_config_bool(config, &remote_key)? {
        return Ok(value);
    }
    if let Some(value) = optional_config_bool(config, "fetch.prune")? {
        return Ok(value);
    }
    Ok(false)
}

/// Resolve tag-following policy for a fetch.
///
/// When `cli_tags` is [`Some`], `true` maps to [`TagMode::All`] (`--tags`) and
/// `false` to [`TagMode::None`] (`--no-tags`). Otherwise reads
/// `remote.<remote>.tagOpt` (`--tags` / `--no-tags`, Git-compatible) and defaults
/// to [`TagMode::Following`].
#[must_use]
pub fn resolve_fetch_tags(
    config: &crate::config::ConfigSet,
    remote: &str,
    cli_tags: Option<bool>,
) -> TagMode {
    if let Some(all) = cli_tags {
        return if all { TagMode::All } else { TagMode::None };
    }
    let key = format!("remote.{remote}.tagOpt");
    match config.get(&key).as_deref().map(str::trim) {
        Some("--tags") | Some("tags") | Some("all") => TagMode::All,
        Some("--no-tags") | Some("no-tags") | Some("no") | Some("never") => TagMode::None,
        _ => TagMode::Following,
    }
}

fn optional_config_bool(config: &crate::config::ConfigSet, key: &str) -> Result<Option<bool>> {
    match config.get_bool(key) {
        None => Ok(None),
        Some(Ok(value)) => Ok(Some(value)),
        Some(Err(message)) => Err(Error::Config(crate::error::ConfigError::Other(format!(
            "invalid boolean for '{key}': {message}"
        )))),
    }
}

/// Object and byte counts for objects received during a fetch or clone.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct TransferStats {
    /// Objects installed from the received pack (zero when nothing was transferred).
    pub objects: u64,
    /// Size in bytes of the received pack file (zero when nothing was transferred).
    pub bytes: u64,
    /// Objects by type, when the pack was indexed locally (`None` when packs were
    /// linked or copied from a local repository without being re-indexed).
    pub kinds: Option<crate::index_pack::ObjectKindCounts>,
}

impl TransferStats {
    /// Build stats from an ingested pack (installed under `objects/pack/` or written
    /// into a custom primary store).
    #[must_use]
    pub fn from_ingested_pack(ingested: &crate::index_pack::IngestedPack) -> Self {
        Self {
            objects: u64::try_from(ingested.object_ids.len()).unwrap_or(u64::MAX),
            bytes: ingested.pack_bytes,
            kinds: Some(ingested.object_counts),
        }
    }
}

/// Totals across several transfers, such as every remote of a fetch-all.
///
/// Counts and bytes add up. [`TransferStats::kinds`] stays known only while every
/// transfer that received objects reported its type breakdown.
impl std::ops::AddAssign for TransferStats {
    fn add_assign(&mut self, other: Self) {
        self.kinds = match (self.kinds, other.kinds) {
            (Some(mut mine), Some(theirs)) => {
                mine += theirs;
                Some(mine)
            }
            (mine, None) if other.objects == 0 => mine,
            (None, theirs) if self.objects == 0 => theirs,
            _ => None,
        };
        self.objects += other.objects;
        self.bytes += other.bytes;
    }
}

/// The structured result of a fetch, ready for the embedder's ref-store apply.
#[derive(Clone, Debug, Default)]
pub struct FetchOutcome {
    /// Per-ref resolved updates.
    pub updates: Vec<RefUpdate>,
    /// The remote's default branch (from `HEAD` symref), if known.
    pub default_branch: Option<String>,
    /// New shallow boundary commits the server reported (`shallow <oid>`), already
    /// applied to the local `shallow` file. The commits' parents are intentionally
    /// absent from the local object store after this fetch.
    pub new_shallow: Vec<ObjectId>,
    /// Commits the server reported as no longer shallow (`unshallow <oid>`), i.e.
    /// boundaries removed from the local `shallow` file because their history is
    /// now complete. Populated by a deepen / `--unshallow` fetch.
    pub new_unshallow: Vec<ObjectId>,
    /// Objects and pack bytes ingested during this fetch.
    pub transfer: TransferStats,
}

impl FetchOutcome {
    /// Count of remote branch refs represented in [`Self::updates`].
    #[must_use]
    pub fn branch_count(&self) -> u64 {
        self.updates
            .iter()
            .filter(|u| u.remote_ref.starts_with("refs/heads/"))
            .count() as u64
    }

    /// Count of remote tag refs represented in [`Self::updates`] (peeled lines excluded).
    #[must_use]
    pub fn tag_count(&self) -> u64 {
        self.updates
            .iter()
            .filter(|u| u.remote_ref.starts_with("refs/tags/") && !u.remote_ref.ends_with("^{}"))
            .count() as u64
    }

    /// Updates worth reporting to a user, in fetch order ([`UpdateMode::is_reported`]).
    pub fn reported_updates(&self) -> impl Iterator<Item = &RefUpdate> {
        self.updates.iter().filter(|u| u.mode.is_reported())
    }

    /// Number of local refs this fetch changed ([`UpdateMode::modified_repo`]).
    ///
    /// For a [`FetchOptions::dry_run`] fetch this counts the refs that *would* change;
    /// nothing was written.
    #[must_use]
    pub fn changed_ref_count(&self) -> usize {
        self.updates
            .iter()
            .filter(|u| u.mode.modified_repo())
            .count()
    }

    /// Whether any ref update was refused ([`UpdateMode::is_rejected`]).
    #[must_use]
    pub fn has_rejections(&self) -> bool {
        self.updates.iter().any(|u| u.mode.is_rejected())
    }
}

fn object_kind_counts_for_oids(
    odb: &Odb,
    oids: &HashSet<ObjectId>,
) -> Result<crate::index_pack::ObjectKindCounts> {
    use crate::index_pack::ObjectKindCounts;
    let mut kinds = ObjectKindCounts::default();
    for oid in oids {
        kinds.add(odb.read_info(oid)?.kind);
    }
    Ok(kinds)
}

fn loose_transfer_stats(objects_dir: &Path, oids: &HashSet<ObjectId>) -> Result<(u64, u64)> {
    let mut objects = 0u64;
    let mut bytes = 0u64;
    for oid in oids {
        let path = oid.loose_path_in(objects_dir);
        if path.is_file() {
            objects += 1;
            bytes += fs::metadata(&path).map_err(Error::Io)?.len();
        }
    }
    Ok((objects, bytes))
}

fn pack_file_basenames(pack_dir: &Path) -> HashSet<String> {
    let mut names = HashSet::new();
    let Ok(entries) = fs::read_dir(pack_dir) else {
        return names;
    };
    for entry in entries.flatten() {
        let path = entry.path();
        if path.extension().is_some_and(|e| e == "pack") {
            if let Some(name) = path.file_name() {
                names.insert(name.to_string_lossy().into_owned());
            }
        }
    }
    names
}

const LOCAL_LINK_KINDS_OID_PROBE_MAX: u64 = 4096;

fn oids_from_newly_linked_pack_indexes(
    local_odb: &Odb,
    pack_dir_before: &HashSet<String>,
) -> Result<Vec<ObjectId>> {
    let pack_dir = local_odb.objects_dir().join("pack");
    let mut oids = Vec::new();
    if !pack_dir.is_dir() {
        return Ok(oids);
    }
    for entry in fs::read_dir(&pack_dir).map_err(Error::Io)? {
        let path = entry.map_err(Error::Io)?.path();
        if path.extension().is_none_or(|e| e != "pack") {
            continue;
        }
        let Some(name) = path.file_name() else {
            continue;
        };
        if pack_dir_before.contains(name.to_string_lossy().as_ref()) {
            continue;
        }
        let idx = path.with_extension("idx");
        if !idx.is_file() {
            continue;
        }
        let index = crate::pack::read_pack_index_no_verify(&idx)?;
        for entry in index.iter() {
            let oid = ObjectId::from_bytes(entry.oid())?;
            oids.push(oid);
        }
    }
    Ok(oids)
}

fn enrich_local_link_transfer_kinds(
    local_odb: &Odb,
    pack_dir_before: &HashSet<String>,
    loose_reachable: &HashSet<ObjectId>,
    transfer: &mut TransferStats,
) -> Result<()> {
    if transfer.kinds.is_some() || transfer.objects == 0 {
        return Ok(());
    }
    if transfer.objects > LOCAL_LINK_KINDS_OID_PROBE_MAX {
        return Ok(());
    }
    let mut probe: HashSet<ObjectId> = loose_reachable.clone();
    for oid in oids_from_newly_linked_pack_indexes(local_odb, pack_dir_before)? {
        probe.insert(oid);
    }
    if u64::try_from(probe.len()).unwrap_or(u64::MAX) != transfer.objects {
        return Ok(());
    }
    transfer.kinds = Some(object_kind_counts_for_oids(local_odb, &probe)?);
    Ok(())
}

/// Per-fetch transfer stats for objects installed during one local copy operation.
fn transfer_stats_for_local_operation(
    local_odb: &Odb,
    installed: &HashSet<ObjectId>,
    pack_dir_before: &HashSet<String>,
) -> Result<TransferStats> {
    let objects_dir = local_odb.objects_dir();
    let pack_dir = objects_dir.join("pack");
    let mut bytes = 0u64;
    let mut objects_from_new_packs = 0u64;
    if pack_dir.is_dir() {
        for entry in fs::read_dir(&pack_dir).map_err(Error::Io)? {
            let path = entry.map_err(Error::Io)?.path();
            if path.extension().is_none_or(|e| e != "pack") {
                continue;
            }
            let Some(name) = path.file_name() else {
                continue;
            };
            let name = name.to_string_lossy();
            if pack_dir_before.contains(name.as_ref()) {
                continue;
            }
            bytes += fs::metadata(&path).map_err(Error::Io)?.len();
            let idx = path.with_extension("idx");
            if idx.is_file() {
                let index = crate::pack::read_pack_index_no_verify(&idx)?;
                objects_from_new_packs += u64::try_from(index.len()).unwrap_or(u64::MAX);
            }
        }
    }
    let (loose_objects, loose_bytes) = loose_transfer_stats(objects_dir, installed)?;
    bytes = bytes.saturating_add(loose_bytes);
    let installed_count = u64::try_from(installed.len()).unwrap_or(u64::MAX);
    let objects = if objects_from_new_packs > 0 {
        objects_from_new_packs.saturating_add(loose_objects)
    } else {
        installed_count
    };
    let kinds = if objects == installed_count {
        Some(object_kind_counts_for_oids(local_odb, installed)?)
    } else {
        None
    };
    Ok(TransferStats {
        objects,
        bytes,
        kinds,
    })
}

/// A single ref update requested by a push.
#[derive(Clone, Debug, Default)]
pub struct PushRefSpec {
    /// The source object to push (`None` for a deletion).
    pub src: Option<ObjectId>,
    /// Local source ref name when known (for reports and `git push --porcelain`).
    pub src_ref: Option<String>,
    /// The destination ref on the remote (e.g. `refs/heads/main`).
    pub dst: String,
    /// Whether a non-fast-forward update is allowed.
    pub force: bool,
    /// Whether this update deletes the remote ref.
    pub delete: bool,
    /// Compare-and-swap expectation: the remote ref's current value must match
    /// this (force-with-lease). `None` disables the value check.
    pub expected_old: Option<ObjectId>,
    /// Force-with-lease expectation that the remote ref does **not** currently
    /// exist. When `true`, a push whose destination already exists on the remote
    /// is rejected as stale (used for "create only" pushes whose lease is the
    /// ref's absence). Independent of [`Self::expected_old`].
    pub expect_absent: bool,
}

/// Local ref name carried on a push spec for status reporting.
#[must_use]
pub(crate) fn push_spec_local_ref(spec: &PushRefSpec) -> Option<String> {
    spec.src_ref.clone()
}

/// Resolve the remote ref name for a push deletion from a short name or full ref.
///
/// When `spec` is a short name, a local `refs/heads/<spec>` wins, then an
/// advertised remote branch, then an advertised remote tag, otherwise
/// `refs/heads/<spec>` (same default as `git push <remote> :<branch>`).
#[must_use]
pub fn resolve_push_delete_ref(
    git_dir: &Path,
    remote_ref_names: &HashSet<String>,
    spec: &str,
) -> String {
    if spec.starts_with("refs/") {
        return spec.to_owned();
    }
    let branch = format!("refs/heads/{spec}");
    let tag = format!("refs/tags/{spec}");
    if crate::refs::resolve_ref(git_dir, &branch).is_ok() {
        return branch;
    }
    if remote_ref_names.contains(&branch) {
        return branch;
    }
    if crate::refs::resolve_ref(git_dir, &tag).is_ok() {
        return tag;
    }
    if remote_ref_names.contains(&tag) {
        return tag;
    }
    branch
}

/// Options controlling a push.
#[derive(Clone, Default)]
pub struct PushOptions {
    /// Apply all updates atomically (all-or-nothing).
    pub atomic: bool,
    /// Compute results without writing to the remote.
    pub dry_run: bool,
    /// When set, update `refs/remotes/<name>/...` in the local repo after a
    /// successful push (matching Git's post-push tracking ref maintenance).
    pub tracking_remote: Option<String>,
    /// Server-side push options to transmit (`git push --push-option <value>`).
    ///
    /// When non-empty, the negotiated capability list includes `push-options`
    /// and one `push-option <value>` pkt-line per entry is written after the
    /// ref-update command block and before the flush/pack. The remote exposes
    /// these to its hooks via `GIT_PUSH_OPTION_COUNT` / `GIT_PUSH_OPTION_<n>`.
    ///
    /// If this is non-empty but the remote `git-receive-pack` does not advertise
    /// the `push-options` capability, the push fails with
    /// [`crate::error::Error::PushOptionsUnsupported`] (matching Git).
    pub push_options: Vec<String>,
    /// Diagnostic sink for [`crate::diagnostics::Trace::Network`] events.
    pub diagnostics: Option<crate::diagnostics::DiagnosticsHandle>,
    /// When true with [`Self::diagnostics`], push paths may emit network trace events.
    pub network_trace: bool,
    /// Ref tips already read from the remote, for example while planning
    /// ([`PushPlanContext::remote_tips`](crate::remote::push_plan::PushPlanContext)).
    ///
    /// A local push uses them as the objects the remote already has instead of listing the
    /// remote's refs again; ref updates still check each destination's current value. They
    /// must come from the push destination, or objects the destination lacks may be left out
    /// of the pack. Wire pushes read the advertisement from their connection and ignore this.
    pub remote_ref_tips: Option<Vec<ObjectId>>,
}

/// The structured result of a push. Reuses [`PushRefResult`] for per-ref status.
#[derive(Clone, Debug, Default)]
pub struct PushOutcome {
    /// Per-ref resolved results (status, old/new oid, reason).
    pub results: Vec<PushRefResult>,
    /// Hook and diagnostic lines from the remote (side-band channel 2), in order.
    ///
    /// Empty for local `file://` pushes and when the server did not use side-band.
    pub remote_messages: Vec<String>,
}

pub use crate::pack_objects::{
    build_pack, build_pack_from_send_list, reachable_closure, PackBuildOptions, PackObjects,
    PackObjectsOptions, PackStats,
};

pub(crate) use crate::pack_objects::build_pack_for_local_fetch;

fn pack_build_options_for_send(
    source_odb: &Odb,
    oids: &[ObjectId],
    base: &PackBuildOptions,
) -> PackBuildOptions {
    if oids.is_empty() || oids.len() > 512 {
        return *base;
    }
    let commits_tags_only = oids.iter().all(|oid| {
        source_odb
            .read(oid)
            .map(|obj| matches!(obj.kind, ObjectKind::Commit | ObjectKind::Tag))
            .unwrap_or(false)
    });
    if !commits_tags_only {
        return *base;
    }
    let mut fast = *base;
    fast.delta = false;
    fast.reuse_deltas = false;
    fast.thin = false;
    fast
}

fn ingest_pack_options_for_fetch(
    local_git_dir: &Path,
    fetch_opts: &FetchOptions,
) -> Result<crate::index_pack::IngestPackOptions> {
    let pack_config =
        crate::config::ConfigSet::load(&fetch_opts.environment, Some(local_git_dir), true)?;
    let index_threads = crate::parallel::index_pack_threads_for_ingest(
        fetch_opts.thread_policy,
        &pack_config,
        fetch_opts.index_pack_threads,
    )?;
    Ok(crate::index_pack::IngestPackOptions {
        fix_thin: true,
        skip_post_index_verify: true,
        threads: index_threads,
        merged_config: Some(pack_config),
        environment: fetch_opts.environment.clone(),
        mark_promisor_pack: fetch_opts.mark_received_pack_promisor,
        trusted_index_rows: None,
    })
}

/// Copy objects for a local fetch: link remote packfiles when possible, then pack any remainder.
#[allow(clippy::too_many_arguments)]
fn copy_objects_for_local_fetch(
    local_git_dir: &Path,
    remote_odb: &Odb,
    local_odb: &Odb,
    wants: &[ObjectId],
    haves: &[ObjectId],
    source_shallow: &HashSet<ObjectId>,
    have_shallow: &HashSet<ObjectId>,
    try_pack_link: bool,
    pack_build: &PackBuildOptions,
    fetch_opts: &FetchOptions,
    progress: &mut dyn ProgressSink,
    cancel: &dyn crate::progress::Cancel,
) -> Result<(HashSet<ObjectId>, TransferStats)> {
    use crate::local_object_copy::{
        link_loose_objects, link_loose_objects_for_oids, link_remote_pack_files, linked_object_set,
        refresh_local_odb_after_link, try_satisfy_via_object_link,
    };

    let remote_objects = remote_odb.objects_dir();
    let local_objects = local_odb.objects_dir();
    let pack_dir_before = pack_file_basenames(&local_objects.join("pack"));
    let remote_has_alternates = !crate::pack::read_alternates_recursive(remote_objects)?.is_empty();

    if !remote_has_alternates
        && !pack_build.thin
        && try_pack_link
        && haves.is_empty()
        && !wants.is_empty()
    {
        progress.start(ProgressPhase::CopyingObjects, Some(wants.len() as u64));
        crate::progress::poll_cancel(progress, cancel)?;
        let linked_packs =
            link_remote_pack_files(remote_objects, local_objects, progress, cancel, wants.len())?;
        if linked_packs > 0 {
            refresh_local_odb_after_link(local_odb);
        }
        let pack_alone_satisfies_wants =
            linked_packs > 0 && wants.iter().all(|oid| local_odb.exists(oid));
        let linked_loose = link_loose_objects(remote_objects, local_objects, progress, cancel)?;
        if linked_packs > 0 || linked_loose > 0 {
            refresh_local_odb_after_link(local_odb);
        }
        if (linked_packs > 0 || linked_loose > 0) && wants.iter().all(|oid| local_odb.exists(oid)) {
            if linked_loose == 0 {
                progress.set(wants.len() as u64);
            }
            progress.finish();
            let stats_loose = if pack_alone_satisfies_wants {
                HashSet::new()
            } else {
                crate::local_object_copy::loose_reachable_from_roots(
                    local_odb,
                    local_objects,
                    wants,
                    source_shallow,
                    cancel,
                )?
            };
            let oids = if stats_loose.is_empty() {
                linked_object_set(local_odb, wants)
            } else {
                stats_loose.clone()
            };
            let mut transfer =
                transfer_stats_for_local_operation(local_odb, &stats_loose, &pack_dir_before)?;
            enrich_local_link_transfer_kinds(
                local_odb,
                &pack_dir_before,
                &stats_loose,
                &mut transfer,
            )?;
            return Ok((oids, transfer));
        }
        progress.finish();
    }

    let have_tips: HashSet<ObjectId> = haves.iter().copied().collect();
    let send_set = collect_reachable_excluding_peer(
        remote_odb,
        wants,
        local_odb,
        &have_tips,
        false,
        source_shallow,
    )?;
    let send = &send_set.objects;

    if !pack_build.thin && try_pack_link {
        progress.start(ProgressPhase::CopyingObjects, Some(send.len() as u64));
        crate::progress::poll_cancel(progress, cancel)?;
        if try_satisfy_via_object_link(
            remote_objects,
            local_objects,
            local_odb,
            send,
            false,
            progress,
            cancel,
        )? {
            progress.set(send.len() as u64);
            progress.finish();
            let oids = linked_object_set(local_odb, send);
            let transfer = transfer_stats_for_local_operation(local_odb, &oids, &pack_dir_before)?;
            return Ok((oids, transfer));
        }
        progress.finish();
    }

    let mut still_missing = objects_missing_from_peer(local_odb, send);
    if still_missing.is_empty() {
        let oids = linked_object_set(local_odb, send);
        let transfer = transfer_stats_for_local_operation(local_odb, &oids, &pack_dir_before)?;
        return Ok((oids, transfer));
    }

    // Hard-link loose objects before building an in-memory pack. Re-reading and
    // recompressing large loose blobs (typical local clone) would spike RSS to several
    // times the object size; linking matches system `git clone` on the same filesystem.
    progress.start(
        ProgressPhase::CopyingObjects,
        Some(still_missing.len() as u64),
    );
    let linked_loose = link_loose_objects_for_oids(
        remote_objects,
        local_objects,
        &still_missing,
        progress,
        cancel,
    )?;
    if linked_loose > 0 {
        refresh_local_odb_after_link(local_odb);
        still_missing = local_odb.missing_oids(still_missing)?;
    }
    progress.finish();
    if still_missing.is_empty() {
        let oids = linked_object_set(local_odb, send);
        let transfer = transfer_stats_for_local_operation(local_odb, &oids, &pack_dir_before)?;
        return Ok((oids, transfer));
    }

    let pack_opts = pack_build_options_for_send(remote_odb, &still_missing, pack_build);
    progress.start(
        ProgressPhase::CopyingObjects,
        Some(still_missing.len() as u64),
    );
    crate::progress::poll_cancel(progress, cancel)?;
    let pack = build_pack_for_local_transfer(
        remote_odb,
        &send_set,
        &still_missing,
        local_odb,
        haves,
        source_shallow,
        have_shallow,
        &pack_opts,
    )?;
    crate::progress::poll_cancel(progress, cancel)?;
    let ingest_opts = ingest_pack_options_for_fetch(local_git_dir, fetch_opts)?;
    let ingested = crate::index_pack::ingest_received_pack_with_progress(
        pack,
        local_odb,
        &ingest_opts,
        None,
        Some(cancel),
    )?;
    progress.inc(ingested.object_ids.len() as u64);
    progress.finish();
    refresh_local_odb_after_link(local_odb);
    let mut out = linked_object_set(local_odb, send);
    out.extend(ingested.object_ids);
    let transfer = transfer_stats_for_local_operation(local_odb, &out, &pack_dir_before)?;
    Ok((out, transfer))
}

/// Copy objects for a local push: link source packfiles into the remote when possible, then pack any remainder.
#[allow(clippy::too_many_arguments)]
fn copy_objects_for_local_push(
    local_odb: &Odb,
    remote_odb: &Odb,
    wants: &[ObjectId],
    haves: &[ObjectId],
    source_shallow: &HashSet<ObjectId>,
    have_shallow: &HashSet<ObjectId>,
    opts: &PackBuildOptions,
) -> Result<()> {
    use crate::local_object_copy::{
        link_loose_objects_for_oids, refresh_local_odb_after_link, try_satisfy_via_object_link,
    };

    let local_objects = local_odb.objects_dir();
    let remote_objects = remote_odb.objects_dir();

    let have_tips: HashSet<ObjectId> = haves.iter().copied().collect();
    let send_set = collect_reachable_excluding_peer(
        local_odb,
        wants,
        remote_odb,
        &have_tips,
        false,
        source_shallow,
    )?;
    let send = &send_set.objects;

    let mut progress = crate::progress::NullProgress;
    let cancel = crate::progress::NeverCancel;
    if !opts.thin
        && try_satisfy_via_object_link(
            local_objects,
            remote_objects,
            remote_odb,
            send,
            false,
            &mut progress,
            &cancel,
        )?
    {
        return Ok(());
    }

    let mut still_missing = objects_missing_from_peer(remote_odb, send);
    if still_missing.is_empty() {
        return Ok(());
    }

    if !haves.is_empty() {
        let linked_loose = link_loose_objects_for_oids(
            local_objects,
            remote_objects,
            &still_missing,
            &mut progress,
            &cancel,
        )?;
        if linked_loose > 0 {
            refresh_local_odb_after_link(remote_odb);
            still_missing = remote_odb.missing_oids(still_missing)?;
        }
        if still_missing.is_empty() {
            return Ok(());
        }
    }

    let pack_opts = pack_build_options_for_send(local_odb, &still_missing, opts);
    let pack = build_pack_for_local_transfer(
        local_odb,
        &send_set,
        &still_missing,
        remote_odb,
        haves,
        source_shallow,
        have_shallow,
        &pack_opts,
    )?;
    crate::index_pack::ingest_received_pack(
        pack,
        remote_odb,
        &crate::index_pack::IngestPackOptions {
            fix_thin: pack_opts.thin,
            ..Default::default()
        },
    )?;
    refresh_local_odb_after_link(remote_odb);
    Ok(())
}

/// The objects of `send` that `peer_odb` does not store on disk.
///
/// Uses the on-disk check rather than [`Odb::exists`], which reports the empty
/// tree as present even without an object file: the result may be packed
/// as-is, and the peer must end up with every object it references.
fn objects_missing_from_peer(peer_odb: &Odb, send: &[ObjectId]) -> Vec<ObjectId> {
    peer_odb.reprepare_local_packs();
    send.iter()
        .copied()
        .filter(|oid| !peer_odb.exists_materialized_local_without_rescan(oid))
        .collect()
}

/// Build the pack that copies `missing` (a subset of `send_set`) into the peer.
///
/// With no `haves` and a non-thin pack, `missing` is already the exact set the
/// peer lacks (the walk stopped at every object the peer stores), so it is
/// packed as-is. Re-enumerating from `missing` instead would walk its whole
/// closure and re-send objects just linked into the peer, such as every packed
/// object reachable from a loose annotated tag. Otherwise defer to
/// [`build_pack_for_local_fetch`], which negotiates against the peer's `haves`.
#[allow(clippy::too_many_arguments)]
fn build_pack_for_local_transfer(
    source_odb: &Odb,
    send_set: &PeerSendSet,
    missing: &[ObjectId],
    peer_odb: &Odb,
    haves: &[ObjectId],
    source_shallow: &HashSet<ObjectId>,
    have_shallow: &HashSet<ObjectId>,
    opts: &PackBuildOptions,
) -> Result<Vec<u8>> {
    if haves.is_empty() && !opts.thin {
        return crate::pack_objects::build_pack_from_send_list(
            source_odb,
            missing,
            &HashSet::new(),
            opts,
            &send_set.name_hashes,
        );
    }
    build_pack_for_local_fetch(
        source_odb,
        missing,
        peer_odb,
        haves,
        source_shallow,
        have_shallow,
        opts,
    )
}

/// Objects a local transfer must send, from [`collect_reachable_excluding_peer`].
struct PeerSendSet {
    /// Objects missing from the peer, in walk order.
    objects: Vec<ObjectId>,
    /// Pack name hash of each blob's and subtree's path, for delta-window ordering.
    name_hashes: HashMap<ObjectId, u32>,
}

/// Like [`collect_reachable_excluding`], but treats every object already stored in
/// `peer_odb` as a boundary (plus explicit `peer_boundary` tips). Used for local
/// file fetch/push so incremental transfers do not walk full history from `haves`.
fn collect_reachable_excluding_peer(
    source_odb: &Odb,
    roots: &[ObjectId],
    peer_odb: &Odb,
    peer_boundary: &HashSet<ObjectId>,
    skip_missing: bool,
    shallow_grafts: &HashSet<ObjectId>,
) -> Result<PeerSendSet> {
    use crate::pack_name_hash::pack_name_hash_extend;

    let mut visited: HashSet<ObjectId> = HashSet::new();
    let mut ordered: Vec<ObjectId> = Vec::new();
    let mut name_hashes: HashMap<ObjectId, u32> = HashMap::new();
    // Each queued object carries the name hash of its directory prefix ("dir/").
    let mut queue: VecDeque<(ObjectId, u32)> = VecDeque::new();

    // The peer is not written during the walk, so one pack-directory refresh is
    // enough; a stale miss only means sending an object the peer already has.
    peer_odb.reprepare_local_packs();
    let at_boundary = |oid: ObjectId| {
        peer_boundary.contains(&oid) || peer_odb.exists_materialized_local_without_rescan(&oid)
    };

    // `visited` also remembers boundary objects so each object is probed once.
    // Blobs named by tree entries are recorded without being read: they have no
    // outgoing links, and a missing one still fails when the pack is written.
    let enqueue = |oid: ObjectId,
                   prefix_hash: Option<u32>,
                   queue: &mut VecDeque<(ObjectId, u32)>,
                   visited: &mut HashSet<ObjectId>,
                   ordered: &mut Vec<ObjectId>|
     -> bool {
        if !visited.insert(oid) || at_boundary(oid) {
            return false;
        }
        ordered.push(oid);
        if let Some(prefix_hash) = prefix_hash {
            queue.push_back((oid, prefix_hash));
        }
        true
    };

    for &root in roots {
        enqueue(root, Some(0), &mut queue, &mut visited, &mut ordered);
    }

    while let Some((oid, prefix_hash)) = queue.pop_front() {
        let obj = match source_odb.read(&oid) {
            Ok(o) => o,
            Err(_) if skip_missing => continue,
            Err(e) => return Err(e),
        };
        match obj.kind {
            ObjectKind::Commit => {
                let commit = parse_commit(&obj.data)?;
                if !shallow_grafts.contains(&oid) {
                    for parent in commit.parents {
                        enqueue(parent, Some(0), &mut queue, &mut visited, &mut ordered);
                    }
                }
                enqueue(commit.tree, Some(0), &mut queue, &mut visited, &mut ordered);
            }
            ObjectKind::Tree => {
                for entry in parse_tree(&obj.data)? {
                    if entry.mode == 0o160000 {
                        continue;
                    }
                    let path_hash = pack_name_hash_extend(prefix_hash, &entry.name);
                    let queued = if entry.mode & 0o170000 == 0o040000 {
                        let dir_hash = pack_name_hash_extend(path_hash, b"/");
                        enqueue(
                            entry.oid,
                            Some(dir_hash),
                            &mut queue,
                            &mut visited,
                            &mut ordered,
                        )
                    } else {
                        enqueue(entry.oid, None, &mut queue, &mut visited, &mut ordered)
                    };
                    if queued {
                        name_hashes.insert(entry.oid, path_hash);
                    }
                }
            }
            ObjectKind::Tag => {
                let tag = parse_tag(&obj.data)?;
                enqueue(
                    tag.object,
                    Some(prefix_hash),
                    &mut queue,
                    &mut visited,
                    &mut ordered,
                );
            }
            ObjectKind::Blob => {}
        }
    }

    Ok(PeerSendSet {
        objects: ordered,
        name_hashes,
    })
}

/// `fetch_local` function.
pub fn fetch_local(
    local_git_dir: &Path,
    remote_git_dir: &Path,
    opts: &FetchOptions,
) -> Result<FetchOutcome> {
    fetch_local_with_progress(
        local_git_dir,
        remote_git_dir,
        opts,
        &mut NoProgress,
        &crate::progress::NeverCancel,
    )
}

/// Result of matching remote refs to fetch refspecs.
pub(crate) struct FetchMatchSet {
    pub(crate) matched: Vec<MatchedRef>,
    pub(crate) positive: Vec<RefspecItem>,
    pub(crate) negatives: Vec<RefspecItem>,
    pub(crate) explicit_refspec_fetch: bool,
}

/// Match remote refs to fetch refspecs (shared by local, wire, and bundle fetch).
pub(crate) fn build_fetch_matched_set(
    opts: &FetchOptions,
    remote_refs: &[(String, ObjectId)],
    remote_head_oid: Option<ObjectId>,
) -> Result<FetchMatchSet> {
    use crate::fetch_request::{FetchRefspecInput, FetchRefspecPlan, FetchRemoteAdvertisement};

    let remote = FetchRemoteAdvertisement {
        refs: remote_refs,
        head_oid: remote_head_oid,
    };
    let plan = FetchRefspecPlan::resolve(
        &FetchRefspecInput {
            user_refspecs: opts.refspecs.clone(),
            configured_refspecs: opts.configured_refspecs.clone(),
            remote_name: opts.remote_name.clone(),
            allow_missing_explicit_for_prune: opts.prune,
        },
        remote,
    )?;
    let explicit_refspec_fetch = plan.is_explicit();

    let mut negatives = match &plan {
        FetchRefspecPlan::Pattern { negatives, .. } | FetchRefspecPlan::Mixed { negatives, .. } => {
            negatives.clone()
        }
    };
    for spec in &opts.negative_refspecs {
        let item = parse_fetch_refspec(spec)
            .map_err(|e| Error::Message(format!("invalid negative refspec '{spec}': {e}")))?;
        negatives.push(item);
    }

    let mut positive = match &plan {
        FetchRefspecPlan::Pattern { positive, .. } => positive.clone(),
        FetchRefspecPlan::Mixed {
            pattern_positive, ..
        } => {
            let mut positive = pattern_positive.clone();
            for spec in &opts.refspecs {
                if let Ok(item) = parse_fetch_refspec(spec) {
                    if !item.negative && !item.pattern && !item.matching {
                        positive.push(item);
                    }
                }
            }
            positive
        }
    };
    if opts.prune {
        crate::fetch_request::supplement_positive_for_explicit_prune(
            &mut positive,
            &opts.configured_refspecs,
            opts.remote_name.as_deref(),
        );
    }

    let matched =
        crate::fetch_request::matched_refs_with_negatives(&plan, remote, &negatives, opts.prune)?;
    Ok(FetchMatchSet {
        matched,
        positive,
        negatives,
        explicit_refspec_fetch,
    })
}

/// Enforce Git fetch ref-update rules for explicit refspec fetches.
pub(crate) fn enforce_fetch_ref_update(
    local_ref: &str,
    old: Option<ObjectId>,
    new: ObjectId,
    mode: UpdateMode,
    opts: &FetchOptions,
    explicit_refspec_fetch: bool,
    local_repo: Option<&crate::repo::Repository>,
) -> Result<()> {
    if opts.dry_run {
        return Ok(());
    }
    let would_write = matches!(
        mode,
        UpdateMode::New | UpdateMode::FastForward | UpdateMode::Forced
    );
    if let Some(short) = local_ref.strip_prefix("refs/heads/") {
        if old.is_some()
            && old != Some(new)
            && (would_write || mode == UpdateMode::NonFastForwardRejected)
        {
            if let Some(repo) = local_repo {
                if crate::worktree::find_worktree_with_branch_checked_out(repo, short, None)?
                    .is_some()
                {
                    return Err(Error::FetchIntoCheckedOutBranch {
                        branch: short.to_owned(),
                    });
                }
            }
        }
    }
    let _ = explicit_refspec_fetch;
    Ok(())
}

/// Fail with [`Error::RemoteUnreadable`] unless `remote_git_dir` is a Git directory.
///
/// A missing or non-repository path must not read as an empty remote.
pub(crate) fn ensure_local_remote_repository(remote_git_dir: &Path) -> Result<()> {
    if crate::submodule_gitdir::is_valid_git_dir(remote_git_dir) {
        return Ok(());
    }
    Err(Error::remote_unreadable(
        remote_git_dir.display().to_string(),
        RemoteUnreadableReason::RepositoryNotFound,
    ))
}

/// Like [`fetch_local`], reporting object copy progress through `progress`.
///
/// The local repository is opened with [`FetchOptions::environment`], so objects land in its
/// `GIT_OBJECT_DIRECTORY` when that is set.
pub fn fetch_local_with_progress(
    local_git_dir: &Path,
    remote_git_dir: &Path,
    opts: &FetchOptions,
    progress: &mut dyn Progress,
    cancel: &dyn crate::progress::Cancel,
) -> Result<FetchOutcome> {
    crate::progress::poll_cancel(progress, cancel)?;
    ensure_local_remote_repository(remote_git_dir)?;
    let local = crate::fetch::open_fetch_target(local_git_dir, opts)?;
    fetch_local_into(&local, remote_git_dir, opts, progress, cancel)
}

/// Like [`fetch_local_with_progress`], copying objects into `local`'s object store.
pub(crate) fn fetch_local_into(
    local: &crate::repo::Repository,
    remote_git_dir: &Path,
    opts: &FetchOptions,
    progress: &mut dyn Progress,
    cancel: &dyn crate::progress::Cancel,
) -> Result<FetchOutcome> {
    crate::progress::poll_cancel(progress, cancel)?;
    ensure_local_remote_repository(remote_git_dir)?;
    let local_git_dir = local.git_dir();
    validate_fetch_depth_local(local_git_dir, opts)?;

    let remote_odb = open_odb(remote_git_dir);
    let remote_algo = remote_odb.hash_algo();
    let reopened = sync_empty_destination_object_format(local_git_dir, remote_algo, opts)?
        .then(|| reopen_after_format_change(local))
        .transpose()?;
    let local = reopened.as_ref().unwrap_or(local);
    let local_git_dir = local.git_dir();
    let local_odb = local.odb();

    // 1. Enumerate remote refs (with HEAD symref for the default branch).
    let remote_entries = crate::remote::list_refs_from_git_dir_for_fetch(
        remote_git_dir,
        Some(&remote_odb),
        &opts.refspecs,
    )
    .map_err(crate::error::Error::from)?;

    let mut default_branch = None;
    let mut remote_head_oid = None;
    // remote ref name -> oid (excluding HEAD and peeled `^{}` entries).
    let mut remote_refs: Vec<(String, ObjectId)> = Vec::new();
    for entry in &remote_entries {
        if entry.name == "HEAD" {
            remote_head_oid = Some(entry.oid);
            default_branch = entry
                .symref_target
                .as_ref()
                .map(|t| t.strip_prefix("refs/heads/").unwrap_or(t).to_owned());
            continue;
        }
        if entry.name.ends_with("^{}") {
            continue;
        }
        if !crate::refs::is_valid_fetch_advertised_ref(&entry.name) {
            continue;
        }
        remote_refs.push((entry.name.clone(), entry.oid));
    }

    let FetchMatchSet {
        mut matched,
        positive,
        negatives,
        explicit_refspec_fetch,
    } = 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();

    let local_shallow = crate::shallow::load_shallow_boundaries(local_git_dir);
    let remote_shallow = crate::shallow::load_shallow_boundaries(remote_git_dir);

    // Explicitly requested tags participate in depth calculation; automatically
    // followed tags must only use the history selected by those fetch roots.
    if opts.tags == TagMode::All {
        apply_tag_mode(
            opts.tags,
            &remote_refs,
            &remote_odb,
            &remote_shallow,
            &negatives,
            &HashSet::new(),
            &mut matched,
            &mut matched_oids,
            &mut seen_remote_ref,
        )?;
    }
    let shallow_response = if opts.has_deepen_request() {
        let shallow_wants: HashSet<ObjectId> = matched
            .iter()
            .map(|m| shallow_want_commit_tip(&remote_odb, m))
            .collect::<Result<_>>()?;
        let remote_repo =
            crate::repo::Repository::open(remote_git_dir, &RepositoryOptions::empty())?;
        let (wire_depth, deepen_relative) = opts
            .wire_deepen()
            .map(|(d, r)| (Some(d), r))
            .unwrap_or((None, false));
        let req = crate::serve::ShallowRequest {
            client_shallow: local_shallow.iter().copied().collect(),
            depth: wire_depth,
            deepen_since: opts
                .deepen_since
                .as_deref()
                .and_then(|s| s.parse::<i64>().ok()),
            deepen_not: Vec::new(),
            deepen_relative,
        };
        Some(crate::serve::compute_shallow_response(
            &remote_repo,
            &shallow_wants.into_iter().collect::<Vec<_>>(),
            &req,
        )?)
    } else {
        None
    };
    let copy_source_shallow = shallow_response
        .as_ref()
        .map(|resp| resp.pack_shallow_grafts.clone())
        .unwrap_or_else(|| remote_shallow.clone());
    let mut tag_shallow = copy_source_shallow.clone();
    if shallow_response.is_none() {
        tag_shallow.extend(local_shallow.iter().copied());
    }

    let following_only = if opts.tags == TagMode::Following {
        let tags_present = crate::fetch::tags_already_local(local.refs(), &remote_refs);
        apply_tag_mode(
            opts.tags,
            &remote_refs,
            &remote_odb,
            &tag_shallow,
            &negatives,
            &tags_present,
            &mut matched,
            &mut matched_oids,
            &mut seen_remote_ref,
        )?
    } else {
        HashSet::new()
    };

    let mut following_tag_wants = if shallow_response.is_some() {
        // Branch roots already bring in lightweight targets. Only annotated tag
        // objects (and their chains) need additional wants in this response.
        local_odb
            .missing_oids(
                matched
                    .iter()
                    .filter(|m| {
                        following_only.contains(&m.oid)
                            && m.advertised_peel.is_some_and(|peel| peel != m.oid)
                    })
                    .map(|m| m.oid),
            )?
            .into_iter()
            .collect()
    } else {
        crate::fetch::following_tag_wants_with_local_peel(&matched, &following_only, local_odb)?
    };
    let shallow_request = opts.has_deepen_request() || !local_shallow.is_empty();
    if shallow_request && shallow_response.is_none() {
        let fetch_commit_oids = matched
            .iter()
            .filter(|m| !m.is_tag)
            .map(|m| m.oid)
            .collect();
        let peel_set = crate::fetch::following_peel_commit_set_for_shallow(
            &remote_odb,
            &fetch_commit_oids,
            opts.absolute_depth_for_tags(),
            &tag_shallow,
        )?;
        following_tag_wants.extend(
            crate::fetch::following_tag_wants_peel_among_fetched_commits(
                &matched,
                &following_only,
                &peel_set,
            ),
        );
    }

    // Determine wants and copy only the required object set.
    let mut wants: Vec<ObjectId> = if shallow_request {
        matched_oids
            .iter()
            .copied()
            .filter(|oid| !following_only.contains(oid) || following_tag_wants.contains(oid))
            .collect()
    } else {
        local_odb.missing_oids(matched_oids.iter().copied())?
    };
    let (shallow_new, shallow_unshallow) = if let Some(resp) = shallow_response {
        for oid in resp.extra_wants.iter().chain(resp.shallow.iter()).copied() {
            if !wants.contains(&oid) {
                wants.push(oid);
            }
        }
        (resp.shallow, resp.unshallow)
    } else {
        (Vec::new(), Vec::new())
    };
    let mut pack_oids: HashSet<ObjectId> = HashSet::new();
    let mut transfer = TransferStats::default();
    if !wants.is_empty() && !opts.dry_run {
        // Resolve exclusion tips only once the final shallow/deepen wants need a copy.
        let mut haves: Vec<ObjectId> = Vec::new();
        let mut have_seen: HashSet<ObjectId> = HashSet::new();
        for m in &matched {
            if let Some(local_ref) = &m.local_ref {
                if let Ok(old) = local.resolve_ref_oid(local_ref) {
                    if have_seen.insert(old) {
                        haves.push(old);
                    }
                }
            }
        }

        let remote_cfg = config_for_git_dir(remote_git_dir);
        let pack_opts = PackBuildOptions::for_local_copy(remote_cfg.as_ref());
        let (oids, fetch_transfer) = copy_objects_for_local_fetch(
            local_git_dir,
            &remote_odb,
            local_odb,
            &wants,
            &haves,
            &copy_source_shallow,
            &local_shallow,
            !opts.has_deepen_request(),
            &pack_opts,
            opts,
            progress,
            cancel,
        )?;
        pack_oids = oids;
        transfer = fetch_transfer;
    }

    if opts.initial_remote_fetch
        && !remote_shallow.is_empty()
        && !opts.has_deepen_request()
        && !opts.dry_run
    {
        let remote_shallow_vec: Vec<ObjectId> = remote_shallow.iter().copied().collect();
        crate::shallow::apply_shallow_updates(local_git_dir, &remote_shallow_vec, &[])?;
    }
    if (!shallow_new.is_empty() || !shallow_unshallow.is_empty()) && !opts.dry_run {
        crate::shallow::apply_shallow_updates(local_git_dir, &shallow_new, &shallow_unshallow)?;
    }

    if opts.tags == TagMode::Following {
        let boundaries = crate::shallow::load_shallow_boundaries(local_git_dir);
        crate::fetch::retain_following_tags(local_odb, &mut matched, &pack_oids, &boundaries)?;
    }

    // 4. Classify and apply ref updates. Ancestry checks use the local repo
    //    (or the bare remote during `--dry-run`, when objects are not copied).
    let dry_run_remote = if opts.dry_run {
        crate::repo::Repository::open(remote_git_dir, &RepositoryOptions::empty()).ok()
    } else {
        None
    };
    let classify_repo = if opts.dry_run {
        dry_run_remote.as_ref()
    } else {
        Some(local)
    };

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

    // Prune BEFORE writing the new tips. A stale tracking ref stored as a file
    // (e.g. `refs/remotes/origin/a`) otherwise blocks creating a nested ref the
    // same fetch introduces (`refs/remotes/origin/a/b`) with a "File exists"
    // directory/file conflict (matches `git fetch --prune` ordering).
    let mut prune_batch: Vec<crate::refs::store::RefUpdate> = Vec::new();
    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 mut store_batch: Vec<crate::refs::store::RefUpdate> = Vec::new();
    for m in &matched {
        let Some(local_ref) = &m.local_ref else {
            // dst empty: fetched but not stored. Report as a no-store update.
            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 old = local.resolve_ref_oid(local_ref).ok();
        let mode = classify_update(
            old.as_ref(),
            &m.oid,
            m.force,
            m.dest_is_tag,
            classify_repo,
            None,
        );

        let write = matches!(
            mode,
            UpdateMode::New | UpdateMode::FastForward | UpdateMode::Forced
        );
        enforce_fetch_ref_update(
            local_ref,
            old,
            m.oid,
            mode,
            opts,
            explicit_refspec_fetch,
            classify_repo,
        )?;
        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;
            }
            if m.dest_is_tag && !ref_target_exists(local_odb, &remote_odb, m.oid)? {
                updates.push(RefUpdate {
                    remote_ref: m.remote_ref.clone(),
                    local_ref: Some(local_ref.clone()),
                    old_oid: old,
                    new_oid: Some(m.oid),
                    mode: UpdateMode::NoChangeNeeded,
                    note: Some("skipped (tag target not present locally)".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(),
            });
        }

        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 let Some(remote) = opts.remote_name.as_deref() {
        if !opts.dry_run {
            if let Some(branch) = default_branch.as_deref() {
                let head_ref = format!("refs/remotes/{remote}/HEAD");
                if crate::refs::resolve_ref(local_git_dir, &head_ref).is_err() {
                    let target = format!("refs/remotes/{remote}/{branch}");
                    let target_will_exist = store_batch.iter().any(|u| u.name == target)
                        || crate::refs::resolve_ref(local_git_dir, &target).is_ok();
                    if target_will_exist {
                        store_batch.push(crate::refs::store::RefUpdate {
                            name: head_ref,
                            new_value: Some(crate::refs::store::RawRef::Symbolic(target)),
                            expected: crate::refs::store::Expected::Any,
                            reflog: None,
                            flags: Default::default(),
                        });
                    }
                }
            }
        }
    }
    if !opts.dry_run && !store_batch.is_empty() {
        apply_fetch_ref_updates(
            local_git_dir,
            &store_batch,
            &mut updates,
            FetchRefRejectHint {
                prune: opts.prune,
                positive: &positive,
                remote_refs: &remote_refs,
                remote_name: opts.remote_name.as_deref(),
            },
        )?;
    }

    crate::fetch::finish_initial_remote_fetch_layout(
        local_git_dir,
        opts,
        default_branch.as_deref(),
    )?;
    if let Some(remote) = opts.remote_name.as_deref() {
        crate::fetch::ensure_remote_tracking_head_symref(
            local_git_dir,
            remote,
            default_branch.as_deref(),
            opts.dry_run,
        )?;
    }
    finish_anonymous_url_fetch(
        local_git_dir,
        opts,
        explicit_refspec_fetch,
        default_branch.as_deref(),
        &remote_refs,
        &matched,
    )?;
    let new_shallow = crate::shallow::dedupe_shallow_list(if opts.has_deepen_request() {
        shallow_new
    } else if opts.initial_remote_fetch {
        remote_shallow.iter().copied().collect()
    } else {
        Vec::new()
    });
    Ok(FetchOutcome {
        updates,
        default_branch,
        new_shallow,
        new_unshallow: shallow_unshallow,
        transfer,
    })
}

/// Push refs and objects from one on-disk git repository into another, entirely
/// in-process (no subprocess, no wire protocol).
///
/// This is the local / `file://` push (send-pack) counterpart to
/// [`fetch_local`]. For each [`PushRefSpec`] it:
///
/// 1. Resolves the source oid from the LOCAL repo (for a non-delete update) and
///    reads the remote's current value of `dst`.
/// 2. Enforces the update rules and produces a [`PushRefResult`] with the right
///    [`crate::push_report::PushRefStatus`]:
///    * `expected_old` set and mismatching the remote's current value →
///      [`PushRefStatus::RejectStale`] (compare-and-swap / force-with-lease).
///    * deletion → succeed when present, or [`PushRefStatus::UpToDate`] when the
///      ref is already gone.
///    * non-fast-forward (remote current is not an ancestor of the source)
///      without `force` → [`PushRefStatus::RejectNonFastForward`]; with `force`
///      it is accepted and reported as forced.
///    * unchanged (remote already at the source) → [`PushRefStatus::UpToDate`].
///    * otherwise [`PushRefStatus::Ok`].
/// 3. For accepted non-delete updates, copies the minimal object closure from the
///    LOCAL odb into the REMOTE odb via [`build_pack`] and
///    [`crate::index_pack::ingest_received_pack`], excluding objects already
///    reachable from the remote's existing ref tips.
/// 4. Applies the ref change on the remote (unless `opts.dry_run`).
///
/// When `opts.atomic` is set and any ref is rejected, no ref or object is
/// written and every otherwise-accepted ref is reported as
/// [`PushRefStatus::AtomicPushFailed`].
///
/// Both repositories must use the same object hash algorithm; the hash width is
/// threaded through [`Odb::hash_algo`] so SHA-256 repos work.
///
/// # Errors
///
/// Returns an error if either repository cannot be opened, if a source object is
/// missing from the local odb, or on I/O failure while writing objects or refs.
//
// TODO(phase: remote transports): `git://`, `http(s)`, and `ssh` push
// (receive-pack handshake + report-status parsing + credential helpers) are out
// of scope here and live in a later phase.
pub fn push_local(
    local_git_dir: &Path,
    remote_git_dir: &Path,
    refs: &[PushRefSpec],
    opts: &PushOptions,
    cancel: &dyn crate::progress::Cancel,
) -> Result<PushOutcome> {
    let local = crate::repo::Repository::open(local_git_dir, &RepositoryOptions::empty())?;
    push_local_from(&local, remote_git_dir, refs, opts, cancel)
}

/// Like [`push_local`], reading the objects to send through `local`'s object store.
pub(crate) fn push_local_from(
    local: &crate::repo::Repository,
    remote_git_dir: &Path,
    refs: &[PushRefSpec],
    opts: &PushOptions,
    cancel: &dyn crate::progress::Cancel,
) -> Result<PushOutcome> {
    let local_git_dir = local.git_dir();
    let mut progress_sink = crate::progress::NullProgress;
    crate::progress::poll_cancel(&mut progress_sink, cancel)?;
    if !opts.push_options.is_empty() {
        return Err(Error::PushOptionsUnsupported);
    }
    ensure_local_remote_repository(remote_git_dir)?;

    let local_odb = local.odb();
    let remote_odb = open_odb(remote_git_dir);

    // The remote's existing ref tips become the `haves` for pack building, so the
    // copied object closure excludes everything the remote already has.
    let remote_have_tips: Vec<ObjectId> = match &opts.remote_ref_tips {
        Some(tips) => tips.clone(),
        None => crate::refs::list_refs(remote_git_dir, "refs/")?
            .into_iter()
            .map(|(_, oid)| oid)
            .collect(),
    };

    // First pass: decide each ref's status without mutating anything.
    let mut decisions: Vec<PushDecision> = Vec::with_capacity(refs.len());
    for spec in refs {
        decisions.push(decide_push(spec, local_odb, remote_git_dir, Some(local))?);
    }

    // Atomic: if any update would be rejected, apply none and demote the
    // otherwise-accepted updates to AtomicPushFailed.
    let any_rejected = decisions.iter().any(|d| d.result.status.is_error());
    if opts.atomic && any_rejected {
        for d in &mut decisions {
            if matches!(d.result.status, PushRefStatus::Ok) {
                d.result.status = PushRefStatus::AtomicPushFailed;
                d.apply = false;
            }
        }
        return Ok(PushOutcome {
            results: decisions.into_iter().map(|d| d.result).collect(),
            remote_messages: Vec::new(),
        });
    }

    let push_wants: Vec<ObjectId> = decisions
        .iter()
        .filter(|d| d.apply && !opts.dry_run)
        .filter_map(|d| match &d.action {
            PushAction::Update(src) => Some(*src),
            _ => None,
        })
        .collect();
    if !push_wants.is_empty() && !opts.dry_run {
        crate::progress::poll_cancel(&mut progress_sink, cancel)?;
        let local_cfg = config_for_git_dir(local_git_dir);
        let pack_opts = if remote_have_tips.is_empty() {
            PackBuildOptions::for_local_copy(local_cfg.as_ref())
        } else {
            PackBuildOptions::for_local_push(local_cfg.as_ref())
        };
        let empty_shallow = HashSet::new();
        let remote_shallow = crate::shallow::load_shallow_boundaries(remote_git_dir);
        copy_objects_for_local_push(
            local_odb,
            &remote_odb,
            &push_wants,
            &remote_have_tips,
            &empty_shallow,
            &remote_shallow,
            &pack_opts,
        )?;
    }

    // Second pass: apply ref updates (one transaction per ref unless `atomic`).
    let mut ref_batch: Vec<crate::refs::store::RefUpdate> = Vec::new();
    for d in &mut decisions {
        if !d.apply || opts.dry_run {
            continue;
        }
        match &d.action {
            PushAction::Update(src) => {
                let update = push_store_ref_update(
                    d.result.remote_ref.as_str(),
                    Some(*src),
                    d.result.old_oid,
                );
                if opts.atomic {
                    ref_batch.push(update);
                } else if let Err(err) = crate::refs::commit_store_update(remote_git_dir, update) {
                    apply_ref_store_error_to_push_decision(d, &err, false);
                }
            }
            PushAction::Delete => {
                let update =
                    push_store_ref_update(d.result.remote_ref.as_str(), None, d.result.old_oid);
                if opts.atomic {
                    ref_batch.push(update);
                } else if let Err(err) = crate::refs::commit_store_update(remote_git_dir, update) {
                    apply_ref_store_error_to_push_decision(d, &err, false);
                }
            }
            PushAction::None => {}
        }
    }
    if opts.atomic && !opts.dry_run && !ref_batch.is_empty() {
        if let Err(err) = crate::refs::commit_ref_store_batch(remote_git_dir, &ref_batch) {
            apply_atomic_batch_ref_store_error(&mut decisions, &ref_batch, err);
        }
    }

    let results: Vec<_> = decisions.into_iter().map(|d| d.result).collect();
    if !opts.dry_run {
        if let Some(remote) = opts.tracking_remote.as_deref() {
            crate::branch_tracking::apply_push_remote_tracking_updates(
                local_git_dir,
                remote,
                &results,
            )?;
        }
    }

    Ok(PushOutcome {
        results,
        remote_messages: Vec::new(),
    })
}

fn push_store_ref_update(
    refname: &str,
    new_oid: Option<ObjectId>,
    expected_old: Option<ObjectId>,
) -> crate::refs::store::RefUpdate {
    crate::refs::store::RefUpdate {
        name: refname.to_owned(),
        new_value: new_oid.map(crate::refs::store::RawRef::Direct),
        expected: match expected_old {
            Some(oid) => crate::refs::store::Expected::Oid(oid),
            None => crate::refs::store::Expected::Any,
        },
        reflog: None,
        flags: Default::default(),
    }
}

fn push_status_for_ref_store_error(err: &Error) -> PushRefStatus {
    match err {
        Error::RefStore(crate::refs::store::RefStoreError::ExpectedMismatch { .. }) => {
            PushRefStatus::RejectStale
        }
        _ => PushRefStatus::RemoteRejected,
    }
}

fn ref_store_error_refname(err: &Error) -> Option<String> {
    let Error::RefStore(store_err) = err else {
        return None;
    };
    Some(match store_err {
        crate::refs::store::RefStoreError::ExpectedMismatch { name, .. }
        | crate::refs::store::RefStoreError::DuplicateUpdate { name }
        | crate::refs::store::RefStoreError::LockHeld { name }
        | crate::refs::store::RefStoreError::InvalidRefName { name }
        | crate::refs::store::RefStoreError::NullOid { name }
        | crate::refs::store::RefStoreError::NotFound { name } => name.clone(),
        crate::refs::store::RefStoreError::NameUnavailable { reason } => match reason {
            crate::refs::RefnameUnavailable::SameBatch { refname, .. }
            | crate::refs::RefnameUnavailable::AncestorExists {
                new_ref: refname, ..
            }
            | crate::refs::RefnameUnavailable::DescendantExists {
                new_ref: refname, ..
            } => refname.clone(),
        },
        crate::refs::store::RefStoreError::SymrefLoop
        | crate::refs::store::RefStoreError::Corrupt(_)
        | crate::refs::store::RefStoreError::GitDirPathIo { .. }
        | crate::refs::store::RefStoreError::InvalidReflog(_)
        | crate::refs::store::RefStoreError::Backend(_) => return None,
    })
}

fn apply_ref_store_error_to_push_decision(d: &mut PushDecision, err: &Error, _atomic: bool) {
    d.result.status = push_status_for_ref_store_error(err);
    d.result.message = Some(err.to_string());
    d.apply = false;
}

fn apply_atomic_batch_ref_store_error(
    decisions: &mut [PushDecision],
    batch: &[crate::refs::store::RefUpdate],
    err: Error,
) {
    let fail_ref = ref_store_error_refname(&err);
    let message = err.to_string();
    let batch_names: HashSet<&str> = batch.iter().map(|u| u.name.as_str()).collect();
    let in_batch = decisions
        .iter_mut()
        .filter(|d| d.apply && batch_names.contains(d.result.remote_ref.as_str()));
    if let Some(fail_name) = fail_ref {
        for d in in_batch {
            if d.result.remote_ref == fail_name {
                d.result.status = push_status_for_ref_store_error(&err);
                d.result.message = Some(message.clone());
            } else {
                d.result.status = PushRefStatus::AtomicPushFailed;
                d.result.message = Some("atomic push failed".to_owned());
            }
            d.apply = false;
        }
    } else {
        let mut first = true;
        for d in in_batch {
            if first {
                d.result.status = push_status_for_ref_store_error(&err);
                d.result.message = Some(message.clone());
                first = false;
            } else {
                d.result.status = PushRefStatus::AtomicPushFailed;
                d.result.message = Some("atomic push failed".to_owned());
            }
            d.apply = false;
        }
    }
}

/// What a single accepted push update does once applied.
enum PushAction {
    /// Copy the closure of `src` to the remote and move the ref to `src`.
    Update(ObjectId),
    /// Delete the remote ref.
    Delete,
    /// No mutation (up-to-date or rejected).
    None,
}

/// A decided-but-not-yet-applied push update.
struct PushDecision {
    result: PushRefResult,
    action: PushAction,
    /// Whether the second pass should apply `action`.
    apply: bool,
}

/// Decide the status of a single [`PushRefSpec`] without mutating either repo.
fn decide_push(
    spec: &PushRefSpec,
    local_odb: &Odb,
    remote_git_dir: &Path,
    local_repo: Option<&crate::repo::Repository>,
) -> Result<PushDecision> {
    let remote_current = crate::refs::resolve_ref(remote_git_dir, &spec.dst).ok();

    // Up-to-date trumps every lease: pushing a non-delete to where the remote ref
    // already points is a no-op that succeeds even when the force-with-lease
    // expectation (a specific `expected_old` value, or `expect_absent`) does not
    // hold — "creating/moving a bookmark to the same place it already is is OK".
    // Must precede both the absence-lease and compare-and-swap checks below.
    if !spec.delete {
        if let Some(src) = spec.src {
            if remote_current == Some(src) {
                return Ok(PushDecision {
                    result: PushRefResult {
                        local_ref: push_spec_local_ref(spec),
                        remote_ref: spec.dst.clone(),
                        old_oid: remote_current,
                        new_oid: Some(src),
                        forced: false,
                        deletion: false,
                        status: PushRefStatus::UpToDate,
                        message: None,
                    },
                    action: PushAction::None,
                    apply: false,
                });
            }
        }
    }

    // Absence lease (force-with-lease that the ref not exist): once the value is
    // actually changing (handled above), a destination that already exists fails
    // the lease and is rejected as stale.
    if spec.expect_absent && remote_current.is_some() {
        return Ok(PushDecision {
            result: PushRefResult {
                local_ref: push_spec_local_ref(spec),
                remote_ref: spec.dst.clone(),
                old_oid: remote_current,
                new_oid: spec.src,
                forced: false,
                deletion: spec.delete,
                status: PushRefStatus::RejectStale,
                message: Some("stale info".to_owned()),
            },
            action: PushAction::None,
            apply: false,
        });
    }

    // Compare-and-swap (force-with-lease): the remote's current value must match
    // the caller's expectation, otherwise reject as stale. A `None` expectation
    // disables the value check.
    if let Some(expected) = spec.expected_old {
        if remote_current != Some(expected) {
            return Ok(PushDecision {
                result: PushRefResult {
                    local_ref: None,
                    remote_ref: spec.dst.clone(),
                    old_oid: remote_current,
                    new_oid: spec.src,
                    forced: false,
                    deletion: spec.delete,
                    status: PushRefStatus::RejectStale,
                    message: Some("stale info".to_owned()),
                },
                action: PushAction::None,
                apply: false,
            });
        }
    }

    if spec.delete {
        let (status, action, apply, message) = match remote_current {
            Some(_) => (PushRefStatus::Ok, PushAction::Delete, true, None),
            None => (
                PushRefStatus::RejectRemoteRefMissing,
                PushAction::None,
                false,
                Some("remote ref does not exist".to_owned()),
            ),
        };
        return Ok(PushDecision {
            result: PushRefResult {
                local_ref: push_spec_local_ref(spec),
                remote_ref: spec.dst.clone(),
                old_oid: remote_current,
                new_oid: None,
                forced: false,
                deletion: true,
                status,
                message,
            },
            action,
            apply,
        });
    }

    // Non-delete updates require a source object that exists locally.
    let Some(src) = spec.src else {
        return Err(Error::Message(format!(
            "push to '{}' has no source object and is not a deletion",
            spec.dst
        )));
    };
    if !local_odb.contains(&src)? {
        return Err(Error::Message(format!(
            "source object {src} for '{}' is missing from the local object store",
            spec.dst
        )));
    }

    // Unchanged: the remote is already at the source.
    if remote_current == Some(src) {
        return Ok(PushDecision {
            result: PushRefResult {
                local_ref: push_spec_local_ref(spec),
                remote_ref: spec.dst.clone(),
                old_oid: remote_current,
                new_oid: Some(src),
                forced: false,
                deletion: false,
                status: PushRefStatus::UpToDate,
                message: None,
            },
            action: PushAction::None,
            apply: false,
        });
    }

    // New ref: nothing on the remote yet — always allowed.
    let Some(old) = remote_current else {
        return Ok(PushDecision {
            result: PushRefResult {
                local_ref: push_spec_local_ref(spec),
                remote_ref: spec.dst.clone(),
                old_oid: None,
                new_oid: Some(src),
                forced: false,
                deletion: false,
                status: PushRefStatus::Ok,
                message: None,
            },
            action: PushAction::Update(src),
            apply: true,
        });
    };

    // Existing ref: fast-forward when the remote's current commit is an ancestor
    // of the source. Otherwise it is a non-fast-forward update, allowed only with
    // force (reported as forced).
    let is_ff = local_repo
        .map(|r| crate::merge_base::is_ancestor(r, old, src).unwrap_or(false))
        .unwrap_or(false);

    if is_ff {
        Ok(PushDecision {
            result: PushRefResult {
                local_ref: push_spec_local_ref(spec),
                remote_ref: spec.dst.clone(),
                old_oid: Some(old),
                new_oid: Some(src),
                forced: false,
                deletion: false,
                status: PushRefStatus::Ok,
                message: None,
            },
            action: PushAction::Update(src),
            apply: true,
        })
    } else if spec.force {
        Ok(PushDecision {
            result: PushRefResult {
                local_ref: push_spec_local_ref(spec),
                remote_ref: spec.dst.clone(),
                old_oid: Some(old),
                new_oid: Some(src),
                forced: true,
                deletion: false,
                status: PushRefStatus::Ok,
                message: None,
            },
            action: PushAction::Update(src),
            apply: true,
        })
    } else {
        Ok(PushDecision {
            result: PushRefResult {
                local_ref: push_spec_local_ref(spec),
                remote_ref: spec.dst.clone(),
                old_oid: Some(old),
                new_oid: Some(src),
                forced: false,
                deletion: false,
                status: PushRefStatus::RejectNonFastForward,
                message: Some("non-fast-forward".to_owned()),
            },
            action: PushAction::None,
            apply: false,
        })
    }
}

/// A remote ref selected for fetch, with its computed local destination.
pub(crate) struct MatchedRef {
    pub(crate) remote_ref: String,
    /// Destination local tracking ref, or `None` for an empty (no-store) dst.
    pub(crate) local_ref: Option<String>,
    pub(crate) oid: ObjectId,
    pub(crate) force: bool,
    pub(crate) is_tag: bool,
    /// Local destination is under `refs/tags/` (tag clobber rules apply).
    pub(crate) dest_is_tag: bool,
    /// Peeled target from ls-refs `peeled:` (annotated tags), when known.
    pub(crate) advertised_peel: Option<ObjectId>,
    /// How this ref appears in [`FETCH_HEAD`](https://git-scm.com/docs/git-fetch#_discussion).
    pub(crate) fetch_head: crate::fetch_request::FetchHeadPolicy,
}

/// Open an [`Odb`] for a git directory, attaching the git dir so `hash_algo`
/// (and MIDX config) resolve correctly.
/// `objects/` of `git_dir`, which lives in the common directory for a linked worktree's
/// admin dir.
fn objects_dir(git_dir: &Path) -> PathBuf {
    crate::repo::common_git_dir_for_config(git_dir).join("objects")
}

pub(crate) fn open_odb(git_dir: &Path) -> Odb {
    Odb::new(&objects_dir(git_dir)).with_config_git_dir(git_dir.to_path_buf())
}

/// Infer the remote repository hash algorithm from fetch capability lines and ref oids.
pub(crate) fn remote_hash_algo_from_advertisement(
    capability_blocks: &[&[String]],
    remote_refs: &[(String, ObjectId)],
) -> HashAlgo {
    for caps in capability_blocks {
        for c in *caps {
            if let Some(name) = c.strip_prefix("object-format=") {
                if let Some(algo) = HashAlgo::from_name(name) {
                    return algo;
                }
            }
        }
    }
    remote_refs
        .iter()
        .find_map(|(_, oid)| HashAlgo::from_len(oid.as_bytes().len()))
        .unwrap_or(HashAlgo::Sha1)
}

/// Before the first pack lands in a fresh clone, align `extensions.objectformat` with the remote.
///
/// Returns whether the local config was rewritten.
pub(crate) fn sync_empty_destination_object_format(
    local_git_dir: &Path,
    remote_algo: HashAlgo,
    opts: &FetchOptions,
) -> Result<bool> {
    if opts.dry_run || !opts.initial_remote_fetch {
        return Ok(false);
    }
    let local_odb = open_odb(local_git_dir);
    if local_odb.hash_algo() == remote_algo {
        return Ok(false);
    }
    if local_odb.approximate_object_count()? > 0 {
        return Err(Error::Message(format!(
            "unsupported object format: destination uses {} but remote uses {}; re-clone into an empty directory with a matching format",
            local_odb.hash_algo().name(),
            remote_algo.name()
        )));
    }
    write_empty_destination_object_format(local_git_dir, remote_algo)?;
    Ok(true)
}

/// Reopen `local` after [`sync_empty_destination_object_format`] rewrote its object format, so
/// its object store uses the new hash algorithm.
pub(crate) fn reopen_after_format_change(
    local: &crate::repo::Repository,
) -> Result<crate::repo::Repository> {
    crate::repo::Repository::open(
        local.git_dir(),
        &RepositoryOptions::with_environment(local.environment().clone()),
    )
}

fn write_empty_destination_object_format(git_dir: &Path, algo: HashAlgo) -> Result<()> {
    use crate::config::{ConfigFile, ConfigScope};
    let config_path = crate::repo::local_config_write_path(git_dir);
    let content = fs::read_to_string(&config_path).unwrap_or_default();
    let mut file = ConfigFile::parse(&config_path, &content, ConfigScope::Local)?;
    if algo == HashAlgo::Sha1 {
        file.unset("extensions.objectformat")?;
        file.set("core.repositoryformatversion", "0")?;
    } else {
        file.set("core.repositoryformatversion", "1")?;
        file.set("extensions.objectformat", algo.name())?;
    }
    file.write()?;
    Ok(())
}

fn config_for_git_dir(git_dir: &Path) -> Option<crate::config::ConfigSet> {
    crate::config::ConfigSet::load(
        &crate::environment::Environment::empty(),
        Some(git_dir),
        true,
    )
    .ok()
}

/// Write [`FETCH_HEAD`](https://git-scm.com/docs/git-fetch#_discussion) for anonymous URL/path fetches.
pub(crate) fn finish_anonymous_url_fetch(
    git_dir: &Path,
    opts: &FetchOptions,
    explicit_refspec_fetch: bool,
    default_branch: Option<&str>,
    remote_refs: &[(String, ObjectId)],
    matched: &[MatchedRef],
) -> Result<()> {
    if opts.dry_run {
        return Ok(());
    }
    let write_fetch_head = opts.remote_name.is_none() || explicit_refspec_fetch;
    if !write_fetch_head {
        return Ok(());
    }
    let source = opts.fetch_source.clone().or_else(|| {
        let name = opts.remote_name.as_deref()?;
        let config = config_for_git_dir(git_dir)?;
        config
            .get(&format!("remote.{name}.url"))
            .filter(|url| !url.trim().is_empty())
    });
    let Some(source) = source.as_deref() else {
        return Ok(());
    };
    let lines = fetch_head_lines_for_ad_hoc_fetch(
        matched,
        remote_refs,
        &opts.refspecs,
        default_branch,
        source,
    );
    if lines.is_empty() {
        return Ok(());
    }
    crate::fetch_head::write_fetch_head(git_dir, &lines)
}

fn fetch_head_lines_for_ad_hoc_fetch(
    matched: &[MatchedRef],
    remote_refs: &[(String, ObjectId)],
    refspecs: &[String],
    default_branch: Option<&str>,
    source: &str,
) -> Vec<crate::fetch_head::FetchHeadLine> {
    use crate::fetch_head::FetchHeadLine;
    use crate::remote::Remote;

    let anonymous_only = refspecs.len() == 1
        && refspecs[0].trim_start_matches('+') == Remote::ANONYMOUS_FETCH_REFSPEC;

    if anonymous_only {
        if let Some(branch) = default_branch {
            let head_ref = format!("refs/heads/{branch}");
            if let Some(m) = matched
                .iter()
                .find(|m| m.remote_ref == head_ref && !m.is_tag)
            {
                return vec![FetchHeadLine {
                    oid: m.oid,
                    for_merge: true,
                    description: source.to_owned(),
                }];
            }
        }
        if let Some((_, oid)) = remote_refs.iter().find(|(name, _)| name == "HEAD") {
            return vec![FetchHeadLine {
                oid: *oid,
                for_merge: true,
                description: source.to_owned(),
            }];
        }
        return Vec::new();
    }

    matched
        .iter()
        .filter(|m| m.fetch_head == crate::fetch_request::FetchHeadPolicy::ForMerge)
        .map(|m| FetchHeadLine {
            oid: m.oid,
            for_merge: true,
            description: crate::fetch_request::fetch_head_description(
                &m.remote_ref,
                m.is_tag,
                source,
            ),
        })
        .collect()
}

/// Tip oid of `HEAD` from a ref advertisement (before `HEAD` is stripped).
#[must_use]
pub(crate) fn remote_head_oid_from_advertised(
    advertised: &[(String, ObjectId)],
) -> Option<ObjectId> {
    advertised
        .iter()
        .find(|(name, _)| name == "HEAD")
        .map(|(_, oid)| *oid)
}

/// Match a ref name against the positive refspecs, returning the destination
/// local ref name (`Some(name)`), `None`+stored=false collapsed: returns
/// `Some(Some(dst))` to store, `Some(None)` to fetch-without-store, or `None`
/// when no positive refspec matches.
pub(crate) fn match_positive(refname: &str, positive: &[RefspecItem]) -> Option<Option<String>> {
    for item in positive {
        let Some(src) = item.src.as_deref() else {
            continue;
        };
        if let Some(dst) = apply_refspec(src, item.dst.as_deref(), refname) {
            // Empty dst means "fetch but do not store".
            if dst.is_empty() {
                return Some(None);
            }
            return Some(Some(dst));
        }
    }
    None
}

/// Whether any positive refspec matching `refname` requested force (`+`).
pub(crate) fn refspecs_force(refname: &str, positive: &[RefspecItem]) -> bool {
    positive.iter().any(|item| {
        item.force
            && item
                .src
                .as_deref()
                .is_some_and(|src| apply_refspec(src, item.dst.as_deref(), refname).is_some())
    })
}

/// Whether `refname` is excluded by any negative refspec.
pub(crate) fn ref_excluded(refname: &str, negatives: &[RefspecItem]) -> bool {
    negatives.iter().any(|item| {
        item.src
            .as_deref()
            .is_some_and(|src| glob_matches(src, refname))
    })
}

/// Apply a `<src>[:<dst>]` refspec to `refname`, returning the destination ref.
///
/// Supports a single `*` wildcard (Git's fetch refspec form). When `dst` is
/// `None` the destination equals the matched source (rare for tracking fetches);
/// when `dst` is `Some("")` the empty string is returned (fetch-without-store).
pub(crate) fn apply_refspec(src: &str, dst: Option<&str>, refname: &str) -> Option<String> {
    match src.find('*') {
        Some(star) => {
            let prefix = &src[..star];
            let suffix = &src[star + 1..];
            if !refname.starts_with(prefix)
                || !refname.ends_with(suffix)
                || refname.len() < prefix.len() + suffix.len()
            {
                return None;
            }
            let middle = &refname[prefix.len()..refname.len() - suffix.len()];
            match dst {
                None => Some(refname.to_owned()),
                Some("") => Some(String::new()),
                Some(d) => Some(d.replacen('*', middle, 1)),
            }
        }
        None => {
            if src != refname {
                return None;
            }
            match dst {
                None => Some(refname.to_owned()),
                Some("") => Some(String::new()),
                Some(d) => Some(d.to_owned()),
            }
        }
    }
}

/// Whether `pattern` (a refspec src side, possibly with one `*`) matches `refname`.
fn glob_matches(pattern: &str, refname: &str) -> bool {
    match pattern.find('*') {
        Some(star) => {
            let prefix = &pattern[..star];
            let suffix = &pattern[star + 1..];
            refname.starts_with(prefix)
                && refname.ends_with(suffix)
                && refname.len() >= prefix.len() + suffix.len()
        }
        None => pattern == refname,
    }
}

/// Add tags to the matched set according to [`TagMode`].
///
/// Under `Following`, tags named in `already_local` (see
/// [`crate::fetch::tags_already_local`]) are skipped, matching Git's auto-follow,
/// which never reconsiders a tag the local repository already has.
#[allow(clippy::too_many_arguments)]
pub(crate) fn apply_tag_mode(
    mode: TagMode,
    remote_refs: &[(String, ObjectId)],
    remote_odb: &Odb,
    shallow_boundaries: &HashSet<ObjectId>,
    negatives: &[RefspecItem],
    already_local: &HashSet<String>,
    matched: &mut Vec<MatchedRef>,
    matched_oids: &mut HashSet<ObjectId>,
    seen_remote_ref: &mut HashSet<String>,
) -> Result<HashSet<ObjectId>> {
    let mut following_only = HashSet::new();
    if mode == TagMode::None {
        return Ok(following_only);
    }

    // Tags this call may add, with the object each one peels to (`None` when
    // the tag cannot be peeled; only the tag oid itself can then match).
    let candidates: Vec<(&String, ObjectId, Option<ObjectId>)> = remote_refs
        .iter()
        .filter(|(name, _)| {
            name.starts_with("refs/tags/")
                && !seen_remote_ref.contains(name)
                && !ref_excluded(name, negatives)
                && !(mode == TagMode::Following && already_local.contains(name))
        })
        .map(|(name, oid)| {
            let peeled = match mode {
                TagMode::Following => remote_odb.peel_tags(*oid).ok().map(|(p, _)| p),
                _ => None,
            };
            (name, *oid, peeled)
        })
        .collect();

    // For Following, decide reachability on the source ODB using the effective
    // shallow grafts for this fetch. Post-pack pruning via
    // [`crate::fetch::retain_following_tags`] still drops tags whose objects did
    // not arrive. Using the local ODB here would miss every tip on first fetch.
    let following_closure: HashSet<ObjectId> = if mode == TagMode::Following {
        let roots: Vec<ObjectId> = matched
            .iter()
            .filter(|m| !m.is_tag)
            .map(|m| m.oid)
            .collect();
        let targets: HashSet<ObjectId> = candidates
            .iter()
            .map(|(_, oid, peeled)| peeled.unwrap_or(*oid))
            .collect();
        crate::fetch::reachable_commits(remote_odb, &roots, &targets, shallow_boundaries)?
    } else {
        HashSet::new()
    };

    for (name, oid, peeled) in candidates {
        let keep = match mode {
            TagMode::All => true,
            // Keep when the tag (or what it peels to) is in the fetched closure.
            TagMode::Following => {
                following_closure.contains(&oid)
                    || peeled.is_some_and(|p| following_closure.contains(&p))
            }
            TagMode::None => false,
        };
        if keep && seen_remote_ref.insert(name.clone()) {
            if matched_oids.insert(oid) && mode == TagMode::Following {
                following_only.insert(oid);
            }
            matched.push(MatchedRef {
                remote_ref: name.clone(),
                local_ref: Some(name.clone()),
                oid,
                force: false,
                is_tag: true,
                dest_is_tag: true,
                advertised_peel: peeled,
                fetch_head: crate::fetch_request::FetchHeadPolicy::ForMerge,
            });
        }
    }
    Ok(following_only)
}

/// Commit tip to use when computing shallow boundaries for one matched ref.
fn shallow_want_commit_tip(remote_odb: &Odb, matched: &MatchedRef) -> Result<ObjectId> {
    if !matched.is_tag {
        return Ok(matched.oid);
    }
    if let Some(peel) = matched.advertised_peel.filter(|p| !p.is_zero()) {
        return Ok(peel);
    }
    if remote_odb.contains(&matched.oid)? {
        let (peeled, _) = remote_odb.peel_tags(matched.oid)?;
        return Ok(peeled);
    }
    Ok(matched.oid)
}

/// Whether a tag ref's peeled target (or the oid itself for lightweight tags)
/// exists in the local or remote object database.
fn ref_target_exists(local_odb: &Odb, remote_odb: &Odb, tag_oid: ObjectId) -> Result<bool> {
    let peel_odb = if local_odb.contains(&tag_oid)? {
        local_odb
    } else {
        remote_odb
    };
    match peel_odb.peel_tags(tag_oid) {
        Ok((peeled, _)) => local_odb.contains(&peeled),
        Err(_) => Ok(false),
    }
}

/// Classify a single ref update into an [`UpdateMode`].
///
/// When `classify_odb` is set (network fetch `--dry-run` after quarantined pack
/// ingest), ancestry uses that store with its configured alternates instead of
/// opening a [`crate::repo::Repository`].
pub(crate) fn classify_update(
    old: Option<&ObjectId>,
    new: &ObjectId,
    force: bool,
    is_tag: bool,
    repo: Option<&crate::repo::Repository>,
    classify_odb: Option<&crate::odb::Odb>,
) -> UpdateMode {
    let Some(old) = old else {
        return UpdateMode::New;
    };
    if old == new {
        return UpdateMode::UpToDate;
    }
    // Fast-forward when old is an ancestor of new (commit history only).
    let ff = if let Some(r) = repo {
        crate::merge_base::is_ancestor(r, *old, *new).unwrap_or(false)
    } else if let Some(odb) = classify_odb {
        crate::merge_base::is_ancestor_odb_walk(odb, *old, *new).unwrap_or(false)
    } else {
        false
    };
    if ff && !is_tag {
        return UpdateMode::FastForward;
    }
    if force {
        return UpdateMode::Forced;
    }
    if is_tag {
        return UpdateMode::TagUpdateRejected;
    }
    UpdateMode::NonFastForwardRejected
}

/// Default-branch symref for a wildcard refspec destination (`<prefix>HEAD`).
///
/// Remote names may contain `/` (e.g. `team/origin` → `refs/remotes/team/origin/HEAD`).
fn is_default_head_under_tracking_prefix(refname: &str, tracking_prefix: &str) -> bool {
    refname == format!("{tracking_prefix}HEAD")
}

/// Local tracking refs that would be removed by `git fetch --prune` given the current remote.
///
/// Returns full ref names and their current object ids. The default-branch symref
/// `refs/remotes/<remote>/HEAD` is never included.
///
/// # Errors
///
/// Propagates ref listing failures.
pub fn tracking_refs_gone_on_remote(
    local_git_dir: &Path,
    positive: &[RefspecItem],
    remote_refs: &[(String, ObjectId)],
) -> Result<Vec<(String, ObjectId)>> {
    let mut live: HashSet<String> = HashSet::new();
    for (name, _) in remote_refs {
        if let Some(Some(dst)) = match_positive(name, positive) {
            live.insert(dst);
        }
    }

    let mut pruned: HashMap<String, ObjectId> = HashMap::new();
    for item in positive {
        let Some(dst) = item.dst.as_deref() else {
            continue;
        };
        if let Some(star) = dst.find('*') {
            let prefix = &dst[..star];
            for (name, oid) in crate::refs::list_refs(local_git_dir, prefix)? {
                if !name.starts_with(prefix) {
                    continue;
                }
                if is_default_head_under_tracking_prefix(&name, prefix) {
                    continue;
                }
                if !live.contains(&name) {
                    pruned.entry(name).or_insert(oid);
                }
            }
        } else if !live.contains(dst) {
            let exact_prefix = dst.strip_suffix("HEAD").filter(|p| p.ends_with('/'));
            if exact_prefix.is_some_and(|p| is_default_head_under_tracking_prefix(dst, p)) {
                continue;
            }
            if let Ok(oid) = crate::refs::resolve_ref(local_git_dir, dst) {
                pruned.entry(dst.to_owned()).or_insert(oid);
            }
        }
    }
    Ok(pruned.into_iter().collect())
}

/// Optional prune hint when a fetch ref write is rejected.
pub(crate) struct FetchRefRejectHint<'a> {
    /// Whether this fetch runs with `--prune` (no stale-ref hints when true).
    pub prune: bool,
    /// Positive refspecs for the fetch (stale tracking detection).
    pub positive: &'a [RefspecItem],
    /// Advertised remote refs at fetch time.
    pub remote_refs: &'a [(String, ObjectId)],
    /// Remote nickname (`origin`, …) for hint text.
    pub remote_name: Option<&'a str>,
}

fn fetch_ref_store_error_rejectable(err: &Error) -> bool {
    matches!(
        err,
        Error::RefStore(crate::refs::store::RefStoreError::NameUnavailable { .. })
            | Error::RefStore(crate::refs::store::RefStoreError::LockHeld { .. })
    )
}

fn blocking_ref_from_unavailable(reason: &crate::refs::RefnameUnavailable) -> Option<&str> {
    match reason {
        crate::refs::RefnameUnavailable::AncestorExists { blocking, .. }
        | crate::refs::RefnameUnavailable::DescendantExists { blocking, .. } => Some(blocking),
        crate::refs::RefnameUnavailable::SameBatch { .. } => None,
    }
}

fn append_fetch_prune_hint(
    note: &mut String,
    stale_tracking: &HashSet<String>,
    blocking: &str,
    remote_name: Option<&str>,
) {
    if stale_tracking.contains(blocking) {
        let remote = remote_name.unwrap_or("origin");
        note.push_str(&format!(
            " — stale remote-tracking ref; try `grit fetch --prune` or `grit remote prune {remote}`"
        ));
    }
}

fn mark_fetch_ref_update_rejected(
    updates: &mut [RefUpdate],
    local_ref: &str,
    err: &Error,
    stale_tracking: &HashSet<String>,
    hint: &FetchRefRejectHint<'_>,
) {
    let Some(update) = updates
        .iter_mut()
        .find(|u| u.local_ref.as_deref() == Some(local_ref))
    else {
        return;
    };
    update.mode = UpdateMode::RefLockRejected;
    let mut note = match err {
        Error::RefStore(crate::refs::store::RefStoreError::NameUnavailable { reason }) => {
            let mut text = reason.lock_message_suffix();
            if let Some(blocking) = blocking_ref_from_unavailable(reason) {
                append_fetch_prune_hint(&mut text, stale_tracking, blocking, hint.remote_name);
            }
            text
        }
        Error::RefStore(crate::refs::store::RefStoreError::LockHeld { name }) => {
            format!("ref '{name}' is locked")
        }
        _ => err.to_string(),
    };
    if note.is_empty() {
        note = "unable to update local ref".to_owned();
    }
    update.note = Some(note);
}

/// Commit tracking-ref writes from a fetch, applying every ref that can be updated.
///
/// Uses one batch transaction when the full batch validates; otherwise falls back to
/// per-ref commits and records lock or namespace failures on the matching
/// [`RefUpdate`] rows instead of aborting the whole fetch.
///
/// # Errors
///
/// Propagates non-rejectable ref-store and I/O errors from the fallback path.
pub(crate) fn apply_fetch_ref_updates(
    local_git_dir: &Path,
    store_batch: &[crate::refs::store::RefUpdate],
    updates: &mut [RefUpdate],
    hint: FetchRefRejectHint<'_>,
) -> Result<()> {
    if store_batch.is_empty() {
        return Ok(());
    }
    if crate::refs::commit_ref_store_batch(local_git_dir, store_batch).is_ok() {
        return Ok(());
    }

    let stale_tracking: HashSet<String> = if hint.prune {
        HashSet::new()
    } else {
        tracking_refs_gone_on_remote(local_git_dir, hint.positive, hint.remote_refs)?
            .into_iter()
            .map(|(name, _)| name)
            .collect()
    };

    for store_update in store_batch {
        match crate::refs::commit_store_update(local_git_dir, store_update.clone()) {
            Ok(()) => {}
            Err(err) if fetch_ref_store_error_rejectable(&err) => {
                mark_fetch_ref_update_rejected(
                    updates,
                    &store_update.name,
                    &err,
                    &stale_tracking,
                    &hint,
                );
            }
            Err(err) => return Err(err),
        }
    }
    Ok(())
}

/// Delete local tracking refs whose remote counterpart no longer exists.
///
/// A local tracking ref is a prune candidate when it lives under the destination
/// namespace of some positive wildcard refspec and no current remote ref maps to
/// it. Matches `git fetch --prune` for the common `refs/remotes/<remote>/*` case.
pub(crate) fn prune_tracking_refs(
    local_git_dir: &Path,
    positive: &[RefspecItem],
    remote_refs: &[(String, ObjectId)],
    dry_run: bool,
    updates: &mut Vec<RefUpdate>,
    mut store_batch: Option<&mut Vec<crate::refs::store::RefUpdate>>,
) -> Result<()> {
    let pruned: Vec<(String, ObjectId)> =
        tracking_refs_gone_on_remote(local_git_dir, positive, remote_refs)?;

    let prune_delete_flags = crate::refs::store::RefUpdateFlags {
        no_deref: true,
        ..Default::default()
    };
    for (name, oid) in pruned.into_iter() {
        if !dry_run {
            if let Some(batch) = store_batch.as_mut() {
                batch.push(crate::refs::store::RefUpdate {
                    name: name.clone(),
                    new_value: None,
                    expected: crate::refs::store::Expected::Any,
                    reflog: None,
                    flags: prune_delete_flags,
                });
            } else {
                crate::refs::delete_ref(local_git_dir, &name)?;
            }
        }
        updates.push(RefUpdate {
            remote_ref: String::new(),
            local_ref: Some(name),
            old_oid: Some(oid),
            new_oid: None,
            mode: UpdateMode::DeletedMissing,
            note: Some("pruned (gone on remote)".to_owned()),
        });
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::progress::ProgressPhase;
    use crate::ref_storage::RefStorageFormat;
    use crate::repo::init_repository;
    use std::process::Command;

    #[derive(Debug, Default)]
    struct PhaseRecorder {
        events: Vec<&'static str>,
        started: Vec<(ProgressPhase, Option<u64>)>,
        finished: Vec<ProgressPhase>,
        counts: Vec<u64>,
        active: Option<ProgressPhase>,
        current: u64,
    }

    impl ProgressSink for PhaseRecorder {
        fn start(&mut self, phase: ProgressPhase, total: Option<u64>) {
            self.events.push("start");
            self.started.push((phase, total));
            self.active = Some(phase);
            self.current = 0;
        }
        fn inc(&mut self, units: u64) {
            self.events.push("inc");
            self.current += units;
            self.counts.push(self.current);
        }
        fn set(&mut self, current: u64) {
            self.events.push("set");
            self.current = current;
            self.counts.push(self.current);
        }
        fn finish(&mut self) {
            self.events.push("finish");
            if let Some(phase) = self.active.take() {
                self.finished.push(phase);
            }
        }
    }

    impl Progress for PhaseRecorder {}

    fn git_ok(dir: &Path, args: &[&str]) -> bool {
        Command::new("git") // hygiene: git fixture comparison in cfg(test) module
            .env("GIT_AUTHOR_NAME", "Test")
            .env("GIT_AUTHOR_EMAIL", "test@example.com")
            .env("GIT_COMMITTER_NAME", "Test")
            .env("GIT_COMMITTER_EMAIL", "test@example.com")
            .current_dir(dir)
            .args(args)
            .env("GIT_CONFIG_GLOBAL", "/dev/null")
            .env("GIT_CONFIG_SYSTEM", "/dev/null")
            .status()
            .map(|s| s.success())
            .unwrap_or(false)
    }

    fn assert_monotonic(counts: &[u64]) {
        for pair in counts.windows(2) {
            assert!(pair[1] >= pair[0], "progress counts must be monotonic");
        }
    }

    #[test]
    fn local_fetch_classifies_through_repository_ref_store() {
        use crate::refs::store::{MemoryRefStore, RawRef};
        use std::sync::Arc;

        let tmp = tempfile::tempdir().unwrap();
        assert!(git_ok(tmp.path(), &["init", "-q", "-b", "main"]));
        assert!(git_ok(
            tmp.path(),
            &["commit", "--allow-empty", "-qm", "base"]
        ));
        let repo =
            crate::repo::Repository::open(&tmp.path().join(".git"), &RepositoryOptions::empty())
                .unwrap();
        let tip = repo.resolve_ref_oid("HEAD").unwrap();
        let store = Arc::new(MemoryRefStore::new());
        store
            .set_ref("refs/remotes/origin/main", RawRef::Direct(tip))
            .unwrap();
        let repo = repo.with_ref_store(store);
        let opts = FetchOptions {
            refspecs: vec!["+refs/heads/*:refs/remotes/origin/*".into()],
            tags: TagMode::None,
            dry_run: true,
            ..Default::default()
        };
        let outcome = fetch_local_into(
            &repo,
            &tmp.path().join(".git"),
            &opts,
            &mut PhaseRecorder::default(),
            &crate::progress::NeverCancel,
        )
        .unwrap();
        assert_eq!(outcome.transfer.objects, 0);
        assert_eq!(outcome.updates.len(), 1);
        assert_eq!(outcome.updates[0].old_oid, Some(tip));
        assert_eq!(outcome.updates[0].mode, UpdateMode::UpToDate);
    }

    #[test]
    fn fetch_local_reports_copy_progress() {
        let remote = tempfile::tempdir().expect("remote");
        assert!(git_ok(
            remote.path(),
            &["init", "-q", "-b", "main", "--bare"]
        ));
        let src = tempfile::tempdir().expect("src");
        assert!(git_ok(src.path(), &["init", "-q", "-b", "main"]));
        std::fs::write(src.path().join("f.txt"), "payload\n").unwrap();
        assert!(git_ok(src.path(), &["add", "f.txt"]));
        assert!(git_ok(src.path(), &["commit", "-qm", "c1"]));
        assert!(git_ok(
            src.path(),
            &["push", "-q", remote.path().to_str().unwrap(), "main:main",],
        ));

        let local = tempfile::tempdir().expect("local");
        init_repository(local.path(), false, "main", None, RefStorageFormat::Files).expect("init");

        let opts = FetchOptions {
            refspecs: vec!["+refs/heads/*:refs/remotes/origin/*".to_owned()],
            initial_remote_fetch: true,
            tags: TagMode::None,
            ..Default::default()
        };
        let mut rec = PhaseRecorder::default();
        fetch_local_with_progress(
            &local.path().join(".git"),
            remote.path(),
            &opts,
            &mut rec,
            &crate::progress::NeverCancel,
        )
        .expect("fetch_local");

        assert!(
            rec.started
                .iter()
                .any(|(p, _)| *p == ProgressPhase::CopyingObjects),
            "expected CopyingObjects phase"
        );
        assert!(
            rec.finished.contains(&ProgressPhase::CopyingObjects),
            "expected CopyingObjects finish"
        );
        assert!(!rec.counts.is_empty(), "expected copy progress counts");
        assert_monotonic(&rec.counts);
        let first_progress = rec.events.iter().find(|e| **e != "start").copied();
        assert_eq!(
            rec.events.first().copied(),
            Some("start"),
            "CopyingObjects must start before link/copy work reports progress"
        );
        assert!(
            first_progress.is_some(),
            "expected set/inc after start during copy"
        );
    }

    #[test]
    fn transfer_stats_add_up_and_keep_kinds_only_when_known() {
        use crate::index_pack::ObjectKindCounts;
        let kinds = |commits, blobs| ObjectKindCounts {
            commits,
            trees: 0,
            blobs,
            tags: 0,
        };
        let mut total = TransferStats::default();
        total += TransferStats {
            objects: 3,
            bytes: 100,
            kinds: Some(kinds(1, 2)),
        };
        // An up-to-date remote adds nothing and does not erase the breakdown.
        total += TransferStats::default();
        total += TransferStats {
            objects: 2,
            bytes: 50,
            kinds: Some(kinds(1, 1)),
        };
        assert_eq!(
            total,
            TransferStats {
                objects: 5,
                bytes: 150,
                kinds: Some(kinds(2, 3)),
            }
        );

        // A transfer without a breakdown makes the combined breakdown unknown.
        total += TransferStats {
            objects: 4,
            bytes: 10,
            kinds: None,
        };
        assert_eq!(total.objects, 9);
        assert_eq!(total.bytes, 160);
        assert_eq!(total.kinds, None);
    }

    #[test]
    fn transfer_stats_for_local_operation_counts_new_pack_index() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let repo = tmp.path().join("repo");
        std::fs::create_dir_all(&repo).unwrap();
        assert!(git_ok(&repo, &["init", "-q", "-b", "main", "."]));
        std::fs::write(repo.join("blob.txt"), b"payload\n").unwrap();
        assert!(git_ok(&repo, &["add", "blob.txt"]));
        assert!(git_ok(&repo, &["commit", "-qm", "c"]));
        assert!(git_ok(&repo, &["repack", "-a", "-d", "-f"]));

        let git_dir = repo.join(".git");
        let pack_dir = git_dir.join("objects/pack");
        let odb = Odb::new(&git_dir.join("objects")).with_config_git_dir(git_dir.clone());
        let stats = transfer_stats_for_local_operation(&odb, &HashSet::new(), &HashSet::new())
            .expect("stats");
        assert!(stats.bytes > 0);
        assert!(stats.objects > 0);

        let pack_path = std::fs::read_dir(&pack_dir)
            .expect("read pack dir")
            .filter_map(|e| e.ok())
            .map(|e| e.path())
            .find(|p| p.extension().is_some_and(|e| e == "pack"))
            .expect("pack file");
        let idx = pack_path.with_extension("idx");
        let index = crate::pack::read_pack_index_no_verify(&idx).expect("idx");
        assert_eq!(
            stats.objects,
            u64::try_from(index.len()).expect("count"),
            "object count must match the on-disk index entry count"
        );
    }

    #[test]
    fn update_mode_modified_repo_excludes_rejected_tag_updates() {
        assert!(UpdateMode::New.modified_repo());
        assert!(UpdateMode::DeletedMissing.modified_repo());
        assert!(!UpdateMode::TagUpdateRejected.modified_repo());
        assert!(!UpdateMode::UpToDate.modified_repo());
    }

    #[test]
    fn update_mode_report_classification_covers_every_mode() {
        use UpdateMode::*;
        for (mode, reported, rejected) in [
            (New, true, false),
            (FastForward, true, false),
            (Forced, true, false),
            (DeletedMissing, true, false),
            (NonFastForwardRejected, true, true),
            (TagUpdateRejected, true, true),
            (SourceObjectNotFound, true, true),
            (UpToDate, false, false),
            (NoChangeNeeded, false, false),
            (Unborn, false, false),
        ] {
            assert_eq!(mode.is_reported(), reported, "{mode:?}");
            assert_eq!(mode.is_rejected(), rejected, "{mode:?}");
            assert!(!(mode.modified_repo() && mode.is_rejected()), "{mode:?}");
        }
    }

    #[test]
    fn fetch_outcome_summarizes_reported_changed_and_rejected_refs() {
        let update = |name: &str, mode| RefUpdate {
            remote_ref: format!("refs/heads/{name}"),
            local_ref: Some(format!("refs/remotes/origin/{name}")),
            old_oid: None,
            new_oid: None,
            mode,
            note: None,
        };
        let outcome = FetchOutcome {
            updates: vec![
                update("same", UpdateMode::UpToDate),
                update("new", UpdateMode::New),
                update("ff", UpdateMode::FastForward),
                update("noop", UpdateMode::NoChangeNeeded),
            ],
            ..Default::default()
        };
        let reported: Vec<&str> = outcome
            .reported_updates()
            .map(|u| u.remote_ref.as_str())
            .collect();
        assert_eq!(reported, ["refs/heads/new", "refs/heads/ff"]);
        assert_eq!(outcome.changed_ref_count(), 2);
        assert!(!outcome.has_rejections());

        let rejected = FetchOutcome {
            updates: vec![update("diverged", UpdateMode::NonFastForwardRejected)],
            ..Default::default()
        };
        assert_eq!(rejected.changed_ref_count(), 0);
        assert!(rejected.has_rejections());
        assert_eq!(rejected.reported_updates().count(), 1);
        assert!(FetchOutcome::default().reported_updates().next().is_none());
    }
}