//! Fingerprint-keyed build-artifact cache (Phase 5).
//!
//! The killer feature. Given a cargo workspace, compute a
//! deterministic 32-byte BLAKE3 fingerprint over the inputs that
//! determine what artifacts should be produced ([`FingerprintInputs`])
//! and use it as the primary key for cached target dirs.
//!
//! # Flow
//!
//! On a CI runner or dev machine:
//!
//! ```text
//! 1. compute_fingerprint(workspace, profile, features)
//! 2. cluster::rpc::call_blob_stat(conn, fingerprint_as_blob_id)
//!    → hit: BlobGetStream → restore_target(bytes, target/) → cargo builds
//!      the workspace's OWN crates in seconds
//!    → miss: cargo build --release ... (full build)
//!             then capture_target(target/) → BlobPutStream + PutManifest
//! ```
//!
//! # What's captured, what's not
//!
//! Portable subset only:
//!   * `target/<profile>/deps/`  — the 95% of size, compiled dep rlibs
//!   * `target/<profile>/.fingerprint/` — cargo's own per-crate mtime
//!     state; without it cargo doesn't trust the deps and rebuilds them
//!   * `target/<profile>/build/` — build.rs outputs (OUT_DIR files)
//!   * `target/<profile>/examples/` — sometimes referenced by test bins
//!   * Small top-level files: `.cargo-lock`, `.rustc_info.json`, `CACHEDIR.TAG`
//!
//! Explicitly NOT captured:
//!   * `incremental/` — rustc's per-machine incremental cache. It's tied
//!     to absolute paths + rustc's own machine state; restoring it on
//!     another host silently corrupts the build. Cargo re-populates it
//!     locally on the first rebuild (~1 minute penalty).
//!   * `*.d` dependency-info files with absolute paths — cargo regenerates
//!     these from `.fingerprint/`.

use anyhow::{bail, Context, Result};
use blake3::Hasher;
use serde::{Deserialize, Serialize};
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::process::Command;

/// Top-level directories captured from `target/<profile>/`.
const CAPTURED_SUBDIRS: &[&str] = &["deps", ".fingerprint", "build", "examples"];

/// Small top-level files worth preserving from `target/<profile>/`.
const CAPTURED_TOP_FILES: &[&str] = &[".cargo-lock", ".rustc_info.json", "CACHEDIR.TAG"];

/// zstd compression level for the tarball. Level 3 is the crate's
/// default: near-instant compress, ~5–10× ratio on cargo `.rlib`.
const ZSTD_LEVEL: i32 = 3;

/// Compressed-tarball payload cap. Same as the RPC bounded cap so
/// small blobs go over `BlobPut`; larger ones must use `BlobPutStream`.
/// Used only in tests to guard against runaway captures.
pub const CAPTURE_INLINE_CEILING: usize = 16 * 1024 * 1024;

/// The set of build inputs whose combined value determines whether two
/// cached artifact sets are interchangeable. All fields must be
/// captured deterministically — random ordering or wall-clock
/// dependence would blow the cache.
///
/// Missing / absent files (e.g. no `rust-toolchain.toml`) are
/// represented as empty strings, so a project that adds an empty
/// `.cargo/config.toml` gets the same fingerprint it had before
/// creating the empty file (the file's absence == the file's emptiness).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct FingerprintInputs {
    /// Full text of `Cargo.lock`. Determines the resolved dep graph.
    pub cargo_lock: String,
    /// Output of `rustc --version --verbose`. Includes the rustc
    /// commit hash, LLVM version, and host triple — anything that
    /// changes compiler output.
    pub rustc_version_verbose: String,
    /// Full text of the workspace's `.cargo/config.toml`, or `""` if
    /// absent. Captures build flags and registry overrides.
    pub cargo_config: String,
    /// Full text of the workspace's `rust-toolchain.toml`, or `""` if
    /// absent. Rustup uses this to pin a toolchain version.
    pub rust_toolchain: String,
    /// Cargo profile: typically `"dev"` or `"release"`.
    pub profile: String,
    /// Enabled features. Sorted + deduped at collection time.
    pub features: Vec<String>,
    /// `RUSTFLAGS` environment variable at build time, or `""` if unset.
    /// Affects rustc codegen (e.g. `-C target-cpu=native`).
    pub rustflags: String,
    /// Extracted host triple from `rustc_version_verbose`. Kept
    /// separately so callers can inspect it without re-parsing.
    pub target_triple: String,
}

impl FingerprintInputs {
    /// Collect from a workspace on disk. Reads Cargo.lock, shells
    /// out to `rustc --version --verbose`, reads optional config
    /// files. Errors when Cargo.lock is missing (the workspace isn't
    /// a cargo project) or rustc isn't in PATH.
    pub fn collect(workspace: &Path, profile: &str, features: &[String]) -> Result<Self> {
        let cargo_lock = std::fs::read_to_string(workspace.join("Cargo.lock"))
            .with_context(|| format!("reading Cargo.lock under {}", workspace.display()))?;

        let out = Command::new("rustc")
            .args(["--version", "--verbose"])
            .output()
            .context("running `rustc --version --verbose`")?;
        if !out.status.success() {
            bail!(
                "rustc --version --verbose failed with exit {}",
                out.status
            );
        }
        let rustc_verbose = String::from_utf8(out.stdout)
            .context("rustc --version --verbose produced non-UTF-8 output")?;

        let cargo_config = read_optional(&workspace.join(".cargo/config.toml"))?;
        let rust_toolchain = read_optional(&workspace.join("rust-toolchain.toml"))?;
        let rustflags = std::env::var("RUSTFLAGS").unwrap_or_default();

        let target_triple = rustc_verbose
            .lines()
            .find_map(|l| l.strip_prefix("host: "))
            .unwrap_or("unknown-triple")
            .to_string();

        let mut features = features.to_vec();
        features.sort();
        features.dedup();

        Ok(Self {
            cargo_lock,
            rustc_version_verbose: rustc_verbose,
            cargo_config,
            rust_toolchain,
            profile: profile.to_string(),
            features,
            rustflags,
            target_triple,
        })
    }

    /// Compute the fingerprint hash. Deterministic given identical
    /// inputs, order-independent for features (already sorted), and
    /// null-byte-separated so `["ab","c"]` hashes differently from
    /// `["a","bc"]`.
    pub fn compute(&self) -> Fingerprint {
        let mut h = Hasher::new();
        // Domain-separated by field with a fixed sentinel — different
        // versions of this struct produce different hashes without a
        // manual version tag.
        h.update(b"clawstor.fingerprint.v1\0");
        update_field(&mut h, b"cargo_lock", self.cargo_lock.as_bytes());
        update_field(
            &mut h,
            b"rustc_version_verbose",
            self.rustc_version_verbose.as_bytes(),
        );
        update_field(&mut h, b"cargo_config", self.cargo_config.as_bytes());
        update_field(&mut h, b"rust_toolchain", self.rust_toolchain.as_bytes());
        update_field(&mut h, b"profile", self.profile.as_bytes());
        h.update(b"features\0");
        for f in &self.features {
            h.update(f.as_bytes());
            h.update(b"\0");
        }
        h.update(b"\0");
        update_field(&mut h, b"rustflags", self.rustflags.as_bytes());
        update_field(&mut h, b"target_triple", self.target_triple.as_bytes());
        Fingerprint(h.finalize().into())
    }
}

/// Read a file; return `""` when it doesn't exist. Errors on any other
/// filesystem failure so a permission-denied doesn't silently produce
/// a "no file" hash.
fn read_optional(path: &Path) -> Result<String> {
    match std::fs::read_to_string(path) {
        Ok(s) => Ok(s),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(String::new()),
        Err(e) => Err(anyhow::Error::from(e))
            .with_context(|| format!("reading optional {}", path.display())),
    }
}

/// Feed one labelled field into the fingerprint hasher.
fn update_field(h: &mut Hasher, name: &[u8], value: &[u8]) {
    h.update(name);
    h.update(b"\0");
    h.update(value);
    h.update(b"\0");
}

/// 32-byte BLAKE3 fingerprint over build inputs. Shape parallels
/// [`crate::cluster::blob::BlobId`] — same content-addressing spirit,
/// but keyed to inputs rather than outputs. A fingerprint identifies
/// a cached target dir; a BlobId identifies its bytes.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct Fingerprint([u8; 32]);

impl Fingerprint {
    pub fn from_bytes(hash: [u8; 32]) -> Self {
        Self(hash)
    }

    pub fn as_bytes(&self) -> &[u8; 32] {
        &self.0
    }

    /// Lowercase hex string, 64 chars. Suitable for filenames + logs.
    pub fn to_hex(&self) -> String {
        let mut out = String::with_capacity(64);
        for b in self.0 {
            out.push_str(&format!("{b:02x}"));
        }
        out
    }
}

impl std::fmt::Display for Fingerprint {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.to_hex())
    }
}

/// Bundle the portable subset of `target/<profile>/` into a
/// zstd-compressed tarball. Returns the bytes ready to hand to
/// [`crate::cluster::blob::BlobStore::put_bytes`] (bounded inputs)
/// or `put_stream` (larger workspaces).
///
/// `target_dir` should be `<workspace>/target/<profile>` — the
/// caller resolves the profile so this function doesn't have to
/// know about cargo's directory layout beyond "the deps live here".
/// Field finding 2026-07-12: walk `src` recursively in sorted order
/// and append every regular file + directory to `tar` under
/// `<archive_prefix>/<rel>`. Two calls on byte-identical trees produce
/// byte-identical tar output (given `HeaderMode::Deterministic`), even
/// across nodes whose `read_dir` returns entries in different orders.
///
/// Symlinks are appended as symlinks (the tar crate handles the header
/// bookkeeping); anything else — sockets, fifos — is skipped.
fn append_dir_sorted<W: std::io::Write>(
    tar: &mut tar::Builder<W>,
    archive_prefix: &str,
    src: &Path,
) -> Result<()> {
    let mut stack: Vec<(PathBuf, String)> =
        vec![(src.to_path_buf(), archive_prefix.to_string())];
    while let Some((dir, archive_dir)) = stack.pop() {
        let mut entries: Vec<_> = std::fs::read_dir(&dir)
            .with_context(|| format!("reading {}", dir.display()))?
            .filter_map(|e| e.ok())
            .collect();
        // Sort by filename bytes — stable across filesystems.
        entries.sort_by(|a, b| a.file_name().cmp(&b.file_name()));
        for entry in entries {
            let ft = entry.file_type()?;
            let name = entry.file_name();
            let name_str = match name.to_str() {
                Some(s) => s,
                None => continue,
            };
            let archive_path = format!("{}/{}", archive_dir, name_str);
            let full = entry.path();
            if ft.is_dir() {
                // Push for later processing; also emit the directory
                // header so an empty dir survives the roundtrip.
                let mut header = tar::Header::new_gnu();
                header.set_entry_type(tar::EntryType::Directory);
                header.set_size(0);
                header.set_mode(0o755);
                header.set_mtime(0);
                header.set_cksum();
                tar.append_data(
                    &mut header,
                    format!("{}/", archive_path),
                    std::io::empty(),
                )?;
                stack.push((full, archive_path));
            } else if ft.is_file() {
                let mut f = std::fs::File::open(&full)
                    .with_context(|| format!("opening {}", full.display()))?;
                tar.append_file(&archive_path, &mut f)
                    .with_context(|| format!("appending {}", full.display()))?;
            } else if ft.is_symlink() {
                let link_target = std::fs::read_link(&full)
                    .with_context(|| format!("reading symlink {}", full.display()))?;
                let mut header = tar::Header::new_gnu();
                header.set_entry_type(tar::EntryType::Symlink);
                header.set_size(0);
                header.set_mode(0o777);
                header.set_mtime(0);
                header
                    .set_link_name(&link_target)
                    .context("setting symlink header link_name")?;
                header.set_cksum();
                tar.append_data(&mut header, &archive_path, std::io::empty())?;
            }
            // Other types (sockets, fifos) are skipped.
        }
    }
    Ok(())
}

/// Field finding 2026-07-12 (clawverse measurement): capture the tar
/// into `out` via a streaming writer, so the whole ~1 GB blob never
/// sits in RAM at once. Returns the byte count actually written.
///
/// Callers stream the result into `BlobPutStream` by opening `out`
/// with `tokio::fs::File::open` — that's `AsyncRead + Unpin`, which
/// is what `call_blob_put_stream` accepts. Peak RAM stays at ~zstd
/// sliding window size (few MB) regardless of source size.
pub fn capture_target_to_writer<W: std::io::Write>(
    target_dir: &Path,
    out: W,
) -> Result<u64> {
    if !target_dir.is_dir() {
        bail!(
            "target dir {} does not exist or is not a directory",
            target_dir.display()
        );
    }
    let counter = ByteCounter::new(out);
    let encoder = zstd::stream::write::Encoder::new(counter, ZSTD_LEVEL)
        .context("initialising zstd encoder")?;
    let mut tar = tar::Builder::new(encoder);
    tar.mode(tar::HeaderMode::Deterministic);
    tar.follow_symlinks(false);

    for sub in CAPTURED_SUBDIRS {
        let path = target_dir.join(sub);
        if path.is_dir() {
            append_dir_sorted(&mut tar, sub, &path)
                .with_context(|| format!("archiving {}", path.display()))?;
        }
    }
    for file in CAPTURED_TOP_FILES {
        let path = target_dir.join(file);
        if path.is_file() {
            let mut f = std::fs::File::open(&path)
                .with_context(|| format!("opening {}", path.display()))?;
            tar.append_file(file, &mut f)
                .with_context(|| format!("appending {}", path.display()))?;
        }
    }

    let encoder = tar.into_inner().context("closing tar builder")?;
    let counter = encoder.finish().context("finalising zstd stream")?;
    Ok(counter.into_bytes_written())
}

/// A small wrapper that counts bytes written to an underlying writer.
/// Used by [`capture_target_to_writer`] so the streaming path returns
/// a byte count without buffering the output.
struct ByteCounter<W> {
    inner: W,
    written: u64,
}

impl<W> ByteCounter<W> {
    fn new(inner: W) -> Self {
        Self { inner, written: 0 }
    }
    fn into_bytes_written(self) -> u64 {
        self.written
    }
}

impl<W: std::io::Write> std::io::Write for ByteCounter<W> {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        let n = self.inner.write(buf)?;
        self.written = self.written.saturating_add(n as u64);
        Ok(n)
    }
    fn flush(&mut self) -> std::io::Result<()> {
        self.inner.flush()
    }
}

/// Legacy in-memory capture. Kept as a thin wrapper over the streaming
/// variant so existing tests + callers keep working; new code should
/// prefer `capture_target_to_writer` for bounded memory.
pub fn capture_target(target_dir: &Path) -> Result<Vec<u8>> {
    let mut buf = Vec::new();
    capture_target_to_writer(target_dir, &mut buf)?;
    Ok(buf)
}

/// Reverse of [`capture_target`]. Decompresses + unpacks the captured
/// bytes into `target_dir`, which is created if missing. Existing files
/// with the same relative path are overwritten; files outside the
/// captured subdirs are untouched.
pub fn restore_target(bytes: &[u8], target_dir: &Path) -> Result<()> {
    std::fs::create_dir_all(target_dir)
        .with_context(|| format!("creating {}", target_dir.display()))?;
    let decoder =
        zstd::stream::read::Decoder::new(bytes).context("initialising zstd decoder")?;
    let mut archive = tar::Archive::new(decoder);
    archive.set_preserve_permissions(true);
    archive.set_preserve_mtime(true);
    archive
        .unpack(target_dir)
        .with_context(|| format!("unpacking into {}", target_dir.display()))?;
    Ok(())
}

/// Bare-metal convenience: capture from a workspace + profile,
/// keeping the pipeline explicit for callers who don't want the
/// full config-driven flow yet.
pub fn capture_workspace(workspace: &Path, profile: &str) -> Result<Vec<u8>> {
    let target_dir = workspace.join("target").join(profile);
    capture_target(&target_dir)
}

/// Compute a fingerprint from a workspace + profile + features.
/// Returns `(inputs, fingerprint)` so the caller can log what went
/// into the hash.
pub fn compute_workspace_fingerprint(
    workspace: &Path,
    profile: &str,
    features: &[String],
) -> Result<(FingerprintInputs, Fingerprint)> {
    let inputs = FingerprintInputs::collect(workspace, profile, features)?;
    let fp = inputs.compute();
    Ok((inputs, fp))
}

/// Return true if two byte slices are equal when interpreted as tar
/// archives (i.e. same set of entries, same content per entry). Used
/// by tests to compare captures across two workspaces that differ
/// only in fingerprint-irrelevant ways.
#[cfg(test)]
fn zstd_tar_contents_equal(a: &[u8], b: &[u8]) -> Result<bool> {
    fn extract(bytes: &[u8]) -> Result<Vec<(PathBuf, Vec<u8>)>> {
        let decoder = zstd::stream::read::Decoder::new(bytes)?;
        let mut archive = tar::Archive::new(decoder);
        let mut out: Vec<(PathBuf, Vec<u8>)> = Vec::new();
        for entry in archive.entries()? {
            let mut entry = entry?;
            let path = entry.path()?.into_owned();
            let mut contents = Vec::new();
            entry.read_to_end(&mut contents)?;
            out.push((path, contents));
        }
        out.sort_by(|x, y| x.0.cmp(&y.0));
        Ok(out)
    }
    Ok(extract(a)? == extract(b)?)
}

#[cfg(test)]
mod tests {
    use super::*;

    fn write_file(root: &Path, relative: &str, bytes: &[u8]) {
        let path = root.join(relative);
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent).unwrap();
        }
        let mut f = std::fs::File::create(&path).unwrap();
        f.write_all(bytes).unwrap();
    }

    fn baseline_inputs() -> FingerprintInputs {
        FingerprintInputs {
            cargo_lock: "[[package]]\nname = \"foo\"\nversion = \"0.1.0\"\n".into(),
            rustc_version_verbose:
                "rustc 1.75.0\nhost: aarch64-apple-darwin\nrelease: 1.75.0\n".into(),
            cargo_config: String::new(),
            rust_toolchain: String::new(),
            profile: "dev".into(),
            features: vec!["a".into(), "b".into()],
            rustflags: String::new(),
            target_triple: "aarch64-apple-darwin".into(),
        }
    }

    #[test]
    fn fingerprint_is_deterministic() {
        let a = baseline_inputs().compute();
        let b = baseline_inputs().compute();
        assert_eq!(a, b);
    }

    #[test]
    fn fingerprint_changes_when_cargo_lock_changes() {
        let base = baseline_inputs().compute();
        let mut mutated = baseline_inputs();
        mutated.cargo_lock.push_str("# extra line\n");
        assert_ne!(base, mutated.compute());
    }

    #[test]
    fn fingerprint_changes_when_profile_changes() {
        let base = baseline_inputs().compute();
        let mut mutated = baseline_inputs();
        mutated.profile = "release".into();
        assert_ne!(base, mutated.compute());
    }

    #[test]
    fn fingerprint_changes_when_features_change() {
        let base = baseline_inputs().compute();
        let mut mutated = baseline_inputs();
        mutated.features.push("c".into());
        assert_ne!(base, mutated.compute());
    }

    #[test]
    fn fingerprint_is_feature_order_independent() {
        // Collect() sorts features. Test that two orderings produce
        // the same fingerprint after collection semantics.
        let mut a = baseline_inputs();
        a.features = vec!["b".into(), "a".into()];
        a.features.sort();
        let mut b = baseline_inputs();
        b.features = vec!["a".into(), "b".into()];
        b.features.sort();
        assert_eq!(a.compute(), b.compute());
    }

    #[test]
    fn fingerprint_domain_separation_prevents_field_collision() {
        // If two fields' contents happened to concat identically, a
        // naive hasher would produce the same hash. Ours prefixes each
        // field with a label + null separator, so this doesn't happen.
        let mut a = baseline_inputs();
        a.cargo_lock = "shared".into();
        a.rustflags = "".into();
        let mut b = baseline_inputs();
        b.cargo_lock = "".into();
        b.rustflags = "shared".into();
        assert_ne!(a.compute(), b.compute());
    }

    #[test]
    fn fingerprint_hex_length_and_stability() {
        let fp = baseline_inputs().compute();
        assert_eq!(fp.to_hex().len(), 64);
        // Round-trip the raw bytes.
        let raw = *fp.as_bytes();
        let recomputed = Fingerprint::from_bytes(raw);
        assert_eq!(recomputed, fp);
        assert_eq!(recomputed.to_hex(), fp.to_hex());
    }

    #[test]
    fn read_optional_returns_empty_for_missing() {
        let tmp = tempfile::TempDir::new().unwrap();
        let out = read_optional(&tmp.path().join("does-not-exist")).unwrap();
        assert!(out.is_empty());
    }

    #[test]
    fn read_optional_returns_content_for_existing() {
        let tmp = tempfile::TempDir::new().unwrap();
        let p = tmp.path().join("f");
        std::fs::write(&p, b"hello").unwrap();
        let out = read_optional(&p).unwrap();
        assert_eq!(out, "hello");
    }

    #[test]
    fn collect_errors_when_cargo_lock_absent() {
        let tmp = tempfile::TempDir::new().unwrap();
        let err = FingerprintInputs::collect(tmp.path(), "dev", &[])
            .unwrap_err()
            .to_string();
        assert!(err.contains("Cargo.lock"), "unexpected error: {err}");
    }

    #[test]
    fn collect_reads_cargo_lock_and_computes() {
        let tmp = tempfile::TempDir::new().unwrap();
        write_file(tmp.path(), "Cargo.lock", b"[[package]]\nname = \"x\"\n");
        // rustc must be on PATH — this is the same env our test suite
        // runs under so it's a safe assumption.
        let inputs = FingerprintInputs::collect(tmp.path(), "dev", &["feat".into()]).unwrap();
        assert!(inputs.cargo_lock.contains("package"));
        assert!(!inputs.rustc_version_verbose.is_empty());
        assert!(inputs.target_triple != "unknown-triple");
        assert_eq!(inputs.profile, "dev");
        assert_eq!(inputs.features, vec!["feat".to_string()]);
        // And computing works.
        let _fp = inputs.compute();
    }

    #[test]
    fn capture_target_errors_when_dir_missing() {
        let tmp = tempfile::TempDir::new().unwrap();
        let err = capture_target(&tmp.path().join("nope"))
            .unwrap_err()
            .to_string();
        assert!(err.contains("does not exist"), "unexpected: {err}");
    }

    #[test]
    fn capture_and_restore_round_trip_preserves_files() {
        let tmp = tempfile::TempDir::new().unwrap();
        let target = tmp.path().join("target");
        write_file(&target, "deps/libfoo.rlib", b"fake rlib content");
        write_file(&target, "deps/libbar.rlib", b"another rlib");
        write_file(&target, ".fingerprint/foo/lib-foo", b"fingerprint blob");
        write_file(&target, "build/foo-abc/output", b"build script output");
        write_file(&target, ".rustc_info.json", b"{\"stub\":\"info\"}");
        // Not captured — should be absent post-restore.
        write_file(&target, "incremental/foo/session", b"incremental data");
        write_file(&target, "examples/demo", b"demo binary");

        let bytes = capture_target(&target).unwrap();
        assert!(!bytes.is_empty());
        assert!(
            bytes.len() < CAPTURE_INLINE_CEILING,
            "test capture stayed inline"
        );

        let dst = tmp.path().join("restored");
        restore_target(&bytes, &dst).unwrap();

        // Captured files are present + equal.
        assert_eq!(
            std::fs::read(dst.join("deps/libfoo.rlib")).unwrap(),
            b"fake rlib content"
        );
        assert_eq!(
            std::fs::read(dst.join("deps/libbar.rlib")).unwrap(),
            b"another rlib"
        );
        assert_eq!(
            std::fs::read(dst.join(".fingerprint/foo/lib-foo")).unwrap(),
            b"fingerprint blob"
        );
        assert_eq!(
            std::fs::read(dst.join("build/foo-abc/output")).unwrap(),
            b"build script output"
        );
        assert_eq!(
            std::fs::read(dst.join(".rustc_info.json")).unwrap(),
            b"{\"stub\":\"info\"}"
        );
        assert_eq!(std::fs::read(dst.join("examples/demo")).unwrap(), b"demo binary");

        // `incremental/` is intentionally NOT captured.
        assert!(
            !dst.join("incremental").exists(),
            "incremental/ must be excluded — it's not portable"
        );
    }

    #[test]
    fn capture_is_order_independent_of_filesystem_readdir() {
        // Field finding 2026-07-12: `append_dir_all` used
        // `read_dir`'s native order, which differs across filesystems.
        // Two byte-identical trees produced different tars whose only
        // difference was entry order. Guard: create two trees whose
        // files are the same but written in DIFFERENT orders (which
        // biases readdir on many FS layouts), and require the
        // captures to match. `append_dir_sorted` — the new walker —
        // sorts by filename so the order at capture time is fixed.
        let tmp = tempfile::TempDir::new().unwrap();
        let a = tmp.path().join("A");
        let b = tmp.path().join("B");
        // Tree A: write a, b, c
        write_file(&a, "deps/aaa.rlib", b"aaa content");
        write_file(&a, "deps/bbb.rlib", b"bbb content");
        write_file(&a, "deps/ccc.rlib", b"ccc content");
        write_file(&a, ".fingerprint/aaa/xxx", b"fp-aaa");
        write_file(&a, ".fingerprint/bbb/xxx", b"fp-bbb");
        // Tree B: same files, reverse creation order
        write_file(&b, "deps/ccc.rlib", b"ccc content");
        write_file(&b, "deps/bbb.rlib", b"bbb content");
        write_file(&b, "deps/aaa.rlib", b"aaa content");
        write_file(&b, ".fingerprint/bbb/xxx", b"fp-bbb");
        write_file(&b, ".fingerprint/aaa/xxx", b"fp-aaa");

        let cap_a = capture_target(&a).unwrap();
        let cap_b = capture_target(&b).unwrap();
        assert_eq!(
            cap_a, cap_b,
            "captures must be byte-identical after sorted walk"
        );
    }

    #[test]
    fn capture_yields_identical_bytes_for_identical_input() {
        // With HeaderMode::Deterministic on the tar builder, two
        // captures of identical trees produce byte-identical tarballs.
        // That's what lets the same target dir hash to the same BlobId
        // across builds.
        let tmp = tempfile::TempDir::new().unwrap();
        let t1 = tmp.path().join("t1");
        let t2 = tmp.path().join("t2");
        write_file(&t1, "deps/a.rlib", b"content");
        write_file(&t1, ".fingerprint/x", b"fp");
        write_file(&t2, "deps/a.rlib", b"content");
        write_file(&t2, ".fingerprint/x", b"fp");

        let a = capture_target(&t1).unwrap();
        let b = capture_target(&t2).unwrap();
        assert!(zstd_tar_contents_equal(&a, &b).unwrap());
    }

    #[test]
    fn capture_streaming_matches_buffered_and_restores_correctly() {
        // Field finding 2026-07-12: the buffered `capture_target` used
        // 2.8 GB peak RAM on clawverse. `capture_target_to_writer`
        // streams into a caller-provided Writer. This guards two
        // properties: (1) the streamed bytes match the buffered
        // variant exactly, (2) the reported byte count agrees, and
        // (3) restore roundtrip works from a file-backed writer.
        let src = tempfile::TempDir::new().unwrap();
        let target = src.path().join("target");
        write_file(&target, "deps/a.rlib", b"aaaaaaaaaaa");
        write_file(&target, "deps/b.rlib", b"bbbbbbbbbbb");
        write_file(&target, ".fingerprint/aa/xxx", b"aa-fp");
        write_file(&target, ".fingerprint/bb/xxx", b"bb-fp");
        write_file(&target, "build/cc/cc.rlib", b"cc");

        let buffered = capture_target(&target).unwrap();

        let out_dir = tempfile::TempDir::new().unwrap();
        let out_path = out_dir.path().join("capture.tar.zst");
        let file = std::fs::File::create(&out_path).unwrap();
        let mut writer = std::io::BufWriter::new(file);
        let reported = capture_target_to_writer(&target, &mut writer).unwrap();
        use std::io::Write;
        writer.flush().unwrap();
        let streamed = std::fs::read(&out_path).unwrap();

        assert_eq!(
            streamed, buffered,
            "streaming capture must match buffered capture byte-for-byte"
        );
        assert_eq!(
            reported as usize,
            buffered.len(),
            "reported byte count must equal actual bytes written"
        );

        // Roundtrip: restore from the streamed file, verify contents.
        let restored = tempfile::TempDir::new().unwrap();
        restore_target(&streamed, restored.path()).unwrap();
        assert_eq!(
            std::fs::read(restored.path().join("deps/a.rlib")).unwrap(),
            b"aaaaaaaaaaa"
        );
        assert_eq!(
            std::fs::read(restored.path().join(".fingerprint/bb/xxx")).unwrap(),
            b"bb-fp"
        );
    }

    #[test]
    fn capture_skips_top_level_files_not_in_allowlist() {
        let tmp = tempfile::TempDir::new().unwrap();
        let target = tmp.path().join("target");
        write_file(&target, "deps/keep.rlib", b"keep");
        // random top-level file NOT in CAPTURED_TOP_FILES
        write_file(&target, "random.txt", b"drop me");

        let bytes = capture_target(&target).unwrap();
        let dst = tmp.path().join("restored");
        restore_target(&bytes, &dst).unwrap();

        assert!(dst.join("deps/keep.rlib").exists());
        assert!(
            !dst.join("random.txt").exists(),
            "random top-level file should not be captured"
        );
    }

    #[test]
    fn capture_workspace_resolves_profile_dir() {
        let tmp = tempfile::TempDir::new().unwrap();
        let workspace = tmp.path();
        let target_release = workspace.join("target").join("release");
        write_file(&target_release, "deps/x.rlib", b"released");

        let bytes = capture_workspace(workspace, "release").unwrap();
        let dst = tmp.path().join("restored");
        restore_target(&bytes, &dst).unwrap();
        assert_eq!(std::fs::read(dst.join("deps/x.rlib")).unwrap(), b"released");
    }

    #[test]
    fn compute_workspace_fingerprint_returns_inputs_and_hash() {
        let tmp = tempfile::TempDir::new().unwrap();
        write_file(tmp.path(), "Cargo.lock", b"[[package]]\nname=\"z\"\n");
        let (inputs, fp) =
            compute_workspace_fingerprint(tmp.path(), "release", &["one".into()]).unwrap();
        assert_eq!(inputs.profile, "release");
        assert_eq!(inputs.features, vec!["one".to_string()]);
        // Recomputing must match.
        assert_eq!(fp, inputs.compute());
    }

    // ── Full pipeline: fingerprint → capture → blob → restore ────────

    #[tokio::test]
    async fn end_to_end_fingerprint_capture_blob_restore() {
        // Given the same workspace state on two machines, the
        // fingerprint is identical → BlobId is deterministic once the
        // captured bytes are stored → the second machine can find the
        // cache by fingerprint alone. This is the full loop, sans RPC.
        use crate::cluster::blob::{BlobId, BlobStore};

        let tmp = tempfile::TempDir::new().unwrap();
        let workspace = tmp.path().join("ws");
        write_file(&workspace, "Cargo.lock", b"[[package]]\nname=\"x\"\n");

        // Fake target dir populated with a couple of "deps".
        let target = workspace.join("target").join("dev");
        write_file(&target, "deps/foo.rlib", b"1234567890" .repeat(50).as_slice());
        write_file(&target, "deps/bar.rlib", b"abcdefghij" .repeat(50).as_slice());
        write_file(&target, ".fingerprint/foo/lib", b"cargo state");

        let (_inputs, fp) =
            compute_workspace_fingerprint(&workspace, "dev", &[]).unwrap();

        let bytes = capture_target(&target).unwrap();

        // Store via BlobStore — the BlobId is content-addressed and
        // independent of the fingerprint. Both sides need to end up
        // with the same bytes, which they do because capture is
        // deterministic.
        let store_dir = tmp.path().join("blobs");
        let store = BlobStore::open(store_dir).unwrap();
        let blob_id = store.put_bytes(&bytes).await.unwrap();

        // A machine that knows only the fingerprint would look up
        // `fingerprint → BlobId` via a metadata layer (Phase 3). For
        // this spike, we assume that mapping via an out-of-band step
        // — the test does it explicitly:
        let claimed_bid = BlobId::from_bytes(blake3::hash(&bytes).into());
        assert_eq!(blob_id, claimed_bid);

        // Now: peer B fetches by BlobId and restores.
        let round = store.get_bytes(&blob_id).await.unwrap().unwrap();
        assert_eq!(round, bytes);

        let restored_target = tmp.path().join("restored/target/dev");
        restore_target(&round, &restored_target).unwrap();

        // Every captured file lands intact.
        assert_eq!(
            std::fs::read(restored_target.join("deps/foo.rlib")).unwrap(),
            "1234567890".repeat(50).as_bytes()
        );
        assert_eq!(
            std::fs::read(restored_target.join(".fingerprint/foo/lib")).unwrap(),
            b"cargo state"
        );

        // Same fingerprint is what a builder on another host would
        // compute from the same workspace. So this end-to-end proves:
        //   workspace → fingerprint  (deterministic)
        //   target → bytes → BlobId  (deterministic)
        //   BlobId → bytes → target  (round-trip)
        let _ = fp; // usage-only assertion above
    }
}