repo/

repository_materialization.rs

// SPDX-License-Identifier: Apache-2.0
//! Tree materialization helpers.

use std::{
    collections::BTreeSet,
    fs,
    num::NonZeroUsize,
    path::{Path, PathBuf},
    sync::atomic::{AtomicBool, Ordering},
    thread,
    time::Instant,
};

use objects::{
    fs_atomic::{enrich_fs_error, is_directory_not_empty},
    object::{Blob, ContentHash, StateId, Tree, TreeEntryTarget},
    store::ObjectStore,
    util::gitlink_placeholder_bytes,
};
use sley::ObjectId as GitObjectId;
use tracing::{debug, instrument};

use super::{HeddleError, Repository, Result};
use crate::{
    worktree_index::IndexEntry,
    worktree_walk::{build_cached_entry, cache_key, validate_symlink_target},
};

/// State threaded through a single `materialize_write_ops_seeded` call.
/// Tracks whether filesystem-level reflinks (CoW clones) are viable on
/// this destination filesystem, so we don't pay the per-blob
/// `clonefile`/`FICLONE` retry tax once we've seen
/// `EXDEV`/`EOPNOTSUPP`/`ENOSYS` from one of them. Reflink and copy
/// counts are emitted at the end for observability.
///
/// SAFETY/CORRECTNESS NOTE on isolated blobs:
///   We materialize blobs via filesystem-level copy-on-write
///   ("reflink") where supported (`clonefile(2)` on macOS APFS,
///   `ioctl(FICLONE)` on Linux btrfs/XFS-with-reflinks/ZFS), and via
///   `fs::copy` everywhere else. **Both paths give the destination
///   its own inode.** A worktree file is never an alias of the
///   canonical loose blob nor of any other worktree's file — so an
///   agent that runs `chmod +w file && echo new > file` only mutates
///   *that* worktree's bytes. The OS handles the divergence: with a
///   reflink the kernel forks the underlying allocation on first
///   write; with a real copy the dest is a separate file from the
///   start. Either way, no shared-inode hazard exists.
///
///   This replaces an earlier hardlink-plus-`chmod 0o444` defense
///   that turned out to be trivially bypassable. The hardlink made
///   the worktree file an alias of the canonical loose blob; the
///   read-only mode was a soft hint that any agent could (and did)
///   undo with `chmod 644`. The new model is filesystem-level and
///   not bypassable from userspace.
struct MaterializationContext {
    reflink_supported: AtomicBool,
    reflink_count: std::sync::atomic::AtomicUsize,
    copy_count: std::sync::atomic::AtomicUsize,
}

impl MaterializationContext {
    fn new() -> Self {
        Self {
            // Optimistic: try reflink on the first blob; a single
            // `EXDEV`/`EOPNOTSUPP` flips this for the rest of the batch.
            reflink_supported: AtomicBool::new(true),
            reflink_count: std::sync::atomic::AtomicUsize::new(0),
            copy_count: std::sync::atomic::AtomicUsize::new(0),
        }
    }

    fn reflinks_enabled(&self) -> bool {
        self.reflink_supported.load(Ordering::Relaxed)
    }

    fn record_reflink(&self) {
        self.reflink_count.fetch_add(1, Ordering::Relaxed);
    }

    fn record_copy(&self) {
        self.copy_count.fetch_add(1, Ordering::Relaxed);
    }

    /// Disable reflink attempts for the rest of this materialization
    /// after the kernel told us the filesystem won't ever clone.
    fn disable_reflinks(&self) {
        self.reflink_supported.store(false, Ordering::Relaxed);
    }
}

const MATERIALIZE_PARALLEL_THRESHOLD: usize = 32;
const MATERIALIZE_THREADS_ENV: &str = "HEDDLE_MATERIALIZE_THREADS";

struct MaterializationPlan {
    validation_root: PathBuf,
    directories: Vec<PathBuf>,
    directory_contexts: Vec<MaterializedDirectoryContext>,
    leaves: Vec<WorktreeWriteOp>,
    file_count: usize,
    symlink_count: usize,
}

#[derive(Debug)]
pub(crate) struct MaterializedTree {
    pub(crate) file_entries: Vec<SeededWorktreeEntry>,
    pub(crate) directory_contexts: Vec<MaterializedDirectoryContext>,
}

#[derive(Debug)]
pub(crate) struct SeededWorktreeEntry {
    pub(crate) key: String,
    pub(crate) entry: IndexEntry,
}

#[derive(Debug)]
pub(crate) struct MaterializedDirectoryContext {
    pub(crate) key: String,
    pub(crate) path: PathBuf,
    pub(crate) child_names: Vec<String>,
    pub(crate) tree_hash: ContentHash,
}

#[derive(Clone, Debug)]
pub(crate) enum WorktreeWriteOp {
    Blob {
        path: PathBuf,
        hash: ContentHash,
        executable: bool,
    },
    Symlink {
        path: PathBuf,
        hash: ContentHash,
        validation_root: PathBuf,
    },
    GitlinkPlaceholder {
        path: PathBuf,
        target: GitObjectId,
    },
}

impl WorktreeWriteOp {
    pub(crate) fn path(&self) -> &Path {
        match self {
            Self::Blob { path, .. }
            | Self::Symlink { path, .. }
            | Self::GitlinkPlaceholder { path, .. } => path,
        }
    }

    pub(crate) fn hash(&self) -> ContentHash {
        match self {
            Self::Blob { hash, .. } | Self::Symlink { hash, .. } => *hash,
            Self::GitlinkPlaceholder { target, .. } => {
                Blob::new(gitlink_placeholder_bytes(target)).hash()
            }
        }
    }

    pub(crate) fn executable(&self) -> bool {
        match self {
            Self::Blob { executable, .. } => *executable,
            Self::Symlink { .. } | Self::GitlinkPlaceholder { .. } => false,
        }
    }

    pub(crate) fn index_kind(&self) -> crate::worktree_index::IndexEntryKind {
        match self {
            Self::Blob { .. } | Self::GitlinkPlaceholder { .. } => {
                crate::worktree_index::IndexEntryKind::File
            }
            Self::Symlink { .. } => crate::worktree_index::IndexEntryKind::Symlink,
        }
    }
}

/// Result of `Repository::warm_canonical_store_for_state(s)`.
///
/// The reflink-first materializer can only clone from a canonical
/// loose-uncompressed file. After `pack_objects + prune_loose_objects`
/// (the steady state for any non-fresh repo) every blob is pack-only
/// and `loose_blob_path` returns `None`. The warm pass walks a
/// state's tree(s) and promotes every reachable blob in advance so
/// the next N materializations of that state across N worktrees all
/// hit the fast path.
///
/// This is the proactive twin of the lazy promotion that already
/// fires inside `materialize_blob`. Lazy is correct on its own; warm
/// is a latency optimization for the "I'm about to materialize this
/// state to N worktrees" case (e.g. `heddle delegate`).
#[derive(Debug, Default, Clone, Copy)]
pub struct WarmCanonicalStoreStats {
    /// Blobs we wrote to the canonical loose-uncompressed path
    /// because they were either pack-only or compressed-loose.
    pub promoted: usize,
    /// Blobs that were already loose+uncompressed; no work done.
    pub already_loose: usize,
    /// Blobs we tried to promote but `promote_to_loose_uncompressed`
    /// returned an error (e.g. the blob isn't in the store, or a
    /// transient I/O failure during the atomic write). Kept
    /// non-fatal: the lazy path will retry on materialize, and a
    /// real corruption shows up there with a louder error.
    pub errors: usize,
}

impl WarmCanonicalStoreStats {
    /// Total blobs visited.
    pub fn total(&self) -> usize {
        self.promoted + self.already_loose + self.errors
    }
}

impl Repository {
    /// Promote every reachable blob from `state_id`'s tree(s) into
    /// the canonical loose-uncompressed store, so a subsequent
    /// `materialize_tree` (or N parallel materializations) can
    /// reflink from the canonical store without paying the
    /// decompress-on-first-clone tax.
    ///
    /// Returns counts of work done. Errors per blob are accumulated
    /// rather than bubbled up so a single corrupt or missing object
    /// doesn't poison the whole warm pass — the lazy path inside
    /// `materialize_blob` will surface that loudly when it actually
    /// matters.
    #[instrument(skip(self), fields(state_id = %state_id))]
    pub fn warm_canonical_store_for_state(
        &self,
        state_id: &StateId,
    ) -> Result<WarmCanonicalStoreStats> {
        self.warm_canonical_store_for_states(std::slice::from_ref(state_id))
    }

    /// Multi-state variant. Walks each state's tree once, dedupes
    /// the union of reachable blob hashes across all of them, and
    /// promotes them. Useful when materializing several sibling
    /// states from the same parent in quick succession (the
    /// `heddle delegate`-style flow).
    #[instrument(skip(self, state_ids), fields(state_count = state_ids.len()))]
    pub fn warm_canonical_store_for_states(
        &self,
        state_ids: &[StateId],
    ) -> Result<WarmCanonicalStoreStats> {
        let mut blob_hashes = BTreeSet::new();
        for state_id in state_ids {
            let state = self
                .store
                .get_state(state_id)?
                .ok_or_else(|| HeddleError::NotFound(format!("state {} not in store", state_id)))?;
            let tree = self.store.get_tree(&state.tree)?.ok_or_else(|| {
                HeddleError::NotFound(format!("tree {} (for state {})", state.tree, state_id))
            })?;
            self.collect_blob_hashes(&tree, &mut blob_hashes)?;
        }

        let mut stats = WarmCanonicalStoreStats::default();
        for hash in &blob_hashes {
            match self.store.promote_to_loose_uncompressed(hash) {
                Ok(true) => stats.promoted += 1,
                Ok(false) => stats.already_loose += 1,
                Err(err) => {
                    debug!(
                        ?err,
                        hash = %hash,
                        "promote_to_loose_uncompressed failed during warm pass"
                    );
                    stats.errors += 1;
                }
            }
        }

        debug!(
            promoted = stats.promoted,
            already_loose = stats.already_loose,
            errors = stats.errors,
            "Warm canonical store pass complete"
        );

        Ok(stats)
    }

    fn collect_blob_hashes(&self, tree: &Tree, out: &mut BTreeSet<ContentHash>) -> Result<()> {
        for entry in tree.entries() {
            // Symlink targets are stored as blobs too — they're
            // small, so promotion cost is negligible, and a stored
            // symlink is materialized via `get_blob` (not hardlink),
            // so promoting them is technically wasted work. But
            // skipping symlinks would mean walking the tree with
            // the same defensive `is_symlink` guard we use in
            // `plan_materialization`, and the cost of warming a few
            // tiny symlink-target blobs is dwarfed by the
            // decompress cost of even one real source file. Keep
            // it simple: promote everything reachable.
            match entry.target() {
                TreeEntryTarget::Blob { hash, .. } | TreeEntryTarget::Symlink { hash } => {
                    out.insert(*hash);
                }
                TreeEntryTarget::Tree { hash } => {
                    let subtree = self
                        .store
                        .get_tree(hash)?
                        .ok_or_else(|| HeddleError::NotFound(format!("tree {}", hash)))?;
                    self.collect_blob_hashes(&subtree, out)?;
                }
                TreeEntryTarget::Gitlink { .. } => {}
                // Native child-spool edge: carries no local blob/tree hash to
                // promote (its target lives in a separate spool graph).
                TreeEntryTarget::Spoollink { .. } => {}
            }
        }
        Ok(())
    }

    /// Materialize a tree to the filesystem.
    ///
    /// Crate-private on purpose: this blob-keyed primitive carries no
    /// `StateId`/audience, so it cannot apply the visibility gate. External
    /// callers serving a *named committed state* to a checkout must go through
    /// [`Repository::checkout_state_gated`] (gated); callers applying a
    /// *locally-computed* tree (a merge/cherry-pick result) use
    /// [`Repository::materialize_computed_tree`]. Keeping this one inside the
    /// crate is what makes "every state checkout is gated" true by
    /// construction rather than by remembering to add a check (#316 Finding 2).
    #[instrument(skip(self, tree), fields(dir = %dir.display(), entries = tree.len()))]
    pub(crate) fn materialize_tree(&self, tree: &Tree, dir: &Path) -> Result<()> {
        self.materialize_tree_seeded(tree, dir).map(|_| ())
    }

    /// Materialize a *locally-computed* tree to `dir` — a merge or cherry-pick
    /// result that is not a single named committed state and so carries no
    /// audience to gate against. The operator already holds every byte the
    /// computation combined; this is a working-tree write, not a serve to an
    /// audience. For serving a named state to a checkout, use
    /// [`Repository::checkout_state_gated`] instead.
    pub fn materialize_computed_tree(&self, tree: &Tree, dir: &Path) -> Result<()> {
        self.materialize_tree(tree, dir)
    }

    pub(crate) fn materialize_tree_seeded(
        &self,
        tree: &Tree,
        dir: &Path,
    ) -> Result<MaterializedTree> {
        let plan_start = Instant::now();
        let mut plan = MaterializationPlan {
            validation_root: dir.to_path_buf(),
            directories: Vec::new(),
            directory_contexts: Vec::new(),
            leaves: Vec::new(),
            file_count: 0,
            symlink_count: 0,
        };
        self.plan_materialization(tree, Path::new(""), dir, &mut plan)?;
        let plan_duration_ms = plan_start.elapsed().as_millis();

        let execution_start = Instant::now();
        let requested_threads = requested_materialization_threads();
        fs::create_dir_all(dir)
            .map_err(|e| HeddleError::Io(enrich_fs_error(dir, "creating", e)))?;
        for directory in &plan.directories {
            fs::create_dir_all(directory)
                .map_err(|e| HeddleError::Io(enrich_fs_error(directory, "creating", e)))?;
        }

        let (worker_count, file_entries) = self.materialize_write_ops_seeded(&plan.leaves)?;

        debug!(
            directories = plan.directories.len(),
            files = plan.file_count,
            symlinks = plan.symlink_count,
            workers = worker_count,
            requested_workers = requested_threads.map(NonZeroUsize::get),
            plan_duration_ms,
            execution_duration_ms = execution_start.elapsed().as_millis(),
            parallel = worker_count > 1,
            "Tree materialization complete"
        );

        Ok(MaterializedTree {
            file_entries,
            directory_contexts: plan.directory_contexts,
        })
    }

    fn plan_materialization(
        &self,
        tree: &Tree,
        rel_dir: &Path,
        dir: &Path,
        plan: &mut MaterializationPlan,
    ) -> Result<()> {
        plan.directory_contexts.push(MaterializedDirectoryContext {
            key: cache_key(rel_dir),
            path: dir.to_path_buf(),
            child_names: tree
                .entries()
                .iter()
                .map(|entry| entry.name().to_string())
                .collect(),
            tree_hash: tree.hash(),
        });

        for entry in tree.entries() {
            let path = dir.join(entry.name());
            let rel_path = rel_dir.join(entry.name());
            match entry.target() {
                TreeEntryTarget::Blob { hash, executable } => {
                    plan.file_count += 1;
                    plan.leaves.push(WorktreeWriteOp::Blob {
                        path,
                        hash: *hash,
                        executable: *executable,
                    });
                }
                TreeEntryTarget::Tree { hash } => {
                    let subtree = self
                        .store
                        .get_tree(hash)?
                        .ok_or_else(|| HeddleError::NotFound(format!("tree {}", hash)))?;
                    plan.directories.push(path.clone());
                    self.plan_materialization(&subtree, &rel_path, &path, plan)?;
                }
                TreeEntryTarget::Symlink { hash } => {
                    plan.symlink_count += 1;
                    plan.leaves.push(WorktreeWriteOp::Symlink {
                        path,
                        hash: *hash,
                        validation_root: plan.validation_root.clone(),
                    });
                }
                TreeEntryTarget::Gitlink { target } => {
                    plan.file_count += 1;
                    plan.leaves.push(WorktreeWriteOp::GitlinkPlaceholder {
                        path,
                        target: *target,
                    });
                }
                // Native child-spool edge: not materialized to the worktree
                // in this phase, so it contributes no write op.
                TreeEntryTarget::Spoollink { .. } => {}
            }
        }

        Ok(())
    }

    pub(crate) fn materialize_write_ops(&self, writes: &[WorktreeWriteOp]) -> Result<usize> {
        self.materialize_write_ops_seeded(writes)
            .map(|(worker_count, _)| worker_count)
    }

    pub(crate) fn materialize_write_ops_seeded(
        &self,
        writes: &[WorktreeWriteOp],
    ) -> Result<(usize, Vec<SeededWorktreeEntry>)> {
        prepare_parent_directories(writes)?;

        let requested_threads = requested_materialization_threads();
        let worker_count = materialization_worker_count(writes.len(), requested_threads);

        // No probe — the per-blob path tries `clonefile`/FICLONE
        // first and flips a batch-wide flag on the first
        // `EXDEV`/`EOPNOTSUPP`/`ENOSYS` verdict, so the rest of the
        // batch falls straight through to `fs::copy` without paying
        // the syscall tax. The cost of one failed reflink call on a
        // non-CoW filesystem is one syscall; it's not worth a
        // dedicated probe.
        let context = MaterializationContext::new();

        // Drive the repository's live progress handle: one `inc` per
        // materialized write op. On the common null handle this is a relaxed
        // atomic add plus a predicted-not-taken branch (no render). When a CLI
        // command has installed a real sink, this paints a live "checking out
        // files (n/total)" line. The handle is a cheap `Arc` clone, so it
        // survives the `thread::scope` seam below — each worker gets its own
        // clone and increments the shared atomic counter.
        let progress = self.progress();
        progress.set_total(writes.len());

        let result = if worker_count <= 1 {
            let mut seeded = Vec::with_capacity(writes.len());
            for write in writes {
                seeded.push(self.materialize_write_op(write, &context)?);
                progress.inc(1);
            }
            Ok((worker_count, seeded))
        } else {
            let chunk_size = writes.len().div_ceil(worker_count);
            let seeded = thread::scope(|scope| -> Result<Vec<SeededWorktreeEntry>> {
                let mut workers = Vec::new();
                let context = &context;
                for chunk in writes.chunks(chunk_size) {
                    let progress = progress.clone();
                    workers.push(scope.spawn(move || -> Result<Vec<SeededWorktreeEntry>> {
                        let mut seeded = Vec::with_capacity(chunk.len());
                        for write in chunk {
                            seeded.push(self.materialize_write_op(write, context)?);
                            progress.inc(1);
                        }
                        Ok(seeded)
                    }));
                }

                let mut seeded = Vec::with_capacity(writes.len());
                for worker in workers {
                    seeded.extend(worker.join().map_err(|_| {
                        HeddleError::Config("materialization worker panicked".to_string())
                    })??);
                }

                Ok(seeded)
            })?;

            Ok((worker_count, seeded))
        };

        let reflinks = context.reflink_count.load(Ordering::Relaxed);
        let copies = context.copy_count.load(Ordering::Relaxed);
        if reflinks + copies > 0 {
            debug!(
                reflinks,
                copies,
                reflinks_enabled = context.reflinks_enabled(),
                "Materialized blobs"
            );
        }

        result
    }

    fn materialize_write_op(
        &self,
        write: &WorktreeWriteOp,
        context: &MaterializationContext,
    ) -> Result<SeededWorktreeEntry> {
        match write {
            WorktreeWriteOp::Blob {
                path,
                hash,
                executable,
            } => {
                self.materialize_blob(path, hash, *executable, context)?;
            }
            WorktreeWriteOp::Symlink {
                path,
                hash,
                validation_root,
            } => {
                let blob = self
                    .store
                    .get_blob(hash)?
                    .ok_or_else(|| HeddleError::NotFound(format!("blob {}", hash)))?;
                #[cfg(unix)]
                {
                    let target = std::str::from_utf8(blob.content()).map_err(|_| {
                        HeddleError::InvalidObject("invalid symlink target".to_string())
                    })?;
                    let target_path = Path::new(target);
                    let symlink_dir = path.parent().unwrap_or(validation_root);
                    if !validate_symlink_target(validation_root, symlink_dir, target_path) {
                        return Err(HeddleError::InvalidSymlinkTarget {
                            path: path
                                .strip_prefix(validation_root)
                                .unwrap_or(path)
                                .to_path_buf(),
                            target: target_path.to_path_buf(),
                        });
                    }
                    remove_materialized_leaf(path)?;
                    std::os::unix::fs::symlink(target, path)?;
                }
                // Windows symlink materialization is unimplemented;
                // the projection layer (ProjFS) handles symlinks
                // through reparse points instead of native symlinks,
                // and `heddle materialize` on Windows isn't part of
                // the daily-use mount story. Suppress the unused
                // bindings rather than ship a half-implementation.
                #[cfg(not(unix))]
                {
                    let _ = (blob, path, validation_root);
                }
            }
            WorktreeWriteOp::GitlinkPlaceholder { path, target } => {
                remove_materialized_leaf(path)?;
                fs::write(path, gitlink_placeholder_bytes(target))
                    .map_err(|err| HeddleError::Io(enrich_fs_error(path, "writing", err)))?;
            }
        }

        let metadata = fs::symlink_metadata(write.path())?;
        let entry = build_cached_entry(
            write.hash(),
            &metadata,
            write.executable(),
            write.index_kind(),
        )
        .ok_or_else(|| {
            HeddleError::Config(format!(
                "seed materialized worktree entry for {}",
                write.path().display()
            ))
        })?;

        Ok(SeededWorktreeEntry {
            key: cache_key(
                write
                    .path()
                    .strip_prefix(self.root())
                    .unwrap_or(write.path()),
            ),
            entry,
        })
    }

    /// Materialize a single blob into the worktree.
    ///
    /// Strategy (in order):
    ///   1. Filesystem reflink (`clonefile(2)` on macOS APFS,
    ///      `ioctl(FICLONE)` on Linux btrfs/XFS/ZFS) from the
    ///      canonical loose-uncompressed blob into `dest`. The dest
    ///      gets its own inode; the kernel forks the underlying
    ///      allocation on first write to either side. On reflink-
    ///      capable filesystems this preserves the storage win
    ///      (~1× disk for N worktrees of the same state) without
    ///      any shared-inode hazard.
    ///   2. Lazy promotion + retry. If the canonical loose blob
    ///      isn't on disk (e.g. post-`pack_objects + prune_loose`),
    ///      promote it once and retry the reflink.
    ///   3. `fs::write` of the decompressed blob bytes. Used when the
    ///      filesystem doesn't support reflinks at all
    ///      (`EXDEV`/`EOPNOTSUPP`/`ENOSYS`), in which case we flip a
    ///      batch-wide flag and stop trying for the rest of this
    ///      materialization.
    ///
    /// Permission bits are normalized to `0o644` (or `0o755` for
    /// executables) on every path. There is no read-only-mode
    /// defense — agents can `chmod +w` and overwrite freely; the
    /// filesystem-level isolation is what keeps sibling worktrees
    /// safe.
    fn materialize_blob(
        &self,
        dest: &Path,
        hash: &ContentHash,
        executable: bool,
        context: &MaterializationContext,
    ) -> Result<()> {
        // Redaction short-circuit: if any redaction declares this
        // blob's bytes off-limits, materialize the human-readable
        // stub instead. The stub names who redacted it, when, why,
        // and whether the bytes have already been purged. Safe to
        // include in worktrees, semantic diffs, and Git projection
        // exports (which themselves call through `materialize_tree`).
        // Errors loading the redactions store are propagated rather
        // than swallowed — a partial redaction read shouldn't
        // silently leak the original bytes.
        if let Some(stub) = self
            .redaction_stub_for_blob(hash)
            .map_err(|err| HeddleError::Config(format!("redaction lookup failed: {err}")))?
        {
            let _ = fs::remove_file(dest);
            fs::write(dest, stub.as_bytes())?;
            // Stubs are never executable — overwriting a tracked
            // executable with a stub correctly drops the +x bit so
            // operators don't accidentally run the redaction notice.
            set_file_mode(dest, false)?;
            // The redaction stub path doesn't reflink/clone — count
            // it as a copy so observability stays accurate.
            context.record_copy();
            let _ = executable;
            return Ok(());
        }

        if context.reflinks_enabled() {
            // First-pass: blob is already loose+uncompressed.
            if let Some(source) = self.store.loose_blob_path(hash)
                && self.try_clone(&source, dest, executable, context)?
            {
                return Ok(());
            }
            // Second-pass: lazy promotion. Pack-resident or
            // compressed-loose blob — promote it to the canonical
            // uncompressed-loose path, then retry the reflink.
            // Without this step `pack_objects + prune_loose_objects`
            // permanently degrades materialize to slow `fs::write`.
            //
            // The first materialize of any given hash pays
            // decompress + atomic write, but every subsequent one
            // (other worktrees, future `goto`s) is a single
            // `clonefile`/FICLONE. Net win for any N > 1
            // materializations on a CoW filesystem.
            match self.store.promote_to_loose_uncompressed(hash) {
                Ok(_) => {
                    if let Some(source) = self.store.loose_blob_path(hash)
                        && self.try_clone(&source, dest, executable, context)?
                    {
                        return Ok(());
                    }
                }
                Err(err) => {
                    debug!(
                        ?err,
                        hash = %hash,
                        "promote_to_loose_uncompressed failed; falling back to fs::write"
                    );
                }
            }
        }

        let blob = self
            .store
            .get_blob(hash)?
            .ok_or_else(|| HeddleError::NotFound(format!("blob {}", hash)))?;
        // Remove any stale dest before writing. We don't share inodes
        // with the canonical store anymore (no hardlinks), but a
        // previous `goto` could still have left an unrelated file
        // here that we should overwrite cleanly.
        let _ = fs::remove_file(dest);
        fs::write(dest, blob.content())?;
        set_file_mode(dest, executable)?;
        context.record_copy();
        Ok(())
    }

    /// One clone attempt: returns `Ok(true)` on a successful reflink,
    /// `Ok(false)` when the caller should fall back to the in-memory
    /// `fs::write` path. The two `Ok(false)` causes are deliberately
    /// handled differently:
    ///
    /// * `ReflinkOutcome::Unsupported` (`EXDEV`/`EOPNOTSUPP`/`ENOSYS`/
    ///   `EINVAL`) — a filesystem-capability verdict, so the context is
    ///   flipped (`disable_reflinks`) and the rest of the batch skips
    ///   straight to `fs::write` without paying the failed-syscall tax.
    /// * `ReflinkOutcome::SourceVanished` — the loose mirror was pruned
    ///   mid-flight. A per-blob race, so we fall back for this blob only
    ///   and leave reflinks ENABLED for the rest of the batch
    ///   (heddle#571 r3).
    ///
    /// Genuine I/O errors bubble up (attributed to the offending side by
    /// `classify_clone_failure`).
    fn try_clone(
        &self,
        source: &Path,
        dest: &Path,
        executable: bool,
        context: &MaterializationContext,
    ) -> Result<bool> {
        // `clonefile`/`FICLONE` fail if `dest` already exists, so
        // make sure we're starting from a clean slate. A previous
        // `goto` could have left a regular file or a stale link here.
        let _ = fs::remove_file(dest);
        // Reflink is a pure optimization; correctness must never depend on the
        // loose source still being present at the syscall. `loose_blob_path`
        // verified it existed, but a concurrent prune or a torn NoSync promote
        // can remove it before we get here. On macOS, handing `clonefile(2)` a
        // missing source surfaces as ENOENT — which `reflink_unsupported`
        // deliberately does NOT swallow (ENOENT means a genuinely missing file,
        // not "reflink unsupported") — so without this guard the whole
        // `heddle start` hard-fails (heddle#571). Fall back to the bytes-write
        // path for THIS blob only by returning `Ok(false)`; do NOT
        // `disable_reflinks`, so sibling blobs whose mirrors are intact still
        // clone. The caller (`materialize_blob`) then writes the blob from its
        // decompressed bytes via `get_blob` — the same path Linux's
        // ext4-EOPNOTSUPP short-circuit already takes.
        if !source.exists() {
            debug!(
                source = %source.display(),
                dest = %dest.display(),
                "loose reflink source missing before clone; falling back to bytes-write for this blob"
            );
            return Ok(false);
        }
        use objects::fs_clone::ReflinkOutcome;
        match objects::fs_clone::try_reflink(source, dest) {
            Ok(ReflinkOutcome::Cloned) => {
                set_file_mode(dest, executable)?;
                context.record_reflink();
                Ok(true)
            }
            Ok(ReflinkOutcome::Unsupported) => {
                // Filesystem doesn't support reflinks. Disable for
                // the rest of the batch and let the caller fall
                // through to `fs::write` (which decompresses from
                // memory rather than reading the loose file twice).
                debug!(
                    source = %source.display(),
                    dest = %dest.display(),
                    "reflink not supported on this filesystem; switching batch to fs::write fallback"
                );
                context.disable_reflinks();
                Ok(false)
            }
            Ok(ReflinkOutcome::SourceVanished) => {
                // [heddle#571 r3] The loose mirror was pruned out from under
                // us between the pre-check and the clone. That's a per-blob
                // race, NOT a filesystem-capability verdict — so degrade to
                // the bytes-write fallback for THIS blob only and DO NOT
                // `disable_reflinks`. Sibling blobs whose mirrors are intact
                // still get the CoW win. A blob genuinely absent from the
                // store still errors loudly when the bytes-write fallback's
                // `get_blob` can't find it.
                debug!(
                    source = %source.display(),
                    dest = %dest.display(),
                    "loose reflink source vanished before clone; falling back to bytes-write for this blob (reflinks stay enabled batch-wide)"
                );
                Ok(false)
            }
            Err(err) => {
                debug!(
                    ?err,
                    source = %source.display(),
                    dest = %dest.display(),
                    "reflink failed with I/O error"
                );
                match classify_clone_failure(source, dest, &err) {
                    // [heddle#571 r2, finding 2] Source vanished mid-flight
                    // (TOCTOU): degrade to the bytes-write fallback for this
                    // blob rather than hard-erroring. See `classify_clone_failure`.
                    None => {
                        debug!(
                            source = %source.display(),
                            dest = %dest.display(),
                            "loose reflink source vanished between pre-check and clone syscall; falling back to bytes-write for this blob"
                        );
                        Ok(false)
                    }
                    // [heddle#571 r2, finding 3] Attribute to the offending side.
                    Some((offender, action)) => {
                        Err(HeddleError::Io(enrich_fs_error(offender, action, err)))
                    }
                }
            }
        }
    }
}

/// Classify a clone-syscall (`clonefile`/`FICLONE`) I/O failure into either a
/// bytes-write fallback or an enriched, correctly-attributed error.
///
/// * `None` — the loose source vanished between the `!source.exists()`
///   pre-check and the syscall (concurrent prune / torn NoSync promote), so the
///   syscall raised `ENOENT`. The caller should degrade to the bytes-write
///   fallback (`Ok(false)`), which re-reads the authoritative bytes from the
///   store; a blob genuinely absent from the store still errors there with its
///   hash. This closes the TOCTOU race rather than blindly masking `ENOENT`
///   (heddle#571 r2, finding 2).
/// * `Some((path, action))` — a real failure to surface, attributed to the side
///   that actually failed. The source survived (the vanished case returned
///   `None`), so a create/permission/read-only/no-space failure is the
///   DESTINATION's (read-only checkout dir, unwritable target). We blame the
///   source only when it is the unreadable party (probed directly). Reporting a
///   dest-side failure against the blob path would point the user at the wrong
///   file (heddle#571 r2, finding 3).
fn classify_clone_failure<'a>(
    source: &'a Path,
    dest: &'a Path,
    err: &std::io::Error,
) -> Option<(&'a Path, &'static str)> {
    if err.kind() == std::io::ErrorKind::NotFound && !source.exists() {
        return None;
    }
    if fs::File::open(source).is_ok() {
        Some((dest, "reflinking into"))
    } else {
        Some((source, "reflinking"))
    }
}

fn prepare_parent_directories(writes: &[WorktreeWriteOp]) -> Result<()> {
    let mut parents = BTreeSet::new();
    for write in writes {
        if let Some(parent) = write.path().parent() {
            parents.insert(parent.to_path_buf());
        }
    }

    for parent in parents {
        fs::create_dir_all(&parent)
            .map_err(|e| HeddleError::Io(enrich_fs_error(&parent, "creating", e)))?;
    }

    Ok(())
}

/// Best-effort removal of a leaf path before replacing it with another
/// materialized leaf, such as a symlink on Unix or a gitlink placeholder
/// on every platform.
///
/// Tolerates `ENOTEMPTY` from `remove_dir` for the same reason the
/// incremental apply path does: untracked or explicitly ignored siblings
/// may still occupy the directory after the planner has cleaned out the
/// tracked children. Without this
/// tolerance, a `goto` over a real-world worktree that mutates a
/// tracked directory into a symlink aborts mid-apply with `os error
/// 66`, leaving HEAD stuck and disk diverged from state.
///
fn remove_materialized_leaf(path: &Path) -> Result<()> {
    match fs::symlink_metadata(path) {
        Ok(metadata) => {
            let file_type = metadata.file_type();
            if file_type.is_symlink() || file_type.is_file() {
                fs::remove_file(path)
                    .map_err(|e| HeddleError::Io(enrich_fs_error(path, "removing", e)))?;
            } else if file_type.is_dir() {
                match fs::remove_dir(path) {
                    Ok(()) => {}
                    Err(error) if is_directory_not_empty(&error) => {}
                    Err(error) => {
                        return Err(HeddleError::Io(enrich_fs_error(path, "removing", error)));
                    }
                }
            }
            Ok(())
        }
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(error) => Err(HeddleError::Io(enrich_fs_error(path, "inspecting", error))),
    }
}

fn set_file_mode(path: &Path, executable: bool) -> Result<()> {
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;

        // `OpenOptions::mode(0o644)` is still filtered by the
        // process umask, and reflink/copy paths preserve the source
        // mode. Normalize the worktree-visible file mode here so
        // materialized checkouts do not inherit a restrictive object
        // store mode such as `0o600`.
        let mode = if executable { 0o755 } else { 0o644 };
        fs::set_permissions(path, fs::Permissions::from_mode(mode))?;
    }
    #[cfg(not(unix))]
    {
        let _ = (path, executable);
    }
    Ok(())
}

fn materialization_worker_count(
    operation_count: usize,
    requested_threads: Option<NonZeroUsize>,
) -> usize {
    if operation_count < MATERIALIZE_PARALLEL_THRESHOLD {
        return 1;
    }

    let available = requested_threads.unwrap_or_else(default_materialization_threads);
    available.get().min(operation_count.max(1))
}

fn default_materialization_threads() -> NonZeroUsize {
    std::thread::available_parallelism().unwrap_or(NonZeroUsize::MIN)
}

fn requested_materialization_threads() -> Option<NonZeroUsize> {
    let raw = std::env::var(MATERIALIZE_THREADS_ENV).ok()?;
    raw.trim().parse::<usize>().ok().and_then(NonZeroUsize::new)
}

#[cfg(test)]
mod tests {
    use std::{num::NonZeroUsize, path::PathBuf};

    use objects::{fs_clone::filesystem_supports_reflink, object::Blob, store::ObjectStore};
    use tempfile::TempDir;

    use super::{
        MaterializationContext, Repository, WorktreeWriteOp, classify_clone_failure,
        materialization_worker_count, remove_materialized_leaf,
    };

    /// The generic [`Progress`](objects::Progress) handle installed on a
    /// repository must survive the `thread::scope` parallel-materialization seam:
    /// a single shared `Arc` handle, cloned into each worker, whose atomic
    /// counter ends at exactly the file count no matter how the work was split
    /// across threads. This is the load-bearing acceptance criterion for #844.
    #[test]
    fn progress_handle_survives_parallel_materialization_seam() {
        use std::sync::{
            Arc,
            atomic::{AtomicUsize, Ordering},
        };

        use objects::{Progress, ProgressSnapshot, Sink};

        /// Active sink that just counts render calls, proving the active path
        /// ran concurrently across workers without panicking.
        struct CountingSink(Arc<AtomicUsize>);
        impl Sink for CountingSink {
            fn render(&self, _snap: ProgressSnapshot) {
                self.0.fetch_add(1, Ordering::Relaxed);
            }
        }

        let temp_dir = TempDir::new().unwrap();
        let repo = Repository::init_default(temp_dir.path()).unwrap();

        // Well over MATERIALIZE_PARALLEL_THRESHOLD (32): on any multi-core host
        // this exercises the `thread::scope` branch; on a single core it
        // degrades to the serial branch and the counter assertion still holds.
        let file_count = 256usize;
        for i in 0..file_count {
            std::fs::write(
                temp_dir.path().join(format!("seam-{i:04}.txt")),
                format!("seam payload {i} {}", "y".repeat(48)),
            )
            .unwrap();
        }
        let state = repo.snapshot(Some("seam".to_string()), None).unwrap();
        let tree = repo.store().get_tree(&state.tree).unwrap().unwrap();

        let render_count = Arc::new(AtomicUsize::new(0));
        let progress = Progress::with_sink(Box::new(CountingSink(render_count.clone())));
        repo.set_progress(progress.clone());

        let dest = temp_dir.path().join("materialized");
        repo.materialize_tree(&tree, &dest).unwrap();

        // Every write op incremented the shared counter exactly once, across
        // however many worker threads split the batch.
        assert_eq!(
            progress.done(),
            file_count,
            "shared progress counter must equal the file count after the parallel seam"
        );
        assert_eq!(
            progress.total(),
            file_count,
            "total set from the write batch"
        );
        // The active handle rendered at least once through the seam (throttled,
        // so not once-per-file) — proving the sink is driven across threads.
        assert!(
            render_count.load(Ordering::Relaxed) > 0,
            "active sink should have rendered during materialization"
        );
    }

    /// heddle#571 (round 2, finding 2): a clone syscall that raises ENOENT
    /// because the loose source vanished AFTER the pre-check (TOCTOU) must
    /// degrade to the bytes-write fallback (`None`), not hard-error.
    #[test]
    fn classify_clone_failure_vanished_source_falls_back() {
        let temp = TempDir::new().unwrap();
        let source = temp.path().join("gone.blob");
        let dest = temp.path().join("checkout/file");
        assert!(!source.exists());

        let enoent = std::io::Error::from(std::io::ErrorKind::NotFound);
        assert!(
            classify_clone_failure(&source, &dest, &enoent).is_none(),
            "a vanished-source ENOENT must signal the bytes-write fallback"
        );
    }

    /// heddle#571 (round 2, finding 3): when the source is still present and
    /// readable, the clone failed on the DESTINATION side — attribute the error
    /// to the dest (checkout) path, not the blob.
    #[test]
    fn classify_clone_failure_present_source_blames_dest() {
        let temp = TempDir::new().unwrap();
        let source = temp.path().join("present.blob");
        std::fs::write(&source, b"bytes").unwrap();
        let dest = temp.path().join("readonly-checkout/file");

        // A dest-side failure shape (e.g. read-only checkout dir).
        let erofs = std::io::Error::from(std::io::ErrorKind::PermissionDenied);
        let attributed = classify_clone_failure(&source, &dest, &erofs);
        assert_eq!(
            attributed,
            Some((dest.as_path(), "reflinking into")),
            "a failure with the source still readable must be attributed to dest"
        );
    }

    /// heddle#571 (round 2, finding 3): a non-ENOENT failure whose source is no
    /// longer openable is attributed to the SOURCE path (it is the unreadable
    /// party), not the destination.
    #[test]
    fn classify_clone_failure_unreadable_source_blames_source() {
        let temp = TempDir::new().unwrap();
        let source = temp.path().join("missing.blob"); // not created → unopenable
        let dest = temp.path().join("file");

        // Not ENOENT (so the vanished-source fast path doesn't fire), but the
        // source can't be opened → blame the source.
        let other = std::io::Error::from(std::io::ErrorKind::PermissionDenied);
        assert_eq!(
            classify_clone_failure(&source, &dest, &other),
            Some((source.as_path(), "reflinking")),
            "an unreadable source must be attributed to the source path"
        );
    }

    /// heddle#571 (round 3): a vanished loose source is a per-blob race, not a
    /// filesystem-capability verdict — `try_clone` must degrade to the
    /// bytes-write fallback for that blob WITHOUT flipping the batch-wide
    /// `reflink_supported` flag. Previously the helper's pre-check returned a
    /// bare `Ok(false)`, indistinguishable from "filesystem can't reflink", so
    /// one concurrently-pruned mirror needlessly disabled reflinks (forcing
    /// copy-writes) for every remaining blob in the batch.
    #[test]
    fn try_clone_vanished_source_keeps_batch_reflinks_enabled() {
        let temp = TempDir::new().unwrap();
        let repo = Repository::init_default(temp.path()).unwrap();
        let context = MaterializationContext::new();
        assert!(context.reflinks_enabled(), "context starts optimistic");

        let missing = temp.path().join("pruned.blob");
        let dest = temp.path().join("wt/out.txt");
        assert!(!missing.exists());

        let cloned = repo
            .try_clone(&missing, &dest, false, &context)
            .expect("a vanished source must fall back, not error");
        assert!(!cloned, "a vanished source cannot have been reflinked");
        assert!(
            context.reflinks_enabled(),
            "a vanished source must NOT disable reflinks for the rest of the batch"
        );
    }

    /// Regression: `remove_materialized_leaf` must tolerate `ENOTEMPTY` on
    /// the directory branch, mirroring `remove_existing_path` in the
    /// incremental apply path. Both tolerances are needed because the
    /// apply planner only removes tracked descendants — when the planner asks
    /// the materializer to clear a directory whose tracked children are gone
    /// but whose untracked or explicitly ignored children remain, `remove_dir` errors
    /// with `os error 66` (macOS/BSD) / `39` (Linux). Pre-fix the
    /// materialization branch propagated that error and aborted apply
    /// mid-walk, leaving HEAD stuck and disk diverged from state.
    #[test]
    fn remove_materialized_leaf_tolerates_directory_not_empty() {
        let temp = TempDir::new().unwrap();
        let dir = temp.path().join("web");
        std::fs::create_dir_all(dir.join("node_modules/lodash")).unwrap();
        std::fs::write(dir.join("node_modules/lodash/index.js"), "ignored").unwrap();

        // Pre-fix this would propagate ENOTEMPTY; post-fix it returns Ok
        // and leaves the directory (with its ignored content) on disk.
        remove_materialized_leaf(&dir).expect("must tolerate ENOTEMPTY");
        assert!(
            dir.join("node_modules/lodash/index.js").exists(),
            "ignored content must survive the tolerated removal"
        );
    }

    /// Regression: empty directories still get cleaned up (the common
    /// case). The `ENOTEMPTY` tolerance must not regress the happy path.
    #[test]
    fn remove_materialized_leaf_removes_empty_directory() {
        let temp = TempDir::new().unwrap();
        let dir = temp.path().join("emptydir");
        std::fs::create_dir(&dir).unwrap();

        remove_materialized_leaf(&dir).expect("must remove empty dir");
        assert!(!dir.exists(), "empty directory must be removed");
    }

    /// Regression: missing paths are a no-op (NotFound), not an error.
    #[test]
    fn remove_materialized_leaf_is_noop_for_missing_path() {
        let temp = TempDir::new().unwrap();
        remove_materialized_leaf(&temp.path().join("does-not-exist"))
            .expect("missing path must be a no-op");
    }

    /// Regression: regular files are still removed (the common symlink-
    /// replacement case where the existing leaf was a tracked file).
    #[test]
    fn remove_materialized_leaf_removes_regular_file() {
        let temp = TempDir::new().unwrap();
        let file = temp.path().join("a.txt");
        std::fs::write(&file, "content").unwrap();

        remove_materialized_leaf(&file).expect("must remove regular file");
        assert!(!file.exists(), "regular file must be removed");
    }

    #[test]
    fn materialization_parallelism_stays_sequential_for_small_workloads() {
        assert_eq!(materialization_worker_count(31, Some(NonZeroUsize::MIN)), 1);
    }

    #[test]
    fn materialization_parallelism_respects_requested_thread_cap() {
        assert_eq!(materialization_worker_count(128, NonZeroUsize::new(4)), 4);
    }

    #[test]
    fn materialize_write_ops_prepares_missing_parent_directories() {
        let temp_dir = TempDir::new().unwrap();
        let repo = Repository::init_default(temp_dir.path()).unwrap();

        let blob = Blob::from("cold pull payload");
        let hash = repo.store().put_blob(&blob).unwrap();
        let file_path = temp_dir.path().join("nested/deep/file.txt");

        repo.materialize_write_ops(&[WorktreeWriteOp::Blob {
            path: file_path.clone(),
            hash,
            executable: false,
        }])
        .unwrap();

        assert_eq!(
            std::fs::read_to_string(&file_path).unwrap(),
            "cold pull payload"
        );
    }

    /// Materialized blobs must be writable by default. The
    /// previous hardlink+chmod-0o444 approach was a footgun:
    /// `chmod 644` then in-place write would mutate the canonical
    /// store inode, corrupting every other worktree. The fix is
    /// filesystem-level CoW (or full copy), so each worktree gets
    /// its own inode and a normal `0o644`/`0o755` mode.
    #[test]
    #[cfg(unix)]
    fn materialized_blob_uses_normal_writable_mode() {
        use std::os::unix::fs::PermissionsExt;

        let temp_dir = TempDir::new().unwrap();
        let repo = Repository::init_default(temp_dir.path()).unwrap();

        let blob = Blob::from("normal mode payload");
        let hash = repo.store().put_blob(&blob).unwrap();
        let regular = temp_dir.path().join("worktree/file.txt");
        let exec = temp_dir.path().join("worktree/run.sh");

        repo.materialize_write_ops(&[
            WorktreeWriteOp::Blob {
                path: regular.clone(),
                hash,
                executable: false,
            },
            WorktreeWriteOp::Blob {
                path: exec.clone(),
                hash,
                executable: true,
            },
        ])
        .unwrap();

        let regular_mode = std::fs::metadata(&regular).unwrap().permissions().mode() & 0o777;
        let exec_mode = std::fs::metadata(&exec).unwrap().permissions().mode() & 0o777;
        assert_eq!(
            regular_mode, 0o644,
            "regular blob must be 0o644 (got 0o{:o})",
            regular_mode
        );
        assert_eq!(
            exec_mode, 0o755,
            "executable blob must be 0o755 (got 0o{:o})",
            exec_mode
        );

        // Sanity: a plain in-place write on the materialized file
        // must succeed (no chmod gymnastics required).
        std::fs::write(&regular, b"agent edits this").unwrap();
        assert_eq!(std::fs::read(&regular).unwrap(), b"agent edits this");
    }

    /// THE core isolation property. An agent in worktree-A that
    /// chmods +w (no-op since we already ship 0o644) and writes
    /// in-place must not affect worktree-B's bytes. Under the old
    /// hardlink+chmod model this exact sequence corrupted sibling
    /// worktrees through the shared inode. Under the new
    /// CoW/copy model the worktrees have distinct inodes and the
    /// kernel guarantees isolation.
    #[test]
    #[cfg(unix)]
    fn materialize_then_chmod_and_write_does_not_affect_sibling_worktree() {
        use std::os::unix::fs::PermissionsExt;

        let temp_dir = TempDir::new().unwrap();
        let repo = Repository::init_default(temp_dir.path()).unwrap();

        let blob = Blob::from("canonical bytes that must never change");
        let hash = repo.store().put_blob(&blob).unwrap();

        let worktree_a = temp_dir.path().join("wt-a/file.txt");
        let worktree_b = temp_dir.path().join("wt-b/file.txt");

        repo.materialize_write_ops(&[WorktreeWriteOp::Blob {
            path: worktree_a.clone(),
            hash,
            executable: false,
        }])
        .unwrap();
        repo.materialize_write_ops(&[WorktreeWriteOp::Blob {
            path: worktree_b.clone(),
            hash,
            executable: false,
        }])
        .unwrap();

        // Simulate a misbehaving agent: re-assert mode 0o644 (the
        // old defense rendered this a no-op for blocking writes),
        // then truncate-and-overwrite in place via the shell-style
        // `> file` pathway.
        std::fs::set_permissions(&worktree_a, std::fs::Permissions::from_mode(0o644)).unwrap();
        std::fs::write(&worktree_a, b"AGENT_TAMPERED_WITH_WORKTREE_A").unwrap();

        // Sibling worktree's bytes are unchanged.
        assert_eq!(
            std::fs::read(&worktree_b).unwrap(),
            blob.content(),
            "sibling worktree must keep canonical bytes despite in-place write to worktree-a"
        );
        // And the canonical loose blob in the store is untouched.
        if let Some(loose) = repo.store().loose_blob_path(&hash) {
            assert_eq!(
                std::fs::read(&loose).unwrap(),
                blob.content(),
                "canonical loose blob must keep canonical bytes despite in-place write to worktree-a"
            );
        }
    }

    /// Atomic-rename writes (write-tempfile + `rename(2)` over
    /// target) must also leave sibling worktrees untouched. This
    /// path was always safe under the old model too — proving it
    /// keeps working with the new isolation strategy.
    #[test]
    #[cfg(unix)]
    fn materialize_atomic_rename_does_not_affect_sibling_worktree() {
        let temp_dir = TempDir::new().unwrap();
        let repo = Repository::init_default(temp_dir.path()).unwrap();

        let blob = Blob::from("atomic-rename canonical bytes");
        let hash = repo.store().put_blob(&blob).unwrap();

        let worktree_a = temp_dir.path().join("wt-a/file.txt");
        let worktree_b = temp_dir.path().join("wt-b/file.txt");

        repo.materialize_write_ops(&[WorktreeWriteOp::Blob {
            path: worktree_a.clone(),
            hash,
            executable: false,
        }])
        .unwrap();
        repo.materialize_write_ops(&[WorktreeWriteOp::Blob {
            path: worktree_b.clone(),
            hash,
            executable: false,
        }])
        .unwrap();

        let tmp = temp_dir.path().join("wt-a/file.txt.tmp");
        std::fs::write(&tmp, b"NEW_CONTENT_VIA_ATOMIC_RENAME").unwrap();
        std::fs::rename(&tmp, &worktree_a).unwrap();

        assert_eq!(
            std::fs::read(&worktree_a).unwrap(),
            b"NEW_CONTENT_VIA_ATOMIC_RENAME"
        );
        assert_eq!(
            std::fs::read(&worktree_b).unwrap(),
            blob.content(),
            "sibling worktree must keep canonical bytes despite atomic rename in worktree-a"
        );
    }

    /// On a CoW filesystem (APFS, btrfs, XFS-with-reflinks, ZFS)
    /// the materialized worktree file must have a **distinct**
    /// inode from the canonical loose blob. This is the key
    /// correctness assertion that distinguishes reflinks from
    /// hardlinks: hardlinks share inodes (the bug we fixed),
    /// reflinks do not.
    ///
    /// Ignored by default because `fs::copy` also gives distinct
    /// inodes on non-CoW filesystems while this test specifically
    /// targets the reflink path.
    #[test]
    #[cfg(unix)]
    #[ignore = "requires a reflink-capable filesystem"]
    fn materialize_uses_reflink_when_filesystem_supports_it() {
        use std::os::unix::fs::MetadataExt;

        let temp_dir = TempDir::new().unwrap();
        assert!(
            filesystem_supports_reflink(temp_dir.path()),
            "materialize_uses_reflink_when_filesystem_supports_it requires reflink support"
        );

        let repo = Repository::init_default(temp_dir.path()).unwrap();
        let blob = Blob::from("reflink correctness check, kept under compression threshold");
        let hash = repo.store().put_blob(&blob).unwrap();
        let worktree = temp_dir.path().join("wt/file.txt");

        repo.materialize_write_ops(&[WorktreeWriteOp::Blob {
            path: worktree.clone(),
            hash,
            executable: false,
        }])
        .unwrap();

        let loose = repo
            .store()
            .loose_blob_path(&hash)
            .expect("blob must be loose+uncompressed (under threshold)");
        let loose_inode = std::fs::metadata(&loose).unwrap().ino();
        let worktree_inode = std::fs::metadata(&worktree).unwrap().ino();
        assert_ne!(
            loose_inode, worktree_inode,
            "reflinked worktree file must have a distinct inode from canonical loose blob (got {} for both — that's a hardlink, the bug we fixed)",
            loose_inode
        );
        // And nlink on the canonical blob is 1: nothing aliases it.
        let nlink = std::fs::metadata(&loose).unwrap().nlink();
        assert_eq!(
            nlink, 1,
            "canonical loose blob must not be aliased (nlink={}); reflinks share blocks, not inodes",
            nlink
        );
    }

    /// Functional readback after N materializations of the same
    /// blob across N worktrees on the same filesystem. Replaces
    /// the old "shared inode" assertion which is no longer the
    /// correctness model. Now we just assert every worktree reads
    /// back the canonical bytes (and they're independent — see
    /// the isolation tests above).
    #[test]
    #[cfg(unix)]
    fn materialize_blob_into_two_worktrees_reads_back_canonical_bytes() {
        let temp_dir = TempDir::new().unwrap();
        let repo = Repository::init_default(temp_dir.path()).unwrap();

        let blob = Blob::from("two-worktree readback payload");
        let hash = repo.store().put_blob(&blob).unwrap();

        let worktree_a = temp_dir.path().join("worktree-a/file.txt");
        let worktree_b = temp_dir.path().join("worktree-b/file.txt");

        repo.materialize_write_ops(&[WorktreeWriteOp::Blob {
            path: worktree_a.clone(),
            hash,
            executable: false,
        }])
        .unwrap();
        repo.materialize_write_ops(&[WorktreeWriteOp::Blob {
            path: worktree_b.clone(),
            hash,
            executable: false,
        }])
        .unwrap();

        assert_eq!(std::fs::read(&worktree_a).unwrap(), blob.content());
        assert_eq!(std::fs::read(&worktree_b).unwrap(), blob.content());
    }

    /// Symlinks are routed through the existing path; introducing
    /// hardlinks must not regress the symlink test that lives in
    /// `repository_tests.rs`. Locally we just confirm a symlink op
    /// still produces a real symlink (not a hardlink to the target
    /// blob's loose path).
    #[test]
    #[cfg(unix)]
    fn materialize_symlink_op_produces_real_symlink_not_hardlink() {
        let temp_dir = TempDir::new().unwrap();
        let repo = Repository::init_default(temp_dir.path()).unwrap();

        let symlink_blob = Blob::new(b"../canonical".to_vec());
        let symlink_hash = repo.store().put_blob(&symlink_blob).unwrap();
        let path = temp_dir.path().join("worktree/link.txt");

        repo.materialize_write_ops(&[WorktreeWriteOp::Symlink {
            path: path.clone(),
            hash: symlink_hash,
            validation_root: temp_dir.path().to_path_buf(),
        }])
        .unwrap();

        let meta = std::fs::symlink_metadata(&path).unwrap();
        assert!(
            meta.file_type().is_symlink(),
            "Symlink op must produce a real symlink, not a hardlinked regular file"
        );
        assert_eq!(
            std::fs::read_link(&path).unwrap(),
            PathBuf::from("../canonical")
        );
    }

    #[test]
    #[cfg(unix)]
    fn materialize_symlink_op_replaces_existing_symlink() {
        let temp_dir = TempDir::new().unwrap();
        let repo = Repository::init_default(temp_dir.path()).unwrap();

        let first_hash = repo.store().put_blob(&Blob::from("first")).unwrap();
        let second_hash = repo.store().put_blob(&Blob::from("second")).unwrap();
        let path = temp_dir.path().join("worktree/link.txt");

        repo.materialize_write_ops(&[WorktreeWriteOp::Symlink {
            path: path.clone(),
            hash: first_hash,
            validation_root: temp_dir.path().to_path_buf(),
        }])
        .unwrap();
        repo.materialize_write_ops(&[WorktreeWriteOp::Symlink {
            path: path.clone(),
            hash: second_hash,
            validation_root: temp_dir.path().to_path_buf(),
        }])
        .unwrap();

        assert_eq!(std::fs::read_link(&path).unwrap(), PathBuf::from("second"));
    }

    #[test]
    #[cfg(unix)]
    fn materialize_write_ops_reuses_prepared_parent_for_multiple_writes() {
        let temp_dir = TempDir::new().unwrap();
        let repo = Repository::init_default(temp_dir.path()).unwrap();

        let symlink_target = Blob::new(b"../target.txt".to_vec());
        let target_hash = repo.store().put_blob(&Blob::from("target")).unwrap();
        let symlink_hash = repo.store().put_blob(&symlink_target).unwrap();
        let base_dir = temp_dir.path().join("nested/deep");
        let target_path = base_dir.join("target.txt");
        let link_path = base_dir.join("link.txt");

        repo.materialize_write_ops(&[
            WorktreeWriteOp::Blob {
                path: target_path.clone(),
                hash: target_hash,
                executable: false,
            },
            WorktreeWriteOp::Symlink {
                path: link_path.clone(),
                hash: symlink_hash,
                validation_root: temp_dir.path().to_path_buf(),
            },
        ])
        .unwrap();

        assert_eq!(std::fs::read_to_string(&target_path).unwrap(), "target");
        assert_eq!(
            std::fs::read_link(&link_path).unwrap(),
            PathBuf::from("../target.txt")
        );
    }

    /// After `pack_objects + prune_loose_objects`, every blob is
    /// pack-only. The lazy-promotion path inside `materialize_blob`
    /// must (a) succeed without errors, (b) read back the canonical
    /// bytes in both worktrees, and (c) leave a real loose
    /// uncompressed mirror on disk under
    /// `.heddle/objects/blobs/<2-char>/<rest>` so subsequent
    /// reflinks have something to clone from.
    #[test]
    #[cfg(unix)]
    fn lazy_promotion_after_pack_and_prune_restores_loose_mirror() {
        let temp_dir = TempDir::new().unwrap();
        let repo = Repository::init_default(temp_dir.path()).unwrap();

        let blob = Blob::from(
            "lazy-promotion payload, packed-then-pruned, kept under compression threshold",
        );
        let hash = repo.store().put_blob(&blob).unwrap();

        // Move the loose copy into a packfile, then drop the loose
        // copy. The store now has only the pack-resident blob.
        repo.store().pack_objects(false).unwrap();
        repo.store().prune_loose_objects().unwrap();
        assert!(
            repo.store().loose_blob_path(&hash).is_none(),
            "after pack+prune, the canonical loose path must be empty"
        );

        let worktree_a = temp_dir.path().join("worktree-a/file.txt");
        let worktree_b = temp_dir.path().join("worktree-b/file.txt");
        repo.materialize_write_ops(&[WorktreeWriteOp::Blob {
            path: worktree_a.clone(),
            hash,
            executable: false,
        }])
        .unwrap();
        repo.materialize_write_ops(&[WorktreeWriteOp::Blob {
            path: worktree_b.clone(),
            hash,
            executable: false,
        }])
        .unwrap();

        // (a)+(b) read back ok.
        assert_eq!(std::fs::read(&worktree_a).unwrap(), blob.content());
        assert_eq!(std::fs::read(&worktree_b).unwrap(), blob.content());

        // (c) the loose-uncompressed mirror exists.
        let loose = repo
            .store()
            .loose_blob_path(&hash)
            .expect("after lazy promotion the canonical loose path must exist");
        assert_eq!(std::fs::read(&loose).unwrap(), blob.content());
    }

    /// Proactive warm: walk a state's tree, promote every reachable
    /// blob, then materialize. Every blob must be loose-uncompressed
    /// after warm so the materialize step can reflink directly
    /// without paying the decompress tax. Cross-worktree readback
    /// must give the canonical bytes.
    #[test]
    #[cfg(unix)]
    fn proactive_warm_promotes_all_state_blobs() {
        let temp_dir = TempDir::new().unwrap();
        let repo = Repository::init_default(temp_dir.path()).unwrap();

        // Materialize a few files and snapshot.
        for i in 0..4 {
            std::fs::write(
                temp_dir.path().join(format!("file-{i}.txt")),
                format!("warm-pass payload {i} {}", "x".repeat(140)),
            )
            .unwrap();
        }
        let state = repo
            .snapshot(Some("warm-pass test".to_string()), None)
            .unwrap();

        // Pack + prune so every blob is pack-only.
        repo.store().pack_objects(false).unwrap();
        repo.store().prune_loose_objects().unwrap();

        // Sanity: with a packed-then-pruned store, no canonical loose
        // file exists yet for the snapshot's blobs.
        let tree = repo.store().get_tree(&state.tree).unwrap().unwrap();
        let mut hashes = std::collections::BTreeSet::new();
        repo.collect_blob_hashes(&tree, &mut hashes).unwrap();
        for hash in &hashes {
            assert!(
                repo.store().loose_blob_path(hash).is_none(),
                "blob {} should be pack-only before warm",
                hash
            );
        }

        // Warm: every blob should now be loose-uncompressed.
        let stats = repo.warm_canonical_store_for_state(&state.id()).unwrap();
        assert_eq!(stats.errors, 0, "warm pass produced errors: {:?}", stats);
        assert_eq!(stats.total(), hashes.len());
        assert!(
            stats.promoted >= hashes.len(),
            "expected to promote all {} blobs, got {} (already_loose={})",
            hashes.len(),
            stats.promoted,
            stats.already_loose
        );
        for hash in &hashes {
            assert!(
                repo.store().loose_blob_path(hash).is_some(),
                "blob {} should be loose+uncompressed after warm",
                hash
            );
        }

        // Materialize across two worktrees on the same FS. Reading
        // back from each must yield the canonical bytes; isolation
        // is guaranteed by filesystem-level CoW (or full copy).
        let worktree_a = temp_dir.path().join("wt-a");
        let worktree_b = temp_dir.path().join("wt-b");
        repo.materialize_tree(&tree, &worktree_a).unwrap();
        repo.materialize_tree(&tree, &worktree_b).unwrap();

        for entry in tree.entries() {
            let path_a = worktree_a.join(entry.name());
            let path_b = worktree_b.join(entry.name());
            assert_eq!(
                std::fs::read(&path_a).unwrap(),
                std::fs::read(&path_b).unwrap(),
                "{} must read back identically across worktrees",
                entry.name()
            );
        }
    }

    /// Idempotent warm: a second pass over the same state must not
    /// rewrite anything. Every blob is `already_loose`.
    #[test]
    #[cfg(unix)]
    fn warm_canonical_store_is_idempotent() {
        let temp_dir = TempDir::new().unwrap();
        let repo = Repository::init_default(temp_dir.path()).unwrap();

        for i in 0..3 {
            std::fs::write(
                temp_dir.path().join(format!("idem-{i}.txt")),
                format!("idem payload {i} {}", "x".repeat(160)),
            )
            .unwrap();
        }
        let state = repo
            .snapshot(Some("idempotent warm".to_string()), None)
            .unwrap();
        repo.store().pack_objects(false).unwrap();
        repo.store().prune_loose_objects().unwrap();

        let first = repo.warm_canonical_store_for_state(&state.id()).unwrap();
        let second = repo.warm_canonical_store_for_state(&state.id()).unwrap();

        assert_eq!(first.total(), second.total(), "blob count must be stable");
        assert_eq!(
            second.promoted, 0,
            "second warm must not promote anything (got {})",
            second.promoted
        );
        assert_eq!(
            second.already_loose,
            second.total(),
            "every blob must be already_loose on second pass"
        );
        assert_eq!(second.errors, 0);
    }

    /// Storage win after warm + materialize on a CoW filesystem.
    /// We can no longer dedupe via inode (reflinks have distinct
    /// inodes by design), so on CoW filesystems we instead assert
    /// that **every materialized file has its own inode**, distinct
    /// from the canonical loose blob — proving the materializer
    /// took the reflink path (which gives the storage win on CoW
    /// without aliasing) rather than the in-memory `fs::write` path
    /// (which costs full duplicates).
    ///
    /// Ignored by default because the materializer uses `fs::copy`
    /// on non-CoW filesystems and the reflink storage win is not
    /// available there.
    #[test]
    #[cfg(unix)]
    #[ignore = "requires a reflink-capable filesystem"]
    fn packed_repo_storage_win_after_warm_and_materialize() {
        use std::{collections::HashSet, os::unix::fs::MetadataExt};

        let temp_dir = TempDir::new().unwrap();
        assert!(
            filesystem_supports_reflink(temp_dir.path()),
            "packed_repo_storage_win_after_warm_and_materialize requires reflink support"
        );

        let repo = Repository::init_default(temp_dir.path()).unwrap();

        let blob_count = 5;
        for i in 0..blob_count {
            std::fs::write(
                temp_dir.path().join(format!("file-{i}.txt")),
                format!("packed-storage-win payload {i} {}", "x".repeat(140 + i * 8)),
            )
            .unwrap();
        }
        let state = repo
            .snapshot(Some("packed storage win".to_string()), None)
            .unwrap();
        // Realistic steady state.
        repo.store().pack_objects(false).unwrap();
        repo.store().prune_loose_objects().unwrap();

        // Warm so the first materialize doesn't pay decompress cost.
        let stats = repo.warm_canonical_store_for_state(&state.id()).unwrap();
        assert_eq!(stats.errors, 0);

        let n_worktrees = 6;
        let tree = repo.store().get_tree(&state.tree).unwrap().unwrap();
        let mut all_paths = Vec::new();
        for w in 0..n_worktrees {
            let worktree = temp_dir.path().join(format!("wt-{w}"));
            repo.materialize_tree(&tree, &worktree).unwrap();
            for i in 0..blob_count {
                all_paths.push(worktree.join(format!("file-{i}.txt")));
            }
        }

        // Every materialized file has its own inode (reflinks, not
        // hardlinks). Total inodes = files materialized.
        let mut inodes = HashSet::new();
        for path in &all_paths {
            inodes.insert(std::fs::metadata(path).unwrap().ino());
        }
        assert_eq!(
            inodes.len(),
            all_paths.len(),
            "every reflinked worktree file must have its own inode (got {} for {} files)",
            inodes.len(),
            all_paths.len()
        );

        // No materialized file shares an inode with the canonical
        // loose blob — that would be the hardlink bug.
        let mut canonical_inodes = HashSet::new();
        for hash in tree.entries().iter().filter_map(|e| e.blob_hash()) {
            if let Some(loose) = repo.store().loose_blob_path(&hash) {
                canonical_inodes.insert(std::fs::metadata(&loose).unwrap().ino());
            }
        }
        for inode in &inodes {
            assert!(
                !canonical_inodes.contains(inode),
                "worktree file inode {} aliases the canonical loose blob — that's the hardlink bug",
                inode
            );
        }

        eprintln!(
            "[packed-storage-win] n_worktrees={} blobs/tree={} reflink_path_confirmed=true",
            n_worktrees, blob_count
        );
    }

    /// `promote_to_loose_uncompressed` is idempotent for an already
    /// loose+uncompressed blob — fast-path returns `Ok(false)` so a
    /// caller can distinguish "no work needed" from "promoted".
    #[test]
    fn promote_to_loose_uncompressed_idempotent_on_loose_blob() {
        let temp_dir = TempDir::new().unwrap();
        let repo = Repository::init_default(temp_dir.path()).unwrap();

        let blob = Blob::from("idempotent promote payload");
        let hash = repo.store().put_blob(&blob).unwrap();
        // Already loose+uncompressed (under compression threshold).
        assert!(repo.store().loose_blob_path(&hash).is_some());

        let did_work = repo.store().promote_to_loose_uncompressed(&hash).unwrap();
        assert!(
            !did_work,
            "promote on already-loose+uncompressed blob must be a no-op"
        );
    }

    /// `promote_to_loose_uncompressed` on a missing blob bubbles a
    /// `NotFound`, not a silent success. Callers can degrade
    /// gracefully (e.g. lazy-path falls back to `fs::write`), but
    /// the failure must not be invisible.
    #[test]
    fn promote_to_loose_uncompressed_returns_error_for_missing_blob() {
        use objects::object::ContentHash;

        let temp_dir = TempDir::new().unwrap();
        let repo = Repository::init_default(temp_dir.path()).unwrap();

        let bogus = ContentHash::compute_typed("blob", b"never-stored");
        let result = repo.store().promote_to_loose_uncompressed(&bogus);
        assert!(
            result.is_err(),
            "promote on missing blob must error, got {:?}",
            result
        );
    }
}