mkit-core 0.5.0

//! Worktree → tree-object builder.
//!
//! Walks a directory, applies `.mkitignore`, hashes each file as a
//! [`Blob`](crate::object::Blob), recurses on subdirectories, validates
//! symlink targets against path-traversal, and writes a single root
//! [`Tree`] into the supplied [`ObjectStore`].
//!
//! Notes:
//!
//! - Files at or below [`CHUNK_THRESHOLD`] are stored as a single
//!   [`Blob`](crate::object::Blob). Files above the threshold are
//!   chunked with [`crate::chunker::FastCdc::v1`]; each chunk is
//!   stored as a `Blob` and the file is represented by a
//!   [`ChunkedBlob`] manifest whose hash
//!   is what lands in the parent tree.
//! - We never follow symlinks while walking. Linux/macOS `read_link`
//!   reports the target verbatim and we hash it as a blob.

use std::fs;
use std::io::{self, Read};
use std::path::{Path, PathBuf};

use crate::chunker::{ChunkIterator, ChunkReader, FastCdc};
use crate::hash::Hash;
use crate::ignore::{self, IgnoreList};
use crate::index::{self, Index};
use crate::object::{ChunkedBlob, EntryMode, Object, Tree, TreeEntry};
use crate::serialize;
use crate::store::{ObjectSink, ObjectStore};

/// Files larger than this go through the chunker (1 MiB).
pub const CHUNK_THRESHOLD: u64 = 1024 * 1024;

/// Hard cap on a single file (1 GiB).
pub const MAX_FILE_BYTES: u64 = 1024 * 1024 * 1024;

/// Errors returned by this module.
#[derive(Debug, thiserror::Error)]
pub enum WorktreeError {
    /// `read_link` returned a target that fails [`validate_symlink_target`].
    #[error("symlink target '{0}' is invalid (absolute or contains '..')")]
    InvalidSymlinkTarget(String),
    /// File exceeded [`MAX_FILE_BYTES`].
    #[error("file '{0}' exceeds the {MAX_FILE_BYTES} byte limit")]
    FileTooLarge(PathBuf),
    /// Path component had non-UTF-8 bytes; tree entry names must be UTF-8.
    #[error("path component is not valid UTF-8")]
    InvalidUtf8,
    /// Underlying I/O failure.
    #[error(transparent)]
    Io(#[from] io::Error),
    /// Error encoding/serialising an object on its way into the store.
    #[error(transparent)]
    Object(#[from] crate::object::MkitError),
    /// Error returned by the object store.
    #[error(transparent)]
    Store(#[from] crate::store::StoreError),
    /// A `hash_chunks` callback passed to
    /// [`store_large_file_streaming_with`]/[`hash_file_with_metadata_with`]
    /// returned a different number of hashes than the batch it was given —
    /// a contract violation by the caller (see those functions' docs),
    /// never a normal runtime condition. Reported as a typed error rather
    /// than a panic since the caller here is `mkit-cli`, a
    /// long(er)-running process where a hard panic is a worse failure mode
    /// than a clean, exit-coded error.
    #[error("hash_chunks callback returned {actual} hashes for a {expected}-chunk batch")]
    ChunkBatchLengthMismatch { expected: usize, actual: usize },
}

/// Result alias used throughout this module.
pub type WorktreeResult<T> = Result<T, WorktreeError>;

mod blob;
pub use blob::{LoadedBlob, content_eq, content_eq_bytes, content_fingerprint, read_blob};

/// Validate a symlink target: must be relative and contain no `..`
/// segments.
#[must_use]
pub fn validate_symlink_target(target: &str) -> bool {
    if target.is_empty() {
        return false;
    }
    if target.starts_with('/') {
        return false;
    }
    for part in target.split('/') {
        if part == ".." {
            return false;
        }
    }
    true
}

/// A hash-time stat observation: while building a tree we re-hashed
/// `path` (its cache was absent or racy-smudged) and the result equals
/// the staging index's hash — so the stat captured from the OPENED file
/// descriptor *before* its content was read proves the entry clean.
/// `status` consumes these to heal the stat cache without ever pairing
/// a post-verification stat with a pre-verification hash (the unsound
/// verify-then-stat order).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StatObservation {
    /// Repo-relative path, `/`-separated (index path form).
    pub path: String,
    /// The content hash the re-hash produced (== the index entry's).
    pub object_hash: Hash,
    /// Stat fields captured from the opened fd before the read, in
    /// [`stat_cache_fields`] order.
    pub mtime_ns: u64,
    pub size: u64,
    pub ino: u64,
    pub ctime_ns: u64,
}

/// Build a tree object for `dir` and its subdirectories. Honours the
/// `.gitignore` + `.mkitignore` ignore files loaded from `dir`.
///
/// Ignore rules only exclude **untracked** content: a path that is tracked
/// (or whose subtree holds tracked content) is always included even if it
/// matches an ignore rule, so a tracked file matching `.gitignore` is never
/// dropped from the worktree snapshot (which would misreport it as a deletion
/// in status/diff). The staging index at `<dir>/.mkit/index` provides the
/// tracked set; an absent index means nothing is tracked.
///
/// # Errors
/// See [`WorktreeError`].
pub fn build_tree<S: ObjectSink + Sync + ?Sized>(sink: &S, dir: &Path) -> WorktreeResult<Hash> {
    build_tree_filtered(sink, dir, None)
}

/// Like [`build_tree`], but the caller supplies the authoritative tracked
/// set (`index`). Callers that seed their index from `HEAD` when no index
/// file exists yet (status, restore safety) MUST pass it here so a tracked
/// file that matches an ignore rule is not dropped right after a checkout.
/// `None` falls back to the on-disk `<dir>/.mkit/index` (empty if absent).
///
/// # Errors
/// See [`WorktreeError`].
pub fn build_tree_filtered<S: ObjectSink + Sync + ?Sized>(
    sink: &S,
    dir: &Path,
    index: Option<&Index>,
) -> WorktreeResult<Hash> {
    build_tree_filtered_observed(sink, dir, index, &mut Vec::new())
}

/// [`build_tree_filtered`] that additionally reports every
/// [`StatObservation`] (file re-hashed to a hash matching its index
/// entry) into `observations`, so callers can heal the stat cache from
/// hash-time stats.
///
/// # Errors
/// See [`WorktreeError`].
pub fn build_tree_filtered_observed<S: ObjectSink + Sync + ?Sized>(
    sink: &S,
    dir: &Path,
    index: Option<&Index>,
    observations: &mut Vec<StatObservation>,
) -> WorktreeResult<Hash> {
    build_tree_observed_impl(sink, None, dir, index, observations)
}

/// Build a snapshot with read access to newly written and indexed objects.
/// On a cache miss, equal bytes preserve the staged object identity even when
/// the current chunker would choose a different representation.
///
/// # Errors
/// See [`WorktreeError`]; unreadable staged content is an error.
pub fn build_tree_filtered_observed_with_source<S: ObjectSink + Sync + ?Sized>(
    sink: &S,
    source: &(dyn crate::store::ObjectSource + Sync),
    dir: &Path,
    index: Option<&Index>,
    observations: &mut Vec<StatObservation>,
) -> WorktreeResult<Hash> {
    build_tree_observed_impl(sink, Some(source), dir, index, observations)
}

fn build_tree_observed_impl<S: ObjectSink + Sync + ?Sized>(
    sink: &S,
    source: Option<&(dyn crate::store::ObjectSource + Sync)>,
    dir: &Path,
    index: Option<&Index>,
    observations: &mut Vec<StatObservation>,
) -> WorktreeResult<Hash> {
    let ignores = ignore::load(dir).map_err(|e| match e {
        crate::ignore::IgnoreError::Io(io) => WorktreeError::Io(io),
        crate::ignore::IgnoreError::FileTooLarge => {
            WorktreeError::Io(io::Error::other("ignore file exceeds 1 MiB"))
        }
    })?;
    // Tracked set for ignore exemption: the caller's index if given, else the
    // on-disk index (missing/unreadable = empty = nothing tracked).
    let loaded;
    let index = if let Some(i) = index {
        i
    } else {
        // Single-layout assumption: `dir` is treated as a classic
        // single-worktree root. Linked-worktree callers (#493 Phase 1+)
        // must pass `Some(index)` or this fallback reads the wrong
        // index; the CLI always passes the discovered layout's index.
        loaded = index::read_index(&crate::layout::RepoLayout::single(dir))
            .map_err(|e| WorktreeError::Io(io::Error::other(e)))?;
        &loaded
    };
    // O(1) per-file entry lookups; `Index::find_entry` is a linear scan
    // and the walk consults it once per regular file.
    let by_path: std::collections::HashMap<&str, &crate::index::IndexEntry> =
        index.entries.iter().map(|e| (e.path.as_str(), e)).collect();
    build_tree_inner(
        sink,
        source,
        dir,
        "",
        &ignores,
        index,
        &by_path,
        false,
        observations,
    )
}

fn retained_content_hash(
    source: Option<&(dyn crate::store::ObjectSource + Sync)>,
    indexed: Option<&crate::index::IndexEntry>,
    fresh: Hash,
) -> WorktreeResult<Hash> {
    if let (Some(entry), Some(source)) = (indexed, source)
        && entry.status != crate::index::EntryStatus::Removed
        && content_eq(source, &entry.object_hash, &fresh)?
    {
        return Ok(entry.object_hash);
    }
    Ok(fresh)
}

/// `rel_dir` is the path of `dir` relative to the repo root (empty at the
/// root), so ignore patterns can be matched against full repo-relative paths
/// rather than bare basenames. `parent_ignored` carries down whether an
/// ancestor directory is ignored (git "everything under an excluded dir is
/// excluded"); `index` is the tracked set used to exempt tracked content.
#[allow(clippy::too_many_arguments)]
fn build_tree_inner<S: ObjectSink + Sync + ?Sized>(
    sink: &S,
    source: Option<&(dyn crate::store::ObjectSource + Sync)>,
    dir: &Path,
    rel_dir: &str,
    ignores: &IgnoreList,
    index: &Index,
    by_path: &std::collections::HashMap<&str, &crate::index::IndexEntry>,
    parent_ignored: bool,
    observations: &mut Vec<StatObservation>,
) -> WorktreeResult<Hash> {
    let mut entries: Vec<TreeEntry> = Vec::new();
    // Cache-miss regular files collected during the walk below, hashed
    // together (sequentially or fanned out) once the walk finishes —
    // see `hash_pending_files`.
    let mut misses: Vec<PendingFileHash<'_>> = Vec::new();

    for entry in fs::read_dir(dir)? {
        let entry = entry?;
        let file_name = entry.file_name();
        let name_str = file_name
            .to_str()
            .ok_or(WorktreeError::InvalidUtf8)?
            .to_string();
        // `symlink_metadata` does not follow symlinks.
        let meta = entry.path().symlink_metadata()?;
        let is_dir = meta.is_dir();
        let rel_path = if rel_dir.is_empty() {
            name_str.clone()
        } else {
            format!("{rel_dir}/{name_str}")
        };
        // Exclude ignored content, but only when it is UNTRACKED — a tracked
        // path (or a dir holding tracked content) is always kept so status/
        // diff see it. An ignored dir with tracked content is descended into
        // (carrying the ignored bit) so its untracked children stay excluded.
        let entry_ignored = parent_ignored || ignores.is_ignored(&rel_path, is_dir);
        if entry_ignored && !index.tracks_path_or_descendant(&rel_path) {
            continue;
        }

        let name_bytes = name_str.as_bytes();
        if !TreeEntry::validate_name(name_bytes) {
            return Err(WorktreeError::Io(io::Error::new(
                io::ErrorKind::InvalidInput,
                format!("invalid tree entry name: {name_str:?}"),
            )));
        }

        if meta.file_type().is_file() {
            // Stat cache: when the staging index proves this file's
            // content via mtime+size+ino+ctime (+exec class), reuse the
            // staged hash without opening the file — O(stat) instead of
            // O(content) for unchanged files. The object was stored at
            // `add` time, so the tree reference stays resolvable.
            let indexed = by_path.get(rel_path.as_str()).copied();
            let cached = indexed.filter(|e| stat_matches(e, &meta));
            if let Some(e) = cached {
                entries.push(TreeEntry {
                    name: name_str.into_bytes(),
                    mode: entry_mode_from_file_metadata(&meta),
                    object_hash: e.object_hash,
                });
            } else {
                // Placeholder, overwritten by `hash_pending_files` below
                // once every miss in this directory has been hashed —
                // its slot is this entry's fixed position in `entries`.
                let slot = entries.len();
                entries.push(TreeEntry {
                    name: name_str.into_bytes(),
                    mode: EntryMode::Blob,
                    object_hash: crate::hash::ZERO,
                });
                misses.push(PendingFileHash {
                    slot,
                    abs_path: entry.path(),
                    rel_path,
                    indexed,
                });
            }
        } else if meta.file_type().is_dir() {
            // A directory on disk at a path tracked as a *file* shadows that
            // tracked entry. git reports only the tracked-side deletion and
            // suppresses the directory's contents as untracked (#288); mirror
            // that by leaving the whole subtree out of the snapshot, so the
            // tracked file reads as deleted and nothing inside surfaces.
            if index.has_tracked_file_at(&rel_path) {
                continue;
            }
            let h = build_tree_inner(
                sink,
                source,
                &entry.path(),
                &rel_path,
                ignores,
                index,
                by_path,
                entry_ignored,
                observations,
            )?;
            entries.push(TreeEntry {
                name: name_str.into_bytes(),
                mode: EntryMode::Tree,
                object_hash: h,
            });
        } else if meta.file_type().is_symlink() {
            let target = fs::read_link(entry.path())?;
            let target_str = target
                .to_str()
                .ok_or(WorktreeError::InvalidUtf8)?
                .to_string();
            if !validate_symlink_target(&target_str) {
                return Err(WorktreeError::InvalidSymlinkTarget(target_str));
            }
            let target_bytes = target_str.as_bytes();
            let prologue = serialize::blob_prologue(target_bytes.len())?;
            let h = sink.put_parts(&[&prologue, target_bytes])?;
            entries.push(TreeEntry {
                name: name_str.into_bytes(),
                mode: EntryMode::Symlink,
                object_hash: h,
            });
        } else {
            // Block / char / fifo / socket — silently skip.
        }
    }

    if !misses.is_empty() {
        hash_pending_files(sink, source, &misses, &mut entries, observations)?;
    }

    entries.sort_by(|a, b| a.name.cmp(&b.name));
    let tree = Object::Tree(Tree { entries });
    let bytes = serialize::serialize(&tree)?;
    Ok(sink.put(&bytes)?)
}

/// One cache-miss regular file discovered by [`build_tree_inner`]'s walk
/// (its stat cache was absent or racy-smudged): still needs
/// [`hash_file_with_metadata`], off the main walk thread once there's
/// enough of them in one directory to amortize the dispatch. `'e`
/// borrows from the same [`Index`] `by_path` does.
struct PendingFileHash<'e> {
    /// This miss's fixed position in `build_tree_inner`'s `entries` —
    /// stable regardless of how the batch below is chunked or ordered,
    /// so results always land back on the right entry.
    slot: usize,
    abs_path: PathBuf,
    rel_path: String,
    indexed: Option<&'e crate::index::IndexEntry>,
}

/// Below this many cache-miss files per available thread, hashing them
/// in [`build_tree_inner`]'s own loop wins over thread-spawn overhead —
/// same crossover shape as [`probe_staged_objects`]. The value borrows
/// `mkit-cli`'s `commands::add::HASH_FANOUT_FILES_PER_THREAD`
/// (measured on PR #951's Slack thread): both fan out the exact same
/// per-item cost, `hash_file_with_metadata`'s open+read+BLAKE3, so the
/// same crossover point applies here. `worktree_walk_fanout` is this
/// path's own regression guard, not a from-scratch tuning.
#[cfg(not(target_arch = "wasm32"))]
const FILE_HASH_MISSES_PER_THREAD: usize = 8;

/// Hash every cache-miss file in `misses` (see [`PendingFileHash`]),
/// filling in `entries[slot]` and pushing a [`StatObservation`] for each
/// one whose fresh hash proves the stat cache healable. Sequential
/// below a thread-count-scaled threshold, fanned out across scoped
/// worker threads at or above it — [`probe_staged_objects_parallel`]'s
/// shape (`mkit-core` stays dependency-neutral and wasm-clean, so this
/// uses `std::thread::scope` directly rather than rayon). Every file's
/// hash result is written back to `entries`/`observations` in `misses`'
/// original directory-walk order regardless of which thread computed
/// it, so the output is byte-identical to the old fully-sequential
/// loop; only which of several simultaneous I/O errors is reported can
/// vary between runs — the same accepted nondeterminism
/// `probe_staged_objects_parallel` already has.
fn hash_pending_files<S: ObjectSink + Sync + ?Sized>(
    sink: &S,
    source: Option<&(dyn crate::store::ObjectSource + Sync)>,
    misses: &[PendingFileHash<'_>],
    entries: &mut [TreeEntry],
    observations: &mut Vec<StatObservation>,
) -> WorktreeResult<()> {
    #[cfg(not(target_arch = "wasm32"))]
    {
        let threads = available_threads();
        if threads > 1 && misses.len() >= FILE_HASH_MISSES_PER_THREAD.saturating_mul(threads) {
            let results = hash_pending_files_parallel(sink, source, misses, threads)?;
            apply_hash_results(misses, results, entries, observations);
            return Ok(());
        }
    }
    let results = misses
        .iter()
        .map(|m| hash_one_pending_file(sink, source, m))
        .collect::<WorktreeResult<Vec<_>>>()?;
    apply_hash_results(misses, results, entries, observations);
    Ok(())
}

/// Hash one [`PendingFileHash`]: pure function of `sink`/`source` and
/// the miss itself, safe to call concurrently across a batch — shared
/// by [`hash_pending_files`]'s sequential loop and
/// [`hash_pending_files_parallel`]'s worker threads.
fn hash_one_pending_file<S: ObjectSink + ?Sized>(
    sink: &S,
    source: Option<&(dyn crate::store::ObjectSource + Sync)>,
    m: &PendingFileHash<'_>,
) -> WorktreeResult<(Hash, EntryMode, Option<StatObservation>)> {
    let (mut h, opened_meta) = hash_file_with_metadata(sink, &m.abs_path)?;
    h = retained_content_hash(source, m.indexed, h)?;
    // Cache miss that re-hashed back to the staged hash: report the
    // observation (stat captured from the opened fd BEFORE the content
    // read) so callers can heal the racy-smudged cache soundly.
    let observation = if let Some(e) = m.indexed
        && e.object_hash == h
    {
        let (mtime_ns, size, ino, ctime_ns) = stat_cache_fields(&opened_meta);
        Some(StatObservation {
            path: m.rel_path.clone(),
            object_hash: h,
            mtime_ns,
            size,
            ino,
            ctime_ns,
        })
    } else {
        None
    };
    Ok((h, entry_mode_from_file_metadata(&opened_meta), observation))
}

/// Parallel branch of [`hash_pending_files`]: [`chunked_scoped_map`]
/// over `misses`, hashing each one via [`hash_one_pending_file`].
#[cfg(not(target_arch = "wasm32"))]
fn hash_pending_files_parallel<S: ObjectSink + Sync + ?Sized>(
    sink: &S,
    source: Option<&(dyn crate::store::ObjectSource + Sync)>,
    misses: &[PendingFileHash<'_>],
    threads: usize,
) -> WorktreeResult<Vec<(Hash, EntryMode, Option<StatObservation>)>> {
    chunked_scoped_map(misses, threads, |m| hash_one_pending_file(sink, source, m))
}

/// Write [`hash_pending_files`]'s per-miss results back into `entries`
/// and `observations`, in `misses`' original directory-walk order.
fn apply_hash_results(
    misses: &[PendingFileHash<'_>],
    results: Vec<(Hash, EntryMode, Option<StatObservation>)>,
    entries: &mut [TreeEntry],
    observations: &mut Vec<StatObservation>,
) {
    for (m, (hash, mode, observation)) in misses.iter().zip(results) {
        entries[m.slot].object_hash = hash;
        entries[m.slot].mode = mode;
        if let Some(obs) = observation {
            observations.push(obs);
        }
    }
}

/// Build a tree object from an [`Index`] (the staging area).
///
/// Walks the flat list of entries, groups them by directory, and
/// recursively materialises sub-tree objects so the on-disk shape
/// matches what [`build_tree`] would produce for the same set of
/// paths. Entries with [`crate::index::EntryStatus::Removed`] are
/// excluded; everything else maps to an [`EntryMode`] one-to-one.
///
/// A file entry (Blob/Executable) may address either a single
/// [`Blob`](crate::object::Blob) or, for content above
/// [`CHUNK_THRESHOLD`], a [`ChunkedBlob`]
/// manifest — exactly the two shapes `store_file_object` (and hence
/// `add`/`hash_file`/`build_tree`) can produce. Symlink entries must be
/// a single `Blob`. Any other object kind under a file entry is rejected.
///
/// # Errors
/// - [`WorktreeError::Io`] on a [`crate::object::TreeEntry::validate_name`]
///   failure (the path's leaf segment is reserved or alias-prone), or
///   when a file entry points at a non-blob/non-chunked-blob object.
/// - Wraps [`crate::MkitError`] surfaced by `serialize` / `store.write`.
pub fn build_tree_from_index(
    store: &ObjectStore,
    index: &crate::index::Index,
) -> WorktreeResult<Hash> {
    // The convenience wrapper publishes a durable tree (commit/merge/
    // rebase/…), so it integrity-verifies staged objects by default.
    build_tree_from_index_with(store, store, index, true)
}

/// [`build_tree_from_index`] writing tree objects through `sink` —
/// pass a [`WriteBatch`](crate::batch::WriteBatch) to amortise the
/// flush cost of all materialised trees into the batch's single commit.
/// `store` is still needed read-only to validate that staged hashes
/// point at blob-shaped objects (a sink cannot read).
///
/// `verify` selects the staged-object integrity check. With `verify =
/// true` (every path that publishes a durable tree) each referenced
/// object is read and re-hashed before the tree is materialised, so a
/// corrupt staged object can never be published. With `verify = false`
/// (ephemeral status/diff snapshots that publish nothing durable) only
/// the 6-byte prologue is read for the blob-shape check — the read path
/// still integrity-verifies the object whenever it is actually used.
///
/// # Errors
/// See [`build_tree_from_index`].
#[allow(clippy::items_after_statements, clippy::too_many_lines)]
pub fn build_tree_from_index_with<S: ObjectSink + ?Sized>(
    store: &ObjectStore,
    sink: &S,
    index: &crate::index::Index,
    verify: bool,
) -> WorktreeResult<Hash> {
    use crate::index::EntryStatus;

    // Build an in-memory directory tree. Each node is either a leaf
    // (one staged blob/symlink) or a directory containing children.
    #[derive(Default)]
    struct Node {
        // Subdirectory name → child node.
        children: std::collections::BTreeMap<String, Node>,
        // Leaf entries directly under this dir: name → (mode, hash).
        leaves: std::collections::BTreeMap<String, (EntryMode, Hash)>,
    }

    let mut root = Node::default();
    let mut seen_paths = std::collections::HashSet::with_capacity(index.entries.len());

    // Pass 1: validate paths/status and collect the surviving entries'
    // (path, mode, staged hash) — cheap, pure-CPU work, no store access.
    // The original single-pass loop bailed on the *first* entry (in
    // index order) with any problem — duplicate path, reserved status,
    // or bad staged object — never even looking at later entries. A
    // path/status problem here defers its error via `deferred_status_err`
    // and stops collecting instead of returning immediately, so pass 2
    // below still gets a chance to surface an *earlier* staged-object
    // problem among the entries collected before the stopping point —
    // preserving that same index-order precedence across both error
    // classes. Entries at or after the stopping point are unreachable
    // either way (the original loop never got there), so excluding them
    // from `kept` changes nothing observable.
    let mut kept: Vec<(&str, EntryMode, Hash)> = Vec::with_capacity(index.entries.len());
    let mut deferred_status_err: Option<WorktreeError> = None;
    for entry in &index.entries {
        if !seen_paths.insert(entry.path.as_str()) {
            deferred_status_err = Some(WorktreeError::Io(io::Error::other(format!(
                "duplicate index path: '{}'",
                entry.path
            ))));
            break;
        }
        if entry.status == EntryStatus::Removed {
            continue;
        }
        let mode = match entry.status {
            EntryStatus::Blob => EntryMode::Blob,
            EntryStatus::Executable => EntryMode::Executable,
            EntryStatus::Symlink => EntryMode::Symlink,
            EntryStatus::Tree => {
                // Reserved-but-unused per SPEC-INDEX §3. Reject for
                // now; if a subtree-staging design lands later it
                // can populate this branch.
                deferred_status_err = Some(WorktreeError::Io(io::Error::other(
                    "index entry uses reserved Tree status (subtree staging not implemented)",
                )));
                break;
            }
            EntryStatus::Removed => unreachable!("filtered above"),
        };
        kept.push((entry.path.as_str(), mode, entry.object_hash));
    }

    // Pass 2: one batched staged-object check (fetch the type, then the
    // same blob-shape check the original loop ran inline right after
    // its own per-entry fetch) for every entry pass 1 kept, instead of
    // interleaving it one-at-a-time with the pass-3 tree walk below.
    // `probe_staged_objects` fans this out across threads once there's
    // enough work to amortize it (native builds only) — a `status`/
    // `diff` snapshot over a many-file repo was paying one serialized
    // `open`+`read`(+`close`) per tracked file here, dominating wall
    // time well before hashing or tree-materialization cost did. Its
    // sequential (default) path still short-circuits on the first
    // failing entry in `kept`'s (index) order, so — combined with pass
    // 1 stopping at the same point the original loop would have —
    // whichever of the two errors is earlier in index order is the one
    // that surfaces, exactly matching the original interleaved loop.
    probe_staged_objects(store, &kept, verify)?;
    if let Some(err) = deferred_status_err {
        return Err(err);
    }

    // Pass 3: walk each surviving, already-validated entry into the
    // in-memory node tree.
    for (path, mode, object_hash) in kept {
        // Split "a/b/c.txt" into ["a", "b"] + "c.txt".
        let segments: Vec<&str> = path.split('/').collect();
        let Some((leaf, dirs)) = segments.split_last() else {
            return Err(WorktreeError::Io(io::Error::other("empty index path")));
        };
        if leaf.is_empty() {
            return Err(WorktreeError::Io(io::Error::other(
                "trailing slash in index path",
            )));
        }

        let mut node = &mut root;
        let mut walked = String::new();
        for seg in dirs {
            if seg.is_empty() {
                return Err(WorktreeError::Io(io::Error::other(
                    "empty path segment in index",
                )));
            }
            // Collision: this segment was previously staged as a blob
            // (e.g. earlier index entry was `a` as a file, this one
            // is `a/b`). Tree object format requires unique entry
            // names per directory; emitting both would produce an
            // invalid tree the deserializer rejects under its strict
            // ascending-name rule.
            if node.leaves.contains_key(*seg) {
                let conflicting = if walked.is_empty() {
                    (*seg).to_string()
                } else {
                    format!("{walked}/{seg}")
                };
                return Err(WorktreeError::Io(io::Error::other(format!(
                    "index path conflict: '{conflicting}' is staged as both a file and a directory"
                ))));
            }
            walked = if walked.is_empty() {
                (*seg).to_string()
            } else {
                format!("{walked}/{seg}")
            };
            node = node.children.entry((*seg).to_string()).or_default();
        }
        // The reverse collision: this entry's leaf name already exists
        // as a child directory under the same parent (an earlier
        // entry staged `a/b` and now this one stages `a` as a file).
        if node.children.contains_key(*leaf) {
            let conflicting = if walked.is_empty() {
                (*leaf).to_string()
            } else {
                format!("{walked}/{leaf}")
            };
            return Err(WorktreeError::Io(io::Error::other(format!(
                "index path conflict: '{conflicting}' is staged as both a file and a directory"
            ))));
        }
        if node
            .leaves
            .insert((*leaf).to_string(), (mode, object_hash))
            .is_some()
        {
            let duplicate = if walked.is_empty() {
                (*leaf).to_string()
            } else {
                format!("{walked}/{leaf}")
            };
            return Err(WorktreeError::Io(io::Error::other(format!(
                "duplicate index path: '{duplicate}'"
            ))));
        }
    }

    fn write_node<S: ObjectSink + ?Sized>(sink: &S, node: &Node) -> WorktreeResult<Hash> {
        let mut entries: Vec<TreeEntry> = Vec::new();

        // Subdirectories first (alphabetical via BTreeMap).
        for (name, child) in &node.children {
            let h = write_node(sink, child)?;
            let bytes = name.as_bytes().to_vec();
            if !crate::object::TreeEntry::validate_name(&bytes) {
                return Err(WorktreeError::Io(io::Error::other(format!(
                    "invalid tree entry name: {name:?}"
                ))));
            }
            entries.push(TreeEntry {
                name: bytes,
                mode: EntryMode::Tree,
                object_hash: h,
            });
        }

        // Then leaves.
        for (name, (mode, hash)) in &node.leaves {
            let bytes = name.as_bytes().to_vec();
            if !crate::object::TreeEntry::validate_name(&bytes) {
                return Err(WorktreeError::Io(io::Error::other(format!(
                    "invalid tree entry name: {name:?}"
                ))));
            }
            entries.push(TreeEntry {
                name: bytes,
                mode: *mode,
                object_hash: *hash,
            });
        }

        // Tree-entry order is name-ascending per SPEC-OBJECTS §4.
        entries.sort_by(|a, b| a.name.cmp(&b.name));
        let tree = Object::Tree(Tree { entries });
        let bytes = serialize::serialize(&tree)?;
        Ok(sink.put(&bytes)?)
    }

    write_node(sink, &root)
}

/// `std::thread::available_parallelism()`, cached process-wide — the
/// core count never changes during a process's lifetime, so every
/// native fan-out crossover check in this module (there are several,
/// and [`build_tree_inner`]'s recurses once per directory) shares one
/// syscall instead of paying it again at every call site/recursion.
#[cfg(not(target_arch = "wasm32"))]
fn available_threads() -> usize {
    static THREADS: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
    *THREADS
        .get_or_init(|| std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get))
}

/// Shared chunk+scoped-thread+join scaffolding behind this module's two
/// order-preserving, short-circuit-on-first-error native fan-outs
/// ([`probe_staged_objects_parallel`], [`hash_pending_files_parallel`]):
/// split `items` into `threads` contiguous chunks, run each chunk's
/// items through `f` sequentially on its own scoped thread, and join.
/// `std::thread::scope` (not a persistent pool) keeps `mkit-core`
/// dependency-neutral and wasm-clean.
///
/// (`pack::stage_raw_entries_parallel` looks similar but isn't folded
/// in here: it *defers* rather than short-circuits on an item error, via
/// a `first_failure` atomic shared across every chunk so a later pack
/// position can still be skipped once an earlier one fails — genuinely
/// different result semantics from this helper's per-chunk
/// short-circuit, not just a different call shape.)
///
/// Every item's `f` output lands in the returned `Vec` at that item's
/// original index in `items`, regardless of which thread computed it.
/// A chunk stops calling `f` on its own remaining items as soon as one
/// of them errors, but a concurrently-running chunk is not signalled to
/// stop early; on any error, which chunk's is the one returned is
/// unspecified — the same accepted nondeterminism
/// `probe_staged_objects_parallel` had before this helper existed.
///
/// # Panics
/// Propagates a worker thread's panic once `std::thread::scope` joins
/// it, rather than swallowing it.
#[cfg(not(target_arch = "wasm32"))]
fn chunked_scoped_map<T, U, E>(
    items: &[T],
    threads: usize,
    f: impl Fn(&T) -> Result<U, E> + Sync,
) -> Result<Vec<U>, E>
where
    T: Sync,
    U: Send,
    E: Send,
{
    let chunk_size = items.len().div_ceil(threads).max(1);
    let mut out: Vec<Option<U>> = (0..items.len()).map(|_| None).collect();
    let mut first_err: Option<E> = None;

    std::thread::scope(|scope| {
        let f = &f;
        let handles: Vec<_> = items
            .chunks(chunk_size)
            .enumerate()
            .map(|(chunk_idx, chunk)| {
                let base = chunk_idx * chunk_size;
                scope.spawn(move || {
                    let mut results = Vec::with_capacity(chunk.len());
                    for it in chunk {
                        match f(it) {
                            Ok(u) => results.push(u),
                            Err(e) => return (base, results, Some(e)),
                        }
                    }
                    (base, results, None)
                })
            })
            .collect();
        for handle in handles {
            let (base, results, err) = handle
                .join()
                .expect("chunked fan-out worker thread panicked");
            for (i, u) in results.into_iter().enumerate() {
                out[base + i] = Some(u);
            }
            if let Some(e) = err
                && first_err.is_none()
            {
                first_err = Some(e);
            }
        }
    });

    match first_err {
        Some(e) => Err(e),
        None => Ok(out
            .into_iter()
            .map(|o| o.expect("every slot filled when no error was reported"))
            .collect()),
    }
}

/// Staged-object check for every `(path, mode, hash)` triple in
/// `entries`, in order — [`build_tree_from_index_with`]'s batched pass-2
/// step. Below a small-batch threshold this runs a plain sequential
/// loop (short-circuiting on `entries`' first failure, same as a plain
/// `for` loop would); at or above it (native builds only — wasm32 has
/// no threads) fans out across a scoped thread pool sized to the
/// machine, same shape as [`crate::pack`]'s
/// `stage_raw_entries`/`stage_raw_entries_parallel`: each check is one
/// `open`+`read`(+`close`) syscall triplet on an independent loose
/// object file, so the calls parallelize with none of the ordering or
/// shared-mutable-state concerns a write path has to account for.
fn probe_staged_objects(
    store: &ObjectStore,
    entries: &[(&str, EntryMode, Hash)],
    verify: bool,
) -> WorktreeResult<()> {
    #[cfg(not(target_arch = "wasm32"))]
    {
        // Below this many entries per available thread, thread-spawn
        // overhead isn't worth it — same crossover shape as
        // `pack::stage_raw_entries`, tuned here by the
        // `status_snapshot` bench.
        const ENTRIES_PER_THREAD: usize = 32;
        let threads = available_threads();
        if threads > 1 && entries.len() >= ENTRIES_PER_THREAD.saturating_mul(threads) {
            return probe_staged_objects_parallel(store, entries, verify, threads);
        }
    }
    entries.iter().try_for_each(|&(path, mode, hash)| {
        check_one_staged_object(store, path, mode, hash, verify)
    })
}

/// Parallel branch of [`probe_staged_objects`]: [`chunked_scoped_map`]
/// over `entries`, checking each one via [`check_one_staged_object`] and
/// discarding its (unit) results — only whether any entry errored
/// matters here.
///
/// On an index with more than one entry pointing at a missing,
/// malformed, or wrong-shape object, the specific error surfaced here
/// depends on chunk/thread scheduling, not index position —
/// `chunked_scoped_map`'s accepted nondeterminism. The tree build is
/// rejected either way; only which error is reported can vary between
/// runs.
#[cfg(not(target_arch = "wasm32"))]
fn probe_staged_objects_parallel(
    store: &ObjectStore,
    entries: &[(&str, EntryMode, Hash)],
    verify: bool,
    threads: usize,
) -> WorktreeResult<()> {
    chunked_scoped_map(entries, threads, |&(path, mode, hash)| {
        check_one_staged_object(store, path, mode, hash, verify)
    })
    .map(|_: Vec<()>| ())
}

// Pure per-entry step shared by both branches of
// `probe_staged_objects`: fetch `hash`'s object-type tag —
// `verify_object_type` (full read + rehash) when `verify`, else the
// cheap prologue-only `object_type` — then run the same blob-shape
// check the original single-pass loop ran inline right after its own
// per-entry fetch.
//
// A regular file (Blob/Executable) may be stored as a single Blob or,
// for content above CHUNK_THRESHOLD, a ChunkedBlob manifest —
// `add`/`hash_file`/`build_tree` all route through
// `store_file_object`. A Symlink is always a single Blob (its target
// path). Accept both blob shapes for file entries so the commit/index
// path agrees with the worktree-hashing path; a tree/commit/etc.
// under a file entry is still rejected. Publishing paths (`verify`)
// read + re-hash the staged object so a tree never references a
// corrupt blob; the read path's hash check is the same one `add`
// passed, so this only catches post-`add` corruption. Ephemeral
// status/diff snapshots skip it — re-reading every staged blob on
// every status dominates large repos of small files, and they
// publish nothing durable.
fn check_one_staged_object(
    store: &ObjectStore,
    path: &str,
    mode: EntryMode,
    hash: Hash,
    verify: bool,
) -> WorktreeResult<()> {
    let object_type = if verify {
        store.verify_object_type(&hash)?
    } else {
        store.object_type(&hash)?
    };
    match object_type {
        crate::object::ObjectType::Blob => Ok(()),
        crate::object::ObjectType::ChunkedBlob if mode != EntryMode::Symlink => Ok(()),
        other => Err(WorktreeError::Io(io::Error::other(format!(
            "index entry '{path}' points to a non-blob object (got {})",
            other.name()
        )))),
    }
}

/// Read a file from disk, hash it, store it, and return the
/// content-address of the resulting object.
///
/// Files at or below [`CHUNK_THRESHOLD`] become a single
/// [`Blob`](crate::object::Blob). Files above the threshold are split
/// with [`FastCdc::v1`]; each chunk is stored as a `Blob`, and the
/// file is represented by a [`ChunkedBlob`]
/// manifest whose hash is returned and lands in the parent tree. See
/// `SPEC-FASTCDC.md` and `SPEC-OBJECTS.md` §7.
///
/// # Errors
/// See [`WorktreeError`].
pub fn hash_file<S: ObjectSink + ?Sized>(sink: &S, path: &Path) -> WorktreeResult<Hash> {
    hash_file_with_metadata(sink, path).map(|(hash, _)| hash)
}

/// Read a regular file without following the final path component on
/// Unix, enforcing [`MAX_FILE_BYTES`] against both the opened handle's
/// metadata and the actual bytes read.
pub fn read_regular_file_bounded(path: &Path) -> WorktreeResult<(fs::Metadata, Vec<u8>)> {
    let mut file = open_regular_file(path)?;
    let meta = file.metadata()?;
    if !meta.file_type().is_file() {
        return Err(WorktreeError::Io(io::Error::new(
            io::ErrorKind::InvalidInput,
            "path is not a regular file",
        )));
    }
    if meta.len() > MAX_FILE_BYTES {
        return Err(WorktreeError::FileTooLarge(path.to_path_buf()));
    }
    let initial_capacity = usize::try_from(meta.len().min(CHUNK_THRESHOLD))
        .map_err(|_| WorktreeError::FileTooLarge(path.to_path_buf()))?;
    let mut data = Vec::with_capacity(initial_capacity);
    file.by_ref()
        .take(MAX_FILE_BYTES + 1)
        .read_to_end(&mut data)?;
    if u64::try_from(data.len()).unwrap_or(u64::MAX) > MAX_FILE_BYTES {
        return Err(WorktreeError::FileTooLarge(path.to_path_buf()));
    }
    Ok((meta, data))
}

/// Hash and store a regular file, returning its content-address and the
/// [`fs::Metadata`] observed when it was opened (for stat-cache use by
/// callers like `mkit add`).
///
/// Files at or below [`CHUNK_THRESHOLD`] are read fully into memory — a
/// single [`Blob`](crate::object::Blob) needs its bytes contiguous
/// anyway, and the threshold keeps this small (1 MiB). Files above the
/// threshold are streamed chunk-by-chunk directly from the open file
/// handle instead of first buffering the whole file (issue #828):
/// ingest memory is bounded by one `FastCdc` window regardless of the
/// file's total size.
///
/// # Errors
/// See [`WorktreeError`].
pub fn hash_file_with_metadata<S: ObjectSink + ?Sized>(
    sink: &S,
    path: &Path,
) -> WorktreeResult<(Hash, fs::Metadata)> {
    hash_file_with_metadata_using(sink, path, store_large_file_streaming)
}

/// [`hash_file_with_metadata`], but with an explicit `hash_chunks` batch
/// callback in place of the built-in sequential loop over
/// [`store_chunk_blob`] — see [`store_large_file_streaming_with`], which
/// this delegates to for files above [`CHUNK_THRESHOLD`].
///
/// # Errors
/// See [`WorktreeError`].
pub fn hash_file_with_metadata_with<S: ObjectSink + ?Sized>(
    sink: &S,
    path: &Path,
    hash_chunks: impl FnMut(&S, &[Vec<u8>]) -> WorktreeResult<Vec<Hash>>,
) -> WorktreeResult<(Hash, fs::Metadata)> {
    hash_file_with_metadata_using(sink, path, |sink, reader, path| {
        store_large_file_streaming_with(sink, reader, path, hash_chunks)
    })
}

/// Share file opening, size checks, and metadata capture between sequential
/// and batched ingest. Only the large-file streaming strategy differs.
fn hash_file_with_metadata_using<S: ObjectSink + ?Sized>(
    sink: &S,
    path: &Path,
    stream: impl FnOnce(&S, io::Take<fs::File>, &Path) -> WorktreeResult<Hash>,
) -> WorktreeResult<(Hash, fs::Metadata)> {
    let mut file = open_regular_file(path)?;
    let meta = file.metadata()?;
    if !meta.file_type().is_file() {
        return Err(WorktreeError::Io(io::Error::new(
            io::ErrorKind::InvalidInput,
            "path is not a regular file",
        )));
    }
    if meta.len() > MAX_FILE_BYTES {
        return Err(WorktreeError::FileTooLarge(path.to_path_buf()));
    }

    if meta.len() <= CHUNK_THRESHOLD {
        let initial_capacity = usize::try_from(meta.len())
            .map_err(|_| WorktreeError::FileTooLarge(path.to_path_buf()))?;
        let mut data = Vec::with_capacity(initial_capacity);
        file.by_ref()
            .take(MAX_FILE_BYTES + 1)
            .read_to_end(&mut data)?;
        if u64::try_from(data.len()).unwrap_or(u64::MAX) > MAX_FILE_BYTES {
            return Err(WorktreeError::FileTooLarge(path.to_path_buf()));
        }
        let hash = store_file_object(sink, &data)?;
        return Ok((hash, meta));
    }

    let hash = stream(sink, file.take(MAX_FILE_BYTES + 1), path)?;
    Ok((hash, meta))
}

/// Number of chunks [`store_large_file_streaming_with`] buffers before
/// handing a batch to its `hash_chunks` callback. Bounds the extra
/// memory a fan-out `hash_chunks` needs to hold resident to at most
/// `STREAM_HASH_BATCH * chunker::MAX_SIZE` (16 MiB at the current 256
/// KiB `MAX_SIZE`) regardless of the file's total size — the same
/// "independent of file size" bound issue #828 gave the streaming
/// reader itself, just sized for a batch instead of one chunk.
const STREAM_HASH_BATCH: usize = 64;

/// Store one already-cut chunk as a canonical Blob object, returning its
/// content address. The per-chunk unit [`store_large_file_streaming_with`]
/// fans out over — hashing (BLAKE3, via `put_parts`) and staging a temp
/// file are both independent of every other chunk once boundaries are
/// known.
///
/// # Errors
/// See [`WorktreeError`].
pub fn store_chunk_blob<S: ObjectSink + ?Sized>(sink: &S, chunk: &[u8]) -> WorktreeResult<Hash> {
    let prologue = serialize::blob_prologue(chunk.len())?;
    Ok(sink.put_parts(&[&prologue, chunk])?)
}

/// Sequential ingest borrows each chunk directly from the reader window.
/// It does not allocate owned chunks or retain a batch between writes.
fn store_large_file_streaming<S: ObjectSink + ?Sized, R: Read>(
    sink: &S,
    reader: R,
    path: &Path,
) -> WorktreeResult<Hash> {
    let mut chunker = ChunkReader::new(FastCdc::v1(), reader);
    let mut chunks = Vec::new();
    let mut total_size: u64 = 0;
    while let Some(chunk) = chunker.next_chunk_ref()? {
        total_size = total_size
            .checked_add(chunk.len() as u64)
            .filter(|&t| t <= MAX_FILE_BYTES)
            .ok_or_else(|| WorktreeError::FileTooLarge(path.to_path_buf()))?;
        chunks.push(store_chunk_blob(sink, chunk)?);
    }
    store_chunk_manifest(sink, total_size, chunks)
}

/// Both streaming strategies use the same canonical manifest encoding.
fn store_chunk_manifest<S: ObjectSink + ?Sized>(
    sink: &S,
    total_size: u64,
    chunks: Vec<Hash>,
) -> WorktreeResult<Hash> {
    let manifest = Object::ChunkedBlob(ChunkedBlob {
        total_size,
        chunk_size: 0, // 0 = content-defined (FastCDC) per SPEC-OBJECTS §7
        chunks,
    });
    let manifest_bytes = serialize::serialize(&manifest)?;
    Ok(sink.put(&manifest_bytes)?)
}

/// Call `hash_chunks(sink, batch)` and enforce its documented contract —
/// exactly one [`Hash`](tyalias@Hash) per input chunk — before the caller ever sees the
/// result, returning [`WorktreeError::ChunkBatchLengthMismatch`] instead of
/// silently letting a short/long result desync the manifest's `chunks`
/// list from the file's actual chunk sequence. [`store_large_file_streaming_with`]'s
/// only caller of `hash_chunks`; kept as a thin wrapper so the check lives
/// in exactly one place rather than being repeated at both call sites.
fn checked_hash_chunks<S: ObjectSink + ?Sized>(
    hash_chunks: &mut impl FnMut(&S, &[Vec<u8>]) -> WorktreeResult<Vec<Hash>>,
    sink: &S,
    batch: &[Vec<u8>],
) -> WorktreeResult<Vec<Hash>> {
    let hashes = hash_chunks(sink, batch)?;
    if hashes.len() == batch.len() {
        Ok(hashes)
    } else {
        Err(WorktreeError::ChunkBatchLengthMismatch {
            expected: batch.len(),
            actual: hashes.len(),
        })
    }
}

/// Store a large (> [`CHUNK_THRESHOLD`]) file's content as a
/// [`ChunkedBlob`] manifest, streaming chunks directly from `reader`
/// instead of requiring the whole file resident in memory first (issue
/// #828). Bounds ingest memory to one fixed `ChunkReader` window
/// (1 MiB), an owned batch of at most 64 chunks (16 MiB), and the
/// growing chunk-hash list (32 bytes/chunk). The reader and batch stay
/// bounded independently of total file size.
///
/// `path` is used only to name the file in a [`WorktreeError::FileTooLarge`]
/// error if `reader` yields more than [`MAX_FILE_BYTES`].
///
/// Takes an explicit `hash_chunks` batch callback in place of a built-in
/// sequential loop over [`store_chunk_blob`] — `mkit-core` has no
/// thread-pool dependency of its own — it stays usable from wasm
/// targets, which have no OS threads — so the fan-out decision lives
/// with the caller instead of living in this function, the same shape
/// as [`crate::transfer::plan_pack_with`]'s `encode_deltas` callback.
/// Cutting chunk boundaries from `reader` is inherently sequential (each
/// cut's start is the previous cut's end), but once a batch of up to
/// `STREAM_HASH_BATCH` chunks is in hand, hashing and storing each one
/// is independent of every other chunk in the batch — `mkit-cli`'s
/// native, rayon-backed `add` path passes a parallel `hash_chunks`
/// (mirroring the pack-compression/signature-verification/delta-encoding
/// fan-outs already in `mkit-cli`). [`hash_file_with_metadata`] uses a
/// separate sequential path that borrows chunks without batching.
///
/// `hash_chunks` receives each batch in file order and MUST return
/// exactly one [`Hash`](tyalias@Hash) per input chunk, in the same order — the
/// manifest's chunk list, and therefore the file's content address,
/// depends on that order.
///
/// Cutting chunk boundaries is otherwise strictly sequential (each cut's
/// start is the previous cut's end), but cutting *batch N+1* has no data
/// dependency on *hashing batch N* — only on the reader position, which
/// the cutter alone advances. Native builds (see
/// `store_large_file_streaming_pipelined`) overlap the two phases
/// across a scoped worker thread instead of running them strictly one
/// after the other; wasm32 (no threads) falls back to
/// `store_large_file_streaming_batched`, the original phase-serial
/// loop. (Both are private, cfg-gated to exactly one of the two
/// targets, so neither is a valid intra-doc link here — a native `cargo
/// doc` build never sees `store_large_file_streaming_batched` at all,
/// and vice versa on wasm32.)
///
/// # Errors
/// See [`WorktreeError`].
pub fn store_large_file_streaming_with<S: ObjectSink + ?Sized, R: Read + Send>(
    sink: &S,
    reader: R,
    path: &Path,
    mut hash_chunks: impl FnMut(&S, &[Vec<u8>]) -> WorktreeResult<Vec<Hash>>,
) -> WorktreeResult<Hash> {
    #[cfg(not(target_arch = "wasm32"))]
    {
        store_large_file_streaming_pipelined(sink, reader, path, &mut hash_chunks)
    }
    #[cfg(target_arch = "wasm32")]
    {
        store_large_file_streaming_batched(sink, reader, path, &mut hash_chunks)
    }
}

/// Original phase-serial implementation of
/// [`store_large_file_streaming_with`]: cut one batch, then hash it via
/// `hash_chunks`, then cut the next. wasm32 (no threads to pipeline
/// across) uses this directly; native builds use
/// `store_large_file_streaming_pipelined` instead (native-only; not a
/// valid intra-doc link from a wasm32 doc build).
#[cfg(target_arch = "wasm32")]
fn store_large_file_streaming_batched<S: ObjectSink + ?Sized, R: Read>(
    sink: &S,
    reader: R,
    path: &Path,
    hash_chunks: &mut impl FnMut(&S, &[Vec<u8>]) -> WorktreeResult<Vec<Hash>>,
) -> WorktreeResult<Hash> {
    let mut chunker = ChunkReader::new(FastCdc::v1(), reader);
    let mut chunks = Vec::new();
    let mut total_size: u64 = 0;
    let mut batch: Vec<Vec<u8>> = Vec::with_capacity(STREAM_HASH_BATCH);
    while let Some(chunk) = chunker.next_chunk()? {
        total_size = total_size
            .checked_add(chunk.len() as u64)
            .filter(|&t| t <= MAX_FILE_BYTES)
            .ok_or_else(|| WorktreeError::FileTooLarge(path.to_path_buf()))?;
        batch.push(chunk);
        if batch.len() == STREAM_HASH_BATCH {
            chunks.extend(checked_hash_chunks(hash_chunks, sink, &batch)?);
            batch.clear();
        }
    }
    if !batch.is_empty() {
        chunks.extend(checked_hash_chunks(hash_chunks, sink, &batch)?);
    }

    store_chunk_manifest(sink, total_size, chunks)
}

/// Pipelined implementation of [`store_large_file_streaming_with`]: a
/// scoped worker thread runs `reader` through the `FastCdc` cutter and
/// hands off each full [`STREAM_HASH_BATCH`]-sized batch (plus the final
/// partial one) over a rendezvous (zero-capacity) channel, while this
/// (the calling) thread receives batches and runs `hash_chunks` on them
/// — so cutting batch N+1 overlaps hashing batch N instead of waiting
/// for it. A zero-capacity channel's `send` blocks until the matching
/// `recv`, so the cutter can have at most one batch *fully built* ahead
/// of the one currently being hashed (held on its own stack, blocked on
/// send) — never two, the way a buffered channel would allow by letting
/// it dequeue immediately and start a third batch. That caps the extra
/// memory this adds at one more `STREAM_HASH_BATCH` (matching this
/// module's existing "independent of file size" bound) while still
/// overlapping the two phases fully: the blocked send/recv handshake
/// itself is a single unblocked pair of syscalls, not a wait for work.
///
/// The cutter thread performs the exact same running-total overflow
/// check `store_large_file_streaming_batched` (wasm32-only; not a valid
/// intra-doc link here) does, so an oversized
/// file is still rejected as soon as the cutter itself detects it
/// (before hashing catches up) rather than only once every chunk has
/// been received; the receiving thread separately sums each received
/// chunk's length into `total_size` for the final manifest, which is
/// always consistent with the cutter's own total since it receives
/// every chunk the cutter decided to keep.
///
/// If `hash_chunks` (or the overflow check) errors, this thread returns
/// early and drops its end of the channel; the cutter's next blocked
/// [`std::sync::mpsc::SyncSender::send`] then fails and it exits — no
/// deadlock, no unjoined thread (`std::thread::scope` joins it before
/// returning). If the cutter itself panics, `std::thread::scope`
/// resumes that panic here once joined, the same behavior
/// [`crate::pack::PackReader::read`]'s own scoped fan-out relies on.
#[cfg(not(target_arch = "wasm32"))]
fn store_large_file_streaming_pipelined<S: ObjectSink + ?Sized, R: Read + Send>(
    sink: &S,
    reader: R,
    path: &Path,
    hash_chunks: &mut impl FnMut(&S, &[Vec<u8>]) -> WorktreeResult<Vec<Hash>>,
) -> WorktreeResult<Hash> {
    use std::sync::mpsc;

    let (tx, rx) = mpsc::sync_channel::<WorktreeResult<Vec<Vec<u8>>>>(0);

    let (chunks, total_size) = std::thread::scope(|scope| -> WorktreeResult<(Vec<Hash>, u64)> {
        scope.spawn(move || {
            let mut chunker = ChunkReader::new(FastCdc::v1(), reader);
            let mut batch: Vec<Vec<u8>> = Vec::with_capacity(STREAM_HASH_BATCH);
            let mut running_total: u64 = 0;
            loop {
                let chunk = match chunker.next_chunk() {
                    Ok(Some(chunk)) => chunk,
                    Ok(None) => {
                        if !batch.is_empty() {
                            let _ = tx.send(Ok(batch));
                        }
                        return;
                    }
                    Err(e) => {
                        let _ = tx.send(Err(WorktreeError::from(e)));
                        return;
                    }
                };
                running_total = if let Some(t) = running_total
                    .checked_add(chunk.len() as u64)
                    .filter(|&t| t <= MAX_FILE_BYTES)
                {
                    t
                } else {
                    let _ = tx.send(Err(WorktreeError::FileTooLarge(path.to_path_buf())));
                    return;
                };
                batch.push(chunk);
                if batch.len() == STREAM_HASH_BATCH {
                    let full = std::mem::replace(&mut batch, Vec::with_capacity(STREAM_HASH_BATCH));
                    if tx.send(Ok(full)).is_err() {
                        return;
                    }
                }
            }
        });

        // On any error, keep draining (discarding) the channel instead of
        // returning immediately: with a rendezvous channel the cutter's
        // *next* `send` — for a batch already cut, or one it hasn't
        // finished cutting yet — has no buffer to land in and blocks
        // until a matching `recv`, so an early return here that stops
        // calling `recv` would deadlock `thread::scope`'s join waiting
        // for a cutter that is itself waiting for a `recv` that will
        // never come. Draining lets every future send succeed (or the
        // cutter hits its own error/EOF and exits on its own), so the
        // channel always disconnects and this loop always terminates.
        let mut chunks = Vec::new();
        let mut total_size: u64 = 0;
        let mut first_err: Option<WorktreeError> = None;
        while let Ok(received) = rx.recv() {
            if first_err.is_some() {
                continue;
            }
            match received.and_then(|batch| {
                for chunk in &batch {
                    total_size += chunk.len() as u64;
                }
                checked_hash_chunks(hash_chunks, sink, &batch)
            }) {
                Ok(hashes) => chunks.extend(hashes),
                Err(e) => first_err = Some(e),
            }
        }
        match first_err {
            Some(e) => Err(e),
            None => Ok((chunks, total_size)),
        }
    })?;

    store_chunk_manifest(sink, total_size, chunks)
}

/// Store a regular file's bytes as the canonical object and return its
/// content-address.
///
/// This is the single source of truth for how file content maps to an
/// object hash, shared by [`hash_file`], [`build_tree`], and `mkit add`
/// so all three agree on the representation:
///
/// - At or below [`CHUNK_THRESHOLD`]: a single
///   [`Blob`](crate::object::Blob).
/// - Above the threshold: `FastCdc::v1` chunks, each stored as a `Blob`,
///   addressed by a [`ChunkedBlob`] manifest.
///
/// # Errors
/// See [`WorktreeError`].
pub fn store_file_object<S: ObjectSink + ?Sized>(sink: &S, data: &[u8]) -> WorktreeResult<Hash> {
    if u64::try_from(data.len()).unwrap_or(u64::MAX) <= CHUNK_THRESHOLD {
        // Zero-copy: the canonical Blob bytes are `prologue ‖ data`
        // (pinned to serialize() by proptest), so the sink can hash and
        // write straight from the source buffer.
        let prologue = serialize::blob_prologue(data.len())?;
        return Ok(sink.put_parts(&[&prologue, data])?);
    }

    // Large file: split with FastCDC v1 via the public ChunkIterator,
    // store each chunk as a Blob, and assemble a ChunkedBlob manifest.
    // Per-manifest chunk count is bounded by serialize::MAX_CHUNKS
    // (1_000_000); MAX_FILE_BYTES (1 GiB) ÷ FastCDC MIN_SIZE (16 KiB)
    // = ~65k, well under the cap.
    let total_size = data.len() as u64;
    let chunks: Vec<Hash> = ChunkIterator::new(FastCdc::v1(), data)
        .map(|b| {
            let chunk = &data[b.offset..b.offset + b.length];
            let prologue = serialize::blob_prologue(chunk.len())?;
            Ok::<_, WorktreeError>(sink.put_parts(&[&prologue, chunk])?)
        })
        .collect::<Result<_, _>>()?;

    let manifest = Object::ChunkedBlob(ChunkedBlob {
        total_size,
        chunk_size: 0, // 0 = content-defined (FastCDC) per SPEC-OBJECTS §7
        chunks,
    });
    let manifest_bytes = serialize::serialize(&manifest)?;
    Ok(sink.put(&manifest_bytes)?)
}

/// Build the canonical [`ChunkedBlob`] manifest for `data`: split with
/// [`FastCdc::v1`], address each chunk by its **Blob object id** — the hash of
/// the chunk's canonical Blob bytes (`blob_prologue ‖ chunk`), i.e. the id the
/// chunk Blob is stored under — and set `chunk_size = 0` (content-defined).
///
/// This is the single source of the manifest recipe: [`hash_file_object`]
/// (read-only change detection) and the `mkit-wasm` encoder both build through
/// it, and [`store_file_object`] writes the chunk Blobs under these same ids,
/// so every path agrees on the manifest — and thus on its merkle root.
///
/// # Errors
/// [`WorktreeError::Object`] if a chunk length exceeds the wire-format cap.
pub fn chunked_blob_from_bytes(data: &[u8]) -> WorktreeResult<ChunkedBlob> {
    let chunks: Vec<Hash> = ChunkIterator::new(FastCdc::v1(), data)
        .map(|b| {
            let chunk = &data[b.offset..b.offset + b.length];
            // Blob object id = BLAKE3(blob_prologue ‖ chunk) — the id
            // store_file_object writes this chunk Blob under (the prologue ‖
            // payload split is pinned to serialize(Blob) by proptest).
            let prologue = serialize::blob_prologue(chunk.len())?;
            let mut hasher = crate::hash::Hasher::new();
            hasher.update(&prologue);
            hasher.update(chunk);
            Ok::<_, WorktreeError>(hasher.finalize())
        })
        .collect::<Result<_, _>>()?;
    Ok(ChunkedBlob {
        total_size: data.len() as u64,
        chunk_size: 0,
        chunks,
    })
}

/// Content-address `data` exactly as [`store_file_object`] would,
/// **without storing anything**. Backs change detection (`status`, `rm`,
/// restore safety checks) where only the answer "would this file hash to X?"
/// is needed — writing objects there would turn a read-only query into store
/// mutation. Equivalence with `store_file_object` is pinned by test.
///
/// # Errors
/// [`WorktreeError::Object`] if a length exceeds the wire-format cap.
pub fn hash_file_object(data: &[u8]) -> WorktreeResult<Hash> {
    if u64::try_from(data.len()).unwrap_or(u64::MAX) <= CHUNK_THRESHOLD {
        let prologue = serialize::blob_prologue(data.len())?;
        let mut hasher = crate::hash::Hasher::new();
        hasher.update(&prologue);
        hasher.update(data);
        return Ok(hasher.finalize());
    }
    // A ChunkedBlob is addressed by its merkle BMT root, matching what the
    // sink stores it under. This read-only mirror must agree with the write
    // path or change detection breaks.
    Ok(crate::merkle::compute_chunked_id(&chunked_blob_from_bytes(
        data,
    )?))
}

/// A file's mtime as nanoseconds since the Unix epoch, saturating; `0`
/// (the "no cache" sentinel) when the mtime is unavailable or predates
/// the epoch.
#[must_use]
pub fn mtime_nanos(meta: &fs::Metadata) -> u64 {
    meta.modified()
        .ok()
        .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
        .map_or(0, |d| u64::try_from(d.as_nanos()).unwrap_or(u64::MAX))
}

/// The full stat-cache observation for `meta`, in index-entry field
/// order: `(mtime_ns, size, ino, ctime_ns)`. The single producer-side
/// dual of [`stat_matches`] — every site that records the cache uses
/// this so the recorded and compared field sets can never drift.
/// `ino`/`ctime_ns` are 0 (= don't check) on platforms without them.
#[must_use]
pub fn stat_cache_fields(meta: &fs::Metadata) -> (u64, u64, u64, u64) {
    #[cfg(unix)]
    let (ino, ctime_ns) = {
        use std::os::unix::fs::MetadataExt;
        let ctime_ns = u64::try_from(meta.ctime())
            .ok()
            .and_then(|s| s.checked_mul(1_000_000_000))
            .and_then(|ns| ns.checked_add(u64::try_from(meta.ctime_nsec()).unwrap_or(0)))
            .unwrap_or(0);
        (meta.ino(), ctime_ns)
    };
    #[cfg(not(unix))]
    let (ino, ctime_ns) = (0u64, 0u64);
    (mtime_nanos(meta), meta.len(), ino, ctime_ns)
}

/// True iff `meta` proves the worktree file behind `entry` is
/// byte-identical to `entry.object_hash` without reading it: the cached
/// mtime is nonzero (cache present, not racy-smudged) and equal, the
/// size is equal, the inode and ctime match when recorded (catching
/// replace-by-rename and `touch -r`-style timestamp restoration —
/// ctime cannot be set from userspace), and the live mode's exec class
/// matches the staged status. Symlink entries never stat-match — the
/// target re-read is cheap and `meta` semantics differ.
#[must_use]
pub fn stat_matches(entry: &crate::index::IndexEntry, meta: &fs::Metadata) -> bool {
    use crate::index::EntryStatus;
    if entry.mtime_ns == 0 || !meta.is_file() {
        return false;
    }
    let (mtime_ns, size, ino, ctime_ns) = stat_cache_fields(meta);
    if size != entry.size || mtime_ns != entry.mtime_ns {
        return false;
    }
    // ino/ctime: compare only when both sides have a value — a v2 entry
    // recorded on a platform without them (0) stays usable elsewhere.
    if entry.ino != 0 && ino != 0 && ino != entry.ino {
        return false;
    }
    if entry.ctime_ns != 0 && ctime_ns != 0 && ctime_ns != entry.ctime_ns {
        return false;
    }
    match entry.status {
        // On non-unix the exec bit is not observable in the filesystem
        // mode, so the recorded status is the source of truth — both
        // classes stat-match (previously Executable entries could never
        // match there, silently defeating the cache).
        #[cfg(not(unix))]
        EntryStatus::Blob | EntryStatus::Executable => true,
        #[cfg(unix)]
        EntryStatus::Blob => entry_mode_from_file_metadata(meta) == EntryMode::Blob,
        #[cfg(unix)]
        EntryStatus::Executable => entry_mode_from_file_metadata(meta) == EntryMode::Executable,
        EntryStatus::Symlink | EntryStatus::Removed | EntryStatus::Tree => false,
    }
}

#[cfg(unix)]
fn open_regular_file(path: &Path) -> io::Result<fs::File> {
    use std::os::unix::fs::OpenOptionsExt;

    fs::OpenOptions::new()
        .read(true)
        .custom_flags(libc::O_NOFOLLOW)
        .open(path)
}

#[cfg(not(unix))]
fn open_regular_file(path: &Path) -> io::Result<fs::File> {
    // Best-effort direct-symlink rejection on platforms without the
    // Unix O_NOFOLLOW path. This does not close the swap race, but it
    // keeps normal symlinks from being treated as regular files.
    let meta = path.symlink_metadata()?;
    if !meta.file_type().is_file() {
        return Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            "path is not a regular file",
        ));
    }
    fs::File::open(path)
}

#[cfg(unix)]
fn entry_mode_from_file_metadata(meta: &fs::Metadata) -> EntryMode {
    use std::os::unix::fs::PermissionsExt;

    if meta.permissions().mode() & 0o111 != 0 {
        EntryMode::Executable
    } else {
        EntryMode::Blob
    }
}

#[cfg(not(unix))]
fn entry_mode_from_file_metadata(_meta: &fs::Metadata) -> EntryMode {
    EntryMode::Blob
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::object::ObjectType;
    use tempfile::TempDir;

    fn fresh_store() -> (TempDir, ObjectStore) {
        let dir = TempDir::new().unwrap();
        let store = ObjectStore::init(&crate::layout::RepoLayout::single(dir.path())).unwrap();
        (dir, store)
    }

    #[test]
    fn validate_symlink_targets() {
        assert!(validate_symlink_target("hello"));
        assert!(validate_symlink_target("sub/dir/file"));
        assert!(!validate_symlink_target(""));
        assert!(!validate_symlink_target("/etc/passwd"));
        assert!(!validate_symlink_target("../escape"));
        assert!(!validate_symlink_target("a/../b"));
    }

    #[test]
    fn build_tree_from_empty_dir() {
        let (_sd, store) = fresh_store();
        let work = TempDir::new().unwrap();
        let h = build_tree(&store, work.path()).unwrap();
        let obj = store.read_object(&h).unwrap();
        match obj {
            Object::Tree(t) => assert_eq!(t.entries.len(), 0),
            other => panic!("expected tree, got {other:?}"),
        }
    }

    #[test]
    fn build_tree_with_single_file() {
        let (_sd, store) = fresh_store();
        let work = TempDir::new().unwrap();
        fs::write(work.path().join("hello.txt"), b"hello world").unwrap();
        let h = build_tree(&store, work.path()).unwrap();
        let obj = store.read_object(&h).unwrap();
        let Object::Tree(t) = obj else {
            panic!("expected tree");
        };
        assert_eq!(t.entries.len(), 1);
        assert_eq!(t.entries[0].name.as_slice(), b"hello.txt");
        assert_eq!(t.entries[0].mode, EntryMode::Blob);
        let blob_obj = store.read_object(&t.entries[0].object_hash).unwrap();
        let Object::Blob(b) = blob_obj else {
            panic!("expected blob");
        };
        assert_eq!(b.data, b"hello world");
    }

    #[cfg(unix)]
    #[test]
    fn build_tree_marks_executable_regular_files() {
        use std::os::unix::fs::PermissionsExt;

        let (_sd, store) = fresh_store();
        let work = TempDir::new().unwrap();
        let script = work.path().join("run.sh");
        fs::write(&script, b"#!/bin/sh\n").unwrap();
        let mut perms = fs::metadata(&script).unwrap().permissions();
        perms.set_mode(perms.mode() | 0o111);
        fs::set_permissions(&script, perms).unwrap();

        let h = build_tree(&store, work.path()).unwrap();
        let Object::Tree(t) = store.read_object(&h).unwrap() else {
            panic!("expected tree");
        };
        assert_eq!(t.entries[0].name.as_slice(), b"run.sh");
        assert_eq!(t.entries[0].mode, EntryMode::Executable);
    }

    #[cfg(unix)]
    #[test]
    fn build_tree_rejects_invalid_entry_name_before_writing_tree() {
        let (_sd, store) = fresh_store();
        let work = TempDir::new().unwrap();
        fs::write(work.path().join("bad."), b"bad name").unwrap();

        let err = build_tree(&store, work.path()).unwrap_err();
        assert!(matches!(err, WorktreeError::Io(_)));
    }

    #[cfg(unix)]
    #[test]
    fn hash_file_rejects_final_component_symlink() {
        use std::os::unix::fs::symlink;

        let (_sd, store) = fresh_store();
        let work = TempDir::new().unwrap();
        fs::write(work.path().join("target.txt"), b"target").unwrap();
        symlink("target.txt", work.path().join("link.txt")).unwrap();

        let err = hash_file(&store, &work.path().join("link.txt")).unwrap_err();
        assert!(matches!(err, WorktreeError::Io(_)));
    }

    #[test]
    fn build_tree_with_nested_directories() {
        let (_sd, store) = fresh_store();
        let work = TempDir::new().unwrap();
        fs::write(work.path().join("a.txt"), b"file a").unwrap();
        fs::create_dir(work.path().join("subdir")).unwrap();
        fs::write(work.path().join("subdir/b.txt"), b"file b").unwrap();
        let h = build_tree(&store, work.path()).unwrap();
        let obj = store.read_object(&h).unwrap();
        let Object::Tree(t) = obj else {
            panic!("expected tree");
        };
        assert_eq!(t.entries.len(), 2);
        // Sorted lex: a.txt first, subdir second.
        assert_eq!(t.entries[0].name.as_slice(), b"a.txt");
        assert_eq!(t.entries[1].name.as_slice(), b"subdir");
        assert_eq!(t.entries[1].mode, EntryMode::Tree);
        let sub = store.read_object(&t.entries[1].object_hash).unwrap();
        let Object::Tree(st) = sub else {
            panic!("expected tree");
        };
        assert_eq!(st.entries.len(), 1);
        assert_eq!(st.entries[0].name.as_slice(), b"b.txt");
    }

    #[test]
    fn build_tree_skips_mkit_directory() {
        let (_sd, store) = fresh_store();
        let work = TempDir::new().unwrap();
        fs::create_dir(work.path().join(".mkit")).unwrap();
        fs::write(work.path().join(".mkit/should_skip"), b"").unwrap();
        fs::write(work.path().join("keep.txt"), b"kept").unwrap();
        let h = build_tree(&store, work.path()).unwrap();
        let obj = store.read_object(&h).unwrap();
        let Object::Tree(t) = obj else {
            panic!("expected tree");
        };
        assert_eq!(t.entries.len(), 1);
        assert_eq!(t.entries[0].name.as_slice(), b"keep.txt");
    }

    #[test]
    fn build_tree_is_deterministic() {
        let (_sd, store) = fresh_store();
        let work = TempDir::new().unwrap();
        fs::write(work.path().join("z.txt"), b"z").unwrap();
        fs::write(work.path().join("a.txt"), b"a").unwrap();
        let h1 = build_tree(&store, work.path()).unwrap();
        let h2 = build_tree(&store, work.path()).unwrap();
        assert_eq!(h1, h2);
    }

    #[test]
    fn build_tree_respects_mkitignore() {
        let (_sd, store) = fresh_store();
        let work = TempDir::new().unwrap();
        fs::write(work.path().join(".mkitignore"), b"*.log\n").unwrap();
        fs::write(work.path().join("keep.txt"), b"kept").unwrap();
        fs::write(work.path().join("debug.log"), b"ignored").unwrap();
        let h = build_tree(&store, work.path()).unwrap();
        let obj = store.read_object(&h).unwrap();
        let Object::Tree(t) = obj else {
            panic!("expected tree");
        };
        // .mkitignore + keep.txt, but not debug.log.
        assert_eq!(t.entries.len(), 2);
        assert_eq!(t.entries[0].name.as_slice(), b".mkitignore");
        assert_eq!(t.entries[1].name.as_slice(), b"keep.txt");
    }

    #[cfg(unix)]
    #[test]
    fn rejects_invalid_symlink_targets() {
        use std::os::unix::fs::symlink;
        let (_sd, store) = fresh_store();
        let work = TempDir::new().unwrap();
        symlink("/etc/passwd", work.path().join("bad-link")).unwrap();
        let err = build_tree(&store, work.path()).unwrap_err();
        assert!(matches!(err, WorktreeError::InvalidSymlinkTarget(_)));
    }

    #[cfg(unix)]
    #[test]
    fn rejects_dotdot_symlink_targets() {
        use std::os::unix::fs::symlink;
        let (_sd, store) = fresh_store();
        let work = TempDir::new().unwrap();
        symlink("../../etc/passwd", work.path().join("bad-link")).unwrap();
        let err = build_tree(&store, work.path()).unwrap_err();
        assert!(matches!(err, WorktreeError::InvalidSymlinkTarget(_)));
    }

    #[test]
    fn small_file_stays_as_regular_blob() {
        let (_sd, store) = fresh_store();
        let work = TempDir::new().unwrap();
        fs::write(work.path().join("small.txt"), b"hello world").unwrap();
        let h = build_tree(&store, work.path()).unwrap();
        let obj = store.read_object(&h).unwrap();
        let Object::Tree(t) = obj else {
            panic!("expected tree");
        };
        let entry = store.read_object(&t.entries[0].object_hash).unwrap();
        assert_eq!(entry.object_type(), ObjectType::Blob);
    }

    #[test]
    fn large_file_becomes_chunked_blob() {
        // File > CHUNK_THRESHOLD should land as a ChunkedBlob manifest
        // pointing at one Blob per FastCDC chunk. We pseudo-randomize
        // the buffer so FastCDC sees real boundary candidates instead
        // of running the entire file as one max-sized chunk.
        let (_sd, store) = fresh_store();
        let work = TempDir::new().unwrap();
        let n = usize::try_from(CHUNK_THRESHOLD).unwrap() + 256 * 1024;
        let mut big = Vec::with_capacity(n);
        let mut state: u64 = 0x00C0_FFEE;
        for _ in 0..n {
            // splitmix64-ish; same construction as the gear table seed.
            state = state.wrapping_add(0x9E37_79B9_7F4A_7C15);
            let mut z = state;
            z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
            z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
            z ^= z >> 31;
            big.push((z & 0xFF) as u8);
        }
        fs::write(work.path().join("big.bin"), &big).unwrap();

        let tree_hash = build_tree(&store, work.path()).unwrap();
        let Object::Tree(t) = store.read_object(&tree_hash).unwrap() else {
            panic!("expected tree");
        };
        assert_eq!(t.entries.len(), 1);

        let entry_hash = t.entries[0].object_hash;
        let entry = store.read_object(&entry_hash).unwrap();
        let Object::ChunkedBlob(manifest) = entry else {
            panic!("expected chunked_blob, got {entry:?}");
        };

        assert_eq!(manifest.total_size, n as u64);
        assert_eq!(manifest.chunk_size, 0, "0 = content-defined (FastCDC)");
        assert!(!manifest.chunks.is_empty());
        // Every chunk hash must resolve to a Blob in the store, and
        // the concatenation must reproduce the original file bytes.
        let mut reassembled: Vec<u8> = Vec::with_capacity(n);
        for h in &manifest.chunks {
            let Object::Blob(b) = store.read_object(h).unwrap() else {
                panic!("chunk did not resolve to a Blob");
            };
            reassembled.extend_from_slice(&b.data);
        }
        assert_eq!(reassembled, big, "chunks must round-trip the source");
    }

    #[test]
    fn hash_file_with_metadata_streaming_matches_store_file_object_in_memory() {
        // The streaming disk path (issue #828) must be content-address
        // equivalent to the original whole-buffer path for the same
        // bytes — same test data/PRNG shape as `large_file_becomes_chunked_blob`.
        let work = TempDir::new().unwrap();
        let n = usize::try_from(CHUNK_THRESHOLD).unwrap() + 700 * 1024;
        let mut big = Vec::with_capacity(n);
        let mut state: u64 = 0xFACE_FEED;
        for _ in 0..n {
            state = state.wrapping_add(0x9E37_79B9_7F4A_7C15);
            let mut z = state;
            z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
            z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
            z ^= z >> 31;
            big.push((z & 0xFF) as u8);
        }
        let path = work.path().join("big.bin");
        fs::write(&path, &big).unwrap();

        let (_sd1, store1) = fresh_store();
        let (streamed_hash, meta) = hash_file_with_metadata(&store1, &path).unwrap();
        assert_eq!(meta.len(), n as u64);

        let (_sd2, store2) = fresh_store();
        let in_memory_hash = store_file_object(&store2, &big).unwrap();

        assert_eq!(
            streamed_hash, in_memory_hash,
            "streaming a file from disk must produce the same content-address \
             as chunking the fully-buffered bytes"
        );

        // And the objects it actually wrote must round-trip.
        let Object::ChunkedBlob(manifest) = store1.read_object(&streamed_hash).unwrap() else {
            panic!("expected chunked_blob");
        };
        let mut reassembled = Vec::with_capacity(n);
        for h in &manifest.chunks {
            let Object::Blob(b) = store1.read_object(h).unwrap() else {
                panic!("chunk did not resolve to a Blob");
            };
            reassembled.extend_from_slice(&b.data);
        }
        assert_eq!(reassembled, big);
    }

    #[test]
    fn streaming_rejects_miscounted_hashes_in_full_and_partial_batches() {
        for size in [
            crate::chunker::MAX_SIZE,
            STREAM_HASH_BATCH * crate::chunker::MAX_SIZE + 1,
        ] {
            let (_dir, sink) = fresh_store();
            let data = vec![0u8; size];
            let mut expected = 0;
            let err = store_large_file_streaming_with(
                &sink,
                io::Cursor::new(data),
                Path::new("batch.bin"),
                |_sink, batch| {
                    expected = batch.len();
                    Ok(Vec::new())
                },
            )
            .unwrap_err();
            assert!(expected > 0);
            assert!(matches!(err, WorktreeError::ChunkBatchLengthMismatch {
                expected: n, actual: 0
            } if n == expected));
        }
    }

    #[test]
    fn streaming_pipeline_errors_promptly_when_hash_chunks_fails_on_an_early_batch() {
        // Regression test for a real deadlock in the native pipelined
        // path (`store_large_file_streaming_pipelined`): on a rendezvous
        // channel, if the receiving side stops calling `recv` as soon as
        // it sees an error, the cutter thread's *next* `send` — for a
        // batch it already cut, or is about to — has no buffer to land
        // in and blocks forever waiting for a `recv` that will never
        // come, hanging `thread::scope`'s join on that thread. Three
        // full batches (well beyond `STREAM_HASH_BATCH`) ensures the
        // cutter still has more batches to send after the one that
        // triggers the very first error. Runs the call on its own
        // thread with a bounded `recv_timeout` so a regression fails
        // this test instead of hanging the whole suite.
        let size = 3 * STREAM_HASH_BATCH * crate::chunker::MAX_SIZE;
        let (_dir, sink) = fresh_store();
        let data = vec![0u8; size];

        let (done_tx, done_rx) = std::sync::mpsc::channel();
        std::thread::spawn(move || {
            let result = store_large_file_streaming_with(
                &sink,
                io::Cursor::new(data),
                Path::new("batch.bin"),
                |_sink, _batch| Err(WorktreeError::InvalidUtf8),
            );
            let _ = done_tx.send(matches!(result, Err(WorktreeError::InvalidUtf8)));
        });

        let errored_correctly = done_rx
            .recv_timeout(std::time::Duration::from_secs(20))
            .expect(
                "store_large_file_streaming_with deadlocked instead of \
                 returning promptly after an early hash_chunks error",
            );
        assert!(errored_correctly);
    }

    #[test]
    fn hash_file_with_metadata_with_batch_fanout_matches_sequential() {
        // A `hash_chunks` callback that reorders its own work internally
        // (simulating a rayon fan-out that hashes a batch out of order)
        // must still produce the same content-address as the sequential
        // default, as long as it returns results in input order — this
        // pins the `_with` contract `hash_pending` (mkit-cli) relies on.
        // Large enough for several `STREAM_HASH_BATCH`-sized (64-chunk)
        // batches at the ~64 KiB average chunk size.
        let n = usize::try_from(CHUNK_THRESHOLD).unwrap() + 6 * 1024 * 1024;
        let mut big = Vec::with_capacity(n);
        let mut state: u64 = 0xABCD_EF01;
        for _ in 0..n {
            state = state.wrapping_add(0x9E37_79B9_7F4A_7C15);
            let mut z = state;
            z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
            z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
            z ^= z >> 31;
            big.push((z & 0xFF) as u8);
        }
        let work = TempDir::new().unwrap();
        let path = work.path().join("big.bin");
        fs::write(&path, &big).unwrap();

        let (_sd1, sequential_store) = fresh_store();
        let (sequential_hash, _) = hash_file_with_metadata(&sequential_store, &path).unwrap();

        let (_sd2, fanout_store) = fresh_store();
        let (fanout_hash, _) = hash_file_with_metadata_with(&fanout_store, &path, |sink, batch| {
            // Hash in reverse, then un-reverse before returning — proves
            // the contract cares about the *returned* order, not the
            // order chunks are actually processed in.
            let mut out: Vec<Hash> = batch
                .iter()
                .rev()
                .map(|chunk| store_chunk_blob(sink, chunk))
                .collect::<WorktreeResult<_>>()?;
            out.reverse();
            Ok(out)
        })
        .unwrap();

        assert_eq!(
            sequential_hash, fanout_hash,
            "an out-of-order-processing hash_chunks callback must still match \
             the sequential default when it returns results in input order"
        );
    }

    // ---- build_tree_from_index — the staging-area path -------------

    use crate::index::{EntryStatus, Index, IndexEntry};

    fn write_blob(store: &ObjectStore, bytes: &[u8]) -> Hash {
        let blob = Object::Blob(crate::object::Blob {
            data: bytes.to_vec(),
        });
        let body = serialize::serialize(&blob).unwrap();
        store.write(&body).unwrap()
    }

    #[test]
    fn from_index_empty_returns_empty_tree() {
        let (_sd, store) = fresh_store();
        let idx = Index::new();
        let h = build_tree_from_index(&store, &idx).unwrap();
        let Object::Tree(t) = store.read_object(&h).unwrap() else {
            panic!("expected tree");
        };
        assert!(t.entries.is_empty());
    }

    #[test]
    fn from_index_single_file_at_root() {
        let (_sd, store) = fresh_store();
        let blob_hash = write_blob(&store, b"hello world");
        let mut idx = Index::new();
        idx.entries.push(IndexEntry {
            path: "hello.txt".into(),
            status: EntryStatus::Blob,
            object_hash: blob_hash,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        let h = build_tree_from_index(&store, &idx).unwrap();
        let Object::Tree(t) = store.read_object(&h).unwrap() else {
            panic!();
        };
        assert_eq!(t.entries.len(), 1);
        assert_eq!(t.entries[0].name, b"hello.txt");
        assert_eq!(t.entries[0].mode, EntryMode::Blob);
        assert_eq!(t.entries[0].object_hash, blob_hash);
    }

    #[test]
    fn from_index_nested_paths_build_subtrees() {
        let (_sd, store) = fresh_store();
        let a = write_blob(&store, b"file a");
        let b = write_blob(&store, b"file b");
        let mut idx = Index::new();
        idx.entries.push(IndexEntry {
            path: "a.txt".into(),
            status: EntryStatus::Blob,
            object_hash: a,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        idx.entries.push(IndexEntry {
            path: "subdir/b.txt".into(),
            status: EntryStatus::Blob,
            object_hash: b,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        let root_hash = build_tree_from_index(&store, &idx).unwrap();
        let Object::Tree(root) = store.read_object(&root_hash).unwrap() else {
            panic!();
        };
        assert_eq!(root.entries.len(), 2);
        assert_eq!(root.entries[0].name, b"a.txt");
        assert_eq!(root.entries[0].mode, EntryMode::Blob);
        assert_eq!(root.entries[1].name, b"subdir");
        assert_eq!(root.entries[1].mode, EntryMode::Tree);

        let Object::Tree(sub) = store.read_object(&root.entries[1].object_hash).unwrap() else {
            panic!();
        };
        assert_eq!(sub.entries.len(), 1);
        assert_eq!(sub.entries[0].name, b"b.txt");
        assert_eq!(sub.entries[0].object_hash, b);
    }

    #[test]
    fn from_index_removed_entries_are_skipped() {
        let (_sd, store) = fresh_store();
        let a = write_blob(&store, b"keep me");
        let mut idx = Index::new();
        idx.entries.push(IndexEntry {
            path: "keep.txt".into(),
            status: EntryStatus::Blob,
            object_hash: a,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        idx.entries.push(IndexEntry {
            path: "drop.txt".into(),
            status: EntryStatus::Removed,
            object_hash: [0; 32],
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        let h = build_tree_from_index(&store, &idx).unwrap();
        let Object::Tree(t) = store.read_object(&h).unwrap() else {
            panic!();
        };
        assert_eq!(t.entries.len(), 1);
        assert_eq!(t.entries[0].name, b"keep.txt");
    }

    #[test]
    fn from_index_executable_and_symlink_modes_pass_through() {
        let (_sd, store) = fresh_store();
        let exec = write_blob(&store, b"#!/bin/sh");
        let link = write_blob(&store, b"target.txt");
        let mut idx = Index::new();
        idx.entries.push(IndexEntry {
            path: "run.sh".into(),
            status: EntryStatus::Executable,
            object_hash: exec,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        idx.entries.push(IndexEntry {
            path: "link".into(),
            status: EntryStatus::Symlink,
            object_hash: link,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        let h = build_tree_from_index(&store, &idx).unwrap();
        let Object::Tree(t) = store.read_object(&h).unwrap() else {
            panic!();
        };
        let by_name: std::collections::HashMap<&[u8], &TreeEntry> =
            t.entries.iter().map(|e| (e.name.as_slice(), e)).collect();
        assert_eq!(by_name[&b"run.sh"[..]].mode, EntryMode::Executable);
        assert_eq!(by_name[&b"link"[..]].mode, EntryMode::Symlink);
    }

    #[test]
    fn from_index_entries_are_sorted_by_name() {
        let (_sd, store) = fresh_store();
        let a = write_blob(&store, b"x");
        let mut idx = Index::new();
        // Insert out-of-order; the on-disk Tree must still be sorted
        // (SPEC-OBJECTS §4 normative).
        idx.entries.push(IndexEntry {
            path: "z.txt".into(),
            status: EntryStatus::Blob,
            object_hash: a,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        idx.entries.push(IndexEntry {
            path: "a.txt".into(),
            status: EntryStatus::Blob,
            object_hash: a,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        idx.entries.push(IndexEntry {
            path: "m.txt".into(),
            status: EntryStatus::Blob,
            object_hash: a,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        let h = build_tree_from_index(&store, &idx).unwrap();
        let Object::Tree(t) = store.read_object(&h).unwrap() else {
            panic!();
        };
        let names: Vec<&[u8]> = t.entries.iter().map(|e| e.name.as_slice()).collect();
        assert_eq!(names, vec![&b"a.txt"[..], b"m.txt", b"z.txt"]);
    }

    #[test]
    fn from_index_rejects_trailing_slash() {
        let (_sd, store) = fresh_store();
        let h = write_blob(&store, b"x");
        let mut idx = Index::new();
        idx.entries.push(IndexEntry {
            path: "dir/".into(),
            status: EntryStatus::Blob,
            object_hash: h,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        let err = build_tree_from_index(&store, &idx).unwrap_err();
        assert!(matches!(err, WorktreeError::Io(_)));
    }

    #[test]
    fn from_index_rejects_empty_segment() {
        let (_sd, store) = fresh_store();
        let h = write_blob(&store, b"x");
        let mut idx = Index::new();
        idx.entries.push(IndexEntry {
            path: "a//b.txt".into(),
            status: EntryStatus::Blob,
            object_hash: h,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        let err = build_tree_from_index(&store, &idx).unwrap_err();
        assert!(matches!(err, WorktreeError::Io(_)));
    }

    #[test]
    fn from_index_rejects_reserved_name() {
        let (_sd, store) = fresh_store();
        let h = write_blob(&store, b"x");
        let mut idx = Index::new();
        // ".mkit" is rejected by TreeEntry::validate_name as repo
        // metadata aliasing.
        idx.entries.push(IndexEntry {
            path: ".mkit".into(),
            status: EntryStatus::Blob,
            object_hash: h,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        let err = build_tree_from_index(&store, &idx).unwrap_err();
        assert!(matches!(err, WorktreeError::Io(_)));
    }

    /// The most important invariant: for a worktree whose contents
    /// match the index entry-for-entry, `build_tree` and
    /// `build_tree_from_index` MUST produce the identical root hash.
    /// If this drifts, attestations signed under one path won't
    /// verify against trees built under the other.
    #[test]
    fn from_index_matches_build_tree_for_equivalent_worktree() {
        let (_sd, store) = fresh_store();

        // Build the same content two ways:
        //   1. drop files on disk, call build_tree.
        //   2. write blobs to the store directly, populate an index,
        //      call build_tree_from_index.
        let work = TempDir::new().unwrap();
        fs::write(work.path().join("a.txt"), b"alpha").unwrap();
        fs::create_dir(work.path().join("dir")).unwrap();
        fs::write(work.path().join("dir/b.txt"), b"beta").unwrap();
        fs::write(work.path().join("dir/c.txt"), b"gamma").unwrap();
        let worktree_root = build_tree(&store, work.path()).unwrap();

        let a = write_blob(&store, b"alpha");
        let b = write_blob(&store, b"beta");
        let c = write_blob(&store, b"gamma");
        let mut idx = Index::new();
        idx.entries.push(IndexEntry {
            path: "a.txt".into(),
            status: EntryStatus::Blob,
            object_hash: a,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        idx.entries.push(IndexEntry {
            path: "dir/b.txt".into(),
            status: EntryStatus::Blob,
            object_hash: b,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        idx.entries.push(IndexEntry {
            path: "dir/c.txt".into(),
            status: EntryStatus::Blob,
            object_hash: c,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        let index_root = build_tree_from_index(&store, &idx).unwrap();

        assert_eq!(
            worktree_root, index_root,
            "build_tree_from_index must produce the same root hash as build_tree for equivalent contents"
        );
    }

    #[test]
    fn from_index_deeply_nested_paths_build_chain_of_subtrees() {
        let (_sd, store) = fresh_store();
        let h = write_blob(&store, b"deep");
        let mut idx = Index::new();
        idx.entries.push(IndexEntry {
            path: "a/b/c/d/e.txt".into(),
            status: EntryStatus::Blob,
            object_hash: h,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        let root = build_tree_from_index(&store, &idx).unwrap();
        let Object::Tree(t) = store.read_object(&root).unwrap() else {
            panic!();
        };
        assert_eq!(t.entries.len(), 1);
        assert_eq!(t.entries[0].name, b"a");
        assert_eq!(t.entries[0].mode, EntryMode::Tree);
        // Walk down to the leaf.
        let mut cursor = t.entries[0].object_hash;
        for seg in [b"b" as &[u8], b"c", b"d"] {
            let Object::Tree(t) = store.read_object(&cursor).unwrap() else {
                panic!();
            };
            assert_eq!(t.entries.len(), 1);
            assert_eq!(t.entries[0].name, seg);
            cursor = t.entries[0].object_hash;
        }
        let Object::Tree(t) = store.read_object(&cursor).unwrap() else {
            panic!();
        };
        assert_eq!(t.entries[0].name, b"e.txt");
        assert_eq!(t.entries[0].object_hash, h);
    }

    /// Path-collision: an index that stakes the same name as both a
    /// blob and a directory MUST be rejected. Without the check the
    /// builder would happily emit two `TreeEntries` with name `a`
    /// (one Blob, one Tree), which the deserializer rejects under
    /// its strict ascending-name rule. We catch it earlier with a
    /// clearer error so the user knows which path needs unstaging.
    /// (Reviewer finding 2 on PR #103.)
    #[test]
    fn from_index_rejects_blob_then_subdir_collision() {
        let (_sd, store) = fresh_store();
        let h = write_blob(&store, b"x");
        let mut idx = Index::new();
        idx.entries.push(IndexEntry {
            path: "a".into(),
            status: EntryStatus::Blob,
            object_hash: h,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        idx.entries.push(IndexEntry {
            path: "a/b".into(),
            status: EntryStatus::Blob,
            object_hash: h,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        let err = build_tree_from_index(&store, &idx).unwrap_err();
        let msg = format!("{err}");
        assert!(
            msg.contains("conflict") || msg.contains("collision") || msg.contains("'a'"),
            "expected collision error mentioning the path, got: {msg}"
        );
    }

    /// Same collision in the opposite stage order: subdir entry
    /// staged first, then a blob at the parent.
    #[test]
    fn from_index_rejects_subdir_then_blob_collision() {
        let (_sd, store) = fresh_store();
        let h = write_blob(&store, b"x");
        let mut idx = Index::new();
        idx.entries.push(IndexEntry {
            path: "a/b".into(),
            status: EntryStatus::Blob,
            object_hash: h,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        idx.entries.push(IndexEntry {
            path: "a".into(),
            status: EntryStatus::Blob,
            object_hash: h,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        assert!(build_tree_from_index(&store, &idx).is_err());
    }

    #[test]
    fn from_index_rejects_duplicate_exact_path() {
        let (_sd, store) = fresh_store();
        let a = write_blob(&store, b"a");
        let b = write_blob(&store, b"b");
        let mut idx = Index::new();
        idx.entries.push(IndexEntry {
            path: "same.txt".into(),
            status: EntryStatus::Blob,
            object_hash: a,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        idx.entries.push(IndexEntry {
            path: "same.txt".into(),
            status: EntryStatus::Blob,
            object_hash: b,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });

        let err = build_tree_from_index(&store, &idx).unwrap_err();
        let msg = format!("{err}");
        assert!(msg.contains("duplicate index path"), "got: {msg}");
    }

    #[test]
    fn from_index_rejects_duplicate_removed_and_live_path() {
        let (_sd, store) = fresh_store();
        let h = write_blob(&store, b"live");
        let mut idx = Index::new();
        idx.entries.push(IndexEntry {
            path: "same.txt".into(),
            status: EntryStatus::Removed,
            object_hash: [0; 32],
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        idx.entries.push(IndexEntry {
            path: "same.txt".into(),
            status: EntryStatus::Blob,
            object_hash: h,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });

        let err = build_tree_from_index(&store, &idx).unwrap_err();
        let msg = format!("{err}");
        assert!(msg.contains("duplicate index path"), "got: {msg}");
    }

    /// All-Removed index → empty root tree, NOT an error.
    /// (Reviewer finding 1 on PR #103.) `staged_count()` excludes
    /// Removed entries by design; the tree builder does too. The
    /// resulting empty tree is a valid commit target — applying a
    /// removals-only changeset to a tree that previously contained
    /// those paths produces an empty root.
    #[test]
    fn from_index_all_removed_produces_empty_tree() {
        let (_sd, store) = fresh_store();
        let mut idx = Index::new();
        idx.entries.push(IndexEntry {
            path: "gone.txt".into(),
            status: EntryStatus::Removed,
            object_hash: [0; 32],
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        let h = build_tree_from_index(&store, &idx).unwrap();
        let Object::Tree(t) = store.read_object(&h).unwrap() else {
            panic!();
        };
        assert!(t.entries.is_empty());
    }

    /// Sanity: `ObjectType::Tree` is what we materialise. Pin so a
    /// future enum reshuffle catches us.
    #[test]
    fn from_index_root_is_a_tree_object() {
        let (_sd, store) = fresh_store();
        let idx = Index::new();
        let h = build_tree_from_index(&store, &idx).unwrap();
        let obj = store.read_object(&h).unwrap();
        assert_eq!(obj.object_type(), ObjectType::Tree);
    }

    #[test]
    fn from_index_rejects_missing_blob_object() {
        let (_sd, store) = fresh_store();
        let mut idx = Index::new();
        idx.entries.push(IndexEntry {
            path: "missing.txt".into(),
            status: EntryStatus::Blob,
            object_hash: [42; 32],
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });

        let err = build_tree_from_index(&store, &idx).unwrap_err();
        assert!(matches!(err, WorktreeError::Store(_)));
    }

    #[test]
    fn from_index_rejects_non_blob_object_for_blob_status() {
        let (_sd, store) = fresh_store();
        let tree = Object::Tree(Tree { entries: vec![] });
        let body = serialize::serialize(&tree).unwrap();
        let tree_hash = store.write(&body).unwrap();
        let mut idx = Index::new();
        idx.entries.push(IndexEntry {
            path: "not-a-blob.txt".into(),
            status: EntryStatus::Blob,
            object_hash: tree_hash,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });

        let err = build_tree_from_index(&store, &idx).unwrap_err();
        let msg = format!("{err}");
        assert!(
            msg.contains("non-blob"),
            "expected non-blob index object error, got: {msg}"
        );
    }

    /// A file entry whose object is a `ChunkedBlob` (the canonical
    /// representation for > `CHUNK_THRESHOLD` content) is accepted by the
    /// commit/index tree builder, NOT rejected as "non-blob" (#203). The
    /// resulting tree carries an `EntryMode::Blob` pointing at the
    /// manifest, exactly as `build_tree` produces for a large worktree
    /// file.
    #[test]
    fn from_index_accepts_chunked_blob_for_file_entry() {
        let (_sd, store) = fresh_store();
        // Build a > CHUNK_THRESHOLD file's content and store it via the
        // shared object path (lands as a ChunkedBlob).
        let n = usize::try_from(CHUNK_THRESHOLD).unwrap() + 256 * 1024;
        let mut big = Vec::with_capacity(n);
        let mut state: u64 = 0x00C0_FFEE;
        for _ in 0..n {
            state = state.wrapping_add(0x9E37_79B9_7F4A_7C15);
            let mut z = state;
            z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
            z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
            z ^= z >> 31;
            big.push((z & 0xFF) as u8);
        }
        let chunked_hash = store_file_object(&store, &big).unwrap();
        assert!(
            matches!(
                store.read_object(&chunked_hash).unwrap(),
                Object::ChunkedBlob(_)
            ),
            "fixture must be a ChunkedBlob"
        );

        let mut idx = Index::new();
        idx.entries.push(IndexEntry {
            path: "big.bin".into(),
            status: EntryStatus::Blob,
            object_hash: chunked_hash,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        let root = build_tree_from_index(&store, &idx).unwrap();
        let Object::Tree(t) = store.read_object(&root).unwrap() else {
            panic!("expected tree");
        };
        assert_eq!(t.entries.len(), 1);
        assert_eq!(t.entries[0].name, b"big.bin");
        assert_eq!(t.entries[0].mode, EntryMode::Blob);
        assert_eq!(t.entries[0].object_hash, chunked_hash);
        // Reassembly via the shared helper round-trips the source bytes.
        assert_eq!(read_blob(&store, &chunked_hash).unwrap(), big);
    }

    /// SPEC-OBJECTS §7: "The concatenated length MUST equal `total_size`."
    /// A manifest with valid chunks but a forged `total_size` (stored under
    /// its real merkle id — the forgery changes the id, not its validity as
    /// a store entry) must fail reassembly, not silently return
    /// wrong-length content.
    #[test]
    fn read_blob_rejects_chunked_total_size_mismatch() {
        let (_sd, store) = fresh_store();
        let chunk = serialize::serialize(&Object::Blob(crate::object::Blob {
            data: b"twelve bytes".to_vec(),
        }))
        .unwrap();
        let chunk_hash = store.write(&chunk).unwrap();
        let manifest = Object::ChunkedBlob(ChunkedBlob {
            total_size: 999,
            chunk_size: 0,
            chunks: vec![chunk_hash],
        });
        let h = store
            .write(&serialize::serialize(&manifest).unwrap())
            .unwrap();
        let err = read_blob(&store, &h).unwrap_err();
        assert!(
            matches!(
                err,
                WorktreeError::Object(crate::object::MkitError::ChunkedBlobSizeMismatch {
                    expected: 999,
                    actual: 12,
                })
            ),
            "expected ChunkedBlobSizeMismatch, got {err:?}"
        );
    }

    /// A symlink entry MUST still address a single `Blob` (its target
    /// path); a `ChunkedBlob` under a symlink entry is rejected.
    #[test]
    fn from_index_rejects_chunked_blob_for_symlink_entry() {
        let (_sd, store) = fresh_store();
        let n = usize::try_from(CHUNK_THRESHOLD).unwrap() + 256 * 1024;
        let big = vec![0xABu8; n];
        let chunked_hash = store_file_object(&store, &big).unwrap();
        let mut idx = Index::new();
        idx.entries.push(IndexEntry {
            path: "link".into(),
            status: EntryStatus::Symlink,
            object_hash: chunked_hash,
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        let err = build_tree_from_index(&store, &idx).unwrap_err();
        assert!(format!("{err}").contains("non-blob"));
    }

    // ---- batched / zero-copy ingest ----------------------------------

    /// Same chunked input through a batch and through the plain store
    /// must produce the identical manifest hash and identical readable
    /// bytes — this transitively pins the zero-copy `put_parts` chunk
    /// path to the golden `ChunkedBlob` vectors.
    #[test]
    fn store_file_object_via_batch_equals_via_store() {
        // 3 MiB of varied bytes: above CHUNK_THRESHOLD, multiple chunks.
        let data: Vec<u8> = (0..3 * 1024 * 1024u32)
            .map(|i| u8::try_from((i.wrapping_mul(2_654_435_761)) % 251).unwrap())
            .collect();

        let (_d1, store1) = fresh_store();
        let h_store = store_file_object(&store1, &data).unwrap();

        let (_d2, store2) = fresh_store();
        let batch = store2.batch();
        let h_batch = store_file_object(&batch, &data).unwrap();
        batch.commit().unwrap();

        assert_eq!(h_store, h_batch, "sink choice must not change hashes");
        assert_eq!(
            read_blob(&store1, &h_store).unwrap(),
            read_blob(&store2, &h_batch).unwrap(),
        );

        // Small (single-blob) shape too.
        let small = b"under the chunk threshold";
        let h1 = store_file_object(&store1, small).unwrap();
        let batch2 = store2.batch();
        let h2 = store_file_object(&batch2, small).unwrap();
        batch2.commit().unwrap();
        assert_eq!(h1, h2);
    }

    /// Committing a staged index must cost exactly one full flush no
    /// matter how many tree objects it materialises.
    #[test]
    fn build_tree_from_index_with_batch_single_flush() {
        use crate::batch::testing::{Ev, RecordingSyncer};
        use crate::index::{EntryStatus, Index, IndexEntry};
        use std::sync::Arc;

        let (_sd, mut store) = fresh_store();
        // Stage 20 files across nested dirs (many tree objects).
        let mut idx = Index::default();
        for i in 0..20 {
            let blob = Object::Blob(crate::object::Blob {
                data: format!("file {i}").into_bytes(),
            });
            let bytes = serialize::serialize(&blob).unwrap();
            let h = store.write(&bytes).unwrap();
            idx.entries.push(IndexEntry {
                status: EntryStatus::Blob,
                object_hash: h,
                path: format!("d{}/sub/f{i}.txt", i % 5),
                mtime_ns: 0,
                size: 0,
                ino: 0,
                ctime_ns: 0,
            });
        }

        let rec = Arc::new(RecordingSyncer::default());
        store.set_syncer(rec.clone());

        let batch = store.batch();
        let tree_h = build_tree_from_index_with(&store, &batch, &idx, true).unwrap();
        batch.commit().unwrap();

        let fulls = rec
            .events()
            .iter()
            .filter(|e| matches!(e, Ev::Full(_)))
            .count();
        assert_eq!(fulls, 2, "tree materialisation flush cost must be constant");
        assert!(store.read_object(&tree_h).is_ok());

        // Equivalence: the per-object path yields the same root hash.
        let (_sd2, store2) = fresh_store();
        for i in 0..20 {
            let blob = Object::Blob(crate::object::Blob {
                data: format!("file {i}").into_bytes(),
            });
            store2.write(&serialize::serialize(&blob).unwrap()).unwrap();
        }
        assert_eq!(tree_h, build_tree_from_index(&store2, &idx).unwrap());
    }

    /// A staged object corrupted after `add` must NOT be publishable: the
    /// verifying path (commit and friends) rejects it, while the cheap
    /// non-verifying path (status/diff snapshots) still accepts the shape.
    #[test]
    fn build_tree_from_index_verify_rejects_corrupt_staged_object() {
        use crate::index::{EntryStatus, Index, IndexEntry};

        let (_sd, store) = fresh_store();
        let blob = Object::Blob(crate::object::Blob {
            data: b"hello".to_vec(),
        });
        let h = store.write(&serialize::serialize(&blob).unwrap()).unwrap();
        let mut idx = Index::default();
        idx.entries.push(IndexEntry {
            status: EntryStatus::Blob,
            object_hash: h,
            path: "a.txt".to_string(),
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });

        // Clean object: both paths succeed and agree.
        assert!(build_tree_from_index_with(&store, &store, &idx, true).is_ok());
        assert!(build_tree_from_index_with(&store, &store, &idx, false).is_ok());

        // Corrupt a payload byte past the 6-byte prologue (the prologue
        // shape stays valid, so only a re-hash can catch it).
        let path = store.path_for(&h);
        let mut bytes = std::fs::read(&path).unwrap();
        let i = bytes.len() - 1;
        bytes[i] ^= 0xFF;
        std::fs::write(&path, &bytes).unwrap();

        // Verifying path refuses to publish the corrupt object…
        assert!(
            build_tree_from_index_with(&store, &store, &idx, true).is_err(),
            "commit-path tree build must reject a corrupt staged object"
        );
        // …but the cheap snapshot path still passes the prologue shape check.
        assert!(
            build_tree_from_index_with(&store, &store, &idx, false).is_ok(),
            "status/diff snapshot path keeps the cheap prologue-only check"
        );
    }

    /// The batched pass-2 staged-object check (`probe_staged_objects`)
    /// must not change *which* error an index with more than one kind
    /// of problem reports — the sequential (default, non-parallel)
    /// path has to surface the same error the original single
    /// interleaved loop would have: whichever entry comes first in
    /// index order, checked dup-path, then status, then staged-object
    /// shape, in that sub-order. Here entry 0's staged hash was never
    /// written (fails the staged-object check) while entry 1 uses the
    /// reserved `Tree` status (fails the pass-1 status check) — entry
    /// 0 comes first, so its error must win even though pass 1 alone
    /// would reach entry 1's problem without ever touching the store.
    #[test]
    fn build_tree_from_index_earlier_object_error_wins_over_later_status_error() {
        use crate::index::{EntryStatus, Index, IndexEntry};

        let (_sd, store) = fresh_store();
        let mut idx = Index::default();
        idx.entries.push(IndexEntry {
            status: EntryStatus::Blob,
            object_hash: [0xAB; 32],
            path: "a.txt".to_string(),
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });
        idx.entries.push(IndexEntry {
            status: EntryStatus::Tree,
            object_hash: [0; 32],
            path: "b".to_string(),
            mtime_ns: 0,
            size: 0,
            ino: 0,
            ctime_ns: 0,
        });

        let err = build_tree_from_index_with(&store, &store, &idx, false)
            .expect_err("index has no valid entries");
        let msg = err.to_string();
        assert!(
            msg.contains("ab") || msg.to_lowercase().contains("not found"),
            "expected entry 0's missing-object error, got: {msg}"
        );
        assert!(
            !msg.contains("reserved Tree status"),
            "entry 1's status error must not preempt entry 0's earlier object error, got: {msg}"
        );
    }

    // ---- pure hashing + stat cache ------------------------------------

    /// `hash_file_object` must agree with `store_file_object` on every
    /// input shape (single blob, chunked) without touching any store.
    #[test]
    fn hash_file_object_equals_store_file_object() {
        let threshold = usize::try_from(CHUNK_THRESHOLD).unwrap();
        for len in [0usize, 1, 1024, threshold, 3 * 1024 * 1024] {
            let data: Vec<u8> = (0..len)
                .map(|i| u8::try_from((i * 31 + 7) % 251).unwrap())
                .collect();
            let (_sd, store) = fresh_store();
            let stored = store_file_object(&store, &data).unwrap();
            let pure = hash_file_object(&data).unwrap();
            assert_eq!(stored, pure, "len {len}: pure hash must match stored hash");
        }
    }

    #[test]
    fn hash_file_object_writes_nothing() {
        let (_sd, store) = fresh_store();
        let data = vec![0xAB; 2 * 1024 * 1024]; // chunked shape
        let _ = hash_file_object(&data).unwrap();
        assert!(
            store.iter_object_hashes().unwrap().is_empty(),
            "pure hashing must not create objects"
        );
    }

    fn meta_of(p: &Path) -> fs::Metadata {
        p.symlink_metadata().unwrap()
    }

    #[test]
    fn stat_matches_requires_nonzero_mtime_and_equal_fields() {
        let work = TempDir::new().unwrap();
        let f = work.path().join("a.txt");
        fs::write(&f, b"hello").unwrap();
        let meta = meta_of(&f);
        let entry = crate::index::IndexEntry {
            path: "a.txt".into(),
            status: crate::index::EntryStatus::Blob,
            object_hash: crate::hash::hash(b"irrelevant"),
            mtime_ns: mtime_nanos(&meta),
            size: meta.len(),
            ino: 0,
            ctime_ns: 0,
        };
        assert!(stat_matches(&entry, &meta));

        // Zero mtime sentinel: never matches.
        let mut zeroed = entry.clone();
        zeroed.mtime_ns = 0;
        assert!(!stat_matches(&zeroed, &meta), "zero sentinel must re-hash");

        // Size mismatch.
        let mut wrong_size = entry.clone();
        wrong_size.size += 1;
        assert!(!stat_matches(&wrong_size, &meta));

        // Mtime mismatch.
        let mut wrong_time = entry.clone();
        wrong_time.mtime_ns ^= 1;
        assert!(!stat_matches(&wrong_time, &meta));
    }

    #[cfg(unix)]
    #[test]
    fn stat_matches_detects_exec_bit_flip() {
        use std::os::unix::fs::PermissionsExt;
        let work = TempDir::new().unwrap();
        let f = work.path().join("run.sh");
        fs::write(&f, b"#!/bin/sh\n").unwrap();
        let meta = meta_of(&f);
        let entry = crate::index::IndexEntry {
            path: "run.sh".into(),
            status: crate::index::EntryStatus::Blob,
            object_hash: crate::hash::hash(b"x"),
            mtime_ns: mtime_nanos(&meta),
            size: meta.len(),
            ino: 0,
            ctime_ns: 0,
        };
        assert!(stat_matches(&entry, &meta));
        // chmod +x without touching content, then restore the mtime so
        // ONLY the mode differs — the exec-class check must still fire.
        let mtime = meta.modified().unwrap();
        fs::set_permissions(&f, fs::Permissions::from_mode(0o755)).unwrap();
        let f_handle = fs::File::options().write(true).open(&f).unwrap();
        f_handle
            .set_times(fs::FileTimes::new().set_modified(mtime))
            .unwrap();
        drop(f_handle);
        let meta2 = meta_of(&f);
        assert_eq!(mtime_nanos(&meta2), entry.mtime_ns, "mtime restored");
        assert!(
            !stat_matches(&entry, &meta2),
            "exec-bit flip must invalidate a Blob-status cache hit"
        );
    }

    /// The killer observable: a stat-matched file is NEVER opened. With
    /// the file made unreadable (chmod 000), the tree build still
    /// succeeds and reuses the staged hash.
    #[cfg(unix)]
    #[test]
    fn build_tree_reuses_hash_on_stat_match_without_reading_file() {
        use std::os::unix::fs::PermissionsExt;
        let (_sd, store) = fresh_store();
        let work = TempDir::new().unwrap();
        let f = work.path().join("locked.txt");
        fs::write(&f, b"cached content").unwrap();

        let staged_hash = store_file_object(&store, b"cached content").unwrap();
        let meta = meta_of(&f);
        let idx = crate::index::Index::from_entries(vec![crate::index::IndexEntry {
            path: "locked.txt".into(),
            status: crate::index::EntryStatus::Blob,
            object_hash: staged_hash,
            mtime_ns: mtime_nanos(&meta),
            size: meta.len(),
            ino: 0,
            ctime_ns: 0,
        }]);

        // Make any read attempt error out.
        fs::set_permissions(&f, fs::Permissions::from_mode(0o000)).unwrap();
        let result = build_tree_filtered(&store, work.path(), Some(&idx));
        fs::set_permissions(&f, fs::Permissions::from_mode(0o644)).unwrap();
        let tree_h = result.expect("stat match must skip the file read");

        let Object::Tree(t) = store.read_object(&tree_h).unwrap() else {
            panic!("expected tree");
        };
        assert_eq!(t.entries.len(), 1);
        assert_eq!(t.entries[0].object_hash, staged_hash);

        // Same content hashed normally yields the identical tree.
        let (_sd2, store2) = fresh_store();
        let f2_dir = TempDir::new().unwrap();
        fs::write(f2_dir.path().join("locked.txt"), b"cached content").unwrap();
        let plain = build_tree(&store2, f2_dir.path()).unwrap();
        assert_eq!(plain, tree_h, "cache hit must not change tree hashes");
    }

    /// `hash_pending_files_parallel`'s fan-out branch only fires once a
    /// single directory has at least `FILE_HASH_MISSES_PER_THREAD *`
    /// the machine's thread count worth of cache-miss files — every
    /// other `build_tree` test in this module stays well under that, so
    /// none of them ever exercise it. 4096 distinct-content files is
    /// comfortably above that threshold on any real machine (including
    /// CI runners with far more than 4 cores), forcing the parallel
    /// branch to run rather than just its sequential fallback.
    ///
    /// Every file's expected content is unique and includes its index,
    /// so a slot/result misalignment in the parallel path (a chunk's
    /// results landing at the wrong base offset, or `entries`/`misses`
    /// getting out of step) would surface here as an entry whose hash
    /// belongs to a different file than its name says.
    #[test]
    fn build_tree_parallel_fanout_assigns_each_hash_to_the_right_file() {
        const N: usize = 4096;

        let (_sd, store) = fresh_store();
        let work = TempDir::new().unwrap();
        for i in 0..N {
            fs::write(
                work.path().join(format!("f{i:04}.txt")),
                format!("content {i}"),
            )
            .unwrap();
        }

        // No index passed: every file is untracked, so every one is a
        // cache miss (never a stat-match short-circuit) regardless of
        // stat-cache behavior.
        let h = build_tree(&store, work.path()).unwrap();
        let Object::Tree(t) = store.read_object(&h).unwrap() else {
            panic!("expected tree");
        };
        assert_eq!(t.entries.len(), N);

        let mut seen_hashes = std::collections::HashSet::with_capacity(N);
        for (i, entry) in t.entries.iter().enumerate() {
            // Tree entries are sorted by name, and zero-padded decimal
            // names sort in the same order as their numeric index.
            let expected_name = format!("f{i:04}.txt");
            assert_eq!(
                entry.name,
                expected_name.as_bytes(),
                "entry {i} out of order"
            );

            let expected_content = format!("content {i}");
            let expected_hash = hash_file_object(expected_content.as_bytes()).unwrap();
            assert_eq!(
                entry.object_hash, expected_hash,
                "entry {i} ({expected_name}) has the wrong hash — parallel slot misalignment?"
            );
            assert!(
                seen_hashes.insert(entry.object_hash),
                "entry {i} repeats another entry's hash (or the ZERO placeholder)"
            );
        }
    }

    /// Replace-by-rename with preserved mtime+size must be caught by
    /// the inode check: the replacement file has a different ino.
    #[cfg(unix)]
    #[test]
    fn stat_mismatch_on_inode_rehashes() {
        let work = TempDir::new().unwrap();
        let f = work.path().join("swap.txt");
        fs::write(&f, b"original").unwrap();
        let meta = meta_of(&f);
        let (mtime_ns, size, ino, ctime_ns) = stat_cache_fields(&meta);
        let entry = crate::index::IndexEntry {
            path: "swap.txt".into(),
            status: crate::index::EntryStatus::Blob,
            object_hash: crate::hash::hash(b"original"),
            mtime_ns,
            size,
            ino,
            ctime_ns,
        };
        assert!(stat_matches(&entry, &meta));

        // Same-size replacement via rename with timestamps restored —
        // the tar -x / rsync -t / mv-of-prepared-file shape.
        let staging = work.path().join(".swap.new");
        fs::write(&staging, b"REPLACED").unwrap(); // same 8-byte size
        let fh = fs::File::options().write(true).open(&staging).unwrap();
        fh.set_times(fs::FileTimes::new().set_modified(meta.modified().unwrap()))
            .unwrap();
        drop(fh);
        fs::rename(&staging, &f).unwrap();
        let meta2 = meta_of(&f);
        assert_eq!(meta2.len(), entry.size, "size preserved by the swap");
        assert!(
            !stat_matches(&entry, &meta2),
            "a renamed-in replacement must not stat-match (ino differs)"
        );
    }

    /// A recorded ctime that disagrees with the live one must miss —
    /// ctime cannot be restored from userspace, so `touch -r` after an
    /// in-place edit is caught even when mtime+size+ino all match.
    #[test]
    fn stat_mismatch_on_ctime_rehashes() {
        let work = TempDir::new().unwrap();
        let f = work.path().join("touched.txt");
        fs::write(&f, b"content").unwrap();
        let meta = meta_of(&f);
        let (mtime_ns, size, ino, ctime_ns) = stat_cache_fields(&meta);
        if ctime_ns == 0 {
            return; // platform without ctime — check not applicable
        }
        let entry = crate::index::IndexEntry {
            path: "touched.txt".into(),
            status: crate::index::EntryStatus::Blob,
            object_hash: crate::hash::hash(b"content"),
            mtime_ns,
            size,
            ino,
            ctime_ns: ctime_ns ^ 1,
        };
        assert!(
            !stat_matches(&entry, &meta),
            "ctime disagreement must invalidate the cache"
        );
    }

    /// The worktree walk must report hash-time observations for entries
    /// whose cache was absent but whose content re-hashed to the staged
    /// hash — and the observation must carry the fd-stat, enabling the
    /// status command to heal the cache soundly.
    #[test]
    fn build_tree_observed_reports_clean_rehashes() {
        let (_sd, store) = fresh_store();
        let work = TempDir::new().unwrap();
        fs::write(work.path().join("clean.txt"), b"clean bytes").unwrap();
        fs::write(work.path().join("dirty.txt"), b"new content").unwrap();

        let clean_hash = store_file_object(&store, b"clean bytes").unwrap();
        let stale_hash = crate::hash::hash(b"old content");
        let idx = crate::index::Index::from_entries(vec![
            crate::index::IndexEntry {
                path: "clean.txt".into(),
                status: crate::index::EntryStatus::Blob,
                object_hash: clean_hash,
                mtime_ns: 0, // racy-smudged: forces a re-hash
                size: 0,
                ino: 0,
                ctime_ns: 0,
            },
            crate::index::IndexEntry {
                path: "dirty.txt".into(),
                status: crate::index::EntryStatus::Blob,
                object_hash: stale_hash,
                mtime_ns: 0,
                size: 0,
                ino: 0,
                ctime_ns: 0,
            },
        ]);
        let mut obs = Vec::new();
        build_tree_filtered_observed(&store, work.path(), Some(&idx), &mut obs).unwrap();

        assert_eq!(obs.len(), 1, "only the verified-clean entry is observed");
        let o = &obs[0];
        assert_eq!(o.path, "clean.txt");
        assert_eq!(o.object_hash, clean_hash);
        let meta = meta_of(&work.path().join("clean.txt"));
        let (mtime_ns, size, _ino, _ctime) = stat_cache_fields(&meta);
        assert_eq!(o.mtime_ns, mtime_ns, "observation carries the fd stat");
        assert_eq!(o.size, size);
    }

    /// A stat MISMATCH must fall back to re-hashing the live content.
    #[test]
    fn build_tree_rehashes_on_stat_mismatch() {
        let (_sd, store) = fresh_store();
        let work = TempDir::new().unwrap();
        let f = work.path().join("changed.txt");
        fs::write(&f, b"new content").unwrap();
        let stale_hash = crate::hash::hash(b"not the real object");
        let meta = meta_of(&f);
        let idx = crate::index::Index::from_entries(vec![crate::index::IndexEntry {
            path: "changed.txt".into(),
            status: crate::index::EntryStatus::Blob,
            object_hash: stale_hash,
            // size deliberately wrong → mismatch → re-hash.
            mtime_ns: mtime_nanos(&meta),
            size: meta.len() + 1,
            ino: 0,
            ctime_ns: 0,
        }]);
        let tree_h = build_tree_filtered(&store, work.path(), Some(&idx)).unwrap();
        let Object::Tree(t) = store.read_object(&tree_h).unwrap() else {
            panic!("expected tree");
        };
        assert_ne!(
            t.entries[0].object_hash, stale_hash,
            "mismatched stat must not reuse the stale hash"
        );
        assert_eq!(
            t.entries[0].object_hash,
            store_file_object(&store, b"new content").unwrap()
        );
    }
}