kache-store 1.1.0

use anyhow::{Context, Result};
use std::fs;
use std::path::{Path, PathBuf};
use std::sync::OnceLock;
use std::sync::atomic::{AtomicBool, Ordering};

/// Process-global: restore via hardlink on a non-CoW Windows volume (NTFS)
/// instead of the default copy (#429). Set once from `config.windows_hardlink`
/// at wrapper entry; read on the Windows restore path. Off everywhere else.
static WINDOWS_HARDLINK_RESTORE: AtomicBool = AtomicBool::new(false);
/// Unix opt-in: every target directory may share a store blob's inode.
static SHARED_HARDLINK_RESTORES: AtomicBool = AtomicBool::new(false);

/// Let restores hardlink a blob that another target directory already links
/// (`[cache] shared_hardlink_restores`). Call once per process before
/// restoring.
pub fn set_shared_hardlink_restores(enabled: bool) {
    SHARED_HARDLINK_RESTORES.store(enabled, Ordering::Relaxed);
}

/// Process-global: marker file used to dedup storage-layout advisories across
/// the hundreds of wrapper processes a single build spawns (#508, #835). Set
/// at wrapper entry; if unset the advisory falls back to once-per-process.
///
/// Originally Windows-only for the no-CoW advisory; #835 reuses the same
/// marker infrastructure for the Linux EXDEV/EPERM hardlink advisory, with
/// separate buckets so one severity never mutes another.
static COW_WARN_MARKER: OnceLock<PathBuf> = OnceLock::new();

/// Set the Windows hardlink-restore opt-in (from `Config::windows_hardlink`).
/// Call once per process before restoring. No effect off Windows.
pub fn set_windows_hardlink_restore(enabled: bool) {
    WINDOWS_HARDLINK_RESTORE.store(enabled, Ordering::Relaxed);
}

/// Is the `[cache] windows_hardlink` opt-in active for this process?
/// The store's insert-side blob hardlinking keys off the same flag as
/// restore: without it, a hardlink would propagate the blob's read-only
/// attribute to the build's own output (shared MFT record, #429).
#[cfg(windows)]
pub fn windows_hardlink_enabled() -> bool {
    WINDOWS_HARDLINK_RESTORE.load(Ordering::Relaxed)
}

/// Set the cross-process dedup marker for storage-layout advisories. Call once
/// per process before restoring. The marker itself is inert; each advisory
/// derives a per-bucket path via [`bucket_marker`] so severities dedup
/// independently.
pub fn set_cow_warn_marker(path: std::path::PathBuf) {
    let _ = COW_WARN_MARKER.set(path);
}

/// Process-global: surface storage-*layout* advisories ("no CoW",
/// "cross-volume") on copy-restores (#551). On by default; set once from
/// `Config::storage_layout_advice` at wrapper entry. Muting this never mutes
/// fault reports (`UnexpectedOnCowVolume`) — advice and faults have different
/// severities and different audiences.
static STORAGE_LAYOUT_ADVICE: AtomicBool = AtomicBool::new(true);

/// Set the storage-layout advisory toggle (from `Config::storage_layout_advice`).
/// Call once per process before restoring.
pub fn set_storage_layout_advice(enabled: bool) {
    STORAGE_LAYOUT_ADVICE.store(enabled, Ordering::Relaxed);
}

/// A mapped target may intentionally restore from the main-store fallback.
/// Do not suggest another volume mapping for that copy; faults still report.
static MAPPED_TARGET: AtomicBool = AtomicBool::new(false);

pub fn set_mapped_target(mapped: bool) {
    MAPPED_TARGET.store(mapped, Ordering::Relaxed);
}

/// Is `[cache] storage_layout_advice` active for this process?
fn storage_layout_advice_enabled() -> bool {
    STORAGE_LAYOUT_ADVICE.load(Ordering::Relaxed)
}

/// Process-global: send storage-layout advice to stderr (the default, for CLI
/// commands) or to the tracing log. Wrapper entries pick the log: a
/// `RUSTC_WRAPPER`'s stderr is rustc's stderr to cargo and to every tool that
/// reads it, so advice there looks like compiler output and can fail builds
/// that reject unexpected stderr (kunobi-ninja/kache#1067). Faults
/// (`UnexpectedOnCowVolume`) always use stderr.
static LAYOUT_ADVICE_TO_LOG: AtomicBool = AtomicBool::new(false);

/// Route storage-layout advice to the tracing log instead of stderr. Call once
/// per process before restoring or storing.
pub fn set_layout_advice_to_log(enabled: bool) {
    LAYOUT_ADVICE_TO_LOG.store(enabled, Ordering::Relaxed);
}

/// Where storage-layout advice goes in this process.
fn layout_advice_sink() -> crate::markers::WarnSink {
    if LAYOUT_ADVICE_TO_LOG.load(Ordering::Relaxed) {
        crate::markers::WarnSink::Log
    } else {
        crate::markers::WarnSink::Stderr
    }
}

// ── Hardlink-fallback reason classification (#835) ───────────────────────────
//
// `link(2)` failures are not one condition: EXDEV across two bind mounts of
// one filesystem, EPERM under `protected_hardlinks`, and any other errno need
// different responses (co-locate cache + build on one mount vs fix ownership
// vs investigate). This pure helper keeps that decision table unit-testable
// off Linux; callers record the matching opcount and, on Linux, emit the
// once-per-session advisory. Observability only: classification never changes
// what gets linked.

/// Why a hardlink attempt fell back to a copy, for an `io::Error` from
/// `link(2)`. Exclusive-carrier (#794) and kind-ineligible are policy
/// refusals, not io errors, and are recorded by their own call sites.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum HardlinkIoReason {
    CrossDevice,
    Permission,
    Other,
}

impl HardlinkIoReason {
    fn mapped_volume_advice(self, mapped: bool) -> bool {
        self == Self::CrossDevice && mapped
    }
}

/// Classify a `link(2)` io error kind into a copy reason. Pure so the decision
/// table is unit-testable off Linux.
pub(crate) fn classify_hardlink_io_error(kind: std::io::ErrorKind) -> HardlinkIoReason {
    if kind == std::io::ErrorKind::CrossesDevices {
        HardlinkIoReason::CrossDevice
    } else if kind == std::io::ErrorKind::PermissionDenied {
        HardlinkIoReason::Permission
    } else {
        HardlinkIoReason::Other
    }
}

#[cfg(test)]
thread_local! {
    /// Test-only `hard_link` failure injection for the restore path (#835):
    /// when set, `hardlink_or_copy` fails with this errno instead of calling
    /// `link(2)`, so the reason counter and the Linux advisory are observable
    /// without bind mounts. Thread-local so parallel `cargo test` workers do
    /// not leak into each other. Production always sees `None`.
    static INJECT_RESTORE_HARDLINK_ERROR: std::cell::Cell<Option<std::io::ErrorKind>> =
        const { std::cell::Cell::new(None) };
}

/// Enable [`INJECT_RESTORE_HARDLINK_ERROR`] for the guard's lifetime.
#[cfg(test)]
pub(crate) struct InjectRestoreHardlinkError {
    _private: (),
}

#[cfg(test)]
impl InjectRestoreHardlinkError {
    pub(crate) fn enable(kind: std::io::ErrorKind) -> Self {
        INJECT_RESTORE_HARDLINK_ERROR.with(|slot| slot.set(Some(kind)));
        Self { _private: () }
    }
}

#[cfg(test)]
impl Drop for InjectRestoreHardlinkError {
    fn drop(&mut self) {
        INJECT_RESTORE_HARDLINK_ERROR.with(|slot| slot.set(None));
    }
}

/// Injected restore `hard_link` errno, if any. Production stub is always
/// `None` so the real `link(2)` runs.
#[cfg(test)]
fn injected_restore_hardlink_error() -> Option<std::io::ErrorKind> {
    INJECT_RESTORE_HARDLINK_ERROR.with(|slot| slot.get())
}

#[cfg(not(test))]
#[inline(always)]
fn injected_restore_hardlink_error() -> Option<std::io::ErrorKind> {
    None
}

/// Attempt `link(2)`, honouring the test-only error injection seam above.
/// Returns the io error on failure so callers can classify it instead of
/// swallowing it with `.is_ok()`.
fn try_hard_link(src: &Path, dst: &Path) -> std::io::Result<()> {
    if let Some(kind) = injected_restore_hardlink_error() {
        return Err(std::io::Error::new(kind, "injected hardlink failure"));
    }
    fs::hard_link(src, dst)
}

/// Emit the Linux once-per-session hardlink advisory for EXDEV/EPERM (#835),
/// reusing the cross-process marker-dedup helper (`warn_once_per_session`)
/// that the Windows no-CoW advisory uses. Observability only: never changes
/// what gets linked. Respects `[cache] storage_layout_advice = false`, which
/// mutes layout advice but never faults.
///
/// EXDEV across two bind mounts of one filesystem is the most likely cause of
/// 0% multi-link blobs on ext4 CI: same `st_dev` is not enough, Linux refuses
/// `link()` across vfsmounts. EPERM happens under `protected_hardlinks=1`
/// when the wrapper does not own the 0o444 blob. Other errnos stay debug-only.
pub(crate) fn warn_hardlink_fallback_once(
    store_path: &Path,
    target_path: &Path,
    reason: HardlinkIoReason,
    io_err: &std::io::Error,
) {
    if !matches!(
        reason,
        HardlinkIoReason::CrossDevice | HardlinkIoReason::Permission
    ) {
        tracing::debug!(
            "hardlink failed ({}), falling back to copy: {} -> {}",
            io_err,
            store_path.display(),
            target_path.display()
        );
        return;
    }
    if !storage_layout_advice_enabled()
        || reason.mapped_volume_advice(MAPPED_TARGET.load(Ordering::Relaxed))
    {
        tracing::debug!(
            "hardlink failed ({:?}; layout advice muted): {} -> {}",
            reason,
            store_path.display(),
            target_path.display()
        );
        return;
    }
    let (bucket, message) = match reason {
        HardlinkIoReason::CrossDevice => (
            "hardlink-exdev",
            format!(
                "kache: operation fell back to COPY because hardlinking failed with EXDEV \
                 (cross-device link): the two paths are on different mounts or volumes — \
                 hardlinks cannot span them, not across drives, not across bind mounts \
                 even of one filesystem. Put the cache and build tree on the SAME \
                 mount/drive for zero-copy sharing. If this layout is intentional, \
                 silence this advice with `[cache] storage_layout_advice = false`.\n         \
                 source: {}\n         dest: {}\n         error: {}",
                store_path.display(),
                target_path.display(),
                io_err,
            ),
        ),
        HardlinkIoReason::Permission => (
            "hardlink-eperm",
            format!(
                "kache: operation fell back to COPY because hardlinking failed with EPERM \
                 (permission denied): the caller must own the source file (on Linux, \
                 fs.protected_hardlinks refuses the link otherwise). Store blobs are \
                 0o444; if the cache was imported by another uid \
                 (daemon vs wrapper, or a carried-over cache dir), linking copies. Ensure one \
                 uid owns the cache. If this layout is intentional, silence this advice with \
                 `[cache] storage_layout_advice = false`.\n         source: {}\n         \
                 dest: {}\n         error: {}",
                store_path.display(),
                target_path.display(),
                io_err,
            ),
        ),
        HardlinkIoReason::Other => {
            return;
        }
    };
    let sink = layout_advice_sink();
    match COW_WARN_MARKER.get() {
        Some(base) => {
            let marker = bucket_marker(base, bucket);
            let _warned = crate::markers::warn_once_per_session_to(
                &marker,
                crate::markers::WARN_SESSION_SECS,
                &message,
                sink,
            );
        }
        None => {
            use std::sync::Once;
            static WARNED: Once = Once::new();
            WARNED.call_once(|| sink.emit(&message));
        }
    }
}

/// Strategy for restoring a cached file to a build output path.
///
/// Every strategy first tries reflink (CoW: zero-copy *with* an independent
/// inode), then uses its own fallback:
///
/// - `Hardlink`: on Unix, fall back to a hardlink (zero-copy via shared inode)
///   only while the blob has no existing consumer. Later restores use private
///   copies so their required mtime stamp cannot mutate a still-linked artifact
///   another process is reading (#794). This is for immutable artifacts like
///   `.rlib` / `.rmeta` where the build won't modify the content in place.
/// - `Copy`: fall back to a plain byte copy (independent file). For
///   executables, dylibs, and proc-macros that may be mutated post-build
///   (codesigning, stripping, etc.).
/// - `ExecutableHardlink`: an executable or loadable library that nothing
///   rewrites in place after the restore. On Unix it falls back like
///   `Hardlink`, but only when the blob is read-only and already carries every
///   execute bit, because the shared inode cannot be chmod'ed. Any other blob,
///   and every Windows restore, is copied as `Copy` would. The caller decides
///   which artifacts qualify.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum LinkStrategy {
    Hardlink,
    Copy,
    ExecutableHardlink,
}

/// Link a cached file to the target output path.
///
/// Every strategy tries reflink first, then uses its own fallback.
pub fn link_to_target(store_path: &Path, target_path: &Path, strategy: LinkStrategy) -> Result<()> {
    link_to_target_via(store_path, target_path, strategy, try_reflink)
}

/// [`link_to_target`] with the reflink attempt passed in, so tests can run the
/// no-CoW fallbacks on a filesystem that clones.
fn link_to_target_via(
    store_path: &Path,
    target_path: &Path,
    strategy: LinkStrategy,
    reflink: fn(&Path, &Path) -> Result<()>,
) -> Result<()> {
    let do_link = || -> Result<()> {
        // Remove existing file at target (link/clone calls fail if dst exists).
        clear_target(target_path)?;

        // Logical size of the artifact, attributed to whichever restoration
        // mechanism runs below. Best-effort — a metadata failure here must
        // not fail the restore.
        let bytes = fs::metadata(store_path).map(|m| m.len()).unwrap_or(0);

        // Try reflink first. CoW gives us zero-copy *and* mutations don't
        // propagate to the cache blob — strictly better than hardlink when
        // available (APFS, btrfs, XFS-with-reflink).
        // Keep the failure reason: on Windows it separates "this volume can't
        // block-clone" from "this one file couldn't be cloned" (#508).
        let reflink_err = match reflink(store_path, target_path) {
            Ok(()) => {
                match strategy {
                    LinkStrategy::Hardlink => set_owner_write_permission(target_path)?,
                    // Executables and loadable libraries also need execute bits.
                    LinkStrategy::Copy | LinkStrategy::ExecutableHardlink => {
                        set_executable_permissions(target_path)?
                    }
                }
                tracing::debug!(
                    "reflinked {} -> {}",
                    store_path.display(),
                    target_path.display()
                );
                crate::opcounts::record_reflinked(bytes);
                return Ok(());
            }
            Err(e) => e,
        };
        #[cfg(not(windows))]
        let _ = &reflink_err;

        // Reflink unsupported on this filesystem — strategy-specific fallback.
        match strategy {
            // Windows has no reflink on NTFS, so the Hardlink strategy would
            // hardlink the read-only store blob. NTFS stores FILE_ATTRIBUTE_READONLY
            // in the shared MFT record, so EVERY hardlink to a read-only blob is
            // itself read-only — and Windows refuses to delete or rewrite a
            // read-only file (WinError 5). A consumer that owns its output and
            // deletes/rewrites it — e.g. mozbuild's configure `ar_supports_response_files`
            // conftest (#429) — then breaks. There is no way on NTFS to give a
            // hardlink a different read-only state than its blob, so restore via an
            // independent COPY instead: the output is writable and deletable while
            // the store blob stays read-only (integrity preserved). This mirrors
            // `write_restored`, already the proven-safe independent-file path, and
            // costs only working-tree↔store block sharing (LRU is index-based, not
            // mtime-based, so eviction is unaffected). gnu/clang restores keep
            // hardlinking — reflink/hardlink there are writable or CoW-isolated.
            #[cfg(windows)]
            LinkStrategy::Hardlink => {
                if WINDOWS_HARDLINK_RESTORE.load(Ordering::Relaxed) {
                    // Opt-in (#429 / `[cache] windows_hardlink`): the caller accepts
                    // that this build never deletes/rewrites a restored object, so
                    // trade the read-only-output risk for working-tree dedup.
                    hardlink_or_copy(store_path, target_path, bytes)
                } else {
                    copy_file(store_path, target_path, false)?;
                    warn_no_cow_restore_once(store_path, target_path, bytes, &reflink_err, true);
                    crate::opcounts::record_copied(bytes);
                    Ok(())
                }
            }
            #[cfg(not(windows))]
            LinkStrategy::Hardlink => hardlink_or_copy(store_path, target_path, bytes),
            // Share the inode only when the blob can already run as it is: a
            // chmod on the link would change the blob too.
            #[cfg(unix)]
            LinkStrategy::ExecutableHardlink
                if executable_blob_is_shareable(blob_mode(store_path)) =>
            {
                executable_hardlink_or_copy(store_path, target_path, bytes)
            }
            LinkStrategy::Copy | LinkStrategy::ExecutableHardlink => {
                // Copying here is by design (executables/dylibs may be mutated after
                // the build), so no storage-layout advice — but a large artifact that
                // failed to block-clone on a CoW volume is still a real fault and is
                // reported rather than swallowed.
                copy_file(store_path, target_path, true)?;
                #[cfg(windows)]
                warn_no_cow_restore_once(store_path, target_path, bytes, &reflink_err, false);
                crate::opcounts::record_copied(bytes);
                Ok(())
            }
        }
    };

    let mut err = match do_link() {
        Ok(()) => return Ok(()),
        Err(e) => e,
    };

    if err.chain().any(|e| {
        e.downcast_ref::<std::io::Error>()
            .is_some_and(|io| io.kind() == std::io::ErrorKind::NotFound)
    }) && let Some(parent) = target_path.parent()
    {
        fs::create_dir_all(parent)
            .with_context(|| format!("creating parent dir for {}", target_path.display()))?;
        if let Err(e) = do_link() {
            err = e;
        } else {
            return Ok(());
        }
    }

    Err(err)
}

/// Hardlink fallback for the `Hardlink` strategy when reflink is unavailable.
/// On Unix, keep at most one target output linked to the store blob; later
/// consumers get a private copy so their restore-time mtime stamp cannot
/// affect a still-linked active reader in another target tree (#794). Falls
/// back to a plain copy on hardlink failure (including cross-filesystem
/// restores).
///
/// On Windows this runs only under the legacy `[cache] windows_hardlink`
/// opt-in. That explicit tradeoff retains unrestricted hardlink sharing; it is
/// unsafe for concurrent builds sharing a store as well as for consumers that
/// delete or rewrite restored outputs (#429, #794). The default copies.
fn hardlink_or_copy(store_path: &Path, target_path: &Path, bytes: u64) -> Result<()> {
    hardlink_or_copy_with_prelink_hook(store_path, target_path, bytes, false, || {})
}

/// [`hardlink_or_copy`] for an executable blob: the same one-consumer rule and
/// post-link check, with a copy fallback that stays runnable (`0o755`).
#[cfg(unix)]
fn executable_hardlink_or_copy(store_path: &Path, target_path: &Path, bytes: u64) -> Result<()> {
    hardlink_or_copy_with_prelink_hook(store_path, target_path, bytes, true, || {})
}

/// Whether an `ExecutableHardlink` restore may share the blob's inode. The
/// blob must be read-only, so a writer that opens the restored file fails
/// instead of reaching the store, and must carry every execute bit, because
/// the link cannot be chmod'ed on its own. A remote import lands at `0o444`
/// and would not run from a link.
#[cfg_attr(not(unix), allow(dead_code))]
fn executable_blob_is_shareable(blob_mode: Option<u32>) -> bool {
    blob_mode.is_some_and(|mode| mode & 0o111 == 0o111 && mode & 0o222 == 0)
}

/// Permission bits of a store blob, or `None` when it cannot be read.
#[cfg(unix)]
fn blob_mode(store_path: &Path) -> Option<u32> {
    use std::os::unix::fs::PermissionsExt;
    fs::metadata(store_path)
        .ok()
        .map(|meta| meta.permissions().mode())
}

fn hardlink_or_copy_with_prelink_hook(
    store_path: &Path,
    target_path: &Path,
    bytes: u64,
    executable: bool,
    prelink_hook: impl FnOnce(),
) -> Result<()> {
    // A restore is stamped after this function returns. On Unix that stamp is
    // inode metadata, so it is safe to hardlink only when the blob has no
    // existing target consumer. Metadata failure is handled conservatively:
    // a copy preserves isolation and lets the ordinary read report any real
    // source failure.
    #[cfg_attr(not(unix), allow(unused_variables))]
    let shared = SHARED_HARDLINK_RESTORES.load(Ordering::Relaxed);
    // Every fallback below copies the same way, so an executable stays
    // runnable whichever check sends it there.
    let copy_instead = || copy_hardlink_fallback(store_path, target_path, bytes, executable);
    #[cfg(unix)]
    match fs::metadata(store_path) {
        Ok(meta) => {
            use std::os::unix::fs::MetadataExt;
            if !shared && meta.nlink() != 1 {
                tracing::debug!(
                    links = meta.nlink(),
                    "blob already has a hardlink consumer, restoring by copy: {} -> {}",
                    store_path.display(),
                    target_path.display()
                );
                crate::opcounts::record_restore_copy_exclusive(bytes);
                return copy_instead();
            }
        }
        Err(error) => {
            tracing::debug!(
                %error,
                "could not verify exclusive blob, restoring by copy: {} -> {}",
                store_path.display(),
                target_path.display()
            );
            crate::opcounts::record_restore_copy_other(bytes);
            return copy_instead();
        }
    }

    // Test-only callers use this seam to hold simultaneous restorers after
    // both prechecks, proving the post-link validation rather than merely the
    // sequential fast path. Production passes a no-op closure.
    prelink_hook();

    if let Err(e) = try_hard_link(store_path, target_path) {
        let kind = e.kind();
        let reason = classify_hardlink_io_error(kind);
        match reason {
            HardlinkIoReason::CrossDevice => {
                crate::opcounts::record_restore_copy_cross_device(bytes);
            }
            HardlinkIoReason::Permission => {
                crate::opcounts::record_restore_copy_permission(bytes);
            }
            HardlinkIoReason::Other => {
                crate::opcounts::record_restore_copy_other(bytes);
            }
        }
        // The failure is no longer silent: the reason is counted above, and
        // EXDEV/EPERM also emit the once-per-session layout advisory.
        // What gets linked is unchanged — this still falls back to a copy.
        warn_hardlink_fallback_once(store_path, target_path, reason, &e);
        return copy_instead();
    }

    // Two restorers may both observe nlink == 1 before either creates its
    // link. Only a contender that observes exactly blob + itself may retain
    // the link. Others unlink before returning, so no contender can reach its
    // mtime stamp while sharing an inode with an existing consumer.
    #[cfg(unix)]
    if !shared {
        use std::os::unix::fs::MetadataExt;
        match fs::metadata(store_path) {
            Ok(meta) if meta.nlink() == 2 => {}
            Ok(meta) => {
                tracing::debug!(
                    links = meta.nlink(),
                    "hardlink was not exclusive, restoring by copy: {} -> {}",
                    store_path.display(),
                    target_path.display()
                );
                fs::remove_file(target_path).with_context(|| {
                    format!(
                        "removing non-exclusive hardlink at {}",
                        target_path.display()
                    )
                })?;
                crate::opcounts::record_restore_copy_exclusive(bytes);
                return copy_instead();
            }
            Err(source_error) => {
                verify_orphaned_hardlink(target_path, fs::metadata(target_path), source_error)?
            }
        }
    }

    tracing::debug!(
        "hardlinked {} -> {}",
        store_path.display(),
        target_path.display()
    );
    crate::opcounts::record_hardlinked(bytes);
    Ok(())
}

fn copy_hardlink_fallback(
    store_path: &Path,
    target_path: &Path,
    bytes: u64,
    executable: bool,
) -> Result<()> {
    copy_file(store_path, target_path, executable)?;
    crate::opcounts::record_copied(bytes);
    Ok(())
}

/// Validate a completed hardlink whose store blob could not be re-stat'ed.
///
/// Between creating the link and verifying it, the blob name may vanish (a
/// concurrent eviction or another restorer's cleanup). Only when this target
/// became the inode's sole remaining name is the restore provably private and
/// safe to keep; otherwise exclusivity cannot be shown and the failure
/// surfaces. Extracted from its caller so tests can inject both observations
/// instead of racing the validation window (#794).
#[cfg(unix)]
fn verify_orphaned_hardlink(
    target_path: &Path,
    target_meta: std::io::Result<fs::Metadata>,
    source_error: std::io::Error,
) -> Result<()> {
    use std::os::unix::fs::MetadataExt;
    match target_meta {
        // The blob was unlinked after hardlink creation, leaving this
        // target as the inode's sole name. It is already private and
        // safe to stamp, so retain the successful restore.
        Ok(meta) if meta.nlink() == 1 => {
            tracing::debug!(
                %source_error,
                "blob disappeared after hardlink; retained private target {}",
                target_path.display()
            );
            Ok(())
        }
        Ok(meta) => Err(source_error).with_context(|| {
            format!(
                "verifying exclusive hardlink {} (target still has {} links)",
                target_path.display(),
                meta.nlink()
            )
        }),
        Err(target_error) => Err(target_error)
            .with_context(|| format!("verifying hardlink target {}", target_path.display())),
    }
}

/// Why a Windows cache hit was restored by COPY instead of a block-clone.
///
/// A failed `FSCTL_DUPLICATE_EXTENTS_TO_FILE` does NOT imply the volume lacks
/// copy-on-write, and conflating the two is what made kache cry wolf at every
/// Dev Drive user (#508). The three causes need three different responses:
///
/// - **`CrossVolume`**: cache blob and build output are on different volumes.
///   Block-cloning cannot span volumes (nor can hardlinks), so even two ReFS Dev
///   Drives copy. Fix: co-locate cache + build on one volume — not a filesystem
///   problem, so this is worth saying (#490).
/// - **`NoCow`**: the volume genuinely has no block-cloning (NTFS). Fix: move
///   both onto a ReFS Dev Drive, or opt into `[cache] windows_hardlink`. Worth
///   saying — this is the case that really does double disk usage.
/// - **`SubClusterOnCowVolume`**: the volume DOES block-clone, and this file is
///   smaller than one cluster, so it has no cluster-aligned range (ReFS clones
///   whole clusters; cloning past EOF is undefined). Every `.d` under ~4 KB hits
///   this on a healthy Dev Drive: all 670 warnings in #508 were sub-cluster `.d`
///   files while the `.rlib`/`.rmeta` beside them block-cloned fine. Saying
///   nothing is right — the volume is healthy and there is no advice to give.
/// - **`UnknownCow`**: the capability probe itself failed, so we do NOT know
///   whether this volume block-clones. Still worth saying — a copy-restore is
///   happening either way — but say it *honestly*: asserting "no copy-on-write"
///   from a failed probe would repeat the very mistake #508 is about.
/// - **`SubClusterOnCowVolume`**: the volume DOES block-clone, and this file is
///   smaller than one cluster, so it has no cluster-aligned range (ReFS clones
///   whole clusters; cloning past EOF is undefined). Every `.d` under ~4 KB hits
///   this on a healthy Dev Drive: all 670 warnings in #508 were sub-cluster `.d`
///   files while the `.rlib`/`.rmeta` beside them block-cloned fine. Saying
///   nothing is right — the volume is healthy and there is no advice to give.
/// - **`UnexpectedOnCowVolume`**: the volume block-clones and the file was big
///   enough to clone, yet the clone still failed. That is NOT benign — a filter
///   driver, an integrity-stream mismatch, or a kache bug could silently demote
///   every large artifact to a copy. Surface it (with the OS error) rather than
///   hiding it behind the #508 fix.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[cfg_attr(not(windows), allow(dead_code))]
enum CopyRestoreCause {
    CrossVolume,
    NoCow,
    UnknownCow,
    SubClusterOnCowVolume,
    UnexpectedOnCowVolume,
}

impl CopyRestoreCause {
    #[cfg_attr(not(windows), allow(dead_code))]
    fn mute_advice(self, mapped: bool, strategy_advice: bool, config_advice: bool) -> bool {
        (self == Self::CrossVolume && mapped)
            || (self.is_layout_advisory() && (!strategy_advice || !config_advice))
    }

    /// Which dedup bucket this advisory belongs to.
    ///
    /// A storage-*layout* advisory ("no CoW", "cross-volume") must NOT be able
    /// to mute a *fault* ("the clone failed unexpectedly") — they have different
    /// severities and different audiences, so they get separate markers. Sharing
    /// one bucket would let a benign cross-volume note swallow the report of a
    /// filter driver silently demoting every large artifact to a copy.
    #[cfg_attr(not(windows), allow(dead_code))]
    fn warn_bucket(self) -> &'static str {
        match self {
            CopyRestoreCause::UnexpectedOnCowVolume => "cow-fault",
            _ => "cow",
        }
    }

    /// Where this cause's message goes. Advice follows the process's
    /// layout-advice sink; a fault always reaches stderr, even in wrapper mode.
    #[cfg_attr(not(windows), allow(dead_code))]
    fn warn_sink(self) -> crate::markers::WarnSink {
        if self.is_layout_advisory() {
            layout_advice_sink()
        } else {
            crate::markers::WarnSink::Stderr
        }
    }

    /// Is this cause storage-layout *advice* (mutable via the `Copy` strategy
    /// or `[cache] storage_layout_advice = false`, #551) rather than a fault
    /// report? Exhaustive on purpose: a future variant must decide explicitly
    /// which side it is on — a fault must never be muteable by an advice knob.
    #[cfg_attr(not(windows), allow(dead_code))]
    fn is_layout_advisory(self) -> bool {
        match self {
            CopyRestoreCause::CrossVolume
            | CopyRestoreCause::NoCow
            | CopyRestoreCause::UnknownCow => true,
            CopyRestoreCause::SubClusterOnCowVolume | CopyRestoreCause::UnexpectedOnCowVolume => {
                false
            }
        }
    }
}

/// Classify a copy-restore from the two volume roots, the build volume's
/// block-cloning capability, and whether the file was too small to clone. Pure —
/// kept platform-independent so the decision table is unit-testable off Windows.
///
/// Unknowns are never resolved *toward silence*: a failed capability probe warns
/// as `UnknownCow` (hedged) rather than asserting NTFS, and an unknown size
/// resolves to "unexpected" rather than assuming the benign sub-cluster case.
#[cfg_attr(not(windows), allow(dead_code))]
fn classify_copy_restore(
    cache_vol: Option<&str>,
    build_vol: Option<&str>,
    build_vol_has_cow: Option<bool>,
    file_is_sub_cluster: Option<bool>,
) -> CopyRestoreCause {
    // If we can't resolve a volume for either side, don't assert cross-volume.
    if let (Some(cache), Some(build)) = (cache_vol, build_vol)
        && !cache.eq_ignore_ascii_case(build)
    {
        return CopyRestoreCause::CrossVolume;
    }
    match build_vol_has_cow {
        // Same volume, and it block-clones — the other files in this same build
        // reflink fine. Expected only when the file had no clonable range.
        Some(true) => {
            if file_is_sub_cluster == Some(true) {
                CopyRestoreCause::SubClusterOnCowVolume
            } else {
                CopyRestoreCause::UnexpectedOnCowVolume
            }
        }
        Some(false) => CopyRestoreCause::NoCow,
        None => CopyRestoreCause::UnknownCow,
    }
}

/// Tell the user their cache hits are being restored by COPY — but only when
/// that is actually a problem, and at most once per warn-session window per
/// [bucket](CopyRestoreCause::warn_bucket) across all wrapper processes (#508).
///
/// Both gates matter: a sub-cluster file on a working Dev Drive says nothing at
/// all, and the real advisories dedup through a marker file rather than a
/// per-process `Once` — each rustc is its own process, which is why a single
/// build used to emit the same warning hundreds of times.
///
/// `bytes` is the blob's size and `reflink_err` the clone failure, so a file
/// that was big enough to clone but failed anyway can still be surfaced instead
/// of being swallowed along with the benign sub-cluster case.
///
/// `layout_advice` gates the *configuration* advisories ("no CoW", "cross
/// volume"). The `Copy` strategy (executables, dylibs) always intended to copy,
/// so it passes `false`: it must not start nagging about storage layout. But it
/// still routes through here, because a LARGE artifact that fails to clone on a
/// CoW volume is a genuine fault that must not pass silently just because the
/// restore strategy happened to be `Copy`.
///
/// `[cache] storage_layout_advice = false` (#551) mutes the same advisories
/// process-wide — the user saying their layout is intentional (e.g. an
/// NTFS-only machine that can never host a ReFS Dev Drive). Like the `Copy`
/// strategy, it can never mute a fault.
#[cfg(windows)]
fn warn_no_cow_restore_once(
    store_path: &Path,
    target_path: &Path,
    bytes: u64,
    reflink_err: &anyhow::Error,
    layout_advice: bool,
) {
    let cache_vol = windows_volume_root(store_path);
    let build_vol = windows_volume_root(target_path);
    // A file with no cluster-aligned range can't be block-cloned on ANY volume.
    // If the cluster size can't be read, leave it unknown rather than guessing.
    let sub_cluster = kache_fs::windows_cluster_size(target_path)
        .ok()
        .map(|cluster| bytes > 0 && bytes < cluster);
    let cause = classify_copy_restore(
        cache_vol.as_deref(),
        build_vol.as_deref(),
        windows_volume_supports_block_clone(target_path),
        sub_cluster,
    );

    let message = match cause {
        CopyRestoreCause::SubClusterOnCowVolume => {
            // Not a filesystem problem: the volume block-clones, this file is
            // just smaller than a cluster. Debug-log it and stay quiet.
            tracing::debug!(
                "copy-restored {} — {} bytes is smaller than one cluster, so it has \
                 no cluster-aligned range to block-clone; volume {} does support \
                 copy-on-write",
                target_path.display(),
                bytes,
                build_vol.as_deref().unwrap_or("?"),
            );
            return;
        }
        CopyRestoreCause::UnexpectedOnCowVolume => format!(
            "kache: this volume ({vol}) supports copy-on-write and this file is \
             large enough to block-clone, but the clone FAILED — so the cache hit \
             was restored by COPY and does not share storage blocks with the cache. This \
             is unexpected; please report it with the error below.\n         \
             error:           {err:#}\n         affected output: {target} \
             ({bytes} bytes)",
            vol = build_vol.as_deref().unwrap_or("?"),
            err = reflink_err,
            target = target_path.display(),
            bytes = bytes,
        ),
        CopyRestoreCause::CrossVolume => format!(
            "kache: cache hits are restored by COPY because the cache and build \
             tree are on different volumes ({cache_vol} vs {build_vol}), and \
             copy-on-write block-cloning cannot span volumes — so they do not \
             share storage blocks, roughly doubling disk for cached content. Put the \
             cache and build dir on the SAME volume (ideally a ReFS Dev Drive) \
             for zero-copy dedup. If this layout is intentional, silence this \
             advice with `[cache] storage_layout_advice = false` (clone faults \
             are still reported).\n         cache blob:   {store}\n         \
             build output: {target}",
            cache_vol = cache_vol.as_deref().unwrap_or("?"),
            build_vol = build_vol.as_deref().unwrap_or("?"),
            store = store_path.display(),
            target = target_path.display(),
        ),
        CopyRestoreCause::NoCow => format!(
            "kache: this volume ({vol}) has no copy-on-write, so cache hits are \
             restored by COPY — the cache and build tree do not share storage blocks, \
             roughly doubling disk for cached content. For zero-copy dedup put \
             the cache + build dir on a ReFS Dev Drive, or set \
             `[cache] windows_hardlink = true` (only if your build never \
             deletes or rewrites an object output). If this layout is \
             intentional, silence this advice with \
             `[cache] storage_layout_advice = false` (clone faults are still \
             reported).\n         affected output: {target}",
            vol = build_vol.as_deref().unwrap_or("NTFS"),
            target = target_path.display(),
        ),
        // Probe failed: we do NOT know what this volume supports. Say exactly
        // that — claiming "no copy-on-write" here would be the same unfounded
        // assertion that produced #508 in the first place.
        CopyRestoreCause::UnknownCow => format!(
            "kache: cache hits are restored by COPY, so the cache and build tree \
             do not share storage blocks. kache could not determine whether this volume \
             ({vol}) supports copy-on-write — the capability probe failed. For \
             zero-copy dedup the cache + build dir must be on the same ReFS Dev \
             Drive. If this layout is intentional, silence this advice with \
             `[cache] storage_layout_advice = false` (clone faults are still \
             reported).\n         probe error:     {err:#}\n         affected output: \
             {target}",
            vol = build_vol.as_deref().unwrap_or("?"),
            err = reflink_err,
            target = target_path.display(),
        ),
    };

    // The Copy strategy always meant to copy — it must not nag about storage
    // layout — and `[cache] storage_layout_advice = false` is the user saying
    // their layout is intentional (#551). Either way this mutes only the
    // *advice*: a genuine clone fault still reports (see `layout_advice`).
    if cause.mute_advice(
        MAPPED_TARGET.load(Ordering::Relaxed),
        layout_advice,
        storage_layout_advice_enabled(),
    ) {
        tracing::debug!(
            "copy-restored {} ({:?}; layout advice muted: strategy={}, config={})",
            target_path.display(),
            cause,
            !layout_advice,
            !storage_layout_advice_enabled(),
        );
        return;
    }

    let sink = cause.warn_sink();
    match COW_WARN_MARKER.get() {
        Some(base) => {
            // Separate bucket per severity: a layout advisory must never mute a
            // fault report (and vice versa).
            let marker = bucket_marker(base, cause.warn_bucket());
            let _warned = crate::markers::warn_once_per_session_to(
                &marker,
                crate::markers::WARN_SESSION_SECS,
                &message,
                sink,
            );
        }
        // No marker configured (unit tests, non-wrapper entrypoints): fall back
        // to once-per-process rather than going silent.
        None => {
            use std::sync::Once;
            static WARNED: Once = Once::new();
            WARNED.call_once(|| sink.emit(&message));
        }
    }
}

/// Derive a per-bucket marker path from the base marker (`…/kache-cow-warn-<hash>`
/// → `…/kache-cow-warn-<hash>.<bucket>`), so advisories of different severity
/// dedup independently.
#[cfg_attr(not(windows), allow(dead_code))]
fn bucket_marker(base: &Path, bucket: &str) -> std::path::PathBuf {
    let mut name = base.as_os_str().to_os_string();
    name.push(".");
    name.push(bucket);
    std::path::PathBuf::from(name)
}

/// Does the volume holding `path` support ReFS block-cloning (copy-on-write)?
/// `None` if the capability can't be determined.
///
/// This is the question `warn_no_cow_restore_once` actually needs answered —
/// asking the volume directly, rather than inferring "no CoW" from one failed
/// clone of one file (#508).
#[cfg(windows)]
fn windows_volume_supports_block_clone(path: &Path) -> Option<bool> {
    use std::os::windows::ffi::OsStrExt;
    use windows_sys::Win32::Storage::FileSystem::GetVolumeInformationW;
    use windows_sys::Win32::System::SystemServices::FILE_SUPPORTS_BLOCK_REFCOUNTING;

    let root = windows_volume_root(path)?;
    let wide: Vec<u16> = std::ffi::OsStr::new(&root)
        .encode_wide()
        .chain(Some(0))
        .collect();
    let mut flags: u32 = 0;
    let ok = unsafe {
        GetVolumeInformationW(
            wide.as_ptr(),
            std::ptr::null_mut(),
            0,
            std::ptr::null_mut(),
            std::ptr::null_mut(),
            &mut flags,
            std::ptr::null_mut(),
            0,
        )
    };
    if ok == 0 {
        return None;
    }
    Some(flags & FILE_SUPPORTS_BLOCK_REFCOUNTING != 0)
}

/// Volume mount root (e.g. `C:\` or `D:\`) that holds `path`, or `None` if it
/// can't be resolved. `path` need not exist; its nearest existing parent volume
/// is used. Used to tell a cross-volume copy-restore from a no-CoW one (#490).
#[cfg(windows)]
fn windows_volume_root(path: &Path) -> Option<String> {
    use std::os::windows::ffi::OsStrExt;
    use windows_sys::Win32::Storage::FileSystem::GetVolumePathNameW;

    let wide: Vec<u16> = path.as_os_str().encode_wide().chain(Some(0)).collect();
    let mut root = [0u16; 260];
    let ok = unsafe { GetVolumePathNameW(wide.as_ptr(), root.as_mut_ptr(), root.len() as u32) };
    if ok == 0 {
        return None;
    }
    let len = root.iter().position(|&c| c == 0).unwrap_or(root.len());
    Some(String::from_utf16_lossy(&root[..len]))
}

/// Grant owner write access on an independent restore, retaining other bits.
/// Call only after the target is known to have its own inode.
fn set_owner_write_permission(path: &Path) -> Result<()> {
    let mut permissions = fs::metadata(path)?.permissions();
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        permissions.set_mode(permissions.mode() | 0o200);
    }
    #[cfg(not(unix))]
    permissions.set_readonly(false);
    fs::set_permissions(path, permissions)
        .with_context(|| format!("setting owner write permission on {}", path.display()))
}

/// Set 0o755 after an executable/dylib reflink. Reflink preserves the store
/// blob's read-only mode, but runtime-loaded artifacts must be executable.
fn set_executable_permissions(path: &Path) -> Result<()> {
    kache_fs::set_writable_permissions(path, true)
        .with_context(|| format!("setting executable permissions on {}", path.display()))
}

/// Clone a file using the shared filesystem implementation.
pub fn try_reflink(src: &Path, dst: &Path) -> Result<()> {
    kache_fs::try_reflink(src, dst).map_err(Into::into)
}

/// Regular file copy with appropriate permissions.
/// `executable`: if true, sets 0o755 (rwxr-xr-x); otherwise 0o644 (rw-r--r--).
fn copy_file(src: &Path, dst: &Path, executable: bool) -> Result<()> {
    kache_fs::copy_writable(src, dst, executable)
        .with_context(|| format!("copying {} to {}", src.display(), dst.display()))?;
    tracing::debug!("copied {} -> {}", src.display(), dst.display());
    Ok(())
}

/// A fully-written C/C++ cache artifact awaiting publication.
///
/// The staging file is created in the target directory with the same requested
/// mode as a compiler output (`0666`). The kernel therefore applies the current
/// umask and any inherited default ACL exactly where the final file will live.
/// No metadata operation is performed through the final pathname.
///
/// When the filesystem can share extents, the cached bytes are cloned into that
/// stage (or onto a new file in the same directory that then takes the stage's
/// mode). The clone has its own inode. A hardlink of the blob is discarded and
/// the bytes are copied instead, so a later rewrite cannot change the cache.
pub struct PreparedWritableTarget {
    stage: WritableStage,
    target: PathBuf,
    bytes: u64,
    source: Option<PathBuf>,
    reflinked: bool,
}

enum WritableStage {
    Staged(kache_fs::StagedFile),
    Cloned(ClonedFile),
}

/// A clone created under a temporary name. Dropping an unpublished clone
/// removes that name and leaves the cache blob alone.
struct ClonedFile {
    path: PathBuf,
}

impl Drop for ClonedFile {
    fn drop(&mut self) {
        if !self.path.as_os_str().is_empty() {
            let _ = fs::remove_file(&self.path);
        }
    }
}

impl ClonedFile {
    fn persist_noclobber(mut self, target: &Path) -> std::io::Result<()> {
        fs::hard_link(&self.path, target)?;
        let _ = fs::remove_file(&self.path);
        self.path.clear();
        Ok(())
    }

    fn replace(mut self, target: &Path) -> std::io::Result<()> {
        fs::rename(&self.path, target)?;
        self.path.clear();
        Ok(())
    }
}

impl PreparedWritableTarget {
    pub fn target(&self) -> &Path {
        &self.target
    }

    pub fn publish(self) -> Result<()> {
        let PreparedWritableTarget {
            stage,
            target,
            bytes,
            source,
            reflinked,
        } = self;
        match stage {
            WritableStage::Staged(staged) => staged.publish_new(&target).with_context(|| {
                format!(
                    "publishing cc output without replacing {}",
                    target.display()
                )
            })?,
            WritableStage::Cloned(cloned) => {
                cloned.persist_noclobber(&target).with_context(|| {
                    format!(
                        "publishing cc output without replacing {}",
                        target.display()
                    )
                })?
            }
        }
        record_published(reflinked, bytes, source.as_deref(), &target);
        Ok(())
    }

    /// Atomically replace an existing ordinary compiler output.
    ///
    /// The caller must first establish that the target is a private, writable
    /// regular file. Special paths retain the no-clobber path above and are
    /// handled by the selected compiler instead.
    pub fn publish_replacing(self) -> Result<()> {
        let PreparedWritableTarget {
            stage,
            target,
            bytes,
            source,
            reflinked,
        } = self;
        match stage {
            WritableStage::Staged(staged) => staged
                .replace(&target)
                .with_context(|| format!("publishing cc output over {}", target.display()))?,
            WritableStage::Cloned(cloned) => cloned
                .replace(&target)
                .with_context(|| format!("publishing cc output over {}", target.display()))?,
        }
        record_published(reflinked, bytes, source.as_deref(), &target);
        Ok(())
    }
}

fn record_published(reflinked: bool, bytes: u64, source: Option<&Path>, target: &Path) {
    if reflinked {
        if let Some(source) = source {
            tracing::debug!("reflinked {} -> {}", source.display(), target.display());
        }
        crate::opcounts::record_reflinked(bytes);
    } else {
        crate::opcounts::record_copied(bytes);
    }
}

fn new_writable_staging_file(target: &Path) -> Result<kache_fs::StagedFile> {
    let parent = target
        .parent()
        .filter(|path| !path.as_os_str().is_empty())
        .unwrap_or_else(|| Path::new("."));
    kache_fs::StagedFile::new_in(parent, ".kache-cc-restore-")
        .with_context(|| format!("creating cc restore staging file in {}", parent.display()))
}

/// Prepare cached file bytes without touching an existing target entry.
pub fn prepare_writable_target_from_file(
    src: &Path,
    target: &Path,
) -> Result<PreparedWritableTarget> {
    prepare_writable_target_from_file_via(
        src,
        target,
        kache_fs::try_clone_extents_into,
        kache_fs::try_reflink,
    )
}

fn prepare_writable_target_from_file_via(
    src: &Path,
    target: &Path,
    clone_into: fn(&Path, &fs::File) -> std::io::Result<()>,
    clone_new: fn(&Path, &Path) -> std::io::Result<()>,
) -> Result<PreparedWritableTarget> {
    let bytes = fs::metadata(src)
        .with_context(|| format!("reading cached cc artifact {}", src.display()))?
        .len();
    let mut staged = new_writable_staging_file(target)?;
    if clone_into(src, staged.file_mut()).is_ok() {
        return Ok(prepared_reflink(staged, target, src, bytes));
    }
    if let Some(cloned) = private_clone_file(src, target, clone_new)? {
        let permissions = staged
            .file_mut()
            .metadata()
            .with_context(|| format!("reading restore stage mode for {}", target.display()))?
            .permissions();
        if let Err(error) = fs::set_permissions(&cloned, permissions) {
            let _ = fs::remove_file(&cloned);
            tracing::debug!(
                %error,
                "cloned cc output kept the blob mode, restoring by copy: {} -> {}",
                src.display(),
                target.display()
            );
        } else {
            return Ok(PreparedWritableTarget {
                stage: WritableStage::Cloned(ClonedFile { path: cloned }),
                target: target.to_path_buf(),
                bytes,
                source: Some(src.to_path_buf()),
                reflinked: true,
            });
        }
    }

    rewind_stage(&mut staged)
        .with_context(|| format!("rewinding cc restore stage for {}", target.display()))?;
    let mut source = fs::File::open(src)
        .with_context(|| format!("opening cached cc artifact {}", src.display()))?;
    let copied = std::io::copy(&mut source, staged.writer()).with_context(|| {
        format!(
            "copying cached cc artifact {} for {}",
            src.display(),
            target.display()
        )
    })?;
    Ok(PreparedWritableTarget {
        stage: WritableStage::Staged(staged),
        target: target.to_path_buf(),
        bytes: copied,
        source: None,
        reflinked: false,
    })
}

fn prepared_reflink(
    staged: kache_fs::StagedFile,
    target: &Path,
    src: &Path,
    bytes: u64,
) -> PreparedWritableTarget {
    PreparedWritableTarget {
        stage: WritableStage::Staged(staged),
        target: target.to_path_buf(),
        bytes,
        source: Some(src.to_path_buf()),
        reflinked: true,
    }
}

/// Clone `src` to a new file in `target`'s directory.
///
/// `None` means the caller should copy. A clone that shares `src`'s inode is
/// removed when it is an extra name, and is not returned.
fn private_clone_file(
    src: &Path,
    target: &Path,
    clone_new: fn(&Path, &Path) -> std::io::Result<()>,
) -> Result<Option<PathBuf>> {
    let parent = stage_parent(target);
    for _ in 0..8 {
        let dest = temp_clone_path(parent);
        match clone_new(src, &dest) {
            Ok(()) => return Ok(accepted_private_clone(src, &dest)),
            Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => continue,
            Err(_) => return Ok(None),
        }
    }
    Ok(None)
}

fn accepted_private_clone(src: &Path, dest: &Path) -> Option<PathBuf> {
    let Ok(dst_meta) = fs::metadata(dest) else {
        let _ = fs::remove_file(dest);
        return None;
    };
    let Ok(src_meta) = fs::metadata(src) else {
        // The clone moved the blob onto `dest`. Put that name back so the
        // copy still has a source. Deleting `dest` would drop the only copy.
        let _ = fs::rename(dest, src);
        return None;
    };
    if !same_file(&src_meta, &dst_meta) {
        return Some(dest.to_path_buf());
    }
    if link_count(&dst_meta) > 1 {
        let _ = fs::remove_file(dest);
    }
    tracing::debug!(
        "cc restore clone shared the blob inode, copying {} -> {}",
        src.display(),
        dest.display()
    );
    None
}

fn rewind_stage(staged: &mut kache_fs::StagedFile) -> std::io::Result<()> {
    use std::io::{Seek, SeekFrom};
    let file = staged.file_mut();
    file.set_len(0)?;
    file.seek(SeekFrom::Start(0))?;
    Ok(())
}

fn stage_parent(target: &Path) -> &Path {
    target
        .parent()
        .filter(|path| !path.as_os_str().is_empty())
        .unwrap_or_else(|| Path::new("."))
}

fn temp_clone_path(directory: &Path) -> PathBuf {
    static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
    let n = NEXT.fetch_add(1, Ordering::Relaxed);
    directory.join(format!(".kache-cc-clone-{}-{n}", std::process::id()))
}

fn same_file(left: &fs::Metadata, right: &fs::Metadata) -> bool {
    #[cfg(unix)]
    {
        use std::os::unix::fs::MetadataExt;
        left.dev() == right.dev() && left.ino() == right.ino()
    }
    // Windows `Metadata` has no stable file index. Clones come from
    // `try_reflink`, which creates a new file rather than a hardlink.
    #[cfg(not(unix))]
    {
        let _ = (left, right);
        false
    }
}

fn link_count(meta: &fs::Metadata) -> u64 {
    #[cfg(unix)]
    {
        std::os::unix::fs::MetadataExt::nlink(meta)
    }
    #[cfg(not(unix))]
    {
        let _ = meta;
        1
    }
}

/// Prepare transformed C/C++ bytes with the same absent-only guarantee.
pub fn prepare_writable_target_from_bytes(
    target: &Path,
    content: &[u8],
) -> Result<PreparedWritableTarget> {
    use std::io::Write;

    let mut staged = new_writable_staging_file(target)?;
    staged
        .writer()
        .write_all(content)
        .with_context(|| format!("writing staged cc output for {}", target.display()))?;
    Ok(PreparedWritableTarget {
        stage: WritableStage::Staged(staged),
        target: target.to_path_buf(),
        bytes: content.len() as u64,
        source: None,
        reflinked: false,
    })
}

/// Remove any file already at `target_path` so a fresh clone / hardlink /
/// write can take its place. A previous restore may have left a
/// read-only hardlink or reflink of a store blob here.
fn clear_target(target_path: &Path) -> Result<()> {
    if target_path.exists() || target_path.symlink_metadata().is_ok() {
        #[cfg(windows)]
        if let Ok(meta) = fs::metadata(target_path) {
            let mut perms = meta.permissions();
            perms.set_readonly(false);
            let _ = fs::set_permissions(target_path, perms);
        }
        fs::remove_file(target_path)
            .with_context(|| format!("removing existing file at {}", target_path.display()))?;
    }
    Ok(())
}

/// Materialize a restored artifact from content computed in memory.
///
/// Used when a post-restore content transform changed the bytes (dep-info
/// path expansion): the final content is written as a fresh, independent,
/// writable file. By construction it shares no inode with the store blob
/// and is not read-only — this is the "compute the final bytes, then
/// materialize" path, as opposed to linking the blob and patching it in
/// place (which fails on a read-only or inode-shared restore).
///
/// `strategy` mirrors [`link_to_target`]: `Copy` and `ExecutableHardlink` are
/// the OS-loadable set (executables, dylibs) and yield `0o755` so cargo / the
/// OS can run or load the result; `Hardlink` (dep-info `.d` and other
/// immutable kinds) yields `0o644`. Keeping the same strategy ⟺ executable
/// proxy in both restore primitives means the "executables stay executable" contract
/// holds no matter which path materializes the file — including a future
/// content transform applied to an executable artifact (issue #298).
pub fn write_restored(target_path: &Path, content: &[u8], strategy: LinkStrategy) -> Result<()> {
    let do_write = || -> Result<()> {
        clear_target(target_path)?;
        fs::write(target_path, content)
            .with_context(|| format!("writing restored file {}", target_path.display()))?;
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let mode = if matches!(
                strategy,
                LinkStrategy::Copy | LinkStrategy::ExecutableHardlink
            ) {
                0o755
            } else {
                0o644
            };
            fs::set_permissions(target_path, fs::Permissions::from_mode(mode))
                .with_context(|| format!("setting perms on {}", target_path.display()))?;
        }
        #[cfg(not(unix))]
        let _ = strategy;
        Ok(())
    };

    let mut err = match do_write() {
        Ok(()) => return Ok(()),
        Err(e) => e,
    };

    if err.chain().any(|e| {
        e.downcast_ref::<std::io::Error>()
            .is_some_and(|io| io.kind() == std::io::ErrorKind::NotFound)
    }) && let Some(parent) = target_path.parent()
    {
        fs::create_dir_all(parent)
            .with_context(|| format!("creating parent dir for {}", target_path.display()))?;
        if let Err(e) = do_write() {
            err = e;
        } else {
            return Ok(());
        }
    }

    Err(err)
}

/// Stamp a restored file's mtime as "written now", through the same clock
/// the filesystem stamps writes with: `futimens(UTIME_NOW)` on unix, a
/// sibling write-clock sample elsewhere — never a bare `FileTime::now()`.
///
/// The stamp itself is required: cargo re-runs build scripts in a cleaned
/// tree, and its `StaleDependency` freshness rule compares those fresh run
/// outputs against our restored artifacts' mtimes — a restored file older
/// than its unit's build-script `output` is permanently dirty. So restored
/// files must read as "written now".
///
/// The clock choice is the subtle part (kunobi-ninja/kache#677, the #135
/// "flake"): `filetime::FileTime::now()` samples the precise realtime clock,
/// which on Linux runs AHEAD of the coarse clock the kernel stamps file
/// writes with (by up to a tick — the observed inversions are
/// sub-millisecond). A precise-clock touch can therefore postdate files the
/// build tool writes *after* the restore (a build script's `output`), which
/// cargo reads as `StaleDependency` — identical back-to-back builds never
/// reach a no-op. On Linux, `UTIME_NOW` resolves in the kernel through the
/// same timestamp source as writes (fs/utimes.c -> inode current-time), so
/// the restored file's mtime is >= everything written before it and <=
/// everything written after it — exactly the ordering cargo needs. That is a
/// verified kernel property on local Linux filesystems only: macOS resolves
/// `UTIME_NOW` in libsyscall via `gettimeofday` (microseconds, empirically
/// green in the e2e suite on APFS but not the same mechanism), and network /
/// FUSE filesystems make no such guarantee.
pub fn touch_mtime_write_clock(path: &Path) -> Result<()> {
    // Any failure below propagates to `restore_from_cache`'s callers, which
    // treat it as a clean miss and recompile (never a failed build) — so
    // fail-closed checks here cost a recompile, not correctness.
    #[cfg(unix)]
    {
        use std::os::fd::{AsRawFd, FromRawFd};
        use std::os::unix::ffi::OsStrExt;

        // Stamp through an fd rather than the path (kunobi-ninja/kache#682):
        // `O_NOFOLLOW` fails closed if the artifact was swapped for a symlink
        // between materialization and stamp, and `fstat` proves the object is
        // still a regular file before it is re-dated. `O_RDONLY` is
        // deliberate — restored hardlinks may be read-only (0444), and
        // `futimens`'s permission check for an OMIT/NOW pair is against the
        // inode (owner or privileged), not the fd's open mode.
        let cpath = std::ffi::CString::new(path.as_os_str().as_bytes())
            .with_context(|| format!("path contains NUL: {}", path.display()))?;
        // O_NONBLOCK: a swapped-in FIFO would otherwise block `open`
        // forever; for a regular file the flag is a no-op.
        let raw = unsafe {
            libc::open(
                cpath.as_ptr(),
                libc::O_RDONLY | libc::O_NOFOLLOW | libc::O_CLOEXEC | libc::O_NONBLOCK,
            )
        };
        if raw < 0 {
            return Err(std::io::Error::last_os_error())
                .with_context(|| format!("opening {} to touch it", path.display()));
        }
        let fd = unsafe { std::os::fd::OwnedFd::from_raw_fd(raw) };

        let mut st: libc::stat = unsafe { std::mem::zeroed() };
        if unsafe { libc::fstat(fd.as_raw_fd(), &mut st) } != 0 {
            return Err(std::io::Error::last_os_error())
                .with_context(|| format!("stat of {} before touch", path.display()));
        }
        if st.st_mode & libc::S_IFMT != libc::S_IFREG {
            anyhow::bail!(
                "refusing to touch {}: not a regular file (mode {:o})",
                path.display(),
                st.st_mode
            );
        }

        // atime is left untouched (UTIME_OMIT); mtime gets the write clock.
        let times = [
            libc::timespec {
                tv_sec: 0,
                tv_nsec: libc::UTIME_OMIT,
            },
            libc::timespec {
                tv_sec: 0,
                tv_nsec: libc::UTIME_NOW,
            },
        ];
        if unsafe { libc::futimens(fd.as_raw_fd(), times.as_ptr()) } != 0 {
            return Err(std::io::Error::last_os_error())
                .with_context(|| format!("updating mtime of {}", path.display()));
        }
    }

    // Non-unix targets have no UTIME_NOW, so sample the filesystem's write
    // clock directly: write a sibling temp file in the same directory (same
    // volume — timestamp behavior is per-filesystem), read back the mtime it
    // was stamped with, and apply that value to the target. Sampling BEFORE
    // stamping is the safe direction: the stamp can only sit slightly in the
    // past relative to later writes, never ahead of them — the #677/#681
    // inversion was the precise clock stamping AHEAD of write timestamps
    // (measured ~1.8 ms on NTFS by the first Windows CI run of the ordering
    // test below). A sampling failure PROPAGATES: falling back to
    // `FileTime::now()` would silently reintroduce the known-bad precise
    // stamp, whereas an error here degrades to a clean recompile at the
    // restore call sites.
    #[cfg(not(unix))]
    {
        let now = sample_write_clock(path)
            .with_context(|| format!("sampling write clock next to {}", path.display()))?;

        // On Windows, hardlinked files share permissions with the store
        // blob, which is read-only. Temporarily make it writable to update
        // the mtime, then restore the read-only flag.
        #[cfg(windows)]
        {
            let meta = fs::metadata(path)?;
            let was_readonly = meta.permissions().readonly();
            if was_readonly {
                let mut perms = meta.permissions();
                perms.set_readonly(false);
                fs::set_permissions(path, perms)?;
            }
            let result = filetime::set_file_mtime(path, now);
            if was_readonly {
                let mut perms = fs::metadata(path)?.permissions();
                perms.set_readonly(true);
                let _ = fs::set_permissions(path, perms);
            }
            result.with_context(|| format!("updating mtime of {}", path.display()))?;
        }

        #[cfg(not(windows))]
        filetime::set_file_mtime(path, now)
            .with_context(|| format!("updating mtime of {}", path.display()))?;
    }

    Ok(())
}

/// Sample the write clock of the filesystem holding `path`: create a sibling
/// temp file (exclusively — never truncating something that already exists),
/// write to it, close it, and read the mtime the filesystem gave that write.
/// The close between write and stat matters on Windows, where the last-write
/// time of an open handle may not be visible to a path-based stat until the
/// handle closes. Returns an error on failure (e.g. the directory is not
/// writable); the caller propagates it, degrading to a recompile.
#[cfg(not(unix))]
fn sample_write_clock(path: &Path) -> Result<filetime::FileTime> {
    use std::io::Write;
    use std::sync::atomic::{AtomicU64, Ordering};
    static COUNTER: AtomicU64 = AtomicU64::new(0);

    let dir = path
        .parent()
        .filter(|p| !p.as_os_str().is_empty())
        .unwrap_or_else(|| Path::new("."));

    // `create_new` refuses to open an existing file (and does not follow a
    // symlink/reparse point squatting on the name), so a PID-reuse leftover
    // or a planted file can never be truncated or sampled — the attempt
    // just moves to the next name. Leftovers from a crash between create
    // and remove are benign litter under a distinctive dotted name.
    let mut last_err: Option<std::io::Error> = None;
    for _ in 0..8 {
        let probe = dir.join(format!(
            ".kache-clock-{}-{}",
            std::process::id(),
            COUNTER.fetch_add(1, Ordering::Relaxed)
        ));
        let created = fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&probe);
        let mut file = match created {
            Ok(f) => f,
            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => continue,
            Err(e) => {
                last_err = Some(e);
                break;
            }
        };
        let sampled = (|| {
            file.write_all(b"t")?;
            drop(file);
            let meta = fs::metadata(&probe)?;
            Ok::<_, std::io::Error>(filetime::FileTime::from_last_modification_time(&meta))
        })();
        let _ = fs::remove_file(&probe);
        return sampled.with_context(|| format!("write-clock probe {}", probe.display()));
    }
    Err(last_err
        .map(anyhow::Error::from)
        .unwrap_or_else(|| anyhow::anyhow!("no free probe name after 8 attempts")))
    .with_context(|| format!("creating write-clock probe in {}", dir.display()))
}

const DEPINFO_ROOT_SENTINEL: &str = "__kache_root__/";
const DEPINFO_CWD_SENTINEL: &str = "__kache_cwd__/";
const DEPINFO_WORKSPACE_SENTINEL: &str = "__kache_workspace__/";
const DEPINFO_TARGET_SENTINEL: &str = "__kache_target_rule__/";

/// rustc's `# env-dep:NAME=value` records carry their own escape grammar:
/// backslash is written `\\` (and newline/CR as `\n`/`\r`), and cargo's
/// parser hard-rejects every other backslash sequence with
/// "unknown escape character". A path prefix can only contribute the
/// backslash case.
const DEPINFO_ENV_DEP_PREFIX: &str = "# env-dep:";

fn escape_depinfo_env_value(prefix: &str) -> String {
    prefix.replace('\\', "\\\\")
}

fn escape_depinfo_make_path(prefix: &str) -> String {
    prefix.replace(' ', "\\ ")
}

/// Pure dep-info path rewrite: relativize absolute project paths to a
/// kache-only sentinel, or expand that sentinel back to absolute paths.
/// No I/O.
///
/// This is the in-memory half of the transform. The restore side calls
/// it directly — it computes the final `.d` content from the store blob
/// and materializes the result with [`write_restored`], so it never
/// rewrites a restored, possibly read-only, possibly inode-shared file in
/// place. The Rust store side calls the two-root variant while creating a
/// private store snapshot; C/C++ keeps using this one-root form.
pub fn rewrite_depinfo_content(content: &str, project_dir: &Path, mode: DepInfoMode) -> String {
    rewrite_depinfo_content_with_sentinel(content, project_dir, DEPINFO_ROOT_SENTINEL, mode)
}

/// Rust dep-info carries two independently relocatable path families: target
/// outputs and sources below Cargo's invocation directory. Re-root both with
/// distinct sentinels so a cached `.d` never leaves Cargo watching a live donor
/// worktree after the artifact moves (#760).
#[cfg(any(test, feature = "test-support"))]
#[doc(hidden)]
pub fn rewrite_rustc_depinfo_content(
    content: &str,
    target_dir: &Path,
    working_dir: &Path,
    workspace_dir: Option<&Path>,
    mode: DepInfoMode,
) -> String {
    rewrite_rustc_depinfo_content_with_configured_roots(
        content,
        target_dir,
        working_dir,
        workspace_dir,
        &[],
        mode,
    )
}

pub fn rewrite_rustc_depinfo_content_with_configured_roots(
    content: &str,
    target_dir: &Path,
    working_dir: &Path,
    workspace_dir: Option<&Path>,
    configured_roots: &[(PathBuf, String, u8)],
    mode: DepInfoMode,
) -> String {
    let mut automatic_roots = vec![(
        working_dir.to_path_buf(),
        DEPINFO_CWD_SENTINEL.to_string(),
        4_u8,
        2_u8,
    )];
    if let Some(workspace_dir) = workspace_dir {
        automatic_roots.push((
            workspace_dir.to_path_buf(),
            DEPINFO_WORKSPACE_SENTINEL.to_string(),
            4_u8,
            1_u8,
        ));
    }

    if matches!(mode, DepInfoMode::Relativize) {
        let mut roots = configured_roots
            .iter()
            .map(|(root, sentinel, priority)| {
                (root.clone(), sentinel.clone(), *priority, 0_u8, true)
            })
            .chain(
                automatic_roots
                    .iter()
                    .map(|(root, sentinel, priority, tie)| {
                        (root.clone(), sentinel.clone(), *priority, *tie, false)
                    }),
            )
            .collect::<Vec<_>>();
        roots.sort_by_key(|(root, _, priority, tie, exact)| {
            std::cmp::Reverse((
                *priority,
                u8::from(*exact),
                root.components().count(),
                root.as_os_str().len(),
                *tie,
            ))
        });
        let target_rewritten = rewrite_rustc_depinfo_targets(content, target_dir, mode);
        // Dependencies generated beneath Cargo's effective target directory
        // must move with that target even when it lives inside the workspace.
        // Otherwise the higher-ranked WORKSPACE source rule captures
        // `target_1/.../OUT_DIR/private.rs`, and a parallel target_2 consumer
        // restores the donor path (#808). Targets on the left have already
        // become DEPINFO_ROOT_SENTINEL, so this only claims source inputs.
        let target_rewritten = rewrite_depinfo_content_with_sentinel(
            &target_rewritten,
            target_dir,
            DEPINFO_TARGET_SENTINEL,
            mode,
        );
        return roots.into_iter().fold(
            target_rewritten,
            |rewritten, (root, sentinel, _, _, exact)| {
                if exact {
                    rewrite_depinfo_content_with_exact_sentinel(&rewritten, &root, &sentinel, mode)
                } else {
                    rewrite_depinfo_content_with_sentinel(&rewritten, &root, &sentinel, mode)
                }
            },
        );
    }

    // A portable owner may have multiple producer aliases, but the first entry
    // for its sentinel is the designated lexical consumer root. Expand each
    // sentinel exactly once so a longer canonical alias cannot steal restore.
    // Expand the effective target before configured roots. A configured root
    // list can contain the same target sentinel with a stale/relative lexical
    // spelling; the argv-derived target_dir is the authoritative consumer.
    let mut rewritten =
        rewrite_depinfo_content_with_sentinel(content, target_dir, DEPINFO_TARGET_SENTINEL, mode);
    let mut expanded = std::collections::HashSet::new();
    for (root, sentinel, _) in configured_roots {
        if expanded.insert(sentinel) {
            rewritten =
                rewrite_depinfo_content_with_exact_sentinel(&rewritten, root, sentinel, mode);
        }
    }
    let rewritten =
        automatic_roots
            .into_iter()
            .fold(rewritten, |rewritten, (root, sentinel, _, _)| {
                rewrite_depinfo_content_with_sentinel(&rewritten, &root, &sentinel, mode)
            });
    rewrite_rustc_depinfo_targets(&rewritten, target_dir, mode)
}

/// Rebase only Makefile target tokens (left of rustc's `: ` delimiter).
/// Dependencies on the right are source inputs and are handled separately:
/// effective-target sources first, then PathNormalizer ownership.
fn rewrite_rustc_depinfo_targets(content: &str, target_dir: &Path, mode: DepInfoMode) -> String {
    content
        .split_inclusive('\n')
        .map(|line| {
            if line.starts_with(DEPINFO_ENV_DEP_PREFIX) {
                return line.to_string();
            }
            let Some(delimiter) = line.find(": ") else {
                return line.to_string();
            };
            let (targets, dependencies) = line.split_at(delimiter);
            let rewritten = rewrite_depinfo_content_with_sentinel(
                targets,
                target_dir,
                DEPINFO_ROOT_SENTINEL,
                mode,
            );
            format!("{rewritten}{dependencies}")
        })
        .collect()
}

/// Rewrite one exact configured-rule spelling. Configured roots already carry
/// their explicitly constructed lexical/canonical/Windows aliases; adding a
/// fresh canonical alias here could let one configured symlink steal another
/// configured slot's sentinel.
fn rewrite_depinfo_content_with_exact_sentinel(
    content: &str,
    project_dir: &Path,
    sentinel: &str,
    mode: DepInfoMode,
) -> String {
    rewrite_depinfo_content_with_exact_sentinel_for_platform(
        content,
        project_dir,
        sentinel,
        mode,
        cfg!(windows),
    )
}

fn rewrite_depinfo_content_with_exact_sentinel_for_platform(
    content: &str,
    project_dir: &Path,
    sentinel: &str,
    mode: DepInfoMode,
    windows: bool,
) -> String {
    let prefixes = depinfo_prefixes_for_exact_display(&project_dir.to_string_lossy(), windows);
    rewrite_depinfo_content_with_prefixes_and_sentinel(content, &prefixes, sentinel, mode)
}

fn depinfo_prefixes_for_exact_display(raw: &str, windows: bool) -> Vec<String> {
    if windows {
        // Expand uses the first spelling. Preserve the configured spelling:
        // verbatim/native Windows paths require `\`, while slash-form paths
        // (including the platform-neutral paths used by pure tests) keep `/`.
        let (first, second) = if windows_display_prefers_backslash(raw) {
            ('\\', '/')
        } else {
            ('/', '\\')
        };
        vec![
            with_trailing_separator(raw, first),
            with_trailing_separator(raw, second),
        ]
    } else {
        vec![with_trailing_separator(raw, '/')]
    }
}

fn rewrite_depinfo_content_with_sentinel(
    content: &str,
    project_dir: &Path,
    sentinel: &str,
    mode: DepInfoMode,
) -> String {
    rewrite_depinfo_content_with_sentinel_for_platform(
        content,
        project_dir,
        sentinel,
        mode,
        cfg!(windows),
    )
}

fn rewrite_depinfo_content_with_sentinel_for_platform(
    content: &str,
    project_dir: &Path,
    sentinel: &str,
    mode: DepInfoMode,
    windows: bool,
) -> String {
    let mut roots = vec![project_dir.to_path_buf()];
    if matches!(mode, DepInfoMode::Relativize)
        && let Ok(canonical) = project_dir.canonicalize()
        && canonical != project_dir
    {
        roots.push(canonical);
    }
    let mut prefixes = Vec::new();
    for root in roots {
        prefixes.extend(depinfo_prefixes_for_display(
            &root.to_string_lossy(),
            matches!(mode, DepInfoMode::Relativize),
            windows,
        ));
    }
    let mut seen = std::collections::HashSet::new();
    prefixes.retain(|prefix| seen.insert(prefix.clone()));
    rewrite_depinfo_content_with_prefixes_and_sentinel(content, &prefixes, sentinel, mode)
}

fn depinfo_prefixes_for_display(
    raw: &str,
    retain_verbatim_alias: bool,
    windows: bool,
) -> Vec<String> {
    let stripped = raw
        .strip_prefix(r"\\?\UNC\")
        .map(|path| format!(r"\\{path}"))
        .or_else(|| raw.strip_prefix(r"\\?\").map(str::to_string))
        .unwrap_or_else(|| raw.to_string());
    let mut displays = vec![stripped.clone()];
    if retain_verbatim_alias && raw != stripped {
        displays.push(raw.to_string());
    }
    let mut prefixes = Vec::new();
    for display in displays {
        if windows {
            let prefers_backslash = windows_display_prefers_backslash(&display);
            let verbatim = display.starts_with(r"\\?\");
            let preferred = if verbatim {
                display.clone()
            } else if prefers_backslash {
                display.replace('/', "\\")
            } else {
                display.replace('\\', "/")
            };
            let alternate = if prefers_backslash {
                display.replace('\\', "/")
            } else {
                display.replace('/', "\\")
            };
            let (first, second) = if prefers_backslash {
                ('\\', '/')
            } else {
                ('/', '\\')
            };
            prefixes.push(with_trailing_separator(&preferred, first));
            prefixes.push(with_trailing_separator(&display, first));
            prefixes.push(with_trailing_separator(&display, second));
            if retain_verbatim_alias && !verbatim {
                prefixes.push(with_trailing_separator(&alternate, second));
            }
        } else {
            prefixes.push(with_trailing_separator(&display, '/'));
        }
    }
    prefixes
}

fn windows_display_prefers_backslash(display: &str) -> bool {
    let bytes = display.as_bytes();
    display.starts_with('\\')
        || (bytes.first().is_some_and(u8::is_ascii_alphabetic)
            && bytes.get(1) == Some(&b':')
            && bytes.get(2) == Some(&b'\\'))
}

fn with_trailing_separator(display: &str, separator: char) -> String {
    let mut value = display.to_string();
    if value.ends_with('/') || value.ends_with('\\') {
        value.pop();
    }
    value.push(separator);
    value
}

/// Prefix-parameterized core of the Windows [`rewrite_depinfo_content`]
/// branch, split out so the dual-spelling behavior is unit-testable on every
/// host.
///
/// `# env-dep:NAME=value` values are transformed through rustc's env-dep
/// escape grammar: a backslash-bearing prefix appears there as `\\`, so
/// Relativize matches the escaped spelling (never splitting an escape pair
/// in two) and Expand inserts the escaped spelling (never creating an escape
/// cargo rejects, kunobi-ninja/kache#730). The record's key is an environment
/// variable NAME, never a path — it is left untouched, as is a key-only
/// record without `=`.
#[cfg(test)]
fn rewrite_depinfo_content_with_prefixes(
    content: &str,
    prefixes: &[String],
    mode: DepInfoMode,
) -> String {
    rewrite_depinfo_content_with_prefixes_and_sentinel(
        content,
        prefixes,
        DEPINFO_ROOT_SENTINEL,
        mode,
    )
}

fn rewrite_depinfo_content_with_prefixes_and_sentinel(
    content: &str,
    prefixes: &[String],
    sentinel: &str,
    mode: DepInfoMode,
) -> String {
    content
        .split_inclusive('\n')
        .map(|line| {
            if let Some(record) = line.strip_prefix(DEPINFO_ENV_DEP_PREFIX) {
                let Some((key, value)) = record.split_once('=') else {
                    return line.to_string();
                };
                let value = match mode {
                    DepInfoMode::Relativize => prefixes.iter().fold(value.to_string(), |acc, p| {
                        replace_depinfo_text(
                            &acc,
                            &escape_depinfo_env_value(p),
                            sentinel,
                            cfg!(windows),
                            DepInfoTextContext::EnvValue,
                        )
                    }),
                    DepInfoMode::Expand => replace_depinfo_text(
                        value,
                        sentinel,
                        &escape_depinfo_env_value(&prefixes[0]),
                        false,
                        DepInfoTextContext::EnvValue,
                    ),
                };
                format!("{DEPINFO_ENV_DEP_PREFIX}{key}={value}")
            } else {
                match mode {
                    DepInfoMode::Relativize => prefixes.iter().fold(line.to_string(), |acc, p| {
                        let escaped = escape_depinfo_make_path(p);
                        let escaped = replace_depinfo_text(
                            &acc,
                            &escaped,
                            sentinel,
                            cfg!(windows),
                            DepInfoTextContext::MakeLine,
                        );
                        replace_depinfo_text(
                            &escaped,
                            p,
                            sentinel,
                            cfg!(windows),
                            DepInfoTextContext::MakeLine,
                        )
                    }),
                    DepInfoMode::Expand => replace_depinfo_text(
                        line,
                        sentinel,
                        &escape_depinfo_make_path(&prefixes[0]),
                        false,
                        DepInfoTextContext::MakeLine,
                    ),
                }
            }
        })
        .collect()
}

#[derive(Clone, Copy)]
enum DepInfoTextContext {
    MakeLine,
    EnvValue,
}

fn replace_depinfo_text(
    input: &str,
    needle: &str,
    replacement: &str,
    ascii_case_insensitive: bool,
    context: DepInfoTextContext,
) -> String {
    if needle.is_empty() {
        return input.to_string();
    }
    let input_bytes = input.as_bytes();
    let needle_bytes = needle.as_bytes();
    let mut output = String::with_capacity(input.len());
    let mut copied = 0;
    let mut search = 0;
    while search + needle_bytes.len() <= input_bytes.len() {
        // Every path out of this iteration must leave `search` past where it
        // started. Both do: a rejected match steps to the next char boundary,
        // and an accepted one resumes after the replacement. Checked rather
        // than assumed because the failure is not a wrong answer — the scan
        // re-matches the same offset forever while `output` keeps growing, so
        // an index slip here exhausts memory instead of returning. Debug-only,
        // so the release scan is unchanged.
        let scan_start = search;
        let offset = if ascii_case_insensitive {
            input_bytes[search..]
                .windows(needle_bytes.len())
                .position(|window| window.eq_ignore_ascii_case(needle_bytes))
        } else {
            input[search..].find(needle)
        };
        let Some(offset) = offset else {
            break;
        };
        let found = search + offset;
        let end = found + needle_bytes.len();
        // UTF-8 is self-synchronizing: a valid UTF-8 `needle` cannot match
        // across code-point boundaries. `str::find` guarantees this directly;
        // the ASCII-insensitive byte scan changes only ASCII case.
        if !depinfo_path_prefix_has_left_boundary(input, found, context) {
            search = depinfo_next_char_boundary(input, found);
            debug_assert!(scan_start < search, "rejected match must advance");
            continue;
        }
        output.push_str(&input[copied..found]);
        output.push_str(replacement);
        copied = end;
        search = copied;
        debug_assert!(scan_start < search, "replaced match must advance");
    }
    output.push_str(&input[copied..]);
    output
}

/// Byte index of the char boundary after the one starting at `start`, or the
/// end of `input` when `start` holds the last char.
///
/// This is `replace_depinfo_text`'s progress guarantee: the loop calls it to
/// step past a match it has rejected, so the result must be strictly greater
/// than `start` or the search cannot terminate. Named rather than inlined so
/// the guarantee can be asserted directly, by
/// `depinfo_next_char_boundary_always_advances_past_the_current_char`, instead
/// of only showing up as the caller looping.
///
/// The name is prefixed because `config_tui` has an unrelated
/// `next_char_boundary`, and mutation excludes match on function name.
fn depinfo_next_char_boundary(input: &str, start: usize) -> usize {
    input[start..]
        .char_indices()
        .nth(1)
        .map_or(input.len(), |(next, _)| start + next)
}

fn depinfo_path_prefix_has_left_boundary(
    input: &str,
    start: usize,
    context: DepInfoTextContext,
) -> bool {
    if start == 0 {
        return true;
    }
    if matches!(context, DepInfoTextContext::EnvValue) {
        return false;
    }
    let before = &input[..start];
    let Some((delimiter_start, delimiter)) = before.char_indices().next_back() else {
        return false;
    };
    if !delimiter.is_whitespace() && delimiter != ':' {
        return false;
    }
    before[..delimiter_start]
        .chars()
        .rev()
        .take_while(|ch| *ch == '\\')
        .count()
        .is_multiple_of(2)
}

/// Rewrite a `.d` (dep-info) file in place.
///
/// Used on the **store** side, where the file is the build's own
/// freshly-written, writable dep-info. The restore side does NOT use this
/// — it computes the rewritten content in memory via
/// [`rewrite_depinfo_content`] (or its Rust two-root variant), then materializes
/// it with [`write_restored`]. This honors "compute the final bytes, then
/// materialize" rather than patching a restored file in place.
#[cfg(test)]
pub fn rewrite_depinfo(depinfo_path: &Path, project_dir: &Path, mode: DepInfoMode) -> Result<()> {
    let content = fs::read_to_string(depinfo_path)
        .with_context(|| format!("reading dep-info file {}", depinfo_path.display()))?;

    let rewritten = rewrite_depinfo_content(&content, project_dir, mode);
    write_rewritten_depinfo(depinfo_path, rewritten)
}

#[cfg(test)]
fn write_rewritten_depinfo(depinfo_path: &Path, rewritten: String) -> Result<()> {
    // Defense-in-depth: if the file is hardlinked (nlink > 1), unlink
    // first so the in-place write can't mutate a shared inode. On the
    // store side `Store::put` never hardlinks `.d` blobs (DepInfo is
    // excluded from insert-side hardlinking precisely because of this
    // post-put in-place rewrite), so this rarely fires — but it keeps
    // `rewrite_depinfo` safe for any caller. Windows exposes no portable
    // nlink count; remove unconditionally there.
    #[cfg(unix)]
    if let Ok(meta) = fs::metadata(depinfo_path) {
        use std::os::unix::fs::MetadataExt;
        if meta.nlink() > 1 {
            let _ = fs::remove_file(depinfo_path);
        }
    }
    #[cfg(not(unix))]
    if depinfo_path.exists() {
        let _ = fs::remove_file(depinfo_path);
    }

    fs::write(depinfo_path, rewritten)?;
    Ok(())
}

#[derive(Debug, Clone, Copy)]
pub enum DepInfoMode {
    /// Replace absolute project paths with a kache sentinel for cross-project cache sharing.
    Relativize,
    /// Expand the kache sentinel back to absolute project paths after restoring.
    Expand,
}

#[cfg(test)]
mod tests {
    use std::cell::Cell;

    use super::*;

    #[test]
    fn a_mapped_volume_mutes_only_cross_volume_advice() {
        for reason in [
            HardlinkIoReason::CrossDevice,
            HardlinkIoReason::Permission,
            HardlinkIoReason::Other,
        ] {
            assert!(!reason.mapped_volume_advice(false));
            assert_eq!(
                reason.mapped_volume_advice(true),
                reason == HardlinkIoReason::CrossDevice
            );
        }
        for cause in [
            CopyRestoreCause::CrossVolume,
            CopyRestoreCause::NoCow,
            CopyRestoreCause::UnknownCow,
            CopyRestoreCause::SubClusterOnCowVolume,
            CopyRestoreCause::UnexpectedOnCowVolume,
        ] {
            assert!(!cause.mute_advice(false, true, true));
            assert_eq!(
                cause.mute_advice(true, true, true),
                cause == CopyRestoreCause::CrossVolume
            );
        }
    }

    #[test]
    fn copy_advice_switches_preserve_faults_for_every_mapping() {
        for mapped in [false, true] {
            for (strategy, config) in [(false, true), (true, false), (false, false)] {
                for cause in [
                    CopyRestoreCause::CrossVolume,
                    CopyRestoreCause::NoCow,
                    CopyRestoreCause::UnknownCow,
                ] {
                    assert!(cause.mute_advice(mapped, strategy, config));
                }
                for cause in [
                    CopyRestoreCause::SubClusterOnCowVolume,
                    CopyRestoreCause::UnexpectedOnCowVolume,
                ] {
                    assert!(!cause.mute_advice(mapped, strategy, config));
                }
            }
        }
    }

    #[test]
    fn mapped_target_changes_the_emitted_hardlink_advice() {
        const CASE: &str = "KACHE_TEST_MAPPED_HARDLINK_ADVICE";
        if let Ok(case) = std::env::var(CASE) {
            set_mapped_target(case != "unmapped-cross");
            let (reason, kind) = if case == "mapped-permission" {
                (
                    HardlinkIoReason::Permission,
                    std::io::ErrorKind::PermissionDenied,
                )
            } else {
                (
                    HardlinkIoReason::CrossDevice,
                    std::io::ErrorKind::CrossesDevices,
                )
            };
            warn_hardlink_fallback_once(
                Path::new("store"),
                Path::new("target"),
                reason,
                &std::io::Error::from(kind),
            );
            return;
        }
        // Separate processes isolate both the mapping flag and once-only warning.
        for (case, expected) in [
            ("mapped-cross", None),
            ("unmapped-cross", Some("EXDEV")),
            ("mapped-permission", Some("EPERM")),
        ] {
            let output = std::process::Command::new(std::env::current_exe().unwrap())
                .args([
                    "--exact",
                    "link::tests::mapped_target_changes_the_emitted_hardlink_advice",
                    "--nocapture",
                ])
                .env(CASE, case)
                .output()
                .unwrap();
            assert!(output.status.success(), "{case}: {output:?}");
            let stderr = String::from_utf8(output.stderr).unwrap();
            match expected {
                Some(message) => assert!(stderr.contains(message), "{case}: {stderr}"),
                None => assert!(!stderr.contains("EXDEV"), "{case}: {stderr}"),
            }
        }
    }

    /// The regression from #508: on a ReFS Dev Drive that DOES block-clone, a
    /// sub-cluster file (every `.d` under ~4 KB) falls back to copy — and kache
    /// used to blame the filesystem, once per rustc process, hundreds of times
    /// per build. A healthy volume must produce no advisory at all.
    #[test]
    fn sub_cluster_copy_on_a_cow_volume_is_not_a_no_cow_problem() {
        assert_eq!(
            classify_copy_restore(Some("G:\\"), Some("G:\\"), Some(true), Some(true)),
            CopyRestoreCause::SubClusterOnCowVolume,
        );
    }

    /// But silence must not extend to a file that COULD have been cloned and
    /// wasn't — that would hide a filter driver or a kache bug quietly demoting
    /// every large artifact to a copy.
    #[test]
    fn a_clonable_file_that_fails_to_clone_is_still_surfaced() {
        assert_eq!(
            classify_copy_restore(Some("G:\\"), Some("G:\\"), Some(true), Some(false)),
            CopyRestoreCause::UnexpectedOnCowVolume,
        );
        assert_eq!(
            classify_copy_restore(Some("G:\\"), Some("G:\\"), Some(true), None),
            CopyRestoreCause::UnexpectedOnCowVolume,
            "an unknown size must not be assumed benign",
        );
    }

    #[test]
    fn same_volume_without_block_cloning_is_a_real_no_cow_warning() {
        assert_eq!(
            classify_copy_restore(Some("C:\\"), Some("C:\\"), Some(false), Some(true)),
            CopyRestoreCause::NoCow,
        );
    }

    /// Cross-volume wins over capability: two Dev Drives still can't clone
    /// across the volume boundary, so the co-locate advice is the useful one.
    #[test]
    fn different_volumes_report_cross_volume_even_when_both_support_cow() {
        assert_eq!(
            classify_copy_restore(Some("C:\\"), Some("G:\\"), Some(true), Some(false)),
            CopyRestoreCause::CrossVolume,
        );
        assert_eq!(
            classify_copy_restore(Some("c:\\"), Some("C:\\"), Some(false), Some(true)),
            CopyRestoreCause::NoCow,
            "volume roots compare case-insensitively",
        );
    }

    /// A failed probe must still warn (a copy-restore IS happening) — but as
    /// "unknown", not as "no copy-on-write". Asserting NTFS from a failed probe
    /// would be the same unfounded claim that produced #508.
    #[test]
    fn a_failed_capability_probe_warns_without_claiming_no_cow() {
        assert_eq!(
            classify_copy_restore(Some("C:\\"), Some("C:\\"), None, Some(true)),
            CopyRestoreCause::UnknownCow,
        );
        assert_eq!(
            classify_copy_restore(None, None, None, None),
            CopyRestoreCause::UnknownCow,
        );
    }

    /// A benign storage-layout advisory must not be able to mute a genuine clone
    /// FAULT: they dedup in separate buckets, so one can't swallow the other.
    #[test]
    fn a_layout_advisory_cannot_mute_a_clone_fault() {
        let fault = CopyRestoreCause::UnexpectedOnCowVolume.warn_bucket();
        for benign in [
            CopyRestoreCause::NoCow,
            CopyRestoreCause::CrossVolume,
            CopyRestoreCause::UnknownCow,
        ] {
            assert_ne!(
                benign.warn_bucket(),
                fault,
                "{benign:?} must not share a dedup marker with a clone fault",
            );
        }

        let base = Path::new("/tmp/kache-cow-warn-abc123");
        assert_ne!(
            bucket_marker(base, CopyRestoreCause::NoCow.warn_bucket()),
            bucket_marker(base, fault),
        );
    }

    /// #551: `[cache] storage_layout_advice = false` mutes exactly the three
    /// layout advisories. The sub-cluster case is already silent, and the fault
    /// case must stay un-muteable — advice and faults are different severities.
    #[test]
    fn storage_layout_advice_mutes_advice_never_faults() {
        for advisory in [
            CopyRestoreCause::NoCow,
            CopyRestoreCause::CrossVolume,
            CopyRestoreCause::UnknownCow,
        ] {
            assert!(
                advisory.is_layout_advisory(),
                "{advisory:?} is layout advice and must be muteable",
            );
        }
        for non_advisory in [
            CopyRestoreCause::SubClusterOnCowVolume,
            CopyRestoreCause::UnexpectedOnCowVolume,
        ] {
            assert!(
                !non_advisory.is_layout_advisory(),
                "{non_advisory:?} must not be muteable by the advice knob",
            );
        }
    }

    /// #1067: wrapper mode sends layout advice to the log, never faults.
    #[test]
    fn layout_advice_sink_follows_the_process_setting_but_faults_keep_stderr() {
        use crate::markers::WarnSink;
        let _lock = crate::test_support::process_state_test_lock();
        for (to_log, advice_sink) in [(false, WarnSink::Stderr), (true, WarnSink::Log)] {
            set_layout_advice_to_log(to_log);
            assert_eq!(layout_advice_sink(), advice_sink);
            for advisory in [
                CopyRestoreCause::CrossVolume,
                CopyRestoreCause::NoCow,
                CopyRestoreCause::UnknownCow,
            ] {
                assert_eq!(advisory.warn_sink(), advice_sink, "{advisory:?}");
            }
            for fault in [
                CopyRestoreCause::SubClusterOnCowVolume,
                CopyRestoreCause::UnexpectedOnCowVolume,
            ] {
                assert_eq!(fault.warn_sink(), WarnSink::Stderr, "{fault:?}");
            }
        }
        set_layout_advice_to_log(false);
    }

    #[test]
    fn test_hardlink_strategy_restores_content() {
        let _guard = SHARED_TEST_LOCK.lock().unwrap();
        let dir = tempfile::tempdir().unwrap();
        let src = dir.path().join("source.rlib");
        fs::write(&src, b"rlib content").unwrap();

        let dst = dir.path().join("subdir/output.rlib");
        link_to_target(&src, &dst, LinkStrategy::Hardlink).unwrap();

        assert!(dst.exists());
        assert_eq!(fs::read(&dst).unwrap(), b"rlib content");

        // Hardlink strategy promises zero-copy when possible: reflink (CoW,
        // independent inode) on APFS/btrfs/XFS-with-reflink, or hardlink
        // (shared inode) as fallback. We don't assert which mechanism was
        // used — either satisfies the contract.
    }

    #[test]
    fn owner_write_permission_preserves_other_bits() {
        let dir = tempfile::tempdir().unwrap();
        let target = dir.path().join("restored");
        fs::write(&target, b"artifact").unwrap();
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            for (before, after) in [
                (0o400, 0o600),
                (0o440, 0o640),
                (0o444, 0o644),
                (0o555, 0o755),
                (0o600, 0o600),
            ] {
                fs::set_permissions(&target, fs::Permissions::from_mode(before)).unwrap();
                set_owner_write_permission(&target).unwrap();
                assert_eq!(
                    fs::metadata(&target).unwrap().permissions().mode() & 0o777,
                    after
                );
            }
        }
        #[cfg(not(unix))]
        {
            let mut permissions = fs::metadata(&target).unwrap().permissions();
            permissions.set_readonly(true);
            fs::set_permissions(&target, permissions).unwrap();
            set_owner_write_permission(&target).unwrap();
            assert!(!fs::metadata(&target).unwrap().permissions().readonly());
        }
        fs::write(&target, b"rebuilt artifact").unwrap();
        assert!(set_owner_write_permission(&dir.path().join("missing")).is_err());
    }

    #[cfg(unix)]
    #[test]
    fn reflink_restore_adds_owner_write_without_changing_blob_or_other_permissions() {
        use std::os::unix::fs::{MetadataExt, PermissionsExt};

        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.rlib");
        let probe = dir.path().join("probe");
        let target = dir.path().join("output.rlib");
        fs::write(&blob, b"cached artifact").unwrap();
        for mode in [0o400, 0o440, 0o444, 0o555] {
            fs::set_permissions(&blob, fs::Permissions::from_mode(mode)).unwrap();
            if let Err(error) = try_reflink(&blob, &probe) {
                eprintln!("reflink unavailable on test filesystem: {error}");
                return;
            }
            let clone_mode = fs::metadata(&probe).unwrap().permissions().mode() & 0o777;
            fs::remove_file(&probe).unwrap();

            link_to_target(&blob, &target, LinkStrategy::Hardlink).unwrap();
            let restored = fs::metadata(&target).unwrap();
            assert_ne!(restored.ino(), fs::metadata(&blob).unwrap().ino());
            assert_eq!(restored.permissions().mode() & 0o777, clone_mode | 0o200);
            fs::write(&target, b"rebuilt artifact").unwrap();
            assert_eq!(fs::read(&blob).unwrap(), b"cached artifact");
            assert_eq!(
                fs::metadata(&blob).unwrap().permissions().mode() & 0o777,
                mode
            );
        }
    }

    #[cfg(unix)]
    #[test]
    fn hardlink_restore_keeps_shared_inode_readonly_and_copy_writable() {
        let _guard = SHARED_TEST_LOCK.lock().unwrap();
        use std::os::unix::fs::{MetadataExt, PermissionsExt};

        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.rlib");
        let linked = dir.path().join("linked.rlib");
        let copied = dir.path().join("copied.rlib");
        fs::write(&blob, b"cached artifact").unwrap();
        fs::set_permissions(&blob, fs::Permissions::from_mode(0o444)).unwrap();
        let bytes = fs::metadata(&blob).unwrap().len();

        hardlink_or_copy(&blob, &linked, bytes).unwrap();
        assert_eq!(
            fs::metadata(&linked).unwrap().ino(),
            fs::metadata(&blob).unwrap().ino()
        );
        assert!(fs::metadata(&linked).unwrap().permissions().readonly());
        hardlink_or_copy(&blob, &copied, bytes).unwrap();
        assert_ne!(
            fs::metadata(&copied).unwrap().ino(),
            fs::metadata(&blob).unwrap().ino()
        );
        fs::write(&copied, b"rebuilt artifact").unwrap();
        assert_eq!(fs::read(&blob).unwrap(), b"cached artifact");
        assert_eq!(fs::read(&linked).unwrap(), b"cached artifact");
        assert_eq!(
            fs::metadata(&blob).unwrap().permissions().mode() & 0o777,
            0o444
        );
    }

    /// kunobi-ninja/kache#794: restoring the same blob into another target
    /// tree must not re-date an artifact that a linker already has open.
    /// Exercise the fallback directly so APFS reflinks cannot make the test
    /// pass without covering the hardlink path used by Linux/ext4.
    #[cfg(unix)]
    #[test]
    fn hardlink_restore_does_not_retimestamp_existing_consumer() {
        let _guard = SHARED_TEST_LOCK.lock().unwrap();
        use std::fs::File;
        use std::os::unix::fs::MetadataExt;

        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.rlib");
        let tree_a = dir.path().join("tree-a.rlib");
        let tree_b = dir.path().join("tree-b.rlib");
        fs::write(&blob, b"cached rlib").unwrap();
        let bytes = fs::metadata(&blob).unwrap().len();

        hardlink_or_copy(&blob, &tree_a, bytes).unwrap();
        let observed = filetime::FileTime::from_unix_time(1_000_000_000, 0);
        filetime::set_file_mtime(&tree_a, observed).unwrap();

        let reader = File::open(&tree_a).unwrap();
        let before = filetime::FileTime::from_last_modification_time(&reader.metadata().unwrap());
        assert_eq!(
            reader.metadata().unwrap().nlink(),
            2,
            "fixture must engage the hardlink fallback"
        );

        hardlink_or_copy(&blob, &tree_b, bytes).unwrap();
        touch_mtime_write_clock(&tree_b).unwrap();

        let after = filetime::FileTime::from_last_modification_time(&reader.metadata().unwrap());
        assert_eq!(after, before, "restoring tree B re-dated active reader A");
        assert_ne!(
            reader.metadata().unwrap().ino(),
            fs::metadata(&tree_b).unwrap().ino(),
            "simultaneous consumers must not share an inode"
        );
    }

    /// The post-link half of the #794 guard closes the precheck race: even
    /// when two restorers both observe a fresh blob, at most one may retain a
    /// shared inode before either caller reaches its mtime stamp.
    #[cfg(unix)]
    #[test]
    fn simultaneous_hardlink_restores_retain_at_most_one_consumer() {
        let _guard = SHARED_TEST_LOCK.lock().unwrap();
        use std::os::unix::fs::MetadataExt;
        use std::sync::{Arc, Barrier};

        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.rlib");
        let tree_a = dir.path().join("tree-a.rlib");
        let tree_b = dir.path().join("tree-b.rlib");
        fs::write(&blob, b"cached rlib").unwrap();
        let bytes = fs::metadata(&blob).unwrap().len();
        let barrier = Arc::new(Barrier::new(2));

        std::thread::scope(|scope| {
            let barrier_a = Arc::clone(&barrier);
            let blob_a = &blob;
            let target_a = &tree_a;
            let first = scope.spawn(move || {
                hardlink_or_copy_with_prelink_hook(blob_a, target_a, bytes, false, || {
                    barrier_a.wait();
                })
            });
            let barrier_b = Arc::clone(&barrier);
            let blob_b = &blob;
            let target_b = &tree_b;
            let second = scope.spawn(move || {
                hardlink_or_copy_with_prelink_hook(blob_b, target_b, bytes, false, || {
                    barrier_b.wait();
                })
            });
            first.join().unwrap().unwrap();
            second.join().unwrap().unwrap();
        });

        let blob_meta = fs::metadata(&blob).unwrap();
        let shared_consumers = [&tree_a, &tree_b]
            .into_iter()
            .filter(|target| fs::metadata(target).unwrap().ino() == blob_meta.ino())
            .count();
        assert!(
            shared_consumers <= 1,
            "concurrent restorers retained {shared_consumers} blob consumers"
        );
        assert!(
            blob_meta.nlink() <= 2,
            "blob plus at most one target may remain hardlinked"
        );
        assert_eq!(fs::read(&tree_a).unwrap(), b"cached rlib");
        assert_eq!(fs::read(&tree_b).unwrap(), b"cached rlib");
    }

    /// The blob-vanished recovery arm must keep the restore when the target
    /// ended up as the inode's sole name; failing there would turn a benign
    /// eviction race into a lost cache hit (#794).
    #[cfg(unix)]
    #[test]
    fn orphaned_hardlink_with_sole_target_name_is_retained() {
        let dir = tempfile::tempdir().unwrap();
        let target = dir.path().join("tree-a.rlib");
        fs::write(&target, b"cached rlib").unwrap();
        let vanished = fs::metadata(dir.path().join("no-such-blob")).unwrap_err();

        verify_orphaned_hardlink(&target, fs::metadata(&target), vanished).unwrap();
        assert_eq!(fs::read(&target).unwrap(), b"cached rlib");
    }

    /// If the blob vanished but this target still shares its inode with
    /// another name, exclusivity cannot be proven and the restore fails
    /// loudly rather than letting a shared mtime stamp reach the caller
    /// (#794).
    #[cfg(unix)]
    #[test]
    fn orphaned_hardlink_with_shared_target_errors() {
        use std::os::unix::fs::MetadataExt;

        let dir = tempfile::tempdir().unwrap();
        let target = dir.path().join("tree-a.rlib");
        fs::write(&target, b"cached rlib").unwrap();
        fs::hard_link(&target, dir.path().join("alias.rlib")).unwrap();
        assert_eq!(fs::metadata(&target).unwrap().nlink(), 2);
        let vanished = fs::metadata(dir.path().join("no-such-blob")).unwrap_err();

        let error = verify_orphaned_hardlink(&target, fs::metadata(&target), vanished)
            .expect_err("shared orphaned target must fail");
        assert!(error.to_string().contains("still has 2 links"));
    }

    /// A vanished blob plus an unreadable target leaves nothing to validate;
    /// surface the target-side error instead of retaining blindly.
    #[cfg(unix)]
    #[test]
    fn orphaned_hardlink_with_unreadable_target_errors() {
        let dir = tempfile::tempdir().unwrap();
        let target = dir.path().join("missing.rlib");
        let vanished = fs::metadata(dir.path().join("no-such-blob")).unwrap_err();

        let error = verify_orphaned_hardlink(&target, fs::metadata(&target), vanished)
            .expect_err("unreadable target must fail");
        assert!(error.to_string().contains("verifying hardlink target"));
    }

    #[cfg(windows)]
    #[test]
    fn windows_hardlink_restore_yields_writable_deletable_output() {
        // #429: on Windows a Hardlink-strategy restore must NOT leave the
        // output read-only (a hardlink to the read-only store blob would be),
        // or a consumer that owns its output — mozbuild's configure conftest —
        // cannot delete/rewrite it (WinError 5). The output must be writable
        // and deletable, while the store blob stays read-only (integrity).
        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("deadbeef");
        fs::write(&blob, b"obj bytes").unwrap();
        let mut p = fs::metadata(&blob).unwrap().permissions();
        p.set_readonly(true);
        fs::set_permissions(&blob, p).unwrap();

        let out = dir.path().join("conftest.o");
        link_to_target(&blob, &out, LinkStrategy::Hardlink).unwrap();

        assert_eq!(fs::read(&out).unwrap(), b"obj bytes");
        assert!(
            !fs::metadata(&out).unwrap().permissions().readonly(),
            "restored output must be writable on Windows (#429)"
        );
        // The consumer must be able to delete its own output.
        fs::remove_file(&out).expect("consumer must be able to delete its output (#429)");
        // ...without the store blob losing its read-only integrity guard.
        assert!(
            fs::metadata(&blob).unwrap().permissions().readonly(),
            "store blob must stay read-only after a restore"
        );
    }

    #[test]
    fn test_copy_strategy_isolates_writes_from_source() {
        // The Copy strategy guarantees that mutating the destination cannot
        // corrupt the cache blob. This holds whether reflink (CoW) or a
        // plain copy was used; only a hardlink would break it, and Copy
        // never falls back to hardlink.
        let dir = tempfile::tempdir().unwrap();
        let src = dir.path().join("source.bin");
        fs::write(&src, b"original").unwrap();

        let dst = dir.path().join("dest.bin");
        link_to_target(&src, &dst, LinkStrategy::Copy).unwrap();

        fs::write(&dst, b"modified").unwrap();
        assert_eq!(
            fs::read(&src).unwrap(),
            b"original",
            "Copy strategy must isolate dst writes from src"
        );
    }

    #[cfg(unix)]
    #[test]
    fn writable_copy_is_private_and_compiler_writable() {
        use std::os::unix::fs::{MetadataExt, PermissionsExt};

        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.o");
        fs::write(&blob, b"cached object").unwrap();
        fs::set_permissions(&blob, fs::Permissions::from_mode(0o400)).unwrap();

        let output = dir.path().join("output.o");
        prepare_writable_target_from_file(&blob, &output)
            .unwrap()
            .publish()
            .unwrap();

        assert_ne!(
            fs::metadata(&output).unwrap().permissions().mode() & 0o200,
            0,
            "restore must be owner-writable regardless of blob permissions"
        );
        assert_ne!(
            fs::metadata(&blob).unwrap().ino(),
            fs::metadata(&output).unwrap().ino(),
            "writable output must never share the blob inode"
        );
        fs::write(&output, b"changed").unwrap();
        assert_eq!(fs::read(&blob).unwrap(), b"cached object");
    }

    fn refuse_clone_into(_src: &Path, _dst: &fs::File) -> std::io::Result<()> {
        Err(std::io::Error::new(
            std::io::ErrorKind::Unsupported,
            "no in-place clone",
        ))
    }

    fn refuse_clone_new(_src: &Path, _dst: &Path) -> std::io::Result<()> {
        Err(std::io::Error::new(
            std::io::ErrorKind::Unsupported,
            "no new clone",
        ))
    }

    thread_local! {
        static CLONE_CALLS: Cell<u32> = const { Cell::new(0) };
    }

    fn reset_clone_calls() {
        CLONE_CALLS.with(|calls| calls.set(0));
    }

    fn clone_calls() -> u32 {
        CLONE_CALLS.with(Cell::get)
    }

    fn count_clone_call() -> u32 {
        CLONE_CALLS.with(|calls| {
            let next = calls.get() + 1;
            calls.set(next);
            next
        })
    }

    fn clone_names(dir: &Path) -> Vec<String> {
        fs::read_dir(dir)
            .unwrap()
            .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
            .filter(|name| name.starts_with(".kache-cc-clone-"))
            .collect()
    }

    fn restore_deltas() -> (u64, u64) {
        (
            crate::opcounts::thread_copied_bytes(),
            crate::opcounts::thread_reflinked_bytes(),
        )
    }

    #[test]
    fn in_place_clone_publishes_the_cloned_bytes() {
        fn write_marker(_src: &Path, dst: &fs::File) -> std::io::Result<()> {
            use std::io::{Seek, SeekFrom, Write};
            if dst.metadata()?.len() != 0 {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::AlreadyExists,
                    "stage already written",
                ));
            }
            let mut out = dst.try_clone()?;
            out.seek(SeekFrom::Start(0))?;
            out.write_all(b"CLONED")?;
            Ok(())
        }

        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.o");
        fs::write(&blob, b"cached object").unwrap();
        let bytes = fs::metadata(&blob).unwrap().len();
        let output = dir.path().join("output.o");
        let (copied_before, reflinked_before) = restore_deltas();
        prepare_writable_target_from_file_via(&blob, &output, write_marker, refuse_clone_new)
            .unwrap()
            .publish()
            .unwrap();
        assert_eq!(fs::read(&output).unwrap(), b"CLONED");
        assert_eq!(fs::read(&blob).unwrap(), b"cached object");
        let (copied, reflinked) = restore_deltas();
        assert_eq!(copied, copied_before);
        assert_eq!(reflinked, reflinked_before + bytes);

        let occupied = dir.path().join("occupied.o");
        fs::write(&occupied, b"keep").unwrap();
        assert!(
            prepare_writable_target_from_file_via(&blob, &occupied, write_marker, refuse_clone_new)
                .unwrap()
                .publish()
                .is_err()
        );
        assert_eq!(fs::read(&occupied).unwrap(), b"keep");
        assert_eq!(restore_deltas(), (copied_before, reflinked_before + bytes));
        let leftovers: Vec<_> = fs::read_dir(dir.path())
            .unwrap()
            .map(|entry| entry.unwrap().file_name())
            .filter(|name| name.to_string_lossy().starts_with(".kache-cc-"))
            .collect();
        assert!(leftovers.is_empty(), "{leftovers:?}");
    }

    #[cfg(unix)]
    #[test]
    fn new_file_clone_is_private_writable_and_keeps_the_blob_mode() {
        use std::os::unix::fs::{MetadataExt, PermissionsExt};

        fn write_readonly(_src: &Path, dst: &Path) -> std::io::Result<()> {
            fs::write(dst, b"CLONED-NEW")?;
            fs::set_permissions(dst, fs::Permissions::from_mode(0o400))?;
            Ok(())
        }

        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.o");
        fs::write(&blob, b"cached object").unwrap();
        fs::set_permissions(&blob, fs::Permissions::from_mode(0o400)).unwrap();
        let bytes = fs::metadata(&blob).unwrap().len();
        let output = dir.path().join("output.o");
        let (copied_before, reflinked_before) = restore_deltas();
        prepare_writable_target_from_file_via(&blob, &output, refuse_clone_into, write_readonly)
            .unwrap()
            .publish()
            .unwrap();

        let restored = fs::metadata(&output).unwrap();
        assert_eq!(fs::read(&output).unwrap(), b"CLONED-NEW");
        assert_ne!(restored.ino(), fs::metadata(&blob).unwrap().ino());
        assert_ne!(restored.permissions().mode() & 0o200, 0);
        assert_eq!(
            fs::metadata(&blob).unwrap().permissions().mode() & 0o777,
            0o400
        );
        let (copied, reflinked) = restore_deltas();
        assert_eq!(copied, copied_before);
        assert_eq!(reflinked, reflinked_before + bytes);
        fs::write(&output, b"changed").unwrap();
        assert_eq!(fs::read(&blob).unwrap(), b"cached object");
    }

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

        fn hardlink_blob(src: &Path, dst: &Path) -> std::io::Result<()> {
            fs::hard_link(src, dst)
        }

        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.o");
        fs::write(&blob, b"cached object").unwrap();
        let bytes = fs::metadata(&blob).unwrap().len();
        let output = dir.path().join("output.o");
        let (copied_before, reflinked_before) = restore_deltas();
        prepare_writable_target_from_file_via(&blob, &output, refuse_clone_into, hardlink_blob)
            .unwrap()
            .publish()
            .unwrap();

        assert_eq!(fs::read(&output).unwrap(), b"cached object");
        assert_ne!(
            fs::metadata(&output).unwrap().ino(),
            fs::metadata(&blob).unwrap().ino()
        );
        assert_eq!(fs::metadata(&blob).unwrap().nlink(), 1);
        let (copied, reflinked) = restore_deltas();
        assert_eq!(copied, copied_before + bytes);
        assert_eq!(reflinked, reflinked_before);
        fs::write(&output, b"changed").unwrap();
        assert_eq!(fs::read(&blob).unwrap(), b"cached object");
    }

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

        fn rename_blob(src: &Path, dst: &Path) -> std::io::Result<()> {
            fs::rename(src, dst)
        }

        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.o");
        fs::write(&blob, b"cached object").unwrap();
        let inode = fs::metadata(&blob).unwrap().ino();
        let bytes = fs::metadata(&blob).unwrap().len();
        let output = dir.path().join("output.o");
        let (copied_before, reflinked_before) = restore_deltas();
        prepare_writable_target_from_file_via(&blob, &output, refuse_clone_into, rename_blob)
            .unwrap()
            .publish()
            .unwrap();

        let restored = fs::metadata(&blob).unwrap();
        assert_eq!(restored.ino(), inode);
        assert_eq!(restored.nlink(), 1);
        assert_eq!(fs::read(&blob).unwrap(), b"cached object");
        assert_eq!(fs::read(&output).unwrap(), b"cached object");
        assert_ne!(fs::metadata(&output).unwrap().ino(), inode);
        let (copied, reflinked) = restore_deltas();
        assert_eq!(copied, copied_before + bytes);
        assert_eq!(reflinked, reflinked_before);
        fs::write(&output, b"changed").unwrap();
        assert_eq!(fs::read(&blob).unwrap(), b"cached object");
    }

    #[test]
    fn replacing_a_clone_overwrites_the_existing_output() {
        fn write_clone(_src: &Path, dest: &Path) -> std::io::Result<()> {
            fs::write(dest, b"CLONED-REPLACE")
        }

        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.o");
        fs::write(&blob, b"cached object").unwrap();
        let output = dir.path().join("output.o");
        fs::write(&output, b"previous output").unwrap();
        prepare_writable_target_from_file_via(&blob, &output, refuse_clone_into, write_clone)
            .unwrap()
            .publish_replacing()
            .unwrap();
        assert_eq!(fs::read(&output).unwrap(), b"CLONED-REPLACE");
        assert!(clone_names(dir.path()).is_empty());
    }

    #[test]
    fn an_unpublished_clone_is_removed() {
        fn write_clone(_src: &Path, dest: &Path) -> std::io::Result<()> {
            fs::write(dest, b"CLONED-TEMP")
        }

        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.o");
        fs::write(&blob, b"cached object").unwrap();
        let output = dir.path().join("output.o");
        drop(
            prepare_writable_target_from_file_via(&blob, &output, refuse_clone_into, write_clone)
                .unwrap(),
        );
        assert!(
            clone_names(dir.path()).is_empty(),
            "dropping an unpublished clone must remove its temporary name"
        );
        assert!(!output.exists());
    }

    #[test]
    fn a_busy_clone_name_is_retried() {
        fn exists_then_clone(_src: &Path, dest: &Path) -> std::io::Result<()> {
            if count_clone_call() == 1 {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::AlreadyExists,
                    "name taken",
                ));
            }
            fs::write(dest, b"CLONED-RETRY")
        }

        reset_clone_calls();
        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.o");
        fs::write(&blob, b"cached object").unwrap();
        let output = dir.path().join("output.o");
        prepare_writable_target_from_file_via(&blob, &output, refuse_clone_into, exists_then_clone)
            .unwrap()
            .publish()
            .unwrap();
        assert_eq!(fs::read(&output).unwrap(), b"CLONED-RETRY");
        assert_eq!(clone_calls(), 2);
    }

    #[test]
    fn a_clone_error_is_not_retried() {
        fn deny_clone(_src: &Path, _dest: &Path) -> std::io::Result<()> {
            count_clone_call();
            Err(std::io::Error::new(
                std::io::ErrorKind::PermissionDenied,
                "denied",
            ))
        }

        reset_clone_calls();
        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.o");
        fs::write(&blob, b"cached object").unwrap();
        let bytes = fs::metadata(&blob).unwrap().len();
        let output = dir.path().join("output.o");
        let (copied_before, reflinked_before) = restore_deltas();
        prepare_writable_target_from_file_via(&blob, &output, refuse_clone_into, deny_clone)
            .unwrap()
            .publish()
            .unwrap();
        assert_eq!(fs::read(&output).unwrap(), b"cached object");
        assert_eq!(clone_calls(), 1);
        let (copied, reflinked) = restore_deltas();
        assert_eq!(copied, copied_before + bytes);
        assert_eq!(reflinked, reflinked_before);
    }

    #[test]
    fn a_clone_is_created_in_the_output_directory() {
        fn write_beside(src: &Path, dest: &Path) -> std::io::Result<()> {
            if dest.parent() != src.parent() {
                return Err(std::io::Error::other(
                    "clone was not created beside the output",
                ));
            }
            fs::write(dest, b"CLONED-BESIDE")
        }

        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.o");
        fs::write(&blob, b"cached object").unwrap();
        let output = dir.path().join("output.o");
        prepare_writable_target_from_file_via(&blob, &output, refuse_clone_into, write_beside)
            .unwrap()
            .publish()
            .unwrap();
        assert_eq!(fs::read(&output).unwrap(), b"CLONED-BESIDE");
    }

    #[test]
    fn a_failed_in_place_clone_does_not_keep_partial_bytes() {
        fn write_junk_then_fail(_src: &Path, dst: &fs::File) -> std::io::Result<()> {
            use std::io::Write;
            dst.try_clone()?.write_all(b"JUNKJUNKJUNKJUNKJUNK")?;
            Err(std::io::Error::new(
                std::io::ErrorKind::Unsupported,
                "partial clone",
            ))
        }

        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.o");
        fs::write(&blob, b"cached object").unwrap();
        let output = dir.path().join("output.o");
        prepare_writable_target_from_file_via(
            &blob,
            &output,
            write_junk_then_fail,
            refuse_clone_new,
        )
        .unwrap()
        .publish()
        .unwrap();
        assert_eq!(fs::read(&output).unwrap(), b"cached object");
    }

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

        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.o");
        fs::write(&blob, b"cached object").unwrap();
        assert!(accepted_private_clone(&blob, &blob).is_none());
        assert_eq!(fs::metadata(&blob).unwrap().nlink(), 1);
        assert_eq!(fs::read(&blob).unwrap(), b"cached object");
    }

    #[test]
    fn clone_failure_copies_the_blob() {
        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.o");
        fs::write(&blob, b"cached object").unwrap();
        let bytes = fs::metadata(&blob).unwrap().len();
        let output = dir.path().join("output.o");
        let (copied_before, reflinked_before) = restore_deltas();
        prepare_writable_target_from_file_via(&blob, &output, refuse_clone_into, refuse_clone_new)
            .unwrap()
            .publish()
            .unwrap();
        assert_eq!(fs::read(&output).unwrap(), b"cached object");
        let (copied, reflinked) = restore_deltas();
        assert_eq!(copied, copied_before + bytes);
        assert_eq!(reflinked, reflinked_before);
    }

    #[cfg(windows)]
    #[test]
    fn writable_copy_is_private_and_writable_on_windows() {
        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.obj");
        fs::write(&blob, b"cached object").unwrap();
        let mut perms = fs::metadata(&blob).unwrap().permissions();
        perms.set_readonly(true);
        fs::set_permissions(&blob, perms).unwrap();

        let output = dir.path().join("output.obj");
        prepare_writable_target_from_file(&blob, &output)
            .unwrap()
            .publish()
            .unwrap();

        assert!(!fs::metadata(&output).unwrap().permissions().readonly());
        fs::write(&output, b"changed").unwrap();
        assert_eq!(fs::read(&blob).unwrap(), b"cached object");
        assert!(fs::metadata(&blob).unwrap().permissions().readonly());

        let mut perms = fs::metadata(&blob).unwrap().permissions();
        #[allow(clippy::permissions_set_readonly_false)]
        perms.set_readonly(false);
        fs::set_permissions(&blob, perms).unwrap();
    }

    #[test]
    fn writable_materializers_never_replace_existing_target() {
        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.o");
        let linked = dir.path().join("linked.o");
        let written = dir.path().join("written.d");
        fs::write(&blob, b"cached").unwrap();
        fs::write(&linked, b"race winner").unwrap();
        fs::write(&written, b"race winner").unwrap();

        #[cfg(windows)]
        for target in [&linked, &written] {
            let mut permissions = fs::metadata(target).unwrap().permissions();
            permissions.set_readonly(true);
            fs::set_permissions(target, permissions).unwrap();
        }

        assert!(
            prepare_writable_target_from_file(&blob, &linked)
                .unwrap()
                .publish()
                .is_err()
        );
        assert!(
            prepare_writable_target_from_bytes(&written, b"cached depinfo")
                .unwrap()
                .publish()
                .is_err()
        );
        assert_eq!(fs::read(&linked).unwrap(), b"race winner");
        assert_eq!(fs::read(&written).unwrap(), b"race winner");

        #[cfg(windows)]
        for target in [&linked, &written] {
            let mut permissions = fs::metadata(target).unwrap().permissions();
            assert!(permissions.readonly(), "refusal must not chmod the target");
            #[allow(clippy::permissions_set_readonly_false)]
            permissions.set_readonly(false);
            fs::set_permissions(target, permissions).unwrap();
        }
    }

    #[test]
    fn test_copy_strategy() {
        let dir = tempfile::tempdir().unwrap();
        let src = dir.path().join("source.bin");
        fs::write(&src, b"binary content").unwrap();

        let dst = dir.path().join("output.bin");
        link_to_target(&src, &dst, LinkStrategy::Copy).unwrap();

        assert!(dst.exists());
        assert_eq!(fs::read(&dst).unwrap(), b"binary content");

        // Should NOT be a hardlink
        #[cfg(unix)]
        {
            use std::os::unix::fs::MetadataExt;
            let src_ino = fs::metadata(&src).unwrap().ino();
            let dst_ino = fs::metadata(&dst).unwrap().ino();
            assert_ne!(src_ino, dst_ino);
        }
    }

    #[test]
    fn test_overwrite_existing() {
        let dir = tempfile::tempdir().unwrap();
        let src = dir.path().join("source.rlib");
        fs::write(&src, b"new content").unwrap();

        let dst = dir.path().join("output.rlib");
        fs::write(&dst, b"old content").unwrap();

        link_to_target(&src, &dst, LinkStrategy::Hardlink).unwrap();
        assert_eq!(fs::read(&dst).unwrap(), b"new content");
    }

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

        let dir = tempfile::tempdir().unwrap();
        let old_blob = dir.path().join("old-blob.rlib");
        let new_blob = dir.path().join("new-blob.rlib");
        let dst = dir.path().join("output.rlib");

        fs::write(&old_blob, b"old content").unwrap();
        fs::set_permissions(&old_blob, fs::Permissions::from_mode(0o444)).unwrap();
        fs::hard_link(&old_blob, &dst).unwrap();

        fs::write(&new_blob, b"new content").unwrap();
        link_to_target(&new_blob, &dst, LinkStrategy::Hardlink).unwrap();

        assert_eq!(fs::read(&dst).unwrap(), b"new content");
        assert!(
            fs::metadata(&old_blob).unwrap().permissions().readonly(),
            "replacing a restored hardlink must not make the original blob writable"
        );
    }

    /// The write-clock touch invariant (kunobi-ninja/kache#677, #135): a
    /// touched file's mtime must order between files written before and
    /// after it by the SAME clock the kernel stamps writes with. A
    /// precise-clock touch (`FileTime::now()`) can postdate a file written
    /// immediately after — that inversion is the entire #677 bug. The
    /// before-bound also proves the touch moved the mtime forward from an
    /// arbitrarily old value.
    #[test]
    fn test_touch_write_clock_orders_with_file_writes() {
        let dir = tempfile::tempdir().unwrap();
        let before = dir.path().join("before");
        let touched = dir.path().join("touched.rlib");
        let after = dir.path().join("after");

        fs::write(&before, b"b").unwrap();
        fs::write(&touched, b"content").unwrap();
        let past = filetime::FileTime::from_unix_time(1_000_000_000, 0);
        filetime::set_file_mtime(&touched, past).unwrap();

        touch_mtime_write_clock(&touched).unwrap();
        fs::write(&after, b"a").unwrap();

        let mtime =
            |p: &Path| filetime::FileTime::from_last_modification_time(&fs::metadata(p).unwrap());
        assert!(
            mtime(&before) <= mtime(&touched),
            "touch must move the mtime to now: before={:?} touched={:?}",
            mtime(&before),
            mtime(&touched)
        );
        // The after-bound is the #677 regression assertion and runs
        // everywhere. On unix the stamp goes through futimens(UTIME_NOW);
        // on Windows through a sibling write-clock sample
        // (kunobi-ninja/kache#681) — before that fix, the first Windows CI
        // run of this test measured the precise-clock stamp postdating a
        // file written ~2ms later (an ~1.8ms inversion window on NTFS).
        assert!(
            mtime(&touched) <= mtime(&after),
            "write-clock touch must not postdate a subsequent write: touched={:?} after={:?}",
            mtime(&touched),
            mtime(&after)
        );
    }

    /// The fail-closed half of kunobi-ninja/kache#682: the stamp opens with
    /// O_NOFOLLOW, so a path swapped for a symlink between materialization
    /// and touch is refused instead of re-dating whatever it points at.
    /// The error degrades to a recompile at the restore call sites.
    #[cfg(unix)]
    #[test]
    fn test_touch_refuses_symlink() {
        let dir = tempfile::tempdir().unwrap();
        let real = dir.path().join("real.rlib");
        let link = dir.path().join("swapped.rlib");
        fs::write(&real, b"content").unwrap();
        let past = filetime::FileTime::from_unix_time(1_000_000_000, 0);
        filetime::set_file_mtime(&real, past).unwrap();
        std::os::unix::fs::symlink(&real, &link).unwrap();

        assert!(
            touch_mtime_write_clock(&link).is_err(),
            "touching through a symlink must fail closed"
        );
        let real_mtime =
            filetime::FileTime::from_last_modification_time(&fs::metadata(&real).unwrap());
        assert_eq!(
            real_mtime.unix_seconds(),
            1_000_000_000,
            "the symlink target must not have been re-dated"
        );
    }

    /// A FIFO swapped in for the artifact must be refused promptly: with
    /// O_NONBLOCK the open succeeds without a writer, fstat sees S_IFIFO,
    /// and the touch bails. Without O_NONBLOCK the open would block forever
    /// waiting for a writer — so this test also pins the flag (a mutant
    /// dropping it hangs here instead of passing).
    #[cfg(unix)]
    #[test]
    fn test_touch_refuses_fifo() {
        use std::os::unix::ffi::OsStrExt;
        let dir = tempfile::tempdir().unwrap();
        let fifo = dir.path().join("swapped.rlib");
        let cpath = std::ffi::CString::new(fifo.as_os_str().as_bytes()).unwrap();
        assert_eq!(unsafe { libc::mkfifo(cpath.as_ptr(), 0o644) }, 0);

        assert!(
            touch_mtime_write_clock(&fifo).is_err(),
            "touching a FIFO must fail closed (and promptly)"
        );
    }

    #[test]
    fn test_depinfo_rewrite() {
        let dir = tempfile::tempdir().unwrap();
        let depfile = dir.path().join("test.d");
        fs::write(
            &depfile,
            "/home/user/project/target/debug/deps/libserde.rlib: /home/user/project/src/lib.rs",
        )
        .unwrap();

        rewrite_depinfo(
            &depfile,
            Path::new("/home/user/project"),
            DepInfoMode::Relativize,
        )
        .unwrap();

        let content = fs::read_to_string(&depfile).unwrap();
        assert!(content.contains("target/debug"));
        assert!(content.contains("src/lib.rs"));
        assert!(content.contains(DEPINFO_ROOT_SENTINEL));
        assert!(!content.contains("/home/user/project/"));

        // Now expand back
        rewrite_depinfo(
            &depfile,
            Path::new("/home/user/project"),
            DepInfoMode::Expand,
        )
        .unwrap();

        let content = fs::read_to_string(&depfile).unwrap();
        assert!(content.contains("/home/user/project/"));
    }

    #[test]
    fn rustc_depinfo_rewrite_rebases_target_and_worktree_separately() {
        let producer_target = Path::new("/build/worktree-a/target");
        let producer_cwd = Path::new("/build/worktree-a");
        let input = "/build/worktree-a/target/debug/deps/libdemo.rlib: \
/build/worktree-a/src/lib.rs /build/worktree-a/assets/value.txt\n";

        let stored = rewrite_rustc_depinfo_content(
            input,
            producer_target,
            producer_cwd,
            None,
            DepInfoMode::Relativize,
        );
        assert!(stored.contains("__kache_root__/debug/deps/libdemo.rlib"));
        assert!(stored.contains("__kache_cwd__/src/lib.rs"));
        assert!(stored.contains("__kache_cwd__/assets/value.txt"));
        assert!(!stored.contains("worktree-a"));

        let restored = rewrite_rustc_depinfo_content(
            &stored,
            Path::new("/build/worktree-b/target"),
            Path::new("/build/worktree-b"),
            None,
            DepInfoMode::Expand,
        );
        assert!(restored.contains("/build/worktree-b/target/debug/deps/libdemo.rlib"));
        assert!(restored.contains("/build/worktree-b/src/lib.rs"));
        assert!(restored.contains("/build/worktree-b/assets/value.txt"));
        assert!(!restored.contains("worktree-a"));
    }

    #[test]
    fn rustc_depinfo_rewrite_handles_workspace_and_make_escaped_paths() {
        let producer_workspace = Path::new("/build/work tree-a");
        let producer_target = producer_workspace.join("target");
        let producer_cwd = producer_workspace.join("crates/demo");
        let input = "/build/work\\ tree-a/target/debug/deps/libdemo.rlib: \
/build/work\\ tree-a/crates/demo/src/lib.rs \
/build/work\\ tree-a/shared/schema.txt\n\
# env-dep:SCHEMA_ROOT=/build/work tree-a/shared\n";

        let stored = rewrite_rustc_depinfo_content(
            input,
            &producer_target,
            &producer_cwd,
            Some(producer_workspace),
            DepInfoMode::Relativize,
        );
        assert!(stored.contains("__kache_root__/debug/deps/libdemo.rlib"));
        assert!(stored.contains("__kache_cwd__/src/lib.rs"));
        assert!(stored.contains("__kache_workspace__/shared/schema.txt"));
        assert!(stored.contains("# env-dep:SCHEMA_ROOT=__kache_workspace__/shared"));
        assert!(!stored.contains("work tree-a"));
        assert!(!stored.contains("work\\ tree-a"));

        let consumer_workspace = Path::new("/build/work tree-b");
        let restored = rewrite_rustc_depinfo_content(
            &stored,
            &consumer_workspace.join("target"),
            &consumer_workspace.join("crates/demo"),
            Some(consumer_workspace),
            DepInfoMode::Expand,
        );
        assert!(restored.contains("/build/work\\ tree-b/target/debug/deps/libdemo.rlib"));
        assert!(restored.contains("/build/work\\ tree-b/crates/demo/src/lib.rs"));
        assert!(restored.contains("/build/work\\ tree-b/shared/schema.txt"));
        assert!(restored.contains("# env-dep:SCHEMA_ROOT=/build/work tree-b/shared"));
        assert!(!restored.contains("work tree-a"));
        assert!(!restored.contains("work\\ tree-a"));
    }

    #[test]
    fn rustc_depinfo_rewrite_round_trips_configured_external_roots() {
        let producer = PathBuf::from("/mount-a/generated");
        let consumer = PathBuf::from("/mount-b/generated");
        let sentinel = "__kache_base_dir_0__/".to_string();
        let input = "/work-a/target/demo.d: /mount-a/generated/schema/data.json\n";

        let stored = rewrite_rustc_depinfo_content_with_configured_roots(
            input,
            Path::new("/work-a/target"),
            Path::new("/work-a/crate"),
            Some(Path::new("/work-a")),
            &[(producer, sentinel.clone(), 8)],
            DepInfoMode::Relativize,
        );
        assert!(stored.contains("__kache_base_dir_0__/schema/data.json"));
        assert!(!stored.contains("/mount-a/generated"));

        let restored = rewrite_rustc_depinfo_content_with_configured_roots(
            &stored,
            Path::new("/work-b/target"),
            Path::new("/work-b/crate"),
            Some(Path::new("/work-b")),
            &[(consumer, sentinel, 8)],
            DepInfoMode::Expand,
        );
        assert!(restored.contains("/mount-b/generated/schema/data.json"));
        assert!(!restored.contains("/mount-a/generated"));
    }

    #[test]
    fn configured_root_wins_when_it_equals_the_working_directory() {
        let root = PathBuf::from("/work/crate");
        let stored = rewrite_rustc_depinfo_content_with_configured_roots(
            "/work/target/demo.d: /work/crate/generated.rs\n",
            Path::new("/work/target"),
            &root,
            Some(Path::new("/work")),
            &[(root.clone(), "__kache_base_dir_0__/".to_string(), 8)],
            DepInfoMode::Relativize,
        );
        assert!(stored.contains("__kache_base_dir_0__/generated.rs"));
        assert!(!stored.contains(DEPINFO_CWD_SENTINEL));
    }

    #[test]
    fn configured_ancestor_precedes_more_specific_automatic_roots() {
        let configured = PathBuf::from("/sandbox");
        let stored = rewrite_rustc_depinfo_content_with_configured_roots(
            "/sandbox/worktree-a/target/demo.d: /sandbox/worktree-a/crate/generated.rs\n",
            Path::new("/sandbox/worktree-a/target"),
            Path::new("/sandbox/worktree-a/crate"),
            Some(Path::new("/sandbox/worktree-a")),
            &[(configured, "__kache_base_dir_0__/".to_string(), 8)],
            DepInfoMode::Relativize,
        );
        assert!(stored.starts_with("__kache_root__/demo.d:"), "{stored}");
        assert!(stored.contains("__kache_base_dir_0__/worktree-a/crate/generated.rs"));
        assert!(!stored.contains(DEPINFO_CWD_SENTINEL));
        assert!(!stored.contains(DEPINFO_WORKSPACE_SENTINEL));
    }

    #[test]
    fn target_generated_source_uses_target_ownership_before_workspace() {
        let stored = rewrite_rustc_depinfo_content_with_configured_roots(
            "/work/target/demo.d: /work/target/generated.rs\n",
            Path::new("/work/target"),
            Path::new("/work"),
            Some(Path::new("/work")),
            &[],
            DepInfoMode::Relativize,
        );

        assert_eq!(
            stored,
            "__kache_root__/demo.d: __kache_target_rule__/generated.rs\n"
        );
    }

    #[test]
    fn external_target_generated_source_round_trips_independently_from_output() {
        let producer_target = PathBuf::from("/external-a/target");
        let consumer_target = PathBuf::from("/external-b/target");
        let target_sentinel = "__kache_target_rule__/".to_string();
        let input = "/external-a/target/debug/deps/demo.d: \
/external-a/target/debug/build/demo/out/generated.rs\n";

        let stored = rewrite_rustc_depinfo_content_with_configured_roots(
            input,
            &producer_target,
            Path::new("/work-a/crate"),
            Some(Path::new("/work-a")),
            &[(producer_target.clone(), target_sentinel.clone(), 3)],
            DepInfoMode::Relativize,
        );
        assert_eq!(
            stored,
            "__kache_root__/debug/deps/demo.d: \
__kache_target_rule__/debug/build/demo/out/generated.rs\n"
        );

        let restored = rewrite_rustc_depinfo_content_with_configured_roots(
            &stored,
            &consumer_target,
            Path::new("/work-b/crate"),
            Some(Path::new("/work-b")),
            &[(consumer_target.clone(), target_sentinel, 3)],
            DepInfoMode::Expand,
        );
        assert_eq!(
            restored,
            "/external-b/target/debug/deps/demo.d: \
/external-b/target/debug/build/demo/out/generated.rs\n"
        );
    }

    /// kunobi-ninja/kache#808: several Cargo processes can share one local
    /// store while building the same workspace into `target_1`...`target_4`.
    /// A generated dependency beneath an in-workspace target must follow the
    /// consuming target, not the stable workspace root, or a hit restores the
    /// donor worker's `OUT_DIR/private.rs` and validate-on-hit evicts it.
    #[test]
    fn in_workspace_target_generated_source_rebases_between_parallel_workers() {
        let workspace = PathBuf::from("/work");
        let producer_target = workspace.join("target_1");
        let consumer_target = workspace.join("target_2");
        let workspace_sentinel = DEPINFO_WORKSPACE_SENTINEL.to_string();
        let target_sentinel = "__kache_target_rule__/".to_string();
        let input = "/work/target_1/debug/deps/serde_core.d: \
/work/src/lib.rs \
/work/target_1/debug/build/serde_core-fe75/out/private.rs\n";

        let stored = rewrite_rustc_depinfo_content_with_configured_roots(
            input,
            &producer_target,
            &workspace,
            Some(&workspace),
            &[
                (workspace.clone(), workspace_sentinel.clone(), 4),
                (producer_target.clone(), target_sentinel.clone(), 3),
            ],
            DepInfoMode::Relativize,
        );
        assert_eq!(
            stored,
            "__kache_root__/debug/deps/serde_core.d: \
__kache_workspace__/src/lib.rs \
__kache_target_rule__/debug/build/serde_core-fe75/out/private.rs\n"
        );

        let restored = rewrite_rustc_depinfo_content_with_configured_roots(
            &stored,
            &consumer_target,
            &workspace,
            Some(&workspace),
            &[
                (workspace.clone(), workspace_sentinel, 4),
                (consumer_target.clone(), target_sentinel, 3),
            ],
            DepInfoMode::Expand,
        );
        assert_eq!(
            restored,
            "/work/target_2/debug/deps/serde_core.d: \
/work/src/lib.rs \
/work/target_2/debug/build/serde_core-fe75/out/private.rs\n"
        );
    }

    #[test]
    fn configured_root_does_not_match_an_interior_path_component() {
        let stored = rewrite_rustc_depinfo_content_with_configured_roots(
            "out: /mnt/sandbox/file.rs /mnt/@/sandbox/punctuation.rs \
/mnt/foo\\ /sandbox/escaped-space.rs /sandbox/owned.rs\n\
# env-dep:ASSET=/mnt/sandbox/asset.txt\n",
            Path::new("/target"),
            Path::new("/work"),
            None,
            &[(
                PathBuf::from("/sandbox"),
                "__kache_base_dir_0__/".to_string(),
                8,
            )],
            DepInfoMode::Relativize,
        );

        assert!(stored.contains("/mnt/sandbox/file.rs"));
        assert!(stored.contains("/mnt/@/sandbox/punctuation.rs"));
        assert!(stored.contains("/mnt/foo\\ /sandbox/escaped-space.rs"));
        assert!(stored.contains("# env-dep:ASSET=/mnt/sandbox/asset.txt"));
        assert!(stored.contains("__kache_base_dir_0__/owned.rs"));
    }

    #[test]
    fn portable_home_root_does_not_steal_nested_workspace_paths() {
        let producer_roots = [(
            PathBuf::from("/home/alice"),
            "__kache_home__/".to_string(),
            1,
        )];
        let stored = rewrite_rustc_depinfo_content_with_configured_roots(
            "out: /home/alice/project/src/lib.rs /home/alice/shared/schema.json\n",
            Path::new("/target"),
            Path::new("/home/alice/project"),
            Some(Path::new("/home/alice/project")),
            &producer_roots,
            DepInfoMode::Relativize,
        );
        assert!(stored.contains("__kache_cwd__/src/lib.rs"), "{stored}");
        assert!(
            stored.contains("__kache_home__/shared/schema.json"),
            "{stored}"
        );

        let consumer_roots = [(
            PathBuf::from("/Users/bob"),
            "__kache_home__/".to_string(),
            1,
        )];
        let restored = rewrite_rustc_depinfo_content_with_configured_roots(
            &stored,
            Path::new("/other-target"),
            Path::new("/Users/bob/project"),
            Some(Path::new("/Users/bob/project")),
            &consumer_roots,
            DepInfoMode::Expand,
        );
        assert!(restored.contains("/Users/bob/project/src/lib.rs"));
        assert!(restored.contains("/Users/bob/shared/schema.json"));
        assert!(!restored.contains("/home/alice"));
    }

    #[test]
    fn depinfo_prefix_replacement_can_match_windows_ascii_case_aliases() {
        assert_eq!(
            replace_depinfo_text(
                r"c:\\WORKTREE\\src\\lib.rs",
                r"C:\\worktree\\",
                DEPINFO_CWD_SENTINEL,
                true,
                DepInfoTextContext::MakeLine,
            ),
            r"__kache_cwd__/src\\lib.rs"
        );
        assert_eq!(
            replace_depinfo_text(
                "unrelated",
                "C:\\worktree\\",
                DEPINFO_CWD_SENTINEL,
                true,
                DepInfoTextContext::MakeLine,
            ),
            "unrelated"
        );
        assert_eq!(
            replace_depinfo_text(
                r"c:\WORKTREE\",
                r"C:\worktree\",
                DEPINFO_CWD_SENTINEL,
                true,
                DepInfoTextContext::MakeLine,
            ),
            DEPINFO_CWD_SENTINEL,
            "a case-insensitive match ending exactly at EOF must be replaced"
        );
        assert_eq!(
            replace_depinfo_text(
                r"keep c:\WORKTREE\src\lib.rs and C:\worktree\asset.txt",
                r"C:\worktree\",
                DEPINFO_CWD_SENTINEL,
                true,
                DepInfoTextContext::MakeLine,
            ),
            r"keep __kache_cwd__/src\lib.rs and __kache_cwd__/asset.txt"
        );
    }

    /// `replace_depinfo_text`'s search loop terminates only because this step
    /// strictly advances. Asserted here rather than left to the loop, where a
    /// slip shows up as a hang instead of a failure.
    #[test]
    fn depinfo_next_char_boundary_always_advances_past_the_current_char() {
        // Single-byte, mid-string: the plain case the loop hits most.
        assert_eq!(depinfo_next_char_boundary("abcd", 1), 2);
        // Multi-byte: advancing lands on the next boundary, not the next byte.
        assert_eq!(depinfo_next_char_boundary("aéb", 1), 3);
        // Last char: clamps to the end rather than reporting no progress.
        assert_eq!(depinfo_next_char_boundary("abc", 2), 3);
        assert_eq!(depinfo_next_char_boundary("aé", 1), 3);

        // The property the loop actually depends on, over every boundary of a
        // mixed-width string.
        let input = "a/é/b/日/c";
        for (start, _) in input.char_indices() {
            assert!(
                depinfo_next_char_boundary(input, start) > start,
                "must advance past byte {start} of {input:?}"
            );
        }
    }

    #[test]
    fn windows_depinfo_prefixes_retain_drive_and_unc_verbatim_aliases() {
        let drive = depinfo_prefixes_for_display(r"\\?\C:\Work", true, true);
        assert!(drive.contains(&r"C:\Work\".to_string()));
        assert!(drive.contains(&r"\\?\C:\Work\".to_string()));
        assert!(drive.contains(&r"C:\Work/".to_string()));
        assert!(drive.contains(&r"\\?\C:\Work/".to_string()));

        let unc = depinfo_prefixes_for_display(r"\\?\UNC\server\share\root", true, true);
        assert!(unc.contains(&r"\\server\share\root\".to_string()));
        assert!(unc.contains(&r"\\?\UNC\server\share\root\".to_string()));

        let expand = depinfo_prefixes_for_display(r"\\?\C:\Work", false, true);
        assert!(!expand.iter().any(|prefix| prefix.starts_with(r"\\?\")));
        assert!(expand.iter().all(|prefix| prefix.starts_with(r"C:\Work")));
    }

    /// The opposite-separator alias is contributed by the stripped display
    /// only. A verbatim `\\?\` display is backslash-only to Windows, so
    /// slash-expanding it would invent a spelling no dep-info ever carries.
    #[test]
    fn windows_depinfo_alternate_alias_is_expanded_only_for_non_verbatim_displays() {
        let prefixes = depinfo_prefixes_for_display(r"\\?\C:\Work", true, true);

        assert!(
            prefixes.contains(&"C:/Work/".to_string()),
            "stripped display contributes its slash alternate: {prefixes:?}"
        );
        assert!(
            !prefixes.iter().any(|prefix| prefix.starts_with("//?/")),
            "verbatim display must not be slash-expanded: {prefixes:?}"
        );
    }

    #[test]
    fn windows_depinfo_expansion_preserves_native_and_slash_root_styles() {
        assert!(windows_display_prefers_backslash(r"C:\work"));
        assert!(windows_display_prefers_backslash(r"\\?\C:\work"));
        assert!(windows_display_prefers_backslash(r"\\server\share"));
        assert!(!windows_display_prefers_backslash("C:/work"));
        assert!(!windows_display_prefers_backslash("/external/work"));

        assert_eq!(with_trailing_separator("/", '/'), "/");
        assert_eq!(with_trailing_separator("/work/", '/'), "/work/");
        assert_eq!(with_trailing_separator(r"C:\", '\\'), r"C:\");
        assert_eq!(with_trailing_separator(r"C:\work\", '/'), "C:\\work/");

        let native = rewrite_depinfo_content_with_sentinel_for_platform(
            "__kache_root__/debug/deps/demo.d\n",
            Path::new(r"C:\work\target"),
            DEPINFO_ROOT_SENTINEL,
            DepInfoMode::Expand,
            true,
        );
        assert_eq!(native, "C:\\work\\target\\debug/deps/demo.d\n");

        let slash = rewrite_depinfo_content_with_sentinel_for_platform(
            "__kache_root__/debug/deps/demo.d\n",
            Path::new("/external/target"),
            DEPINFO_ROOT_SENTINEL,
            DepInfoMode::Expand,
            true,
        );
        assert_eq!(slash, "/external/target/debug/deps/demo.d\n");

        let mixed = rewrite_depinfo_content_with_sentinel_for_platform(
            "/build/work/target/debug/deps/demo.d\n",
            Path::new(r"/build/work\target"),
            DEPINFO_ROOT_SENTINEL,
            DepInfoMode::Relativize,
            true,
        );
        assert_eq!(mixed, "__kache_root__/debug/deps/demo.d\n");
    }

    #[test]
    fn configured_windows_roots_keep_verbatim_and_plain_slots_distinct() {
        let verbatim_drive = depinfo_prefixes_for_exact_display(r"\\?\C:\Work", true);
        assert_eq!(verbatim_drive.len(), 2);
        assert!(verbatim_drive.contains(&r"\\?\C:\Work/".to_string()));
        assert!(verbatim_drive.contains(&r"\\?\C:\Work\".to_string()));
        assert!(
            verbatim_drive
                .iter()
                .all(|prefix| prefix.starts_with(r"\\?\"))
        );
        assert!(
            !verbatim_drive
                .iter()
                .any(|prefix| prefix.starts_with(r"C:\Work"))
        );

        let verbatim_unc = depinfo_prefixes_for_exact_display(r"\\?\UNC\server\share", true);
        assert_eq!(verbatim_unc.len(), 2);
        assert!(verbatim_unc.contains(&r"\\?\UNC\server\share/".to_string()));
        assert!(verbatim_unc.contains(&r"\\?\UNC\server\share\".to_string()));
        assert!(
            verbatim_unc
                .iter()
                .all(|prefix| prefix.starts_with(r"\\?\UNC\server\share"))
        );
        assert!(
            !verbatim_unc
                .iter()
                .any(|prefix| prefix.starts_with(r"\\server\share"))
        );

        let input = r"out: \\?\C:\Work\src\verbatim.rs C:\Work\src\plain.rs \\?\UNC\server\share\src\verbatim_unc.rs \\server\share\src\plain_unc.rs";
        let producer_roots = [
            (
                PathBuf::from(r"\\?\C:\Work"),
                "__kache_base_dir_0__/".to_string(),
            ),
            (
                PathBuf::from(r"C:\Work"),
                "__kache_base_dir_1__/".to_string(),
            ),
            (
                PathBuf::from(r"\\?\UNC\server\share"),
                "__kache_base_dir_2__/".to_string(),
            ),
            (
                PathBuf::from(r"\\server\share"),
                "__kache_base_dir_3__/".to_string(),
            ),
        ];
        let stored = producer_roots
            .iter()
            .fold(input.to_string(), |content, (root, sentinel)| {
                rewrite_depinfo_content_with_exact_sentinel_for_platform(
                    &content,
                    root,
                    sentinel,
                    DepInfoMode::Relativize,
                    true,
                )
            });
        assert!(
            stored.contains(r"__kache_base_dir_0__/src\verbatim.rs"),
            "{stored}"
        );
        assert!(
            stored.contains(r"__kache_base_dir_1__/src\plain.rs"),
            "{stored}"
        );
        assert!(
            stored.contains(r"__kache_base_dir_2__/src\verbatim_unc.rs"),
            "{stored}"
        );
        assert!(
            stored.contains(r"__kache_base_dir_3__/src\plain_unc.rs"),
            "{stored}"
        );

        let consumer_roots = [
            (
                PathBuf::from(r"\\?\D:\Other"),
                "__kache_base_dir_0__/".to_string(),
            ),
            (
                PathBuf::from(r"D:\Other"),
                "__kache_base_dir_1__/".to_string(),
            ),
            (
                PathBuf::from(r"\\?\UNC\server2\share2"),
                "__kache_base_dir_2__/".to_string(),
            ),
            (
                PathBuf::from(r"\\server2\share2"),
                "__kache_base_dir_3__/".to_string(),
            ),
        ];
        let restored = consumer_roots
            .iter()
            .fold(stored, |content, (root, sentinel)| {
                rewrite_depinfo_content_with_exact_sentinel_for_platform(
                    &content,
                    root,
                    sentinel,
                    DepInfoMode::Expand,
                    true,
                )
            });
        assert!(restored.contains(r"\\?\D:\Other\src\verbatim.rs"));
        assert!(restored.contains(r"D:\Other\src\plain.rs"));
        assert!(restored.contains(r"\\?\UNC\server2\share2\src\verbatim_unc.rs"));
        assert!(restored.contains(r"\\server2\share2\src\plain_unc.rs"));
        assert!(!restored.contains(r"C:\Work"));
        assert!(!restored.contains(r"\\server\share"));
    }

    #[test]
    #[cfg(windows)]
    fn rustc_depinfo_rewrite_handles_native_windows_cwd_and_env_dep() {
        let input = "D:\\WORK TREE\\target\\debug\\deps\\demo.d: \
D:\\WORK TREE\\crate\\src\\lib.rs\n\
# env-dep:ASSET=D:\\\\WORK TREE\\\\crate\\\\asset.txt\n";
        let stored = rewrite_rustc_depinfo_content(
            input,
            Path::new(r"d:\work tree\target"),
            Path::new(r"d:\work tree\crate"),
            Some(Path::new(r"d:\work tree")),
            DepInfoMode::Relativize,
        );
        assert!(stored.contains("__kache_root__/debug\\deps\\demo.d"));
        assert!(stored.contains("__kache_cwd__/src\\lib.rs"));
        assert!(stored.contains("# env-dep:ASSET=__kache_cwd__/asset.txt"));
        assert!(!stored.to_ascii_lowercase().contains(r"d:\work tree"));
    }

    #[test]
    #[cfg(unix)]
    fn depinfo_relativize_matches_a_canonical_symlink_target() {
        let dir = tempfile::tempdir().unwrap();
        let real = dir.path().join("real-target");
        let alias = dir.path().join("target-alias");
        std::fs::create_dir_all(&real).unwrap();
        std::os::unix::fs::symlink(&real, &alias).unwrap();
        let canonical = real.canonicalize().unwrap();
        let input = format!(
            "{}/debug/demo.d: {}/debug/input.rs\n",
            canonical.display(),
            canonical.display()
        );

        let stored = rewrite_depinfo_content(&input, &alias, DepInfoMode::Relativize);
        assert_eq!(stored.matches(DEPINFO_ROOT_SENTINEL).count(), 2, "{stored}");
        assert!(!stored.contains(canonical.to_str().unwrap()), "{stored}");
    }

    /// kunobi-ninja/kache#330: Windows dep-info uses backslash separators and
    /// often mixes them (an env-var join appends with `/`). Relativize must
    /// catch both spellings of the prefix, or the builder's absolute paths
    /// ship in the cached entry and poison every other project sharing it.
    #[test]
    #[cfg(windows)]
    fn test_depinfo_relativize_handles_windows_separators() {
        let input = "S:\\proj\\target\\debug\\deps\\demo.d: \
S:\\proj\\target\\debug\\build\\demo-8a22\\out/generated.rs\n";

        let rewritten = rewrite_depinfo_content(
            input,
            Path::new("S:\\proj\\target"),
            DepInfoMode::Relativize,
        );
        assert!(
            !rewritten.contains("S:\\proj\\target"),
            "the builder's absolute prefix must not survive: {rewritten}"
        );
        assert_eq!(
            rewritten.matches(DEPINFO_ROOT_SENTINEL).count(),
            2,
            "both references relativize: {rewritten}"
        );

        // Expansion re-roots at the consumer using its native separator.
        let expanded = rewrite_depinfo_content(
            &rewritten,
            Path::new("T:\\other\\target"),
            DepInfoMode::Expand,
        );
        assert!(
            expanded.contains("T:\\other\\target\\debug\\build\\demo-8a22\\out/generated.rs"),
            "consumer-rooted mixed-separator path: {expanded}"
        );
        assert!(!expanded.contains(DEPINFO_ROOT_SENTINEL));
    }

    /// Mirrors cargo's `unescape_env` grammar for `# env-dep:` values: only
    /// `\\`, `\n`, and `\r` are legal escapes; anything else makes cargo fail
    /// the whole compile with "unknown escape character" (kunobi-ninja/kache#730).
    fn cargo_env_dep_value_parses(value: &str) -> bool {
        let mut chars = value.trim_end_matches('\n').chars();
        while let Some(c) = chars.next() {
            if c == '\\' {
                match chars.next() {
                    Some('\\') | Some('n') | Some('r') => {}
                    _ => return false,
                }
            }
        }
        true
    }

    /// The oracle itself must be red for the corruption this guards against:
    /// an orphan single backslash before `x86_64` is exactly what cargo
    /// rejected in the nightly Firefox/Windows bench.
    #[test]
    fn test_env_dep_grammar_rejects_orphan_escape() {
        assert!(cargo_env_dep_value_parses(
            r"C:\\proj\\obj\\x86_64-pc-windows-msvc"
        ));
        assert!(!cargo_env_dep_value_parses(
            r"__kache_root__/obj\x86_64-pc-windows-msvc"
        ));
        assert!(!cargo_env_dep_value_parses(r"C:\proj\obj"));
    }

    /// kunobi-ninja/kache#730: `# env-dep:` values escape backslashes as
    /// `\\`. Relativize must match the escaped spelling of a backslash
    /// anchor, never leaving half an escape pair behind, and the result must
    /// stay parseable under cargo's env-dep grammar.
    #[test]
    fn test_depinfo_env_dep_relativize_matches_escaped_prefix() {
        let input = "target\\debug\\deps\\demo.d: C:\\proj\\src\\lib.rs\n\
                     # env-dep:OUT_DIR=C:\\\\proj\\\\obj\\\\x86_64-pc-windows-msvc\\\\out\n";
        let prefixes = ["C:\\proj/".to_string(), "C:\\proj\\".to_string()];

        let rewritten =
            rewrite_depinfo_content_with_prefixes(input, &prefixes, DepInfoMode::Relativize);

        let env_line = rewritten
            .lines()
            .find(|l| l.starts_with(DEPINFO_ENV_DEP_PREFIX))
            .unwrap();
        assert!(
            env_line.contains("# env-dep:OUT_DIR=__kache_root__/obj\\\\x86_64"),
            "escaped anchor must relativize whole, not split an escape pair: {rewritten}"
        );
        assert!(
            cargo_env_dep_value_parses(env_line.split_once('=').unwrap().1),
            "relativized env-dep value must stay parseable: {rewritten}"
        );
        assert!(
            rewritten.contains("__kache_root__/src\\lib.rs"),
            "file-list line keeps the plain-spelling rewrite: {rewritten}"
        );
    }

    /// env-dep record keys are environment variable NAMES, never paths:
    /// a key-only record (no `=`) and the key half of a record must survive
    /// both directions byte-for-byte (cross-model review finding).
    #[test]
    fn test_depinfo_env_dep_never_rewrites_keys() {
        let input = "# env-dep:__kache_root__/WEIRD_KEY\n\
                     # env-dep:__kache_root__/ALSO_WEIRD=__kache_root__/value\n";
        let prefixes = ["C:\\proj/".to_string(), "C:\\proj\\".to_string()];

        let expanded = rewrite_depinfo_content_with_prefixes(input, &prefixes, DepInfoMode::Expand);
        assert!(
            expanded.contains("# env-dep:__kache_root__/WEIRD_KEY\n"),
            "key-only record must be untouched: {expanded}"
        );
        assert!(
            expanded.contains("# env-dep:__kache_root__/ALSO_WEIRD=C:\\\\proj/value"),
            "only the value half may expand: {expanded}"
        );
    }

    /// kunobi-ninja/kache#730, the shape that broke the nightly bench: an
    /// entry relativized under a forward-spelled anchor, expanded under a
    /// backslash-spelled anchor. Expand must insert the escaped spelling into
    /// `# env-dep:` lines or cargo rejects the restored `.d` outright.
    #[test]
    fn test_depinfo_env_dep_expand_inserts_escaped_prefix() {
        let stored = "target/debug/deps/demo.d: __kache_root__/src/lib.rs\n\
                      # env-dep:OUT_DIR=__kache_root__/obj-kache-bench\\\\x86_64-pc-windows-msvc\\\\out\n";
        let prefixes = [
            "C:\\other\\clone-b/".to_string(),
            "C:\\other\\clone-b\\".to_string(),
        ];

        let expanded =
            rewrite_depinfo_content_with_prefixes(stored, &prefixes, DepInfoMode::Expand);

        let env_line = expanded
            .lines()
            .find(|l| l.starts_with(DEPINFO_ENV_DEP_PREFIX))
            .unwrap();
        assert!(
            env_line.contains("OUT_DIR=C:\\\\other\\\\clone-b/obj-kache-bench\\\\x86_64"),
            "expanded env-dep anchor must be escaped: {expanded}"
        );
        assert!(
            cargo_env_dep_value_parses(env_line.split_once('=').unwrap().1),
            "expanded env-dep value must stay parseable: {expanded}"
        );
        assert!(
            expanded.contains("demo.d: C:\\other\\clone-b/src/lib.rs"),
            "file-list line takes the plain anchor (rustc leaves backslashes raw there): {expanded}"
        );

        // Round trip: re-relativizing under the same anchor restores the
        // stored form, so warm entries stay stable across hits.
        let rerelativized =
            rewrite_depinfo_content_with_prefixes(&expanded, &prefixes, DepInfoMode::Relativize);
        assert_eq!(rerelativized, stored, "expand/relativize must round-trip");
    }

    #[test]
    fn test_depinfo_expand_preserves_parent_relative_paths() {
        let input = "\
foo.o: ../../src/foo.cc ../include/foo.h __kache_root__/generated.h foo/./bar.h
";

        let rewritten =
            rewrite_depinfo_content(input, Path::new("/build/worktree/obj"), DepInfoMode::Expand);

        assert!(
            rewritten.contains("../../src/foo.cc"),
            "parent-relative deps must not be expanded: {rewritten}"
        );
        assert!(
            rewritten.contains("../include/foo.h"),
            "single parent-relative deps must not be expanded: {rewritten}"
        );
        assert!(
            rewritten.contains("/build/worktree/obj/generated.h"),
            "sentinel paths must expand: {rewritten}"
        );
        assert!(
            rewritten.contains("foo/./bar.h"),
            "embedded ./ segments are compiler-owned paths: {rewritten}"
        );
    }

    #[test]
    fn test_depinfo_expand_preserves_firefox_parent_relative_depfile_paths() {
        let input = "\
Unified_mm_ettings-WrongChannel0.o: Unified_mm_ettings-WrongChannel0.mm \\
  ../../../../../../../toolkit/mozapps/update/updater/macos-frameworks/UpdateSettings/UpdateSettings.mm \\
  ../../../../../../../toolkit/mozapps/update/updater/macos-frameworks/UpdateSettings/UpdateSettings.h \\
  __kache_root__/mozilla-config.h
";
        let anchor = Path::new(
            "/Users/lenij/work/kache/tmp/bench/clone-a/obj-kache-bench\
             /toolkit/mozapps/update/updater/macos-frameworks/UpdateSettings-WrongChannel",
        );

        let rewritten = rewrite_depinfo_content(input, anchor, DepInfoMode::Expand);

        assert!(
            rewritten.contains(
                "../../../../../../../toolkit/mozapps/update/updater/macos-frameworks\
                 /UpdateSettings/UpdateSettings.mm"
            ),
            "Firefox-style parent-relative source path must survive restore: {rewritten}"
        );
        assert!(
            rewritten.contains(
                "../../../../../../../toolkit/mozapps/update/updater/macos-frameworks\
                 /UpdateSettings/UpdateSettings.h"
            ),
            "Firefox-style parent-relative header path must survive restore: {rewritten}"
        );
        assert!(
            !rewritten.contains("/./Users/") && !rewritten.contains("WrongChannel/./"),
            "restore must not inject the anchor into ../ paths: {rewritten}"
        );
        assert!(
            rewritten.contains(
                "/Users/lenij/work/kache/tmp/bench/clone-a/obj-kache-bench\
                 /toolkit/mozapps/update/updater/macos-frameworks\
                 /UpdateSettings-WrongChannel/mozilla-config.h"
            ),
            "sentinel paths should still expand at the restore anchor: {rewritten}"
        );
    }

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

        let dir = tempfile::tempdir().unwrap();
        // Simulate a blob: read-only, no execute bit (as stored in kache's blob store)
        let src = dir.path().join("blob");
        fs::write(&src, b"ELF fake binary").unwrap();
        fs::set_permissions(&src, fs::Permissions::from_mode(0o444)).unwrap();

        let dst = dir.path().join("test_binary");
        link_to_target(&src, &dst, LinkStrategy::Copy).unwrap();

        let mode = fs::metadata(&dst).unwrap().permissions().mode();
        assert_eq!(mode & 0o111, 0o111, "executable should have +x: {mode:#o}");
        assert_eq!(
            mode & 0o200,
            0o200,
            "executable should be writable: {mode:#o}"
        );
    }

    #[test]
    fn executable_permission_helper_restores_consumer_permissions() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("restored-executable");
        fs::write(&path, b"executable").unwrap();

        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            fs::set_permissions(&path, fs::Permissions::from_mode(0o400)).unwrap();
        }
        #[cfg(not(unix))]
        {
            let mut permissions = fs::metadata(&path).unwrap().permissions();
            permissions.set_readonly(true);
            fs::set_permissions(&path, permissions).unwrap();
        }

        set_executable_permissions(&path).unwrap();

        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            assert_eq!(
                fs::metadata(&path).unwrap().permissions().mode() & 0o777,
                0o755
            );
        }
        #[cfg(not(unix))]
        assert!(!fs::metadata(&path).unwrap().permissions().readonly());
    }

    #[test]
    fn writable_staging_file_is_created_beside_target() {
        let dir = tempfile::tempdir().unwrap();
        let parent = dir.path().join("outputs");
        fs::create_dir(&parent).unwrap();
        let target = parent.join("output.o");

        let _staged = new_writable_staging_file(&target).unwrap();
        let entries: Vec<_> = fs::read_dir(&parent)
            .unwrap()
            .map(|entry| entry.unwrap().file_name())
            .collect();
        assert_eq!(entries.len(), 1);
        assert!(
            entries[0]
                .to_str()
                .unwrap()
                .starts_with(".kache-cc-restore-")
        );
    }

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

        let dir = tempfile::tempdir().unwrap();
        let src_dir = tempfile::tempdir().unwrap(); // different tempdir to force cross-dir

        let src = src_dir.path().join("blob.rlib");
        fs::write(&src, b"rlib content").unwrap();
        fs::set_permissions(&src, fs::Permissions::from_mode(0o444)).unwrap();

        let dst = dir.path().join("output.rlib");

        // Hardlink should succeed (same filesystem), so test copy_file directly
        copy_file(&src, &dst, false).unwrap();

        let mode = fs::metadata(&dst).unwrap().permissions().mode();
        assert_eq!(
            mode & 0o111,
            0,
            "non-executable should NOT have +x: {mode:#o}"
        );
        assert_eq!(mode & 0o200, 0o200, "should be writable: {mode:#o}");
    }

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

        let dir = tempfile::tempdir().unwrap();
        // Blob stored as read-only (exactly how kache stores them)
        let src = dir.path().join("blob");
        fs::write(&src, b"test binary content").unwrap();
        fs::set_permissions(&src, fs::Permissions::from_mode(0o444)).unwrap();

        let dst = dir.path().join("my_test-abc123");
        link_to_target(&src, &dst, LinkStrategy::Copy).unwrap();

        // Must be executable (cargo test will try to run this)
        let mode = fs::metadata(&dst).unwrap().permissions().mode();
        assert_eq!(mode & 0o755, 0o755, "expected 0o755, got {mode:#o}");
    }

    #[cfg(unix)]
    #[test]
    fn write_restored_over_readonly_blob_link_isolates_and_stays_writable() {
        // The bug this guards: on restore, a `.d` was linked to the
        // read-only store blob, then a post-restore rewrite tried to
        // edit it in place — failing on the 0o444 mode (reflink case)
        // or corrupting the shared blob (hardlink case). `write_restored`
        // instead materializes the final content as a fresh file.
        use std::os::unix::fs::{MetadataExt, PermissionsExt};

        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.d");
        let target = dir.path().join("sub/restored.d");

        // A read-only store blob, and a prior restore that hardlinked it
        // into place (the worst case — shared inode + read-only).
        fs::write(&blob, b"OLD RELATIVIZED CONTENT").unwrap();
        fs::set_permissions(&blob, fs::Permissions::from_mode(0o444)).unwrap();
        fs::create_dir_all(target.parent().unwrap()).unwrap();
        fs::hard_link(&blob, &target).unwrap();

        write_restored(&target, b"NEW EXPANDED CONTENT", LinkStrategy::Hardlink).unwrap();

        // Final content is in place...
        assert_eq!(fs::read(&target).unwrap(), b"NEW EXPANDED CONTENT");
        // ...the restored file is writable (an in-place edit could not
        // have failed on it)...
        let mode = fs::metadata(&target).unwrap().permissions().mode();
        assert_eq!(
            mode & 0o200,
            0o200,
            "restored file must be writable: {mode:#o}"
        );
        // ...it shares no inode with the blob...
        assert_ne!(
            fs::metadata(&target).unwrap().ino(),
            fs::metadata(&blob).unwrap().ino(),
            "restored file must not share an inode with the store blob"
        );
        // ...and the store blob is byte-for-byte untouched.
        assert_eq!(fs::read(&blob).unwrap(), b"OLD RELATIVIZED CONTENT");
        assert!(
            fs::metadata(&blob).unwrap().permissions().readonly(),
            "store blob must remain read-only"
        );
    }

    #[test]
    fn write_restored_creates_missing_parent_dirs() {
        let dir = tempfile::tempdir().unwrap();
        let target = dir.path().join("a/b/c/out.d");
        write_restored(&target, b"content", LinkStrategy::Hardlink).unwrap();
        assert_eq!(fs::read(&target).unwrap(), b"content");
    }

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

        let dir = tempfile::tempdir().unwrap();
        let target = dir.path().join("deps/e2e-abc123");

        // Copy strategy is what Executable / DynamicLibrary kinds use:
        // a materialized executable must be runnable by the OS (#298).
        write_restored(&target, b"ELF fake binary", LinkStrategy::Copy).unwrap();

        let mode = fs::metadata(&target).unwrap().permissions().mode();
        assert_eq!(mode & 0o111, 0o111, "executable should have +x: {mode:#o}");
        assert_eq!(
            mode & 0o200,
            0o200,
            "executable should be writable: {mode:#o}"
        );
    }

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

        let dir = tempfile::tempdir().unwrap();
        let target = dir.path().join("sub/restored.d");

        // Hardlink strategy backs dep-info (.d) and other immutable kinds:
        // materialized content stays a plain 0o644 file, never executable.
        write_restored(&target, b"deps: src.rs", LinkStrategy::Hardlink).unwrap();

        let mode = fs::metadata(&target).unwrap().permissions().mode();
        assert_eq!(
            mode & 0o111,
            0,
            "non-executable must NOT have +x: {mode:#o}"
        );
        assert_eq!(mode & 0o200, 0o200, "should be writable: {mode:#o}");
    }

    #[test]
    fn link_to_target_optimistic_existing_parent() {
        let dir = tempfile::TempDir::new().unwrap();
        let src = dir.path().join("source.txt");
        fs::write(&src, "test").unwrap();

        // Parent exists
        let dst_dir = dir.path().join("out");
        fs::create_dir_all(&dst_dir).unwrap();
        let dst = dst_dir.join("dest.txt");

        super::link_to_target(&src, &dst, LinkStrategy::Copy).unwrap();
        assert_eq!(fs::read_to_string(&dst).unwrap(), "test");
    }

    #[test]
    fn link_to_target_optimistic_missing_parent() {
        let dir = tempfile::TempDir::new().unwrap();
        let src = dir.path().join("source.txt");
        fs::write(&src, "test").unwrap();

        // Parent does NOT exist
        let dst = dir.path().join("missing").join("dest.txt");

        super::link_to_target(&src, &dst, LinkStrategy::Copy).unwrap();
        assert_eq!(fs::read_to_string(&dst).unwrap(), "test");
    }

    #[test]
    fn link_to_target_optimistic_nested_directories() {
        let dir = tempfile::TempDir::new().unwrap();
        let src = dir.path().join("source.txt");
        fs::write(&src, "test").unwrap();

        // Deeply nested missing parent
        let dst = dir
            .path()
            .join("foo")
            .join("bar")
            .join("baz")
            .join("dest.txt");

        super::link_to_target(&src, &dst, LinkStrategy::Copy).unwrap();
        assert_eq!(fs::read_to_string(&dst).unwrap(), "test");
    }

    #[test]
    #[cfg(unix)] // Permissions are easier to test on Unix
    fn link_to_target_permission_denied_does_not_loop() {
        let dir = tempfile::TempDir::new().unwrap();
        let src = dir.path().join("source.txt");
        fs::write(&src, "test").unwrap();

        let dst_dir = dir.path().join("readonly");
        fs::create_dir_all(&dst_dir).unwrap();

        // Make directory read-only
        use std::os::unix::fs::PermissionsExt;
        fs::set_permissions(&dst_dir, fs::Permissions::from_mode(0o555)).unwrap();

        let dst = dst_dir.join("dest.txt");

        let err = super::link_to_target(&src, &dst, LinkStrategy::Copy).unwrap_err();
        let io_err = err.downcast_ref::<std::io::Error>().unwrap();
        assert_eq!(io_err.kind(), std::io::ErrorKind::PermissionDenied);

        // Restore permissions so TempDir can clean up
        fs::set_permissions(&dst_dir, fs::Permissions::from_mode(0o755)).unwrap();
    }

    #[test]
    #[cfg(unix)]
    fn link_to_target_missing_parent_permission_denied_surfaces_real_cause() {
        let dir = tempfile::TempDir::new().unwrap();
        let src = dir.path().join("source.txt");
        fs::write(&src, "test").unwrap();

        let dst_dir = dir.path().join("readonly_parent");
        fs::create_dir_all(&dst_dir).unwrap();

        use std::os::unix::fs::PermissionsExt;
        fs::set_permissions(&dst_dir, fs::Permissions::from_mode(0o555)).unwrap();

        let dst = dst_dir.join("nested").join("dest.txt");

        let err = super::link_to_target(&src, &dst, LinkStrategy::Copy).unwrap_err();
        let io_err = err.downcast_ref::<std::io::Error>().unwrap();
        assert_eq!(io_err.kind(), std::io::ErrorKind::PermissionDenied);

        fs::set_permissions(&dst_dir, fs::Permissions::from_mode(0o755)).unwrap();
    }

    #[test]
    #[cfg(unix)]
    fn write_restored_missing_parent_permission_denied_surfaces_real_cause() {
        let dir = tempfile::TempDir::new().unwrap();
        let dst_dir = dir.path().join("readonly_parent");
        fs::create_dir_all(&dst_dir).unwrap();

        use std::os::unix::fs::PermissionsExt;
        fs::set_permissions(&dst_dir, fs::Permissions::from_mode(0o555)).unwrap();

        let dst = dst_dir.join("nested").join("dest.txt");

        let err = super::write_restored(&dst, b"test", LinkStrategy::Copy).unwrap_err();
        let io_err = err.downcast_ref::<std::io::Error>().unwrap();
        assert_eq!(io_err.kind(), std::io::ErrorKind::PermissionDenied);

        fs::set_permissions(&dst_dir, fs::Permissions::from_mode(0o755)).unwrap();
    }

    #[test]
    fn classify_hardlink_cross_device_maps_to_cross_device() {
        assert_eq!(
            classify_hardlink_io_error(std::io::ErrorKind::CrossesDevices),
            HardlinkIoReason::CrossDevice,
        );
    }

    #[test]
    fn classify_hardlink_permission_denied_maps_to_permission() {
        assert_eq!(
            classify_hardlink_io_error(std::io::ErrorKind::PermissionDenied),
            HardlinkIoReason::Permission,
        );
    }

    #[test]
    fn classify_hardlink_other_errno_maps_to_other() {
        // EMLINK (too many links) and any other errno are not EXDEV/EPERM:
        // they must not be misreported as a layout problem.
        assert_eq!(
            classify_hardlink_io_error(std::io::ErrorKind::AlreadyExists),
            HardlinkIoReason::Other,
        );
    }

    #[test]
    fn injected_cross_device_restore_records_cross_device_reason() {
        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.rlib");
        fs::write(&blob, b"cached rlib").unwrap();
        let bytes = fs::metadata(&blob).unwrap().len();
        let target = dir.path().join("tree.rlib");

        let before = crate::opcounts::restore_copy_cross_device_bytes();
        let _guard = InjectRestoreHardlinkError::enable(std::io::ErrorKind::CrossesDevices);
        hardlink_or_copy(&blob, &target, bytes).unwrap();
        assert!(
            crate::opcounts::restore_copy_cross_device_bytes() >= before + bytes,
            "EXDEV injection must record the cross-device reason"
        );
        assert_eq!(fs::read(&target).unwrap(), b"cached rlib");
    }

    #[test]
    fn injected_permission_restore_records_permission_reason() {
        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.rlib");
        fs::write(&blob, b"cached rlib").unwrap();
        let bytes = fs::metadata(&blob).unwrap().len();
        let target = dir.path().join("tree.rlib");

        let before = crate::opcounts::restore_copy_permission_bytes();
        let _guard = InjectRestoreHardlinkError::enable(std::io::ErrorKind::PermissionDenied);
        hardlink_or_copy(&blob, &target, bytes).unwrap();
        assert!(
            crate::opcounts::restore_copy_permission_bytes() >= before + bytes,
            "EPERM injection must record the permission reason"
        );
        assert_eq!(fs::read(&target).unwrap(), b"cached rlib");
    }

    #[test]
    fn injected_other_restore_records_other_reason() {
        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.rlib");
        fs::write(&blob, b"cached rlib").unwrap();
        let bytes = fs::metadata(&blob).unwrap().len();
        let target = dir.path().join("tree.rlib");

        let before = crate::opcounts::restore_copy_other_bytes();
        let _guard = InjectRestoreHardlinkError::enable(std::io::ErrorKind::AlreadyExists);
        hardlink_or_copy(&blob, &target, bytes).unwrap();
        assert!(
            crate::opcounts::restore_copy_other_bytes() >= before + bytes,
            "other errno injection must record the other reason"
        );
        assert_eq!(fs::read(&target).unwrap(), b"cached rlib");
    }

    /// The shared-restore flag is process-global; tests that read or set it
    /// take turns.
    static SHARED_TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());

    #[test]
    #[cfg(unix)]
    fn shared_restores_let_every_tree_link_the_blob() {
        use std::os::unix::fs::MetadataExt;
        let _guard = SHARED_TEST_LOCK.lock().unwrap();
        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.rlib");
        fs::write(&blob, b"cached rlib").unwrap();
        let bytes = fs::metadata(&blob).unwrap().len();
        set_shared_hardlink_restores(true);
        let result = (|| {
            let first = dir.path().join("tree-a.rlib");
            hardlink_or_copy(&blob, &first, bytes)?;
            let second = dir.path().join("tree-b.rlib");
            hardlink_or_copy(&blob, &second, bytes)?;
            Ok::<_, anyhow::Error>((first, second))
        })();
        set_shared_hardlink_restores(false);
        let (first, second) = result.unwrap();
        assert_eq!(
            fs::metadata(&blob).unwrap().nlink(),
            3,
            "both trees link the blob"
        );
        assert_eq!(
            fs::metadata(&first).unwrap().ino(),
            fs::metadata(&second).unwrap().ino()
        );
    }

    #[test]
    #[cfg(unix)]
    fn exclusive_precheck_restore_records_exclusive_reason() {
        let _guard = SHARED_TEST_LOCK.lock().unwrap();
        use std::os::unix::fs::MetadataExt;

        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob.rlib");
        fs::write(&blob, b"cached rlib").unwrap();
        let bytes = fs::metadata(&blob).unwrap().len();

        // Occupy the single exclusive slot: blob + first consumer.
        let first = dir.path().join("tree-a.rlib");
        hardlink_or_copy(&blob, &first, bytes).unwrap();
        assert_eq!(fs::metadata(&blob).unwrap().nlink(), 2);

        let before = crate::opcounts::restore_copy_exclusive_bytes();
        let second = dir.path().join("tree-b.rlib");
        hardlink_or_copy(&blob, &second, bytes).unwrap();
        assert!(
            crate::opcounts::restore_copy_exclusive_bytes() >= before + bytes,
            "second consumer must record the exclusive-carrier reason"
        );
        assert_ne!(
            fs::metadata(&second).unwrap().ino(),
            fs::metadata(&blob).unwrap().ino(),
            "exclusive fallback must be a private copy"
        );
    }

    #[test]
    #[cfg(unix)]
    fn missing_blob_restore_records_other_reason() {
        let dir = tempfile::tempdir().unwrap();
        let missing = dir.path().join("no-such-blob.rlib");
        let target = dir.path().join("tree.rlib");

        let before = crate::opcounts::restore_copy_other_bytes();
        // Metadata cannot verify exclusivity: conservative copy with Other.
        let result = hardlink_or_copy(&missing, &target, 0);
        // Copy of a missing source fails, but the Other reason is recorded
        // before the fallback runs.
        assert!(result.is_err());
        assert!(
            crate::opcounts::restore_copy_other_bytes() >= before,
            "unverifiable blob must record the other reason"
        );
    }

    // ── ExecutableHardlink: proc-macros and build scripts ──────────────────

    /// Stands in for a filesystem without reflink, so the no-CoW fallbacks run
    /// on APFS too.
    #[cfg(unix)]
    fn no_reflink(_: &Path, _: &Path) -> Result<()> {
        anyhow::bail!("reflink disabled for this test")
    }

    #[cfg(unix)]
    fn mode_of(path: &Path) -> u32 {
        use std::os::unix::fs::PermissionsExt;
        fs::metadata(path).unwrap().permissions().mode() & 0o777
    }

    #[cfg(unix)]
    fn ino_of(path: &Path) -> u64 {
        use std::os::unix::fs::MetadataExt;
        fs::metadata(path).unwrap().ino()
    }

    #[cfg(unix)]
    fn running_as_root() -> bool {
        unsafe { libc::geteuid() == 0 }
    }

    #[test]
    fn executable_blob_is_shareable_only_when_runnable_and_read_only() {
        // A local blob of a linked output: 0o755 made read-only.
        assert!(executable_blob_is_shareable(Some(0o555)));
        assert!(executable_blob_is_shareable(Some(0o100555)));
        // Remote imports land at 0o444 and would not run from a link.
        assert!(!executable_blob_is_shareable(Some(0o444)));
        assert!(!executable_blob_is_shareable(Some(0o554)));
        assert!(!executable_blob_is_shareable(Some(0o455)));
        // A writable blob would let a writer in one tree reach the store.
        assert!(!executable_blob_is_shareable(Some(0o755)));
        assert!(!executable_blob_is_shareable(Some(0o557)));
        assert!(!executable_blob_is_shareable(None));
    }

    /// Without CoW, a runnable read-only blob is shared like an rlib. The link
    /// keeps the blob's mode: making it writable or chmod'ing it would change
    /// the store blob as well.
    #[cfg(unix)]
    #[test]
    fn executable_hardlink_without_reflink_shares_a_runnable_blob_read_only() {
        let _guard = SHARED_TEST_LOCK.lock().unwrap();
        use std::os::unix::fs::PermissionsExt;

        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob");
        let target = dir.path().join("deps/libserde_derive-1.so");
        fs::write(&blob, b"proc-macro dylib").unwrap();
        fs::set_permissions(&blob, fs::Permissions::from_mode(0o555)).unwrap();

        link_to_target_via(&blob, &target, LinkStrategy::ExecutableHardlink, no_reflink).unwrap();

        assert_eq!(ino_of(&target), ino_of(&blob), "expected a shared inode");
        assert_eq!(mode_of(&target), 0o555);
        assert_eq!(mode_of(&blob), 0o555, "the restore must not chmod the blob");
        assert_eq!(fs::read(&target).unwrap(), b"proc-macro dylib");
        if !running_as_root() {
            let error = fs::OpenOptions::new()
                .write(true)
                .open(&target)
                .unwrap_err();
            assert_eq!(error.kind(), std::io::ErrorKind::PermissionDenied);
        }
    }

    /// A blob that cannot run as it is, or that is writable, is copied to a
    /// private runnable file, exactly as the `Copy` strategy would.
    #[cfg(unix)]
    #[test]
    fn executable_hardlink_without_reflink_copies_a_blob_it_cannot_share() {
        let _guard = SHARED_TEST_LOCK.lock().unwrap();
        use std::os::unix::fs::PermissionsExt;

        let dir = tempfile::tempdir().unwrap();
        for blob_mode in [0o444, 0o755] {
            let blob = dir.path().join(format!("blob-{blob_mode:o}"));
            let target = dir.path().join(format!("build_script_build-{blob_mode:o}"));
            fs::write(&blob, b"build script").unwrap();
            fs::set_permissions(&blob, fs::Permissions::from_mode(blob_mode)).unwrap();

            link_to_target_via(&blob, &target, LinkStrategy::ExecutableHardlink, no_reflink)
                .unwrap();

            assert_ne!(ino_of(&target), ino_of(&blob), "{blob_mode:o}");
            assert_eq!(mode_of(&target), 0o755, "{blob_mode:o}");
            assert_eq!(mode_of(&blob), blob_mode);
            fs::write(&target, b"rebuilt").unwrap();
            assert_eq!(fs::read(&blob).unwrap(), b"build script");
        }
    }

    /// With reflink the restore is a private clone, runnable and writable
    /// even when the blob itself carries no execute bit.
    #[cfg(unix)]
    #[test]
    fn executable_hardlink_reflink_restore_is_a_private_runnable_file() {
        use std::os::unix::fs::PermissionsExt;

        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob");
        let probe = dir.path().join("probe");
        let target = dir.path().join("libfoo_macros-1.so");
        fs::write(&blob, b"proc-macro dylib").unwrap();
        fs::set_permissions(&blob, fs::Permissions::from_mode(0o444)).unwrap();
        if let Err(error) = try_reflink(&blob, &probe) {
            eprintln!("reflink unavailable on test filesystem: {error}");
            return;
        }

        link_to_target(&blob, &target, LinkStrategy::ExecutableHardlink).unwrap();

        assert_ne!(ino_of(&target), ino_of(&blob));
        assert_eq!(mode_of(&target), 0o755);
        fs::write(&target, b"rebuilt").unwrap();
        assert_eq!(fs::read(&blob).unwrap(), b"proc-macro dylib");
        assert_eq!(mode_of(&blob), 0o444);
    }

    /// kunobi-ninja/kache#677 and #794 for shared executables. Tree A holds
    /// the one link to the blob. Restoring the same blob into tree B must not
    /// re-date or rewrite what cargo sees in tree A, tree B must still get a
    /// runnable file, and a compile in either tree must not write through
    /// into the store or the other tree.
    #[cfg(unix)]
    #[test]
    fn executable_hardlink_restore_into_tree_b_leaves_tree_a_untouched() {
        let _guard = SHARED_TEST_LOCK.lock().unwrap();
        use std::fs::File;
        use std::os::unix::fs::PermissionsExt;

        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob");
        let tree_a = dir.path().join("a/build/pkg-1/build_script_build-1");
        let tree_b = dir.path().join("b/build/pkg-1/build_script_build-1");
        fs::write(&blob, b"build script").unwrap();
        fs::set_permissions(&blob, fs::Permissions::from_mode(0o555)).unwrap();

        link_to_target_via(&blob, &tree_a, LinkStrategy::ExecutableHardlink, no_reflink).unwrap();
        touch_mtime_write_clock(&tree_a).unwrap();
        assert_eq!(
            ino_of(&tree_a),
            ino_of(&blob),
            "fixture must engage the hardlink fallback"
        );
        let observed = filetime::FileTime::from_unix_time(1_000_000_000, 0);
        filetime::set_file_mtime(&tree_a, observed).unwrap();
        let reader = File::open(&tree_a).unwrap();

        link_to_target_via(&blob, &tree_b, LinkStrategy::ExecutableHardlink, no_reflink).unwrap();
        touch_mtime_write_clock(&tree_b).unwrap();

        let seen = filetime::FileTime::from_last_modification_time(&reader.metadata().unwrap());
        assert_eq!(seen, observed, "restoring tree B re-dated tree A");
        assert_eq!(fs::read(&tree_a).unwrap(), b"build script");
        assert_ne!(ino_of(&tree_b), ino_of(&tree_a), "trees must not share");
        assert_eq!(mode_of(&tree_b), 0o755, "tree B must still be runnable");

        // A compile in tree B rewrites its private copy in place.
        fs::write(&tree_b, b"rebuilt in b").unwrap();
        assert_eq!(fs::read(&blob).unwrap(), b"build script");
        assert_eq!(fs::read(&tree_a).unwrap(), b"build script");

        // Tree A's shared inode refuses an in-place write. A compile there
        // replaces the read-only output (the wrapper's pre-clean, or the
        // linker's own unlink) and never reaches the blob.
        if !running_as_root() {
            assert!(fs::OpenOptions::new().write(true).open(&tree_a).is_err());
        }
        fs::remove_file(&tree_a).unwrap();
        fs::write(&tree_a, b"rebuilt in a").unwrap();
        assert_eq!(fs::read(&blob).unwrap(), b"build script");
        assert_eq!(fs::read(&tree_b).unwrap(), b"rebuilt in b");
        assert_eq!(mode_of(&blob), 0o555);
    }

    /// A shareable blob whose `link(2)` fails, for example with the store on
    /// another filesystem, is copied to a private file that still runs.
    #[cfg(unix)]
    #[test]
    fn executable_hardlink_link_failure_copies_a_runnable_file() {
        let _guard = SHARED_TEST_LOCK.lock().unwrap();
        use std::os::unix::fs::PermissionsExt;

        let dir = tempfile::tempdir().unwrap();
        for kind in [
            std::io::ErrorKind::CrossesDevices,
            std::io::ErrorKind::PermissionDenied,
        ] {
            let blob = dir.path().join(format!("blob-{kind:?}"));
            let target = dir.path().join(format!("build_script_build-{kind:?}"));
            fs::write(&blob, b"build script").unwrap();
            fs::set_permissions(&blob, fs::Permissions::from_mode(0o555)).unwrap();
            let _inject = InjectRestoreHardlinkError::enable(kind);

            link_to_target_via(&blob, &target, LinkStrategy::ExecutableHardlink, no_reflink)
                .unwrap();

            assert_ne!(ino_of(&target), ino_of(&blob), "{kind:?}");
            assert_eq!(mode_of(&target), 0o755, "{kind:?}");
            assert_eq!(mode_of(&blob), 0o555, "{kind:?}");
        }
    }

    /// A restorer that another link beats between its precheck and its own
    /// link drops the link and gets a private file that still runs.
    #[cfg(unix)]
    #[test]
    fn executable_hardlink_that_loses_the_link_race_copies_a_runnable_file() {
        let _guard = SHARED_TEST_LOCK.lock().unwrap();
        use std::os::unix::fs::PermissionsExt;

        let dir = tempfile::tempdir().unwrap();
        let blob = dir.path().join("blob");
        let other = dir.path().join("other/build_script_build-1");
        let target = dir.path().join("build_script_build-1");
        fs::create_dir_all(other.parent().unwrap()).unwrap();
        fs::write(&blob, b"build script").unwrap();
        fs::set_permissions(&blob, fs::Permissions::from_mode(0o555)).unwrap();
        let bytes = fs::metadata(&blob).unwrap().len();

        hardlink_or_copy_with_prelink_hook(&blob, &target, bytes, true, || {
            fs::hard_link(&blob, &other).unwrap();
        })
        .unwrap();

        assert_ne!(ino_of(&target), ino_of(&blob));
        assert_eq!(mode_of(&target), 0o755);
        assert_eq!(ino_of(&other), ino_of(&blob), "the winner keeps its link");
    }

    #[cfg(unix)]
    #[test]
    fn write_restored_executable_hardlink_strategy_sets_executable() {
        let dir = tempfile::tempdir().unwrap();
        let target = dir.path().join("deps/libfoo_macros-1.so");

        write_restored(
            &target,
            b"proc-macro dylib",
            LinkStrategy::ExecutableHardlink,
        )
        .unwrap();

        assert_eq!(mode_of(&target), 0o755);
    }
}