mkit_cli/remote_dispatch/

mod.rs

//! URL-scheme → `Transport` dispatch for `mkit push` / `mkit pull`.
//!
//! The Rust binary wires all five shipping schemes here: `mkit+file://`,
//! `mkit+https://` (and `mkit+http://` for local dev), `mkit+s3://`, and
//! `mkit+ssh://`. The memory transport is in-process only, so it is
//! reached via [`push_all`] / [`pull_all`] with an `Arc<MemoryTransport>`
//! constructed in-process rather than URL-based construction. Integration
//! tests in the `mkit-cli` crate exercise the memory path directly.
//!
//! Credentials / environment sources:
//! - HTTP(S): optional `MKIT_API_TOKEN` bearer.
//! - S3/R2: `MKIT_R2_ACCESS_KEY_ID` + `MKIT_R2_SECRET_ACCESS_KEY` (plus
//!   optional `MKIT_R2_REGION`, default `auto`). Missing creds do NOT
//!   fail at connect time; the first signed request returns
//!   `TransportError::AccessDenied`.
//! - SSH: spawns `ssh(1)` subprocess — inherits the user's agent / keys /
//!   `~/.ssh/config`. Per-repo `.mkit/config` SSH options (host-key
//!   checking, known-hosts path, identity file) are wired through via
//!   `SshTransport::connect_with_options` when config is loaded.

// `pub(crate)` so the `remote remove`/`rename` command handlers can drive
// the record's lifecycle ops (#545); everything else stays module-private.
pub(crate) mod applied_packs;
mod envelope_signer;
pub(crate) mod grants;
mod packmap;
mod split;
mod upload_receipts;

use mkit_core::layout::RepoLayout;
use std::path::Path;
use std::sync::Arc;

use applied_packs::AppliedPacks;

use mkit_core::hash::{HASH_LEN, Hash};
use mkit_core::object::Object;
use mkit_core::ops::merge::is_ancestor;
use mkit_core::ops::restore;
use mkit_core::pack::{self, PackError, PackWriter, PreparedDelta, PreparedRaw};
use mkit_core::protocol::{PackKey, Transport, TransportError, UploadLimits};
use mkit_core::refs::{self, Head};
use mkit_core::store::{ObjectStore, StoreError};
use mkit_core::transfer::{self, PackListError};
use mkit_transport_connect::admission::{AdmissionPolicy, is_reserved};
use mkit_transport_connect::{
    ConnectTransport, PENDING_INTERRUPTED_MESSAGE, repository_identity_from_url,
};
use mkit_transport_file::FileTransport;
use mkit_transport_s3::S3Transport;
use mkit_transport_ssh::{SshInitError, SshOptions, SshTransport, parse_mkit_ssh_url};
use rayon::prelude::*;

pub use split::{MAX_CHAIN_COMMITS, MAX_SPLIT_STEPS, PushControl, StepAuthority, plan_push_steps};

use packmap::{
    ChainAction, advance_packmap, apply_fetched_chain, commit_head, packmap_ref, probe_chain,
    rebaseline_depth, resolve_and_download_chain,
};

const DEFAULT_REMOTE: &str = "default";

/// Errors returned by the push / pull helpers. Mapped to exit codes by
/// the commands themselves.
#[derive(Debug, thiserror::Error)]
pub enum DispatchError {
    #[error("unsupported URL scheme: {0}")]
    UnsupportedScheme(String),
    #[error("malformed URL: {0}")]
    MalformedUrl(String),
    #[error(
        "repository `{identity}` not found at {origin}; check the remote URL path against the server's configured name (an empty path selects `default` on a single-repository deployment)"
    )]
    RepositoryNotFound { identity: String, origin: String },
    #[error("no HEAD branch to push")]
    NoHead,
    /// A poll-loop checkpoint observed `signal::is_shutdown() == true`
    /// and aborted partway through. Callers should map this to
    /// `exit::TEMPFAIL` (75) so retries are safe — the transfer is
    /// half-finished but the remote is unmodified for any ref we
    /// hadn't reached yet.
    #[error("interrupted")]
    Interrupted,
    #[error("{0}")]
    UploadInterrupted(String),
    #[error("transport: {0}")]
    Transport(#[from] TransportError),
    #[error("refs: {0}")]
    Refs(#[from] refs::RefError),
    #[error("repo lock: {0}")]
    RepoLock(#[from] mkit_core::repo_lock::LockError),
    #[error("worktree discovery: {0}")]
    Discover(#[from] mkit_core::layout::DiscoverError),
    #[error("io: {0}")]
    Io(#[from] std::io::Error),
    #[error("store: {0}")]
    Store(#[from] StoreError),
    #[error("pack: {0}")]
    Pack(#[from] PackError),
    #[error("packlist: {0}")]
    PackList(#[from] PackListError),
    #[error("ssh init: {0}")]
    SshInit(#[from] SshInitError),
    #[error("pull requires HEAD to point at a branch")]
    DetachedHead,
    #[error("remote branch '{0}' not found")]
    RemoteBranchMissing(String),
    #[error("pull would not fast-forward branch '{branch}'; merge or rebase first")]
    NonFastForwardPull { branch: String },
    #[error("restore safety: {0}")]
    RestoreSafety(String),
    #[error("object is not a commit")]
    NotCommit,
    #[error("restore: {0}")]
    Restore(#[from] restore::RestoreError),
    /// The per-endpoint credential-trust gate (#97) refused to build a
    /// credential-bearing transport for a repo-chosen endpoint the user
    /// has not explicitly trusted. The wrapped string is the actionable
    /// message produced by [`crate::config::endpoint_credential_trust`].
    #[error("{0}")]
    UntrustedRemote(String),
    /// A CAS ref write was rejected because the remote moved under us
    /// (non-fast-forward). Callers map this to an actionable
    /// fetch-then-retry / `--force-with-lease` hint.
    #[error(
        "updates were rejected for branch '{branch}' (non-fast-forward); fetch and merge first, or re-run with --force-with-lease / --force"
    )]
    NonFastForwardPush { branch: String },
    /// The branch's packmap pointer could not be durably advanced under
    /// sustained concurrent pushes. The branch ref was NOT moved, so the
    /// remote stays consistent; the push is safe to retry.
    #[error(
        "could not establish the pack map for branch '{branch}' under concurrent pushes; retry"
    )]
    PackmapContended { branch: String },
    /// A packmap chain was malformed — it exceeded the depth cap, contained
    /// a cycle, or a node could not be downloaded/decoded. Indicates a
    /// corrupt or hostile remote.
    #[error("pack map chain for branch '{branch}' is malformed (too deep, cyclic, or unreadable)")]
    PackChainInvalid { branch: String },
    /// The remote advertised a branch ref but no packmap
    /// (`refs/mkit/packmap/<branch>`) for it. mkit speaks a single,
    /// packmap-only transfer dialect: the push path ALWAYS advertises a
    /// packmap before moving the branch ref, so a branch with a tip but no
    /// packmap is a corrupt/incomplete remote, not a format we degrade to.
    /// We refuse to fetch rather than silently materialise a partial ref.
    #[error("remote advertised branch '{0}' but no pack map to reconstruct it")]
    PackmapMissing(String),
    /// The packmap advertised a pack the remote does not hold. The branch's
    /// closure cannot be reconstructed, so the fetch is aborted rather than
    /// publishing a ref to an incomplete history.
    #[error("remote advertised pack {pack} for branch '{branch}' but does not hold it")]
    AdvertisedPackMissing { branch: String, pack: String },
    /// After unpacking the branch's whole packmap chain, an object
    /// reachable from the fetched tip is still absent from the local store —
    /// the chain did not deliver the full closure. This is a pure integrity
    /// assertion (no recovery download is attempted): the remote's packmap
    /// is incomplete, so fetch aborts before publishing the ref.
    #[error("remote is missing object {0} needed to reconstruct the ref")]
    RemoteMissingObject(String),
    /// The fetched tip's object closure exceeds the
    /// [`mkit_core::ops::graph::MAX_REACHABLE`] verification cap, so
    /// completeness could not be confirmed. On the applied-pack skip path
    /// (#409) this closure walk is the sole guarantee the local store is
    /// whole; a truncated walk could silently pass over missing objects, so
    /// we fail closed rather than publish a ref we can't fully verify. This is
    /// NOT a self-heal trigger.
    #[error(
        "fetched history is too large to verify (closure exceeds the {0}-object cap); refusing to publish an unverified ref"
    )]
    ClosureTooLarge(usize),
    /// A commit/remix/tag newly introduced by this fetch failed Ed25519
    /// signature verification via [`mkit_core::sign::verify_commit`] /
    /// `verify_remix` / `verify_tag` — the exact check `mkit verify <rev>`
    /// runs manually (issue #692). A hostile remote (THREAT-MODEL §3.1) can
    /// otherwise push an unsigned or forged history that `clone`/`pull`/
    /// `fetch` would silently materialise. Deliberately distinct from
    /// [`RemoteMissingObject`](Self::RemoteMissingObject) so it does NOT
    /// feed the applied-pack self-heal retry (#409): an invalid signature
    /// is not evidence of local staleness, and clearing the applied-packs
    /// record would not make a hostile remote's history valid. Fails
    /// closed by default; opt out with `--no-verify-signatures` or the
    /// user-scoped `pull.require_signed = false` config (never settable
    /// from repo-scoped config — see [`crate::config::REPO_FORBIDDEN_KEYS`]).
    #[error("object {hash} failed signature verification: {reason}")]
    UnsignedOrInvalidObject { hash: String, reason: String },
    /// A remote name passed to the applied-packs record (#409) is not a legal
    /// ref name (per [`mkit_core::refs::validate_ref_name`]). A remote name
    /// *is* a ref name, so this should never occur for a config-registered
    /// remote; it is defence-in-depth against a malformed name being used as
    /// a raw path component under `.mkit/applied-packs/`.
    #[error("invalid remote name for applied-packs record: '{0}'")]
    InvalidRemoteName(String),
    /// One advance would need more data packs than the server permits per
    /// advance. `commit` names the single commit or merge that cannot be split
    /// further when the refusal came from the pre-flight dry seal, before
    /// anything was published; without it the seal-time backstop fired.
    #[error(
        "push needs {packs} data packs but this server permits {limit} per advance{}{}; ask the operator to raise max_pack_bytes",
        .commit.as_ref().map_or_else(String::new, |c| format!(" ({c} cannot be split further)")),
        if *.holds_remote_head {
            " (it is the first commit that contains the remote head: rebase onto the remote head, or merge it in a smaller or earlier commit)"
        } else {
            ""
        }
    )]
    PushTooLarge {
        packs: usize,
        limit: usize,
        commit: Option<String>,
        /// The unsplittable commit is the first that contains the remote head,
        /// so the split could not cut before it.
        holds_remote_head: bool,
    },
    /// A retry of one advance (a rejected ticket, a lost delta base) re-planned
    /// the push and the new plan no longer fits one advance. Found by a dry
    /// seal, so nothing was uploaded for the retry.
    #[error(
        "{reason}, and the retry needs {packs} data packs but this server permits {limit} per advance; nothing was uploaded for the retry"
    )]
    RetryTooLarge {
        reason: RetryReason,
        packs: usize,
        limit: usize,
    },
    /// A split push would exceed a bound (steps or chain length). Nothing was
    /// published.
    #[error("{0}")]
    PushSplitLimit(String),
    /// The stored grants do not authorize every advance of a split push.
    /// Nothing was published.
    #[error("{0}")]
    PushNotAuthorized(String),
    /// A split push failed after publishing some advances. `head` is the last
    /// advance this push published, a first-parent ancestor of the local tip
    /// (another pusher may have moved the branch since); re-running the push
    /// resumes from the remote branch unless a concurrent push caused the
    /// failure.
    #[error(
        "{cause}; {published} of {total} advances were published and the last published advance was {head} on branch '{branch}'{}",
        resume_hint(.cause)
    )]
    SplitInterrupted {
        branch: String,
        head: String,
        published: usize,
        total: usize,
        cause: Box<DispatchError>,
    },
    #[error("upload ticket was rejected after the push was retried")]
    TicketRejected,
    #[error("delta base is unavailable after a restart")]
    DeltaBaseUnavailable,
    #[error("packlist still names a pack absent from this repository after retry")]
    PacklistNotInRepository,
}

/// The closing advice of a split push's failure: resume, unless the failure
/// was a concurrent push (the branch then moved, and a re-run would only
/// compare against whatever the other pusher left).
fn resume_hint(cause: &DispatchError) -> &'static str {
    if matches!(cause, DispatchError::NonFastForwardPush { .. }) {
        "; the branch was moved by another push, so fetch and merge before pushing again"
    } else {
        "; re-run the push to resume"
    }
}

/// Why an advance was re-planned.
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
pub enum RetryReason {
    /// The server no longer holds a delta base the plan relied on.
    #[error(
        "the server no longer holds a delta base this push relied on, so it must be re-sent as a full closure (ask the operator whether the repository was restored from an older copy)"
    )]
    LostDeltaBase,
    /// The upload ticket or packlist was rejected, so the push was planned
    /// again against the remote's current head, which may have moved.
    #[error(
        "the upload was rejected and the push was re-planned against the remote's current head, which may have moved (fetch and merge, then push again)"
    )]
    Replanned,
}

/// The advances a split push published before it failed.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PublishedPrefix {
    pub branch: String,
    pub published: usize,
    pub total: usize,
    pub head: String,
    /// Whether re-running the push is the advice (not after a concurrent push).
    pub resumable: bool,
}

impl PublishedPrefix {
    /// The note printed with any failure that left a prefix published.
    #[must_use]
    pub fn note(&self) -> String {
        format!(
            "{} of {} advances were published; the last published advance on branch '{}' was {}{}",
            self.published,
            self.total,
            self.branch,
            self.head,
            if self.resumable {
                "; re-run the push to resume"
            } else {
                "; the branch was moved by another push, so fetch and merge before pushing again"
            }
        )
    }
}

impl DispatchError {
    /// The error a split push failed with, without its published-prefix
    /// wrapper, and that prefix (`None` when no advance of this branch was published).
    #[must_use]
    pub fn into_published_prefix(self) -> (Self, Option<PublishedPrefix>) {
        match self {
            Self::SplitInterrupted {
                branch,
                head,
                published,
                total,
                cause,
            } => {
                let resumable = !matches!(*cause, Self::NonFastForwardPush { .. });
                (
                    *cause,
                    Some(PublishedPrefix {
                        branch,
                        published,
                        total,
                        head,
                        resumable,
                    }),
                )
            }
            other => (other, None),
        }
    }
}

/// Interpret a missing result as a repository failure only for repository-level
/// operations. Pack downloads retain their content-specific missing errors.
fn repository_operation_error(tx: &dyn Transport, error: TransportError) -> DispatchError {
    if let TransportError::RemoteError(message) = &error
        && message.starts_with("upload interrupted; ")
    {
        return DispatchError::UploadInterrupted(message.clone());
    }
    if matches!(&error, TransportError::RemoteError(message) if message == PENDING_INTERRUPTED_MESSAGE)
    {
        return DispatchError::Interrupted;
    }
    if matches!(&error, TransportError::PackNotFound)
        && let Some(address) = tx.repository_address()
    {
        return DispatchError::RepositoryNotFound {
            identity: address.repository.to_owned(),
            origin: address.origin.to_owned(),
        };
    }
    error.into()
}

/// Open a transport for `endpoint` only after the per-endpoint
/// credential-trust gate (#97) approves it.
///
/// This is the single choke point through which push / fetch / pull
/// (and named-remote callers in #175) MUST build a transport: it runs
/// [`crate::config::endpoint_credential_trust`] — keyed on the resolved
/// ENDPOINT and its `repo_chosen` provenance — *before* constructing
/// the transport, so a credential-bearing HTTP/S3 transport is never
/// instantiated for a repo-chosen endpoint the user hasn't trusted.
///
/// `repo_chosen` is `true` when the endpoint came from repo-scoped
/// config (the flat `remote_endpoint` or a `remote.<name>.url`),
/// `false` when it came from the user / an explicit CLI argument. Trust
/// is per ENDPOINT, never per remote name.
///
/// `layout` is needed only to resolve a repo-key-file envelope signer
/// when `cfg.merged.transport_auth == "envelope"` (see
/// `envelope_signer_from_config`) — every caller already has it at
/// hand (it discovered the repo before building `cfg`).
pub fn open_trusted(
    endpoint: &str,
    remote_name: &str,
    repo_chosen: bool,
    cfg: &crate::config::LayeredConfig,
    layout: &RepoLayout,
) -> Result<Arc<dyn Transport>, DispatchError> {
    crate::config::endpoint_credential_trust(cfg, endpoint, repo_chosen)
        .map_err(DispatchError::UntrustedRemote)?;
    open_with_config_for_remote(endpoint, &cfg.merged, layout, Some(remote_name))
}

/// A remote opened for `mkit push`: the transport, and (for a signing Connect
/// client) the check that its stored grants authorize every advance of a split
/// push.
pub(crate) struct PushRemote {
    pub tx: Arc<dyn Transport>,
    pub authority: Option<Arc<dyn StepAuthority>>,
}

/// [`open_trusted`] for a push: the same trust gate and transport, plus the
/// grant pre-check a split push needs (WP-1.17b, B5).
///
/// # Errors
/// As [`open_trusted`].
pub(crate) fn open_trusted_for_push(
    endpoint: &str,
    remote_name: &str,
    repo_chosen: bool,
    cfg: &crate::config::LayeredConfig,
    layout: &RepoLayout,
) -> Result<PushRemote, DispatchError> {
    crate::config::endpoint_credential_trust(cfg, endpoint, repo_chosen)
        .map_err(DispatchError::UntrustedRemote)?;
    if !is_connect_url(endpoint) {
        return Ok(PushRemote {
            tx: open_with_config_for_remote(endpoint, &cfg.merged, layout, Some(remote_name))?,
            authority: None,
        });
    }
    let parts = open_connect_parts(endpoint, &cfg.merged, layout, Some(remote_name), true)?;
    let tx = Arc::new(parts.transport);
    let authority = parts
        .signer_key
        .zip(parts.grants)
        .map(|(signer_key, grants)| {
            Arc::new(grants::ConnectAuthority {
                transport: tx.clone(),
                grants,
                signer_key,
            }) as Arc<dyn StepAuthority>
        });
    Ok(PushRemote { tx, authority })
}

/// The single chokepoint that resolves SSH trust-pinning (issue #389) and
/// `mkit+https://` envelope-signing config from `cfg` and opens a
/// transport. Every config-bearing caller — [`open_trusted`] (push /
/// fetch / pull) and `clone` — routes through here, so both are resolved
/// and threaded in exactly ONE place. A new remote command physically
/// cannot forget them as long as it opens through config; the only
/// un-pinned path is the config-less [`open`], which production never
/// uses for `ssh` or envelope auth.
pub(crate) fn open_with_config(
    url: &str,
    cfg: &crate::config::Config,
    layout: &RepoLayout,
) -> Result<Arc<dyn Transport>, DispatchError> {
    open_with_config_for_remote(url, cfg, layout, None)
}

fn open_with_config_for_remote(
    url: &str,
    cfg: &crate::config::Config,
    layout: &RepoLayout,
    remote_name: Option<&str>,
) -> Result<Arc<dyn Transport>, DispatchError> {
    if is_connect_url(url) {
        return Ok(Arc::new(open_connect_with_config(
            url,
            cfg,
            layout,
            remote_name,
            true,
        )?));
    }
    open_with_ssh_options(url, &ssh_options_from_config(cfg), None)
}

fn is_connect_url(url: &str) -> bool {
    url.starts_with("mkit+https://") || url.starts_with("mkit+http://")
}

/// [`open_trusted`] for a Connect endpoint, returning the concrete
/// [`ConnectTransport`] so callers can reach the epoch and visibility RPCs
/// (WP-2.14). `sign` chooses whether the ambient signing identity is used:
/// `mkit epoch` and statement-mode `mkit visibility set` send unsigned RPCs
/// and pass `false`; envelope-mode visibility passes `true` and so needs
/// `transport_auth = envelope` and a trusted remote, like `mkit push`.
///
/// # Errors
/// The credential-trust gate, a non-Connect endpoint, or connection setup.
pub(crate) fn open_connect_trusted(
    endpoint: &str,
    remote_name: &str,
    repo_chosen: bool,
    cfg: &crate::config::LayeredConfig,
    layout: &RepoLayout,
    sign: bool,
) -> Result<ConnectTransport, DispatchError> {
    crate::config::endpoint_credential_trust(cfg, endpoint, repo_chosen)
        .map_err(DispatchError::UntrustedRemote)?;
    if !is_connect_url(endpoint) {
        return Err(DispatchError::UnsupportedScheme(format!(
            "`{endpoint}` is not an mkit+https:// or mkit+http:// remote; grant epochs and repository visibility are served over Connect"
        )));
    }
    open_connect_with_config(endpoint, &cfg.merged, layout, Some(remote_name), sign)
}

/// Build the Connect transport for `url` from `cfg`: envelope signing (when
/// `sign` and configured), the user grant store as the `GrantSource`, the
/// upload receipt store, progress observers and the admission helper.
pub(crate) fn open_connect_with_config(
    url: &str,
    cfg: &crate::config::Config,
    layout: &RepoLayout,
    remote_name: Option<&str>,
    sign: bool,
) -> Result<ConnectTransport, DispatchError> {
    Ok(open_connect_parts(url, cfg, layout, remote_name, sign)?.transport)
}

/// A Connect transport with what a push needs to check its authority.
struct ConnectParts {
    transport: ConnectTransport,
    /// The user grants installed on the transport (none without a signer).
    grants: Option<Arc<grants::LocalGrants>>,
    /// The signing key's public hex, when requests are signed.
    signer_key: Option<String>,
}

fn open_connect_parts(
    url: &str,
    cfg: &crate::config::Config,
    layout: &RepoLayout,
    remote_name: Option<&str>,
    sign: bool,
) -> Result<ConnectParts, DispatchError> {
    let envelope_signer = if sign {
        if cfg.transport_auth_envelope() && cfg.trusted_remote_endpoint.trim() != url {
            return Err(DispatchError::UntrustedRemote(format!(
                "refusing request signing for untrusted destination `{url}`; run `mkit config trusted_remote_endpoint {url}` before using ambient signing identity"
            )));
        }
        envelope_signer_from_config(cfg, layout)?
    } else {
        None
    };
    validate_connect_repository(url)?;
    let signed = envelope_signer.is_some();
    let signer_key = envelope_signer
        .as_ref()
        .map(|signer| signer.public_key_hex());
    // Resolve only the selected config fallback: an authoritative environment
    // path must not be defeated by an ignored fallback's tilde/HOME error.
    let ca_file = if url.starts_with("mkit+https://")
        && std::env::var_os(mkit_transport_connect::tls::CA_FILE_ENV).is_none()
    {
        cfg.ssl_ca_file_path(layout).map_err(|error| {
            DispatchError::Transport(TransportError::TlsConfiguration(error.to_string()))
        })?
    } else {
        None
    };
    let mut tx = ConnectTransport::connect_with_signer_and_ca_file(
        url,
        envelope_signer,
        ca_file.as_deref(),
    )?
    .with_receipt_store(Arc::new(upload_receipts::FilePartReceiptStore::new(
        layout.upload_parts_dir(),
    )))
    .with_pending_observer(|event| {
        crate::progress::pending_event(event);
        !crate::signal::is_shutdown()
    })
    .with_upload_observer(|event| {
        crate::progress::upload_event(event);
        !crate::signal::is_shutdown()
    })
    .with_admission_receipt_observer(|receipt| {
        eprintln!(
            "note: remote returned a {} receipt for {}",
            receipt.header,
            receipt
                .procedure
                .rsplit('/')
                .next()
                .unwrap_or(receipt.procedure)
        );
    });
    // Grants ride only on signed requests (SPEC-WRITE-GRANTS §4.2), so the
    // store is read only when there is a signer to present them with.
    let grants = signed.then(|| Arc::new(grants::LocalGrants::from_stored(load_user_grants(cfg))));
    if let Some(grants) = &grants {
        tx = tx.with_grant_source(grants.clone());
    }
    if !cfg.admission_helper.is_empty() && cfg.trusted_remote_endpoint.trim() == url {
        let responder = Arc::new(crate::admission_helper::ExecResponder {
            path: cfg.admission_helper.clone().into(),
        });
        let mut policy = AdmissionPolicy::new(responder);
        if let Some(name) = remote_name
            && let Some(headers) = cfg.remote_admission_headers.get(name)
        {
            let bearer = std::env::var("MKIT_API_TOKEN").is_ok_and(|s| !s.is_empty());
            for header in headers.split(',').map(str::trim).filter(|s| !s.is_empty()) {
                if is_reserved(header, bearer) {
                    eprintln!(
                        "warning: ignoring reserved header `{header}` in remote.{name}.admission_headers (see SPEC-TRANSPORT-CONNECT §5.1)"
                    );
                } else {
                    policy = policy.with_extra_allowed(header);
                }
            }
        }
        tx = tx.with_admission(policy);
    }
    Ok(ConnectParts {
        transport: tx,
        grants,
        signer_key,
    })
}

/// The verified grants in the user store, one warning per skipped file.
fn load_user_grants(cfg: &crate::config::Config) -> Vec<crate::grants::store::StoredGrant> {
    let rps = crate::grants::parse_relying_parties(&cfg.grant_webauthn_rp).unwrap_or_else(|e| {
        eprintln!("warning: grant.webauthn_rp: {e}; treating no relying party as pinned");
        Vec::new()
    });
    let store = match crate::grants::store::GrantStore::open_default() {
        Ok(store) => store,
        Err(e) => {
            eprintln!("warning: {e}; using no stored grants");
            return Vec::new();
        }
    };
    let report = store.load(&rps);
    for warning in &report.warnings {
        eprintln!("warning: {warning}");
    }
    report.grants
}

/// The configured mkit signing key as an owner signer for `mkit grant`,
/// `mkit epoch` and `mkit visibility`. Same key resolution as
/// [`envelope_signer_from_config`], without its `transport_auth` gate: the
/// owner asked for this signature explicitly.
pub(crate) fn owner_ed25519_signer(
    cfg: &crate::config::Config,
    layout: &RepoLayout,
) -> Result<Arc<dyn mkit_transport_connect::EnvelopeSigner>, String> {
    match cfg.signer.as_str() {
        "" | "legacy" => {
            let key_path = crate::config::resolve_key_path(layout, &cfg.signing_key)
                .map_err(|e| format!("signing_key: {e}"))?;
            if !key_path.exists() {
                return Err(format!(
                    "no signing key at {} — run `mkit keygen` first",
                    key_path.display()
                ));
            }
            let kp = mkit_core::sign::load_key(&key_path).map_err(|e| format!("load key: {e}"))?;
            Ok(Arc::new(envelope_signer::RepoKeyEnvelopeSigner::new(kp)))
        }
        "keystore" => Ok(Arc::new(envelope_signer::KeystoreEnvelopeSigner::open(
            cfg,
        )?)),
        other => Err(format!(
            "unknown signer `{other}` — expected `legacy` or `keystore`"
        )),
    }
}

/// Resolve an [`mkit_transport_connect::EnvelopeSigner`] from `cfg`, when
/// `cfg.transport_auth_envelope()` is set — `Ok(None)` otherwise (the
/// default: bearer-token-only, unchanged from #700/#701).
///
/// Reuses EXACTLY the same signer resolution as `mkit commit`'s
/// [`crate::commands::commit::load_commit_signer`] (`cfg.signer` ==
/// `""`/`"legacy"` -> the repo key file at `cfg.signing_key`; `"keystore"`
/// -> `cfg.key.ed25519_ref_or_fallback()` via `mkit-keystore`) rather than
/// inventing a parallel key path — the write envelope authenticates with
/// the SAME Ed25519 identity that already signs the user's commits.
///
/// Both signer kinds sign the raw envelope digest directly (no
/// SPEC-SIGNING commit/remix/tag domain prefix): the legacy path delegates
/// to the EXISTING `mkit_attest::RepoKeySigner` (its `sign` already signs
/// the given bytes directly — "the PAE's own `\"DSSEv1 \"` prefix is the
/// domain separator" per its own doc comment — so no new raw-Ed25519 call
/// site is needed here), the keystore path via `KeySigner::sign`, whose
/// own contract already documents "Ed25519 signers return the 64-byte
/// RFC 8032 signature over `msg`" — i.e. no domain digest applied, exactly
/// what the envelope needs. See `envelope_signer.rs` for both adapters.
pub(crate) fn envelope_signer_from_config(
    cfg: &crate::config::Config,
    layout: &RepoLayout,
) -> Result<Option<Arc<dyn mkit_transport_connect::EnvelopeSigner>>, DispatchError> {
    if !cfg.transport_auth_envelope() {
        return Ok(None);
    }
    let remote_error = |msg: String| DispatchError::Transport(TransportError::RemoteError(msg));
    match cfg.signer.as_str() {
        "" | "legacy" => {
            let key_path =
                crate::config::resolve_key_path(layout, &cfg.signing_key).map_err(|e| {
                    remote_error(format!("transport_auth = envelope: signing_key: {e}"))
                })?;
            if !key_path.exists() {
                return Err(remote_error(format!(
                    "transport_auth = envelope requires a signing key at {} — run `mkit keygen` first",
                    key_path.display()
                )));
            }
            let kp = mkit_core::sign::load_key(&key_path)
                .map_err(|e| remote_error(format!("transport_auth = envelope: load key: {e}")))?;
            Ok(Some(
                Arc::new(envelope_signer::RepoKeyEnvelopeSigner::new(kp))
                    as Arc<dyn mkit_transport_connect::EnvelopeSigner>,
            ))
        }
        "keystore" => {
            let signer = envelope_signer::KeystoreEnvelopeSigner::open(cfg)
                .map_err(|e| remote_error(format!("transport_auth = envelope: {e}")))?;
            Ok(Some(
                Arc::new(signer) as Arc<dyn mkit_transport_connect::EnvelopeSigner>
            ))
        }
        other => Err(remote_error(format!(
            "transport_auth = envelope: unknown signer `{other}` — expected `legacy` or `keystore`"
        ))),
    }
}

/// Map the three `ssh.*` trust-pinning keys from a merged [`Config`] into
/// the [`SshOptions`] carried to the spawned `ssh(1)` child. An empty
/// string means "unset" — `build_ssh_command` emits no flag for it, so
/// the user's `ssh(1)` defaults are inherited. The producer half of
/// issue #389 (the consumer half, `build_ssh_command`, wires the fields
/// into argv). Sole caller is [`open_with_config`].
fn ssh_options_from_config(cfg: &crate::config::Config) -> SshOptions {
    SshOptions {
        strict_host_key_checking: cfg.ssh_strict_host_key_checking.clone(),
        user_known_hosts_file: cfg.ssh_user_known_hosts_file.clone(),
        identity_file: cfg.ssh_identity_file.clone(),
    }
}

/// Open a transport for the given URL with **no** SSH trust-pinning.
/// Returns a type-erased `Arc` so callers can treat all schemes
/// uniformly.
///
/// Low-level scheme dispatch only — it neither enforces the credential
/// gate nor threads `ssh.*` config. Any caller that has a [`Config`]
/// must use `open_with_config` (directly, or via `open_trusted`) so
/// the trust-pinning keys reach the spawned `ssh(1)`; `open` stays
/// public only for file/memory integration tests that have no ambient
/// config to resolve.
///
/// [`Config`]: crate::config::Config
pub fn open(url: &str) -> Result<Arc<dyn Transport>, DispatchError> {
    open_with_ssh_options(url, &SshOptions::default(), None)
}

/// Scheme dispatch with explicit SSH options and an optional `mkit+https://`
/// / `mkit+http://` envelope signer. Identical to [`open`] for every
/// non-SSH, non-Connect scheme; the `mkit+ssh://` branch threads
/// `ssh_options` (issue #389) into the spawned `ssh(1)` child via
/// [`SshTransport::connect_with_options`], and the `mkit+https://`/
/// `mkit+http://` branch threads `envelope_signer` (issue #699 follow-up)
/// into [`ConnectTransport::connect_with_signer`]. Reached via [`open`]
/// (no config — both `None`/default) and [`open_with_config`] for non-Connect
/// schemes; `open_with_config` handles Connect URLs itself, so the Connect
/// branch here is reached only from [`open`].
fn open_with_ssh_options(
    url: &str,
    ssh_options: &SshOptions,
    envelope_signer: Option<Arc<dyn mkit_transport_connect::EnvelopeSigner>>,
) -> Result<Arc<dyn Transport>, DispatchError> {
    if url.starts_with("git+") {
        return Err(DispatchError::UnsupportedScheme(format!(
            "'{url}' is a git-bridge remote — native push/pull/fetch/clone do not \
             speak git transports; use `mkit git export` / `mkit git import` / \
             `mkit git pull` (feature git-bridge)"
        )));
    }
    if let Some(rest) = url.strip_prefix("mkit+file://") {
        // mkit+file:///abs/path -> /abs/path
        let path = Path::new(rest);
        return Ok(Arc::new(FileTransport::new(path)));
    }
    if url.starts_with("mkit+memory://") {
        // Memory transport is in-process; the URL-based path is not
        // useful on its own but we accept it so `mkit remote add`
        // round-trips cleanly.
        return Err(DispatchError::UnsupportedScheme(
            "mkit+memory:// must be driven via in-process harness (see tests)".to_string(),
        ));
    }
    if url.starts_with("mkit+https://") || url.starts_with("mkit+http://") {
        validate_connect_repository(url)?;
        // ConnectTransport::connect_with_signer strips the `mkit+` prefix
        // itself and reads MKIT_API_TOKEN from the environment (mkit#701 —
        // the native mkit.transport.v1 ConnectRPC client, replacing the
        // retired mkit-transport-http JSON dialect as of
        // SPEC-TRANSPORT-CONNECT verb parity). Only the config-less `open`
        // reaches this branch (`open_with_config` returns early for Connect
        // URLs), so `envelope_signer` is always `None` here; bearer token and
        // envelope signing are independent, additive auth modes.
        let tx = ConnectTransport::connect_with_signer(url, envelope_signer)?;
        return Ok(Arc::new(tx));
    }
    if url.starts_with("mkit+s3://") {
        // S3Transport::connect reads MKIT_R2_ACCESS_KEY_ID /
        // MKIT_R2_SECRET_ACCESS_KEY from the environment. Missing
        // credentials surface as AccessDenied on the first signed call,
        // not at connect time.
        let tx = S3Transport::connect(url)?;
        return Ok(Arc::new(tx));
    }
    if url.starts_with("mkit+ssh://") {
        // Parse the URL, then spawn `ssh(1)` with the caller-supplied
        // trust-pinning options (issue #389). `connect_with_options`
        // performs the `Hello` / `HelloResponse` handshake. Any failure
        // here tears the child down before returning, so callers never
        // see a half-initialised transport.
        let target = parse_mkit_ssh_url(url).map_err(SshInitError::from)?;
        let tx = SshTransport::connect_with_options(&target, ssh_options)?;
        return Ok(Arc::new(tx));
    }
    #[cfg(feature = "enc-transport")]
    if url.starts_with("mkit+enc://") {
        return open_enc(url);
    }
    Err(DispatchError::MalformedUrl(url.to_string()))
}

fn validate_connect_repository(url: &str) -> Result<(), DispatchError> {
    repository_identity_from_url(url).map_err(|reason| {
        let path = url
            .split_once("://")
            .and_then(|(_, rest)| rest.split_once('/'))
            .map_or("", |(_, path)| path)
            .split(['?', '#'])
            .next()
            .unwrap_or("")
            .trim_matches('/');
        DispatchError::MalformedUrl(format!("repository identity `{path}` in {url}: {reason}"))
    })?;
    Ok(())
}

/// `mkit+enc://` dispatch (issue #156).
///
/// Parses the URL, derives an ephemeral dialer keypair (keystore
/// integration is SPEC-TRANSPORT-ENC §6 item 5, still deferred), and
/// runs the encrypted-stream handshake against the URL-advertised
/// server public key.
///
/// Client identity (issue #178): an allowlisting server pins the
/// dialer's static ed25519 key. To survive across restarts the client
/// can supply a STABLE raw-32 key file via the `MKIT_ENC_CLIENT_KEY`
/// environment variable (a user-scoped / CLI-supplied path — never
/// repo-local `.mkit/config`, which `open_enc` has no access to anyway).
/// When the variable is unset we fall back to a fresh ephemeral key per
/// process, which still works against an allow-any enc listener.
#[cfg(feature = "enc-transport")]
const ENC_CLIENT_KEY_ENV: &str = "MKIT_ENC_CLIENT_KEY";

#[cfg(feature = "enc-transport")]
fn open_enc(url: &str) -> Result<Arc<dyn Transport>, DispatchError> {
    use mkit_transport_enc::url::parse_enc_url;

    let target = parse_enc_url(url).map_err(DispatchError::Transport)?;
    let sk = load_or_ephemeral_client_key()?;
    let tx = mkit_transport_enc::connect_tcp(&target.host, target.port, &target.server_pubkey, sk)
        .map_err(|e| DispatchError::Transport(TransportError::RemoteError(e.to_string())))?;
    Ok(Arc::new(tx))
}

/// Resolve the dialer's static signing key.
///
/// If `MKIT_ENC_CLIENT_KEY` points at a raw 32-byte key file, load it
/// (with the standard `load_raw_32` 0600/owner hardening) so the
/// client's public key is stable — letting an allowlisting server pin
/// it across restarts. Otherwise draw a fresh ephemeral key from the
/// system RNG (≥256 bits) for back-compat with allow-any servers.
#[cfg(feature = "enc-transport")]
fn load_or_ephemeral_client_key()
-> Result<commonware_cryptography::ed25519::PrivateKey, DispatchError> {
    use commonware_codec::DecodeExt as _;
    use commonware_cryptography::ed25519::PrivateKey;
    use zeroize::Zeroizing;

    let map_err = |e: String| DispatchError::Transport(TransportError::RemoteError(e));

    if let Some(path) = std::env::var_os(ENC_CLIENT_KEY_ENV).filter(|s| !s.is_empty()) {
        let seed = mkit_core::sign::load_raw_32(std::path::Path::new(&path))
            .map_err(|e| map_err(format!("load {ENC_CLIENT_KEY_ENV}: {e}")))?;
        return PrivateKey::decode(seed.as_ref())
            .map_err(|e| map_err(format!("client key construction failed: {e}")));
    }

    // Ephemeral fallback. Draw 32 bytes from `getrandom`, wrapped in
    // `Zeroizing` so the stack copy is scrubbed on drop; the resulting
    // `PrivateKey` carries its own `Secret`-based zeroization.
    let mut secret = Zeroizing::new([0u8; 32]);
    getrandom::fill(secret.as_mut()).map_err(|e| map_err(e.to_string()))?;
    PrivateKey::decode(secret.as_ref()).map_err(|e| map_err(e.to_string()))
}

/// Push every ref under `refs/heads/` to the remote. Returns the count of
/// refs pushed. Each branch is published with [`push_branch`], which sends
/// one delta-compressed pack of the objects the remote lacks, advertises it
/// via the `refs/mkit/packmap/<branch>` ref, then moves the branch ref.
pub fn push_all(cwd: &Path, tx: &dyn Transport) -> Result<usize, DispatchError> {
    push_all_with(cwd, tx, None, false, None).map(|pushed| pushed.refs)
}

/// What a successful push published.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Pushed {
    /// Branches pushed.
    pub refs: usize,
    /// Advances published in total (one per branch unless a push was split).
    pub steps: usize,
}

/// CAS-aware mirror push (`mkit push --all`). Pushes every local
/// `refs/heads/*` to the remote, using the remote-tracking ref under
/// `refs/remotes/<remote>/<branch>` as the CAS lease (Missing when no
/// tracking ref exists, Match otherwise). `force` upgrades every write
/// to an unconditional `Any`. On success each pushed branch's
/// remote-tracking ref is advanced to the pushed tip.
///
/// `remote` is the remote NAME used for the local tracking-ref
/// namespace; `None` means the legacy `default`. A branch too large for one
/// advance is split (see [`push_branch_steps`]); its tracking ref follows each
/// published advance, so an interrupted push resumes where it stopped.
/// `authority` is checked before a split branch uploads anything.
pub fn push_all_with(
    cwd: &Path,
    tx: &dyn Transport,
    remote: Option<&str>,
    force: bool,
    authority: Option<&dyn StepAuthority>,
) -> Result<Pushed, DispatchError> {
    let layout = mkit_core::layout::discover(cwd)?;
    let store = crate::commands::open_store_configured(&layout)?;
    let refs_list = crate::commands::list_refs_parallel(&layout)?;
    let remote = remote.unwrap_or(DEFAULT_REMOTE);
    let mut n = 0;
    let mut steps = 0;
    let shallow = shallow_boundaries(&layout)?;
    // Batch every pushed branch's remote-tracking-ref write (#645):
    // publishing lands each ref as soon as its branch's `push_branch`
    // succeeds (same visibility as before), but the directory fsync that
    // makes those renames crash-durable is deferred to one pass over the
    // distinct directories touched, below — instead of once per branch.
    let mut tracking = refs::RemoteRefBatch::new(&layout, remote)?;
    let result: Result<(), DispatchError> = (|| {
        for r in refs_list {
            if crate::signal::is_shutdown() {
                return Err(DispatchError::Interrupted);
            }
            let Some(h) = r.hash else { continue };
            let condition = if force {
                refs::RefWriteCondition::Any
            } else {
                match refs::read_remote_ref(&layout, remote, &r.name)? {
                    Some(tracked) => refs::RefWriteCondition::Match(tracked),
                    None => refs::RefWriteCondition::Missing,
                }
            };
            let control = PushControl {
                authority,
                shallow: shallow.clone(),
                ..PushControl::default()
            };
            steps += push_branch_steps(
                tx,
                &store,
                &r.name,
                h,
                condition,
                rebaseline_depth(),
                pack::MAX_TOTAL_PAYLOAD,
                &control,
                &mut |published| Ok(tracking.write(&r.name, &published)?),
            )?;
            n += 1;
        }
        Ok(())
    })();
    // Commit whatever tracking-ref writes succeeded regardless of how the
    // loop above ended, so a mid-loop failure still durably publishes the
    // prefix that already pushed successfully — matching the old
    // per-branch loop, where each completed ref write was independently
    // durable before the loop moved to the next branch.
    tracking.commit()?;
    result?;
    Ok(Pushed { refs: n, steps })
}

fn shallow_boundaries(
    layout: &RepoLayout,
) -> Result<std::collections::HashSet<Hash>, DispatchError> {
    Ok(refs::load_shallow_boundaries(layout)?
        .unwrap_or_default()
        .into_iter()
        .collect())
}

/// True iff advancing a ref from `old` to `new` is a fast-forward (i.e.
/// `old` is an ancestor of `new`). A missing `old` (brand-new ref) and an
/// unchanged ref both count as fast-forwards. Used by `push`/`fetch` to
/// pick the git-style summary symbol (`..` vs `...`/`(forced update)`).
pub fn is_fast_forward(cwd: &Path, old: Option<Hash>, new: Hash) -> Result<bool, DispatchError> {
    match old {
        None => Ok(true),
        Some(o) if o == new => Ok(true),
        Some(o) => {
            let layout = mkit_core::layout::discover(cwd)?;
            let store = crate::commands::open_store_configured(&layout)?;
            Ok(is_ancestor(&store, o, new)?)
        }
    }
}

/// CAS lease policy for a default (current-branch → upstream) push.
#[derive(Debug, Clone, Copy)]
pub enum PushLease {
    /// Force — unconditional `Any`.
    Force,
    /// `--force-with-lease` — require the remote tip to equal the local
    /// remote-tracking ref (Match), or Missing when there is none.
    /// Identical mechanism to the default safe push; semantically it is
    /// the explicit, opt-in form that overwrites a fast-forward-failing
    /// branch *only* if the remote hasn't moved past what we last saw.
    WithLease,
    /// Default safe push: Match the local remote-tracking ref, or
    /// Missing when absent (first push of this branch).
    FastForward,
}

/// Resolve the CAS condition for a single-branch push from the local
/// remote-tracking ref `refs/remotes/<remote>/<branch>` and the lease
/// policy.
pub fn lease_condition(
    cwd: &Path,
    remote: &str,
    branch: &str,
    lease: PushLease,
) -> Result<refs::RefWriteCondition, DispatchError> {
    if matches!(lease, PushLease::Force) {
        return Ok(refs::RefWriteCondition::Any);
    }
    let layout = mkit_core::layout::discover(cwd)?;
    Ok(match refs::read_remote_ref(&layout, remote, branch)? {
        Some(tracked) => refs::RefWriteCondition::Match(tracked),
        None => refs::RefWriteCondition::Missing,
    })
}

/// Push the current branch to its upstream and, on success, advance the
/// local remote-tracking ref `refs/remotes/<remote>/<branch>` to the
/// pushed tip.
///
/// `remote` is the upstream remote NAME (for the tracking-ref
/// namespace); `branch` is the local branch name; `remote_branch` is the
/// branch name on the remote (`refs/heads/<remote_branch>`).
///
/// Returns the pushed tip and the number of advances it took: more than one
/// when the push was split along first-parent history (see
/// [`push_branch_steps`]). The tracking ref follows every published advance.
pub fn push_branch_tracked(
    cwd: &Path,
    tx: &dyn Transport,
    remote: &str,
    branch: &str,
    remote_branch: &str,
    lease: PushLease,
    authority: Option<&dyn StepAuthority>,
) -> Result<(Hash, usize), DispatchError> {
    let layout = mkit_core::layout::discover(cwd)?;
    let store = crate::commands::open_store_configured(&layout)?;
    let tip = refs::read_ref(&layout, branch)?
        .ok_or_else(|| DispatchError::RemoteBranchMissing(branch.to_owned()))?;
    // Default safe push requires a TRUE fast-forward: the new tip must
    // descend from the last-seen remote-tracking ref. The CAS `Match`
    // lease alone only proves the remote hasn't moved since we last
    // fetched — on its own it would still let a divergent local tip
    // (e.g. after a local `reset` to an unrelated commit) overwrite the
    // remote, which Git rejects as non-fast-forward. `--force-with-lease`
    // (`WithLease`) intentionally skips this check (overwrite as long as
    // the remote matches what we last saw); `Force` skips everything.
    if matches!(lease, PushLease::FastForward)
        && let Some(tracked) = refs::read_remote_ref(&layout, remote, remote_branch)?
        && !is_ancestor(&store, tracked, tip)?
    {
        return Err(DispatchError::NonFastForwardPush {
            branch: remote_branch.to_owned(),
        });
    }
    let condition = lease_condition(cwd, remote, remote_branch, lease)?;
    let control = PushControl {
        authority,
        shallow: shallow_boundaries(&layout)?,
        ..PushControl::default()
    };
    let steps = push_branch_steps(
        tx,
        &store,
        remote_branch,
        tip,
        condition,
        rebaseline_depth(),
        pack::MAX_TOTAL_PAYLOAD,
        &control,
        &mut |published| {
            Ok(refs::write_remote_ref(
                &layout,
                remote,
                remote_branch,
                &published,
            )?)
        },
    )?;
    Ok((tip, steps))
}

/// Push one branch: upload one or more delta-compressed packs carrying
/// every object reachable from `tip` that the remote lacks — split
/// across multiple packs when the plan's payload exceeds a single
/// pack's cap (issue #831) — durably advertise them as one node on the
/// `refs/mkit/packmap/<branch>` metadata ref, then CAS-write
/// `refs/heads/<branch>` under `condition`.
///
/// Objects already present at the remote's current tip are never re-sent
/// (identical-object dedup), and changed `FastCDC` chunks are delta-encoded
/// against the prior version the remote already holds when that saves
/// bytes (see [`mkit_core::transfer::plan_pack`]). The pack is keyed by its
/// own BLAKE3 digest (SPEC-PACKFILE §7) — required because the digest-
/// checking storage server rejects a delta stored under the reconstructed
/// object's hash.
///
/// The packmap is advanced *and confirmed* before the branch ref moves: if
/// the packmap can't be durably established the push aborts without
/// touching the head, so the head never points past a packmap that fails to
/// reconstruct it (even under concurrent pushers to the same branch).
///
/// Plans the pack FIRST (diffing against the remote's current tip). A no-op
/// push (empty plan — the remote already holds this closure) takes the cheap
/// head-only path and walks NO packmap chain (mkit #521 perf). Only when the
/// plan is non-empty does it resolve the branch's current packmap chain depth
/// (walking it exactly once, see `packmap::probe_chain`) and, if the chain
/// would grow past the re-baseline threshold (#406, see
/// `packmap::rebaseline_depth`) AND the transport's `advance_refs` is
/// transactional ([`Transport::supports_atomic_advance`], mkit #521) AND the
/// head write is CAS-conditioned (a force push's `Any` head condition takes
/// the safe append path — an `Any` condition makes even an atomic transport
/// fall back to the ordered two-PUT `advance_refs`, so a reset there is not
/// safe), re-plans as a full closure (diffs against no remote tip) and
/// carries that decision down to `advance_packmap` as
/// `ChainAction::ResetSelfContained` so it resets the chain to a single
/// fresh node instead of appending to it — bounding clone cost, which
/// otherwise grows with chain length.
///
/// On a transport WITHOUT transactional `advance_refs` (the default used by
/// file/S3/SSH/memory), crossing the threshold never triggers a reset: the
/// default `advance_refs` commits the packmap write before the head CAS, and
/// a reset (unlike an append) is not a superset of the prior chain, so a
/// lost head-CAS race after a committed reset would strand the (unmoved)
/// head pointing at a commit the packmap can no longer reconstruct. Such a
/// transport keeps appending — `ChainAction::Append` — past the
/// threshold; `packmap::MAX_PACK_CHAIN_DEPTH` (the pure runaway/cycle guard)
/// remains the only bound on chain growth there, unchanged by this gate.
///
/// A chain read that fails with [`DispatchError::PackChainInvalid`], or a
/// missing prior packmap (first push), is left alone here — depth is only
/// defined for a resolvable chain, and a broken chain already has its own
/// reset path in `advance_packmap` (the broken-chain escape hatch, gated on
/// `self_contained` alone, independent of this transactional-advance gate —
/// see `ChainAction::Append`'s doc comment).
///
/// The already-resolved chain from this probe (when not discarded by a
/// re-baseline decision) is threaded into `advance_packmap` so its first
/// CAS attempt does not have to walk the chain a second time (#521 perf
/// fix).
///
/// On a CAS failure ([`TransportError::RefConflict`]) this reads the head
/// back first (SPEC-TRANSPORT §7: a retried write can report a conflict for
/// its own landed attempt); a head already at `tip` is success, anything
/// else returns [`DispatchError::NonFastForwardPush`] so callers can render
/// an actionable fetch-then-retry hint. Does NOT touch local
/// remote-tracking refs — the caller decides when to advance them.
pub fn push_branch(
    tx: &dyn Transport,
    store: &ObjectStore,
    branch: &str,
    tip: Hash,
    condition: refs::RefWriteCondition,
) -> Result<(), DispatchError> {
    push_branch_with_depth(tx, store, branch, tip, condition, rebaseline_depth())
}

/// [`push_branch`] with an explicit re-baseline threshold in place of the
/// configured one (`packmap::rebaseline_depth`: the
/// `MKIT_PACK_REBASELINE_DEPTH` env var, default 64). `0` disables
/// re-baselining. Semantics are otherwise identical — see [`push_branch`].
///
/// This is the depth-policy seam (#547): integration tests inject a small
/// threshold (e.g. 3) to exercise the re-baseline path in-process with a
/// handful of pushes, where reaching the default threshold would take ~64
/// real pushes and the env-var override cannot be set on the test's own
/// process (`std::env::set_var` is banned by `clippy::disallowed_methods` —
/// it races other threads on POSIX).
pub fn push_branch_with_depth(
    tx: &dyn Transport,
    store: &ObjectStore,
    branch: &str,
    tip: Hash,
    condition: refs::RefWriteCondition,
    rebaseline_threshold: usize,
) -> Result<(), DispatchError> {
    push_branch_with_limits(
        tx,
        store,
        branch,
        tip,
        condition,
        rebaseline_threshold,
        pack::MAX_TOTAL_PAYLOAD,
    )
}

/// [`push_branch_with_depth`] with an explicit per-pack payload cap in
/// place of the format's hardcoded [`pack::MAX_TOTAL_PAYLOAD`] (issue
/// #831). Semantics are otherwise identical — see [`push_branch`].
///
/// This is the payload-cap test seam, mirroring the #547
/// `rebaseline_threshold` pattern: integration tests inject a tiny cap
/// (a few KiB) to exercise multi-pack splitting in-process without
/// moving a multi-GiB plan through it.
pub fn push_branch_with_limits(
    tx: &dyn Transport,
    store: &ObjectStore,
    branch: &str,
    tip: Hash,
    condition: refs::RefWriteCondition,
    rebaseline_threshold: usize,
    pack_payload_cap: u64,
) -> Result<(), DispatchError> {
    push_branch_steps(
        tx,
        store,
        branch,
        tip,
        condition,
        rebaseline_threshold,
        pack_payload_cap,
        &PushControl::default(),
        &mut |_| Ok(()),
    )
    .map(drop)
}

/// [`push_branch_with_limits`] that splits a push too large for one advance
/// along the branch's first-parent history (WP-1.17b, R-164), and reports each
/// published advance to `on_step` with the commit the remote branch now
/// points at. Returns the number of advances.
///
/// The split applies only when the transport ticketed-uploads
/// ([`UploadLimits::tickets_per_advance`]) and the conservative estimate of the
/// push's packs exceeds the data-pack budget of one advance. Otherwise this is
/// exactly one [`push_branch_with_limits`] call. When it applies, the first
/// advance uses `condition`, and every later one is a `Match` on the previous
/// advance's commit, whatever `condition` was, so a long split never
/// overwrites a concurrent pusher. Each advance uploads its own packs just
/// before it lands; nothing is pre-uploaded.
///
/// A failure after the first advance is returned as
/// [`DispatchError::SplitInterrupted`], naming the published prefix; running
/// the push again resumes from it.
///
/// # Errors
/// As [`push_branch`], plus the split's own refusals (see
/// [`plan_push_steps`]), and whatever `on_step` returns.
#[allow(clippy::too_many_arguments)]
pub fn push_branch_steps(
    tx: &dyn Transport,
    store: &ObjectStore,
    branch: &str,
    tip: Hash,
    condition: refs::RefWriteCondition,
    rebaseline_threshold: usize,
    pack_payload_cap: u64,
    control: &PushControl<'_>,
    on_step: &mut dyn FnMut(Hash) -> Result<(), DispatchError>,
) -> Result<usize, DispatchError> {
    let limits = tx.upload_limits();
    let mut single = |plan| {
        push_step_recovering(
            tx,
            store,
            branch,
            tip,
            condition,
            rebaseline_threshold,
            pack_payload_cap,
            plan,
        )?;
        on_step(tip)?;
        Ok(1)
    };
    if limits.tickets_per_advance.is_none() {
        return single(None);
    }
    refs::check_pushable_branch(branch)?;
    let remote_tip = tx.read_ref(&format!("refs/heads/{branch}"))?;
    let plan = transfer::plan_pack_with(store, tip, remote_tip, encode_delta_candidates_batch)?;
    let cap = effective_payload_cap(pack_payload_cap, limits.max_pack_bytes)?;
    if estimate_pack_sizes(store, &plan, cap, limits.max_pack_bytes)?.len()
        <= split::data_pack_budget(limits)
    {
        return single(Some(plan));
    }
    // The estimate is uncompressed: a push that compresses into the budget
    // is not split (exact dry seal), and one against a remote tip this store
    // lacks is left to the head CAS rather than walking all of history.
    let unknown_remote = remote_tip.is_some_and(|remote| !store.contains(&remote));
    if unknown_remote {
        return single(None);
    }
    match build_and_upload_packs(PackSink::Count, store, plan, cap, limits) {
        Err(DispatchError::PushTooLarge { .. }) => {}
        Err(DispatchError::Interrupted) => return Err(DispatchError::Interrupted),
        _ => return single(None),
    }
    let steps = plan_push_steps(
        store,
        tip,
        remote_tip,
        limits,
        pack_payload_cap,
        control,
        condition,
        branch,
    )?;
    if steps.len() == 1 {
        return single(None);
    }
    let total = steps.len();
    let mut published: Option<Hash> = None;
    for (index, step_tip) in steps.into_iter().enumerate() {
        let step_condition = published.map_or(condition, refs::RefWriteCondition::Match);
        let mut landed = false;
        let outcome = (|| {
            if crate::signal::is_shutdown() {
                return Err(DispatchError::Interrupted);
            }
            crate::progress::report(crate::progress::Event::Step {
                index: index + 1,
                total,
            });
            push_step_recovering(
                tx,
                store,
                branch,
                step_tip,
                step_condition,
                rebaseline_threshold,
                pack_payload_cap,
                None,
            )?;
            landed = true;
            on_step(step_tip)?;
            crate::progress::step_committed(index + 1, total, &step_tip);
            Ok(())
        })();
        if landed {
            published = Some(step_tip);
        }
        if let Err(cause) = outcome {
            return Err(match published {
                Some(head) => DispatchError::SplitInterrupted {
                    branch: branch.to_owned(),
                    head: mkit_core::hash::to_hex(&head),
                    published: index + usize::from(landed),
                    total,
                    cause: Box::new(cause),
                },
                None => cause,
            });
        }
    }
    Ok(total)
}

/// One advance to `tip` with the restart-once recovery: a rejected ticket or
/// packlist re-plans the same way, a missing delta base re-plans as the full
/// closure. `plan` is an already-computed plan against the remote's current
/// tip, used for the first attempt only.
#[allow(clippy::too_many_arguments)]
fn push_step_recovering(
    tx: &dyn Transport,
    store: &ObjectStore,
    branch: &str,
    tip: Hash,
    condition: refs::RefWriteCondition,
    rebaseline_threshold: usize,
    pack_payload_cap: u64,
    plan: Option<transfer::PackPlan>,
) -> Result<(), DispatchError> {
    let once = |retry, plan| {
        push_branch_once(
            tx,
            store,
            branch,
            tip,
            condition,
            rebaseline_threshold,
            pack_payload_cap,
            retry,
            plan,
        )
    };
    match once(None, plan) {
        Err(DispatchError::TicketRejected | DispatchError::PacklistNotInRepository) => {
            once(Some(RetryReason::Replanned), None)
        }
        Err(DispatchError::DeltaBaseUnavailable) => once(Some(RetryReason::LostDeltaBase), None),
        result => result,
    }
}

#[allow(clippy::too_many_arguments)]
fn push_branch_once(
    tx: &dyn Transport,
    store: &ObjectStore,
    branch: &str,
    tip: Hash,
    condition: refs::RefWriteCondition,
    rebaseline_threshold: usize,
    pack_payload_cap: u64,
    retry: Option<RetryReason>,
    preplanned: Option<transfer::PackPlan>,
) -> Result<(), DispatchError> {
    let force_self_contained = retry == Some(RetryReason::LostDeltaBase);
    // Both wire names must fit SPEC-REFS §3's bound; a local branch named
    // before `MAX_BRANCH_NAME_BYTES` existed may not. Say so by name
    // rather than as a transport's "invalid ref name".
    refs::check_pushable_branch(branch)?;
    // Diff against the remote's CURRENT tip so we send only what it lacks
    // and can delta against bases it already holds. Planning is an
    // optimization; the head CAS below remains authoritative.
    // A caller that already planned against the remote's current tip (the
    // split decision does) passes its plan, so it is diffed once.
    let mut plan = if let Some(plan) = preplanned {
        plan
    } else {
        let remote_tip = tx.read_ref(&format!("refs/heads/{branch}"))?;
        // Plan FIRST, against the remote's current tip (#521 perf): a no-op push
        // (the remote already holds this closure) yields an empty plan and takes
        // the cheap head-only path below WITHOUT walking the packmap chain. Only
        // a push that actually has objects to send pays the O(depth) chain probe.
        transfer::plan_pack_with(
            store,
            tip,
            if force_self_contained {
                None
            } else {
                remote_tip
            },
            encode_delta_candidates_batch,
        )?
    };

    let limits = tx.upload_limits();
    // Payload and serialized-byte limits are checked separately. Framing
    // overhead grows with the entry count, so a fixed margin is insufficient.
    let effective_cap = effective_payload_cap(pack_payload_cap, limits.max_pack_bytes)?;

    if plan.is_empty() {
        // Nothing to send — the remote already holds the closure; just move
        // the head (no packmap change needed, so no chain walk and no atomic
        // two-ref advance).
        return commit_head(tx, &format!("refs/heads/{branch}"), condition, &tip, branch);
    }

    if crate::signal::is_shutdown() {
        return Err(DispatchError::Interrupted);
    }

    // Re-baseline decision (#406/#521), made only now that we know there IS
    // something to send: walk the current chain exactly once and decide
    // whether this push should reset the chain to a single self-contained
    // node instead of appending. When NOT re-baselining, the walk is cached
    // (`resolved_chain`) so `advance_packmap`'s append path can reuse it
    // instead of re-walking.
    //
    // A reset is committed together with the head; the ordered (non-atomic)
    // `advance_refs` fallback (packmap PUT then head PUT) would strand the
    // head at the old tip on a torn write, and a reset — unlike an append —
    // is not a superset of the prior chain, so it can't rebuild that stranded
    // closure. We therefore re-baseline ONLY when BOTH hold:
    //   * the transport advances both refs transactionally
    //     (`supports_atomic_advance`), AND
    //   * the head write is CAS-conditioned (not `Any`). A force push's `Any`
    //     head condition makes even an atomic transport fall back to the
    //     ordered two-PUT path (`Any` is not expressible on the atomic
    //     endpoint — see `HttpTransport::supports_atomic_advance`), so a
    //     force push MUST take the safe append path instead of resetting.
    let mut rebaseline = false;
    let mut resolved_chain = None;
    if !force_self_contained
        && rebaseline_threshold > 0
        && let Some(pm) = tx.read_ref(&packmap_ref(branch))?
    {
        match probe_chain(tx, branch, pm) {
            Ok(chain)
                if chain.depth + 1 > rebaseline_threshold
                    && tx.supports_atomic_advance()
                    && !matches!(condition, refs::RefWriteCondition::Any) =>
            {
                rebaseline = true;
            }
            Ok(chain) => resolved_chain = Some(chain),
            Err(DispatchError::PackChainInvalid { .. }) => {}
            Err(e) => return Err(e),
        }
    }

    if rebaseline {
        // Force a full-closure plan: no external bases, so the pack is
        // self-contained and safe to reset the chain onto.
        let full = transfer::plan_pack_with(store, tip, None, encode_delta_candidates_batch)?;
        let full_cap = effective_payload_cap(pack_payload_cap, limits.max_pack_bytes)?;
        if estimated_ticketed_count(
            &estimate_pack_sizes(store, &full, full_cap, limits.max_pack_bytes)?,
            limits,
        ) > split::data_pack_budget(limits)
        {
            rebaseline = false;
        } else {
            plan = full;
        }
    }

    // A retry re-planned the push: make sure the new plan still fits before
    // uploading any of it, so the failure leaves no half-uploaded advance.
    if let Some(reason) = retry
        && estimated_ticketed_count(
            &estimate_pack_sizes(store, &plan, effective_cap, limits.max_pack_bytes)?,
            limits,
        ) > split::data_pack_budget(limits)
        && let Err(DispatchError::PushTooLarge { packs, limit, .. }) =
            build_and_upload_packs(PackSink::Count, store, plan.clone(), effective_cap, limits)
    {
        return Err(DispatchError::RetryTooLarge {
            reason,
            packs,
            limit,
        });
    }

    // No pre-flight PushTooLarge: the estimate uses uncompressed sizes, so it
    // would refuse pushes that compress into six packs. The seal-time gate is
    // exact and still stops before the seventh ticketed pack's BeginUpload.
    // (The re-baseline check above may use the conservative estimate: a false
    // "too many" only keeps the append plan.)

    // Build the plan into one or more payload-bounded packs (splitting
    // when the plan exceeds `pack_payload_cap`, issue #831) and upload
    // each as it's sealed. Raws first (non-blobs before blobs), then
    // deltas (their bases are external — resolved from the fetcher's
    // store via earlier packs, never a base introduced earlier in THIS
    // push — so no in-pack ordering is required across the split,
    // SPEC-PACKFILE §4).
    //
    // `self_contained` is read out before `plan` moves into
    // `build_and_upload_packs` (which consumes it by value so its
    // delta streams can be handed to the compression fan-out without
    // an extra clone — see that function's doc comment).
    let self_contained = plan.self_contained;
    let head_ref = format!("refs/heads/{branch}");
    let pack_keys = build_and_upload_packs(
        PackSink::Upload {
            tx,
            head_ref: &head_ref,
        },
        store,
        plan,
        effective_cap,
        limits,
    )?;

    // Chain the pack(s) onto the packmap AND move the head together
    // (#408): a transactional transport applies both atomically, the
    // default does packmap-then-head. Either way the head never lands
    // past a packmap that can't reconstruct it. `Append`'s
    // `self_contained` lets a full-closure push reset a broken chain
    // (unconditionally, on any transport); `ResetSelfContained`
    // proactively resets a healthy chain that has grown too deep, and
    // is only ever chosen above when the transport is atomic-capable
    // AND the head write is CAS-conditioned. A failed advance leaves
    // the head untouched.
    let action = if rebaseline {
        ChainAction::ResetSelfContained
    } else {
        ChainAction::Append { self_contained }
    };
    advance_packmap(
        tx,
        branch,
        &pack_keys,
        action,
        resolved_chain,
        condition,
        tip,
    )
}

/// Build `plan`'s entries into one or more packs, each staying under
/// `payload_cap` bytes of wire payload, uploading each pack to `tx` as
/// soon as it's sealed. Returns the ordered pack keys — build order is
/// apply order, threaded straight into [`advance_packmap`].
///
/// A single linear left-to-right pass over the plan's already-ordered
/// `raw ++ deltas` sequence is enough: [`transfer::plan_pack`] never
/// deltas an entry against a base introduced earlier in THIS push
/// (every delta's base is already on the remote from a prior push), so
/// packs can be sealed independently the instant one would exceed the
/// cap — no intra-push base-ordering hazard to preserve across the
/// split.
///
/// Sizing uses a conservative *uncompressed* upper bound per entry
/// (`bytes.len()` for a raw, `HASH_LEN + stream.len()` for a delta)
/// checked against [`PackWriter::total_payload`]'s real (compressed)
/// running total, so a sealed pack never exceeds `payload_cap` — it may
/// under-fill when compression bites, yielding more packs than the
/// theoretical minimum, never fewer. Peak memory is one pack buffer
/// PLUS one prepared batch, not every pack (or the whole plan) held at
/// once: [`size_capped_batch_lens`] caps each batch at `payload_cap`
/// conservative bytes (the same budget the writer itself is bounded
/// by) as well as [`pack_fanout_threshold`]-many entries, so a plan
/// containing very large objects can't balloon a batch past what one
/// sealed pack would have held anyway — see that function's doc
/// comment for why both caps are needed.
///
/// Takes `plan` BY VALUE (not `&PackPlan`): the delta compression
/// fan-out needs owned `Vec<u8>` streams to hand to rayon, and
/// `plan.deltas` already owns them — draining them out here is zero
/// extra bytes copied, where borrowing would force a clone per entry
/// just to satisfy ownership. The caller reads whatever it needs off
/// `plan` (just `self_contained`) before making this call.
fn effective_payload_cap(requested: u64, advertised: Option<u64>) -> Result<u64, DispatchError> {
    // The serialized limit's overhead margin is exact and dynamic: header,
    // trailer and one frame per entry. `should_seal` applies it on each add.
    let cap = requested
        .min(pack::MAX_TOTAL_PAYLOAD)
        .min(advertised.unwrap_or(pack::MAX_TOTAL_PAYLOAD));
    if cap == 0 {
        return Err(DispatchError::Transport(TransportError::PayloadTooLarge(0)));
    }
    Ok(cap)
}

fn serialized_bound(payload: u64, entries: usize) -> u64 {
    payload
        .saturating_add((entries as u64).saturating_mul(pack::ENTRY_FRAME_LEN as u64))
        .saturating_add((pack::HEADER_LEN + pack::TRAILER_LEN) as u64)
}

fn ticketed_pack(limits: UploadLimits, serialized_size: u64) -> bool {
    limits.tickets_per_advance.is_some()
        && serialized_size >= limits.ticket_threshold_bytes.unwrap_or(0)
}

fn estimated_ticketed_count(sizes: &[u64], limits: UploadLimits) -> usize {
    sizes
        .iter()
        .filter(|&&size| ticketed_pack(limits, size))
        .count()
}

fn estimate_pack_sizes(
    store: &ObjectStore,
    plan: &transfer::PackPlan,
    payload_cap: u64,
    max_pack_bytes: Option<u64>,
) -> Result<Vec<u64>, DispatchError> {
    let mut packs = Vec::new();
    let mut payload = 0_u64;
    let mut entries = 0_usize;
    let sizes = plan
        .raw
        .iter()
        .map(|hash| store.object_metadata(hash).map(|meta| meta.len()))
        .collect::<Result<Vec<_>, _>>()?;
    for size in sizes.into_iter().chain(
        plan.deltas
            .iter()
            .map(|delta| (HASH_LEN + delta.stream.len()) as u64),
    ) {
        let next_payload = payload.saturating_add(size);
        if entries > 0
            && (next_payload > payload_cap
                || max_pack_bytes
                    .is_some_and(|limit| serialized_bound(next_payload, entries + 1) > limit))
        {
            packs.push(serialized_bound(payload, entries));
            payload = 0;
            entries = 0;
        }
        payload = payload.saturating_add(size);
        entries += 1;
    }
    if entries > 0 {
        packs.push(serialized_bound(payload, entries));
    }
    Ok(packs)
}

/// Where sealed packs go: to the remote, or nowhere (an exact local dry run of
/// the seal, for the split's pre-flight check).
#[derive(Clone, Copy)]
enum PackSink<'a> {
    Upload {
        tx: &'a dyn Transport,
        head_ref: &'a str,
    },
    Count,
}

fn build_and_upload_packs(
    sink: PackSink<'_>,
    store: &ObjectStore,
    plan: transfer::PackPlan,
    payload_cap: u64,
    limits: UploadLimits,
) -> Result<Vec<Hash>, DispatchError> {
    let mut pack_keys = Vec::new();
    let mut ticketed_count = 0;
    let mut w = PackWriter::new();
    let max_entries = pack_fanout_threshold().max(1);
    let transfer::PackPlan {
        raw, mut deltas, ..
    } = plan;

    // Raw entries: size each via a metadata `stat` (cheap — no content
    // read) so batches can be capped by real bytes, not just count,
    // *before* any object's content is actually read into memory.
    let raw_sizes: Vec<u64> = raw
        .iter()
        .map(|h| Ok(store.object_metadata(h)?.len()))
        .collect::<Result<_, DispatchError>>()?;
    let mut start = 0;
    for len in size_capped_batch_lens(&raw_sizes, payload_cap, max_entries) {
        let chunk = &raw[start..start + len];
        start += len;
        for entry in prepare_raw_batch(store, chunk)? {
            if should_seal(
                &w,
                entry.conservative_len() as u64,
                payload_cap,
                limits.max_pack_bytes,
            ) {
                seal_pack(sink, &mut w, &mut pack_keys, &mut ticketed_count, limits)?;
            }
            w.push_prepared_raw(entry)?;
            // Honest progress (#711): one real object just got staged into
            // the outgoing pack. Never git's fabricated
            // Enumerating/Counting/Compressing lines — see `crate::progress`.
            report_packed(sink);
        }
    }

    // Delta entries: sizes are already known (`stream.len()`, no I/O),
    // so batch lengths can be computed directly off them; `drain` then
    // moves each batch's `PlannedDelta`s out of `deltas` without
    // cloning their streams.
    let delta_sizes: Vec<u64> = deltas
        .iter()
        .map(|d| (HASH_LEN + d.stream.len()) as u64)
        .collect();
    for len in size_capped_batch_lens(&delta_sizes, payload_cap, max_entries) {
        let chunk: Vec<transfer::PlannedDelta> = deltas.drain(..len).collect();
        for entry in prepare_delta_batch(chunk) {
            if should_seal(
                &w,
                entry.conservative_len() as u64,
                payload_cap,
                limits.max_pack_bytes,
            ) {
                seal_pack(sink, &mut w, &mut pack_keys, &mut ticketed_count, limits)?;
            }
            w.push_prepared_delta(entry)?;
            report_packed(sink);
        }
    }

    seal_pack(sink, &mut w, &mut pack_keys, &mut ticketed_count, limits)?;
    Ok(pack_keys)
}

/// Honest progress (#711): one real object just got staged into the outgoing
/// pack. Never git's fabricated Enumerating/Counting/Compressing lines — see
/// `crate::progress`. A dry seal stages nothing to send and reports nothing.
fn report_packed(sink: PackSink<'_>) {
    if matches!(sink, PackSink::Upload { .. }) {
        crate::progress::report(crate::progress::Event::ObjectsPacked(1));
    }
}

/// Split `sizes` (each item's conservative uncompressed byte size, in
/// original order) into batch lengths such that each batch's summed
/// size stays at-or-under `payload_cap` bytes AND at-or-under
/// `max_entries` items — whichever limit a batch hits first ends it.
/// Mirrors [`should_seal`]'s "never true for an empty writer" escape
/// hatch: a single item over `payload_cap` on its own (an oversized
/// blob against a tiny test cap, say) still gets a one-item batch of
/// its own rather than looping forever or being silently dropped.
///
/// The byte cap exists so [`build_and_upload_packs`]'s per-batch
/// compression fan-out can't hold dramatically more raw content in
/// memory at once than the single pack buffer it's feeding already
/// does — without it, a count-only cap (`max_entries`, sized for
/// rayon dispatch overhead, not memory) could group hundreds of
/// near-`MAX_RAW_OBJECT_SIZE` (1 GiB) blobs into one batch and read +
/// compress all of them before any of `payload_cap`'s own bookkeeping
/// ever runs. The entry cap exists for the opposite corpus shape: many
/// small objects would otherwise form one enormous `payload_cap`-sized
/// (default 4 GiB) batch, both blowing past `pack_fanout_threshold`'s
/// bench-measured parallel-dispatch sizing and stalling progress
/// reporting for the whole batch's compression time.
fn size_capped_batch_lens(sizes: &[u64], payload_cap: u64, max_entries: usize) -> Vec<usize> {
    let mut lens = Vec::new();
    let mut running: u64 = 0;
    let mut count = 0usize;
    for &size in sizes {
        if count > 0 && (running.saturating_add(size) > payload_cap || count >= max_entries) {
            lens.push(count);
            running = 0;
            count = 0;
        }
        running += size;
        count += 1;
    }
    if count > 0 {
        lens.push(count);
    }
    lens
}

/// Entries-per-thread budget below which [`prepare_raw_batch`] /
/// [`prepare_delta_batch`] read + zstd-compress sequentially instead of
/// fanning out across rayon's thread pool, for a pool of a given size.
/// Same crossover shape as `add.rs`'s `HASH_FANOUT_FILES_PER_THREAD`
/// (PR #951) — rayon's pool-dispatch overhead loses to a plain loop
/// below a few entries per thread and wins clearly above it — measured
/// for the pack-build path specifically with `cargo bench -p
/// mkit-benches --bench pack_build_fanout`.
///
/// Also doubles as one of [`size_capped_batch_lens`]'s two caps: a
/// batch that isn't cut short by `payload_cap` first comes out exactly
/// this many entries long (until the final, possibly-shorter
/// remainder) — the same length `prepare_raw_batch`/
/// `prepare_delta_batch` compare against to decide sequential vs.
/// parallel, so a full-length batch always takes the parallel path.
const PACK_FANOUT_ENTRIES_PER_THREAD: usize = 8;

/// The per-batch entry count [`build_and_upload_packs`] fans out across
/// rayon at — see [`PACK_FANOUT_ENTRIES_PER_THREAD`].
fn pack_fanout_threshold() -> usize {
    crate::fanout::threshold(PACK_FANOUT_ENTRIES_PER_THREAD)
}

/// Read + compression-prepare one chunk of raw object hashes —
/// sequentially below [`pack_fanout_threshold`], via rayon's global
/// thread pool at or above it. Output mirrors `chunk`'s order 1:1
/// either way: `plan.raw` is already ordered (non-blobs before blobs,
/// each group in BLAKE3 order — see [`transfer::PackPlan`]'s doc
/// comment), and that order is the wire order the pack is built in, so
/// a parallel fan-out must not reshuffle it.
fn prepare_raw_batch(
    store: &ObjectStore,
    chunk: &[Hash],
) -> Result<Vec<PreparedRaw>, DispatchError> {
    crate::fanout::try_map_seq_or_par(chunk, pack_fanout_threshold(), |h| {
        Ok(PackWriter::prepare_raw(*h, store.read(h)?))
    })
}

/// The delta-entry counterpart of [`prepare_raw_batch`]. Delta streams
/// are already resident in `chunk` (no store read needed), so this
/// only fans out the zstd compression step. Takes `chunk` by value —
/// `build_and_upload_packs` drains it straight out of `plan.deltas` —
/// so `PackWriter::prepare_delta` can take each stream by move instead
/// of cloning it.
fn prepare_delta_batch(chunk: Vec<transfer::PlannedDelta>) -> Vec<PreparedDelta> {
    if chunk.len() < pack_fanout_threshold() {
        return chunk
            .into_iter()
            .map(|d| PackWriter::prepare_delta(d.base, d.stream))
            .collect();
    }
    chunk
        .into_par_iter()
        .map(|d| PackWriter::prepare_delta(d.base, d.stream))
        .collect()
}

/// Entries-per-thread budget below which [`encode_delta_candidates_batch`]
/// diffs candidates sequentially instead of fanning out across rayon's
/// thread pool, for a pool of a given size. Same crossover shape as
/// [`PACK_FANOUT_ENTRIES_PER_THREAD`] but tuned separately, and lower:
/// each entry here pays two disk reads (target + base) plus
/// `delta::encode`'s block-hash-table build and greedy scan over the
/// target's full length — a heavier per-item cost than the
/// zstd-compression fan-out `PACK_FANOUT_ENTRIES_PER_THREAD` guards, so
/// fewer entries per thread are enough to amortize dispatch.
///
/// `cargo bench -p mkit-benches --bench delta_plan_fanout` on a 4-core
/// host, isolating `delta::encode` over synthetic 64 KiB
/// (FastCDC-average-sized) near-duplicate chunks:
///
///   n=256: sequential 46.99ms -> rayon 14.79ms (3.18x)
///   n=128: sequential 22.85ms -> rayon  7.57ms (3.02x)
///   n=64:  sequential 11.65ms -> rayon  3.67ms (3.18x)
///   n=32:  sequential  5.52ms -> rayon  1.74ms (3.18x)
///   n=16:  sequential  3.28ms -> rayon  1.22ms (2.69x)
///   n=8:   sequential  1.38ms -> rayon  0.70ms (1.97x)
///   n=4:   sequential  0.68ms -> rayon  0.42ms (1.63x)
///   n=2:   sequential  0.34ms -> rayon  0.33ms (roughly flat)
///   n=1:   sequential  0.17ms -> rayon  0.17ms (roughly flat)
///
/// Each candidate already costs ~0.17ms, comfortably above rayon's
/// microsecond-scale dispatch overhead, so the crossover falls at the
/// smallest candidate count worth fanning out at all rather than at a
/// larger multiple of the pool size the other fan-outs need.
const DELTA_PLAN_FANOUT_ENTRIES_PER_THREAD: usize = 1;

/// The candidate count [`encode_delta_candidates_batch`] fans out across
/// rayon at — see [`DELTA_PLAN_FANOUT_ENTRIES_PER_THREAD`].
fn delta_plan_fanout_threshold() -> usize {
    crate::fanout::threshold(DELTA_PLAN_FANOUT_ENTRIES_PER_THREAD)
}

/// [`transfer::plan_pack_with`]'s `encode_deltas` callback: diffs every
/// delta candidate `plan_pack_with` found — sequentially below
/// [`delta_plan_fanout_threshold`], via rayon's global thread pool at or
/// above it. Each candidate's `delta::encode` is independent of every
/// other candidate's (same shape as `prepare_raw_batch`'s per-object
/// compression fan-out), so a push touching many changed blobs
/// parallelizes the diffing itself, not just the downstream zstd pass
/// `prepare_delta_batch` already fans out.
///
/// Processes at most [`DELTA_CANDIDATE_BATCH_CAP`] candidates at a time.
/// Repeated bases share a cache capped at [`DELTA_BASE_CACHE_BYTES`];
/// one-use bases and bases that do not fit are read by each encoder as
/// before. Metadata checks avoid reading nonfitting bases twice. The
/// cache is dropped before the next batch. Active encoders' target/base
/// buffers and the accumulated encoded output are additional allocations;
/// imported blobs can reach the store's 1 GiB object limit.
///
/// Order matches `candidates` 1:1, including across batch boundaries,
/// satisfying `plan_pack_with`'s contract.
fn encode_delta_candidates_batch(
    store: &ObjectStore,
    candidates: &[transfer::DeltaCandidate],
) -> Result<Vec<Option<transfer::PlannedDelta>>, StoreError> {
    let mut results = Vec::with_capacity(candidates.len());
    for batch in candidates.chunks(DELTA_CANDIDATE_BATCH_CAP) {
        let base_bytes = cache_delta_bases(store, batch, DELTA_BASE_CACHE_BYTES)?;
        results.extend(crate::fanout::try_map_seq_or_par(
            batch,
            delta_plan_fanout_threshold(),
            |&c| match base_bytes.get(&c.base) {
                Some(bytes) => transfer::encode_delta_candidate_with_base(store, c, bytes),
                None => transfer::encode_delta_candidate(store, c),
            },
        )?);
    }
    Ok(results)
}

/// Maximum number of candidates whose base objects are cached together.
const DELTA_CANDIDATE_BATCH_CAP: usize = 64;

/// Maximum retained payload bytes for reused bases, independent of blob sizes.
const DELTA_BASE_CACHE_BYTES: usize = 64 * 1024 * 1024;

fn cache_delta_bases(
    store: &ObjectStore,
    candidates: &[transfer::DeltaCandidate],
    byte_budget: usize,
) -> Result<std::collections::HashMap<Hash, Vec<u8>>, StoreError> {
    let mut counts = std::collections::BTreeMap::<Hash, usize>::new();
    for candidate in candidates {
        *counts.entry(candidate.base).or_default() += 1;
    }
    let mut cache = std::collections::HashMap::new();
    let mut remaining = byte_budget;
    for (base, count) in counts {
        if count < 2 || remaining == 0 {
            continue;
        }
        if store.object_metadata(&base)?.len() > remaining as u64 {
            continue;
        }
        let bytes = store.read(&base)?;
        // Recheck after reading: metadata is only an optimization, never
        // authority for the retained byte bound if a file changes meanwhile.
        if bytes.len() <= remaining {
            remaining -= bytes.len();
            cache.insert(base, bytes);
        }
    }
    Ok(cache)
}

/// Would pushing an entry of (conservative, uncompressed) size
/// `add_bound` into `w` exceed `payload_cap`? Never true for an empty
/// writer — a single entry over the cap (only reachable with a
/// test-injected tiny cap; production entries are bounded well under
/// [`pack::MAX_TOTAL_PAYLOAD`] by [`mkit_core::store::MAX_RAW_OBJECT_SIZE`])
/// lands alone in its own pack rather than looping forever.
fn should_seal(
    w: &PackWriter,
    add_bound: u64,
    payload_cap: u64,
    max_pack_bytes: Option<u64>,
) -> bool {
    let next_payload = w.total_payload().saturating_add(add_bound);
    w.entry_count() > 0
        && (next_payload > payload_cap
            || max_pack_bytes
                .is_some_and(|limit| serialized_bound(next_payload, w.entry_count() + 1) > limit))
}

/// Finish `w`, upload it, record its key, and replace `w` with a fresh
/// empty writer so the caller can keep pushing entries into the next
/// pack.
fn seal_pack(
    sink: PackSink<'_>,
    w: &mut PackWriter,
    pack_keys: &mut Vec<Hash>,
    ticketed_count: &mut usize,
    limits: UploadLimits,
) -> Result<(), DispatchError> {
    if crate::signal::is_shutdown() {
        return Err(DispatchError::Interrupted);
    }
    let sealed = std::mem::replace(w, PackWriter::new());
    let pack = sealed.finish()?;
    let is_ticketed = ticketed_pack(limits, pack.len() as u64);
    let budget = split::data_pack_budget(limits);
    if is_ticketed && *ticketed_count >= budget {
        return Err(DispatchError::PushTooLarge {
            packs: *ticketed_count + 1,
            limit: budget,
            commit: None,
            holds_remote_head: false,
        });
    }
    if limits
        .max_pack_bytes
        .is_some_and(|limit| pack.len() as u64 > limit)
    {
        return Err(DispatchError::Transport(TransportError::PayloadTooLarge(
            pack.len(),
        )));
    }
    let pack_key = pack::pack_key(&pack);
    if let PackSink::Upload { tx, head_ref } = sink {
        tx.upload_pack_via_ref(&pack, &PackKey::from_hash(pack_key), head_ref)
            .map_err(|error| repository_operation_error(tx, error))?;
        // Upload is complete — report the real byte count handed to the
        // transport, not an estimate.
        crate::progress::report(crate::progress::Event::PackUploaded(pack.len() as u64));
    }
    pack_keys.push(pack_key);
    if is_ticketed {
        *ticketed_count += 1;
    }
    Ok(())
}

/// [`pull_all_with`] with signature verification on — the CLI's default
/// (issue #692). Existing in-process callers (the integration-test suite)
/// that construct only validly-signed histories are unaffected.
pub fn pull_all(
    cwd: &Path,
    tx: &dyn Transport,
    remote: &str,
    target_branch: Option<&str>,
) -> Result<usize, DispatchError> {
    pull_all_with(cwd, tx, remote, target_branch, true)
}

/// Fetch remote refs, then fast-forward the current local branch from
/// `refs/remotes/default/<branch>`. Fresh repos with no local branch tip
/// initialise from the current branch's remote-tracking ref, or the first
/// advertised remote branch when the current default branch is absent.
///
/// `target_branch`, when `Some`, overrides which remote branch to land
/// on (used by `mkit clone -b <branch>`): the branch MUST exist among
/// the remote's advertised refs or the call fails with
/// [`DispatchError::RemoteBranchMissing`] rather than silently falling
/// back to another branch. `None` preserves the historical HEAD-driven
/// selection used by plain `pull`.
///
/// `require_signed` gates the post-fetch commit/remix/tag signature check
/// (issue #692) — `true` (the CLI's default, see [`pull_all`]) verifies
/// every newly-fetched object and fails closed; `false` is the explicit
/// `--no-verify-signatures` / `pull.require_signed = false` opt-out.
pub fn pull_all_with(
    cwd: &Path,
    tx: &dyn Transport,
    remote: &str,
    target_branch: Option<&str>,
    require_signed: bool,
) -> Result<usize, DispatchError> {
    let layout = mkit_core::layout::discover(cwd)?;
    let store = crate::commands::open_store_configured(&layout)?;
    // Fetch phase: `fetch_objects` takes the repo lock itself, narrowly and
    // per branch, around only the local unpack + remote-ref-publish window
    // (#642 — see `packmap::apply_fetched_chain`). No lock is held here
    // across the network transfer.
    let n = fetch_objects(&store, &layout, tx, remote, target_branch, require_signed)?;
    let remote_refs = crate::commands::list_remote_refs_parallel(&layout, remote)?
        .into_iter()
        .filter_map(|r| r.hash.map(|hash| (r.name, hash)))
        .collect::<Vec<_>>();
    if remote_refs.is_empty() {
        return Ok(n);
    }

    // Fast-forward phase (#642): branch ref + HEAD + worktree, narrowly
    // locked around just this window rather than bundled with the fetch
    // above. The objects `remote_tip` points at are already reachable via
    // the remote-tracking ref published during the fetch phase, so this
    // lock's job here is worktree-mutation exclusivity against concurrent
    // commands (e.g. a racing `commit`/`checkout`/`reset`), not GC
    // protection — that hazard was already closed before this lock was
    // taken.
    let _lock = mkit_core::repo_lock::acquire_default(
        layout.worktree_state_dir(),
        crate::commands::WORKTREE_LOCK,
    )?;
    crate::commands::warn_if_served(&layout);
    let original_head = refs::read_head(&layout).ok();
    let (branch, local_tip, remote_tip) = match &original_head {
        Some(Head::Branch(head_branch)) => {
            let want_branch = target_branch.unwrap_or(head_branch.as_str());
            let local_tip = refs::read_ref(&layout, want_branch)?;
            let selected = if local_tip.is_some() || target_branch.is_some() {
                // An explicit `-b <branch>` (or an already-committed local
                // branch of that name) must match exactly — no silent
                // fallback to a different branch.
                remote_refs
                    .iter()
                    .find(|(name, _)| name == want_branch)
                    .ok_or_else(|| DispatchError::RemoteBranchMissing(want_branch.to_owned()))?
            } else {
                remote_refs
                    .iter()
                    .find(|(name, _)| name == want_branch)
                    .unwrap_or(&remote_refs[0])
            };
            (selected.0.clone(), local_tip, selected.1)
        }
        Some(Head::Detached(_)) => return Err(DispatchError::DetachedHead),
        None => (remote_refs[0].0.clone(), None, remote_refs[0].1),
    };

    let ref_condition = if let Some(local_tip) = local_tip {
        if local_tip == remote_tip {
            return Ok(n);
        }
        if !is_ancestor(&store, local_tip, remote_tip)? {
            return Err(DispatchError::NonFastForwardPull { branch });
        }
        refs::RefWriteCondition::Match(local_tip)
    } else {
        refs::RefWriteCondition::Missing
    };

    let tree = load_tree_hash(&store, remote_tip)?;
    crate::commands::ensure_restore_safe(&layout, &store, tree)
        .map_err(DispatchError::RestoreSafety)?;
    crate::commands::write_ref_recording_history(&layout, &branch, ref_condition, &remote_tip)?;
    if let Err(e) = refs::write_head_branch(&layout, &branch) {
        rollback_pull_ref(&layout, &branch, local_tip, remote_tip)?;
        return Err(e.into());
    }
    if let Err(e) = crate::commands::restore_worktree_and_index(&layout, &store, tree) {
        if let Err(rollback) =
            rollback_pull_ref_and_head(&layout, &branch, local_tip, remote_tip, original_head)
        {
            return Err(DispatchError::RestoreSafety(format!(
                "{e}; additionally failed to roll back ref: {rollback}"
            )));
        }
        return Err(DispatchError::RestoreSafety(e));
    }
    Ok(n)
}

fn rollback_pull_ref_and_head(
    layout: &RepoLayout,
    branch: &str,
    local_tip: Option<Hash>,
    remote_tip: Hash,
    original_head: Option<Head>,
) -> Result<(), String> {
    rollback_pull_ref(layout, branch, local_tip, remote_tip).map_err(|e| e.to_string())?;
    match original_head {
        Some(Head::Branch(name)) => refs::write_head_branch(layout, &name),
        Some(Head::Detached(hash)) => refs::write_head_detached(layout, &hash),
        None => Ok(()),
    }
    .map_err(|e| e.to_string())
}

fn rollback_pull_ref(
    layout: &RepoLayout,
    branch: &str,
    local_tip: Option<Hash>,
    remote_tip: Hash,
) -> Result<(), refs::RefError> {
    if let Some(local_tip) = local_tip {
        crate::commands::write_ref_recording_history(
            layout,
            branch,
            refs::RefWriteCondition::Match(remote_tip),
            &local_tip,
        )
    } else if refs::read_ref(layout, branch)? == Some(remote_tip) {
        refs::delete_ref(layout, branch)
    } else {
        Ok(())
    }
}

/// [`fetch_all_with`] with signature verification on — the CLI's default
/// (issue #692). Existing in-process callers (the integration-test suite)
/// that construct only validly-signed histories are unaffected.
pub fn fetch_all(cwd: &Path, tx: &dyn Transport, remote: &str) -> Result<usize, DispatchError> {
    fetch_all_with(cwd, tx, remote, true)
}

/// `fetch` — `pull_all` without the HEAD update. Downloads every object
/// reachable from each remote ref (via [`Transport::download_pack`] on
/// the object's own digest) and writes the ref into
/// `refs/remotes/default/<branch>`.
///
/// See [`pull_all_with`] for the `require_signed` contract (issue #692).
pub fn fetch_all_with(
    cwd: &Path,
    tx: &dyn Transport,
    remote: &str,
    require_signed: bool,
) -> Result<usize, DispatchError> {
    let layout = mkit_core::layout::discover(cwd)?;
    // No outer lock here (#642): `fetch_objects` takes the repo lock
    // itself, narrowly and per branch, around only the local unpack +
    // remote-ref-publish window for that branch — never across the
    // network transfer. See `packmap::resolve_and_download_chain` /
    // `apply_fetched_chain` and `fetch_objects_inner` below.
    let store = crate::commands::open_store_configured(&layout)?;
    fetch_objects(&store, &layout, tx, remote, None, require_signed)
}

/// Reconstruct every remote `refs/heads/*` from its packmap chain and
/// publish the remote-tracking refs. Each branch's local object writes and
/// its ref publish happen under a repo lock acquired fresh for that branch
/// (#642) — see [`fetch_objects_inner`] — so the caller does NOT need to
/// hold the repo lock around this call.
///
/// mkit speaks a single, packmap-only transfer dialect (the legacy
/// per-object download path was removed): for every advertised branch the
/// flow is exactly
///
/// 1. read the branch's packmap ref (`refs/mkit/packmap/<branch>`),
/// 2. walk its chain oldest-first and download any pack the local
///    applied-pack record doesn't already have
///    ([`packmap::resolve_and_download_chain`]) — no repo lock held,
/// 3. acquire the repo lock, unpack the downloaded packs
///    ([`packmap::apply_fetched_chain`]), then
/// 4. assert the tip's closure is fully present
///    ([`verify_closure_present`]) — a pure integrity check that downloads
///    nothing (still under the lock from step 3), then publish the
///    branch's remote-tracking ref and release the lock (#642).
///
/// Steps 3 and 4 are both performed *inside* [`packmap::apply_fetched_chain`]
/// (rather than sequenced here) so a closure-completeness failure counts
/// toward that function's applied-pack self-heal retry (#409): if the
/// local record wrongly claims every pack in the chain is already applied
/// (e.g. `.mkit/objects` was wiped out-of-band while `applied-packs/`
/// survived), the very first symptom is exactly this closure check
/// failing, not a download/unpack error — the retry has to cover both.
///
/// Both ends fail loudly: an absent packmap beside a present head is
/// [`DispatchError::PackmapMissing`] (a head absent too is a stale listing,
/// STC §7.9: the branch is skipped and its tracking ref left untouched)
/// and a present-but-incomplete packmap (even after the self-heal retry) is
/// [`DispatchError::RemoteMissingObject`]. We never publish a
/// remote-tracking ref to a closure we couldn't fully materialise locally.
///
/// # Concurrent re-baseline (mkit #521)
///
/// We list branch tips (`list_refs`) BEFORE reading each branch's packmap,
/// so a concurrent push that re-baselines (resets the packmap to a fresh
/// single node, #406) between those two reads can leave us verifying an
/// OLD tip `h` against a packmap whose closure only covers the NEW tip —
/// surfacing as [`DispatchError::RemoteMissingObject`] (the reset chain
/// isn't a superset of the prior one, and the applied-pack self-heal can't
/// rescue a genuinely stale tip). This is transient — no bad ref was
/// published — so on that specific error we re-read the branch's CURRENT tip
/// and packmap and retry the chain once with the fresh pair, publishing the
/// fresh tip. A second failure (or a vanished branch) propagates unchanged.
///
/// # Applied-packs record: load once, persist once (mkit #546)
///
/// The applied-pack record (`<common dir>/applied-packs/<remote>`, #409) is
/// keyed by remote, not by branch, so this function — not
/// [`packmap::resolve_and_download_chain`] / [`packmap::apply_fetched_chain`]
/// — owns its lifecycle for the WHOLE fetch: loaded once before the branch
/// loop, persisted once after, however many branches are fetched. Those two
/// functions only mutate the record in memory (inserting applied digests,
/// or clearing on self-heal); neither touches disk. The final persist is
/// unconditional and best-effort — it runs on every outcome so a fetch that
/// applied packs before failing never loses that progress, and the record
/// is a pure performance cache whose own I/O must never fail a fetch whose
/// objects durably landed.
///
/// # Repo-lock scope (mkit #642)
///
/// Each branch acquires the repo lock fresh, right before
/// [`packmap::apply_fetched_chain`] unpacks that branch's downloaded packs,
/// and releases it right after the branch's remote-tracking ref is
/// published — see [`fetch_objects_inner`]. Nothing here holds the lock
/// during [`packmap::resolve_and_download_chain`]'s network I/O, and the
/// lock is released between branches, so only the local write + ref-publish
/// window for one branch at a time is ever locked. This still closes the
/// #267 GC-prune race: `gc` takes the very same lock before computing its
/// live set, so it can never observe a branch's objects on disk without
/// that branch's ref already published.
///
/// `target_branch` (a `clone -b` / `pull` target) is fetched even when an
/// eventual `ListRefs` (D34, STC §7.9) has not caught up with it: the branch
/// is strongly read by name, fetched if present, and
/// [`DispatchError::RemoteBranchMissing`] if absent. Never an exit-0 empty
/// clone for `-b`.
fn fetch_objects(
    store: &ObjectStore,
    layout: &RepoLayout,
    tx: &dyn Transport,
    remote: &str,
    target_branch: Option<&str>,
    require_signed: bool,
) -> Result<usize, DispatchError> {
    let mut applied = AppliedPacks::load_or_empty(layout, remote);
    let result = fetch_objects_inner(
        store,
        layout,
        tx,
        remote,
        target_branch,
        &mut applied,
        require_signed,
    );
    persist_record(&mut applied, remote);
    result
}

/// The branch loop proper — see [`fetch_objects`] for the load-once /
/// persist-once applied-packs contract this is called under.
fn fetch_objects_inner(
    store: &ObjectStore,
    layout: &RepoLayout,
    tx: &dyn Transport,
    remote: &str,
    target_branch: Option<&str>,
    applied: &mut AppliedPacks,
    require_signed: bool,
) -> Result<usize, DispatchError> {
    let mut remote_refs = tx
        .list_refs("refs/heads/")
        .map_err(|error| repository_operation_error(tx, error))?;
    if let Some(branch) = target_branch
        && !remote_refs.iter().any(|r| r.name == branch)
    {
        // The listing may be stale (eventual under D34): a strong read by
        // name decides whether the branch exists. A transport error here
        // propagates; only `Ok(None)` is the "no such branch" verdict.
        match tx.read_ref(&format!("refs/heads/{branch}"))? {
            Some(hash) => remote_refs.push(refs::Ref {
                name: branch.to_owned(),
                hash: Some(hash),
            }),
            None => return Err(DispatchError::RemoteBranchMissing(branch.to_owned())),
        }
    }
    let mut n = 0;
    // Batch every fetched branch's remote-tracking-ref write (#645): see
    // `push_all_with` for the same pattern and its rationale. `tracking.write`
    // still runs inside this branch's `_lock` scope below, so ref visibility
    // timing is unchanged from the per-branch-publish-then-unlock model
    // (#642) — only the parent-directory fsync is deferred to one pass after
    // the loop. GC's #267 protection is unaffected: it depends on the
    // repo lock covering the object-write-to-ref-publish window (still true
    // per branch below), not on when the ref directory itself is fsynced.
    let mut tracking = refs::RemoteRefBatch::new(layout, remote)?;
    let result: Result<(), DispatchError> = (|| {
        for r in remote_refs {
            if crate::signal::is_shutdown() {
                return Err(DispatchError::Interrupted);
            }
            let Some(h) = r.hash else { continue };
            // A branch tip without a packmap is a corrupt/incomplete remote, not
            // a format we degrade to: the push path ALWAYS advertises a packmap
            // before moving the branch ref. A real transport error (network blip,
            // auth) propagates unchanged — only `Ok(None)` is the explicit
            // "no packmap" verdict.
            //
            // The one exception is a stale listing (STC §7.9): `ListRefs` is
            // eventual, so under D34 a branch deleted since the listing can
            // still be named. Re-read the head strongly: absent too means the
            // listing is stale, so skip the branch and write no tracking ref;
            // present means corruption (a head and its packmap share a shard),
            // which stays `PackmapMissing`. A transport error on the re-read
            // propagates and never becomes a skip.
            let Some(chain_head) = tx.read_ref(&packmap_ref(&r.name))? else {
                if tx.read_ref(&format!("refs/heads/{}", r.name))?.is_none() {
                    if crate::progress::should_report(false) {
                        eprintln!(
                            "fetch: skipping branch `{}`: listed by an eventual ListRefs but \
                             no longer present (deleted since the listing)",
                            r.name
                        );
                    }
                    continue;
                }
                return Err(DispatchError::PackmapMissing(r.name.clone()));
            };

            // Phase 1 (#642): resolve this branch's chain and download its
            // packs — pure network I/O, no repo lock held (see
            // `packmap::resolve_and_download_chain`).
            let fetched = resolve_and_download_chain(tx, &r.name, chain_head, applied)?;

            // Phase 2 (#642): unpack + verify + publish this branch's ref,
            // under a repo lock acquired fresh for this branch and released
            // once this loop iteration ends — never held across another
            // branch's download. See `packmap::apply_fetched_chain`'s doc
            // comment for the safety contract (closing the #267 GC-prune
            // race) this depends on.
            let lock = mkit_core::repo_lock::acquire_default(
                layout.worktree_state_dir(),
                crate::commands::WORKTREE_LOCK,
            )?;
            crate::commands::warn_if_served(layout);
            // The tip we publish: normally the listed `h`, but if the chain fails
            // because a concurrent re-baseline moved the branch under us, the
            // freshly re-read tip (see this fn's doc comment). The match also
            // carries the lock guard out, so whichever branch runs, the
            // correct (possibly re-acquired) guard is what's held at
            // `tracking.write` below — see the retry branch's comment for why
            // there are two guards, not one held across the whole match.
            let (published_tip, _lock) = match apply_fetched_chain(
                store,
                tx,
                remote,
                &r.name,
                fetched,
                h,
                applied,
                require_signed,
            ) {
                Ok(()) => (h, lock),
                Err(e @ DispatchError::RemoteMissingObject(_)) => {
                    // Re-read the branch's CURRENT tip + packmap. If either
                    // is gone (branch deleted mid-fetch) the original error
                    // stands. Otherwise retry the chain ONCE with the fresh
                    // pair; a second failure propagates via `?`.
                    let (Some(fresh_h), Some(fresh_head)) = (
                        tx.read_ref(&format!("refs/heads/{}", r.name))?,
                        tx.read_ref(&packmap_ref(&r.name))?,
                    ) else {
                        return Err(e);
                    };
                    // Release the lock for the retry's network
                    // re-download too — mirrors phase 1's unlocked
                    // download exactly, rather than the previously
                    // "accepted trade" of holding the lock across a
                    // second network round-trip on this rare
                    // race-recovery path. Re-acquire before the
                    // retry's local unpack + publish, which still
                    // needs the same #267 protection phase 2 always
                    // has.
                    drop(lock);
                    let fresh_fetched =
                        resolve_and_download_chain(tx, &r.name, fresh_head, applied)?;
                    let lock = mkit_core::repo_lock::acquire_default(
                        layout.worktree_state_dir(),
                        crate::commands::WORKTREE_LOCK,
                    )?;
                    crate::commands::warn_if_served(layout);
                    apply_fetched_chain(
                        store,
                        tx,
                        remote,
                        &r.name,
                        fresh_fetched,
                        fresh_h,
                        applied,
                        require_signed,
                    )?;
                    (fresh_h, lock)
                }
                Err(e) => return Err(e),
            };
            // Still inside `_lock`'s scope (the original guard, or the
            // retry's re-acquired one — see above).
            tracking.write(&r.name, &published_tip)?;
            n += 1;
        }
        Ok(())
    })();
    // Commit whatever tracking-ref writes succeeded regardless of how the
    // loop above ended — a mid-loop failure still durably publishes the
    // prefix that already verified successfully, matching the old
    // per-branch loop's per-write durability.
    tracking.commit()?;
    result?;
    Ok(n)
}

/// Best-effort persist of `applied` (a write failure is logged and
/// swallowed), called exactly once per fetch — see [`fetch_objects`] for
/// the load-once / persist-once contract.
fn persist_record(applied: &mut AppliedPacks, remote: &str) {
    if let Err(e) = applied.persist() {
        eprintln!(
            "warning: could not persist applied-packs record for remote '{remote}' ({e}); it will be rebuilt on the next fetch"
        );
    }
}

/// Assert that every object reachable from `tip` is already present in the
/// local store after the packmap chain has been unpacked. This is a pure
/// integrity check — it walks the closure via
/// [`mkit_core::ops::reachable_closure_checked`] (which reads each object and
/// re-verifies its digest) and performs NO network access. A reachable object
/// that the chain failed to deliver surfaces as [`StoreError::ObjectNotFound`],
/// which we re-tag as [`DispatchError::RemoteMissingObject`] so the fetch
/// aborts loudly rather than publishing a ref to a closure we can't
/// reconstruct.
///
/// When packs were skipped (the applied-pack fast path, #409) this walk is
/// the *sole* guarantee the store is complete, so it must not silently pass
/// on an unverified frontier: a closure exceeding the
/// [`mkit_core::ops::graph::MAX_REACHABLE`] cap leaves objects past the cap
/// unchecked, which over a partially-wiped store could hide missing objects.
/// We therefore surface truncation as a hard [`DispatchError::ClosureTooLarge`]
/// rather than dropping the flag. `ClosureTooLarge` is deliberately distinct
/// from `RemoteMissingObject` so it does NOT feed the self-heal retry — a
/// too-large history is not evidence of local staleness.
///
/// Called from [`packmap::fetch_pack_chain`] (not sequenced after it) so its
/// `RemoteMissingObject` result participates in that function's applied-pack
/// self-heal retry — see [`fetch_objects`]'s doc comment.
pub(crate) fn verify_closure_present(store: &ObjectStore, tip: &Hash) -> Result<(), DispatchError> {
    match mkit_core::ops::reachable_closure_checked(store, std::iter::once(tip)) {
        Ok((_, false)) => Ok(()),
        Ok((_, true)) => Err(DispatchError::ClosureTooLarge(
            mkit_core::ops::graph::MAX_REACHABLE,
        )),
        Err(StoreError::ObjectNotFound(hex)) => Err(DispatchError::RemoteMissingObject(hex)),
        Err(e) => Err(e.into()),
    }
}

fn load_tree_hash(store: &ObjectStore, commit_hash: Hash) -> Result<Hash, DispatchError> {
    match store.read_object(&commit_hash)? {
        Object::Commit(c) => Ok(c.tree_hash),
        Object::Remix(r) => Ok(r.tree_hash),
        _ => Err(DispatchError::NotCommit),
    }
}

#[cfg(test)]
mod tests {
    use super::ssh_options_from_config;
    use crate::config::Config;
    use mkit_core::layout::RepoLayout;
    use mkit_core::pack::{PreparedDelta, PreparedRaw};
    use mkit_core::store::ObjectStore;
    use mkit_core::transfer;

    // =================================================================
    // `size_capped_batch_lens` — the batch-boundary arithmetic
    // `build_and_upload_packs` uses to bound each compression-fan-out
    // batch by both bytes (`payload_cap`) and entry count
    // (`pack_fanout_threshold`). Pure and deterministic (no rayon pool
    // size involved), so these pin the boundary conditions directly
    // rather than relying on an end-to-end push to happen to exercise
    // them on whatever hardware runs the test.
    // =================================================================

    #[test]
    fn envelope_destination_trust_precedes_missing_key_resolution() {
        let directory = tempfile::tempdir().unwrap();
        let layout = RepoLayout::single(directory.path());
        for endpoint in ["mkit+https://untrusted.example", "mkit+http://127.0.0.1:1"] {
            let mut cfg = Config {
                transport_auth: "envelope".into(),
                signer: "legacy".into(),
                signing_key: ".mkit/keys/missing-signing-key".into(),
                ..Config::default()
            };
            let error = super::open_with_config(endpoint, &cfg, &layout)
                .err()
                .expect("untrusted signing must fail");
            assert!(
                matches!(error, super::DispatchError::UntrustedRemote(_)),
                "destination rejection must precede key access: {error}"
            );
            cfg.trusted_remote_endpoint = endpoint.into();
            let error = super::open_with_config(endpoint, &cfg, &layout)
                .err()
                .expect("trusted endpoint now resolves the absent key");
            assert!(
                error.to_string().contains("requires a signing key"),
                "trusted destination should reach key resolution: {error}"
            );
        }
    }

    #[test]
    fn configured_connect_open_reports_malformed_repository_identity() {
        let directory = tempfile::tempdir().unwrap();
        let layout = RepoLayout::single(directory.path());
        let url = "mkit+https://host/Uppercase";
        let error = super::open_with_config(url, &Config::default(), &layout)
            .err()
            .expect("uppercase repository identity must be rejected");
        assert!(
            matches!(error, super::DispatchError::MalformedUrl(_)),
            "expected malformed URL, got {error:?}"
        );
        let message = error.to_string();
        assert!(message.contains("Uppercase"), "{message}");
        assert!(!message.contains("invalid response"), "{message}");
    }

    #[test]
    fn batch_lens_empty_input_is_no_batches() {
        assert_eq!(
            super::size_capped_batch_lens(&[], 1000, 10),
            Vec::<usize>::new()
        );
    }

    #[test]
    fn batch_lens_fits_one_batch_under_both_caps() {
        assert_eq!(
            super::size_capped_batch_lens(&[10, 10, 10], 1000, 10),
            vec![3]
        );
    }

    #[test]
    fn batch_lens_splits_on_byte_cap() {
        // Each item is 40 bytes; a 100-byte cap fits 2 per batch (80 <=
        // 100), not 3 (120 > 100) — entry cap (10) never binds here.
        assert_eq!(
            super::size_capped_batch_lens(&[40, 40, 40, 40, 40], 100, 10),
            vec![2, 2, 1]
        );
    }

    #[test]
    fn batch_lens_splits_on_entry_cap() {
        // Every item is tiny (1 byte), so the byte cap (10_000) never
        // binds — only the entry cap (2) does.
        assert_eq!(
            super::size_capped_batch_lens(&[1, 1, 1, 1, 1], 10_000, 2),
            vec![2, 2, 1]
        );
    }

    #[test]
    fn batch_lens_single_oversized_item_gets_its_own_batch() {
        // A lone item over `payload_cap` must not loop forever or be
        // dropped — mirrors `should_seal`'s "never true for an empty
        // writer" escape hatch.
        assert_eq!(super::size_capped_batch_lens(&[500], 100, 10), vec![1]);
        assert_eq!(
            super::size_capped_batch_lens(&[500, 10, 10], 100, 10),
            vec![1, 2]
        );
    }

    // =================================================================
    // `prepare_raw_batch` / `prepare_delta_batch` — order preservation
    // and error propagation across BOTH the sequential and the rayon
    // branch. `pack_fanout_threshold()` is read directly (rather than
    // hardcoding a count) so a chunk built at exactly that length
    // reliably lands in the parallel branch regardless of how many
    // cores the test happens to run on.
    // =================================================================

    fn store() -> (tempfile::TempDir, ObjectStore) {
        let dir = tempfile::tempdir().expect("tempdir");
        let layout = RepoLayout::single(dir.path());
        let store = ObjectStore::init(&layout).expect("init store");
        (dir, store)
    }

    #[test]
    fn prepare_raw_batch_preserves_order_sequential_and_parallel() {
        let (_dir, store) = store();
        for &n in &[1usize, super::pack_fanout_threshold()] {
            let hashes: Vec<mkit_core::hash::Hash> = (0..n)
                .map(|i| store.write(format!("raw entry #{i}").as_bytes()).unwrap())
                .collect();
            let prepared = super::prepare_raw_batch(&store, &hashes).expect("prepare raw batch");
            assert_eq!(
                prepared.iter().map(PreparedRaw::hash).collect::<Vec<_>>(),
                hashes,
                "prepare_raw_batch must preserve input order at n={n}"
            );
        }
    }

    #[test]
    fn prepare_raw_batch_propagates_a_missing_object_error_sequential_and_parallel() {
        let (_dir, store) = store();
        for &n in &[1usize, super::pack_fanout_threshold()] {
            let mut hashes: Vec<mkit_core::hash::Hash> = (0..n.saturating_sub(1))
                .map(|i| store.write(format!("raw entry #{i}").as_bytes()).unwrap())
                .collect();
            // A hash of content never written to the store — `store.read`
            // must fail with `ObjectNotFound`, and that failure must
            // surface through `prepare_raw_batch` rather than being
            // silently dropped or panicking on either branch.
            hashes.push(mkit_core::hash::hash(b"never written"));
            let err = super::prepare_raw_batch(&store, &hashes).expect_err(
                "a missing object must fail prepare_raw_batch instead of being dropped",
            );
            assert!(matches!(
                err,
                super::DispatchError::Store(mkit_core::store::StoreError::ObjectNotFound(_))
            ));
        }
    }

    #[test]
    fn prepare_delta_batch_preserves_order_sequential_and_parallel() {
        use mkit_core::transfer::PlannedDelta;
        for &n in &[1usize, super::pack_fanout_threshold()] {
            let deltas: Vec<PlannedDelta> = (0..n)
                .map(|i| PlannedDelta {
                    target: mkit_core::hash::hash(format!("target #{i}").as_bytes()),
                    base: mkit_core::hash::hash(format!("base #{i}").as_bytes()),
                    stream: format!("delta stream #{i}").into_bytes(),
                })
                .collect();
            let expected_bases: Vec<_> = deltas.iter().map(|d| d.base).collect();
            let prepared = super::prepare_delta_batch(deltas);
            assert_eq!(
                prepared.iter().map(PreparedDelta::base).collect::<Vec<_>>(),
                expected_bases,
                "prepare_delta_batch must preserve input order at n={n}"
            );
        }
    }

    #[test]
    fn encode_delta_candidates_batch_preserves_order_sequential_and_parallel() {
        let (_dir, store) = store();
        for &n in &[
            1usize,
            super::delta_plan_fanout_threshold(),
            super::DELTA_CANDIDATE_BATCH_CAP * 2 + 1,
        ] {
            // Each candidate's base/target pair is a small, localized edit
            // of distinct-per-index content — a real delta candidate (the
            // encoded stream is smaller than the raw target), and distinct
            // across candidates so no candidate's result could accidentally
            // match another's and mask an ordering bug.
            let candidates: Vec<transfer::DeltaCandidate> = (0..n)
                .map(|i| {
                    let mut base_bytes =
                        format!("encode-delta-candidates-batch fixture #{i}\n").into_bytes();
                    base_bytes.extend_from_slice(
                        b"the quick brown fox jumps over the lazy dog\n"
                            .repeat(4)
                            .as_slice(),
                    );
                    let base = store.write(&base_bytes).unwrap();
                    let mut target_bytes = base_bytes.clone();
                    target_bytes.extend_from_slice(b"-- edited --\n");
                    let target = store.write(&target_bytes).unwrap();
                    transfer::DeltaCandidate { target, base }
                })
                .collect();

            // Ground truth: encode each candidate one at a time, in order.
            let expected: Vec<Option<Vec<u8>>> = candidates
                .iter()
                .map(|&c| {
                    transfer::encode_delta_candidate(&store, c)
                        .expect("encode delta candidate")
                        .map(|p| p.stream)
                })
                .collect();
            assert!(
                expected.iter().all(Option::is_some),
                "every fixture candidate must actually delta-encode smaller than raw"
            );

            let actual: Vec<Option<Vec<u8>>> =
                super::encode_delta_candidates_batch(&store, &candidates)
                    .expect("encode delta candidates batch")
                    .into_iter()
                    .map(|r| r.map(|p| p.stream))
                    .collect();
            assert_eq!(
                actual, expected,
                "encode_delta_candidates_batch must preserve input order at n={n}"
            );
        }
    }

    #[test]
    fn encode_delta_candidates_batch_propagates_a_missing_object_error_sequential_and_parallel() {
        let (_dir, store) = store();
        for &n in &[
            1usize,
            super::delta_plan_fanout_threshold(),
            super::DELTA_CANDIDATE_BATCH_CAP * 2 + 1,
        ] {
            let mut candidates: Vec<transfer::DeltaCandidate> = (0..n.saturating_sub(1))
                .map(|i| {
                    let base = store
                        .write(format!("present base #{i}").as_bytes())
                        .unwrap();
                    let target = store
                        .write(format!("present target #{i}").as_bytes())
                        .unwrap();
                    transfer::DeltaCandidate { target, base }
                })
                .collect();
            // A target hash never written to the store — `store.read` must
            // fail with `ObjectNotFound`, surfaced through
            // `encode_delta_candidates_batch` rather than dropped or
            // panicking, on either branch.
            candidates.push(transfer::DeltaCandidate {
                target: mkit_core::hash::hash(b"never written target"),
                base: mkit_core::hash::hash(b"never written base"),
            });
            let err = super::encode_delta_candidates_batch(&store, &candidates).expect_err(
                "a missing object must fail encode_delta_candidates_batch instead of being dropped",
            );
            assert!(matches!(
                err,
                mkit_core::store::StoreError::ObjectNotFound(_)
            ));
        }
    }

    #[test]
    fn delta_base_cache_respects_bytes_and_skips_single_use_bases() {
        let (_dir, store) = store();
        let mut candidates = Vec::new();
        for (value, size, uses) in [
            (b'a', 128usize, 2usize),
            (b'b', 128, 2),
            (b'c', 1024, 2),
            (b'd', 64, 1),
        ] {
            let base_bytes = vec![value; size];
            let base = store.write(&base_bytes).unwrap();
            let mut target_bytes = base_bytes;
            target_bytes.extend_from_slice(b"edited");
            let target = store.write(&target_bytes).unwrap();
            candidates.extend(std::iter::repeat_n(
                transfer::DeltaCandidate { target, base },
                uses,
            ));
        }
        // Only one of the two repeated 128-byte bases fits. The 1 KiB
        // base exceeds the budget and the 64-byte base is used only once.
        let cache = super::cache_delta_bases(&store, &candidates, 192).unwrap();
        assert_eq!(cache.len(), 1);
        assert_eq!(cache.values().map(Vec::len).sum::<usize>(), 128);
    }

    #[test]
    fn delta_base_cache_does_not_read_past_a_failed_batch() {
        let (_dir, store) = store();
        let base = store.write(b"present base").unwrap();
        let missing_target = mkit_core::hash::hash(b"missing first-batch target");
        let mut candidates = vec![
            transfer::DeltaCandidate {
                target: missing_target,
                base,
            };
            super::DELTA_CANDIDATE_BATCH_CAP
        ];
        // If all bases are prefetched for the whole push, this missing base
        // fails first. Bounded processing must encounter the first batch's
        // missing target before it even tries to read the later base.
        candidates.extend(
            [transfer::DeltaCandidate {
                target: missing_target,
                base: mkit_core::hash::hash(b"missing later-batch base"),
            }; 2],
        );
        let err = super::encode_delta_candidates_batch(&store, &candidates).unwrap_err();
        assert!(matches!(
            err,
            mkit_core::store::StoreError::ObjectNotFound(h) if h == mkit_core::hash::to_hex(&missing_target)
        ));
    }

    /// The three `ssh.*` trust-pinning keys, when set in `Config`, must
    /// map 1:1 into the `SshOptions` carried to the spawned `ssh(1)`
    /// child. This is the producer half of issue #389: without it the
    /// keys are parsed but never reach the subprocess.
    #[test]
    fn populated_config_maps_to_ssh_options() {
        let cfg = Config {
            ssh_strict_host_key_checking: "yes".to_string(),
            ssh_user_known_hosts_file: "/path/to/project.known_hosts".to_string(),
            ssh_identity_file: "/path/to/id_ed25519".to_string(),
            ..Config::default()
        };
        let opts = ssh_options_from_config(&cfg);
        assert_eq!(opts.strict_host_key_checking, "yes");
        assert_eq!(opts.user_known_hosts_file, "/path/to/project.known_hosts");
        assert_eq!(opts.identity_file, "/path/to/id_ed25519");
    }

    /// Empty `ssh.*` fields must map to empty `SshOptions` fields so
    /// `build_ssh_command` emits NO `-o`/`-i` flags and the user's
    /// `ssh(1)` defaults are inherited unchanged.
    #[test]
    fn empty_config_maps_to_empty_ssh_options() {
        let opts = ssh_options_from_config(&Config::default());
        assert!(opts.strict_host_key_checking.is_empty());
        assert!(opts.user_known_hosts_file.is_empty());
        assert!(opts.identity_file.is_empty());
    }
}