fdu-core 0.3.0

//! Persisting an index to disk and reading it back.
//!
//! # Status: format v5 is a bounded bootstrap format
//!
//! This module implements a flat, uncompressed writer and a bounded streaming reader.
//! It exists so the cache *lifecycle* — semantic-scope invalidation, atomic replacement,
//! complete-only persistence, resource limits, and corrupt-equals-empty — is nailed down
//! before the optimized wire format is designed. The loader checks file size, trailer,
//! and payload checksum before parsing, then checks the header, record count, and path
//! lengths before allocating record data and rebuilds one entry at a time.
//!
//! The target format is different in every other respect: zstd-compressed blocks with an
//! index block at the tail, so opening costs one small read and directory listings
//! decompress on demand into a small LRU cache; item references encoded as
//! `(block << k) | offset` and delta-encoded when they point inside the same block;
//! sibling groups written contiguously so one directory listing costs one block
//! decompression; front-coded names; and pre-computed roll-ups stored per directory so a
//! query never re-aggregates. O(1) open with lazy materialization matters more than raw
//! decode throughput, because it matches how a UI actually navigates: open now, expand
//! later.
//!
//! Two things here are **not** placeholders and must survive the format change:
//!
//! - **A corrupt or unrecognized snapshot is treated as absent, never as an error and
//!   never as data.** A cache that fails closed costs a rescan; a cache that fails open
//!   silently lies.
//! - **Replacement is atomic.** Write a temporary file, then rename over the target, so a
//!   crash mid-write leaves the previous snapshot intact rather than a half-written one.

use std::ffi::{OsStr, OsString};
use std::fs::{self, OpenOptions};
use std::io::{self, BufReader, Read, Seek, SeekFrom, Write};
use std::path::{Component, Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};

use crate::engine_contract::{Attrs, EntryKind, Error, Observation, Op, Result, Source};
use crate::index::{EntryId, Index, IndexHandle};
use crate::stored_state::{
    ControlTierIdentity, SNAPSHOT_IDENTITY_BYTES, Serves, SnapshotIdentity, serves_snapshot,
};

/// Result of loading a snapshot for one requested stored-state identity.
#[derive(Debug)]
#[allow(clippy::large_enum_variant)] // One transient load result; avoid boxing the returned index.
pub enum LoadOutcome {
    /// The snapshot supplied an index, either exactly or through a lawful projection.
    Served {
        /// The index in the requested scope.
        index: Index,
        /// The identity the snapshot declares, which is what a plan admits: a projected
        /// index reports the requested identity, not this one.
        stored: SnapshotIdentity,
    },
    /// The snapshot was valid, but its identity cannot answer this request.
    Refused {
        /// The validated identity that did not serve the request.
        identity: SnapshotIdentity,
        /// The validated snapshot root, used before explaining a scope mismatch.
        root: PathBuf,
    },
    /// No valid snapshot was present.
    Absent,
}

/// Leading magic. Distinguishes an fdu snapshot from any other file that lands here.
const MAGIC: &[u8; 8] = b"FDUSNAP\x00";

/// Trailing magic. Present only once the whole snapshot has been written, so a truncated
/// file is detected on load rather than parsed into a plausible-looking partial tree.
const TRAILER: &[u8; 8] = b"FDUEND\x00\x00";

/// CRC-32C of every byte before the footer. This detects plausible payload corruption
/// that remains structurally valid; it is an integrity check, not authentication.
const CHECKSUM_BYTES: usize = std::mem::size_of::<u32>();

/// Reversed Castagnoli polynomial used by CRC-32C.
const CRC32C_POLYNOMIAL: u32 = 0x82f6_3b78;

/// One table lookup per byte keeps integrity validation from dominating snapshot load.
const CRC32C_TABLES: [[u32; 256]; 8] = make_crc32c_tables();

/// On-disk format version. Bump on any layout change; old snapshots are then discarded
/// rather than misread.
///
/// 4: the control section also carries both control limits, the budget and the line
/// limit, and every refused control file, so a snapshot of an index that refused some
/// reloads with the same coverage rather than claiming every rule applied.
///
/// 5: the header records the identity of each tier the snapshot holds, in the
/// stored-state model's fixed-width encodings: the entry tier's scope, type rules, and
/// reducer set, and the control tier's observation and limits. Observation and limits are
/// no longer mixed into the scope as one hash, and the limits moved out of the control
/// section, so a section is re-admitted under the header's limits and one they would not
/// admit is refused. The header also records `writing_pass_started_at_ns`, the start of the
/// pass that last wrote the image.
///
/// 6: the entry tier records ignored population and, when narrowed, the governing
/// control policy. Narrowed scans cannot reuse broader metadata without projection.
const FORMAT_VERSION: u32 = 6;

/// Version of the rules that decide which bucket an entry's bytes are tallied under.
///
/// Separate from [`FORMAT_VERSION`] because the two answer different questions: the
/// layout can be unchanged while the meaning of what was written moves. A snapshot stores
/// the bucket an entry was assigned, not the name it was assigned from, so a rule change
/// cannot be re-derived on load — the entries have to be discarded and re-walked.
///
/// 1: initial rules. 2: files with no extension are tallied under `(none)` instead of
/// being left out of the extension roll-up entirely.
const CLASSIFICATION_VERSION: u32 = 2;

/// Version of the per-entry facts used to decide whether retained state is still valid.
///
/// Version 2 adds Windows change time, volume identity, and file index. A pre-v2 Windows
/// snapshot stores zeroes in those fields and cannot safely serve cache-only as if those
/// facts had been observed. This is mixed into the engine fingerprint on every platform
/// so one build has one store identity.
const VALIDITY_VERSION: u32 = 2;

/// Snapshot path encoding used by Unix targets.
#[cfg(unix)]
const PATH_ENCODING_UNIX_BYTES: u8 = 1;

/// Snapshot path encoding used by Windows targets.
#[cfg(windows)]
const PATH_ENCODING_WINDOWS_WIDE: u8 = 2;

/// Portable UTF-8 fallback for targets outside Unix and Windows.
#[cfg(not(any(unix, windows)))]
const PATH_ENCODING_UTF8: u8 = 3;

/// Marks the root's absent parent.
const NO_PARENT: u32 = u32::MAX;

/// Byte offset of `writing_pass_started_at_ns`: after the magic, the format version, the
/// engine fingerprint, and the path encoding.
///
/// Fixed, so a save can read the stamp of the image already at its path without parsing
/// the rest of it.
const WRITING_PASS_STARTED_AT_OFFSET: usize = MAGIC.len() + 4 + 8 + 1;

/// Byte offset of the tier identities, which follow the stamp.
const IDENTITY_OFFSET: usize = WRITING_PASS_STARTED_AT_OFFSET + 8;

/// Byte offset of the root path, which follows the tier identities.
#[cfg(test)]
const ROOT_OFFSET: usize = IDENTITY_OFFSET + SNAPSHOT_IDENTITY_BYTES;

/// Smallest possible on-disk record: parent slot, kind, name length, and six 8-byte
/// attribute fields, with a zero-length name. Used to sanity-check a declared entry
/// count against the bytes actually present.
const MIN_RECORD_BYTES: usize = 4 + 1 + 4 + 8 * 6;

/// Binary size unit used by the snapshot resource limits.
const GIBIBYTE: u64 = 1024 * 1024 * 1024;

/// Largest snapshot image accepted by the bootstrap streaming reader.
const MAX_SNAPSHOT_BYTES: u64 = 64 * GIBIBYTE;

/// Process-local discriminator for exclusive sibling temporary files.
static NEXT_TEMP_FILE: AtomicU64 = AtomicU64::new(0);

/// Per-process entropy mixed into temporary file names.
///
/// Correctness never depended on this: `O_EXCL` is what guarantees one creator wins,
/// and the counter above is what keeps two threads of this process from even trying the
/// same name. Randomness closes two gaps the counter cannot.
///
/// A killed writer leaves its temporary behind and nothing reaps it. With a
/// pid-and-counter name, a later process that recycles that pid restarts its counter at
/// zero and collides with the corpse on every run — correct, because it retries, but it
/// accumulates litter and in the pathological case walks the whole retry budget. And
/// the names are otherwise entirely predictable, which matters because
/// [`MAX_TEMP_CREATE_ATTEMPTS`] exists specifically to survive a hostile directory: an
/// attacker who can guess the names can pre-create the whole budget and deny every save.
///
/// Seeded from `RandomState`, whose keys the standard library takes from the operating
/// system, so this needs no dependency and no clock.
static TEMP_FILE_ENTROPY: std::sync::LazyLock<u64> = std::sync::LazyLock::new(|| {
    use std::hash::{BuildHasher, Hasher};
    std::collections::hash_map::RandomState::new().build_hasher().finish()
});

/// Stale files can survive a killed writer. Try enough unique sequence values to step
/// over them without turning an attacker-controlled directory into an unbounded loop.
const MAX_TEMP_CREATE_ATTEMPTS: usize = 1024;

/// How old an abandoned temporary must be before a later writer removes it.
///
/// Per-process entropy in the name means a corpse can never collide with a future
/// writer — which also means no future writer will ever reuse or overwrite it. Unique
/// names turn an occasional collision into permanent litter, and each corpse is a whole
/// snapshot image, so something has to collect them.
///
/// A day is far longer than any real save and far shorter than "never". The threshold
/// is what makes this safe without any liveness check: a temporary this old cannot
/// belong to a writer that is still running, and pid-based liveness tests are both
/// unportable and wrong under pid reuse.
///
/// Shared with the cache lifecycle, which reclaims the same corpses on request rather
/// than waiting for the next writer, and must answer "old enough to be nobody's" the same
/// way this does.
pub(crate) const STALE_TEMP_AGE: std::time::Duration = std::time::Duration::from_secs(24 * 60 * 60);

/// Largest encoded root or entry name accepted from a snapshot.
const MAX_PATH_BYTES: u32 = 1024 * 1024;

/// Upper bound on records accepted even when a sparse file could physically hold more.
const MAX_SNAPSHOT_ENTRIES: u64 = 100_000_000;

/// Binary size unit for the control-section ceiling.
const MEBIBYTE: usize = 1024 * 1024;

/// Largest control table, in retained charge and in total source bytes, a snapshot may
/// carry.
///
/// A parser guard over untrusted lengths, deliberately independent of both control limits:
/// they are a caller's runtime settings and may be unbounded, while this bounds what a
/// corrupt or hostile file can make the loader allocate. [`save`] refuses a table above
/// it, so a snapshot the loader would reject is never written as if it were usable.
const SNAPSHOT_CONTROL_TABLE_CEILING: usize = 256 * MEBIBYTE;

/// Encoded refusal reasons.
const REFUSED_FOR_BUDGET: u8 = 1;
const REFUSED_FOR_LINE_LIMIT: u8 = 2;

/// A fingerprint of everything that would change how the engine interprets a tree.
///
/// When this does not match, the whole snapshot is discarded. That is the cheap,
/// wholesale answer to "the rules changed, so every derived verdict in the cache might
/// be wrong" — far simpler than trying to work out which entries a rule change affected.
///
/// Currently derived from the crate version, the format version, and the versions of the
/// classification rules, of the per-entry validity facts, and of the fixed `.gitignore`
/// semantics. When compiled type-recognition rules and reducer registrations arrive, their
/// hashes belong here too.
pub fn engine_fingerprint() -> u64 {
    engine_fingerprint_under(crate::stored_state::IGNORE_RULES_VERSION)
}

/// [`engine_fingerprint`] as a build whose `.gitignore` semantics are
/// `ignore_rules_version` computes it.
///
/// The rules version is the one input a snapshot's tier identities cannot tell apart for
/// the default population, so a test forges a snapshot of another version through it.
fn engine_fingerprint_under(ignore_rules_version: u64) -> u64 {
    let mut hash = 0xcbf2_9ce4_8422_2325_u64;
    let mut mix = |bytes: &[u8]| {
        for byte in bytes {
            hash ^= u64::from(*byte);
            hash = hash.wrapping_mul(0x1000_0000_01b3);
        }
    };
    mix(env!("CARGO_PKG_VERSION").as_bytes());
    mix(&FORMAT_VERSION.to_le_bytes());
    mix(&CLASSIFICATION_VERSION.to_le_bytes());
    mix(&VALIDITY_VERSION.to_le_bytes());
    mix(&ignore_rules_version.to_le_bytes());
    hash
}

/// Write `index` to `path`, replacing any existing snapshot atomically.
pub fn save(index: &Index, path: &Path) -> Result<()> {
    debug_assert!(!index.is_folded(), "a folded index omits files, so it is never persisted");
    if !crate::stored_state::entries_writable(index) {
        return Err(Error::Snapshot(
            "refusing to persist an index that is stale, reconciling, or incomplete".into(),
        ));
    }
    // The loader refuses a table whose limits its scope does not claim, so writing one
    // would publish a snapshot that every later open discards.
    index.require_control_limits_in_scope(index.control_table().limits())?;
    let mut buf: Vec<u8> = Vec::new();
    buf.extend_from_slice(MAGIC);
    buf.extend_from_slice(&FORMAT_VERSION.to_le_bytes());
    buf.extend_from_slice(&engine_fingerprint().to_le_bytes());
    buf.push(path_encoding());
    debug_assert_eq!(buf.len(), WRITING_PASS_STARTED_AT_OFFSET);
    buf.extend_from_slice(&index.writing_pass_started_at_ns().to_le_bytes());
    // Every tier identity shares the prologue's engine fingerprint, which is this build's.
    buf.extend_from_slice(&index.snapshot_identity().encode());

    put_os_str(&mut buf, index.root_path().as_os_str())?;

    // Pre-order, so a parent's record always precedes its children's and the loader can
    // rebuild the tree in one forward pass with no fixups.
    let mut records: Vec<(u32, EntryId)> = Vec::new();
    let mut stack: Vec<(u32, EntryId)> = vec![(NO_PARENT, EntryId::ROOT)];
    while let Some((parent_slot, id)) = stack.pop() {
        let slot = u32::try_from(records.len())
            .map_err(|_| Error::Snapshot("snapshot exceeds u32 entry capacity".into()))?;
        records.push((parent_slot, id));
        let children = index
            .children_of(id)
            .ok_or_else(|| Error::Snapshot("stale entry handle while saving".into()))?;
        for (_, child) in children.rev() {
            stack.push((slot, child));
        }
    }

    let count = u64::try_from(records.len())
        .map_err(|_| Error::Snapshot("snapshot entry count overflow".into()))?;
    buf.extend_from_slice(&count.to_le_bytes());

    for (parent_slot, id) in records {
        buf.extend_from_slice(&parent_slot.to_le_bytes());
        let kind = index
            .kind_of(id)
            .ok_or_else(|| Error::Snapshot("stale entry handle while saving".into()))?;
        buf.push(kind as u8);
        let name = index
            .name_of(id)
            .ok_or_else(|| Error::Snapshot("stale entry handle while saving".into()))?;
        put_os_str(&mut buf, name)?;
        let attrs = index
            .attrs_of(id)
            .ok_or_else(|| Error::Snapshot("stale entry handle while saving".into()))?;
        buf.extend_from_slice(&attrs.size.to_le_bytes());
        buf.extend_from_slice(&attrs.allocated.to_le_bytes());
        buf.extend_from_slice(&attrs.mtime_ns.to_le_bytes());
        buf.extend_from_slice(&attrs.ctime_ns.to_le_bytes());
        buf.extend_from_slice(&attrs.inode.to_le_bytes());
        buf.extend_from_slice(&attrs.dev.to_le_bytes());
    }
    put_controls(&mut buf, index.control_table())?;
    publish(path, buf)
}

/// Seal a snapshot payload and write it to `path`, keeping an image already there that
/// encodes the same facts.
///
/// Two passes over an unchanged tree encode identical images except for each pass's
/// start, and rewriting for the stamp alone would give back the skip [`write_atomically`]
/// exists for. A default run never reads its snapshot for a metadata query, so on that path
/// the rewrite was the whole cost of having a cache: about 70 ms of a 375 ms run over 175k
/// entries (exp-066), and a 14 MB write and `F_FULLFSYNC` on every run (exp-067). So an image at `path` that differs only in its stamp is kept, stamp
/// included, and only its mtime moves. The kept stamp is the start of an earlier pass that
/// wrote exactly these facts, which can understate how recently they were verified and
/// never overstates it.
fn publish(path: &Path, mut payload: Vec<u8>) -> Result<()> {
    if keep_equivalent_image(path, &payload) {
        return Ok(());
    }
    seal(&mut payload);
    // Nothing at `path` is this image under any stamp, this pass's included, so comparing
    // again before replacing it could only repeat the answer.
    replace_atomically(path, &payload)
}

/// Keep the file at `path` when it is `payload` sealed under any pass's stamp, moving only
/// its mtime, to the stamp in `payload`.
///
/// The mtime is cache-maintenance metadata, not filesystem-observation provenance: a copy
/// or a touch must not change when the tree was observed. Moving it to the start of the pass
/// that would have stamped the image lets the equivalent-image fast path retain its useful
/// monotonic hint without changing the persisted observation stamp. It never moves backwards:
/// a save of an index loaded under an older stamp leaves a later mtime in place.
///
/// One read through one handle, comparing every payload byte before computing the
/// checksum, so a changed tree stops at its first difference and costs no checksum here,
/// and an unchanged one costs the one checksum sealing it would have. The checksum is
/// still verified before the file is kept: a file whose own checksum is wrong reads as
/// absent, and keeping it would keep every later run cold.
fn keep_equivalent_image(path: &Path, payload: &[u8]) -> bool {
    const FOOTER_BYTES: usize = CHECKSUM_BYTES + TRAILER.len();
    let Ok(mut file) = fs::File::open(path) else { return false };
    let Ok(metadata) = file.metadata() else { return false };
    let image_len = payload.len().checked_add(FOOTER_BYTES).and_then(|len| u64::try_from(len).ok());
    if image_len != Some(metadata.len()) {
        return false;
    }
    let (before_stamp, after_stamp) =
        (&payload[..WRITING_PASS_STARTED_AT_OFFSET], &payload[IDENTITY_OFFSET..]);
    let mut stamp = [0u8; 8];
    if !reads_as(&mut file, before_stamp)
        || file.read_exact(&mut stamp).is_err()
        || !reads_as(&mut file, after_stamp)
    {
        return false;
    }
    let checksum =
        ![before_stamp, &stamp[..], after_stamp].into_iter().fold(u32::MAX, crc32c_update);
    let mut footer = [0u8; FOOTER_BYTES];
    if file.read_exact(&mut footer).is_err()
        || footer[..CHECKSUM_BYTES] != checksum.to_le_bytes()
        || footer[CHECKSUM_BYTES..] != *TRAILER
        // A file that grew under the comparison is not this image.
        || !matches!(file.read(&mut [0u8; 1]), Ok(0))
    {
        return false;
    }
    let pass_started = payload
        .get(WRITING_PASS_STARTED_AT_OFFSET..IDENTITY_OFFSET)
        .and_then(|stamp| <[u8; 8]>::try_from(stamp).ok())
        .map(i64::from_le_bytes)
        .and_then(|nanos| u64::try_from(nanos).ok())
        .and_then(|nanos| {
            std::time::UNIX_EPOCH.checked_add(std::time::Duration::from_nanos(nanos))
        });
    if let Some(pass_started) = pass_started {
        if !matches!(metadata.modified(), Ok(modified) if modified >= pass_started) {
            // Best effort, as for any kept image: a stale mtime understates freshness.
            let _ = touch(path, pass_started);
        }
    }
    true
}

/// Append the checksum of every byte so far, then the trailer.
fn seal(payload: &mut Vec<u8>) {
    let checksum = crc32c(payload);
    payload.extend_from_slice(&checksum.to_le_bytes());
    payload.extend_from_slice(TRAILER);
}

/// Capture and persist a coherent shared-index image without holding its lock during
/// serialization or filesystem I/O.
pub fn save_handle(index: &IndexHandle, path: &Path) -> Result<()> {
    let snapshot = index.snapshot()?;
    save(&snapshot, path)
}

/// Load a snapshot, or return `None` when there is nothing usable at `path`.
///
/// Returns `None` — not an error — for a missing file, a foreign file, a version or
/// engine-fingerprint mismatch, and a truncated or corrupt file. Every one of those
/// means the same thing to a caller: there is no warm cache, so scan.
pub fn load(path: &Path) -> Result<Option<Index>> {
    load_with_size_limit(path, MAX_SNAPSHOT_BYTES)
}

/// Load a snapshot under the registry that will classify and analyze its entries.
///
/// A snapshot made under different rules is a clean cache miss. The snapshot carries
/// only their derived identity, so the caller must supply the matching registry rather
/// than allowing the loader to attach unrelated default rules.
pub fn load_with_types(
    path: &Path,
    types: std::sync::Arc<crate::classify::TypeRegistry>,
) -> Result<Option<Index>> {
    load_with_types_and_size_limit(path, MAX_SNAPSHOT_BYTES, types, None).map(|outcome| {
        match outcome {
            LoadOutcome::Served { index, .. } => Some(index),
            LoadOutcome::Refused { .. } | LoadOutcome::Absent => None,
        }
    })
}

/// Load a snapshot that can serve `wanted`, projecting observed control state away when
/// the entry tier is equal and the request does not observe controls.
pub fn load_serving(
    path: &Path,
    types: std::sync::Arc<crate::classify::TypeRegistry>,
    wanted: SnapshotIdentity,
) -> Result<LoadOutcome> {
    load_with_types_and_size_limit(path, MAX_SNAPSHOT_BYTES, types, Some(wanted))
}

fn load_with_size_limit(path: &Path, max_snapshot_bytes: u64) -> Result<Option<Index>> {
    load_with_types_and_size_limit(
        path,
        max_snapshot_bytes,
        crate::classify::TypeRegistry::compiled_shared(),
        None,
    )
    .map(|outcome| match outcome {
        LoadOutcome::Served { index, .. } => Some(index),
        LoadOutcome::Refused { .. } | LoadOutcome::Absent => None,
    })
}

fn load_with_types_and_size_limit(
    path: &Path,
    max_snapshot_bytes: u64,
    types: std::sync::Arc<crate::classify::TypeRegistry>,
    wanted: Option<SnapshotIdentity>,
) -> Result<LoadOutcome> {
    let mut file = match fs::File::open(path) {
        Ok(file) => file,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(LoadOutcome::Absent),
        Err(e) => return Err(Error::io(path, e)),
    };
    let file_len = file.metadata().map_err(|e| Error::io(path, e))?.len();
    let footer_bytes = CHECKSUM_BYTES
        .checked_add(TRAILER.len())
        .and_then(|bytes| u64::try_from(bytes).ok())
        .ok_or_else(|| Error::Snapshot("snapshot footer size overflow".into()))?;
    if file_len > max_snapshot_bytes || file_len < footer_bytes {
        return Ok(LoadOutcome::Absent);
    }

    let footer_offset = i64::try_from(footer_bytes)
        .map_err(|_| Error::Snapshot("snapshot footer size overflow".into()))?;
    file.seek(SeekFrom::End(-footer_offset)).map_err(|e| Error::io(path, e))?;
    let expected_checksum = match read_footer_checksum(&mut file) {
        Ok(checksum) => checksum,
        Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => {
            return Ok(LoadOutcome::Absent);
        }
        Err(error) => return Err(Error::io(path, error)),
    };
    let mut trailer = [0u8; TRAILER.len()];
    match file.read_exact(&mut trailer) {
        Ok(()) if &trailer == TRAILER => {}
        Ok(()) => return Ok(LoadOutcome::Absent),
        Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(LoadOutcome::Absent),
        Err(e) => return Err(Error::io(path, e)),
    }
    let payload_len = file_len
        .checked_sub(footer_bytes)
        .ok_or_else(|| Error::Snapshot("snapshot length underflow".into()))?;
    file.seek(SeekFrom::Start(0)).map_err(|e| Error::io(path, e))?;

    // One pass, not two: the checksum accumulates as the parser consumes, instead of
    // a separate full read of the image before parsing begins. The verdict still
    // gates the data — the parsed index is returned only after the digest over the
    // complete payload matches, so nothing is ever served from bytes that failed
    // their checksum. What changes is the failure mode for structurally-valid
    // corruption: the parser may do work before the mismatch is known, and the
    // result is then discarded. Structural corruption is caught by the parser's own
    // bounds and consistency checks exactly as before, fail-closed either way.
    let mut reader = Crc32cReader::new(BufReader::new(file.take(payload_len)));
    let outcome = parse_stream(&mut reader, payload_len, types, wanted);
    match outcome {
        Ok(outcome) => {
            // A successful parse consumed every payload byte (the trailing-byte check
            // proves it), so the running digest covers the whole image.
            if reader.finish() == expected_checksum { Ok(outcome) } else { Ok(LoadOutcome::Absent) }
        }
        Err(ParseError::Invalid) => Ok(LoadOutcome::Absent),
        Err(ParseError::Io(source)) => Err(Error::io(path, source)),
    }
}

/// A reader that folds CRC-32C over every byte the caller consumes.
struct Crc32cReader<R> {
    inner: R,
    state: u32,
}

impl<R: Read> Crc32cReader<R> {
    fn new(inner: R) -> Self {
        Self { inner, state: u32::MAX }
    }

    fn finish(&self) -> u32 {
        !self.state
    }
}

impl<R: Read> Read for Crc32cReader<R> {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        let read = self.inner.read(buf)?;
        self.state = crc32c_update(self.state, &buf[..read]);
        Ok(read)
    }
}

fn read_footer_checksum(reader: &mut impl Read) -> std::io::Result<u32> {
    let mut bytes = [0u8; CHECKSUM_BYTES];
    reader.read_exact(&mut bytes)?;
    Ok(u32::from_le_bytes(bytes))
}

pub(crate) fn crc32c(bytes: &[u8]) -> u32 {
    !crc32c_update(u32::MAX, bytes)
}

/// Slicing-by-8: fold eight input bytes per step through eight derived tables
/// instead of one byte through one. Table 0 is the classic byte table, so the
/// remainder loop and the 8-byte path share one source of truth; the digest is
/// bit-identical to the byte-at-a-time form, which a test asserts over uneven
/// lengths alongside the standard check value.
fn crc32c_update(mut state: u32, bytes: &[u8]) -> u32 {
    let mut chunks = bytes.chunks_exact(8);
    for chunk in &mut chunks {
        let low = (state ^ u32::from_le_bytes(chunk[..4].try_into().expect("chunk holds 8 bytes")))
            .to_le_bytes();
        let high =
            u32::from_le_bytes(chunk[4..].try_into().expect("chunk holds 8 bytes")).to_le_bytes();
        state = CRC32C_TABLES[7][usize::from(low[0])]
            ^ CRC32C_TABLES[6][usize::from(low[1])]
            ^ CRC32C_TABLES[5][usize::from(low[2])]
            ^ CRC32C_TABLES[4][usize::from(low[3])]
            ^ CRC32C_TABLES[3][usize::from(high[0])]
            ^ CRC32C_TABLES[2][usize::from(high[1])]
            ^ CRC32C_TABLES[1][usize::from(high[2])]
            ^ CRC32C_TABLES[0][usize::from(high[3])];
    }
    for byte in chunks.remainder() {
        let index = usize::from(state.to_le_bytes()[0] ^ *byte);
        state = CRC32C_TABLES[0][index] ^ (state >> 8);
    }
    state
}

const fn make_crc32c_tables() -> [[u32; 256]; 8] {
    let mut tables = [[0u32; 256]; 8];
    let mut index = 0usize;
    let mut value = 0u32;
    while index < 256 {
        let mut crc = value;
        let mut bit = 0;
        while bit < u8::BITS {
            crc = (crc >> 1) ^ (CRC32C_POLYNOMIAL & 0u32.wrapping_sub(crc & 1));
            bit += 1;
        }
        tables[0][index] = crc;
        index += 1;
        value += 1;
    }
    // tables[k] advances a byte's contribution k further positions: one more
    // byte-table step applied to the previous table's value.
    let mut table = 1usize;
    while table < tables.len() {
        let mut index = 0usize;
        while index < 256 {
            let previous = tables[table - 1][index];
            tables[table][index] = tables[0][(previous & 0xFF) as usize] ^ (previous >> 8);
            index += 1;
        }
        table += 1;
    }
    tables
}

/// Read only a snapshot's header, without materializing its index.
///
/// Returns `None` for anything this build cannot serve — absent, truncated, foreign,
/// a different format version, or a mismatched engine fingerprint. Corrupt equals
/// absent here exactly as it does on the load path: a caller asking what is in the cache
/// must not be stopped by one unreadable file.
pub fn read_header(path: &Path) -> Result<Option<crate::cache::SnapshotInfo>> {
    Ok(match identify(path)? {
        Some(Identity::Current(info)) => Some(info),
        Some(Identity::Stale(_) | Identity::Foreign) | None => None,
    })
}

/// What a file's leading bytes and trailer say it is.
#[derive(Debug)]
pub(crate) enum Identity {
    /// A snapshot this build reads.
    Current(crate::cache::SnapshotInfo),
    /// It begins with the snapshot magic, so fdu wrote it, but this build cannot serve it.
    Stale(crate::cache::StaleReason),
    /// It does not begin with the snapshot magic.
    Foreign,
}

/// Identify the file at `path` by its contents, or return `None` when nothing is there.
///
/// The magic alone decides whether a file is fdu's; the rest of the prologue decides
/// whether this build can serve it. Every format fdu has written puts the version and the
/// engine fingerprint at the same offsets after the magic, so a snapshot from another
/// release is recognised as stale rather than mistaken for a foreign file. That is what
/// lets the cache lifecycle reclaim the snapshots an upgrade leaves behind.
///
/// Opening follows a symbolic link at `path`, so a caller that must not follow one checks
/// the path's own metadata first.
pub(crate) fn identify(path: &Path) -> Result<Option<Identity>> {
    let file = match fs::File::open(path) {
        Ok(file) => file,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(error) => return Err(Error::io(path, error)),
    };
    let trailer_intact = has_intact_trailer(&file).map_err(|error| Error::io(path, error))?;
    // The trailer check left the cursor at the end; the header lives at the start.
    let mut file = file;
    file.seek(SeekFrom::Start(0)).map_err(|error| Error::io(path, error))?;
    identify_prologue(&mut BufReader::new(file), trailer_intact)
        .map(Some)
        .map_err(|error| Error::io(path, error))
}

/// Classify a file from its prologue. Only an I/O failure is an error; every malformed
/// byte is an answer.
fn identify_prologue(reader: &mut impl Read, trailer_intact: bool) -> io::Result<Identity> {
    use crate::cache::StaleReason;

    match read_array::<_, 8>(reader) {
        Ok(magic) if magic == *MAGIC => {}
        Ok(_) | Err(ParseError::Invalid) => return Ok(Identity::Foreign),
        Err(ParseError::Io(error)) => return Err(error),
    }
    let Some(version) = invalid_as_none(read_u32(reader))? else {
        return Ok(Identity::Stale(StaleReason::Unreadable));
    };
    match version.cmp(&FORMAT_VERSION) {
        std::cmp::Ordering::Less => {
            return Ok(Identity::Stale(StaleReason::OlderFormat { version }));
        }
        std::cmp::Ordering::Greater => {
            return Ok(Identity::Stale(StaleReason::NewerFormat { version }));
        }
        std::cmp::Ordering::Equal => {}
    }
    let engine = match invalid_as_none(read_u64(reader))? {
        Some(fingerprint) if fingerprint == engine_fingerprint() => fingerprint,
        Some(_) => return Ok(Identity::Stale(StaleReason::OtherEngine)),
        None => return Ok(Identity::Stale(StaleReason::Unreadable)),
    };
    if !trailer_intact {
        // Truncation removes the tail and leaves the prologue readable, so a header-only
        // check would call a half-written file current.
        return Ok(Identity::Stale(StaleReason::Unreadable));
    }
    Ok(match invalid_as_none(parse_header_fields(reader, engine))? {
        Some(header) => Identity::Current(crate::cache::SnapshotInfo {
            root: header.root,
            identity: header.identity,
            entries: header.entries,
        }),
        None => Identity::Stale(StaleReason::Unreadable),
    })
}

/// Separate a malformed value, which is an answer, from an I/O failure, which is not.
fn invalid_as_none<T>(result: ParseResult<T>) -> io::Result<Option<T>> {
    match result {
        Ok(value) => Ok(Some(value)),
        Err(ParseError::Invalid) => Ok(None),
        Err(ParseError::Io(error)) => Err(error),
    }
}

/// Whether a file ends with the snapshot trailer.
///
/// Cheap enough for a status listing — one seek and eight bytes — and it is exactly what
/// a truncated write destroys.
fn has_intact_trailer(file: &fs::File) -> io::Result<bool> {
    let footer_bytes = CHECKSUM_BYTES + TRAILER.len();
    let file_len = file.metadata()?.len();
    if file_len < u64::try_from(footer_bytes).unwrap_or(u64::MAX) {
        return Ok(false);
    }

    let mut handle = file;
    handle.seek(SeekFrom::End(-(i64::try_from(TRAILER.len()).unwrap_or(0))))?;
    let mut trailer = [0u8; TRAILER.len()];
    match handle.read_exact(&mut trailer) {
        Ok(()) => Ok(&trailer == TRAILER),
        Err(error) if error.kind() == io::ErrorKind::UnexpectedEof => Ok(false),
        Err(error) => Err(error),
    }
}

/// The header fields after a snapshot's prologue.
struct Header {
    /// The start of the pass that last wrote the image.
    ///
    /// A later pass that verifies the same facts keeps the image and this stamp rather
    /// than rewriting for the stamp alone, so the value is a lower bound on when the facts
    /// were last verified: it can predate many verifying passes and never postdates the one
    /// that wrote them. Until P1.4.4 stamps completed passes, reconciliation does not
    /// advance it either.
    writing_pass_started_at_ns: i64,
    /// The identity of every tier the snapshot holds.
    identity: SnapshotIdentity,
    /// The absolute root the snapshot describes.
    root: PathBuf,
    /// How many entry records follow.
    entries: u64,
}

/// Parse the header fields after the magic, format version, and `engine` fingerprint.
fn parse_header_fields(reader: &mut impl Read, engine: u64) -> ParseResult<Header> {
    if read_u8(reader)? != path_encoding() {
        return Err(ParseError::Invalid);
    }
    let writing_pass_started_at_ns = read_i64(reader)?;
    let identity =
        SnapshotIdentity::decode(engine, &read_array::<_, SNAPSHOT_IDENTITY_BYTES>(reader)?)
            .ok_or(ParseError::Invalid)?;
    let root = PathBuf::from(read_os_string(reader)?);
    let entries = read_u64(reader)?;
    if entries == 0 || entries > MAX_SNAPSHOT_ENTRIES {
        return Err(ParseError::Invalid);
    }
    Ok(Header { writing_pass_started_at_ns, identity, root, entries })
}

/// Parse a bounded payload. Records are applied one at a time so bootstrap paths are not
/// retained in a second full-tree allocation.
fn parse_stream(
    reader: &mut impl Read,
    payload_len: u64,
    types: std::sync::Arc<crate::classify::TypeRegistry>,
    wanted: Option<SnapshotIdentity>,
) -> ParseResult<LoadOutcome> {
    if read_array::<_, 8>(reader)? != *MAGIC {
        return Err(ParseError::Invalid);
    }
    let engine = engine_fingerprint();
    if read_u32(reader)? != FORMAT_VERSION || read_u64(reader)? != engine {
        return Err(ParseError::Invalid);
    }
    let Header { writing_pass_started_at_ns, identity, root: root_path, entries: count } =
        parse_header_fields(reader, engine)?;
    let serves = wanted.map_or(Serves::Exact, |wanted| serves_snapshot(identity, wanted));
    let mut scope = match (serves, wanted) {
        (Serves::ProjectControlsOff, Some(wanted)) => wanted.scan_scope(),
        _ => identity.scan_scope(),
    };
    if scope.type_rules_fingerprint != types.fingerprint() {
        if serves != Serves::Refuse {
            return Err(ParseError::Invalid);
        }
        // A foreign registry cannot be attached to a served index. For a refusal,
        // reconstruct only to validate every record and the checksum, then discard it.
        // The stored identity below remains unchanged for the caller's diagnostic.
        //
        // That costs a full index build for a diagnostic: every record is decoded and
        // inserted so the checksum can be verified over the bytes it covers, and the
        // result is dropped. It is paid only under a type-rules mismatch, which a
        // cache-only request reports and every other policy answers by scanning, so it
        // buys a truthful refusal (valid but foreign, rather than absent) at the price of
        // one load on a path that has no answer anyway.
        scope.type_rules_fingerprint = types.fingerprint();
    }
    let minimum_body = count
        .checked_mul(u64::try_from(MIN_RECORD_BYTES).map_err(|_| ParseError::Invalid)?)
        .ok_or(ParseError::Invalid)?;
    if minimum_body > payload_len {
        return Err(ParseError::Invalid);
    }

    let mut index = Index::new_with_scope_and_types(&root_path, scope, types);
    // The facts below were verified by the pass that wrote them, not by this load.
    index.set_writing_pass_started_at_ns(writing_pass_started_at_ns);
    // Everything this loader inserts describes the tree as the snapshot found it, not
    // as this process has seen it. Stamping the entries `Cached` is what lets a
    // consumer paint them immediately and label them honestly; without it a loaded
    // index claims to be fresh when nothing has been checked since the file was read.
    index.set_applying_source(Source::Cached, writing_pass_started_at_ns);
    // The record count is validated against the bytes actually present above, so it is
    // safe to size from: reserving here removes the geometric regrowth of a 450k-element
    // vector without letting a corrupt count drive the allocation.
    let mut ids: Vec<EntryId> = Vec::with_capacity(usize::try_from(count).unwrap_or(0));
    // Roll-ups are u64 totals and every record is merged into them as it lands, so an
    // image whose file sizes sum past u64 is corrupt however sound its checksum: it is
    // refused before the record that would carry a total out of range is inserted, and
    // the index built so far is dropped with the error (fdu-sqyk).
    let mut bytes_total = 0_u64;
    let mut allocated_total = 0_u64;
    for slot in 0..count {
        let parent_slot = read_u32(reader)?;
        let kind = EntryKind::from_u8(read_u8(reader)?).ok_or(ParseError::Invalid)?;
        let name = read_os_string(reader)?;
        let attrs = Attrs {
            size: read_u64(reader)?,
            allocated: read_u64(reader)?,
            mtime_ns: read_i64(reader)?,
            ctime_ns: read_i64(reader)?,
            inode: read_u64(reader)?,
            dev: read_u64(reader)?,
        };

        if parent_slot == NO_PARENT {
            if slot != 0 || kind != EntryKind::Dir || !name.is_empty() {
                return Err(ParseError::Invalid);
            }
            index
                .apply_baseline(&Observation::new(vec![Op::Upsert {
                    path: PathBuf::new(),
                    kind,
                    attrs,
                }]))
                .map_err(|_| ParseError::Invalid)?;
            ids.push(EntryId::ROOT);
            continue;
        }

        let parent = *ids
            .get(usize::try_from(parent_slot).map_err(|_| ParseError::Invalid)?)
            .ok_or(ParseError::Invalid)?;
        if !is_snapshot_name(&name) {
            return Err(ParseError::Invalid);
        }
        if kind == EntryKind::File {
            bytes_total = bytes_total.checked_add(attrs.size).ok_or(ParseError::Invalid)?;
            allocated_total =
                allocated_total.checked_add(attrs.allocated).ok_or(ParseError::Invalid)?;
        }
        // The parent's id is already in hand, so the record is inserted straight beneath
        // it. Resolving a path to rediscover that parent, and then searching the parent's
        // children to rediscover the id just created, were both work the format had
        // already answered. A snapshot naming the same path twice, or parenting an entry
        // to a non-directory, is corrupt, and `insert_loaded_child` fails closed on both.
        let id = index.insert_loaded_child(parent, name, kind, attrs).ok_or(ParseError::Invalid)?;
        ids.push(id);
    }

    let controls = read_controls(reader, identity.controls)?;
    if serves == Serves::Exact {
        index.install_controls(controls).map_err(|_| ParseError::Invalid)?;
    }

    let mut extra = [0u8; 1];
    if reader.read(&mut extra).map_err(ParseError::Io)? != 0 {
        return Err(ParseError::Invalid);
    }
    // Once, rather than once per record: the loaded tree is the process baseline, and
    // nothing it inserted was journalled.
    index.establish_baseline();
    // Anything applied after the load is this process checking what the snapshot
    // claimed, which is a revalidation rather than a first sighting.
    index.set_applying_source(Source::Revalidated, 0);
    // The image on disk is this index, so nothing is owed until a pass mutates it.
    index.set_persistence_owed(false);
    Ok(if serves == Serves::Refuse {
        LoadOutcome::Refused { identity, root: root_path }
    } else {
        LoadOutcome::Served { index, stored: identity }
    })
}

#[derive(Debug)]
enum ParseError {
    Invalid,
    Io(std::io::Error),
}

type ParseResult<T> = std::result::Result<T, ParseError>;

fn read_array<R: Read, const N: usize>(reader: &mut R) -> ParseResult<[u8; N]> {
    let mut bytes = [0u8; N];
    match reader.read_exact(&mut bytes) {
        Ok(()) => Ok(bytes),
        Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => Err(ParseError::Invalid),
        Err(error) => Err(ParseError::Io(error)),
    }
}

fn read_u8(reader: &mut impl Read) -> ParseResult<u8> {
    Ok(read_array::<_, 1>(reader)?[0])
}

fn read_u32(reader: &mut impl Read) -> ParseResult<u32> {
    Ok(u32::from_le_bytes(read_array(reader)?))
}

fn read_u64(reader: &mut impl Read) -> ParseResult<u64> {
    Ok(u64::from_le_bytes(read_array(reader)?))
}

fn read_i64(reader: &mut impl Read) -> ParseResult<i64> {
    Ok(i64::from_le_bytes(read_array(reader)?))
}

fn read_bytes(reader: &mut impl Read) -> ParseResult<Vec<u8>> {
    let len = read_u32(reader)?;
    if len > MAX_PATH_BYTES {
        return Err(ParseError::Invalid);
    }
    let mut bytes = vec![0u8; usize::try_from(len).map_err(|_| ParseError::Invalid)?];
    match reader.read_exact(&mut bytes) {
        Ok(()) => Ok(bytes),
        Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => Err(ParseError::Invalid),
        Err(error) => Err(ParseError::Io(error)),
    }
}

/// Read the control section, retained sources then refusals, for the control tier the
/// header records.
///
/// Every source is admitted again under the header's limits. The charge does not depend on
/// admission order, so a table a scan retained always fits; one that does not, a repeated
/// path, a refusal naming a retained or repeated path, a refusal by a limit the header
/// records as unbounded, or any source or refusal under a tier that observed nothing, was
/// not written by [`save`].
fn read_controls(
    reader: &mut impl Read,
    tier: ControlTierIdentity,
) -> ParseResult<crate::control::ControlTable> {
    let limits = match tier {
        ControlTierIdentity::Observed { limits } => limits,
        // Limits decide nothing when nothing was observed, and the section must be empty.
        // So a loaded blind table holds the default limits where the scanning index kept
        // the request's; no reader consults a blind table's limits, and a cache status or
        // projection that reports them must not show that as a difference.
        ControlTierIdentity::NotObserved => crate::control::ControlLimits::default(),
    };
    let control_count = read_u32(reader)?;
    if control_count != 0 && !tier.is_observed() {
        return Err(ParseError::Invalid);
    }
    if usize::try_from(control_count)
        .map_err(|_| ParseError::Invalid)?
        .saturating_mul(crate::control::CONTROL_SOURCE_OVERHEAD)
        > SNAPSHOT_CONTROL_TABLE_CEILING
    {
        return Err(ParseError::Invalid);
    }
    let mut sources = Vec::new();
    let mut source_bytes = 0_usize;
    for _ in 0..control_count {
        let path = PathBuf::from(read_os_string(reader)?);
        let source = read_control_bytes(reader)?;
        source_bytes = source_bytes.saturating_add(source.len());
        if source_bytes > SNAPSHOT_CONTROL_TABLE_CEILING {
            return Err(ParseError::Invalid);
        }
        sources.push((path, source));
    }
    let mut controls = crate::control::ControlTable::with_limits(limits);
    for (path, source) in sources {
        let admission = controls.upsert(&path, source).map_err(|_| ParseError::Invalid)?;
        if admission != (crate::control::ControlAdmission::Retained { changed: true }) {
            return Err(ParseError::Invalid);
        }
    }
    if controls.retained_cost() > SNAPSHOT_CONTROL_TABLE_CEILING {
        return Err(ParseError::Invalid);
    }
    let refused = read_u32(reader)?;
    if refused != 0 && !tier.is_observed() {
        return Err(ParseError::Invalid);
    }
    for _ in 0..refused {
        let path = PathBuf::from(read_os_string(reader)?);
        let reason = match read_u8(reader)? {
            REFUSED_FOR_BUDGET => crate::control::ControlRefusalReason::Budget,
            REFUSED_FOR_LINE_LIMIT => crate::control::ControlRefusalReason::LineLimit,
            _ => return Err(ParseError::Invalid),
        };
        // An unbounded limit refuses nothing, so no save records a refusal by one.
        if !crate::control::is_control_file(&path)
            || controls.contains(&path)
            || limits.limit_for(reason).is_none()
        {
            return Err(ParseError::Invalid);
        }
        controls.record_refusal(&path, reason).map_err(|_| ParseError::Invalid)?;
    }
    Ok(controls)
}

fn read_control_bytes(reader: &mut impl Read) -> ParseResult<Vec<u8>> {
    let len = read_u32(reader)?;
    if usize::try_from(len).map_err(|_| ParseError::Invalid)? > SNAPSHOT_CONTROL_TABLE_CEILING {
        return Err(ParseError::Invalid);
    }
    let mut bytes = vec![0u8; usize::try_from(len).map_err(|_| ParseError::Invalid)?];
    match reader.read_exact(&mut bytes) {
        Ok(()) => Ok(bytes),
        Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => Err(ParseError::Invalid),
        Err(error) => Err(ParseError::Io(error)),
    }
}

#[cfg(unix)]
fn read_os_string(reader: &mut impl Read) -> ParseResult<OsString> {
    Ok(os_string_from_bytes(&read_bytes(reader)?))
}

#[cfg(not(unix))]
fn read_os_string(reader: &mut impl Read) -> ParseResult<OsString> {
    os_string_from_bytes(&read_bytes(reader)?).ok_or(ParseError::Invalid)
}

fn put_bytes(buf: &mut Vec<u8>, bytes: &[u8]) -> Result<()> {
    let len = u32::try_from(bytes.len())
        .map_err(|_| Error::Snapshot("string too long for snapshot".into()))?;
    if len > MAX_PATH_BYTES {
        return Err(Error::Snapshot("path exceeds snapshot limit".into()));
    }
    buf.extend_from_slice(&len.to_le_bytes());
    buf.extend_from_slice(bytes);
    Ok(())
}

/// Write the control section [`read_controls`] reads: sources, then refusals. The limits
/// they were admitted under are the header's control tier identity.
fn put_controls(buf: &mut Vec<u8>, controls: &crate::control::ControlTable) -> Result<()> {
    if controls.retained_cost() > SNAPSHOT_CONTROL_TABLE_CEILING
        || controls.source_bytes() > SNAPSHOT_CONTROL_TABLE_CEILING
    {
        return Err(Error::Snapshot(format!(
            "the control table retains {} bytes of charge, above the {} bytes a snapshot can \
             carry; set a control budget below it, or open without a cache",
            controls.retained_cost(),
            SNAPSHOT_CONTROL_TABLE_CEILING
        )));
    }
    let sources: Vec<_> = controls.sources().collect();
    let control_count = u32::try_from(sources.len())
        .map_err(|_| Error::Snapshot("control table exceeds u32 capacity".into()))?;
    buf.extend_from_slice(&control_count.to_le_bytes());
    for (path, source) in sources {
        put_os_str(buf, path.as_os_str())?;
        let len = u32::try_from(source.len())
            .map_err(|_| Error::Snapshot("control source exceeds u32 capacity".into()))?;
        buf.extend_from_slice(&len.to_le_bytes());
        buf.extend_from_slice(source);
    }
    let refused = u32::try_from(controls.refused_len())
        .map_err(|_| Error::Snapshot("refused controls exceed u32 capacity".into()))?;
    buf.extend_from_slice(&refused.to_le_bytes());
    for refusal in controls.refusals() {
        put_os_str(buf, refusal.path.as_os_str())?;
        buf.push(match refusal.reason {
            crate::control::ControlRefusalReason::Budget => REFUSED_FOR_BUDGET,
            crate::control::ControlRefusalReason::LineLimit => REFUSED_FOR_LINE_LIMIT,
        });
    }
    Ok(())
}

/// A snapshot entry name is one filesystem component, spelled exactly as stored.
///
/// `Path::components` treats `a` and `a/` as the same `Normal("a")`. The child map
/// distinguishes the raw names, but a `PathBuf`-keyed restore map merges them and
/// shrinks the cache-only completeness denominator. The raw name must equal that
/// single normal component so a checksummed alias cannot load.
fn is_snapshot_name(name: &OsStr) -> bool {
    let mut components = Path::new(name).components();
    let Some(Component::Normal(component)) = components.next() else { return false };
    components.next().is_none() && component == name
}

#[cfg(unix)]
pub(crate) fn path_encoding() -> u8 {
    PATH_ENCODING_UNIX_BYTES
}

#[cfg(windows)]
pub(crate) fn path_encoding() -> u8 {
    PATH_ENCODING_WINDOWS_WIDE
}

#[cfg(not(any(unix, windows)))]
pub(crate) fn path_encoding() -> u8 {
    PATH_ENCODING_UTF8
}

#[cfg(unix)]
pub(crate) fn put_os_str(buf: &mut Vec<u8>, value: &OsStr) -> Result<()> {
    use std::os::unix::ffi::OsStrExt;
    put_bytes(buf, value.as_bytes())
}

#[cfg(windows)]
pub(crate) fn put_os_str(buf: &mut Vec<u8>, value: &OsStr) -> Result<()> {
    use std::os::windows::ffi::OsStrExt;
    let mut bytes = Vec::new();
    for unit in value.encode_wide() {
        bytes.extend_from_slice(&unit.to_le_bytes());
    }
    put_bytes(buf, &bytes)
}

#[cfg(not(any(unix, windows)))]
pub(crate) fn put_os_str(buf: &mut Vec<u8>, value: &OsStr) -> Result<()> {
    let text = value
        .to_str()
        .ok_or_else(|| Error::Snapshot("path is not valid UTF-8 on this platform".into()))?;
    put_bytes(buf, text.as_bytes())
}

#[cfg(unix)]
fn os_string_from_bytes(bytes: &[u8]) -> OsString {
    use std::os::unix::ffi::OsStringExt;
    OsString::from_vec(bytes.to_vec())
}

#[cfg(windows)]
fn os_string_from_bytes(bytes: &[u8]) -> Option<OsString> {
    use std::os::windows::ffi::OsStringExt;
    // Stable since 1.87, against an MSRV of 1.85 — see the note in content_cache.rs.
    if bytes.len() % std::mem::size_of::<u16>() != 0 {
        return None;
    }
    let units: Vec<u16> = bytes
        .chunks_exact(std::mem::size_of::<u16>())
        .map(|chunk| u16::from_le_bytes([chunk[0], chunk[1]]))
        .collect();
    Some(OsString::from_wide(&units))
}

#[cfg(not(any(unix, windows)))]
fn os_string_from_bytes(bytes: &[u8]) -> Option<OsString> {
    Some(OsString::from(String::from_utf8(bytes.to_vec()).ok()?))
}

/// Write to a sibling temporary file, then rename over the target.
///
/// Rename is atomic within a filesystem, so a reader either sees the whole old snapshot
/// or the whole new one. The temporary must be a sibling for that to hold — a rename
/// across filesystems is a copy, and copies are not atomic.
pub(crate) fn write_atomically(path: &Path, bytes: &[u8]) -> Result<()> {
    // Serialization is deterministic, so unchanged input encodes to the bytes already on
    // disk, and replacing them is a full write, an `F_FULLFSYNC`, and a rename that leave
    // the file exactly as it was. Comparing against the page-cached file costs a few
    // milliseconds and cannot go stale, because the bytes are the same bytes. The metadata
    // snapshot, whose images also differ in their pass-start stamp, compares in `publish`
    // instead, which records what the skip is worth (exp-066, exp-067).
    if keep_identical(path, bytes) {
        return Ok(());
    }
    replace_atomically(path, bytes)
}

/// [`write_atomically`] without first comparing against the file at `path`, for a caller
/// that has already compared.
fn replace_atomically(path: &Path, bytes: &[u8]) -> Result<()> {
    let parent = parent_dir(path);
    fs::create_dir_all(parent).map_err(|e| Error::io(parent, e))?;
    let (tmp, mut file) = create_temp_file(path, parent)?;
    let write_then_sync = file.write_all(bytes).and_then(|()| file.sync_all());
    if let Err(e) = write_then_sync {
        let _ = fs::remove_file(&tmp);
        return Err(Error::io(&tmp, e));
    }
    drop(file);

    if let Err(e) = fs::rename(&tmp, path) {
        let _ = fs::remove_file(&tmp);
        return Err(Error::io(path, e));
    }
    reap_stale_temporaries(parent, path, STALE_TEMP_AGE);
    Ok(())
}

/// Leave `path` in place when it already holds exactly `bytes`, moving only its mtime to
/// now.
///
/// The bytes were just produced again, so now is when they were last known to be current,
/// and moving the mtime says so without writing the payload. Best effort: a stale mtime
/// understates that, which fails safe.
fn keep_identical(path: &Path, bytes: &[u8]) -> bool {
    if !same_bytes_on_disk(path, bytes) {
        return false;
    }
    let _ = touch(path, std::time::SystemTime::now());
    true
}

/// Whether `path` already holds exactly `bytes`.
///
/// Streamed in 1 MiB pieces rather than read whole, so deciding not to write a 14 MB
/// snapshot does not cost a second 14 MB allocation beside the one being compared. Any
/// error -- missing file, short read, a file that grew under the comparison -- reads as
/// "different", and the caller writes; the comparison can only ever save work.
fn same_bytes_on_disk(path: &Path, bytes: &[u8]) -> bool {
    let Ok(mut file) = fs::File::open(path) else { return false };
    let Ok(metadata) = file.metadata() else { return false };
    if metadata.len() != u64::try_from(bytes.len()).unwrap_or(u64::MAX) {
        return false;
    }
    // A file that holds every byte and then more is not the same file.
    reads_as(&mut file, bytes) && matches!(file.read(&mut [0u8; 1]), Ok(0))
}

/// Whether the next bytes `file` yields are exactly `bytes`.
///
/// Streamed in 1 MiB pieces, and stopping at the first piece that differs; any read error
/// or early end reads as "different".
fn reads_as(file: &mut fs::File, bytes: &[u8]) -> bool {
    let mut buffer = vec![0u8; bytes.len().min(1 << 20)];
    let mut offset = 0usize;
    while offset < bytes.len() {
        let want = buffer.len().min(bytes.len() - offset);
        let read = match file.read(&mut buffer[..want]) {
            Ok(0) | Err(_) => return false,
            Ok(read) => read,
        };
        let end = offset + read;
        if buffer[..read] != bytes[offset..end] {
            return false;
        }
        offset = end;
    }
    true
}

/// Move `path`'s modification time to `when` without touching its contents.
fn touch(path: &Path, when: std::time::SystemTime) -> io::Result<()> {
    let file = OpenOptions::new().write(true).open(path)?;
    file.set_modified(when)
}

/// Remove long-abandoned temporaries beside `path`.
///
/// Best effort and deliberately silent: this is housekeeping, and a snapshot that was
/// written correctly must not fail because a directory could not be tidied. Anything
/// that cannot be read or removed is left for the next writer.
fn reap_stale_temporaries(parent: &Path, path: &Path, older_than: std::time::Duration) {
    let Some(prefix) = temp_prefix(path) else { return };
    let Ok(entries) = fs::read_dir(parent) else { return };
    let now = std::time::SystemTime::now();
    for entry in entries.flatten() {
        let name = entry.file_name();
        if !name.as_encoded_bytes().starts_with(prefix.as_encoded_bytes()) {
            continue;
        }
        let stale = entry
            .metadata()
            .and_then(|meta| meta.modified())
            .ok()
            .and_then(|modified| now.duration_since(modified).ok())
            .is_some_and(|age| age >= older_than);
        if stale {
            let _ = fs::remove_file(entry.path());
        }
    }
}

/// The shared leading portion of every temporary written for `path`.
///
/// Matching on this rather than on a full parse keeps the reaper from depending on the
/// pid, entropy, and sequence encoding: a temporary written by an older build with a
/// different suffix layout is still recognised and still collected.
fn temp_prefix(path: &Path) -> Option<OsString> {
    let mut prefix = OsString::from(".");
    prefix.push(path.file_name()?);
    prefix.push(".tmp.");
    Some(prefix)
}

/// The directory a target lives in, as a path that can actually be opened.
///
/// A bare relative target such as `snap.fdu` has an *empty* parent rather than none, and
/// the empty path is not the current directory: `create_dir_all` and `rename` tolerate
/// it, but `read_dir` does not — which silently disabled the reaper for exactly those
/// targets while the write itself kept working.
fn parent_dir(path: &Path) -> &Path {
    match path.parent() {
        Some(parent) if !parent.as_os_str().is_empty() => parent,
        _ => Path::new("."),
    }
}

/// The temporary name this process uses for `path` at `sequence`.
///
/// Split out so a test can plant a collision at a name the writer will actually try.
/// Guessing the shape does not work — the entropy is per process — and getting that
/// wrong is how the first version of the stale-temporary test came to exercise nothing.
fn temp_name(path: &Path, sequence: u64) -> OsString {
    let mut name = OsString::from(".");
    name.push(path.file_name().unwrap_or_else(|| OsStr::new("snapshot")));
    name.push(format!(".tmp.{}.{:016x}.{}", std::process::id(), *TEMP_FILE_ENTROPY, sequence));
    name
}

fn create_temp_file(path: &Path, parent: &Path) -> Result<(PathBuf, fs::File)> {
    for _ in 0..MAX_TEMP_CREATE_ATTEMPTS {
        let sequence = NEXT_TEMP_FILE.fetch_add(1, Ordering::Relaxed);
        let tmp = parent.join(temp_name(path, sequence));
        let mut options = OpenOptions::new();
        options.write(true).create_new(true);
        #[cfg(unix)]
        {
            use std::os::unix::fs::OpenOptionsExt;
            options.mode(0o600);
        }
        match options.open(&tmp) {
            Ok(file) => return Ok((tmp, file)),
            Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
            Err(error) => return Err(Error::io(&tmp, error)),
        }
    }
    Err(Error::io(
        parent,
        std::io::Error::new(
            std::io::ErrorKind::AlreadyExists,
            "could not reserve a unique snapshot temporary file",
        ),
    ))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::engine_contract::Observation;
    use crate::index::ExtTally;

    fn attrs(size: u64, mtime_ns: i64) -> Attrs {
        Attrs {
            size,
            allocated: size.div_ceil(512) * 512,
            mtime_ns,
            ctime_ns: mtime_ns,
            inode: size.wrapping_mul(7).wrapping_add(1),
            dev: 3,
        }
    }

    fn sample_index() -> Index {
        let mut index = Index::new("/some/root");
        index.apply_ok(&Observation::new(vec![
            Op::Upsert { path: PathBuf::from("src"), kind: EntryKind::Dir, attrs: attrs(0, 1) },
            Op::Upsert {
                path: PathBuf::from("src/main.rs"),
                kind: EntryKind::File,
                attrs: attrs(100, 10),
            },
            Op::Upsert {
                path: PathBuf::from("src/deep"),
                kind: EntryKind::Dir,
                attrs: attrs(0, 2),
            },
            Op::Upsert {
                path: PathBuf::from("src/deep/nested.rs"),
                kind: EntryKind::File,
                attrs: attrs(50, 20),
            },
            Op::Upsert {
                path: PathBuf::from("notes.md"),
                kind: EntryKind::File,
                attrs: attrs(7, 30),
            },
        ]));
        index
    }

    fn entry_count_offset(bytes: &[u8]) -> usize {
        let root_len_at = ROOT_OFFSET;
        let root_len = u32::from_le_bytes(
            bytes[root_len_at..root_len_at + 4]
                .try_into()
                .expect("saved snapshot has a root length"),
        );
        root_len_at + 4 + usize::try_from(root_len).expect("root length fits usize")
    }

    fn rewrite_checksum(bytes: &mut [u8]) {
        let payload_len = bytes.len() - CHECKSUM_BYTES - TRAILER.len();
        let checksum = crc32c(&bytes[..payload_len]);
        bytes[payload_len..payload_len + CHECKSUM_BYTES].copy_from_slice(&checksum.to_le_bytes());
    }

    /// The encoded name field and following attributes for every entry record.
    fn entry_record_fields(bytes: &[u8]) -> Vec<(std::ops::Range<usize>, std::ops::Range<usize>)> {
        let count_at = entry_count_offset(bytes);
        let count = u64::from_le_bytes(
            bytes[count_at..count_at + 8].try_into().expect("saved snapshot has an entry count"),
        );
        let mut at = count_at + 8;
        let mut fields = Vec::new();
        for _ in 0..count {
            let name_len_at = at + 5;
            let name_len = usize::try_from(u32::from_le_bytes(
                bytes[name_len_at..name_len_at + 4]
                    .try_into()
                    .expect("saved entry has a name length"),
            ))
            .expect("name length fits usize");
            let name_end = name_len_at + 4 + name_len;
            fields.push((name_len_at..name_end, name_end..name_end + 6 * 8));
            at = name_end + 6 * 8;
        }
        fields
    }

    fn replace_entry_name(image: &[u8], record: usize, name: &OsStr) -> Vec<u8> {
        let mut encoded = Vec::new();
        put_os_str(&mut encoded, name).expect("encode replacement name");
        let field = entry_record_fields(image)[record].0.clone();
        let mut rewritten = image.to_vec();
        rewritten.splice(field, encoded);
        rewrite_checksum(&mut rewritten);
        rewritten
    }

    #[test]
    fn crc32c_matches_the_standard_check_value() {
        assert_eq!(crc32c(b"123456789"), 0xe306_9283);
    }

    #[test]
    fn crc32c_slicing_matches_the_byte_reference_on_uneven_lengths() {
        // The 8-byte path and the remainder loop must agree with the classic
        // byte-at-a-time recurrence at every alignment, or an old snapshot's digest
        // stops verifying. The reference below IS that recurrence, against table 0.
        fn reference(bytes: &[u8]) -> u32 {
            let mut state = u32::MAX;
            for byte in bytes {
                let index = usize::from(state.to_le_bytes()[0] ^ *byte);
                state = CRC32C_TABLES[0][index] ^ (state >> 8);
            }
            !state
        }
        let mut data = Vec::new();
        let mut seed = 0x9e37_79b9u32;
        for length in [0usize, 1, 7, 8, 9, 15, 16, 63, 64, 65, 1000] {
            data.clear();
            for _ in 0..length {
                seed = seed.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
                data.push(seed.to_le_bytes()[0]);
            }
            assert_eq!(crc32c(&data), reference(&data), "length {length}");
        }
    }

    #[test]
    fn snapshot_names_must_equal_their_single_normal_component() {
        assert!(is_snapshot_name(OsStr::new("notes.md")));
        assert!(!is_snapshot_name(OsStr::new("notes.md/")));
        assert!(!is_snapshot_name(OsStr::new("a/b")));
        assert!(!is_snapshot_name(OsStr::new("../bad")));
        assert!(!is_snapshot_name(OsStr::new("")));
        assert!(!is_snapshot_name(OsStr::new(".")));
        assert!(!is_snapshot_name(OsStr::new("..")));
        #[cfg(unix)]
        {
            use std::os::unix::ffi::OsStringExt;
            let native = OsString::from_vec(vec![b'n', 0x80]);
            assert!(is_snapshot_name(&native), "canonical validation is OsStr identity");
            let aliased = OsString::from_vec(vec![b'n', 0x80, b'/']);
            assert!(!is_snapshot_name(&aliased));
        }
    }

    #[test]
    fn a_valid_forged_rename_loads_and_is_found_by_lookup() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snapshot.fdu");
        let mut index = Index::new("/some/root");
        index.apply_ok(&Observation::new(vec![Op::Upsert {
            path: PathBuf::from("valid"),
            kind: EntryKind::File,
            attrs: attrs(1, 1),
        }]));
        save(&index, &path).expect("save");
        let saved = fs::read(&path).expect("read");
        let forged = replace_entry_name(&saved, 1, OsStr::new("renamed"));
        fs::write(&path, forged).expect("write valid rename");
        let restored = load(&path).expect("load").expect("a valid rename is a snapshot");
        assert!(restored.lookup(Path::new("renamed")).is_some());
        assert!(restored.lookup(Path::new("valid")).is_none());
    }

    #[test]
    fn noncanonical_entry_names_are_rejected_after_integrity_checks() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snapshot.fdu");
        let mut index = Index::new("/some/root");
        index.apply_ok(&Observation::new(vec![Op::Upsert {
            path: PathBuf::from("valid"),
            kind: EntryKind::File,
            attrs: attrs(1, 1),
        }]));
        save(&index, &path).expect("save");
        let saved = fs::read(&path).expect("read");

        #[cfg(windows)]
        let names = ["a/", "a//", "a/.", "./a", r"a\", r"a\\", r"a\."];
        #[cfg(not(windows))]
        let names = ["a/", "a//", "a/.", "./a"];
        for name in names {
            let forged = replace_entry_name(&saved, 1, OsStr::new(name));
            fs::write(&path, forged).expect("write forged snapshot");
            assert!(load(&path).expect("malformed snapshot is a miss").is_none(), "{name:?}");
        }
    }

    #[test]
    fn a_checksummed_name_alias_cannot_serve_cache_only_content() {
        let dir = tempfile::tempdir().expect("tempdir");
        let root = dir.path().join("root");
        fs::create_dir(&root).expect("create root");
        fs::write(root.join("a"), b"one two\n").expect("write first");
        fs::write(root.join("bb"), b"one two\n").expect("write second");
        let snapshot_path = dir.path().join("snapshot.fdu");
        let config = crate::OpenFixture {
            cache_path: Some(snapshot_path.clone()),
            policy: crate::CachePolicy::Auto,
            analysis: crate::content::AnalysisRequest {
                profile: crate::content::AnalysisSet::NONE.with_lines(),
                ..crate::content::AnalysisRequest::default()
            },
            ..crate::OpenFixture::default()
        };
        crate::open_fixture(&root, &config).expect("seed snapshot and sidecar");

        let mut image = fs::read(&snapshot_path).expect("read snapshot");
        let fields = entry_record_fields(&image);
        assert_eq!(fields.len(), 3, "root and two files");
        let first_attrs = image[fields[1].1.clone()].to_vec();
        image[fields[2].1.clone()].copy_from_slice(&first_attrs);
        let forged = replace_entry_name(&image, 2, OsStr::new("a/"));
        fs::write(&snapshot_path, forged).expect("write checksummed alias");

        let only = crate::OpenFixture { stale_ok: true, ..config };
        assert!(
            matches!(crate::open_fixture(&root, &only), Err(Error::Snapshot(_))),
            "a malformed snapshot cannot shrink the cache-only completeness denominator"
        );
    }

    #[test]
    fn a_loaded_index_reports_cached_provenance_not_fresh() {
        // The gap that motivated the provenance model. A snapshot is complete when it
        // is written, so a loaded index used to claim `Fresh` — true of when the file
        // was made, and exactly backwards for a consumer painting on load, which needs
        // to know nothing has been checked since.
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snapshot.fdu");
        let original = sample_index();
        save(&original, &path).expect("save");

        let restored = load(&path).expect("load").expect("snapshot present");
        let provenance =
            restored.provenance(Path::new("src/main.rs")).expect("the loaded entry is present");
        assert_eq!(provenance.source, crate::Source::Cached);
        assert!(!provenance.is_verified(), "nothing has been stat'd since the load");
        assert!(
            provenance.observed_at_ns > 0,
            "a cached value must say as of when, or a UI cannot label it"
        );

        // A freshly scanned index is the contrasting case.
        assert_eq!(
            original.provenance(Path::new("src/main.rs")).expect("present").source,
            crate::Source::Scanned
        );
    }

    #[test]
    fn a_loaded_root_reports_cached_provenance_not_fresh() {
        // The root is the one entry the child-only test above cannot cover, and it is
        // the one that matters most: whole-tree totals hang off it, so `fdu ~` reads
        // the root's provenance to label its headline number. `apply_upsert` handles
        // the root in a separate branch, and that branch used to skip the source stamp
        // entirely — leaving a snapshot-loaded root claiming `Scanned` and
        // `is_verified()`, the precise silent lie the model exists to prevent.
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snapshot.fdu");
        save(&sample_index(), &path).expect("save");

        let restored = load(&path).expect("load").expect("snapshot present");
        let provenance = restored.provenance(Path::new("")).expect("the root is always present");
        assert_eq!(provenance.source, crate::Source::Cached, "the root came off disk too");
        assert!(!provenance.is_verified(), "nothing has been stat'd since the load");
        assert!(
            provenance.observed_at_ns > 0,
            "a cached total must say as of when, or a UI cannot label it"
        );
    }

    #[test]
    fn cached_observation_time_comes_from_the_header_after_touch_or_copy() {
        use std::time::{Duration, UNIX_EPOCH};

        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snapshot.fdu");
        let copied = dir.path().join("copied.fdu");
        let mut original = sample_index();
        original.set_writing_pass_started_at_ns(1_000);
        save(&original, &path).expect("save");

        // Cache files are routinely kept in place or copied between cache locations. Neither
        // operation observes the tree, so neither file mtime may become value provenance.
        touch(&path, UNIX_EPOCH + Duration::from_secs(10)).expect("touch cache image");
        fs::copy(&path, &copied).expect("copy cache image");
        touch(&copied, UNIX_EPOCH + Duration::from_secs(20)).expect("touch copied image");

        for cache in [&path, &copied] {
            let restored = load(cache).expect("load").expect("present");
            for entry in [Path::new(""), Path::new("src/main.rs")] {
                assert_eq!(
                    restored.provenance(entry).expect("present").observed_at_ns,
                    1_000,
                    "{} uses its persisted pass start, not its cache-file mtime",
                    cache.display()
                );
            }
        }
    }

    #[test]
    fn revalidating_a_loaded_index_promotes_entries_out_of_cached() {
        // The failure a reviewer caught on PR #6: entries were stamped only when
        // allocated, so a warm sweep could stat every entry and leave them all
        // reporting `Cached`. A consumer could then never clear a stale-value
        // indicator no matter how much verification ran, which defeats the point.
        let dir = tempfile::tempdir().expect("tempdir");
        let tree = dir.path().join("tree");
        std::fs::create_dir_all(tree.join("sub")).expect("create dirs");
        std::fs::write(tree.join("sub/file.txt"), b"contents").expect("write");
        let snapshot_path = dir.path().join("snapshot.fdu");

        let config = crate::ScanConfig::default();
        let (original, report) = crate::scan::scan_into_index(&tree, &config).expect("scan");
        assert!(report.is_complete());
        save(&original, &snapshot_path).expect("save");

        let mut restored = load(&snapshot_path).expect("load").expect("present");
        let target = Path::new("sub/file.txt");
        assert_eq!(
            restored.provenance(target).expect("present").source,
            crate::Source::Cached,
            "straight off disk, nothing has been checked"
        );

        // A sweep that finds nothing changed still verified every entry it stat'd.
        let reconciled =
            crate::scan::reconcile(&mut restored, &config, &mut |_| {}).expect("reconcile");
        assert!(reconciled.is_complete());
        assert_eq!(reconciled.apply.updated, 0, "the tree did not change");

        let provenance = restored.provenance(target).expect("present");
        assert_eq!(
            provenance.source,
            crate::Source::Revalidated,
            "an unchanged entry that was freshly stat'd has still been verified"
        );
        assert!(provenance.is_verified());
    }

    /// Every entry, not just the root.
    ///
    /// The loader inserts beneath a known parent instead of replaying an observation, so
    /// it no longer shares a code path with the producer that built the original. Root
    /// totals cannot catch a roll-up that is wrong halfway down — the errors would have
    /// to cancel at the root to hide, but a single misplaced subtree does not touch the
    /// root at all. This walks both trees and compares each directory's own roll-up,
    /// each entry's kind, attributes and extension tallies, and the shape of the tree.
    #[test]
    fn round_trip_preserves_every_entry_not_just_the_root() {
        let dir = tempfile::tempdir().expect("tempdir");
        let tree = dir.path().join("tree");
        let snapshot_path = dir.path().join("cache").join("snap.fdu");
        // Depth and fan-out both matter: a parent-relative insert that mis-parents an
        // entry shows up as a wrong roll-up on an interior directory.
        for (relative, contents) in [
            ("a.rs", &b"fn main() {}"[..]),
            ("deep/one/two/three/leaf.txt", b"leaf"),
            ("deep/one/two/sibling.rs", b"sibling"),
            ("deep/one/other.md", b"# other"),
            ("wide/w1.txt", b"1"),
            ("wide/w2.txt", b"22"),
            ("wide/w3.rs", b"333"),
            ("empty/.keep", b""),
        ] {
            let path = tree.join(relative);
            fs::create_dir_all(path.parent().expect("parent")).expect("create dirs");
            fs::write(&path, contents).expect("write");
        }

        let config = crate::ScanConfig::default();
        let (original, report) = crate::scan::scan_into_index(&tree, &config).expect("scan");
        assert!(report.is_complete());
        save(&original, &snapshot_path).expect("save");
        let restored = load(&snapshot_path).expect("load").expect("present");

        assert_eq!(restored.len(), original.len(), "entry count");

        // Walk the original and demand the same entry, in the same place, with the same
        // numbers, in the restored index.
        let mut stack = vec![crate::index::EntryId::ROOT];
        let mut compared = 0_u64;
        while let Some(id) = stack.pop() {
            let path = original.path_of(id).expect("original path");
            let mirrored = restored.lookup(&path).expect("restored entry at the same path");
            assert_eq!(
                original.kind_of(id),
                restored.kind_of(mirrored),
                "kind at {}",
                path.display()
            );
            assert_eq!(
                original.attrs_of(id),
                restored.attrs_of(mirrored),
                "attrs at {}",
                path.display()
            );
            // Roll-ups and children exist only on directories; a file legitimately has
            // neither, and demanding them would fail on the tree rather than the loader.
            if original.kind_of(id) == Some(crate::EntryKind::Dir) {
                let (before, after) = (
                    original.rollup_of(id).expect("original rollup"),
                    restored.rollup_of(mirrored).expect("restored rollup"),
                );
                assert_eq!(
                    (
                        before.files,
                        before.dirs,
                        before.bytes,
                        before.allocated,
                        before.newest_mtime_ns
                    ),
                    (after.files, after.dirs, after.bytes, after.allocated, after.newest_mtime_ns),
                    "rollup at {}",
                    path.display()
                );
                assert_eq!(before.by_ext, after.by_ext, "extension tallies at {}", path.display());
                let names: Vec<_> = original
                    .children_of(id)
                    .expect("original children")
                    .map(|(name, _)| name.to_os_string())
                    .collect();
                let mirrored_names: Vec<_> = restored
                    .children_of(mirrored)
                    .expect("restored children")
                    .map(|(name, _)| name.to_os_string())
                    .collect();
                assert_eq!(names, mirrored_names, "children of {}", path.display());
                stack.extend(original.children_of(id).expect("children").map(|(_, child)| child));
            }
            compared += 1;
        }
        assert_eq!(compared, original.len(), "every entry was compared");

        // The loader must not leave the index looking like it has pending history.
        assert_eq!(restored.clock(), crate::Clock::ZERO);
        assert!(restored.since(crate::Clock::ZERO).commits.is_empty());
    }

    #[test]
    fn round_trip_preserves_tree_and_rollups() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("cache").join("snap.fdu");
        let original = sample_index();

        save(&original, &path).expect("save");
        let restored = load(&path).expect("load").expect("snapshot present");

        assert_eq!(restored.root_path(), Path::new("/some/root"));
        assert_eq!(restored.clock(), crate::Clock::ZERO);
        assert!(restored.since(crate::Clock::ZERO).commits.is_empty());
        assert_eq!(restored.len(), original.len());
        // Public roll-ups resolve assignment-ordered extension ids to stable names.
        let (restored_total, original_total) = (restored.total(), original.total());
        assert_eq!(
            (restored_total.files, restored_total.dirs, restored_total.bytes),
            (original_total.files, original_total.dirs, original_total.bytes)
        );
        assert_eq!(restored_total.allocated, original_total.allocated);
        assert_eq!(restored_total.newest_mtime_ns, original_total.newest_mtime_ns);
        assert_eq!(restored_total.by_ext, original_total.by_ext);
        assert!(!original.serving_indexes_enabled());
        assert!(!restored.serving_indexes_enabled());
        assert_eq!(restored.total().files, 3);
        assert_eq!(restored.total().dirs, 2);
        assert_eq!(restored.total().bytes, 157);
        assert_eq!(
            restored.total().by_ext[".rs"],
            ExtTally { files: 2, bytes: 150, allocated: 1024 }
        );
        assert_eq!(
            restored.attrs(Path::new("src/deep/nested.rs")),
            original.attrs(Path::new("src/deep/nested.rs"))
        );
    }

    #[test]
    fn round_trip_preserves_exact_controls_and_fixed_partitions() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("controls.fdu");
        let mut original =
            Index::new_with_scope("/some/root", crate::test_support::observing_controls());
        original.apply_ok(&Observation::new(vec![
            Op::Upsert {
                path: PathBuf::from(".gitignore"),
                kind: EntryKind::File,
                attrs: attrs(6, 1),
            },
            Op::Upsert {
                path: PathBuf::from("debug.log"),
                kind: EntryKind::File,
                attrs: attrs(10, 2),
            },
            Op::Upsert {
                path: PathBuf::from("keep.rs"),
                kind: EntryKind::File,
                attrs: attrs(20, 3),
            },
            Op::ControlUpsert { path: PathBuf::from(".gitignore"), source: b"*.log\n".to_vec() },
        ]));

        save(&original, &path).expect("save");
        let restored = load(&path).expect("load").expect("snapshot present");

        assert!(
            restored
                .controls()
                .expect("control state observed")
                .source_is(Path::new(".gitignore"), b"*.log\n")
        );
        assert_eq!(restored.controls().expect("control state observed").source_bytes(), 6);
        assert_eq!(
            restored.is_ignored(Path::new("debug.log")).expect("control state observed"),
            Some(true)
        );
        assert_eq!(
            restored.is_ignored(Path::new("keep.rs")).expect("control state observed"),
            Some(false)
        );
        let partitions = restored.partition_total().expect("control state observed");
        assert_eq!(partitions, original.partition_total().expect("control state observed"));
        assert_eq!(partitions.all.files, 3);
        assert_eq!(partitions.unignored.files, 2);
    }

    #[test]
    fn removing_the_last_control_before_save_round_trips_an_empty_table() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("no-controls.fdu");
        let mut original =
            Index::new_with_scope("/some/root", crate::test_support::observing_controls());
        original.apply_ok(&Observation::new(vec![
            Op::Upsert {
                path: PathBuf::from(".gitignore"),
                kind: EntryKind::File,
                attrs: attrs(6, 1),
            },
            Op::Upsert {
                path: PathBuf::from("debug.log"),
                kind: EntryKind::File,
                attrs: attrs(10, 2),
            },
            Op::ControlUpsert { path: PathBuf::from(".gitignore"), source: b"*.log\n".to_vec() },
        ]));
        original
            .apply_ok(&Observation::new(vec![Op::Remove { path: PathBuf::from(".gitignore") }]));

        save(&original, &path).expect("save");
        let restored = load(&path).expect("load").expect("snapshot present");

        assert!(restored.controls().expect("control state observed").is_empty());
        assert_eq!(
            restored.is_ignored(Path::new("debug.log")).expect("control state observed"),
            Some(false)
        );
        let partitions = restored.partition_total().expect("control state observed");
        assert_eq!(partitions.all, partitions.unignored);
    }

    #[test]
    fn a_control_table_at_its_shared_bound_round_trips() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("bounded-controls.fdu");
        let mut original =
            Index::new_with_scope("/some/root", crate::test_support::observing_controls());
        let source = crate::control::source_at_test_limit();
        original.apply_ok(&Observation::new(vec![Op::ControlUpsert {
            path: PathBuf::from(".gitignore"),
            source: source.clone(),
        }]));
        assert_eq!(
            original.controls().expect("control state observed").retained_cost(),
            crate::control::DEFAULT_CONTROL_BUDGET
        );

        save(&original, &path).expect("save at bound");
        let restored = load(&path).expect("load").expect("snapshot present");

        assert_eq!(
            restored.controls().expect("control state observed").retained_cost(),
            original.controls().expect("control state observed").retained_cost()
        );
        assert!(
            restored
                .controls()
                .expect("control state observed")
                .source_is(Path::new(".gitignore"), &source)
        );
    }

    /// A control table must enforce the limits its scope was taken under. A hand-built index
    /// whose scope claims other limits is refused at save with a typed error, a snapshot
    /// whose control section its header's control tier would not admit fails closed at
    /// load, and `Index::new_with_config` builds an index whose table and scope agree.
    #[test]
    fn control_limits_that_disagree_with_the_scope_are_refused_at_save_and_load() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("limits.fdu");
        let lifted = crate::ScanConfig {
            control_limits: crate::control::ControlLimits {
                budget: None,
                ..crate::control::ControlLimits::default()
            },
            ..crate::ScanConfig::default()
        };

        let mismatched = Index::new_with_scope("/some/root", lifted.scope());
        let error = save(&mismatched, &path).expect_err("the scope claims no budget");
        assert!(
            matches!(
                error,
                Error::ControlLimitsOutsideScope { limits }
                    if limits == crate::control::ControlLimits::default()
            ),
            "{error}"
        );
        assert!(!path.exists(), "nothing is written");

        let mut agreeing = Index::new_with_config("/some/root", &lifted);
        agreeing.apply_ok(&Observation::new(vec![Op::ControlUpsert {
            path: PathBuf::from(".gitignore"),
            source: b"*.log\n".to_vec(),
        }]));
        assert_eq!(agreeing.scope(), lifted.scope());
        save(&agreeing, &path).expect("save an index whose table enforces its scope's limits");
        let restored = load(&path).expect("load").expect("snapshot present");
        assert_eq!(restored.control_coverage(), agreeing.control_coverage());
        assert_eq!(restored.snapshot_identity(), lifted.snapshot_identity());

        // The limits live only in the header's control tier. A CRC-valid snapshot whose
        // header claims a line limit the retained source exceeds, or claims no control
        // state beside a retained source, was written by no save, so it fails closed like
        // any other corruption.
        let saved = fs::read(&path).expect("read snapshot");
        let controls_at = IDENTITY_OFFSET + crate::stored_state::ENTRY_TIER_BYTES;
        let controls = controls_at..controls_at + crate::stored_state::CONTROL_TIER_BYTES;
        let blind = crate::ScanConfig { read_controls: false, ..lifted.clone() };
        let forge = |tier: ControlTierIdentity| {
            let mut forged = saved.clone();
            forged[controls.clone()].copy_from_slice(&tier.encode());
            rewrite_checksum(&mut forged);
            fs::write(&path, &forged).expect("write forged limits");
            (
                load(&path).expect("forged equals absent"),
                load_serving(&path, blind.types_shared(), blind.snapshot_identity())
                    .expect("projected forged equals absent"),
            )
        };
        let tight = crate::control::ControlLimits { line_limit: Some(1), ..lifted.control_limits };
        let (exact, projected) = forge(ControlTierIdentity::Observed { limits: tight });
        assert!(exact.is_none());
        assert!(matches!(projected, LoadOutcome::Absent));
        let (exact, projected) = forge(ControlTierIdentity::NotObserved);
        assert!(exact.is_none());
        assert!(matches!(projected, LoadOutcome::Absent));
        // The same splice with the saved tier restores, so the splice is not what failed.
        let (exact, projected) = forge(lifted.control_identity());
        assert!(exact.is_some());
        let LoadOutcome::Served { index: projected, stored } = projected else {
            panic!("the valid control payload projects");
        };
        assert_eq!(serves_snapshot(stored, blind.snapshot_identity()), Serves::ProjectControlsOff);
        assert_eq!(projected.snapshot_identity(), blind.snapshot_identity());
        assert_eq!(projected.scope(), blind.scope());
        assert!(matches!(projected.controls(), Err(Error::ControlStateNotObserved)));
    }

    /// An index whose control table refused some sources, and its path-ordered tail.
    fn index_with_refused_controls() -> Index {
        let mut index =
            Index::new_with_scope("/some/root", crate::test_support::observing_controls());
        let mut line = vec![b'x'; crate::control::DEFAULT_CONTROL_LINE_LIMIT + 1];
        line.push(b'\n');
        index.apply_ok(&Observation::new(vec![
            Op::Upsert {
                path: PathBuf::from(".gitignore"),
                kind: EntryKind::File,
                attrs: attrs(6, 1),
            },
            Op::Upsert { path: PathBuf::from("a"), kind: EntryKind::Dir, attrs: attrs(0, 1) },
            Op::Upsert {
                path: PathBuf::from("a/.gitignore"),
                kind: EntryKind::File,
                attrs: attrs(1, 1),
            },
            Op::Upsert { path: PathBuf::from("b"), kind: EntryKind::Dir, attrs: attrs(0, 1) },
            Op::Upsert {
                path: PathBuf::from("b/.gitignore"),
                kind: EntryKind::File,
                attrs: attrs(1, 1),
            },
            Op::Upsert {
                path: PathBuf::from("b/debug.log"),
                kind: EntryKind::File,
                attrs: attrs(9, 1),
            },
            Op::ControlUpsert { path: PathBuf::from(".gitignore"), source: b"*.log\n".to_vec() },
            Op::ControlUpsert { path: PathBuf::from("a/.gitignore"), source: line },
            Op::ControlUpsert {
                path: PathBuf::from("b/.gitignore"),
                source: b"y\n".repeat(crate::control::DEFAULT_CONTROL_BUDGET / 2),
            },
        ]));
        index
    }

    /// A snapshot of an index that refused control files reloads with the same coverage,
    /// rather than as an index whose every rule applied.
    #[test]
    fn a_snapshot_with_refused_controls_reloads_its_coverage_exactly() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("refused.fdu");
        let original = index_with_refused_controls();
        let crate::control::ControlCoverage::Observed(coverage) = original.control_coverage()
        else {
            panic!("observed");
        };
        assert_eq!((coverage.applied, coverage.refused), (1, 2));

        save(&original, &path).expect("save a partially covered index");
        let restored = load(&path).expect("load").expect("snapshot present");

        assert_eq!(restored.control_coverage(), original.control_coverage());
        assert_eq!(
            restored.is_ignored(Path::new("b/debug.log")).expect("observed"),
            original.is_ignored(Path::new("b/debug.log")).expect("observed")
        );
        assert_eq!(
            restored.partition_total().expect("observed"),
            original.partition_total().expect("observed")
        );
    }

    /// The refusal records are parsed as strictly as the rest: an unknown reason, and a
    /// refusal naming a retained source, are corrupt.
    #[test]
    fn corrupt_refusal_records_fail_closed() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("refused.fdu");
        save(&index_with_refused_controls(), &path).expect("save");
        let saved = fs::read(&path).expect("read snapshot");
        let footer = CHECKSUM_BYTES + TRAILER.len();
        // The last refusal is `b/.gitignore` and its one-byte reason ends the payload.
        let reason_at = saved.len() - footer - 1;
        assert_eq!(saved[reason_at], REFUSED_FOR_BUDGET);

        let mut unknown_reason = saved;
        unknown_reason[reason_at] = 9;
        rewrite_checksum(&mut unknown_reason);
        fs::write(&path, &unknown_reason).expect("write corrupt reason");
        assert!(load(&path).expect("corrupt equals absent").is_none());

        // A refusal naming a retained source, or one already refused, is corrupt, and so is a
        // refusal by a limit the section records as unbounded, which refuses nothing. Built
        // with the platform's own path encoding, so the same bytes mean the same on every
        // target.
        let defaults = crate::control::ControlLimits::default();
        let section = |refusals: &[(&str, u8)]| {
            let mut section = Vec::new();
            section.extend_from_slice(&1_u32.to_le_bytes());
            put_os_str(&mut section, OsStr::new(".gitignore")).expect("retained path");
            section.extend_from_slice(&6_u32.to_le_bytes());
            section.extend_from_slice(b"*.log\n");
            let count = u32::try_from(refusals.len()).expect("few refusals");
            section.extend_from_slice(&count.to_le_bytes());
            for (refusal, reason) in refusals {
                put_os_str(&mut section, OsStr::new(refusal)).expect("refused path");
                section.push(*reason);
            }
            section
        };
        let no_budget = crate::control::ControlLimits { budget: None, ..defaults };
        let no_line_limit = crate::control::ControlLimits { line_limit: None, ..defaults };
        for (limits, refusals) in [
            (defaults, &[("a/.gitignore", REFUSED_FOR_BUDGET)][..]),
            (no_budget, &[("a/.gitignore", REFUSED_FOR_LINE_LIMIT)][..]),
            (no_line_limit, &[("a/.gitignore", REFUSED_FOR_BUDGET)][..]),
        ] {
            let tier = ControlTierIdentity::Observed { limits };
            let valid = read_controls(&mut section(refusals).as_slice(), tier)
                .expect("a well-formed control section");
            assert_eq!((valid.len(), valid.refused_len()), (1, 1));
        }
        for (limits, refusals) in [
            (defaults, &[(".gitignore", REFUSED_FOR_BUDGET)][..]),
            (
                defaults,
                &[("a/.gitignore", REFUSED_FOR_BUDGET), ("a/.gitignore", REFUSED_FOR_BUDGET)][..],
            ),
            (no_budget, &[("a/.gitignore", REFUSED_FOR_BUDGET)][..]),
            (no_line_limit, &[("a/.gitignore", REFUSED_FOR_LINE_LIMIT)][..]),
        ] {
            let tier = ControlTierIdentity::Observed { limits };
            assert!(
                matches!(
                    read_controls(&mut section(refusals).as_slice(), tier),
                    Err(ParseError::Invalid)
                ),
                "{limits:?} {refusals:?}"
            );
        }
        // A tier that observed nothing holds no source and no refusal.
        for refusals in [&[][..], &[("a/.gitignore", REFUSED_FOR_BUDGET)][..]] {
            assert!(matches!(
                read_controls(&mut section(refusals).as_slice(), ControlTierIdentity::NotObserved),
                Err(ParseError::Invalid)
            ));
        }
    }

    /// A snapshot with no control state loads without walking the tree to reclassify it.
    ///
    /// Installing the loaded control table re-evaluated every entry's ignored bit, allocating
    /// a path per entry, even when the table was empty and no bit could move. That is every
    /// warm open of a tree with no `.gitignore`, and it scaled with the tree.
    #[test]
    fn loading_a_snapshot_without_controls_skips_the_reclassification_walk() {
        fn visits_while_loading(path: &Path) -> (u64, Index) {
            crate::index::RECLASSIFY_VISITS.with(|visits| visits.set(0));
            let restored = load(path).expect("load").expect("snapshot present");
            (crate::index::RECLASSIFY_VISITS.with(std::cell::Cell::get), restored)
        }

        let dir = tempfile::tempdir().expect("tempdir");
        let mut original =
            Index::new_with_scope("/some/root", crate::test_support::observing_controls());
        let mut ops = vec![Op::Upsert {
            path: PathBuf::from("src"),
            kind: EntryKind::Dir,
            attrs: attrs(0, 1),
        }];
        ops.extend((0..64).map(|sequence| Op::Upsert {
            path: PathBuf::from(format!("src/file-{sequence:02}.rs")),
            kind: EntryKind::File,
            attrs: attrs(sequence + 1, 2),
        }));
        original.apply_baseline_ok(&Observation::new(ops));
        let plain = dir.path().join("plain.fdu");
        save(&original, &plain).expect("save");

        let (visits, restored) = visits_while_loading(&plain);

        assert_eq!(visits, 0, "an empty control table has nothing to reclassify");
        assert!(restored.control_table().is_empty());
        // The load still answers as the saved index did.
        assert_eq!(
            restored.is_ignored(Path::new("src/file-00.rs")).expect("control state observed"),
            Some(false)
        );
        assert_eq!(
            restored.partition_total().expect("control state observed"),
            original.partition_total().expect("control state observed")
        );

        // The probe sees the walk when there is something to walk for.
        original.apply_ok(&Observation::new(vec![Op::ControlUpsert {
            path: PathBuf::from("src/.gitignore"),
            source: b"file-0*.rs\n".to_vec(),
        }]));
        let controlled = dir.path().join("controlled.fdu");
        save(&original, &controlled).expect("save");

        let (visits, restored) = visits_while_loading(&controlled);

        assert!(visits > 64, "{visits}");
        assert_eq!(
            restored.is_ignored(Path::new("src/file-00.rs")).expect("control state observed"),
            Some(true)
        );
        assert_eq!(
            restored.is_ignored(Path::new("src/file-10.rs")).expect("control state observed"),
            Some(false)
        );
    }

    #[test]
    fn round_trip_handles_wide_directory_fanout() {
        // Load resolves each record's id from its parent. Doing that by scanning the
        // parent's children costs O(width^2) per directory, which is invisible on the
        // handful of siblings every other test uses and dominates a real one — a
        // node_modules or a Maildir is thousands wide.
        const CHILDREN: u64 = 4_096;
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("wide.fdu");
        let mut original = Index::new("/some/root");
        let ops = (0..CHILDREN)
            .map(|sequence| Op::Upsert {
                path: PathBuf::from(format!("child-{sequence:04}.dat")),
                kind: EntryKind::File,
                attrs: attrs(sequence + 1, i64::try_from(sequence).expect("fanout fits i64")),
            })
            .collect();
        original.apply_baseline_ok(&Observation::new(ops));

        save(&original, &path).expect("save wide snapshot");
        let restored = load(&path).expect("load wide snapshot").expect("snapshot present");

        assert_eq!(restored.total().files, CHILDREN);
        assert_eq!(restored.len(), original.len());
        // The last child is the one a linear scan reaches last, so it is the one that
        // proves the resolution used the parent's map rather than its sibling order.
        assert_eq!(
            restored.attrs(Path::new("child-4095.dat")),
            original.attrs(Path::new("child-4095.dat"))
        );
    }

    #[test]
    fn shared_save_captures_before_filesystem_io() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("shared.fdu");
        let handle = IndexHandle::new(sample_index());

        save_handle(&handle, &path).expect("save shared snapshot");
        handle
            .apply(&Observation::new(vec![Op::Upsert {
                path: PathBuf::from("after.txt"),
                kind: EntryKind::File,
                attrs: attrs(9, 40),
            }]))
            .expect("mutate after capture");

        let restored = load(&path).expect("load").expect("snapshot present");
        assert!(restored.lookup(Path::new("after.txt")).is_none());
        assert!(handle.kind(Path::new("after.txt")).expect("query").is_some());
    }

    #[test]
    fn missing_snapshot_is_absent_not_an_error() {
        let dir = tempfile::tempdir().expect("tempdir");
        let loaded = load(&dir.path().join("nope.fdu")).expect("load must not error");
        assert!(loaded.is_none());
    }

    #[test]
    fn configured_size_limit_rejects_before_body_allocation() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");
        save(&sample_index(), &path).expect("save");
        let file_len = fs::metadata(&path).expect("metadata").len();

        assert!(load_with_size_limit(&path, file_len - 1).expect("load must not error").is_none());
    }

    #[test]
    fn foreign_file_is_treated_as_absent() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");
        fs::write(&path, b"this is not a snapshot at all").expect("write");
        assert!(load(&path).expect("load must not error").is_none());
    }

    #[test]
    fn truncated_snapshot_is_treated_as_absent() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");
        save(&sample_index(), &path).expect("save");

        let full = fs::read(&path).expect("read");
        for cut in [full.len() - 1, full.len() / 2, MAGIC.len() + 2] {
            fs::write(&path, &full[..cut]).expect("truncate");
            assert!(
                load(&path).expect("load must not error").is_none(),
                "a snapshot truncated to {cut} bytes must not parse"
            );
        }
    }

    #[test]
    fn corrupt_body_with_intact_magic_and_trailer_is_rejected() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");
        save(&sample_index(), &path).expect("save");

        let mut bytes = fs::read(&path).expect("read");
        // Claim far more entries than the body holds.
        let count_at = entry_count_offset(&bytes);
        bytes[count_at..count_at + 8].copy_from_slice(&u64::MAX.to_le_bytes());
        rewrite_checksum(&mut bytes);
        fs::write(&path, &bytes).expect("write");

        assert!(load(&path).expect("load must not error").is_none());
    }

    #[test]
    fn a_snapshot_whose_sizes_sum_past_u64_fails_closed() {
        // The image is structurally valid and its checksum is right; only the arithmetic
        // is impossible. A load that summed it would panic in debug builds and wrap in
        // release builds, so it is refused like any other corrupt image (fdu-sqyk).
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");
        save(&sample_index(), &path).expect("save");

        let mut bytes = fs::read(&path).expect("read");
        let mut files = 0;
        for (name, attrs) in entry_record_fields(&bytes) {
            let kind_at = name.start - 1;
            if EntryKind::from_u8(bytes[kind_at]) != Some(EntryKind::File) {
                continue;
            }
            bytes[attrs.start..attrs.start + 8].copy_from_slice(&u64::MAX.to_le_bytes());
            files += 1;
            if files == 2 {
                break;
            }
        }
        assert_eq!(files, 2, "the sample has two files to inflate");
        rewrite_checksum(&mut bytes);
        fs::write(&path, &bytes).expect("write");

        assert!(load(&path).expect("load must not error").is_none());
    }

    #[test]
    fn plausible_attribute_corruption_is_rejected() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");
        save(&sample_index(), &path).expect("save");

        let mut bytes = fs::read(&path).expect("read");
        let records_at = entry_count_offset(&bytes) + 8;
        // The root record has an empty name, so its first attribute starts immediately
        // after parent, kind, and encoded-name length. Changing a low size byte keeps the
        // image structurally valid and would silently alter the restored state without an
        // integrity check.
        let root_size_at = records_at + 4 + 1 + 4;
        bytes[root_size_at] ^= 1;
        fs::write(&path, &bytes).expect("write");

        assert!(load(&path).expect("load must not error").is_none());
    }

    #[test]
    fn entry_names_with_path_components_are_rejected() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");
        save(&sample_index(), &path).expect("save");

        let mut bytes = fs::read(&path).expect("read");
        let count_at = entry_count_offset(&bytes);
        let records_at = count_at + 8;
        let first_child_name_len_at = records_at + MIN_RECORD_BYTES + 4 + 1;
        let old_name_len = u32::from_le_bytes(
            bytes[first_child_name_len_at..first_child_name_len_at + 4]
                .try_into()
                .expect("saved snapshot has a child name length"),
        );
        let old_name_end = first_child_name_len_at
            + 4
            + usize::try_from(old_name_len).expect("name length fits usize");
        let mut invalid_name = Vec::new();
        put_os_str(&mut invalid_name, OsStr::new("../bad")).expect("encode invalid name");
        bytes.splice(first_child_name_len_at..old_name_end, invalid_name);
        rewrite_checksum(&mut bytes);
        fs::write(&path, &bytes).expect("write");

        assert!(load(&path).expect("load must not error").is_none());
    }

    #[test]
    fn oversized_declared_path_is_rejected_before_allocation() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");
        save(&sample_index(), &path).expect("save");

        let mut bytes = fs::read(&path).expect("read");
        let root_len_at = ROOT_OFFSET;
        bytes[root_len_at..root_len_at + 4].copy_from_slice(&(MAX_PATH_BYTES + 1).to_le_bytes());
        rewrite_checksum(&mut bytes);
        fs::write(&path, &bytes).expect("write");

        assert!(load(&path).expect("load must not error").is_none());
    }

    #[test]
    fn engine_fingerprint_mismatch_discards_the_snapshot() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");
        save(&sample_index(), &path).expect("save");

        let mut bytes = fs::read(&path).expect("read");
        let fp_at = MAGIC.len() + 4;
        bytes[fp_at] ^= 0xff;
        rewrite_checksum(&mut bytes);
        fs::write(&path, &bytes).expect("write");

        assert!(load(&path).expect("load must not error").is_none());
    }

    #[test]
    fn save_replaces_an_existing_snapshot_and_leaves_no_temp_files() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");

        save(&sample_index(), &path).expect("first save");
        let mut smaller = Index::new("/some/root");
        smaller.apply_ok(&Observation::new(vec![Op::Upsert {
            path: PathBuf::from("only.txt"),
            kind: EntryKind::File,
            attrs: attrs(1, 1),
        }]));
        save(&smaller, &path).expect("second save");

        let restored = load(&path).expect("load").expect("present");
        assert_eq!(restored.total().files, 1);

        let leftovers: Vec<_> = fs::read_dir(dir.path())
            .expect("read_dir")
            .filter_map(std::result::Result::ok)
            .map(|e| e.file_name().to_string_lossy().into_owned())
            .filter(|name| name != "snap.fdu")
            .collect();
        assert!(leftovers.is_empty(), "temp files left behind: {leftovers:?}");
    }

    #[test]
    fn an_abandoned_temporary_is_collected_once_it_is_old_enough() {
        // Per-process entropy means no future writer will ever generate a corpse's
        // name again, so nothing would otherwise reclaim it: unique names turn an
        // occasional collision into permanent litter, one whole snapshot image at a
        // time. The reaper closes that, and the age threshold is what makes it safe
        // without a liveness check — pid-based liveness is unportable and wrong under
        // pid reuse.
        //
        // The threshold is a parameter so both sides are testable without setting
        // mtimes, which the standard library cannot do and which is not worth a
        // dependency.
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");
        let prefix = temp_prefix(&path).expect("a named target has a prefix");

        let mut corpse = prefix.clone();
        corpse.push("111.0123456789abcdef.7");
        let unrelated = OsString::from("notes.txt");
        fs::write(dir.path().join(&corpse), b"a writer killed long ago").expect("plant");
        fs::write(dir.path().join(&unrelated), b"not ours").expect("plant");

        // The shipped threshold spares anything that could still be in flight.
        reap_stale_temporaries(dir.path(), &path, STALE_TEMP_AGE);
        assert!(dir.path().join(&corpse).exists(), "a fresh corpse must be left alone");

        // Past the threshold it is collected, and only it.
        reap_stale_temporaries(dir.path(), &path, std::time::Duration::ZERO);
        assert!(!dir.path().join(&corpse).exists(), "an old corpse must be collected");
        assert!(dir.path().join(&unrelated).exists(), "unrelated files are never touched");
    }

    #[test]
    fn the_reaper_only_matches_its_own_targets_temporaries() {
        // The prefix carries the target's file name, so two snapshots sharing a
        // directory cannot collect each other's work in progress.
        let mine = temp_prefix(Path::new("/cache/snap.fdu")).expect("prefix");
        let theirs = temp_prefix(Path::new("/cache/other.fdu")).expect("prefix");
        assert_eq!(mine, OsString::from(".snap.fdu.tmp."));
        assert_ne!(mine, theirs);
        assert!(temp_prefix(Path::new("/")).is_none(), "a rootless path has no name");
    }

    #[test]
    fn a_stale_temporary_does_not_block_a_later_write() {
        // A killed writer leaves its temporary behind and nothing reaps it until it is
        // a day old, so a later write can find a corpse sitting exactly where it wants
        // to go. `O_EXCL` plus the retry loop is what makes that safe: the collision is
        // detected and stepped over, never shared.
        //
        // The corpse is planted at names this process will really try, via `temp_name`.
        // An earlier version of this test guessed the shape and planted
        // `.snap.fdu.tmp.{pid}.0`, which the entropy in the name means the writer can
        // never generate — so it collided with nothing and proved nothing, while
        // reading as though it had.
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");

        // Block a window of upcoming sequence values. A window rather than one name
        // because the counter is process-global and other tests draw from it too;
        // whichever value this write lands on, it starts inside the blocked range.
        let first = NEXT_TEMP_FILE.load(Ordering::Relaxed);
        let planted: Vec<OsString> =
            (first..first + 32).map(|sequence| temp_name(&path, sequence)).collect();
        for name in &planted {
            fs::write(dir.path().join(name), b"corpse").expect("plant a stale temporary");
        }

        write_atomically(&path, b"payload").expect("write past the stale temporaries");
        assert_eq!(fs::read(&path).expect("read back"), b"payload");

        // Every corpse survives: the writer stepped over them rather than reusing or
        // removing one, and they are far too young for the reaper.
        let mut survivors: Vec<_> = fs::read_dir(dir.path())
            .expect("read_dir")
            .filter_map(std::result::Result::ok)
            .map(|entry| entry.file_name())
            .filter(|name| name != "snap.fdu")
            .collect();
        survivors.sort();
        let mut expected = planted;
        expected.sort();
        assert_eq!(survivors, expected, "the write must step over corpses, not consume them");
    }

    #[test]
    fn an_identical_payload_leaves_the_file_in_place() {
        // The default command encodes an unchanged tree to the bytes already on disk.
        // Replacing them was a full write, an fsync and a rename for nothing; now the
        // file is left alone, which on a filesystem is observable as the same inode.
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");
        let payload: Vec<u8> =
            (0u32..(3 << 20)).map(|i| u8::try_from(i % 251).unwrap_or(0)).collect();

        write_atomically(&path, &payload).expect("first write");
        let first = fs::metadata(&path).expect("metadata");
        let before = std::time::SystemTime::now();
        write_atomically(&path, &payload).expect("identical write");
        let second = fs::metadata(&path).expect("metadata");

        assert_eq!(fs::read(&path).expect("read back"), payload);
        assert_same_file(&first, &second);
        // The "as of" moved even though the bytes did not: the loader reads the mtime
        // as every cached entry's observation time, and the tree was just verified.
        assert!(second.modified().expect("mtime") >= before);
        // No temporary was created and abandoned on the way to deciding not to write.
        let extras = fs::read_dir(dir.path())
            .expect("read_dir")
            .filter_map(std::result::Result::ok)
            .filter(|entry| entry.file_name() != "snap.fdu")
            .count();
        assert_eq!(extras, 0);
    }

    #[test]
    fn a_different_payload_of_the_same_length_is_written() {
        // Length is the cheap pre-check, not the decision: two snapshots of the same
        // tree with one mtime changed are the same length and must not be confused.
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");
        write_atomically(&path, b"payload-a").expect("first write");
        write_atomically(&path, b"payload-b").expect("second write");
        assert_eq!(fs::read(&path).expect("read back"), b"payload-b");

        // And a file that holds the payload as a prefix is not the payload.
        fs::write(&path, b"payload-b-and-more").expect("lengthen");
        assert!(!same_bytes_on_disk(&path, b"payload-b"));
        assert!(!same_bytes_on_disk(&path, b"payload-b-and-mor"));
        assert!(same_bytes_on_disk(&path, b"payload-b-and-more"));
        assert!(!same_bytes_on_disk(&dir.path().join("absent"), b""));
    }

    #[cfg(unix)]
    fn assert_same_file(first: &fs::Metadata, second: &fs::Metadata) {
        use std::os::unix::fs::MetadataExt;
        assert_eq!(first.ino(), second.ino(), "the snapshot was replaced, not left in place");
    }

    #[cfg(not(unix))]
    fn assert_same_file(first: &fs::Metadata, second: &fs::Metadata) {
        // Creation time survives a skipped write and changes across a rename of a fresh
        // temporary, on the platforms that report it.
        if let (Ok(a), Ok(b)) = (first.created(), second.created()) {
            assert_eq!(a, b, "the snapshot was replaced, not left in place");
        }
    }

    #[test]
    fn a_bare_filename_target_resolves_to_an_openable_directory() {
        // `Path::parent` returns `Some("")` for a bare name, not `None`, so an
        // `unwrap_or(".")` fallback never fires. The write still worked — `rename`
        // accepts the empty path — but `read_dir("")` fails, so the reaper returned
        // immediately and collected nothing for those targets.
        assert_eq!(parent_dir(Path::new("snap.fdu")), Path::new("."));
        assert!(fs::read_dir(parent_dir(Path::new("snap.fdu"))).is_ok(), "must be openable");
        assert_eq!(parent_dir(Path::new("/cache/snap.fdu")), Path::new("/cache"));
        assert_eq!(parent_dir(Path::new("cache/snap.fdu")), Path::new("cache"));
    }

    #[test]
    fn a_temporary_name_is_unique_per_sequence_and_carries_process_entropy() {
        // The two components that make a corpse unreachable by a future process, and a
        // writer unable to collide with itself.
        let path = Path::new("/cache/snap.fdu");
        assert_ne!(temp_name(path, 0), temp_name(path, 1), "the counter separates files");
        let name = temp_name(path, 0).to_string_lossy().into_owned();
        assert!(name.starts_with(".snap.fdu.tmp."), "reaper prefix must match: {name}");
        assert!(
            name.contains(&format!("{:016x}", *TEMP_FILE_ENTROPY)),
            "entropy must be in the name, or a recycled pid regenerates a corpse: {name}"
        );
    }

    #[test]
    fn concurrent_atomic_writes_do_not_share_a_temporary_file() {
        use std::sync::{Arc, Barrier};

        const WRITERS: usize = 8;
        const PAYLOAD_BYTES: usize = 1024 * 1024;

        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");
        let barrier = Arc::new(Barrier::new(WRITERS));
        let results = std::thread::scope(|scope| {
            let handles: Vec<_> = (0..WRITERS)
                .map(|writer| {
                    let barrier = Arc::clone(&barrier);
                    let path = path.clone();
                    scope.spawn(move || {
                        let byte = u8::try_from(writer + 1).expect("writer id fits");
                        let bytes = vec![byte; PAYLOAD_BYTES];
                        barrier.wait();
                        write_atomically(&path, &bytes)
                    })
                })
                .collect();
            handles.into_iter().map(std::thread::ScopedJoinHandle::join).collect::<Vec<_>>()
        });

        for result in results {
            result.expect("writer thread did not panic").expect("concurrent atomic write");
        }
        let final_bytes = fs::read(&path).expect("read final image");
        assert_eq!(final_bytes.len(), PAYLOAD_BYTES);
        assert!(final_bytes.iter().all(|byte| *byte == final_bytes[0]));

        let leftovers: Vec<_> = fs::read_dir(dir.path())
            .expect("read_dir")
            .filter_map(std::result::Result::ok)
            .map(|entry| entry.file_name())
            .filter(|name| name != "snap.fdu")
            .collect();
        assert!(leftovers.is_empty(), "temp files left behind: {leftovers:?}");
    }

    #[test]
    fn concurrent_snapshot_reader_sees_only_a_complete_old_or_new_image() {
        use std::sync::{Arc, Barrier};

        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");
        let mut old_index = Index::new("/some/root");
        old_index.apply_ok(&Observation::new(vec![Op::Upsert {
            path: PathBuf::from("old.txt"),
            kind: EntryKind::File,
            attrs: attrs(11, 1),
        }]));
        let mut new_index = Index::new("/some/root");
        new_index.apply_ok(&Observation::new(vec![
            Op::Upsert {
                path: PathBuf::from("new-a.txt"),
                kind: EntryKind::File,
                attrs: attrs(20, 2),
            },
            Op::Upsert {
                path: PathBuf::from("new-b.txt"),
                kind: EntryKind::File,
                attrs: attrs(30, 3),
            },
        ]));
        save(&old_index, &path).expect("save old image");

        let start: Arc<Barrier> = Arc::new(Barrier::new(2));
        let (done_tx, done_rx) = std::sync::mpsc::sync_channel(1);
        std::thread::scope(|scope| {
            let writer_start: Arc<Barrier> = Arc::clone(&start);
            let writer_path: PathBuf = path.clone();
            scope.spawn(move || {
                writer_start.wait();
                done_tx.send(save(&new_index, &writer_path)).expect("report snapshot write");
            });

            let reader_start: Arc<Barrier> = Arc::clone(&start);
            let reader_path: PathBuf = path.clone();
            scope.spawn(move || {
                reader_start.wait();
                let deadline: std::time::Instant =
                    std::time::Instant::now() + std::time::Duration::from_secs(10);
                loop {
                    assert!(
                        std::time::Instant::now() < deadline,
                        "snapshot replacement did not finish before the deadline"
                    );
                    let image: Index =
                        load(&reader_path).expect("load during replacement").expect("image");
                    match image.total().files {
                        1 => {
                            assert_eq!(image.total().bytes, 11);
                            assert!(image.lookup(Path::new("old.txt")).is_some());
                            assert!(image.lookup(Path::new("new-a.txt")).is_none());
                        }
                        2 => {
                            assert_eq!(image.total().bytes, 50);
                            assert!(image.lookup(Path::new("old.txt")).is_none());
                            assert!(image.lookup(Path::new("new-a.txt")).is_some());
                            assert!(image.lookup(Path::new("new-b.txt")).is_some());
                        }
                        partial => panic!("reader observed partial image with {partial} files"),
                    }

                    match done_rx.try_recv() {
                        Ok(result) => {
                            result.expect("replace snapshot");
                            break;
                        }
                        Err(std::sync::mpsc::TryRecvError::Empty) => {}
                        Err(std::sync::mpsc::TryRecvError::Disconnected) => {
                            panic!("snapshot writer disconnected")
                        }
                    }
                }
            });
        });

        let final_image: Index = load(&path).expect("load final").expect("final image");
        assert_eq!(final_image.total().files, 2);
        assert_eq!(final_image.total().bytes, 50);
    }

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

        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");
        save(&sample_index(), &path).expect("save");

        let mode = fs::metadata(&path).expect("metadata").permissions().mode() & 0o777;
        assert_eq!(mode, 0o600);
    }

    #[test]
    fn empty_index_round_trips() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");
        let empty = Index::new("/some/root");

        save(&empty, &path).expect("save");
        let restored = load(&path).expect("load").expect("present");
        assert!(restored.is_empty());
        assert_eq!(restored.total(), empty.total());
    }

    #[test]
    fn partial_index_is_never_persisted() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("partial.fdu");
        save(&Index::new("/root"), &path).expect("complete baseline");
        let complete_bytes = fs::read(&path).expect("read complete snapshot");

        let mut index = Index::new("/root");
        index.set_initial_freshness(false);

        assert!(matches!(save(&index, &path), Err(Error::Snapshot(_))));
        assert_eq!(fs::read(&path).expect("old snapshot remains"), complete_bytes);
    }

    #[test]
    fn semantic_scan_scope_round_trips() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");
        let entries = crate::EntryTierIdentity {
            engine: engine_fingerprint(),
            scope: crate::EntryScope {
                max_depth: Some(7),
                follow_symlinks: false,
                one_filesystem: true,
                hidden_fingerprint: 5,
                exclude_special: true,
                population: crate::query::IgnoredEntries::Include,
                control_fingerprint: 0,
            },
            type_rules_fingerprint: crate::classify::type_rule_fingerprint(),
            reducers_fingerprint: 33,
        };
        // The default control limits, which the table of an index built with
        // `new_with_scope` enforces; any other limits are refused at save.
        for controls in [
            ControlTierIdentity::Observed { limits: crate::control::ControlLimits::default() },
            ControlTierIdentity::NotObserved,
        ] {
            let identity = SnapshotIdentity { entries, controls };
            let index = Index::new_with_scope("/some/root", identity.scan_scope());

            save(&index, &path).expect("save");
            let restored = load(&path).expect("load").expect("present");
            assert_eq!(restored.snapshot_identity(), identity);
            assert_eq!(restored.scope(), identity.scan_scope());
            let info = read_header(&path).expect("read header").expect("current");
            assert_eq!(info.identity, identity);
            assert_eq!(info.scope(), identity.scan_scope());
        }
    }

    /// Reading control files decides which entries are ignored, never which entries exist,
    /// so the snapshots of one tree with observation on and off record one entry tier, byte
    /// for byte, and differ only in the control tier.
    #[test]
    fn controls_on_and_off_snapshots_share_the_entry_identity() {
        let tree = tempfile::tempdir().expect("tree");
        fs::write(tree.path().join(".gitignore"), b"*.log\n").expect("write");
        fs::write(tree.path().join("debug.log"), b"log").expect("write");
        fs::create_dir(tree.path().join("src")).expect("mkdir");
        fs::write(tree.path().join("src/main.rs"), b"fn main() {}\n").expect("write");
        let dir = tempfile::tempdir().expect("tempdir");
        let observed = crate::ScanConfig::default();
        let blind = crate::ScanConfig { read_controls: false, ..observed.clone() };

        let saved = [(&observed, "on.fdu"), (&blind, "off.fdu")].map(|(config, name)| {
            let (index, _) = crate::scan::scan_into_index(tree.path(), config).expect("scan");
            let path = dir.path().join(name);
            save(&index, &path).expect("save");
            let restored = load(&path).expect("load").expect("present");
            assert_eq!(restored.snapshot_identity(), config.snapshot_identity());
            (restored.snapshot_identity(), fs::read(&path).expect("read"))
        });
        let [(on, on_bytes), (off, off_bytes)] = saved;

        assert_eq!(on.entries, off.entries);
        assert_ne!(on.controls, off.controls);
        let entry_tier = IDENTITY_OFFSET..IDENTITY_OFFSET + crate::stored_state::ENTRY_TIER_BYTES;
        let control_tier = entry_tier.end..ROOT_OFFSET;
        assert_eq!(on_bytes[entry_tier.clone()], off_bytes[entry_tier]);
        assert_ne!(on_bytes[control_tier.clone()], off_bytes[control_tier]);

        let projected = load_serving(
            &dir.path().join("on.fdu"),
            blind.types_shared(),
            blind.snapshot_identity(),
        )
        .expect("load projection");
        let LoadOutcome::Served { index: projected, stored } = projected else {
            panic!("an observed snapshot should project to the blind request");
        };
        assert_eq!(serves_snapshot(stored, blind.snapshot_identity()), Serves::ProjectControlsOff);
        let (cold, _) = crate::scan::scan_into_index(tree.path(), &blind).expect("blind scan");
        assert_eq!(projected.snapshot_identity(), blind.snapshot_identity());
        assert_eq!(projected.scope(), blind.scope());
        assert_eq!(projected.len(), cold.len());
        assert_eq!(projected.total(), cold.total());
        assert!(matches!(projected.controls(), Err(Error::ControlStateNotObserved)));
        assert_eq!(
            projected.path_state(Path::new("debug.log")),
            cold.path_state(Path::new("debug.log"))
        );

        let mut corrupt = on_bytes;
        corrupt[WRITING_PASS_STARTED_AT_OFFSET] ^= 1;
        fs::write(dir.path().join("on.fdu"), corrupt).expect("corrupt checksum");
        assert!(matches!(
            load_serving(
                &dir.path().join("on.fdu"),
                blind.types_shared(),
                blind.snapshot_identity(),
            )
            .expect("corrupt projection is absent"),
            LoadOutcome::Absent
        ));
    }

    /// A snapshot written under other `.gitignore` semantics is not served to the default
    /// population, although its tier identities cannot say so: the control tier records only
    /// its limits, from which a loading build recomputes the ignore-rules fingerprint under
    /// its own rules version, and an Include entry tier records no governing control policy.
    /// The engine fingerprint mixes the rules version, and it alone refuses the snapshot.
    #[test]
    fn a_snapshot_under_other_ignore_rules_is_refused_for_the_default_population() {
        use crate::stored_state::IGNORE_RULES_VERSION;

        let tree = tempfile::tempdir().expect("tree");
        fs::write(tree.path().join(".gitignore"), b"*.log\n").expect("write");
        fs::write(tree.path().join("debug.log"), b"log").expect("write");
        let config = crate::ScanConfig::default();
        let wanted = config.snapshot_identity();
        assert_eq!(wanted.entries.scope.population, crate::query::IgnoredEntries::Include);
        assert!(wanted.controls.is_observed(), "the default population reads .gitignore");

        let (index, _) = crate::scan::scan_into_index(tree.path(), &config).expect("scan");
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");
        save(&index, &path).expect("save");
        let current = fs::read(&path).expect("read");
        let serve = || load_serving(&path, config.types_shared(), wanted).expect("load");
        assert!(matches!(serve(), LoadOutcome::Served { .. }), "this build's snapshot serves");

        let engine_at = MAGIC.len() + 4;
        let tiers = IDENTITY_OFFSET..ROOT_OFFSET;
        for version in [IGNORE_RULES_VERSION - 1, IGNORE_RULES_VERSION + 1] {
            let label = format!("rules version {version}");
            let other = engine_fingerprint_under(version);
            assert_ne!(other, engine_fingerprint(), "{label}");
            let mut forged = current.clone();
            forged[engine_at..engine_at + 8].copy_from_slice(&other.to_le_bytes());
            rewrite_checksum(&mut forged);
            fs::write(&path, &forged).expect("write forged");

            // The tier identities are this build's, byte for byte, and would serve the
            // request exactly under this build's engine.
            assert_eq!(forged[tiers.clone()], current[tiers.clone()], "{label}");
            let bytes = forged[tiers.clone()].try_into().expect("the tier identities");
            let stored = SnapshotIdentity::decode(wanted.entries.engine, bytes).expect("decode");
            assert_eq!(serves_snapshot(stored, wanted), Serves::Exact, "{label}");

            assert!(matches!(serve(), LoadOutcome::Absent), "{label}");
            assert!(
                matches!(
                    identify(&path).expect("identify"),
                    Some(Identity::Stale(crate::cache::StaleReason::OtherEngine))
                ),
                "{label}"
            );
            assert_eq!(
                crate::cache::cache_status(&path).expect("status").state,
                crate::cache::CacheState::Stale(crate::cache::StaleReason::OtherEngine),
                "{label}"
            );
        }
    }

    /// A format-4 snapshot, laid out as that format wrote it, is fdu's and stale: the
    /// prologue kept its offsets, so the cache lifecycle names its version rather than
    /// calling it foreign, and loading it is a miss.
    #[test]
    fn a_v4_snapshot_is_older_format() {
        const V4: u32 = 4;
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("v4.fdu");

        let mut image = MAGIC.to_vec();
        image.extend_from_slice(&V4.to_le_bytes());
        // The v4 engine fingerprint. Recognition reads the version before the engine, so a
        // format-4 image is older whatever these eight bytes hold.
        image.extend_from_slice(&0x4444_4444_4444_4444_u64.to_le_bytes());
        image.push(path_encoding());
        // The v4 scope: depth, flags, and the hidden, ignore-rules, type-rules, and reducer
        // fingerprints.
        image.extend_from_slice(&u64::MAX.to_le_bytes());
        image.push(0);
        let scope = crate::ScanConfig::default().scope();
        for fingerprint in [
            scope.hidden_fingerprint,
            scope.ignore_rules_fingerprint,
            scope.type_rules_fingerprint,
            scope.reducers_fingerprint,
        ] {
            image.extend_from_slice(&fingerprint.to_le_bytes());
        }
        put_os_str(&mut image, OsStr::new("/some/root")).expect("root");
        image.extend_from_slice(&1_u64.to_le_bytes());
        image.extend_from_slice(&NO_PARENT.to_le_bytes());
        image.push(EntryKind::Dir as u8);
        put_os_str(&mut image, OsStr::new("")).expect("root name");
        image.extend_from_slice(&[0; 6 * 8]);
        // The v4 control section: no sources, both default limits, no refusals.
        image.extend_from_slice(&0_u32.to_le_bytes());
        for limit in [crate::DEFAULT_CONTROL_BUDGET, crate::DEFAULT_CONTROL_LINE_LIMIT] {
            image.push(1);
            image.extend_from_slice(&u64::try_from(limit).expect("limit").to_le_bytes());
        }
        image.extend_from_slice(&0_u32.to_le_bytes());
        let checksum = crc32c(&image);
        image.extend_from_slice(&checksum.to_le_bytes());
        image.extend_from_slice(TRAILER);
        fs::write(&path, &image).expect("write v4 image");

        assert!(matches!(
            identify(&path).expect("identify"),
            Some(Identity::Stale(crate::cache::StaleReason::OlderFormat { version: V4 }))
        ));
        assert!(read_header(&path).expect("read header").is_none());
        assert!(load(&path).expect("load").is_none());
        assert_eq!(
            crate::cache::cache_status(&path).expect("status").state,
            crate::cache::CacheState::Stale(crate::cache::StaleReason::OlderFormat { version: V4 })
        );
    }

    /// A later pass over an unchanged tree encodes the same facts under a newer pass start.
    /// The image already on disk is kept, its stamp included, rather than rewritten for the
    /// stamp alone, and its mtime moves to the later pass's start, never backwards; a
    /// changed tree is written with the new pass's stamp.
    #[test]
    fn a_later_pass_over_the_same_facts_keeps_the_snapshot_in_place() {
        use std::time::{Duration, UNIX_EPOCH};

        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");
        let stamp_at = |bytes: &[u8]| {
            i64::from_le_bytes(
                bytes[WRITING_PASS_STARTED_AT_OFFSET..IDENTITY_OFFSET].try_into().expect("stamp"),
            )
        };
        let mtime = || fs::metadata(&path).expect("metadata").modified().expect("mtime");
        let nanos = |time: std::time::SystemTime| {
            i64::try_from(time.duration_since(UNIX_EPOCH).expect("after the epoch").as_nanos())
                .expect("nanoseconds")
        };

        let mut first = sample_index();
        first.set_writing_pass_started_at_ns(1_000);
        save(&first, &path).expect("first save");
        let written = fs::metadata(&path).expect("metadata");
        assert_eq!(stamp_at(&fs::read(&path).expect("read")), 1_000);
        assert_eq!(
            load(&path).expect("load").expect("present").writing_pass_started_at_ns(),
            1_000
        );

        // A pass that started after the image was written, on a whole second so every
        // filesystem's mtime holds the instant exactly.
        let later_started = UNIX_EPOCH
            + Duration::from_secs(
                mtime().duration_since(UNIX_EPOCH).expect("after the epoch").as_secs() + 1,
            );
        let mut later = sample_index();
        later.set_writing_pass_started_at_ns(nanos(later_started));
        save(&later, &path).expect("unchanged save");
        let kept = fs::metadata(&path).expect("metadata");
        assert_same_file(&written, &kept);
        assert_eq!(kept.modified().expect("mtime"), later_started, "the kept image's as-of");
        assert_eq!(stamp_at(&fs::read(&path).expect("read")), 1_000);
        assert!(load(&path).expect("load").is_some(), "the kept image is still valid");

        // An equivalent save under an older stamp, as of an index loaded from an earlier
        // image, keeps the later mtime, which was already true of these facts.
        save(&first, &path).expect("older unchanged save");
        assert_same_file(&written, &fs::metadata(&path).expect("metadata"));
        assert_eq!(mtime(), later_started);

        // A kept image must still verify: one whose own checksum is wrong reads as absent,
        // so it is replaced rather than kept under an older stamp forever.
        let mut corrupt = fs::read(&path).expect("read");
        let checksum_at = corrupt.len() - TRAILER.len() - CHECKSUM_BYTES;
        corrupt[checksum_at] ^= 0xff;
        fs::write(&path, &corrupt).expect("corrupt the checksum");
        assert!(load(&path).expect("load").is_none());
        save(&later, &path).expect("save over a corrupt image");
        assert_eq!(stamp_at(&fs::read(&path).expect("read")), nanos(later_started));
        assert!(load(&path).expect("load").is_some());

        let mut changed = sample_index();
        changed.set_writing_pass_started_at_ns(3_000);
        changed.apply_ok(&Observation::new(vec![Op::Upsert {
            path: PathBuf::from("notes.md"),
            kind: EntryKind::File,
            attrs: attrs(8, 30),
        }]));
        save(&changed, &path).expect("changed save");
        let rewritten = fs::read(&path).expect("read");
        assert_eq!(stamp_at(&rewritten), 3_000);
        let restored = load(&path).expect("load").expect("present");
        assert_eq!(restored.writing_pass_started_at_ns(), 3_000);
        assert_eq!(restored.total(), changed.total());
    }

    #[cfg(unix)]
    #[test]
    fn non_utf8_names_round_trip_without_aliasing() {
        use std::os::unix::ffi::OsStringExt;

        let first = PathBuf::from(OsString::from_vec(vec![b'n', 0x80]));
        let second = PathBuf::from(OsString::from_vec(vec![b'n', 0x81]));
        let mut root = PathBuf::from("/some");
        root.push(OsString::from_vec(vec![b'r', 0x82]));
        let mut index = Index::new(&root);
        index.apply_baseline_ok(&Observation::new(vec![
            Op::Upsert { path: first.clone(), kind: EntryKind::File, attrs: attrs(10, 1) },
            Op::Upsert { path: second.clone(), kind: EntryKind::File, attrs: attrs(20, 2) },
        ]));

        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("snap.fdu");
        save(&index, &path).expect("save");
        let restored = load(&path).expect("load").expect("present");

        assert_eq!(restored.root_path(), root);
        assert_eq!(restored.total().files, 2);
        assert_eq!(restored.total().bytes, 30);
        assert!(restored.lookup(&first).is_some());
        assert!(restored.lookup(&second).is_some());
        assert!(!restored.serving_indexes_enabled());
    }
}