//! HNSW index persistence (save/load)

use std::cell::UnsafeCell;
use std::path::Path;
use std::sync::atomic::{AtomicBool, Ordering};

use hnsw_rs::hnswio::HnswIo;

use crate::index::{DistanceMetric, VectorIndex};

use super::{HnswError, HnswGraph, HnswIndex, HnswInner, HnswIoCell, LoadedHnsw};

/// Configurable HNSW graph file size limit via `CQS_HNSW_MAX_GRAPH_BYTES`
/// env var. Defaults to 500MB. Cached in OnceLock for single parse.
/// Parse/warn/default goes through the shared `crate::limits::parse_env_u64`.
fn hnsw_max_graph_bytes() -> u64 {
    static MAX: std::sync::OnceLock<u64> = std::sync::OnceLock::new();
    *MAX.get_or_init(|| crate::limits::parse_env_u64("CQS_HNSW_MAX_GRAPH_BYTES", 500 * 1024 * 1024))
}

/// Configurable HNSW data file size limit via `CQS_HNSW_MAX_DATA_BYTES`
/// env var. Defaults to 1GB. Cached in OnceLock for single parse.
/// Parse/warn/default goes through the shared `crate::limits::parse_env_u64`.
fn hnsw_max_data_bytes() -> u64 {
    static MAX: std::sync::OnceLock<u64> = std::sync::OnceLock::new();
    *MAX.get_or_init(|| crate::limits::parse_env_u64("CQS_HNSW_MAX_DATA_BYTES", 1024 * 1024 * 1024))
}

/// Configurable HNSW ID map file size limit via `CQS_HNSW_MAX_ID_MAP_BYTES`
/// env var. Defaults to 500MB. Cached in OnceLock for single parse.
/// Parse/warn/default goes through the shared `crate::limits::parse_env_u64`.
fn hnsw_max_id_map_bytes() -> u64 {
    static MAX: std::sync::OnceLock<u64> = std::sync::OnceLock::new();
    *MAX.get_or_init(|| {
        crate::limits::parse_env_u64("CQS_HNSW_MAX_ID_MAP_BYTES", 500 * 1024 * 1024)
    })
}

/// Whether the WSL advisory locking warning has been emitted (once per process)
static WSL_LOCK_WARNED: AtomicBool = AtomicBool::new(false);

/// Emit a one-time warning about advisory-only file locking on WSL/NTFS mounts.
/// Delegates the `/mnt/<letter>/` detection to `config::is_wsl_drvfs_path`.
fn warn_wsl_advisory_locking(dir: &Path) {
    if crate::config::is_wsl()
        && crate::config::is_wsl_drvfs_path(dir)
        && !WSL_LOCK_WARNED.swap(true, Ordering::Relaxed)
    {
        tracing::warn!(
            "HNSW file locking is advisory-only on WSL/NTFS — avoid concurrent index operations"
        );
    }
}

/// Core HNSW file extensions (graph, data, IDs, meta header). These are the
/// extensions allowed in the checksum manifest. `hnsw.meta` is absent from
/// legacy (pre-#1351) indexes — the load path treats a missing meta as
/// cosine, which is what every legacy index is.
const HNSW_EXTENSIONS: &[&str] = &["hnsw.graph", "hnsw.data", "hnsw.ids", "hnsw.meta"];

/// All HNSW file extensions including checksum (for cleanup/deletion).
/// NOTE: Keep in sync with HNSW_EXTENSIONS above — first 4 elements must match.
pub const HNSW_ALL_EXTENSIONS: &[&str] = &[
    "hnsw.graph",
    "hnsw.data",
    "hnsw.ids",
    "hnsw.meta",
    "hnsw.checksum",
    "hnsw.lock",
];

/// Magic string for the `{basename}.hnsw.meta` JSON header. A
/// mismatch is a typed load error, not a silent default.
const HNSW_META_MAGIC: &str = "CQSHNSW";

/// Version of the `.hnsw.meta` header. Bump when adding fields an older
/// binary must not misinterpret; the load path rejects versions newer than
/// this binary understands.
const HNSW_META_VERSION: u32 = 1;

/// JSON header persisted next to the HNSW graph/data/ids files.
///
/// Carries index-level configuration that the raw hnsw_rs dump does not
/// expose up front — the distance metric, which the load path
/// needs *before* it can instantiate the right `Hnsw<f32, D>` type
/// parameter — plus the store-state stamp the sidecar contents reflect.
/// Mirrors the CAGRA `.cagra.meta` sidecar pattern (`src/cagra.rs`).
#[derive(serde::Serialize, serde::Deserialize, Debug, PartialEq, Eq)]
struct HnswMeta {
    /// Format magic. See [`HNSW_META_MAGIC`].
    magic: String,
    /// Header schema version. See [`HNSW_META_VERSION`].
    version: u32,
    /// [`DistanceMetric::as_str`] value the index was built with.
    metric: String,
    /// `Store::chunk_count()` of the store state this index was built from.
    /// `None` on sidecars written before the stamp existed; clean (non-dirty)
    /// loads of such sidecars keep working, but the dirty-flag self-heal
    /// treats a missing stamp as a mismatch — a stamp-less sidecar cannot
    /// prove it postdates the last chunk write.
    #[serde(default)]
    chunk_count: Option<u64>,
    /// `Store::splade_generation()` of the store state this index was built
    /// from. The counter is bumped by every sparse-vector write batch, by
    /// the chunk-delete trigger, and by dense-embedding rewrites
    /// (`update_embeddings_with_hashes_batch`, the enrichment pass), so it
    /// advances with every vector-affecting write the index pipeline
    /// performs. Same `None` semantics as `chunk_count`.
    #[serde(default)]
    splade_generation: Option<u64>,
}

impl HnswMeta {
    /// The store-state stamp recorded at save, if this sidecar has one.
    /// Both fields are written together; a half-present pair (hand-edited
    /// meta) counts as no stamp.
    fn store_stamp(&self) -> Option<StoreStamp> {
        match (self.chunk_count, self.splade_generation) {
            (Some(chunk_count), Some(splade_generation)) => Some(StoreStamp {
                chunk_count,
                splade_generation,
            }),
            _ => None,
        }
    }
}

/// Store state an HNSW sidecar set reflects, captured at save time.
///
/// The pair plays the same role CAGRA's `chunk_count` + `splade_generation`
/// sidecar fields play (`src/cagra.rs`): a cheap discriminator that lets the
/// load path ask "does this index reflect the live store?" instead of only
/// "are these files internally consistent?". `splade_generation` advances on
/// every sparse-vector write batch, on every chunk delete (SQLite trigger),
/// and on every dense-embedding rewrite
/// (`update_embeddings_with_hashes_batch` — the enrichment pass — bumps it
/// explicitly so enrichment-only reindexes move the stamp); `chunk_count`
/// catches insert-only paths that might not touch sparse rows. Every
/// vector-affecting write path therefore moves the stamp.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct StoreStamp {
    /// `Store::chunk_count()` at capture time.
    pub chunk_count: u64,
    /// `Store::splade_generation()` at capture time.
    pub splade_generation: u64,
}

impl StoreStamp {
    /// Capture the current stamp from a live store.
    pub fn read<Mode>(store: &crate::store::Store<Mode>) -> Result<Self, crate::store::StoreError> {
        Ok(Self {
            chunk_count: store.chunk_count()?,
            splade_generation: store.splade_generation()?,
        })
    }
}

/// What a stamped save actually did.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum SaveOutcome {
    /// The sidecar set on disk now reflects this index.
    Saved,
    /// The save was discarded under the exclusive lock: the live store moved
    /// past the snapshot this index was built from, so persisting it would
    /// overwrite a newer index with stale contents. Nothing on disk changed.
    /// Callers must NOT clear the `hnsw_dirty` flag on this outcome.
    DiscardedStale,
}

/// Stamp context for [`HnswIndex::save_stamped`]: the snapshot the index
/// contents reflect, plus a re-reader invoked under the exclusive save lock.
struct SaveStamp<'a> {
    /// Re-reads the live store stamp. Called only while the exclusive save
    /// lock is held, so the comparison against `snapshot` cannot race a
    /// concurrent in-process saver.
    live: &'a dyn Fn() -> Result<StoreStamp, HnswError>,
    /// Store state the index contents were built from (including any delta
    /// replayed on top of the build snapshot).
    snapshot: StoreStamp,
}

/// Read and validate the `{basename}.hnsw.meta` header, if present.
///
/// - Missing meta file → `Ok(None)` (legacy index predating the meta header).
/// - Present but malformed / unknown magic / newer version → typed error.
fn read_hnsw_meta(dir: &Path, basename: &str) -> Result<Option<HnswMeta>, HnswError> {
    let meta_path = dir.join(format!("{}.hnsw.meta", basename));
    if !meta_path.exists() {
        return Ok(None);
    }

    // Size cap: the header is a 3-field JSON object; anything beyond a few
    // KB is corruption or tampering.
    const MAX_META_SIZE: u64 = 64 * 1024;
    let meta_size = std::fs::metadata(&meta_path)
        .map_err(|e| {
            HnswError::Internal(format!(
                "Failed to stat HNSW meta {}: {}",
                meta_path.display(),
                e
            ))
        })?
        .len();
    if meta_size > MAX_META_SIZE {
        return Err(HnswError::Internal(format!(
            "HNSW meta {} is {} bytes, exceeds {} byte limit",
            meta_path.display(),
            meta_size,
            MAX_META_SIZE
        )));
    }

    let raw = std::fs::read(&meta_path).map_err(|e| {
        HnswError::Internal(format!(
            "Failed to read HNSW meta {}: {}",
            meta_path.display(),
            e
        ))
    })?;
    let meta: HnswMeta = serde_json::from_slice(&raw).map_err(|e| {
        HnswError::Internal(format!(
            "Failed to parse HNSW meta {}: {}",
            meta_path.display(),
            e
        ))
    })?;

    if meta.magic != HNSW_META_MAGIC {
        return Err(HnswError::Internal(format!(
            "HNSW meta {} has unexpected magic {:?} (want {:?})",
            meta_path.display(),
            meta.magic,
            HNSW_META_MAGIC
        )));
    }
    if meta.version > HNSW_META_VERSION {
        return Err(HnswError::Internal(format!(
            "HNSW meta {} has version {} — newer than this binary understands ({}). \
             Run 'cqs index --force' or upgrade cqs.",
            meta_path.display(),
            meta.version,
            HNSW_META_VERSION
        )));
    }

    Ok(Some(meta))
}

/// Read the persisted distance metric for an index, if any.
///
/// - Missing meta file → `Ok(Cosine)`: every index predating the meta header
///   was built with `DistCosine` by construction.
/// - Present but malformed / unknown magic / newer version / unknown metric
///   → typed error. Never a silent cosine fallback: misreading a dot-product
///   index as cosine produces wrong distances, not a crash.
fn read_hnsw_meta_metric(dir: &Path, basename: &str) -> Result<DistanceMetric, HnswError> {
    let Some(meta) = read_hnsw_meta(dir, basename)? else {
        tracing::debug!(
            dir = %dir.display(),
            basename,
            "No HNSW meta header (legacy index) — metric is cosine"
        );
        return Ok(DistanceMetric::Cosine);
    };
    meta.metric.parse::<DistanceMetric>().map_err(|e| {
        HnswError::Internal(format!(
            "HNSW meta for {}/{} has invalid metric: {}",
            dir.display(),
            basename,
            e
        ))
    })
}

/// Verify that a persisted HNSW sidecar set is both intact AND reflects the
/// live store. This is the dirty-flag self-heal check: the `hnsw_dirty` flag
/// is set before chunk writes and cleared after a successful HNSW save, so a
/// set flag means "chunks may have been committed after the last save". A
/// checksum pass alone cannot refute that — the manifest is written by the
/// same save it describes, so a complete previous-generation set always
/// verifies. The store-state stamp is what distinguishes "died after the
/// save landed" (heal) from "died after the chunk commit, before the save"
/// (stay dirty, brute-force until rebuild).
///
/// A sidecar without a stamp (written before the stamp existed) fails this
/// check: it cannot prove it postdates the last chunk write. Clean
/// (non-dirty) loads of such sidecars are unaffected — this function is only
/// on the dirty path.
pub fn verify_hnsw_current<Mode>(
    dir: &Path,
    basename: &str,
    store: &crate::store::Store<Mode>,
) -> Result<(), HnswError> {
    verify_hnsw_checksums(dir, basename)?;
    let live = StoreStamp::read(store).map_err(|e| {
        HnswError::Internal(format!(
            "Failed to read live store stamp for HNSW staleness check: {}",
            e
        ))
    })?;
    let stamped = read_hnsw_meta(dir, basename)?.and_then(|m| m.store_stamp());
    match stamped {
        Some(s) if s == live => Ok(()),
        Some(s) => Err(HnswError::Internal(format!(
            "HNSW sidecar stamp (chunks={}, generation={}) does not match the live store \
             (chunks={}, generation={}) — the index predates the last chunk write",
            s.chunk_count, s.splade_generation, live.chunk_count, live.splade_generation
        ))),
        None => Err(HnswError::Internal(
            "HNSW sidecar has no store-state stamp — cannot prove the index postdates \
             the last chunk write"
                .to_string(),
        )),
    }
}

/// Verify HNSW index file checksums using blake3.
///
/// # Security Model
///
/// **WARNING:** These checksums detect accidental corruption only (disk errors,
/// incomplete writes). They do NOT provide tamper-detection or authenticity
/// guarantees - an attacker with filesystem access can update both files and
/// checksums. For tamper-proofing, the checksum file would need to be signed
/// or stored separately in a trusted location.
///
/// Returns Ok if checksums match or no checksum file exists (with warning).
pub fn verify_hnsw_checksums(dir: &Path, basename: &str) -> Result<(), HnswError> {
    // Entry span so operators can see this on the startup path. Hashing
    // potentially hundreds of MB of HNSW manifests is the kind of thing that
    // ends up at the top of "why is `cqs index` slow on cold start"
    // investigations; an untimed pass is opaque.
    let started = std::time::Instant::now();
    let _span = tracing::info_span!(
        "verify_hnsw_checksums",
        dir = %dir.display(),
        basename
    )
    .entered();
    let checksum_path = dir.join(format!("{}.hnsw.checksum", basename));

    if !checksum_path.exists() {
        return Err(HnswError::Internal(
            "No checksum file for HNSW index — run 'cqs index --force' to regenerate".to_string(),
        ));
    }

    let checksum_content = std::fs::read_to_string(&checksum_path).map_err(|e| {
        tracing::warn!(
            error = %e,
            path = %checksum_path.display(),
            kind = ?e.kind(),
            "verify_hnsw_checksums IO failure"
        );
        HnswError::Internal(format!("Failed to read {}: {}", checksum_path.display(), e))
    })?;
    for line in checksum_content.lines() {
        if let Some((ext, expected)) = line.split_once(':') {
            // Only allow known extensions to prevent path traversal
            if !HNSW_EXTENSIONS.contains(&ext) {
                tracing::warn!(ext = %ext, "Ignoring unknown extension in checksum file");
                continue;
            }
            let path = dir.join(format!("{}.{}", basename, ext));
            // A missing file referenced by the checksum manifest is a hard
            // failure, not a skip. This catches the partial-save case: a
            // crash between writing the manifest and renaming the graph/data
            // files into place would otherwise leave the verifier returning
            // Ok, then the load path falling into `hnsw_rs::file_load`
            // against a missing file. The `HNSW_EXTENSIONS` whitelist at the
            // top of the loop already gates which paths can reach this
            // branch, so the strict check is safe.
            if !path.exists() {
                return Err(HnswError::Internal(format!(
                    "HNSW checksum manifest references missing file: {} (partial save?)",
                    path.display()
                )));
            }
            // Stream file through blake3 hasher to avoid loading entire file into memory
            let file = std::fs::File::open(&path).map_err(|e| {
                tracing::warn!(
                    error = %e,
                    path = %path.display(),
                    kind = ?e.kind(),
                    "verify_hnsw_checksums IO failure"
                );
                HnswError::Internal(format!(
                    "Failed to open {} for checksum: {}",
                    path.display(),
                    e
                ))
            })?;
            let mut hasher = blake3::Hasher::new();
            std::io::copy(&mut std::io::BufReader::new(file), &mut hasher).map_err(|e| {
                tracing::warn!(
                    error = %e,
                    path = %path.display(),
                    kind = ?e.kind(),
                    "verify_hnsw_checksums IO failure"
                );
                HnswError::Internal(format!(
                    "Failed to read {} for checksum: {}",
                    path.display(),
                    e
                ))
            })?;
            let actual = hasher.finalize().to_hex().to_string();
            if actual != expected {
                return Err(HnswError::ChecksumMismatch {
                    file: path.display().to_string(),
                    expected: expected.to_string(),
                    actual,
                });
            }
        }
    }
    tracing::info!(
        elapsed_ms = started.elapsed().as_millis() as u64,
        "HNSW checksums verified"
    );
    Ok(())
}

/// Test-only fault injection seams for the save/promote path. Thread-local
/// so parallel tests can't observe each other's injected faults (saves run
/// synchronously on the calling thread).
#[cfg(test)]
pub(crate) mod test_hooks {
    use std::cell::RefCell;

    thread_local! {
        /// When set, `save()` deletes the staged file with this extension
        /// (e.g. `"hnsw.ids"`) right before the promote pass, simulating a
        /// staging directory that was tampered with mid-save.
        pub(crate) static DELETE_STAGED_EXT_BEFORE_PROMOTE: RefCell<Option<String>> =
            const { RefCell::new(None) };
    }
}

impl HnswIndex {
    /// Save the index to disk
    ///
    /// Creates files in the directory:
    /// - `{basename}.hnsw.data` - Vector data
    /// - `{basename}.hnsw.graph` - HNSW graph structure
    /// - `{basename}.hnsw.ids` - Chunk ID mapping
    /// - `{basename}.hnsw.meta` - JSON header (distance metric, #1351)
    /// - `{basename}.hnsw.checksum` - Blake3 checksums for integrity
    ///
    /// # Crash safety
    /// The ID map and checksum files are written atomically (write-to-temp, then rename).
    /// The checksum file is written last, so if the process crashes during save:
    /// - If checksum is missing/incomplete, load() will fail verification
    /// - If graph/data are incomplete, load() will fail checksum verification
    ///
    /// Note: The underlying library writes graph/data non-atomically. However, the
    /// checksum verification on load ensures we never use a corrupted index.
    ///
    /// This variant writes no store-state stamp into the meta header, so the
    /// dirty-flag self-heal will refuse the resulting sidecars (see
    /// [`verify_hnsw_current`]). Production callers that have a `Store` in
    /// scope must use [`save_stamped`](Self::save_stamped); this remains for
    /// tests and store-less callers.
    pub fn save(&self, dir: &Path, basename: &str) -> Result<(), HnswError> {
        self.save_impl(dir, basename, None).map(|_| ())
    }

    /// Save with a store-state stamp, discarding the save if the store has
    /// moved past `snapshot`.
    ///
    /// `snapshot` is the [`StoreStamp`] of the store state this index's
    /// contents reflect (the build snapshot, plus any delta replayed on top).
    /// Under the exclusive save lock the live stamp is re-read and compared:
    /// a mismatch means a writer this index never observed committed chunks
    /// after the snapshot — persisting would overwrite a newer on-disk index
    /// with stale contents while stamping it current. The save is discarded
    /// instead ([`SaveOutcome::DiscardedStale`]), and even if a stale save
    /// raced in through the remaining window between the caller's snapshot
    /// capture and this lock, its stamp would be the old generation — the
    /// dirty-path check ([`verify_hnsw_current`]) refuses it at next load.
    pub fn save_stamped<Mode>(
        &self,
        dir: &Path,
        basename: &str,
        store: &crate::store::Store<Mode>,
        snapshot: StoreStamp,
    ) -> Result<SaveOutcome, HnswError> {
        let live = || {
            StoreStamp::read(store).map_err(|e| {
                HnswError::Internal(format!(
                    "Failed to re-read store stamp during HNSW save: {}",
                    e
                ))
            })
        };
        self.save_impl(
            dir,
            basename,
            Some(SaveStamp {
                live: &live,
                snapshot,
            }),
        )
    }

    /// Shared save body. See [`save`](Self::save) for the file layout and
    /// crash-safety contract, [`save_stamped`](Self::save_stamped) for the
    /// stamp semantics.
    fn save_impl(
        &self,
        dir: &Path,
        basename: &str,
        stamp: Option<SaveStamp<'_>>,
    ) -> Result<SaveOutcome, HnswError> {
        let _span = tracing::debug_span!("hnsw_save", dir = %dir.display(), basename).entered();
        tracing::info!(dir = %dir.display(), basename, "Saving HNSW index");

        // Verify ID map matches HNSW vector count before saving
        let hnsw_count = self.inner.with_hnsw(|h| h.get_nb_point());
        if hnsw_count != self.id_map.len() {
            return Err(HnswError::Internal(format!(
                "HNSW/ID map count mismatch on save: HNSW has {} vectors but id_map has {}. This is a bug.",
                hnsw_count,
                self.id_map.len()
            )));
        }

        // Ensure target directory exists
        std::fs::create_dir_all(dir).map_err(|e| {
            HnswError::Internal(format!(
                "Failed to create directory {}: {}",
                dir.display(),
                e
            ))
        })?;

        // Acquire exclusive lock for save. This happens BEFORE the stale-.bak
        // guard below: a concurrent saver's transient .baks (it renames the
        // live files to .bak between its backup loop and promote pass) must
        // not be observable here. Once the exclusive lock is held, any .bak
        // we see is from a save that already terminated.
        // NOTE: File locking is advisory only on WSL over 9P.
        // This prevents concurrent cqs processes from corrupting the index,
        // but cannot protect against external Windows process modifications.
        let lock_path = dir.join(format!("{}.hnsw.lock", basename));
        #[allow(clippy::suspicious_open_options)] // Intentional: create if missing, don't truncate
        let lock_file = std::fs::OpenOptions::new()
            .create(true)
            .write(true)
            .open(&lock_path)?;
        lock_file.lock().map_err(HnswError::Io)?;
        warn_wsl_advisory_locking(dir);
        tracing::debug!(lock_path = %lock_path.display(), "Acquired HNSW save lock");

        // Live store stamp, read once under the exclusive lock. Used by both
        // the debris check below and the staleness discard after it.
        let live_stamp: Option<StoreStamp> = match &stamp {
            Some(s) => Some((s.live)()?),
            None => None,
        };

        let all_exts = [
            "hnsw.graph",
            "hnsw.data",
            "hnsw.ids",
            "hnsw.meta",
            "hnsw.checksum",
        ];
        let stale_baks: Vec<std::path::PathBuf> = all_exts
            .iter()
            .filter_map(|ext| {
                let bak = dir.join(format!("{}.{}.bak", basename, ext));
                if bak.exists() {
                    Some(bak)
                } else {
                    None
                }
            })
            .collect();
        if !stale_baks.is_empty() {
            // .bak files exist from a save that didn't finish its cleanup.
            // Two distinct states:
            //
            // - Post-success debris: the previous save promoted its complete
            //   new set and was killed before deleting the .baks. A partial
            //   promote cannot pass `verify_hnsw_checksums` (the manifest is
            //   promoted last and references the other files' new hashes), so
            //   a live set that verifies AND carries a store-state stamp from
            //   this store's history is a completed save. The .baks hold the
            //   generation BEFORE it — redundant. Delete them and proceed.
            //
            // - Recovery breadcrumbs: the live set is absent, incomplete, or
            //   fails verification — the .baks may be the only good copy of
            //   the index. Refuse so the backup loop below can't overwrite
            //   them.
            //
            // The debris check deliberately does NOT require the live set's
            // stamp to equal the CURRENT store stamp: writes that landed
            // between the kill and this save would re-wedge persistence
            // (the exact lockout this branch fixes). Staleness of the live
            // set is the dirty-path load check's job, not this guard's; the
            // stamp here only has to prove the set came from a completed,
            // stamp-aware save of this store (generation not in the future).
            // An ahead-of-store generation also occurs after restoring
            // index.db from a backup (the restore rolls the counter back
            // under sidecars saved later); the refusal is conservative
            // there — remove the .baks or run `cqs index --force` to
            // realign sidecars with the restored store.
            let debris_reason: Result<(), String> = (|| {
                verify_hnsw_checksums(dir, basename)
                    .map_err(|e| format!("live set failed verification: {}", e))?;
                let set_stamp = read_hnsw_meta(dir, basename)
                    .map_err(|e| format!("live set meta unreadable: {}", e))?
                    .and_then(|m| m.store_stamp())
                    .ok_or_else(|| "live set has no store-state stamp".to_string())?;
                if let Some(live) = live_stamp {
                    if set_stamp.splade_generation > live.splade_generation {
                        return Err(format!(
                            "live set stamp generation {} is ahead of the store ({}) — \
                             not from this store's history",
                            set_stamp.splade_generation, live.splade_generation
                        ));
                    }
                } else {
                    return Err("no store available to validate the live set stamp".to_string());
                }
                Ok(())
            })();
            match debris_reason {
                Ok(()) => {
                    tracing::warn!(
                        baks = ?stale_baks,
                        "Removing post-success .bak debris: live HNSW set is complete, \
                         checksum-valid, and stamped — the .baks are the superseded generation"
                    );
                    for bak in &stale_baks {
                        if let Err(e) = std::fs::remove_file(bak) {
                            return Err(HnswError::Internal(format!(
                                "Failed to remove .bak debris {}: {}",
                                bak.display(),
                                e
                            )));
                        }
                    }
                }
                Err(reason) => {
                    tracing::warn!(
                        baks = ?stale_baks,
                        reason = %reason,
                        "Refusing HNSW save: .bak files present and live set is not a \
                         verified completed save"
                    );
                    return Err(HnswError::Internal(format!(
                        "stale .bak files from prior failed save: {:?} ({}); manual recovery \
                         required (the .baks are the previous good index)",
                        stale_baks, reason,
                    )));
                }
            }
        }

        // Staleness discard: if the store moved past the snapshot this index
        // was built from, persisting would replace a newer on-disk index with
        // stale contents stamped as current. Nothing has been written yet, so
        // discarding here is side-effect-free.
        if let (Some(s), Some(live)) = (&stamp, live_stamp) {
            if live != s.snapshot {
                tracing::warn!(
                    basename,
                    snapshot_chunks = s.snapshot.chunk_count,
                    snapshot_generation = s.snapshot.splade_generation,
                    live_chunks = live.chunk_count,
                    live_generation = live.splade_generation,
                    "Discarding HNSW save: store moved past the build snapshot — \
                     an unobserved writer committed chunks; the on-disk index is \
                     left untouched and the dirty flag must stay set"
                );
                return Ok(SaveOutcome::DiscardedStale);
            }
        }

        // Use a temporary directory for atomic writes so that a crash
        // mid-save leaves the old index intact. Unpredictable suffix
        // prevents symlink TOCTOU.
        let suffix = crate::temp_suffix();
        let temp_dir = dir.join(format!(".{}.{:016x}.tmp", basename, suffix));
        if temp_dir.exists() {
            std::fs::remove_dir_all(&temp_dir).map_err(|e| {
                HnswError::Internal(format!(
                    "Failed to clean up temp dir {}: {}",
                    temp_dir.display(),
                    e
                ))
            })?;
        }
        std::fs::create_dir_all(&temp_dir).map_err(|e| {
            HnswError::Internal(format!(
                "Failed to create temp dir {}: {}",
                temp_dir.display(),
                e
            ))
        })?;

        // Save the HNSW graph and data to temp directory
        self.inner
            .with_hnsw(|g| g.file_dump(&temp_dir, basename))
            .map_err(|e| {
                HnswError::Internal(format!(
                    "Failed to dump HNSW to {}/{}: {}",
                    temp_dir.display(),
                    basename,
                    e
                ))
            })?;

        // Write the meta header (distance metric) to the temp directory.
        // Small JSON; written with mode 0600 + fsync like the id map so a
        // power cut can't promote a checksum manifest that references a
        // page-cache-only meta file.
        let meta_temp = temp_dir.join(format!("{}.hnsw.meta", basename));
        {
            let meta = HnswMeta {
                magic: HNSW_META_MAGIC.to_string(),
                version: HNSW_META_VERSION,
                metric: self.metric().as_str().to_string(),
                chunk_count: stamp.as_ref().map(|s| s.snapshot.chunk_count),
                splade_generation: stamp.as_ref().map(|s| s.snapshot.splade_generation),
            };
            let meta_json = serde_json::to_vec(&meta).map_err(|e| {
                HnswError::Internal(format!("Failed to serialize HNSW meta: {}", e))
            })?;
            let mut file = {
                #[cfg(unix)]
                {
                    use std::os::unix::fs::OpenOptionsExt;
                    std::fs::OpenOptions::new()
                        .write(true)
                        .create(true)
                        .truncate(true)
                        .mode(0o600)
                        .open(&meta_temp)
                }
                #[cfg(not(unix))]
                {
                    std::fs::File::create(&meta_temp)
                }
            }
            .map_err(|e| {
                HnswError::Internal(format!("Failed to create {}: {}", meta_temp.display(), e))
            })?;
            use std::io::Write;
            file.write_all(&meta_json).map_err(|e| {
                HnswError::Internal(format!("Failed to write {}: {}", meta_temp.display(), e))
            })?;
            file.sync_all().map_err(|e| {
                HnswError::Internal(format!("Failed to fsync {}: {}", meta_temp.display(), e))
            })?;
        }

        // Stream ID map directly to file via BufWriter rather than
        // serializing to an in-memory JSON string first.
        let id_map_temp = temp_dir.join(format!("{}.hnsw.ids", basename));
        {
            // Create with mode 0o600 so file is never world-readable
            let file = {
                #[cfg(unix)]
                {
                    use std::os::unix::fs::OpenOptionsExt;
                    std::fs::OpenOptions::new()
                        .write(true)
                        .create(true)
                        .truncate(true)
                        .mode(0o600)
                        .open(&id_map_temp)
                }
                #[cfg(not(unix))]
                {
                    std::fs::File::create(&id_map_temp)
                }
            }
            .map_err(|e| {
                HnswError::Internal(format!("Failed to create {}: {}", id_map_temp.display(), e))
            })?;
            let mut writer = std::io::BufWriter::new(file);
            serde_json::to_writer(&mut writer, &self.id_map)
                .map_err(|e| HnswError::Internal(format!("Failed to serialize ID map: {}", e)))?;
            // Flush the BufWriter and fsync the underlying file before it is
            // dropped so the id_map bytes are durable on disk. Without this
            // the id_map could survive in the page cache through the
            // subsequent rename and get lost on a power cut, leaving the
            // graph without any string IDs to look up.
            use std::io::Write;
            writer.flush().map_err(|e| {
                HnswError::Internal(format!("Failed to flush {}: {}", id_map_temp.display(), e))
            })?;
            writer.get_ref().sync_all().map_err(|e| {
                HnswError::Internal(format!("Failed to fsync {}: {}", id_map_temp.display(), e))
            })?;
        }

        // Compute checksum by reading back the file (avoids holding JSON in memory)
        let ids_hash = {
            let file = std::fs::File::open(&id_map_temp).map_err(|e| {
                HnswError::Internal(format!(
                    "Failed to open {} for checksum: {}",
                    id_map_temp.display(),
                    e
                ))
            })?;
            let mut hasher = blake3::Hasher::new();
            hasher.update_reader(file).map_err(|e| {
                HnswError::Internal(format!(
                    "Failed to read {} for checksum: {}",
                    id_map_temp.display(),
                    e
                ))
            })?;
            hasher.finalize()
        };
        let mut checksums = vec![format!("hnsw.ids:{}", ids_hash.to_hex())];
        for ext in &["hnsw.graph", "hnsw.data", "hnsw.meta"] {
            let path = temp_dir.join(format!("{}.{}", basename, ext));
            if path.exists() {
                let file = std::fs::File::open(&path).map_err(|e| {
                    HnswError::Internal(format!(
                        "Failed to open {} for checksum: {}",
                        path.display(),
                        e
                    ))
                })?;
                let mut hasher = blake3::Hasher::new();
                hasher.update_reader(file).map_err(|e| {
                    HnswError::Internal(format!(
                        "Failed to read {} for checksum: {}",
                        path.display(),
                        e
                    ))
                })?;
                let hash = hasher.finalize();
                checksums.push(format!("{}:{}", ext, hash.to_hex()));
            }
        }

        // Write checksum with mode 0o600 from creation
        let checksum_temp = temp_dir.join(format!("{}.hnsw.checksum", basename));
        {
            #[cfg(unix)]
            {
                use std::io::Write;
                use std::os::unix::fs::OpenOptionsExt;
                let mut f = std::fs::OpenOptions::new()
                    .write(true)
                    .create(true)
                    .truncate(true)
                    .mode(0o600)
                    .open(&checksum_temp)
                    .map_err(|e| {
                        HnswError::Internal(format!(
                            "Failed to write {}: {}",
                            checksum_temp.display(),
                            e
                        ))
                    })?;
                f.write_all(checksums.join("\n").as_bytes()).map_err(|e| {
                    HnswError::Internal(format!(
                        "Failed to write {}: {}",
                        checksum_temp.display(),
                        e
                    ))
                })?;
            }
            #[cfg(not(unix))]
            {
                std::fs::write(&checksum_temp, checksums.join("\n")).map_err(|e| {
                    HnswError::Internal(format!(
                        "Failed to write {}: {}",
                        checksum_temp.display(),
                        e
                    ))
                })?;
            }
        }

        // Set restrictive permissions on remaining temp files (graph/data written by library)
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let restrictive = std::fs::Permissions::from_mode(0o600);
            for ext in &["hnsw.graph", "hnsw.data"] {
                let path = temp_dir.join(format!("{}.{}", basename, ext));
                if path.exists() {
                    if let Err(e) = std::fs::set_permissions(&path, restrictive.clone()) {
                        tracing::debug!(path = %path.display(), error = %e, "Failed to set HNSW file permissions");
                    }
                }
            }
        }

        // Test-only fault injection: delete one staged file just before the
        // promote pass so tests can pin the missing-temp hard-error path.
        #[cfg(test)]
        test_hooks::DELETE_STAGED_EXT_BEFORE_PROMOTE.with(|hook| {
            if let Some(ext) = hook.borrow_mut().take() {
                let p = temp_dir.join(format!("{}.{}", basename, ext));
                let _ = std::fs::remove_file(&p);
            }
        });

        // Atomically rename each file from temp to final location. Track
        // which files were successfully moved so we can roll back on failure.
        // `all_exts` is reused from the stale-bak guard above.
        let mut moved_exts: Vec<&str> = Vec::new();

        let rename_result: Result<(), HnswError> = (|| {
            // Back up existing files before overwriting so rollback can
            // restore them.
            //
            // Propagate the rename error rather than warning-and-continuing.
            // If the backup never landed, the rollback path further down
            // can't restore the original file when atomic_replace later
            // fails — we'd delete the (newly promoted) file in `moved_exts`
            // then look for a `.bak` that was never created and silently lose
            // the prior index. Better to bail BEFORE the atomic_replace pass
            // touches anything.
            for ext in &all_exts {
                let final_path = dir.join(format!("{}.{}", basename, ext));
                let bak_path = dir.join(format!("{}.{}.bak", basename, ext));
                if final_path.exists() {
                    std::fs::rename(&final_path, &bak_path).map_err(|e| {
                        HnswError::Internal(format!(
                            "Failed to back up {} -> {} before save: {}",
                            final_path.display(),
                            bak_path.display(),
                            e
                        ))
                    })?;
                }
            }

            // Fsync the parent directory so the `.bak` rename entries are
            // durable before the atomic_replace pass proceeds. Without this,
            // a power cut between the backup loop and atomic_replace can
            // leave the directory in a state where the `.bak` file exists in
            // the page cache but not on disk. Best-effort fsync: log at debug
            // on platforms that don't support directory fsync.
            match std::fs::File::open(dir) {
                Ok(f) => {
                    if let Err(e) = f.sync_all() {
                        tracing::debug!(
                            error = %e,
                            dir = %dir.display(),
                            "fsync of HNSW parent directory after backup loop failed (non-fatal)"
                        );
                    }
                }
                Err(e) => {
                    tracing::debug!(
                        error = %e,
                        dir = %dir.display(),
                        "could not open HNSW parent directory for fsync after backup loop"
                    );
                }
            }

            for ext in &all_exts {
                let temp_path = temp_dir.join(format!("{}.{}", basename, ext));
                let final_path = dir.join(format!("{}.{}", basename, ext));
                // Every staged file is written unconditionally earlier in
                // this function, so a missing one means the staging dir was
                // deleted out from under us (e.g. by another process's
                // stale-temp cleanup). Silently skipping it would promote a
                // partial set and let the success path below delete the
                // `.bak`s of the previous good index — hard-error into the
                // rollback instead.
                if !temp_path.exists() {
                    return Err(HnswError::Internal(format!(
                        "HNSW staging file missing during promote: {} — \
                         aborting save and rolling back",
                        temp_path.display()
                    )));
                }
                // `atomic_replace` fsyncs the temp file, renames with
                // cross-device fallback (the in-process temp dir sits
                // on a different device from `dir` on Docker overlayfs
                // and WSL `/mnt/c`), and fsyncs the parent directory.
                crate::fs::atomic_replace(&temp_path, &final_path).map_err(|e| {
                    HnswError::Internal(format!(
                        "Failed to promote {} -> {}: {}",
                        temp_path.display(),
                        final_path.display(),
                        e
                    ))
                })?;
                // Keep the restrictive mode on the promoted file now that
                // it lives in the final directory. The library
                // `file_dump` output uses the process umask; we want
                // 0o600 regardless.
                #[cfg(unix)]
                {
                    use std::os::unix::fs::PermissionsExt;
                    let _ = std::fs::set_permissions(
                        &final_path,
                        std::fs::Permissions::from_mode(0o600),
                    );
                }
                moved_exts.push(ext);
            }
            Ok(())
        })();

        if let Err(e) = rename_result {
            // Roll back: remove new files and restore originals from .bak
            for ext in &moved_exts {
                let final_path = dir.join(format!("{}.{}", basename, ext));
                let _ = std::fs::remove_file(&final_path);
            }
            // Track which `.bak` files failed to restore rather than
            // warning-and-continuing. The unrestored ones are left in place
            // as recovery breadcrumbs and the operator gets an actionable
            // error. The start-of-save guard will also refuse to start the
            // next save while any `.bak` survives, but surfacing the failure
            // here gives an immediate signal that manual recovery is needed.
            let mut restore_failures: Vec<(String, std::io::Error)> = Vec::new();
            for ext in &all_exts {
                let bak_path = dir.join(format!("{}.{}.bak", basename, ext));
                let final_path = dir.join(format!("{}.{}", basename, ext));
                if bak_path.exists() {
                    if let Err(rename_err) = std::fs::rename(&bak_path, &final_path) {
                        tracing::error!(
                            path = %final_path.display(),
                            error = %rename_err,
                            "Failed to restore backup during HNSW save rollback"
                        );
                        restore_failures.push((final_path.display().to_string(), rename_err));
                    }
                }
            }
            // Fsync the parent directory after restoring backups so the
            // restore renames are durable. Without this, a second power cut
            // during rollback can leave the index with missing files even
            // though the `.bak` existed on disk.
            match std::fs::File::open(dir) {
                Ok(f) => {
                    if let Err(sync_err) = f.sync_all() {
                        tracing::debug!(
                            error = %sync_err,
                            dir = %dir.display(),
                            "fsync of HNSW parent directory after rollback failed (non-fatal)"
                        );
                    }
                }
                Err(open_err) => {
                    tracing::debug!(
                        error = %open_err,
                        dir = %dir.display(),
                        "could not open HNSW parent directory for fsync after rollback"
                    );
                }
            }
            tracing::warn!(error = %e, "HNSW save failed mid-rename, rolled back to original files");
            let _ = std::fs::remove_dir_all(&temp_dir);

            // Surface partial-rollback as an actionable error so the
            // operator can clear the orphaned .bak files. The next `save`
            // would refuse to start anyway under the stale-bak guard above;
            // this turns the deferred refusal into an immediate breadcrumb at
            // the point of failure.
            if !restore_failures.is_empty() {
                let detail: String = restore_failures
                    .iter()
                    .map(|(p, err)| format!("  {p}: {err}"))
                    .collect::<Vec<_>>()
                    .join("\n");
                tracing::error!(
                    %detail,
                    dir = %dir.display(),
                    "HNSW rollback INCOMPLETE — manual recovery required: \
                     restore .bak files matching {basename}.*.bak in dir",
                );
                return Err(HnswError::Internal(format!(
                    "HNSW rollback partially failed; .bak files present, \
                     manual recovery required:\n{detail}"
                )));
            }
            return Err(e);
        }

        // Fsync the parent directory so the renames themselves are durable
        // on a power cut — otherwise the files exist on disk but the
        // directory entries can be reordered or lost, leaving a half-saved
        // index. Best-effort: on platforms where opening a directory for
        // fsync isn't supported we log at debug level and continue.
        match std::fs::File::open(dir) {
            Ok(f) => {
                if let Err(e) = f.sync_all() {
                    tracing::debug!(
                        error = %e,
                        dir = %dir.display(),
                        "fsync of HNSW parent directory failed (non-fatal)"
                    );
                }
            }
            Err(e) => {
                tracing::debug!(
                    error = %e,
                    dir = %dir.display(),
                    "could not open HNSW parent directory for fsync"
                );
            }
        }

        // Clean up temp directory and .bak files from successful save
        let _ = std::fs::remove_dir_all(&temp_dir);
        for ext in &all_exts {
            let bak_path = dir.join(format!("{}.{}.bak", basename, ext));
            let _ = std::fs::remove_file(&bak_path);
        }

        tracing::info!(
            "HNSW index saved: {} vectors (with checksums)",
            self.id_map.len()
        );

        Ok(SaveOutcome::Saved)
    }

    /// Load an index from disk
    ///
    /// Verifies blake3 checksums before loading to mitigate bincode deserialization risks.
    /// Memory is properly freed when the HnswIndex is dropped.
    pub fn load_with_dim(dir: &Path, basename: &str, dim: usize) -> Result<Self, HnswError> {
        let _span = tracing::debug_span!("hnsw_load", dir = %dir.display(), basename).entered();
        let graph_path = dir.join(format!("{}.hnsw.graph", basename));
        let data_path = dir.join(format!("{}.hnsw.data", basename));
        let id_map_path = dir.join(format!("{}.hnsw.ids", basename));

        // The stale-temp cleanup and the existence check both happen
        // INSIDE the shared-lock critical section.
        //
        // Cleanup before the lock would race a concurrent `save()`: the
        // saver creates its `.{basename}.<suffix>.tmp` staging directory
        // under the EXCLUSIVE lock and writes into it while a pre-lock
        // loader could `remove_dir_all` that live staging dir. The saver
        // would then rename the previous good files to `.bak`, find every
        // staged file missing during promote, and the success path would
        // delete the `.bak`s — destroying the previous good index. Once
        // the shared lock is held, no exclusive saver is mid-flight, so
        // any surviving `.tmp` directory is genuinely stale.
        //
        // Checking file existence before acquiring the lock
        // would open a TOCTOU window: a concurrent `save()` could take
        // the exclusive lock, rename the four files into `.bak`, and
        // complete; the reader would then proceed with files that had
        // already been replaced, and `verify_hnsw_checksums` would fail
        // with a confusing checksum-mismatch error during a normal
        // concurrent rebuild.
        //
        // The lock-acquisition path doesn't depend on the four data
        // files existing (the lock file is a separate `.hnsw.lock`),
        // so the check is done only AFTER the shared lock is held: any
        // saver that wants to rename files MUST first take the
        // exclusive lock.
        //
        // The deeper hazard — a half-renamed set where save N+1 has
        // moved graph but not yet data — remains and is tracked as
        // the larger "bundle into single .hnsw.bundle for atomic
        // replace" refactor. The lock-then-existence check at least
        // ensures we don't observe in-progress renames.

        // Acquire shared lock for the read phase only (released before
        // return). Protects the on-disk graph/data/ids/checksum files
        // from being overwritten by a concurrent `save()` while this
        // function is reading them — once the data is in memory the
        // copy is independent and the lock is no longer needed.
        //
        // NOTE: File locking is advisory only on WSL over 9P. The lock
        // is best-effort cross-process; an external Windows tool can
        // still modify the files.
        //
        // **Why the lock must NOT outlive the read phase**: keeping the
        // shared lock alive for the index's entire lifetime (by storing
        // it on the returned `HnswIndex`) deadlocks the daemon. The same
        // process's HNSW rebuild thread calls `save()` from a *second*
        // file descriptor on the same lock path; Linux `flock(2)`'s
        // exclusive-waits-for-shared-even-self semantics block the
        // rebuild thread permanently. Dropping the lock at the bottom of
        // this function makes the load and save paths use disjoint open
        // descriptions so the in-process self-deadlock can't form.
        // Cross-process exclusion during runtime is theoretical anyway —
        // concurrent writers race on the SQLite writer lock first.
        let lock_path = dir.join(format!("{}.hnsw.lock", basename));
        let lock_file = std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(&lock_path)?;
        // Bound the wait with try-lock-and-retry so a wedged peer process
        // holding the exclusive save lock doesn't hang every reader
        // indefinitely. 5 attempts × 200 ms (with jitter) = ~1s ceiling;
        // daemons can retry the load and CLI users see a clear error
        // instead of a silent hang. Without this, a paused `cqs index`
        // (debugger / SIGSTOP) takes down all readers.
        let lock_acquired = (0..5).any(|attempt| {
            if attempt > 0 {
                let jitter = std::time::Duration::from_millis(200 + (attempt * 37) as u64);
                std::thread::sleep(jitter);
            }
            lock_file.try_lock_shared().is_ok()
        });
        if !lock_acquired {
            return Err(HnswError::Internal(format!(
                "HNSW load lock contended after 5 retries — concurrent save in progress at {}",
                lock_path.display()
            )));
        }
        warn_wsl_advisory_locking(dir);
        tracing::debug!(lock_path = %lock_path.display(), "Acquired HNSW load lock (shared, released after read)");

        // Clean up stale temp dirs from interrupted saves. This runs under
        // the shared lock: a live `save()` holds the exclusive lock while
        // its staging directory exists, so any `.tmp` directory observed
        // here is from a crashed/killed save and safe to remove. Temp dirs
        // have unpredictable suffixes, so match by prefix+suffix pattern.
        if let Ok(entries) = std::fs::read_dir(dir) {
            let prefix = format!(".{}.", basename);
            for entry in entries.flatten() {
                let name = entry.file_name();
                let name = name.to_string_lossy();
                if name.starts_with(&prefix) && name.ends_with(".tmp") && entry.path().is_dir() {
                    tracing::info!(dir = %entry.path().display(), "Cleaning up interrupted HNSW save");
                    let _ = std::fs::remove_dir_all(entry.path());
                }
            }
        }

        // Existence check happens under the shared lock. A concurrent
        // `save()` that wanted to rename these files would need the
        // exclusive lock — which it can't get while we hold the shared
        // lock. So the files we observe here are guaranteed to remain in
        // place for the duration of the load.
        if !graph_path.exists() || !data_path.exists() || !id_map_path.exists() {
            return Err(HnswError::NotFound(dir.display().to_string()));
        }

        tracing::info!(dir = %dir.display(), basename, "Loading HNSW index");
        verify_hnsw_checksums(dir, basename)?;

        // Resolve the distance metric. The stored value (meta
        // header; legacy indexes without one are cosine) always wins — but
        // an *explicit* conflicting CQS_DISTANCE_METRIC is a typed error
        // rather than a silent reinterpretation of distances.
        let stored_metric = read_hnsw_meta_metric(dir, basename)?;
        if let Some(requested) = DistanceMetric::from_env().map_err(HnswError::Internal)? {
            if requested != stored_metric {
                return Err(HnswError::MetricMismatch {
                    stored: stored_metric,
                    requested,
                });
            }
        }

        // Check ID map file size to prevent OOM (limit configurable via CQS_HNSW_MAX_ID_MAP_BYTES)
        let max_id_map_size = hnsw_max_id_map_bytes();
        let id_map_size = std::fs::metadata(&id_map_path)
            .map_err(|e| {
                HnswError::Internal(format!(
                    "Failed to stat ID map {}: {}",
                    id_map_path.display(),
                    e
                ))
            })?
            .len();
        if id_map_size > max_id_map_size {
            return Err(HnswError::Internal(format!(
                "ID map too large: {}MB > {}MB limit",
                id_map_size / (1024 * 1024),
                max_id_map_size / (1024 * 1024)
            )));
        }

        // Check graph and data file sizes to prevent OOM before deserialization
        for (path, limit, label) in [
            (&graph_path, hnsw_max_graph_bytes(), "graph"),
            (&data_path, hnsw_max_data_bytes(), "data"),
        ] {
            let size = std::fs::metadata(path)
                .map_err(|e| {
                    HnswError::Internal(format!(
                        "Failed to stat HNSW {} file {}: {}",
                        label,
                        path.display(),
                        e
                    ))
                })?
                .len();
            if size > limit {
                return Err(HnswError::Internal(format!(
                    "HNSW {} file too large: {}MB > {}MB limit",
                    label,
                    size / (1024 * 1024),
                    limit / (1024 * 1024)
                )));
            }
        }

        // Load ID map via streaming parse to avoid holding raw JSON + parsed Vec simultaneously
        let id_map_file = std::fs::File::open(&id_map_path).map_err(|e| {
            HnswError::Internal(format!(
                "Failed to open ID map {}: {}",
                id_map_path.display(),
                e
            ))
        })?;
        let id_map_reader = std::io::BufReader::new(id_map_file);
        // Deserialize as `Vec<String>` (serde's default), then convert in
        // place to `Vec<Box<str>>`. The wire format is JSON `[String, ...]`.
        // `String::into_boxed_str()` is zero-copy (shrinks the existing heap
        // allocation, drops the `cap` field).
        let id_map_strings: Vec<String> = serde_json::from_reader(id_map_reader)
            .map_err(|e| HnswError::Internal(format!("Failed to parse ID map: {}", e)))?;
        let id_map: Vec<Box<str>> = id_map_strings
            .into_iter()
            .map(String::into_boxed_str)
            .collect();

        // Cap element count to prevent memory exhaustion from crafted id_map files.
        // 10M entries at ~64 bytes average ID = ~640MB — well above any real codebase.
        const MAX_ID_MAP_ENTRIES: usize = 10_000_000;
        if id_map.len() > MAX_ID_MAP_ENTRIES {
            return Err(HnswError::Internal(format!(
                "ID map has {} entries, exceeding {} limit — possible corruption",
                id_map.len(),
                MAX_ID_MAP_ENTRIES
            )));
        }

        // Validate data file size against id_map before bincode deserialization.
        // A crafted file could claim more vectors than the id_map supports, causing
        // unbounded allocation during deserialization. Each vector is `dim` f32s,
        // with 2x headroom for HNSW graph overhead (neighbor lists, metadata).
        //
        // Use checked_mul so a pathological `dim` argument (e.g. a model with
        // a huge embedding dimension) can't overflow usize silently on 32-bit
        // targets. On 64-bit the product fits for any realistic corpus, but
        // defense-in-depth is cheap here.
        if !id_map.is_empty() {
            let expected_max_data = id_map
                .len()
                .checked_mul(dim)
                .and_then(|v| v.checked_mul(std::mem::size_of::<f32>()))
                .and_then(|v| v.checked_mul(2))
                .ok_or_else(|| {
                    HnswError::Internal(format!(
                        "expected_max_data overflow: id_map={} dim={}",
                        id_map.len(),
                        dim
                    ))
                })?;
            let data_meta = std::fs::metadata(&data_path).map_err(|e| {
                HnswError::Internal(format!(
                    "Failed to stat data file {}: {}",
                    data_path.display(),
                    e
                ))
            })?;
            if data_meta.len() as usize > expected_max_data {
                return Err(HnswError::Internal(format!(
                    "HNSW data file ({} bytes) too large for {} vectors (max {} bytes)",
                    data_meta.len(),
                    id_map.len(),
                    expected_max_data
                )));
            }
        }

        // Load HNSW graph using self_cell for safe self-referential ownership
        //
        // hnsw_rs returns Hnsw<'a> borrowing from &'a mut HnswIo.
        // self_cell ties these lifetimes together without transmute.
        let hnsw_io_cell = Box::new(HnswIoCell(UnsafeCell::new(HnswIo::new(dir, basename))));

        let loaded = LoadedHnsw::try_new(hnsw_io_cell, |cell| {
            // SAFETY: Exclusive access during construction — no other references exist.
            // After this closure returns, the UnsafeCell is never accessed again directly.
            let io = unsafe { &mut *cell.0.get() };
            // Instantiate the dist type recorded in the meta header. hnsw_rs
            // additionally cross-checks the distance name embedded in the
            // graph file, so a forged meta cannot misread distances.
            HnswGraph::load(io, stored_metric)
        })?;

        // Validate id_map size matches HNSW vector count
        let hnsw_count = loaded.with_dependent(|_, hnsw| hnsw.get_nb_point());
        if hnsw_count != id_map.len() {
            return Err(HnswError::Internal(format!(
                "ID map size mismatch: HNSW has {} vectors but id_map has {}",
                hnsw_count,
                id_map.len()
            )));
        }

        tracing::info!(count = id_map.len(), metric = %stored_metric, "HNSW index loaded");

        // Release the load-phase shared lock before returning. The
        // in-memory `Loaded(...)` variant is fully self-contained
        // (vectors copied into rust-allocated buffers via `bincode`,
        // not mmap'd), so the on-disk file can be modified or removed
        // without affecting this instance. Keeping the lock alive
        // across the return would self-deadlock the daemon's rebuild
        // thread on its next `save()` (Linux flock; see the lock
        // acquisition site above).
        drop(lock_file);

        Ok(Self {
            inner: HnswInner::Loaded(loaded),
            id_map,
            ef_search: super::ef_search(),
            dim,
            _lock_file: None,
        })
    }

    /// Read the stored distance metric for a persisted index without
    /// loading it. Missing meta header → `Cosine` (legacy indexes,
    /// and the build default when no index exists yet). Used by derived
    /// backends (CAGRA rebuild path) to stay consistent with the slot's
    /// HNSW metric when `CQS_DISTANCE_METRIC` is unset.
    pub fn stored_metric(dir: &Path, basename: &str) -> Result<DistanceMetric, HnswError> {
        read_hnsw_meta_metric(dir, basename)
    }

    /// Check if an HNSW index exists at the given path
    pub fn exists(dir: &Path, basename: &str) -> bool {
        let graph_path = dir.join(format!("{}.hnsw.graph", basename));
        let data_path = dir.join(format!("{}.hnsw.data", basename));
        let id_map_path = dir.join(format!("{}.hnsw.ids", basename));

        graph_path.exists() && data_path.exists() && id_map_path.exists()
    }

    /// Get vector count without loading the full index (fast, for stats).
    ///
    /// Uses `BufReader` + `serde_json::from_reader` to avoid reading the entire
    /// id map file into a String first. The file is a JSON array of chunk ID strings.
    pub fn count_vectors(dir: &Path, basename: &str) -> Option<usize> {
        let id_map_path = dir.join(format!("{}.hnsw.ids", basename));
        let file = match std::fs::File::open(&id_map_path) {
            Ok(f) => f,
            Err(e) => {
                tracing::debug!(
                    path = %id_map_path.display(),
                    error = %e,
                    "Could not read HNSW id map"
                );
                return None;
            }
        };
        // Acquire shared lock for consistency with load()
        if let Err(e) = file.lock_shared() {
            tracing::debug!(error = %e, "Could not lock HNSW id map");
            return None;
        }
        // Guard against oversized id map files.
        //
        // 1 GB cap aligns with the hard-load path's MAX_ID_MAP_ENTRIES =
        // 10M × ~64 byte strings = ~640 MB. A lower cap would make
        // `cqs stats` / health report "unknown vector count" for large
        // corpora well below the 1M+ scaling target.
        const MAX_ID_MAP_SIZE: u64 = 1024 * 1024 * 1024; // 1GB
        match file.metadata() {
            Ok(meta) if meta.len() > MAX_ID_MAP_SIZE => {
                tracing::warn!(
                    size_bytes = meta.len(),
                    path = %id_map_path.display(),
                    "HNSW id map too large"
                );
                return None;
            }
            Err(e) => {
                tracing::debug!(
                    path = %id_map_path.display(),
                    error = %e,
                    "Could not stat HNSW id map"
                );
                return None;
            }
            _ => {}
        }
        // Count array elements by streaming JSON without allocating all strings.
        // The id map is a JSON array of strings: ["id1","id2",...].
        // We iterate the stream and count SeqAccess elements rather than
        // deserializing into a Vec<String>.
        use serde::de::{Deserializer, SeqAccess, Visitor};
        use std::fmt;

        struct CountVisitor;
        impl<'de> Visitor<'de> for CountVisitor {
            type Value = usize;
            fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
                write!(f, "an array")
            }
            /// Count sequence elements without materializing them.
            fn visit_seq<A: SeqAccess<'de>>(self, mut seq: A) -> Result<usize, A::Error> {
                let mut count = 0usize;
                while seq.next_element::<serde::de::IgnoredAny>()?.is_some() {
                    count += 1;
                }
                Ok(count)
            }
        }

        let reader = std::io::BufReader::new(file);
        let mut de = serde_json::Deserializer::from_reader(reader);
        match de.deserialize_seq(CountVisitor) {
            Ok(count) => Some(count),
            Err(e) => {
                tracing::warn!(path = %id_map_path.display(), error = %e, "Corrupted HNSW id map");
                None
            }
        }
    }

    /// Load HNSW index with optional ef_search override and an explicit
    /// runtime dim.
    ///
    /// `dim` is required — it must match the dimension the index was built
    /// with. Mismatches surface as load errors rather than silent misreads.
    pub fn try_load_with_ef(
        cq_dir: &Path,
        ef_search: Option<usize>,
        dim: usize,
    ) -> Option<Box<dyn VectorIndex>> {
        Self::try_load_named(cq_dir, "index", ef_search, dim)
    }

    /// Load the base (non-enriched) HNSW index.
    ///
    /// Returns `None` when `index_base.hnsw.*` files are absent or corrupt —
    /// the router treats that as a signal to fall back to the enriched index.
    ///
    /// `dim` is required for the same reasons as `try_load_with_ef`.
    pub fn try_load_base_with_ef(
        cq_dir: &Path,
        ef_search: Option<usize>,
        dim: usize,
    ) -> Option<Box<dyn VectorIndex>> {
        Self::try_load_named(cq_dir, "index_base", ef_search, dim)
    }

    /// Internal: load any named HNSW index (enriched, base, or future variants).
    fn try_load_named(
        cq_dir: &Path,
        basename: &str,
        ef_search: Option<usize>,
        dim: usize,
    ) -> Option<Box<dyn VectorIndex>> {
        if Self::exists(cq_dir, basename) {
            match Self::load_with_dim(cq_dir, basename, dim) {
                Ok(mut index) => {
                    if let Some(ef) = ef_search {
                        index.set_ef_search(ef);
                        tracing::debug!(
                            basename = basename,
                            ef_search = ef,
                            "Applied config ef_search override"
                        );
                    }
                    tracing::info!(
                        basename = basename,
                        vectors = index.len(),
                        "HNSW index loaded"
                    );
                    Some(Box::new(index))
                }
                Err(e) => {
                    tracing::warn!(
                        basename = basename,
                        error = %e,
                        "HNSW index corrupted or incomplete — falling back to brute-force search. \
                         Run 'cqs index' to rebuild."
                    );
                    None
                }
            }
        } else {
            tracing::debug!(basename = basename, "No HNSW index found");
            None
        }
    }
}

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

    use crate::hnsw::make_test_embedding as make_embedding;

    /// Write a checksum file matching the given HNSW files
    fn write_checksums(dir: &Path, basename: &str) {
        let mut lines = Vec::new();
        for ext in &["hnsw.graph", "hnsw.data", "hnsw.ids"] {
            let path = dir.join(format!("{}.{}", basename, ext));
            if path.exists() {
                let mut hasher = blake3::Hasher::new();
                let mut file = std::fs::File::open(&path).unwrap();
                std::io::copy(&mut file, &mut hasher).unwrap();
                let hash = hasher.finalize().to_hex().to_string();
                lines.push(format!("{}:{}", ext, hash));
            }
        }
        std::fs::write(
            dir.join(format!("{}.hnsw.checksum", basename)),
            lines.join("\n"),
        )
        .unwrap();
    }

    #[test]
    fn test_load_rejects_oversized_graph_file() {
        let tmp = TempDir::new().unwrap();

        // Create valid-looking HNSW files, but make graph oversized
        let graph_path = tmp.path().join("test.hnsw.graph");
        let data_path = tmp.path().join("test.hnsw.data");
        let ids_path = tmp.path().join("test.hnsw.ids");

        // Write oversized graph file (just over 500MB limit)
        // We use set_len to create a sparse file — no actual disk I/O
        let f = std::fs::File::create(&graph_path).unwrap();
        f.set_len(501 * 1024 * 1024).unwrap();

        std::fs::write(&data_path, b"dummy").unwrap();
        std::fs::write(&ids_path, b"[]").unwrap();
        write_checksums(tmp.path(), "test");

        match HnswIndex::load_with_dim(tmp.path(), "test", crate::EMBEDDING_DIM) {
            Err(e) => {
                let msg = format!("{}", e);
                assert!(
                    msg.contains("graph") && msg.contains("too large"),
                    "Expected graph size error, got: {}",
                    msg
                );
            }
            Ok(_) => panic!("Expected error for oversized graph file"),
        }
    }

    #[test]
    fn test_load_rejects_oversized_data_file() {
        let tmp = TempDir::new().unwrap();

        let graph_path = tmp.path().join("test.hnsw.graph");
        let data_path = tmp.path().join("test.hnsw.data");
        let ids_path = tmp.path().join("test.hnsw.ids");

        std::fs::write(&graph_path, b"dummy").unwrap();

        // Write oversized data file (just over 1GB limit)
        let f = std::fs::File::create(&data_path).unwrap();
        f.set_len(1025 * 1024 * 1024).unwrap();

        std::fs::write(&ids_path, b"[]").unwrap();
        write_checksums(tmp.path(), "test");

        match HnswIndex::load_with_dim(tmp.path(), "test", crate::EMBEDDING_DIM) {
            Err(e) => {
                let msg = format!("{}", e);
                assert!(
                    msg.contains("data") && msg.contains("too large"),
                    "Expected data size error, got: {}",
                    msg
                );
            }
            Ok(_) => panic!("Expected error for oversized data file"),
        }
    }

    #[test]
    fn test_load_rejects_data_too_large_for_id_map() {
        let tmp = TempDir::new().unwrap();

        let graph_path = tmp.path().join("test.hnsw.graph");
        let data_path = tmp.path().join("test.hnsw.data");
        let ids_path = tmp.path().join("test.hnsw.ids");

        std::fs::write(&graph_path, b"dummy").unwrap();

        // id_map claims 2 vectors, but data file is far larger than
        // 2 * 768 * 4 * 2 = 12,288 bytes would allow
        std::fs::write(&ids_path, r#"["a","b"]"#).unwrap();
        let f = std::fs::File::create(&data_path).unwrap();
        f.set_len(1_000_000).unwrap(); // ~1MB >> 12KB limit
        write_checksums(tmp.path(), "test");

        match HnswIndex::load_with_dim(tmp.path(), "test", crate::EMBEDDING_DIM) {
            Err(e) => {
                let msg = format!("{}", e);
                assert!(
                    msg.contains("data file") && msg.contains("too large for"),
                    "Expected data/id_map size mismatch error, got: {}",
                    msg
                );
            }
            Ok(_) => panic!("Expected error for data file exceeding id_map capacity"),
        }
    }

    #[test]
    fn test_load_rejects_missing_checksum() {
        let tmp = TempDir::new().unwrap();

        std::fs::write(tmp.path().join("test.hnsw.graph"), b"data").unwrap();
        std::fs::write(tmp.path().join("test.hnsw.data"), b"data").unwrap();
        std::fs::write(tmp.path().join("test.hnsw.ids"), b"[]").unwrap();
        // No checksum file

        match HnswIndex::load_with_dim(tmp.path(), "test", crate::EMBEDDING_DIM) {
            Err(e) => {
                let msg = format!("{}", e);
                assert!(
                    msg.contains("No checksum file"),
                    "Expected checksum error, got: {}",
                    msg
                );
            }
            Ok(_) => panic!("Expected error for missing checksum file"),
        }
    }

    #[test]
    fn test_save_creates_lock_file() {
        let tmp = TempDir::new().unwrap();
        let basename = "test_lock";

        let embeddings = vec![
            ("chunk1".to_string(), make_embedding(1)),
            ("chunk2".to_string(), make_embedding(2)),
        ];

        let index = HnswIndex::build_with_dim(embeddings, crate::EMBEDDING_DIM).unwrap();
        index.save(tmp.path(), basename).unwrap();

        let lock_path = tmp.path().join(format!("{}.hnsw.lock", basename));
        assert!(lock_path.exists(), "Lock file should exist after save");
    }

    #[test]
    fn test_concurrent_load_shared() {
        let tmp = TempDir::new().unwrap();
        let basename = "test_shared";

        let embeddings = vec![
            ("chunk1".to_string(), make_embedding(1)),
            ("chunk2".to_string(), make_embedding(2)),
        ];

        let index = HnswIndex::build_with_dim(embeddings, crate::EMBEDDING_DIM).unwrap();
        index.save(tmp.path(), basename).unwrap();

        // Load twice — shared locks should not block each other
        let loaded1 = HnswIndex::load_with_dim(tmp.path(), basename, crate::EMBEDDING_DIM).unwrap();
        let loaded2 = HnswIndex::load_with_dim(tmp.path(), basename, crate::EMBEDDING_DIM).unwrap();
        assert_eq!(loaded1.len(), 2);
        assert_eq!(loaded2.len(), 2);
    }

    /// Regression for the in-process flock self-deadlock between
    /// `load_with_dim`'s shared lock and a subsequent `save()`'s
    /// exclusive lock. If the load held a `shared` flock for the entire
    /// lifetime of the returned `HnswIndex` via the `_lock_file` field,
    /// the daemon's HNSW rebuild thread would open a *second* fd on the
    /// same lock path inside `save()` and call `lock()` (exclusive),
    /// which blocks forever on Linux because `flock(2)` exclusive-waits
    /// for *all* shared holders, including ones held by the same process
    /// via a different open description. That manifests as a permanently
    /// `state == rebuilding` daemon with one `cqs-hnsw-rebuild` thread
    /// parked in `locks_lock_inode_wait`.
    ///
    /// `load_with_dim` releases the load lock at the end (the in-memory
    /// `Loaded(...)` is independent of the on-disk files), so a
    /// same-process `save()` after a `load()` can proceed. We model this
    /// by loading the on-disk index, then saving a *fresh* `HnswIndex`
    /// (built in memory) to the same directory while the loaded handle is
    /// still alive. We bound the save in a worker thread with a 5-second
    /// timeout so the test fails fast on regression instead of hanging the
    /// suite.
    #[test]
    fn load_does_not_block_subsequent_save_in_same_process() {
        use std::sync::mpsc::{channel, RecvTimeoutError};
        use std::time::Duration;

        let tmp = TempDir::new().unwrap();
        let basename = "deadlock_repro";

        let initial = HnswIndex::build_with_dim(
            vec![
                ("chunk1".to_string(), make_embedding(1)),
                ("chunk2".to_string(), make_embedding(2)),
            ],
            crate::EMBEDDING_DIM,
        )
        .unwrap();
        initial.save(tmp.path(), basename).unwrap();

        // Step 2: load the on-disk HNSW. The shared flock used during the
        // read phase is released before this returns.
        let _loaded = HnswIndex::load_with_dim(tmp.path(), basename, crate::EMBEDDING_DIM)
            .expect("load should succeed");

        // Step 3: build a fresh HnswIndex in memory and save it to the
        // same directory while the loaded handle is still alive. This
        // re-enters `save()`, which opens a second fd on the same lock
        // path and calls `lock()` (exclusive) — the deadlock surface.
        let dir = tmp.path().to_path_buf();
        let basename_owned = basename.to_string();
        let (tx, rx) = channel();
        std::thread::spawn(move || {
            let new_index = HnswIndex::build_with_dim(
                vec![("chunk3".to_string(), make_embedding(3))],
                crate::EMBEDDING_DIM,
            )
            .unwrap();
            // Save into a NEW basename so we don't trip the
            // stale-`.bak` guard from a partial first save; the lock
            // path is keyed on basename so we keep the original to
            // exercise the deadlock surface.
            let result = new_index.save(&dir, &basename_owned);
            let _ = tx.send(result);
        });

        match rx.recv_timeout(Duration::from_secs(5)) {
            Ok(Ok(())) => { /* save completed cleanly — fix is in place */ }
            Ok(Err(e)) => panic!("save failed: {e:?}"),
            Err(RecvTimeoutError::Timeout) => panic!(
                "save deadlocked — regression: load's shared flock is still held while save \
                 tries to acquire exclusive. Verify `load_with_dim` drops `lock_file` before \
                 returning."
            ),
            Err(RecvTimeoutError::Disconnected) => {
                panic!("save thread panicked without sending a result")
            }
        }
    }

    #[test]
    fn test_load_cleans_stale_temp_dir() {
        let dir = tempfile::tempdir().unwrap();
        let basename = "test_index";
        let temp_dir = dir.path().join(format!(".{}.tmp", basename));
        std::fs::create_dir_all(&temp_dir).unwrap();

        // Load should clean up the temp dir even though no index exists.
        // The cleanup happens after the shared lock is acquired (no saver
        // is mid-flight here, so the load proceeds straight through).
        let result = HnswIndex::load_with_dim(dir.path(), basename, crate::EMBEDDING_DIM);
        assert!(result.is_err()); // no index to load
        assert!(!temp_dir.exists()); // but temp dir should be cleaned
    }

    /// Pins the cleanup-under-lock ordering: while a saver holds the
    /// exclusive lock, its staging directory is LIVE and a concurrent load
    /// must not delete it. If the cleanup ever moves back in front of the
    /// shared-lock acquisition, this test fails because the staging dir
    /// gets removed while the "saver" still holds the exclusive lock.
    #[test]
    fn test_load_does_not_clean_staging_dir_while_saver_holds_lock() {
        let dir = tempfile::tempdir().unwrap();
        let basename = "test_index";

        // Simulate a mid-flight save: staging dir exists and the exclusive
        // lock is held (flock is per open file description, so this blocks
        // the load's shared lock even within the same process).
        let staging = dir
            .path()
            .join(format!(".{}.{:016x}.tmp", basename, 0xdeadbeefu64));
        std::fs::create_dir_all(&staging).unwrap();
        let lock_path = dir.path().join(format!("{}.hnsw.lock", basename));
        #[allow(clippy::suspicious_open_options)]
        let lock_file = std::fs::OpenOptions::new()
            .create(true)
            .write(true)
            .open(&lock_path)
            .unwrap();
        lock_file.lock().unwrap();

        // Load must fail to acquire the shared lock (bounded retries) and
        // must NOT have touched the live staging directory.
        let result = HnswIndex::load_with_dim(dir.path(), basename, crate::EMBEDDING_DIM);
        let err = format!(
            "{}",
            result.err().expect("load must fail under lock contention")
        );
        assert!(
            err.contains("lock contended"),
            "load should fail on lock contention, got: {}",
            err
        );
        assert!(
            staging.exists(),
            "live staging dir must survive a load that never got the shared lock"
        );
    }

    /// A staged file that goes missing between staging and promote must
    /// hard-error into the rollback path — never report a successful save.
    /// Silently skipping missing temps used to promote a partial set and
    /// then delete the `.bak`s of the previous good index on the success
    /// path, destroying it.
    #[test]
    fn test_save_errors_and_rolls_back_when_staged_file_missing() {
        let tmp = TempDir::new().unwrap();
        let basename = "index";

        // Establish a good v1 index on disk.
        let v1: Vec<(String, crate::embedder::Embedding)> = (1..=2)
            .map(|i| (format!("vec{}", i), make_embedding(i)))
            .collect();
        let v1_idx = HnswIndex::build_with_dim(v1, crate::EMBEDDING_DIM).unwrap();
        v1_idx.save(tmp.path(), basename).unwrap();
        let v1_ids = std::fs::read(tmp.path().join(format!("{}.hnsw.ids", basename))).unwrap();

        // Save v2 with the ids staging file deleted just before promote.
        // graph/data promote first, so this exercises mid-loop rollback.
        let v2: Vec<(String, crate::embedder::Embedding)> = (1..=3)
            .map(|i| (format!("newvec{}", i), make_embedding(i + 10)))
            .collect();
        let v2_idx = HnswIndex::build_with_dim(v2, crate::EMBEDDING_DIM).unwrap();
        test_hooks::DELETE_STAGED_EXT_BEFORE_PROMOTE
            .with(|hook| *hook.borrow_mut() = Some("hnsw.ids".to_string()));
        let result = v2_idx.save(tmp.path(), basename);

        let err = format!("{}", result.unwrap_err());
        assert!(
            err.contains("staging file missing during promote"),
            "save must surface the missing staged file, got: {}",
            err
        );

        // Rollback must restore the previous good index: v1 files back in
        // place, no `.bak` breadcrumbs, no staging dir, and the index loads.
        for ext in ["hnsw.graph", "hnsw.data", "hnsw.ids", "hnsw.checksum"] {
            let bak = tmp.path().join(format!("{}.{}.bak", basename, ext));
            assert!(
                !bak.exists(),
                "rollback must consume .bak files; {} still exists",
                bak.display()
            );
        }
        let restored_ids =
            std::fs::read(tmp.path().join(format!("{}.hnsw.ids", basename))).unwrap();
        assert_eq!(restored_ids, v1_ids, "rollback must restore the v1 id map");
        let loaded = HnswIndex::load_with_dim(tmp.path(), basename, crate::EMBEDDING_DIM)
            .expect("previous good index must still load after rolled-back save");
        assert_eq!(loaded.len(), 2, "loaded index must be the v1 index");
    }

    #[test]
    fn test_save_and_load() {
        // This test pins the save → exists → load roundtrip. The recall check
        // (chunk1 reachable post-load) goes through the shared
        // `assert_self_match_reachable` helper, which retries the build to
        // absorb the rare degenerate concurrent-build graph (see the helper
        // docs in hnsw/mod.rs). The roundtrip itself (exists/len) is asserted
        // unconditionally on every build inside the closure.
        let build = || {
            let tmp = TempDir::new().unwrap();
            let embeddings = vec![
                ("chunk1".to_string(), make_embedding(1)),
                ("chunk2".to_string(), make_embedding(2)),
            ];
            let index = HnswIndex::build_with_dim(embeddings, crate::EMBEDDING_DIM).unwrap();
            index.save(tmp.path(), "index").unwrap();
            assert!(HnswIndex::exists(tmp.path(), "index"));
            let loaded =
                HnswIndex::load_with_dim(tmp.path(), "index", crate::EMBEDDING_DIM).unwrap();
            assert_eq!(loaded.len(), 2);
            // TempDir drops here; the loaded index is self-contained.
            loaded
        };
        crate::hnsw::assert_self_match_reachable(build, &make_embedding(1), "chunk1", 2);
    }

    // ===== multi-model dim-threading (HNSW persist) =====

    /// Create a deterministic normalized embedding of arbitrary dimension.
    fn make_embedding_dim(seed: u32, dim: usize) -> crate::embedder::Embedding {
        let mut v = vec![0.0f32; dim];
        for (i, val) in v.iter_mut().enumerate() {
            *val = ((seed as f32 * 0.1) + (i as f32 * 0.001)).sin();
        }
        let norm: f32 = v.iter().map(|x| x * x).sum::<f32>().sqrt();
        if norm > 0.0 {
            for val in &mut v {
                *val /= norm;
            }
        }
        crate::embedder::Embedding::new(v)
    }

    #[test]
    fn tc31_save_and_load_with_dim_1024() {
        // Save a 1024-dim HNSW index, load with load_with_dim(1024),
        // verify it loads and searches correctly.
        //
        // The save/load roundtrip (dim + len) is asserted on every build
        // inside the closure. The recall check (vec1 reachable post-load)
        // goes through the shared `assert_self_match_reachable` helper, which
        // retries the build to absorb the rare degenerate concurrent-build
        // graph (see the helper docs in hnsw/mod.rs): the five sin-derived
        // vectors are near-parallel (adjacent seeds ~0.995 cosine), making
        // them especially exposed to it.
        let basename = "test_1024";
        let build = || {
            let tmp = TempDir::new().unwrap();
            let embeddings: Vec<(String, crate::embedder::Embedding)> = (1..=5)
                .map(|i| (format!("vec{}", i), make_embedding_dim(i, 1024)))
                .collect();
            let index = HnswIndex::build_with_dim(embeddings, 1024).unwrap();
            assert_eq!(index.dim, 1024);
            index.save(tmp.path(), basename).unwrap();

            let loaded = HnswIndex::load_with_dim(tmp.path(), basename, 1024).unwrap();
            assert_eq!(loaded.len(), 5, "Loaded index should have 5 vectors");
            assert_eq!(loaded.dim, 1024, "Loaded index dim should be 1024");
            // TempDir drops here; the loaded index is self-contained.
            loaded
        };
        crate::hnsw::assert_self_match_reachable(build, &make_embedding_dim(1, 1024), "vec1", 3);
    }

    #[test]
    fn tc31_load_with_wrong_dim_data_size_rejected() {
        // Build with dim=1024, try to load with a much smaller dim.
        // The size check: expected_max_data = id_map.len() * dim * sizeof(f32) * 2
        // We use dim=128 for the load so the expected_max is small enough that
        // the actual data file (sized for 1024-dim vectors + HNSW overhead)
        // exceeds it, triggering the "data file too large" error.
        //
        // Math: 20 vectors * 128 * 4 * 2 = 20,480 expected_max
        //       Actual file: 20 * 1024 * 4 + HNSW overhead >> 20,480
        let tmp = TempDir::new().unwrap();
        let basename = "test_dim_mismatch";

        let embeddings: Vec<(String, crate::embedder::Embedding)> = (1..=20)
            .map(|i| (format!("vec{}", i), make_embedding_dim(i, 1024)))
            .collect();

        let index = HnswIndex::build_with_dim(embeddings, 1024).unwrap();
        index.save(tmp.path(), basename).unwrap();

        // Load with much smaller dim — expected_max_data will be far too small
        let result = HnswIndex::load_with_dim(tmp.path(), basename, 128);
        assert!(
            result.is_err(),
            "Loading 1024-dim index with dim=128 should fail due to data size mismatch"
        );
        let err_msg = match result {
            Err(e) => format!("{}", e),
            Ok(_) => panic!("Expected error, got Ok"),
        };
        assert!(
            err_msg.contains("too large"),
            "Error should mention data file size: {}",
            err_msg
        );
    }

    /// `try_load_base_with_ef` returns `None` when the index_base files
    /// don't exist. The caller treats this as "fall back to enriched index".
    #[test]
    fn test_try_load_base_returns_none_when_missing() {
        let tmp = TempDir::new().unwrap();
        // No index_base.* files written; only the enriched index.
        let embeddings: Vec<(String, crate::embedder::Embedding)> = (1..=10)
            .map(|i| (format!("vec{}", i), make_embedding(i)))
            .collect();
        let index = HnswIndex::build_with_dim(embeddings, crate::EMBEDDING_DIM).unwrap();
        index.save(tmp.path(), "index").unwrap();

        // Enriched load should succeed.
        let enriched = HnswIndex::try_load_with_ef(tmp.path(), None, crate::EMBEDDING_DIM);
        assert!(enriched.is_some(), "enriched HNSW should load");

        // Base load should return None — no index_base.* files exist.
        let base = HnswIndex::try_load_base_with_ef(tmp.path(), None, crate::EMBEDDING_DIM);
        assert!(
            base.is_none(),
            "base HNSW should return None when index_base files are absent"
        );
    }

    /// `try_load_base_with_ef` succeeds when index_base files exist.
    /// Verifies the basename routing is correct — loading "index_base" when
    /// the base files are present and "index" when only enriched is present.
    #[test]
    fn test_try_load_base_loads_when_present() {
        let tmp = TempDir::new().unwrap();
        let embeddings: Vec<(String, crate::embedder::Embedding)> = (1..=10)
            .map(|i| (format!("vec{}", i), make_embedding(i)))
            .collect();
        let index = HnswIndex::build_with_dim(embeddings, crate::EMBEDDING_DIM).unwrap();
        index.save(tmp.path(), "index_base").unwrap();

        // Base load succeeds.
        let base = HnswIndex::try_load_base_with_ef(tmp.path(), None, crate::EMBEDDING_DIM);
        assert!(base.is_some(), "base HNSW should load when files present");
        assert_eq!(base.unwrap().len(), 10);

        // Enriched should still return None — only the base files exist.
        let enriched = HnswIndex::try_load_with_ef(tmp.path(), None, crate::EMBEDDING_DIM);
        assert!(
            enriched.is_none(),
            "enriched should return None when only index_base files exist"
        );
    }

    /// A backup-rename failure during save MUST bubble up rather than being
    /// warning-and-continued. Swallowing the error would leave the rollback
    /// path unable to restore the original file because no `.bak` had ever
    /// been created — silently losing the prior index.
    ///
    /// Force the failure by pre-creating a `.bak` path as a non-empty directory
    /// (Linux `rename(file, dir)` returns EISDIR / ENOTDIR depending on kernel,
    /// either way an error). The save must surface that error and leave the
    /// original `index.hnsw.*` files untouched.
    #[cfg(unix)]
    #[test]
    fn test_save_propagates_backup_rename_failure() {
        let tmp = TempDir::new().unwrap();

        // Build + save a valid index so the directory has the v1 files.
        let embeddings: Vec<(String, crate::embedder::Embedding)> = (1..=5)
            .map(|i| (format!("vec{}", i), make_embedding(i)))
            .collect();
        let v1 = HnswIndex::build_with_dim(embeddings.clone(), crate::EMBEDDING_DIM).unwrap();
        v1.save(tmp.path(), "index").unwrap();

        // Snapshot the v1 graph file size so we can detect post-failure damage.
        let graph_path = tmp.path().join("index.hnsw.graph");
        let v1_graph_size = std::fs::metadata(&graph_path).unwrap().len();
        assert!(
            v1_graph_size > 0,
            "v1 graph file should be non-empty after first save"
        );

        // Pre-create a non-empty directory at the `.bak` path for one of the
        // extensions. `std::fs::rename(file, non_empty_dir)` fails with
        // ENOTDIR/EISDIR on Linux, exercising the backup-rename error branch.
        let bak_blocker = tmp.path().join("index.hnsw.graph.bak");
        std::fs::create_dir(&bak_blocker).unwrap();
        // Put a file inside so the rename can't succeed by replacing an empty dir.
        std::fs::write(bak_blocker.join("blocker"), b"x").unwrap();

        // Attempt a second save with a different (still valid) embeddings set.
        let v2_embeddings: Vec<(String, crate::embedder::Embedding)> = (1..=8)
            .map(|i| (format!("vec{}", i + 100), make_embedding(i + 100)))
            .collect();
        let v2 = HnswIndex::build_with_dim(v2_embeddings, crate::EMBEDDING_DIM).unwrap();

        let result = v2.save(tmp.path(), "index");
        assert!(
            result.is_err(),
            "save MUST surface the backup-rename failure (P2 #30)"
        );
        let err_msg = format!("{}", result.unwrap_err());
        // The stale-bak guard runs before the backup-rename pass, so a
        // pre-existing `.bak` (whether file or directory) trips the guard
        // with a "stale .bak files" message before the rename failure path
        // runs. Either error path is a valid surfacing of the underlying
        // scenario: save bails without touching the original index. Accept
        // both messages so this test pins the correctness invariant ("save
        // bails before damage") independent of which guard fires.
        assert!(
            err_msg.contains("back up")
                || err_msg.contains("backup")
                || err_msg.contains("stale .bak files"),
            "error should mention backup failure or stale-bak guard, got: {}",
            err_msg
        );

        // Original v1 graph file MUST still be intact and the same size.
        let post_size = std::fs::metadata(&graph_path).unwrap().len();
        assert_eq!(
            post_size, v1_graph_size,
            "original index file must not be touched when save bails on backup failure"
        );
    }

    /// A successful save must leave NO `.bak` files behind. This test pins
    /// it so a future refactor of the cleanup loop can't silently regress
    /// and let the next save trip the stale-bak guard.
    #[cfg(unix)]
    #[test]
    fn test_save_cleans_up_baks_on_success() {
        let tmp = TempDir::new().unwrap();

        let v1: Vec<(String, crate::embedder::Embedding)> = (1..=3)
            .map(|i| (format!("vec{}", i), make_embedding(i)))
            .collect();
        let v1_idx = HnswIndex::build_with_dim(v1, crate::EMBEDDING_DIM).unwrap();
        v1_idx.save(tmp.path(), "index").unwrap();

        let v2: Vec<(String, crate::embedder::Embedding)> = (1..=5)
            .map(|i| (format!("vec{}", i + 10), make_embedding(i + 10)))
            .collect();
        let v2_idx = HnswIndex::build_with_dim(v2, crate::EMBEDDING_DIM).unwrap();
        // Successful overwrite. The save creates `.bak` files mid-flight
        // and must clean them up before returning Ok.
        v2_idx.save(tmp.path(), "index").unwrap();

        for ext in ["hnsw.graph", "hnsw.data", "hnsw.ids", "hnsw.checksum"] {
            let bak = tmp.path().join(format!("index.{}.bak", ext));
            assert!(
                !bak.exists(),
                "successful save must remove .bak files; {} still exists",
                bak.display(),
            );
        }
    }

    /// Stale `.bak` files from a prior failed rollback must be detected at
    /// save start. Manual recovery is required so
    /// we don't silently overwrite a known-good index with a stale
    /// `.bak` on a future rollback. The guard returns
    /// `HnswError::Internal` with an actionable message naming the
    /// stale paths.
    #[cfg(unix)]
    #[test]
    fn test_save_refuses_when_stale_bak_present() {
        let tmp = TempDir::new().unwrap();

        // Plant a stale .bak file (no current index — fresh dir).
        let stale_bak = tmp.path().join("index.hnsw.graph.bak");
        std::fs::write(&stale_bak, b"stale-content").unwrap();

        let v: Vec<(String, crate::embedder::Embedding)> = (1..=3)
            .map(|i| (format!("vec{}", i), make_embedding(i)))
            .collect();
        let idx = HnswIndex::build_with_dim(v, crate::EMBEDDING_DIM).unwrap();

        let result = idx.save(tmp.path(), "index");
        assert!(
            result.is_err(),
            "save MUST refuse to start with a stale .bak present"
        );
        let err_msg = format!("{}", result.unwrap_err());
        assert!(
            err_msg.contains("stale .bak files"),
            "error must name the stale-bak guard, got: {}",
            err_msg,
        );
        assert!(
            err_msg.contains("manual recovery required"),
            "error must direct operator to manual recovery, got: {}",
            err_msg,
        );

        // Stale .bak must still be on disk — the guard does NOT clean
        // up; it surfaces the breadcrumb for the operator.
        assert!(
            stale_bak.exists(),
            "guard must NOT delete the stale .bak (it's the operator's recovery breadcrumb)"
        );
    }

    // ===== distance metric persistence =====

    /// Roundtrip: build with DotProduct → save → load → metric preserved
    /// and search still finds the self-match with a near-1 score (the
    /// embeddings are unit-norm, so self-dot ≈ 1 and `1 − dist` ≈ 1).
    /// Uses the shared retry helper so the rare degenerate concurrent-build
    /// graph can't flake the assertion; the metric roundtrip itself is
    /// asserted on every build inside the closure.
    #[test]
    fn test_metric_dot_save_load_roundtrip() {
        use crate::index::DistanceMetric;
        let basename = "test_dot_metric";
        let build = || {
            let tmp = TempDir::new().unwrap();
            let embeddings = vec![
                ("chunk1".to_string(), make_embedding(1)),
                ("chunk2".to_string(), make_embedding(2)),
            ];
            let index = HnswIndex::build_with_dim_and_metric(
                embeddings,
                crate::EMBEDDING_DIM,
                DistanceMetric::DotProduct,
            )
            .unwrap();
            assert_eq!(index.metric(), DistanceMetric::DotProduct);
            index.save(tmp.path(), basename).unwrap();

            // The meta header must be on disk and covered by the manifest.
            assert!(
                tmp.path().join(format!("{basename}.hnsw.meta")).exists(),
                "save must write the .hnsw.meta header"
            );

            let loaded =
                HnswIndex::load_with_dim(tmp.path(), basename, crate::EMBEDDING_DIM).unwrap();
            assert_eq!(
                loaded.metric(),
                DistanceMetric::DotProduct,
                "metric must survive the save/load roundtrip"
            );
            assert_eq!(loaded.len(), 2);
            // TempDir drops here; the loaded index is self-contained.
            loaded
        };
        let score =
            crate::hnsw::assert_self_match_reachable(build, &make_embedding(1), "chunk1", 2);
        assert!(
            score > 0.9,
            "unit-norm self-match under DistDot should score ≈ 1, got {score}"
        );
    }

    /// A pre-#1351 index has no `.hnsw.meta` and a 3-line checksum manifest.
    /// It must load as cosine — that is what every legacy index was built
    /// with.
    #[test]
    fn test_legacy_index_without_meta_loads_as_cosine() {
        use crate::index::DistanceMetric;
        let tmp = TempDir::new().unwrap();
        let basename = "test_legacy";

        let embeddings = vec![
            ("a".to_string(), make_embedding(1)),
            ("b".to_string(), make_embedding(2)),
        ];
        let index = HnswIndex::build_with_dim(embeddings, crate::EMBEDDING_DIM).unwrap();
        index.save(tmp.path(), basename).unwrap();

        // Simulate the legacy layout: drop the meta header and rewrite the
        // manifest with only the three legacy extensions (`write_checksums`
        // covers exactly graph/data/ids).
        std::fs::remove_file(tmp.path().join(format!("{basename}.hnsw.meta"))).unwrap();
        write_checksums(tmp.path(), basename);

        let loaded = HnswIndex::load_with_dim(tmp.path(), basename, crate::EMBEDDING_DIM)
            .expect("legacy index without meta header must load");
        assert_eq!(
            loaded.metric(),
            DistanceMetric::Cosine,
            "legacy index must load as cosine"
        );
        assert_eq!(loaded.len(), 2);
    }

    /// Loading with `CQS_DISTANCE_METRIC` explicitly set to a different
    /// metric than the stored one is a typed `MetricMismatch` error, never
    /// a silent reinterpretation.
    ///
    /// Test discipline: the env var is only ever set to "cosine" —
    /// a concurrent unlocked test loading a cosine index then still sees
    /// `requested == stored` and is unaffected. The conflict is created by
    /// the *stored* side (a DotProduct index built via the explicit-metric
    /// API). Held under `HNSW_ENV_LOCK` for the var's whole lifetime.
    #[test]
    fn test_metric_mismatch_is_typed_error() {
        use crate::hnsw::{EnvVarGuard, HNSW_ENV_LOCK};
        use crate::index::DistanceMetric;

        let tmp = TempDir::new().unwrap();
        let basename = "test_metric_mismatch";

        let _lock = HNSW_ENV_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        let embeddings = vec![
            ("a".to_string(), make_embedding(1)),
            ("b".to_string(), make_embedding(2)),
        ];
        let index = HnswIndex::build_with_dim_and_metric(
            embeddings.clone(),
            crate::EMBEDDING_DIM,
            DistanceMetric::DotProduct,
        )
        .unwrap();
        index.save(tmp.path(), basename).unwrap();

        let _var = EnvVarGuard::set("CQS_DISTANCE_METRIC", "cosine");

        // Explicit conflicting request → typed error.
        match HnswIndex::load_with_dim(tmp.path(), basename, crate::EMBEDDING_DIM) {
            Err(crate::hnsw::HnswError::MetricMismatch { stored, requested }) => {
                assert_eq!(stored, DistanceMetric::DotProduct);
                assert_eq!(requested, DistanceMetric::Cosine);
            }
            Err(other) => panic!("expected MetricMismatch, got: {other}"),
            Ok(_) => panic!("conflicting metric must not load"),
        }

        // Matching explicit request → loads fine. Same env value ("cosine"),
        // conflict-free because this index IS cosine.
        let cosine_index = HnswIndex::build_with_dim_and_metric(
            embeddings,
            crate::EMBEDDING_DIM,
            DistanceMetric::Cosine,
        )
        .unwrap();
        let basename2 = "test_metric_match";
        cosine_index.save(tmp.path(), basename2).unwrap();
        let loaded = HnswIndex::load_with_dim(tmp.path(), basename2, crate::EMBEDDING_DIM)
            .expect("matching explicit metric must load");
        assert_eq!(loaded.metric(), DistanceMetric::Cosine);
        // _var's Drop removes the env var before the lock releases.
    }

    /// The meta header is covered by the blake3 manifest: tampering with it
    /// after save must fail checksum verification before the metric is even
    /// read.
    #[test]
    fn test_tampered_meta_fails_checksum() {
        let tmp = TempDir::new().unwrap();
        let basename = "test_meta_tamper";

        let embeddings = vec![
            ("a".to_string(), make_embedding(1)),
            ("b".to_string(), make_embedding(2)),
        ];
        let index = HnswIndex::build_with_dim(embeddings, crate::EMBEDDING_DIM).unwrap();
        index.save(tmp.path(), basename).unwrap();

        // Rewrite the meta to claim a different metric WITHOUT updating the
        // manifest.
        let meta_path = tmp.path().join(format!("{basename}.hnsw.meta"));
        std::fs::write(
            &meta_path,
            r#"{"magic":"CQSHNSW","version":1,"metric":"dot"}"#,
        )
        .unwrap();

        match HnswIndex::load_with_dim(tmp.path(), basename, crate::EMBEDDING_DIM) {
            Err(HnswError::ChecksumMismatch { file, .. }) => {
                assert!(
                    file.contains("hnsw.meta"),
                    "checksum failure should point at the meta file: {file}"
                );
            }
            Err(other) => panic!("expected ChecksumMismatch, got: {other}"),
            Ok(_) => panic!("tampered meta must not load"),
        }
    }

    // ===== id_map edge cases =====
    //
    // Three id_map shapes exercise rare but reachable states after a corrupt
    // save, a user-crafted index, or a bincode deserialisation glitch:
    //
    // * duplicate string entries — the id_map is not a set, so duplicates
    //   are accepted. Pins that behaviour so a future "dedup on load"
    //   refactor is deliberate.
    // * empty string entries — the store uses non-empty chunk IDs, but the
    //   loader has no minimum-length check, so `""` in the id_map is
    //   accepted and eventually returned by `search()` as a bogus ID.
    // * NUL-byte string entries — JSON preserves NUL (` `), serde_json
    //   decodes it to a real NUL in the `String`. Pins that the loader
    //   accepts such ids today.

    /// Build a valid on-disk HNSW index, then rewrite its `.hnsw.ids` file
    /// with a crafted id_map while keeping graph + data intact. The
    /// checksum file must also be rewritten or `verify_hnsw_checksums`
    /// rejects the index before the id_map is touched.
    fn rewrite_id_map_and_checksums(dir: &Path, basename: &str, ids: &[&str]) {
        let id_map_path = dir.join(format!("{}.hnsw.ids", basename));
        let json = serde_json::to_string(ids).unwrap();
        std::fs::write(&id_map_path, json).unwrap();

        // Re-write checksums so the load path accepts the mutated id_map.
        let mut lines = Vec::new();
        for ext in &["hnsw.graph", "hnsw.data", "hnsw.ids"] {
            let path = dir.join(format!("{}.{}", basename, ext));
            let mut hasher = blake3::Hasher::new();
            let mut file = std::fs::File::open(&path).unwrap();
            std::io::copy(&mut file, &mut hasher).unwrap();
            let hash = hasher.finalize().to_hex().to_string();
            lines.push(format!("{}:{}", ext, hash));
        }
        std::fs::write(
            dir.join(format!("{}.hnsw.checksum", basename)),
            lines.join("\n"),
        )
        .unwrap();
    }

    /// Duplicate id_map entries are NOT rejected. `load_with_dim` only
    /// asserts `id_map.len() == hnsw_count`, not that the entries are
    /// unique. This test pins the current behaviour — a duplicate id_map
    /// survives load and is returned from search as-is.
    #[test]
    fn test_load_accepts_duplicate_id_map_entries() {
        let tmp = TempDir::new().unwrap();
        let basename = "test_dup";

        // Build an index with 3 unique vectors, so hnsw_count == 3.
        let embeddings = vec![
            ("a".to_string(), make_embedding(1)),
            ("b".to_string(), make_embedding(2)),
            ("c".to_string(), make_embedding(3)),
        ];
        let index = HnswIndex::build_with_dim(embeddings, crate::EMBEDDING_DIM).unwrap();
        index.save(tmp.path(), basename).unwrap();

        // Rewrite id_map so two slots point at the same chunk id. The
        // hnsw_count still matches 3, so the mismatch check doesn't fire.
        rewrite_id_map_and_checksums(tmp.path(), basename, &["a", "a", "c"]);

        let loaded = HnswIndex::load_with_dim(tmp.path(), basename, crate::EMBEDDING_DIM)
            .expect("duplicate id_map entries must not cause load failure");
        assert_eq!(loaded.len(), 3);
        // AUDIT-FOLLOWUP: if a future dedup/validation pass rejects
        // duplicates, update this assertion accordingly.
    }

    /// Empty-string id_map entries are accepted. The id_map carries
    /// `Vec<String>` and there is no "must be non-empty" check. Search
    /// eventually returns the bogus empty id to the caller.
    #[test]
    fn test_load_accepts_empty_string_id_map_entries() {
        let tmp = TempDir::new().unwrap();
        let basename = "test_empty_id";

        let embeddings = vec![
            ("a".to_string(), make_embedding(1)),
            ("b".to_string(), make_embedding(2)),
        ];
        let index = HnswIndex::build_with_dim(embeddings, crate::EMBEDDING_DIM).unwrap();
        index.save(tmp.path(), basename).unwrap();

        rewrite_id_map_and_checksums(tmp.path(), basename, &["", "b"]);

        let loaded = HnswIndex::load_with_dim(tmp.path(), basename, crate::EMBEDDING_DIM)
            .expect("empty-string id_map entries must not cause load failure");
        assert_eq!(loaded.len(), 2);
        // AUDIT-FOLLOWUP: once a non-empty-string guard lands, change to
        // `assert!(result.is_err())` with the rejection message.
    }

    /// NUL-byte id_map entries are preserved verbatim. JSON encodes NUL as
    /// ` ` and serde_json decodes it to a real NUL in `String`. The
    /// loader has no sanitation pass.
    #[test]
    fn test_load_accepts_nul_byte_id_map_entries() {
        let tmp = TempDir::new().unwrap();
        let basename = "test_nul_id";

        let embeddings = vec![
            ("a".to_string(), make_embedding(1)),
            ("b".to_string(), make_embedding(2)),
        ];
        let index = HnswIndex::build_with_dim(embeddings, crate::EMBEDDING_DIM).unwrap();
        index.save(tmp.path(), basename).unwrap();

        // A chunk id containing a NUL byte. Write via serde_json::to_string so
        // the NUL gets its ` ` escape — we can't hand-hardcode a
        // JSON-with-real-NUL byte sequence into the file.
        let nul_id = String::from("has\0nul");
        let normal_id = String::from("b");
        let ids: Vec<&str> = vec![nul_id.as_str(), normal_id.as_str()];
        rewrite_id_map_and_checksums(tmp.path(), basename, &ids);

        let loaded = HnswIndex::load_with_dim(tmp.path(), basename, crate::EMBEDDING_DIM)
            .expect("NUL-byte id_map entries must not cause load failure");
        assert_eq!(loaded.len(), 2);
        // AUDIT-FOLLOWUP: accepting NUL in chunk ids is a downstream hazard
        // (SQL queries, log lines). Once a reject-NUL guard lands, flip this
        // to `result.is_err()`.
    }

    // ===== store-state stamp: save_stamped / verify_hnsw_current =====

    /// Build a tiny saveable index for the stamp tests.
    fn small_index() -> HnswIndex {
        HnswIndex::build_with_dim(
            vec![
                ("chunk1".to_string(), make_embedding(1)),
                ("chunk2".to_string(), make_embedding(2)),
            ],
            crate::EMBEDDING_DIM,
        )
        .unwrap()
    }

    /// Advance the store's state stamp by inserting a chunk: `chunk_count`
    /// changes, so any earlier `StoreStamp` no longer matches.
    fn advance_store(store: &crate::Store, id: &str) {
        use crate::parser::{Chunk, ChunkType, Language};
        let content = format!("fn {}() {{ }}", id);
        let content_hash = blake3::hash(content.as_bytes()).to_hex().to_string();
        let chunk = Chunk {
            id: id.to_string(),
            file: std::path::PathBuf::from("src/lib.rs"),
            language: Language::Rust,
            chunk_type: ChunkType::Function,
            name: id.to_string(),
            signature: format!("fn {}()", id),
            content,
            doc: None,
            line_start: 1,
            line_end: 5,
            byte_start: 0,
            content_hash,
            canonical_hash: String::new(),
            parent_id: None,
            window_idx: None,
            parent_type_name: None,
            parser_version: 0,
        };
        store
            .upsert_chunks_batch(&[(chunk, make_embedding(42))], Some(0))
            .expect("seed chunk");
    }

    /// All five sidecar extensions a completed save promotes.
    const STAMP_TEST_EXTS: [&str; 5] = [
        "hnsw.graph",
        "hnsw.data",
        "hnsw.ids",
        "hnsw.meta",
        "hnsw.checksum",
    ];

    /// Dirty-path contract: a stamped save whose stamp still matches the
    /// live store verifies (died after the save → heal); once the store
    /// moves past the stamp, verification refuses (died after the chunk
    /// commit, before the save → stay dirty / brute-force).
    #[test]
    fn verify_current_distinguishes_save_landed_from_save_missed() {
        let (store, tmp) = crate::test_helpers::setup_store();
        let index = small_index();
        let snapshot = StoreStamp::read(&store).unwrap();
        let outcome = index
            .save_stamped(tmp.path(), "index", &store, snapshot)
            .unwrap();
        assert_eq!(outcome, SaveOutcome::Saved);

        verify_hnsw_current(tmp.path(), "index", &store)
            .expect("stamp matches the live store — the dirty flag was a false positive");

        advance_store(&store, "late_chunk");
        let err = verify_hnsw_current(tmp.path(), "index", &store)
            .expect_err("store moved past the sidecar stamp — must refuse");
        assert!(
            format!("{err}").contains("does not match the live store"),
            "want stamp-mismatch error, got: {err}"
        );
    }

    /// Enrichment-style false-heal scenario: an embedding-only rewrite
    /// (`update_embeddings_with_hashes_batch`) changes no chunk rows and no
    /// sparse vectors — `chunk_count` is unchanged — so the stamp moves only
    /// because the update bumps the write generation. A crash between that
    /// commit and the HNSW save (dirty flag set, sidecars from before the
    /// rewrite) must refuse the self-heal; healing would serve the
    /// pre-enrichment vectors as current.
    #[test]
    fn verify_current_refuses_after_embedding_only_update() {
        let (store, tmp) = crate::test_helpers::setup_store();
        // Seed a chunk first so the sidecar stamp includes it — the later
        // embedding rewrite must be the ONLY thing that moves the stamp.
        advance_store(&store, "enriched_chunk");
        let chunks_at_save = store.chunk_count().unwrap();

        let index = small_index();
        let snapshot = StoreStamp::read(&store).unwrap();
        assert_eq!(
            index
                .save_stamped(tmp.path(), "index", &store, snapshot)
                .unwrap(),
            SaveOutcome::Saved
        );
        verify_hnsw_current(tmp.path(), "index", &store).expect("precondition: stamp matches");

        // The enrichment pass: rewrite the chunk's embedding in place.
        store
            .update_embeddings_with_hashes_batch(&[(
                "enriched_chunk".to_string(),
                make_embedding(77),
                Some("enrichment_hash".to_string()),
            )])
            .unwrap();
        assert_eq!(
            store.chunk_count().unwrap(),
            chunks_at_save,
            "fixture invariant: the rewrite must not change chunk_count — \
             only the write generation may move"
        );

        let err = verify_hnsw_current(tmp.path(), "index", &store)
            .expect_err("sidecars predate the enrichment rewrite — must refuse");
        assert!(
            format!("{err}").contains("does not match the live store"),
            "want stamp-mismatch error, got: {err}"
        );
    }

    /// A sidecar set without a stamp (plain `save()`, or pre-stamp binaries)
    /// cannot prove it postdates the last chunk write — the dirty path must
    /// refuse it even though checksums pass.
    #[test]
    fn verify_current_refuses_unstamped_sidecars() {
        let (store, tmp) = crate::test_helpers::setup_store();
        small_index().save(tmp.path(), "index").unwrap();

        let err = verify_hnsw_current(tmp.path(), "index", &store)
            .expect_err("unstamped sidecars must refuse on the dirty path");
        assert!(
            format!("{err}").contains("no store-state stamp"),
            "want missing-stamp error, got: {err}"
        );
    }

    /// The stale-rebuild seam: store generation bumps between build-snapshot
    /// capture and save → the save is discarded under the lock and nothing
    /// lands on disk.
    #[test]
    fn save_stamped_discards_when_store_moves_past_snapshot() {
        let (store, tmp) = crate::test_helpers::setup_store();
        let index = small_index();
        let snapshot = StoreStamp::read(&store).unwrap();

        // The unobserved writer: commits a chunk after the snapshot was
        // captured but before the save runs.
        advance_store(&store, "raced_in");

        let outcome = index
            .save_stamped(tmp.path(), "index", &store, snapshot)
            .unwrap();
        assert_eq!(
            outcome,
            SaveOutcome::DiscardedStale,
            "a save built from a superseded snapshot must be discarded"
        );
        assert!(
            !HnswIndex::exists(tmp.path(), "index"),
            "a discarded save must not write sidecars"
        );
    }

    /// Post-success debris: a kill between the last promote and the .bak
    /// cleanup leaves a healthy stamped live set plus orphan .baks. The next
    /// save must treat the .baks as debris (delete, proceed) instead of
    /// wedging persistence with a manual-recovery refusal.
    #[test]
    fn save_stamped_clears_post_success_bak_debris() {
        let (store, tmp) = crate::test_helpers::setup_store();
        let v1 = small_index();
        let snapshot = StoreStamp::read(&store).unwrap();
        assert_eq!(
            v1.save_stamped(tmp.path(), "index", &store, snapshot)
                .unwrap(),
            SaveOutcome::Saved
        );

        // Simulate the kill window: every promoted file also has a .bak left
        // behind (contents don't matter to the guard — only the live set's
        // health does).
        for ext in STAMP_TEST_EXTS {
            let live = tmp.path().join(format!("index.{ext}"));
            let bak = tmp.path().join(format!("index.{ext}.bak"));
            std::fs::copy(&live, &bak).unwrap();
        }

        let v2 = small_index();
        let snapshot2 = StoreStamp::read(&store).unwrap();
        assert_eq!(
            v2.save_stamped(tmp.path(), "index", &store, snapshot2)
                .unwrap(),
            SaveOutcome::Saved,
            "healthy stamped live set + .baks = debris; the save must proceed"
        );
        for ext in STAMP_TEST_EXTS {
            let bak = tmp.path().join(format!("index.{ext}.bak"));
            assert!(!bak.exists(), "debris {} must be cleaned", bak.display());
        }
        let loaded =
            HnswIndex::load_with_dim(tmp.path(), "index", crate::EMBEDDING_DIM).expect("load v2");
        assert_eq!(loaded.len(), 2);
    }

    /// When .baks exist and the live set does NOT verify, the .baks may be
    /// the only good copy — the save must still hard-refuse and leave them
    /// in place.
    #[test]
    fn save_stamped_refuses_baks_when_live_set_invalid() {
        let (store, tmp) = crate::test_helpers::setup_store();
        let v1 = small_index();
        let snapshot = StoreStamp::read(&store).unwrap();
        assert_eq!(
            v1.save_stamped(tmp.path(), "index", &store, snapshot)
                .unwrap(),
            SaveOutcome::Saved
        );
        for ext in STAMP_TEST_EXTS {
            let live = tmp.path().join(format!("index.{ext}"));
            let bak = tmp.path().join(format!("index.{ext}.bak"));
            std::fs::copy(&live, &bak).unwrap();
        }
        // Corrupt the live graph so verification fails.
        std::fs::File::create(tmp.path().join("index.hnsw.graph"))
            .unwrap()
            .set_len(0)
            .unwrap();

        let v2 = small_index();
        let snapshot2 = StoreStamp::read(&store).unwrap();
        let err = v2
            .save_stamped(tmp.path(), "index", &store, snapshot2)
            .expect_err("invalid live set + .baks must refuse");
        assert!(
            format!("{err}").contains("manual recovery"),
            "want manual-recovery refusal, got: {err}"
        );
        for ext in STAMP_TEST_EXTS {
            let bak = tmp.path().join(format!("index.{ext}.bak"));
            assert!(
                bak.exists(),
                "breadcrumb {} must be preserved",
                bak.display()
            );
        }
    }

    /// Plain (store-less) `save()` cannot validate the live set's stamp, so
    /// any .bak keeps the conservative hard refusal.
    #[test]
    fn plain_save_still_refuses_any_baks() {
        let tmp = TempDir::new().unwrap();
        let v1 = small_index();
        v1.save(tmp.path(), "index").unwrap();
        let live = tmp.path().join("index.hnsw.ids");
        std::fs::copy(&live, tmp.path().join("index.hnsw.ids.bak")).unwrap();

        let err = small_index()
            .save(tmp.path(), "index")
            .expect_err("store-less save must refuse while .baks exist");
        assert!(
            format!("{err}").contains("manual recovery"),
            "want manual-recovery refusal, got: {err}"
        );
    }

    /// The meta header roundtrips the stamp, and pre-stamp sidecars parse
    /// with `store_stamp() == None` (clean loads keep working).
    #[test]
    fn meta_stamp_roundtrip_and_legacy_compat() {
        let (store, tmp) = crate::test_helpers::setup_store();
        let snapshot = StoreStamp::read(&store).unwrap();
        small_index()
            .save_stamped(tmp.path(), "index", &store, snapshot)
            .unwrap();
        let meta = read_hnsw_meta(tmp.path(), "index")
            .unwrap()
            .expect("meta written");
        assert_eq!(meta.store_stamp(), Some(snapshot));

        // Legacy (pre-stamp) meta: only magic/version/metric.
        let legacy = serde_json::json!({
            "magic": HNSW_META_MAGIC,
            "version": HNSW_META_VERSION,
            "metric": "cosine",
        });
        std::fs::write(
            tmp.path().join("legacy.hnsw.meta"),
            serde_json::to_vec(&legacy).unwrap(),
        )
        .unwrap();
        let legacy_meta = read_hnsw_meta(tmp.path(), "legacy")
            .unwrap()
            .expect("legacy meta parses");
        assert_eq!(legacy_meta.store_stamp(), None);
        assert_eq!(
            read_hnsw_meta_metric(tmp.path(), "legacy").unwrap(),
            DistanceMetric::Cosine
        );
    }

    // ===== property tests: HnswMeta save→load codec + the stamp/version
    // contracts =====
    //
    // The example tests above pin hand-picked values: one fixed stamp from
    // `setup_store()`, the (Some,Some) and (None,None) presence shapes, the
    // `dot`/`cosine` metrics, `version == 1`. The structural null is the rest
    // of the field space — boundary stamp values (0, 1, u64::MAX), the two
    // *half-present* presence shapes (the hand-edited-meta case the
    // `store_stamp()` doc calls out), version boundaries around
    // `HNSW_META_VERSION` (including 0 and u32::MAX), and arbitrary metric /
    // magic strings. The properties below drive the REAL functions
    // (`serde_json` serialize == the save path's encoder; `read_hnsw_meta` /
    // `read_hnsw_meta_metric` == the load path's decoder + validators) over a
    // generator that lands deliberately on those boundaries.
    mod properties {
        use super::*;
        use proptest::prelude::*;

        /// Serialize a typed `HnswMeta` exactly as `save_impl` does
        /// (`serde_json::to_vec`), write it to the basename's `.hnsw.meta`
        /// path under `dir`, and return that path's parent for the readers.
        fn write_meta_json(dir: &Path, basename: &str, meta: &HnswMeta) {
            let path = dir.join(format!("{}.hnsw.meta", basename));
            let bytes = serde_json::to_vec(meta).expect("HnswMeta serializes");
            std::fs::write(&path, bytes).expect("write meta");
        }

        /// Generator coverage claim: every (chunk_count, splade_generation)
        /// presence pairing — (None,None), (Some,None), (None,Some),
        /// (Some,Some) — at stamp values drawn from the boundary set
        /// {0, 1, u64::MAX} unioned with a uniform u64. Distrust: proptest
        /// weights the explicit boundary set so the two half-present shapes
        /// (which `store_stamp()` must collapse to `None`) and the u64::MAX
        /// edge are not left to uniform luck.
        fn stamp_field() -> impl Strategy<Value = Option<u64>> {
            prop_oneof![
                Just(None),
                prop_oneof![Just(0u64), Just(1u64), Just(u64::MAX), any::<u64>(),].prop_map(Some),
            ]
        }

        /// Generator coverage claim: `version` lands on the validation
        /// boundary deliberately — 0 (below current), HNSW_META_VERSION
        /// (the accept/reject seam), HNSW_META_VERSION+1 (first rejected),
        /// u32::MAX (max) — plus a uniform u32 so no in-between value is
        /// unreachable.
        fn version_field() -> impl Strategy<Value = u32> {
            prop_oneof![
                Just(0u32),
                Just(HNSW_META_VERSION),
                Just(HNSW_META_VERSION.saturating_add(1)),
                Just(u32::MAX),
                any::<u32>(),
            ]
        }

        /// Generator coverage claim: `magic` is the correct constant about
        /// half the time and arbitrary unicode (incl. the empty string and
        /// near-misses) otherwise, so the magic-gate's accept and reject arms
        /// both get hit.
        fn magic_field() -> impl Strategy<Value = String> {
            prop_oneof![
                Just(HNSW_META_MAGIC.to_string()),
                Just(String::new()),
                Just("cqshnsw".to_string()),  // case near-miss
                Just("CQSHNSW ".to_string()), // trailing-space near-miss
                ".*",                         // arbitrary, incl. unicode
            ]
        }

        /// Generator coverage claim: `metric` covers every accepted spelling
        /// `FromStr` recognizes (so the metric round-trips through
        /// `read_hnsw_meta_metric`), the canonical persisted spellings, and
        /// arbitrary garbage (so the typed-error arm is exercised).
        fn metric_field() -> impl Strategy<Value = String> {
            prop_oneof![
                Just("cosine".to_string()),
                Just("dot".to_string()),
                Just("dotproduct".to_string()),
                Just("inner_product".to_string()),
                Just("ip".to_string()),
                Just("  Cosine  ".to_string()), // whitespace + case (FromStr trims/lowercases)
                ".*",                           // arbitrary, incl. unicode garbage
            ]
        }

        /// A fully-arbitrary, always-well-formed `HnswMeta` (correct magic +
        /// supported version + parseable metric). Used by the round-trip and
        /// stamp properties where the validators must ACCEPT every input, so
        /// the generator stays inside the valid region but still sweeps the
        /// full stamp boundary set.
        fn valid_meta() -> impl Strategy<Value = HnswMeta> {
            (
                prop_oneof![Just("cosine".to_string()), Just("dot".to_string())],
                0u32..=HNSW_META_VERSION, // supported versions only
                stamp_field(),
                stamp_field(),
            )
                .prop_map(|(metric, version, chunk_count, splade_generation)| {
                    HnswMeta {
                        magic: HNSW_META_MAGIC.to_string(),
                        version,
                        metric,
                        chunk_count,
                        splade_generation,
                    }
                })
        }

        proptest! {
            // Determinism: proptest's default RNG is seeded and a falsifier is
            // pinned in `.proptest-regressions`; we additionally pin the
            // boundary cases below as explicit `#[test]`s.

            /// Invariant (round-trip / codec):
            ///   read(write(meta)) ≡ meta   ∀ well-formed meta.
            /// The save path encodes with `serde_json::to_vec`; the load path
            /// decodes with `serde_json::from_slice` inside `read_hnsw_meta`.
            /// This drives both real functions and asserts byte-faithful field
            /// reconstruction — the `#[serde(default)] Option<u64>` fields are
            /// the bite risk (a `None` that decodes as `Some(0)`, or a stamp
            /// dropped, would silently break the staleness guard).
            #[test]
            fn prop_meta_round_trips(meta in valid_meta()) {
                let tmp = TempDir::new().unwrap();
                write_meta_json(tmp.path(), "rt", &meta);
                let loaded = read_hnsw_meta(tmp.path(), "rt")
                    .expect("well-formed meta must read")
                    .expect("meta file exists");
                prop_assert_eq!(loaded, meta);
            }

            /// Invariant (stamp pairing):
            ///   store_stamp() = Some{cc,sg}  iff  chunk_count=Some(cc) ∧ splade_generation=Some(sg)
            ///                 = None          otherwise (incl. both half-present shapes).
            /// Driven through the persisted form so the `#[serde(default)]`
            /// decode of half-present pairs is the actual subject. The example
            /// suite only covers (Some,Some) at one fixed stamp and (None,None).
            #[test]
            fn prop_store_stamp_full_pairing(
                cc in stamp_field(),
                sg in stamp_field(),
            ) {
                let meta = HnswMeta {
                    magic: HNSW_META_MAGIC.to_string(),
                    version: HNSW_META_VERSION,
                    metric: "cosine".to_string(),
                    chunk_count: cc,
                    splade_generation: sg,
                };
                let tmp = TempDir::new().unwrap();
                write_meta_json(tmp.path(), "sp", &meta);
                let loaded = read_hnsw_meta(tmp.path(), "sp").unwrap().unwrap();
                let expected = match (cc, sg) {
                    (Some(chunk_count), Some(splade_generation)) => {
                        Some(StoreStamp { chunk_count, splade_generation })
                    }
                    _ => None,
                };
                prop_assert_eq!(loaded.store_stamp(), expected);
            }

            /// Invariant (version/magic gate — the generation-stamp seam):
            ///   read_hnsw_meta accepts  iff  magic == HNSW_META_MAGIC ∧ version ≤ HNSW_META_VERSION.
            /// A meta from a *newer* generation (version > this binary) must be
            /// REJECTED, never silently used as-current — that is exactly the
            /// sidecar-from-a-future-generation class. Both arms are generated.
            #[test]
            fn prop_version_magic_gate(
                magic in magic_field(),
                version in version_field(),
                metric in metric_field(),
                cc in stamp_field(),
                sg in stamp_field(),
            ) {
                let meta = HnswMeta {
                    magic: magic.clone(),
                    version,
                    metric,
                    chunk_count: cc,
                    splade_generation: sg,
                };
                let tmp = TempDir::new().unwrap();
                write_meta_json(tmp.path(), "vg", &meta);
                let result = read_hnsw_meta(tmp.path(), "vg");
                let should_accept = magic == HNSW_META_MAGIC && version <= HNSW_META_VERSION;
                if should_accept {
                    prop_assert!(
                        result.is_ok() && result.as_ref().unwrap().is_some(),
                        "magic+version valid → must accept; got {:?}",
                        result.err()
                    );
                } else {
                    // A mismatched magic or future version is a typed error —
                    // never a silent Ok that the load path would treat as a
                    // current sidecar.
                    prop_assert!(
                        result.is_err(),
                        "magic={:?} version={} should reject (magic_ok={}, ver_ok={})",
                        magic,
                        version,
                        magic == HNSW_META_MAGIC,
                        version <= HNSW_META_VERSION,
                    );
                }
            }

            /// Invariant (metric round-trip):
            ///   read_hnsw_meta_metric(write(meta)) = meta.metric.parse::<DistanceMetric>()
            /// whenever magic+version are valid. A parseable metric survives;
            /// an unparseable one is a typed error (never a silent cosine).
            #[test]
            fn prop_metric_round_trip(metric in metric_field()) {
                let meta = HnswMeta {
                    magic: HNSW_META_MAGIC.to_string(),
                    version: HNSW_META_VERSION,
                    metric: metric.clone(),
                    chunk_count: None,
                    splade_generation: None,
                };
                let tmp = TempDir::new().unwrap();
                write_meta_json(tmp.path(), "mr", &meta);
                let got = read_hnsw_meta_metric(tmp.path(), "mr");
                match metric.parse::<DistanceMetric>() {
                    Ok(expected) => {
                        prop_assert_eq!(
                            got.expect("parseable metric must read"),
                            expected
                        );
                    }
                    Err(_) => {
                        prop_assert!(
                            got.is_err(),
                            "unparseable metric {:?} must be a typed error, not a silent cosine",
                            metric
                        );
                    }
                }
            }

            /// Invariant (generation-mismatch self-heal — the headline target):
            ///   verify_hnsw_current Ok  iff  the persisted full stamp == the live store stamp.
            /// For ANY (saved, live) stamp pair and ANY presence shape, the
            /// dirty-path check accepts ONLY when a complete stamp is present
            /// AND equals the live store. A persisted stamp from a *different*
            /// generation (saved ≠ live) must be refused — silently accepting
            /// it is the sidecar-from-previous-generation bug. The example
            /// suite drives this with the store's *actual* generation; here we
            /// forge the persisted stamp directly so saved>live, saved<live,
            /// saved==live, and the half/empty shapes are all reachable.
            ///
            /// `setup_store()` gives a fresh store at a fixed live stamp; we
            /// read it once and forge persisted stamps around it.
            #[test]
            fn prop_verify_current_accepts_iff_stamp_equals_live(
                saved_cc in stamp_field(),
                saved_sg in stamp_field(),
            ) {
                let (store, tmp) = crate::test_helpers::setup_store();
                let live = StoreStamp::read(&store).unwrap();

                // Forge a verified sidecar SET (graph/data/ids/checksum) so
                // `verify_hnsw_checksums` passes — then the only thing left to
                // decide the verdict is the stamp comparison. Build+save a real
                // tiny index, then overwrite ONLY the meta with the forged
                // stamp and refresh its checksum line.
                small_index().save(tmp.path(), "index").unwrap();
                let forged = HnswMeta {
                    magic: HNSW_META_MAGIC.to_string(),
                    version: HNSW_META_VERSION,
                    metric: "cosine".to_string(),
                    chunk_count: saved_cc,
                    splade_generation: saved_sg,
                };
                write_meta_json(tmp.path(), "index", &forged);
                // Rebuild the checksum manifest over the (now-forged) meta plus
                // the real graph/data/ids so checksum verification still passes
                // and the stamp comparison is the sole discriminator.
                rewrite_all_checksums(tmp.path(), "index");

                let result = verify_hnsw_current(tmp.path(), "index", &store);

                let saved_stamp = match (saved_cc, saved_sg) {
                    (Some(c), Some(g)) => Some(StoreStamp {
                        chunk_count: c,
                        splade_generation: g,
                    }),
                    _ => None,
                };
                let should_accept = saved_stamp == Some(live);
                if should_accept {
                    prop_assert!(
                        result.is_ok(),
                        "stamp equals live → must heal; got {:?}",
                        result.err()
                    );
                } else {
                    prop_assert!(
                        result.is_err(),
                        "saved stamp {:?} ≠ live {:?} (or absent) → must refuse, not silently heal",
                        saved_stamp,
                        live,
                    );
                }
            }
        }

        /// Rewrite the blake3 checksum manifest over the four content files
        /// (graph/data/ids/meta) as they currently sit on disk. Used by the
        /// `verify_hnsw_current` property after forging the meta so the
        /// checksum pass stays green and the stamp comparison is the sole
        /// discriminator.
        fn rewrite_all_checksums(dir: &Path, basename: &str) {
            let mut lines = Vec::new();
            for ext in &["hnsw.ids", "hnsw.graph", "hnsw.data", "hnsw.meta"] {
                let path = dir.join(format!("{}.{}", basename, ext));
                if path.exists() {
                    let mut hasher = blake3::Hasher::new();
                    let mut file = std::fs::File::open(&path).unwrap();
                    std::io::copy(&mut file, &mut hasher).unwrap();
                    lines.push(format!("{}:{}", ext, hasher.finalize().to_hex()));
                }
            }
            std::fs::write(
                dir.join(format!("{}.hnsw.checksum", basename)),
                lines.join("\n"),
            )
            .unwrap();
        }

        // ===== pinned boundary regressions =====
        //
        // The boundary cases the properties sweep, pinned as explicit examples
        // so a generator change can never silently stop covering them. These
        // are the values a hand-written suite would have to enumerate by hand;
        // they are kept here as the durable floor under the generated search.

        /// version == HNSW_META_VERSION is the accept/reject seam: it must be
        /// accepted (the boundary is `>`, not `>=`).
        #[test]
        fn boundary_version_at_limit_accepts() {
            let tmp = TempDir::new().unwrap();
            let meta = HnswMeta {
                magic: HNSW_META_MAGIC.to_string(),
                version: HNSW_META_VERSION,
                metric: "cosine".to_string(),
                chunk_count: Some(u64::MAX),
                splade_generation: Some(0),
            };
            write_meta_json(tmp.path(), "b", &meta);
            let loaded = read_hnsw_meta(tmp.path(), "b").unwrap().unwrap();
            assert_eq!(
                loaded.store_stamp(),
                Some(StoreStamp {
                    chunk_count: u64::MAX,
                    splade_generation: 0
                })
            );
        }

        /// version == HNSW_META_VERSION + 1 is the first rejected value: a
        /// sidecar written by a newer binary must be refused, not used.
        #[test]
        fn boundary_version_above_limit_rejects() {
            let tmp = TempDir::new().unwrap();
            let meta = HnswMeta {
                magic: HNSW_META_MAGIC.to_string(),
                version: HNSW_META_VERSION + 1,
                metric: "cosine".to_string(),
                chunk_count: None,
                splade_generation: None,
            };
            write_meta_json(tmp.path(), "b", &meta);
            let err =
                read_hnsw_meta(tmp.path(), "b").expect_err("future-version sidecar must reject");
            assert!(format!("{err}").contains("newer than this binary"));
        }

        /// Both half-present stamp shapes collapse to `None` — a hand-edited
        /// meta carrying only one of the pair cannot prove a generation.
        #[test]
        fn boundary_half_present_stamp_is_none() {
            for (cc, sg) in [(Some(7u64), None), (None, Some(9u64))] {
                let tmp = TempDir::new().unwrap();
                let meta = HnswMeta {
                    magic: HNSW_META_MAGIC.to_string(),
                    version: HNSW_META_VERSION,
                    metric: "cosine".to_string(),
                    chunk_count: cc,
                    splade_generation: sg,
                };
                write_meta_json(tmp.path(), "b", &meta);
                let loaded = read_hnsw_meta(tmp.path(), "b").unwrap().unwrap();
                assert_eq!(
                    loaded.store_stamp(),
                    None,
                    "half-present pair {cc:?}/{sg:?} must not yield a stamp"
                );
            }
        }
    }
}
