//! CAGRA GPU-accelerated vector search
//!
//! Uses NVIDIA cuVS for GPU-accelerated nearest neighbor search.
//! Only available when compiled with the `cuda-index` feature.
//!
//! ## Usage
//!
//! CAGRA indexes are built from embeddings at runtime and can be persisted to
//! disk via [`CagraIndex::save`] / [`CagraIndex::load`] (cuVS native serialize
//! plus a small JSON sidecar with our metadata and a blake3 checksum).
//!
//! When GPU is available and this feature is enabled, CAGRA provides
//! faster search than CPU-based HNSW for large indexes.
//!
//! ## Ownership Model
//!
//! The cuVS `search()` method takes `&self` (non-consuming). The index is
//! built once and reused for all searches.
//!
//! ## Persistence
//!
//! The persisted form is two files next to each other:
//!
//! - `{cqs_dir}/index.cagra`      — binary blob written by `cuvsCagraSerialize`
//! - `{cqs_dir}/index.cagra.meta` — JSON sidecar: magic, version, dim,
//!   chunk_count, splade_generation (coarse staleness check), id_map,
//!   and a blake3 checksum over the `.cagra` blob.
//!
//! On load we:
//!   1. Parse the sidecar and verify magic + version.
//!   2. Check dim and chunk_count match the current store; bail out to a
//!      rebuild if either has drifted.
//!   3. Verify the blake3 checksum over the `.cagra` blob to catch corruption.
//!   4. Call `cuvsCagraDeserialize` to reconstitute the GPU index.
//!
//! Any failure logs a warn and returns `Err`, and the caller rebuilds from
//! the store. The save path warn-logs failures (non-fatal); we rebuild on
//! next startup.
//!
//! Set `CQS_CAGRA_PERSIST=0` to disable save+load entirely (A/B testing or
//! reducing on-disk footprint). Default: enabled.

#[cfg(feature = "cuda-index")]
use std::path::Path;
#[cfg(feature = "cuda-index")]
use std::sync::atomic::{AtomicBool, Ordering};
#[cfg(feature = "cuda-index")]
use std::sync::Mutex;

#[cfg(feature = "cuda-index")]
use ndarray_015::Array2;

#[cfg(feature = "cuda-index")]
use thiserror::Error;

#[cfg(feature = "cuda-index")]
use crate::embedder::Embedding;
#[cfg(feature = "cuda-index")]
use crate::index::{DistanceMetric, IndexResult, VectorIndex};

/// True when every component of `query` is finite (no NaN/Inf).
///
/// Extracted as a free function so the `CagraIndex::search` finite-query
/// guard is unit-testable without a GPU — constructing a `CagraIndex`
/// requires live cuVS resources, but the guard predicate itself is pure
/// CPU. Mirrors the up-front check on the HNSW search path.
#[cfg(feature = "cuda-index")]
fn query_is_finite(query: &Embedding) -> bool {
    query.as_slice().iter().all(|v| v.is_finite())
}

/// On-disk magic bytes for the CAGRA sidecar. Changes force a rebuild.
#[cfg(feature = "cuda-index")]
const CAGRA_META_MAGIC: &str = "CAGRA01";

/// On-disk version for the CAGRA sidecar. Bump when adding fields that an
/// older binary can't parse; the parse-fail path falls through to rebuild.
#[cfg(feature = "cuda-index")]
const CAGRA_META_VERSION: u32 = 1;

/// Sentinel distance marking an output slot cuVS did not write.
///
/// # Why a sentinel?
///
/// cuVS does not promise to fill the neighbor / distance buffers beyond
/// `index.len()` rows. When we request `k` neighbors against an index with
/// fewer than `k` vectors — or against a filtered set smaller than `k` —
/// the kernel writes exactly `index.len()` (or `|filter|`) real
/// `(neighbor, distance)` pairs and leaves the remaining slots untouched.
///
/// A zero-initialized output buffer decodes those untouched slots as
/// `(chunk_id = 0, distance = 0.0)` → `score = 1.0`, emitting phantom
/// perfect-match hits pointing at whichever chunk happens to hold internal
/// index 0.
///
/// The fix is to pre-fill `distances_host` with this sentinel before the
/// kernel launch and drop any slot whose distance still holds it after
/// copy-back. CAGRA writes finite distances for every real hit — squared-L2
/// values are non-negative, InnerProduct (raw dot product, #1351) values may
/// be negative but are still finite — so the `+∞` sentinel is unambiguous
/// for both metrics.
///
/// # Why `f32::INFINITY` specifically?
///
/// - Distinct from any real squared-L2 distance cuVS produces.
/// - `!dist.is_finite()` also captures any NaN a future cuVS release
///   might emit on exotic inputs, giving us a zero-cost second line of
///   defense without a dedicated branch.
/// - Cheap to initialise — [`ndarray::Array2::from_elem`] materialises
///   the constant once per query; no broadcast or copy overhead.
///
/// `f32::NAN` was considered and rejected: `dist == NAN` is always false
/// (NaN is not comparable), so the sentinel check would have to use
/// `is_nan()` exclusively, losing the "real distances are finite"
/// structural guarantee.
///
/// # cuVS API audit
///
/// Two upstream mechanisms would make this sentinel unnecessary:
///   1. A `fill_with_invalid` (or equivalent) option on
///      [`cuvs::cagra::SearchParams`] that pre-fills unused rows.
///   2. An `n_valid_results` output field on the search call.
///
/// Neither exists: `SearchParams` exposes only `itopk_size`,
/// `max_queries`, `max_iterations`, algo/team/block tuning, and hashmap
/// knobs. The search entrypoint returns `Result<()>` with no per-row
/// validity information. Re-audit this when bumping the `cuvs` pin; if
/// either mechanism lands upstream, the sentinel scheme can be removed
/// in favour of the native API.
#[cfg(feature = "cuda-index")]
const INVALID_DISTANCE: f32 = f32::INFINITY;

#[cfg(feature = "cuda-index")]
#[derive(Error, Debug)]
pub enum CagraError {
    #[error("cuVS error: {0}")]
    Cuvs(String),
    #[error("No GPU available")]
    NoGpu,
    #[error("Dimension mismatch: expected {expected}, got {actual}")]
    DimensionMismatch { expected: usize, actual: usize },
    #[error("Build error: {0}")]
    Build(String),
    #[error("Index not built")]
    NotBuilt,
    #[error("Persistence IO error: {0}")]
    Io(String),
    #[error("Persistence metadata invalid: {0}")]
    BadMeta(String),
    #[error("Persisted CAGRA index is stale ({reason})")]
    Stale { reason: String },
    #[error("Persisted CAGRA index checksum mismatch (file: {0})")]
    ChecksumMismatch(String),
}

/// Configurable CAGRA CPU memory cap via `CQS_CAGRA_MAX_BYTES` env var.
/// Defaults to 2GB. Cached in OnceLock for single parse.
#[cfg(feature = "cuda-index")]
fn cagra_max_bytes() -> usize {
    static MAX: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
    *MAX.get_or_init(|| {
        std::env::var("CQS_CAGRA_MAX_BYTES")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(2 * 1024 * 1024 * 1024)
    })
}

/// Configurable CAGRA streaming batch size for `build_from_store_with_metric`, scaled by
/// `dim` and overridable via `CQS_CAGRA_STREAM_BATCH_SIZE` (which wins
/// verbatim). The base 10_000 at dim=1024 is a 40 MB allocation per batch
/// (10_000 × 1024 × 4 bytes). Higher-dim models shrink it to keep per-batch
/// heap bounded; lower-dim models (E5-base, dim=768) grow it for fewer SQL
/// round trips.
#[cfg(feature = "cuda-index")]
fn cagra_stream_batch_size(dim: usize) -> usize {
    if let Ok(val) = std::env::var("CQS_CAGRA_STREAM_BATCH_SIZE") {
        if let Ok(n) = val.parse::<usize>() {
            if n > 0 {
                return n;
            }
        }
    }
    crate::limits::dim_scaled_batch(10_000, dim, 500, 50_000)
}

/// Scale `itopk_max` ceiling with corpus size. At 1k chunks we
/// want the library default (~320); at 1M chunks we want ~640 or more.
/// Logarithmic scaling based on chunk count:
///   ceiling = (log2(n_vectors) * 32).clamp(128, 4096)
/// 1k → 320, 13k → 447, 100k → 532, 1M → 640, 10M → 744
/// Then `(k * 2).clamp(min, max)` narrows the actual `itopk_size` at
/// query time. Overridable via `CQS_CAGRA_ITOPK_MAX`.
#[cfg(feature = "cuda-index")]
fn cagra_itopk_max_default(n_vectors: usize) -> usize {
    let log2 = (n_vectors.max(1) as f64).log2();
    let scaled = (log2 * 32.0) as usize;
    scaled.clamp(128, 4096)
}

/// Build-time CAGRA graph degrees, overridable via env.
/// `CQS_CAGRA_GRAPH_DEGREE` (default 64) is the output graph degree.
/// `CQS_CAGRA_INTERMEDIATE_GRAPH_DEGREE` (default 128) is the pruned-input
/// graph degree. Both map to the corresponding cuVS `IndexParams` setters.
/// Returns `IndexParams` with those setters applied (and traces the choice).
///
/// `metric` threads the cqs [`DistanceMetric`] into cuVS:
/// - `Cosine` keeps cuVS's default `L2Expanded`, which is rank-equivalent to
///   cosine on the unit-norm embeddings cqs produces (`d² = 2 − 2·cos`) and
///   is exactly what every pre-#1351 build used.
/// - `DotProduct` sets `InnerProduct`.
#[cfg(feature = "cuda-index")]
fn cagra_build_params(metric: DistanceMetric) -> Result<cuvs::cagra::IndexParams, CagraError> {
    // Use parse_env_usize_clamped so a literal "0" or empty string falls back
    // to the default. cuvs treats 0 as "library default" on some versions,
    // errors on others — silent-misconfig surface.
    let graph_degree =
        crate::limits::parse_env_usize_clamped("CQS_CAGRA_GRAPH_DEGREE", 64, 1, 4096);
    let intermediate_graph_degree =
        crate::limits::parse_env_usize_clamped("CQS_CAGRA_INTERMEDIATE_GRAPH_DEGREE", 128, 1, 4096);
    let params = cuvs::cagra::IndexParams::new()
        .map_err(|e| CagraError::Cuvs(e.to_string()))?
        .set_graph_degree(graph_degree)
        .set_intermediate_graph_degree(intermediate_graph_degree);
    match metric {
        // Default L2Expanded — deliberately untouched (see doc comment).
        DistanceMetric::Cosine => {}
        DistanceMetric::DotProduct => {
            // The cuvs 26.6 Rust wrapper exposes no `set_metric` builder, but
            // `IndexParams.0` is the pub raw `cuvsCagraIndexParams_t`.
            // SAFETY: identical access pattern to the wrapper's own setters
            // (e.g. `set_graph_degree`) — the pointer is valid for the
            // lifetime of `params` and we hold exclusive access here.
            unsafe {
                (*params.0).metric = cuvs::distance_type::DistanceType::InnerProduct;
            }
        }
    }
    tracing::info!(
        graph_degree,
        intermediate_graph_degree,
        metric = %metric,
        "CAGRA build params"
    );
    Ok(params)
}

/// CAGRA GPU index for vector search.
///
/// # Thread Safety
/// `resources` and `index` are protected by a single Mutex to ensure safe
/// concurrent access. CUDA contexts (managed by cuVS Resources) are not
/// inherently thread-safe, so we serialize all GPU operations.
#[cfg(feature = "cuda-index")]
pub struct CagraIndex {
    /// Embedding dimensionality (runtime, from model config)
    dim: usize,
    /// Distance metric the index was built with. Drives the
    /// distance → similarity conversion in `search_impl` and is stamped
    /// into the persisted sidecar.
    metric: DistanceMetric,
    /// cuVS resources + index, protected by Mutex (CUDA contexts require serialized access)
    gpu: Mutex<GpuState>,
    /// Mapping from internal index to chunk ID.
    ///
    /// `Box<str>` rather than `String` drops the 8-byte `cap` field per
    /// entry (~8 MB reduction at 1M chunks). Mirrors `HnswIndex::id_map`.
    id_map: Vec<Box<str>>,
    /// Set when a mutex poison is observed. The CUDA stream
    /// may be in an inconsistent posture after a mid-op panic
    /// (cudaMalloc'd buffer unfreed, stream corked, resources leaked),
    /// so subsequent searches against the same `GpuState` could
    /// double-free or CUDA-fault. `BatchContext::vector_index` checks
    /// `is_poisoned()` and forces a rebuild via `build_from_store_with_metric`
    /// rather than reusing the poisoned state.
    poisoned: AtomicBool,
}

#[cfg(feature = "cuda-index")]
struct GpuState {
    // Drop order is declaration order: `index` must drop before `resources`
    // because the cuVS Index holds handles into the Resources' CUDA context
    // and stream.
    index: cuvs::cagra::Index,
    resources: cuvs::Resources,
}

#[cfg(feature = "cuda-index")]
impl Drop for GpuState {
    fn drop(&mut self) {
        // Block until any pending CUDA work on this stream completes
        // before the Index + Resources fields drop. Without this, async
        // kernels launched by a prior search/deserialize can still be in
        // flight when cuvsResourcesDestroy fires, causing SIGSEGV on the
        // next test's cuvsResourcesCreate / kernel launch. Observed
        // deterministically when `test_save_load_round_trip` was followed
        // by `test_search_dimension_mismatch_query`.
        if let Err(e) = self.resources.sync_stream() {
            tracing::warn!(error = ?e, "cuvsStreamSync failed during GpuState drop");
        }
    }
}

// Debug impl needed because cuvs types don't implement Debug
#[cfg(feature = "cuda-index")]
impl std::fmt::Debug for CagraIndex {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("CagraIndex")
            .field("dim", &self.dim)
            .field("len", &self.id_map.len())
            .finish()
    }
}

#[cfg(feature = "cuda-index")]
impl CagraIndex {
    /// Check if GPU is available for CAGRA.
    ///
    /// Equivalent to `gpu_available_for(0, 0)`. Prefer
    /// [`Self::gpu_available_for`], which estimates the build memory budget
    /// and refuses to claim GPU availability when the corpus would OOM the
    /// device.
    pub fn gpu_available() -> bool {
        Self::gpu_available_for(0, 0)
    }

    /// GPU-availability + VRAM-budget check.
    ///
    /// `cuvs::Resources::new().is_ok()` only verifies that the CUDA driver
    /// loads; it doesn't guard against the *build* peak memory exceeding
    /// the device's free VRAM. On 8 GB GPUs an unguarded build OOMs during
    /// CAGRA construction with no graceful fallback to HNSW.
    ///
    /// Pass the actual `(n_vectors, dim)` of the corpus you're about to
    /// index. With `(0, 0)` this collapses to a plain driver-loadable check.
    pub fn gpu_available_for(n_vectors: usize, dim: usize) -> bool {
        if cuvs::Resources::new().is_err() {
            return false;
        }
        if n_vectors == 0 || dim == 0 {
            // Callers asking only "is the driver loadable?" — the caller has
            // no corpus shape to size.
            return true;
        }
        // Estimate build peak memory: dataset + graph + ~30% slack for cuVS
        // intermediate buffers. graph_degree default is 64 (matches
        // `cagra_build_params`).
        let dataset_bytes = (n_vectors as u64)
            .saturating_mul(dim as u64)
            .saturating_mul(4);
        let graph_bytes = (n_vectors as u64).saturating_mul(64).saturating_mul(4);
        let estimated = dataset_bytes
            .saturating_add(graph_bytes)
            .saturating_mul(130)
            / 100;
        // env override `CQS_CAGRA_MAX_GPU_BYTES` lets operators with
        // workloads they understand opt out of the conservative default
        // (2 GiB). Without `nvml-wrapper` we can't probe free VRAM here;
        // 2 GiB keeps RTX 4000 8 GB safe for most realistic corpora.
        let cap = std::env::var("CQS_CAGRA_MAX_GPU_BYTES")
            .ok()
            .and_then(|s| s.parse::<u64>().ok())
            .unwrap_or(2 * 1024 * 1024 * 1024);
        if estimated > cap.saturating_mul(80) / 100 {
            tracing::warn!(
                estimated_bytes = estimated,
                cap_bytes = cap,
                n_vectors,
                dim,
                "CAGRA: estimated build memory exceeds 80% of cap — falling back to HNSW. \
                 Set CQS_CAGRA_MAX_GPU_BYTES to override."
            );
            return false;
        }
        true
    }

    /// Build a CAGRA index from embeddings. The distance metric is resolved
    /// from `CQS_DISTANCE_METRIC` (default cosine); use
    /// [`Self::build_with_metric`] to pin it explicitly.
    ///
    /// Test-only: production builds go through `build_from_store_with_metric`,
    /// which streams embeddings from SQLite. This in-memory constructor exists
    /// to seed fixtures from explicit `(id, embedding)` pairs.
    #[cfg(test)]
    pub fn build(embeddings: Vec<(String, Embedding)>, dim: usize) -> Result<Self, CagraError> {
        let metric = DistanceMetric::resolve().map_err(CagraError::Build)?;
        Self::build_with_metric(embeddings, dim, metric)
    }

    /// [`Self::build`] with an explicit [`DistanceMetric`].
    ///
    /// Test-only: see [`Self::build`].
    #[cfg(test)]
    pub fn build_with_metric(
        embeddings: Vec<(String, Embedding)>,
        dim: usize,
        metric: DistanceMetric,
    ) -> Result<Self, CagraError> {
        let _span = tracing::debug_span!("cagra_build", metric = %metric).entered();
        let (id_map, flat_data, n_vectors) = crate::hnsw::prepare_index_data(embeddings, dim)
            .map_err(|e| CagraError::Build(e.to_string()))?;

        tracing::info!(n_vectors, metric = %metric, "Building CAGRA index");

        let resources = cuvs::Resources::new().map_err(|e| CagraError::Cuvs(e.to_string()))?;

        let dataset = Array2::from_shape_vec((n_vectors, dim), flat_data)
            .map_err(|e| CagraError::Cuvs(format!("Failed to create array: {}", e)))?;

        let build_params = cagra_build_params(metric)?;

        let index = cuvs::cagra::Index::build(&resources, &build_params, &dataset)
            .map_err(|e| CagraError::Cuvs(e.to_string()))?;

        tracing::info!("CAGRA index built successfully");

        Ok(Self {
            dim,
            metric,
            gpu: Mutex::new(GpuState { resources, index }),
            id_map,
            poisoned: AtomicBool::new(false),
        })
    }

    /// The distance metric this index was built (or loaded) with.
    pub fn metric(&self) -> DistanceMetric {
        self.metric
    }

    /// Number of vectors in the index
    pub fn len(&self) -> usize {
        self.id_map.len()
    }

    /// Checks whether this collection contains any elements.
    pub fn is_empty(&self) -> bool {
        self.id_map.is_empty()
    }

    /// Search for nearest neighbors
    pub fn search(&self, query: &Embedding, k: usize) -> Vec<IndexResult> {
        let _span = tracing::debug_span!("cagra_search", k).entered();
        if self.id_map.is_empty() || k == 0 {
            return Vec::new();
        }

        if query.len() != self.dim {
            tracing::warn!(
                expected_dim = self.dim,
                actual_dim = query.len(),
                "Query dimension mismatch"
            );
            return Vec::new();
        }

        // Reject non-finite query vectors before they reach the cuVS kernel.
        // The HNSW path guards this up front (a NaN distance trips an
        // `anndists` assert); CAGRA accepted them silently — a backend
        // asymmetry. Mirror the HNSW empty-result contract: a malformed
        // query (encoder bug, query-cache corruption, bit rot) returns no
        // results rather than dispatching a kernel on poisoned input.
        if !query_is_finite(query) {
            tracing::warn!(
                "CAGRA query embedding contains non-finite values (NaN/Inf), \
                 returning empty results"
            );
            return Vec::new();
        }

        let gpu = self.gpu.lock().unwrap_or_else(|poisoned| {
            // A prior panic left the CUDA stream in an unknown state. Flag
            // the index so the caller forces a rebuild on the next
            // `vector_index()` access; return an empty result here rather
            // than run a new kernel against possibly-corrupted resources.
            self.poisoned.store(true, Ordering::Release);
            tracing::warn!(
                "CAGRA GPU mutex poisoned — results from this call are discarded \
                 and the index will be rebuilt on the next vector_index() access"
            );
            poisoned.into_inner()
        });

        if self.poisoned.load(Ordering::Acquire) {
            // Don't dispatch new kernels; the caller should have already
            // rebuilt us. This is a safety net for racing clients.
            return Vec::new();
        }

        self.search_impl(&gpu, query, k, None)
    }

    /// Core search implementation shared by filtered and unfiltered paths.
    fn search_impl(
        &self,
        gpu: &GpuState,
        query: &Embedding,
        k: usize,
        bitset_device: Option<&cuvs::ManagedTensor>,
    ) -> Vec<IndexResult> {
        let itopk_min = std::env::var("CQS_CAGRA_ITOPK_MIN")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(128);
        let itopk_max = std::env::var("CQS_CAGRA_ITOPK_MAX")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or_else(|| cagra_itopk_max_default(self.len()));
        // cuVS CAGRA hard-requires `itopk_size >= k`. A plain
        // `(k * 2).clamp(min, max)` could clamp `itopk_size` *below* `k`
        // when `k > itopk_max` (e.g. `cqs search --limit 500` on a small
        // corpus); CAGRA then errors out and the result reaches the caller
        // as an empty Vec — a silent zero-result. Force `itopk_size >= k`
        // and refuse the search if the cap can't honour it. `saturating_mul`
        // mirrors the HNSW search path — pathological `k` saturates rather
        // than panics/wraps.
        // Defense-in-depth: dispatch sites read `VectorIndex::max_k()`
        // (returns `Some(itopk_max)` for this backend) and trim `k` so
        // the call below succeeds. This branch only fires when a caller
        // bypasses the dispatch helper — log loudly and bail.
        let itopk_size = k.saturating_mul(2).clamp(itopk_min, itopk_max).max(k);
        if itopk_size > itopk_max {
            tracing::warn!(
                k,
                itopk_max,
                n_vectors = self.len(),
                "CAGRA: k exceeds itopk_max — caller bypassed `cap_k_to_backend`"
            );
            return Vec::new();
        }
        tracing::debug!(
            itopk_size,
            itopk_min,
            itopk_max,
            k,
            n_vectors = self.len(),
            "CAGRA itopk resolved"
        );

        let search_params = match cuvs::cagra::SearchParams::new() {
            Ok(params) => params.set_itopk_size(itopk_size),
            Err(e) => {
                tracing::error!(error = %e, "Failed to create search params");
                return Vec::new();
            }
        };

        let query_host = match Array2::from_shape_vec((1, self.dim), query.as_slice().to_vec()) {
            Ok(arr) => arr,
            Err(e) => {
                tracing::error!(expected_dim = self.dim, error = %e, "Invalid query shape");
                return Vec::new();
            }
        };

        // IMPORTANT: host arrays must outlive device tensors — ManagedTensor::to_device()
        // copies data to GPU but the DLTensor shape pointer still references the host
        // ndarray's internal shape storage.
        let mut neighbors_host: Array2<u32> = Array2::zeros((1, k));
        // Sentinel-init: cuVS does not write slots beyond the number of
        // real neighbours it found, so we seed every slot with
        // `INVALID_DISTANCE` and filter against it after copy-back. See
        // the `INVALID_DISTANCE` doc for the cuVS API audit that motivates
        // the sentinel approach.
        let mut distances_host: Array2<f32> = Array2::from_elem((1, k), INVALID_DISTANCE);

        let query_device = match cuvs::ManagedTensor::from(&query_host).to_device(&gpu.resources) {
            Ok(t) => t,
            Err(e) => {
                tracing::error!(error = %e, "Failed to copy query to device");
                return Vec::new();
            }
        };

        let neighbors_device =
            match cuvs::ManagedTensor::from(&neighbors_host).to_device(&gpu.resources) {
                Ok(t) => t,
                Err(e) => {
                    tracing::error!(error = %e, "Failed to allocate neighbors on device");
                    return Vec::new();
                }
            };

        let distances_device =
            match cuvs::ManagedTensor::from(&distances_host).to_device(&gpu.resources) {
                Ok(t) => t,
                Err(e) => {
                    tracing::error!(error = %e, "Failed to allocate distances on device");
                    return Vec::new();
                }
            };

        // Perform search — non-consuming.
        let result = if let Some(bitset) = bitset_device {
            gpu.index.search_with_filter(
                &gpu.resources,
                &search_params,
                &query_device,
                &neighbors_device,
                &distances_device,
                bitset,
            )
        } else {
            gpu.index.search(
                &gpu.resources,
                &search_params,
                &query_device,
                &neighbors_device,
                &distances_device,
            )
        };

        if let Err(e) = result {
            tracing::error!(error = %e, "CAGRA search failed");
            return Vec::new();
        }

        // Copy results back to host
        if let Err(e) = neighbors_device.to_host(&gpu.resources, &mut neighbors_host) {
            tracing::error!(error = %e, "Failed to copy neighbors from device");
            return Vec::new();
        }
        if let Err(e) = distances_device.to_host(&gpu.resources, &mut distances_host) {
            tracing::error!(error = %e, "Failed to copy distances from device");
            return Vec::new();
        }

        // Convert results: distance → similarity, per build metric.
        // - Cosine: the index is built with cuVS's default L2Expanded
        //   (squared L2). For unit-norm vectors d = 2 - 2*cos_sim, so
        //   cos_sim = 1 - d/2.
        // - DotProduct: cuVS InnerProduct returns the RAW inner product in
        //   the distances buffer, best-first (descending) — verified
        //   empirically on cuVS 26.06 by `test_dot_product_build_and_search`
        //   (self-match slot holds ≈ +1.0 on unit-norm vectors). The
        //   similarity is the value itself.
        let mut results = Vec::with_capacity(k);
        let neighbor_row = neighbors_host.row(0);
        let distance_row = distances_host.row(0);

        for i in 0..k {
            let idx = neighbor_row[i] as usize;
            let dist = distance_row[i];
            // Sentinel check: a slot still holding
            // `INVALID_DISTANCE` means cuVS did not overwrite it, so the
            // paired `neighbor_row[i]` is garbage and must be dropped.
            // `!is_finite()` is a superset of `dist == INVALID_DISTANCE`
            // (since `INVALID_DISTANCE == +∞`) and also catches any NaN
            // cuVS might emit on exotic inputs — real squared-L2
            // distances are always finite and non-negative.
            if !dist.is_finite() {
                continue;
            }
            if idx < self.id_map.len() {
                let score = match self.metric {
                    // Clamp the f32 overshoot: a near-self match where the
                    // unit-norm self-dot rounds just above 1.0 makes cos = 1 - d/2
                    // exceed 1.0; cap it so a downstream acos/sqrt-of-(1-score)
                    // consumer can't NaN.
                    DistanceMetric::Cosine => (1.0 - dist / 2.0).min(1.0),
                    DistanceMetric::DotProduct => dist,
                };
                results.push(IndexResult {
                    id: self.id_map[idx].to_string(),
                    score,
                });
            }
        }

        results
    }
}

#[cfg(feature = "cuda-index")]
impl VectorIndex for CagraIndex {
    fn search(&self, query: &Embedding, k: usize) -> Vec<IndexResult> {
        CagraIndex::search(self, query, k)
    }

    fn len(&self) -> usize {
        CagraIndex::len(self)
    }

    fn is_empty(&self) -> bool {
        CagraIndex::is_empty(self)
    }

    fn name(&self) -> &'static str {
        "CAGRA"
    }

    /// Expose the poison flag so `BatchContext::vector_index` can force a
    /// full rebuild instead of reusing a possibly-corrupt CUDA context
    /// after a prior panic.
    fn is_poisoned(&self) -> bool {
        self.poisoned.load(Ordering::Acquire)
    }

    fn dim(&self) -> usize {
        self.dim
    }

    /// CAGRA enforces `itopk_size >= k` and `itopk_size <= itopk_max`.
    /// `cagra_itopk_max_default(n_vectors)` produces 441 at 14k chunks,
    /// 532 at 100k, 640 at 1M. A request for `k > itopk_max` would cause
    /// `search_impl` to return an empty Vec — see the safety net at the
    /// top of `search_impl`. Dispatch sites read this cap and trim `k`
    /// before calling, so the cap there is defense-in-depth and this
    /// trait method is the primary contract.
    ///
    /// Honors `CQS_CAGRA_ITOPK_MAX` (the same env override `search_impl`
    /// uses) so a single knob controls both the actual cap and the
    /// reported cap; otherwise dispatch could trim to the default value
    /// while the search-time read of the env raised the real ceiling and
    /// vice versa.
    fn max_k(&self) -> Option<usize> {
        let cap = std::env::var("CQS_CAGRA_ITOPK_MAX")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or_else(|| cagra_itopk_max_default(self.len()));
        Some(cap)
    }

    /// GPU-native filtered search: builds a bitset from the predicate and
    /// passes it to CAGRA for traversal-time filtering. No over-fetching needed.
    fn search_with_filter(
        &self,
        query: &Embedding,
        k: usize,
        filter: &dyn Fn(&str) -> bool,
    ) -> Vec<IndexResult> {
        let _span = tracing::debug_span!("cagra_search_filtered", k).entered();
        if self.id_map.is_empty() || k == 0 {
            return Vec::new();
        }

        if query.len() != self.dim {
            tracing::warn!(
                expected_dim = self.dim,
                actual_dim = query.len(),
                "Query dimension mismatch"
            );
            return Vec::new();
        }

        // Build bitset on host: evaluate predicate for each vector
        let n = self.id_map.len();
        let n_words = n.div_ceil(32);
        let mut bitset = vec![0u32; n_words];
        let mut included = 0usize;
        for (i, id) in self.id_map.iter().enumerate() {
            if filter(id) {
                bitset[i / 32] |= 1u32 << (i % 32);
                included += 1;
            }
        }

        // If everything passes the filter, use unfiltered search (faster)
        if included == n {
            return CagraIndex::search(self, query, k);
        }

        // If nothing passes, no results
        if included == 0 {
            return Vec::new();
        }

        // Cap effective `k` at the bitset's `included` count. Asking
        // CAGRA for more slots than feasible silently under-fills (or, when
        // `k > itopk_max`, errors and returns empty). Both modes hide a
        // "candidate pool was small" answer behind the same empty Vec a
        // genuine "no matches" would produce. Trim explicitly so the caller
        // sees an honest, smaller-than-requested result.
        let effective_k = k.min(included);
        if effective_k < k {
            tracing::debug!(
                requested = k,
                effective = effective_k,
                included,
                "CAGRA filtered search: capping k at included to avoid under-fill"
            );
        }

        tracing::debug!(
            total = n,
            included,
            excluded = n - included,
            "CAGRA bitset filter"
        );

        let gpu = self.gpu.lock().unwrap_or_else(|poisoned| {
            // Same recovery path as `search()`. CUDA state may be corrupt;
            // flag for rebuild and drop the work rather than kernel-launch
            // against a bad stream.
            self.poisoned.store(true, Ordering::Release);
            tracing::warn!(
                "CAGRA GPU mutex poisoned (filtered path) — results discarded \
                 and index will be rebuilt on next vector_index()"
            );
            poisoned.into_inner()
        });

        if self.poisoned.load(Ordering::Acquire) {
            return Vec::new();
        }

        // Upload bitset to device
        let bitset_host = ndarray_015::Array1::from_vec(bitset);
        let bitset_device = match cuvs::ManagedTensor::from(&bitset_host).to_device(&gpu.resources)
        {
            Ok(t) => t,
            Err(e) => {
                tracing::error!(error = %e, "Failed to upload bitset to device");
                return Vec::new();
            }
        };

        self.search_impl(&gpu, query, effective_k, Some(&bitset_device))
    }
}

// SAFETY: CagraIndex is thread-safe because:
// - `gpu` (resources + index) is protected by Mutex (CUDA contexts require serialized access)
// - `id_map` is immutable after construction
#[cfg(feature = "cuda-index")]
unsafe impl Send for CagraIndex {}
#[cfg(feature = "cuda-index")]
unsafe impl Sync for CagraIndex {}

#[cfg(feature = "cuda-index")]
impl CagraIndex {
    /// Build CAGRA index from all embeddings in a Store.
    /// CAGRA (cuVS) requires all data upfront for GPU index building, so we
    /// can't stream incrementally like HNSW. We stream from SQLite to avoid
    /// double-buffering in memory.
    /// Notes are excluded — they use brute-force search from SQLite.
    ///
    /// `CagraBackend::try_open` uses this to keep a rebuilt CAGRA index
    /// consistent with the slot's stored HNSW metric instead of silently
    /// re-resolving to the default.
    pub fn build_from_store_with_metric<Mode>(
        store: &crate::Store<Mode>,
        dim: usize,
        metric: DistanceMetric,
    ) -> Result<Self, CagraError> {
        let _span = tracing::debug_span!("cagra_build_from_store", metric = %metric).entered();
        let chunk_count = store
            .chunk_count()
            .map_err(|e| CagraError::Cuvs(format!("Failed to count chunks: {}", e)))?
            as usize;

        if chunk_count == 0 {
            return Err(CagraError::Cuvs("No embeddings in store".into()));
        }

        tracing::info!(chunk_count, "Building CAGRA index from chunk embeddings");

        // Guard against OOM: estimate CPU memory needed for flat data + id map
        let max_bytes = cagra_max_bytes();
        let estimated_bytes = chunk_count.saturating_mul(dim).saturating_mul(4);
        if estimated_bytes > max_bytes {
            return Err(CagraError::Cuvs(format!(
                "Dataset too large for GPU indexing: {}MB estimated (limit {}MB)",
                estimated_bytes / (1024 * 1024),
                max_bytes / (1024 * 1024)
            )));
        }

        // Sanity-bound chunk_count before with_capacity in case a corrupt
        // store reports usize::MAX (matching SPLADE's defensive pattern in
        // splade/index.rs). 1<<28 = ~268M chunks, well above any realistic
        // corpus.
        const MAX_CHUNKS_SANITY: usize = 1 << 28;
        if chunk_count > MAX_CHUNKS_SANITY {
            return Err(CagraError::Io(format!(
                "Refusing to allocate id_map for chunk_count={} > {}",
                chunk_count, MAX_CHUNKS_SANITY
            )));
        }
        let mut id_map = Vec::with_capacity(chunk_count);
        let mut flat_data = Vec::with_capacity(chunk_count.saturating_mul(dim));

        // Streaming batch size is env-overridable so higher-dim models can
        // shrink the per-batch heap footprint without a recompile. At
        // dim=1024 the default is 40 MB / batch (10_000 × 1024 × 4 bytes);
        // at dim=4096, 160 MB / batch.
        let batch_size = cagra_stream_batch_size(dim);
        let mut loaded_chunks = 0usize;
        for batch_result in store.embedding_batches(batch_size) {
            let batch = batch_result
                .map_err(|e| CagraError::Cuvs(format!("Failed to fetch batch: {}", e)))?;

            let batch_len = batch.len();
            for (chunk_id, embedding) in batch {
                if embedding.len() != dim {
                    return Err(CagraError::DimensionMismatch {
                        expected: dim,
                        actual: embedding.len(),
                    });
                }
                id_map.push(chunk_id);
                flat_data.extend(embedding.into_inner());
            }

            loaded_chunks += batch_len;
            let progress_pct = (loaded_chunks * 100)
                .checked_div(chunk_count)
                .unwrap_or(100);
            tracing::info!(
                "CAGRA loading progress: {} / {} chunks ({}%)",
                loaded_chunks,
                chunk_count,
                progress_pct
            );
        }

        Self::build_from_flat(id_map, flat_data, dim, metric)
    }

    /// Build CAGRA index from pre-collected flat data (also used by tests)
    pub(crate) fn build_from_flat(
        id_map: Vec<String>,
        flat_data: Vec<f32>,
        dim: usize,
        metric: DistanceMetric,
    ) -> Result<Self, CagraError> {
        let n_vectors = id_map.len();
        if n_vectors == 0 {
            return Err(CagraError::Cuvs("Cannot build empty index".into()));
        }

        tracing::info!(n_vectors, metric = %metric, "Building CAGRA index");

        let resources = cuvs::Resources::new().map_err(|e| CagraError::Cuvs(e.to_string()))?;

        let dataset = Array2::from_shape_vec((n_vectors, dim), flat_data)
            .map_err(|e| CagraError::Cuvs(format!("Failed to create array: {}", e)))?;

        let build_params = cagra_build_params(metric)?;

        let index = cuvs::cagra::Index::build(&resources, &build_params, &dataset)
            .map_err(|e| CagraError::Cuvs(e.to_string()))?;

        tracing::info!("CAGRA index built successfully");

        // Convert Vec<String> → Vec<Box<str>> at the CagraIndex
        // construction boundary. `String::into_boxed_str()` is zero-copy
        // (shrinks the existing heap allocation, drops the `cap` field),
        // saving ~8 bytes per entry of resident memory.
        let id_map_boxed: Vec<Box<str>> = id_map.into_iter().map(String::into_boxed_str).collect();

        Ok(Self {
            dim,
            metric,
            gpu: Mutex::new(GpuState { resources, index }),
            id_map: id_map_boxed,
            poisoned: AtomicBool::new(false),
        })
    }
}

/// JSON sidecar next to the `.cagra` binary blob.
///
/// Carries everything we need to validate the blob before handing it to
/// `cuvsCagraDeserialize` and to reject stale persisted data after a
/// reindex.
///
/// The JSON format is versioned via `magic` + `version` so a future binary
/// that can't parse an older sidecar just falls through to rebuilding.
#[cfg(feature = "cuda-index")]
#[derive(serde::Serialize, serde::Deserialize, Debug)]
struct CagraMeta {
    /// Format magic. See [`CAGRA_META_MAGIC`].
    magic: String,
    /// Sidecar schema version. See [`CAGRA_META_VERSION`].
    version: u32,
    /// Embedding dimensionality, captured at save. Must match the store on load.
    dim: usize,
    /// Number of vectors in the persisted index. Must match the store on load.
    chunk_count: usize,
    /// `Store::splade_generation()` at save time. Bumped by the delete trigger.
    /// Coarse staleness check; a mismatch is not fatal because CAGRA builds
    /// survive deletion-free INSERTs, but we log it as an informational warn.
    splade_generation: u64,
    /// Chunk IDs in the same order as the persisted CAGRA internal indices.
    /// Rebuilding this is trivially cheap (serde_json), and cuVS gives us
    /// nothing to translate internal ids → chunk ids on its own.
    id_map: Vec<String>,
    /// Blake3 checksum over the `.cagra` binary blob as a hex string.
    blake3: String,
    /// [`DistanceMetric::as_str`] value the index was built with.
    /// Legacy sidecars (pre-#1351) omit the field and default to cosine —
    /// which is what every legacy build used.
    #[serde(default = "default_metric_string")]
    metric: String,
}

/// serde default for [`CagraMeta::metric`]: legacy sidecars are cosine.
#[cfg(feature = "cuda-index")]
fn default_metric_string() -> String {
    DistanceMetric::Cosine.as_str().to_string()
}

/// Whether CAGRA persistence is enabled (via `CQS_CAGRA_PERSIST`).
///
/// Defaults to `true`. Setting `CQS_CAGRA_PERSIST=0` disables both the save
/// path (build_from_store_with_metric won't write the file) and the load path
/// (build_vector_index_with_config won't even check for persisted indices).
///
/// Cached in a `OnceLock` so we parse the env var exactly once per process.
#[cfg(feature = "cuda-index")]
pub fn cagra_persist_enabled() -> bool {
    static ENABLED: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
    *ENABLED.get_or_init(|| match std::env::var("CQS_CAGRA_PERSIST").as_deref() {
        Ok("0") | Ok("false") | Ok("no") => {
            tracing::info!("CQS_CAGRA_PERSIST=0 — CAGRA persistence disabled");
            false
        }
        _ => true,
    })
}

#[cfg(feature = "cuda-index")]
impl CagraIndex {
    /// Persist the index to disk with a zero `splade_generation` stamp.
    ///
    /// Test-only convenience wrapper over [`save_inner`](Self::save_inner);
    /// see [`save_with_store`](Self::save_with_store) for the production path.
    #[cfg(test)]
    pub fn save(&self, path: &Path) -> Result<(), CagraError> {
        // Test-only convenience: stamps `splade_generation = 0`. Production
        // callers must use `save_with_store` so the deletion counter is
        // recorded for the coarse staleness check on load.
        self.save_inner(path, 0)
    }

    /// Persist with an explicit `splade_generation` stamp from the caller's
    /// `Store`. Preferred over the test-only `save` because it records the
    /// deletion counter for coarse staleness checks on load.
    pub fn save_with_store<Mode>(
        &self,
        path: &Path,
        store: &crate::Store<Mode>,
    ) -> Result<(), CagraError> {
        let _span = tracing::info_span!("cagra_save_with_store", path = %path.display()).entered();
        // splade_generation is not a perfect staleness signal (deletes only,
        // not inserts) but it still catches the common reindex-with-GC case.
        // Read before delegating so a store error surfaces in this span.
        let generation = if cagra_persist_enabled() {
            match store.splade_generation() {
                Ok(g) => g,
                Err(e) => {
                    tracing::warn!(
                        error = %e,
                        "Failed to read splade_generation for CAGRA meta; defaulting to 0"
                    );
                    0
                }
            }
        } else {
            0
        };
        self.save_inner(path, generation)
    }

    /// Shared persistence body for [`save`](Self::save) and
    /// [`save_with_store`](Self::save_with_store).
    ///
    /// Writes two files:
    /// - `{path}` — the cuVS binary blob (via `cuvsCagraSerialize`)
    /// - `{path}.meta` — a JSON sidecar with magic/version/dim/chunk_count/
    ///   id_map/blake3/splade_generation so `load()` can validate before
    ///   handing the blob back to cuVS.
    ///
    /// The cuVS blob is staged with `.bak` rollback and checksummed, then the
    /// sidecar is written atomically (write-temp → rename) with the given
    /// `splade_generation` stamp. If the sidecar write fails the partial
    /// `.cagra` file is removed so we don't leave an orphan that would later
    /// fail metadata validation. Mirrors `src/hnsw/persist.rs` and the SPLADE
    /// pattern.
    ///
    /// Persistence is a best-effort optimisation: callers should warn-log
    /// failures and continue rather than propagate errors. The caller will
    /// rebuild on next startup.
    fn save_inner(&self, path: &Path, generation: u64) -> Result<(), CagraError> {
        let _span = tracing::info_span!("cagra_save_inner", path = %path.display()).entered();
        if !cagra_persist_enabled() {
            // Per-call warn — a silent skip would surface only as a missing
            // blob at next load.
            tracing::warn!(path = %path.display(), "CAGRA save skipped — CQS_CAGRA_PERSIST=0");
            return Err(CagraError::Io(
                "CAGRA persistence disabled via CQS_CAGRA_PERSIST=0".to_string(),
            ));
        }

        let gpu = self.gpu.lock().map_err(|_| {
            // The mutex was poisoned by a prior panic in the search path.
            // Rather than try to serialize a potentially corrupt CUDA
            // context, refuse and let the caller rebuild.
            self.poisoned.store(true, Ordering::Release);
            CagraError::Io("CAGRA mutex poisoned, refusing to save".to_string())
        })?;

        if self.poisoned.load(Ordering::Acquire) {
            return Err(CagraError::Io(
                "CAGRA index is poisoned, refusing to save".to_string(),
            ));
        }

        // Ensure the parent exists so cuVS can open the file for writing.
        if let Some(parent) = path.parent() {
            if let Err(e) = std::fs::create_dir_all(parent) {
                return Err(CagraError::Io(format!(
                    "Failed to create parent dir {}: {}",
                    parent.display(),
                    e
                )));
            }
        }

        // Atomic save with `.bak` rollback: stage to a tmp path,
        // atomic_replace onto the live name, restore from `.bak` on failure
        // so a mid-save failure can't destroy the previous good blob.
        let meta_path = meta_path_for(path);
        let blob_hash = save_blob_atomic_with_rollback(self, &gpu, path)?;

        let meta = CagraMeta {
            magic: CAGRA_META_MAGIC.to_string(),
            version: CAGRA_META_VERSION,
            dim: self.dim,
            chunk_count: self.id_map.len(),
            splade_generation: generation,
            // Convert Vec<Box<str>> → Vec<String> at the sidecar boundary
            // (the on-disk schema is Vec<String> for `serde_json` decode).
            id_map: self.id_map.iter().map(|s| s.to_string()).collect(),
            blake3: blob_hash,
            metric: self.metric.as_str().to_string(),
        };

        // Meta-write failure after the blob is in place leaves a blob
        // without a sidecar — load() rejects → next caller rebuilds. Both
        // sides are safe; clean up so we don't leave the pair half-formed.
        if let Err(e) = write_meta_atomic(&meta_path, &meta) {
            let _ = std::fs::remove_file(path);
            let _ = std::fs::remove_file(&meta_path);
            return Err(e);
        }

        tracing::info!(
            path = %path.display(),
            n_vectors = self.id_map.len(),
            splade_generation = generation,
            "CAGRA index persisted"
        );
        Ok(())
    }

    /// Load a previously-saved index from disk.
    ///
    /// Verifies the sidecar magic/version, confirms `dim` / `chunk_count`
    /// match the caller's expectation (passed as `expected_dim` /
    /// `expected_chunks`), verifies the blake3 checksum over the `.cagra`
    /// blob, then hands the blob to `cuvsCagraDeserialize`.
    ///
    /// Any validation failure returns `Err(CagraError::Stale { .. })` or
    /// `Err(CagraError::ChecksumMismatch { .. })`. The caller should warn-log
    /// and rebuild from the store.
    ///
    /// The `expected_chunks` check is what prevents us from handing cuVS a
    /// blob whose id_map no longer matches the live store: an incremental
    /// reindex that added or removed chunks will change the count even if
    /// `splade_generation` didn't bump.
    pub fn load(
        path: &Path,
        expected_dim: usize,
        expected_chunks: usize,
    ) -> Result<Self, CagraError> {
        let _span = tracing::info_span!("cagra_load", path = %path.display()).entered();
        if !cagra_persist_enabled() {
            tracing::warn!(path = %path.display(), "CAGRA load skipped — CQS_CAGRA_PERSIST=0");
            return Err(CagraError::Io(
                "CAGRA persistence disabled via CQS_CAGRA_PERSIST=0".to_string(),
            ));
        }

        // Sweep orphan tmp files from interrupted saves before loading. A hard
        // kill during the cuVS serialize step leaves a `.<blob>.<hex>.tmp` (or
        // sidecar `.<blob>.meta.<hex>.tmp`) behind; the random suffix means no
        // later save ever overwrites it, so they accumulate. Both `.<blob>.`-
        // prefixed `.tmp` names are caught here. Mirrors the HNSW load sweep
        // (hnsw/persist.rs) and SPLADE loader sweep (splade/index.rs).
        cleanup_orphan_temp_files(path);

        if !path.exists() {
            return Err(CagraError::Io(format!(
                "CAGRA blob not found at {}",
                path.display()
            )));
        }

        let meta_path = meta_path_for(path);
        if !meta_path.exists() {
            return Err(CagraError::BadMeta(format!(
                "CAGRA sidecar missing at {}",
                meta_path.display()
            )));
        }

        // Bounded read of the sidecar so a corrupt or hostile file can't
        // OOM us. 128MB is generous even for multi-million-vector id_maps.
        const MAX_META_SIZE: u64 = 128 * 1024 * 1024;
        let meta_size = std::fs::metadata(&meta_path)
            .map_err(|e| {
                CagraError::Io(format!(
                    "Failed to stat CAGRA sidecar {}: {}",
                    meta_path.display(),
                    e
                ))
            })?
            .len();
        if meta_size > MAX_META_SIZE {
            return Err(CagraError::BadMeta(format!(
                "CAGRA sidecar {} is {}MB, exceeds {}MB limit",
                meta_path.display(),
                meta_size / (1024 * 1024),
                MAX_META_SIZE / (1024 * 1024)
            )));
        }

        let meta_file = std::fs::File::open(&meta_path).map_err(|e| {
            CagraError::Io(format!(
                "Failed to open CAGRA sidecar {}: {}",
                meta_path.display(),
                e
            ))
        })?;
        let meta: CagraMeta =
            serde_json::from_reader(std::io::BufReader::new(meta_file)).map_err(|e| {
                CagraError::BadMeta(format!(
                    "Failed to parse CAGRA sidecar {}: {}",
                    meta_path.display(),
                    e
                ))
            })?;

        if meta.magic != CAGRA_META_MAGIC {
            return Err(CagraError::BadMeta(format!(
                "Unexpected magic {:?} (want {:?})",
                meta.magic, CAGRA_META_MAGIC
            )));
        }
        if meta.version != CAGRA_META_VERSION {
            return Err(CagraError::Stale {
                reason: format!(
                    "sidecar version {} != current {}",
                    meta.version, CAGRA_META_VERSION
                ),
            });
        }
        if meta.dim != expected_dim {
            return Err(CagraError::Stale {
                reason: format!("dim {} != expected {}", meta.dim, expected_dim),
            });
        }
        if meta.chunk_count != expected_chunks {
            return Err(CagraError::Stale {
                reason: format!(
                    "chunk_count {} != expected {} (reindex occurred)",
                    meta.chunk_count, expected_chunks
                ),
            });
        }
        if meta.id_map.len() != meta.chunk_count {
            return Err(CagraError::BadMeta(format!(
                "sidecar id_map has {} entries but claims {} chunks",
                meta.id_map.len(),
                meta.chunk_count
            )));
        }

        // Distance metric: the stored value wins; an explicitly set,
        // conflicting CQS_DISTANCE_METRIC marks the blob stale so the caller
        // rebuilds with the requested metric (CAGRA is a derived index — the
        // rebuild path is the remedy, unlike HNSW where the same conflict is
        // a hard typed error).
        let stored_metric: DistanceMetric = meta
            .metric
            .parse()
            .map_err(|e| CagraError::BadMeta(format!("sidecar has invalid metric: {e}")))?;
        if let Some(requested) = DistanceMetric::from_env().map_err(CagraError::Build)? {
            if requested != stored_metric {
                return Err(CagraError::Stale {
                    reason: format!("distance metric '{stored_metric}' != requested '{requested}'"),
                });
            }
        }

        // Verify the `.cagra` blob matches the hash in the sidecar before
        // handing it to cuVS — cheap insurance against silent disk rot
        // and against someone (or us) overwriting the blob without
        // updating the sidecar.
        let actual_hash = blake3_of_path(path)?;
        if actual_hash != meta.blake3 {
            return Err(CagraError::ChecksumMismatch(path.display().to_string()));
        }

        let path_str = path.to_str().ok_or_else(|| {
            CagraError::Io(format!(
                "CAGRA load path is not valid UTF-8: {}",
                path.display()
            ))
        })?;

        let resources = cuvs::Resources::new().map_err(|e| CagraError::Cuvs(e.to_string()))?;
        let index = cuvs::cagra::Index::deserialize(&resources, path_str)
            .map_err(|e| CagraError::Cuvs(format!("cuvsCagraDeserialize failed: {}", e)))?;

        tracing::info!(
            n_vectors = meta.chunk_count,
            dim = meta.dim,
            splade_generation = meta.splade_generation,
            metric = %stored_metric,
            "CAGRA index loaded from disk"
        );

        Ok(Self {
            dim: meta.dim,
            metric: stored_metric,
            gpu: Mutex::new(GpuState { resources, index }),
            // Convert Vec<String> → Vec<Box<str>> at the load boundary.
            // Zero-copy via `String::into_boxed_str()`.
            id_map: meta
                .id_map
                .into_iter()
                .map(String::into_boxed_str)
                .collect(),
            poisoned: AtomicBool::new(false),
        })
    }

    /// Delete a persisted CAGRA index and its sidecar. Best-effort — missing
    /// files are treated as success so the caller can use this as a cleanup
    /// before a rebuild without first checking existence.
    pub fn delete_persisted(path: &Path) {
        let _span =
            tracing::debug_span!("cagra_delete_persisted", path = %path.display()).entered();
        for p in [path.to_path_buf(), meta_path_for(path)] {
            match std::fs::remove_file(&p) {
                Ok(()) => {}
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
                Err(e) => tracing::warn!(
                    path = %p.display(),
                    error = %e,
                    "CAGRA cleanup failed — next rebuild may re-hit the same corrupt blob"
                ),
            }
        }
    }
}

/// Sidecar path for a given CAGRA blob path.
#[cfg(feature = "cuda-index")]
fn meta_path_for(path: &Path) -> std::path::PathBuf {
    let mut s = path.as_os_str().to_os_string();
    s.push(".meta");
    std::path::PathBuf::from(s)
}

/// Remove orphan tmp files left by interrupted CAGRA saves.
///
/// Both the blob (`save_blob_atomic_with_rollback`) and the meta sidecar
/// (`write_meta_atomic`) stage to `.<blob_file_name>.<hex16>.tmp` (the sidecar
/// adds a `.meta` segment, so it shares the `.<blob_file_name>.` prefix) before
/// atomic-replacing the live path. A hard kill (SIGKILL on daemon-stop timeout,
/// OOM-kill, power loss) during the cuVS serialize — the longest save window —
/// leaves the tmp behind, and the random suffix means no later save overwrites
/// it. Sweeping on load matches the HNSW (`hnsw/persist.rs`) and SPLADE
/// (`splade/index.rs`) loaders. Best-effort: a leftover tmp is wasteful, not
/// fatal, so errors are logged and not propagated.
///
/// GPU-free (pure filesystem) so the sweep test runs without touching cuVS,
/// though the whole `cagra` module is `cuda-index`-gated.
#[cfg(feature = "cuda-index")]
fn cleanup_orphan_temp_files(path: &std::path::Path) {
    let parent = match path.parent() {
        Some(p) => p,
        None => return,
    };
    let blob_name = match path.file_name().and_then(|s| s.to_str()) {
        Some(n) => n,
        None => return,
    };
    // Live blob is `<blob_name>` and live meta is `<blob_name>.meta`; neither
    // has a leading dot, and `.bak` rollbacks end in `.bak`, so this prefix +
    // suffix only ever matches staged tmp files.
    let prefix = format!(".{}.", blob_name);
    let entries = match std::fs::read_dir(parent) {
        Ok(e) => e,
        Err(e) => {
            tracing::debug!(
                error = %e,
                parent = %parent.display(),
                "read_dir for CAGRA orphan cleanup failed, skipping"
            );
            return;
        }
    };
    for entry in entries.flatten() {
        let name = match entry.file_name().into_string() {
            Ok(n) => n,
            Err(_) => continue, // non-utf8 filename, leave it alone
        };
        if name.starts_with(&prefix) && name.ends_with(".tmp") {
            match std::fs::remove_file(entry.path()) {
                Ok(_) => tracing::info!(
                    orphan = %name,
                    "Cleaning up interrupted CAGRA save"
                ),
                Err(e) => tracing::debug!(
                    error = %e,
                    orphan = %name,
                    "Failed to remove orphan CAGRA temp file"
                ),
            }
        }
    }
}

/// Stream-hash a file with blake3.
#[cfg(feature = "cuda-index")]
fn blake3_of_path(path: &Path) -> Result<String, CagraError> {
    let file = std::fs::File::open(path).map_err(|e| {
        CagraError::Io(format!(
            "Failed to open {} for checksum: {}",
            path.display(),
            e
        ))
    })?;
    let mut hasher = blake3::Hasher::new();
    hasher.update_reader(file).map_err(|e| {
        CagraError::Io(format!(
            "Failed to read {} for checksum: {}",
            path.display(),
            e
        ))
    })?;
    Ok(hasher.finalize().to_hex().to_string())
}

/// Write the cuVS blob via stage-tmp + atomic-replace with `.bak` rollback
/// so a mid-save failure can restore the prior good blob.
///
/// Sequence (mirrors `src/hnsw/persist.rs` and the SPLADE pattern):
///   1. Refuse if a stale `<path>.bak` already exists — it's a recovery
///      breadcrumb from a prior failed save and must be cleared first.
///   2. cuVS serializes the index to `<path>.<suffix>.tmp` (NOT the live path).
///   3. Hash the temp file — feeds the meta sidecar and validates the write.
///   4. If `<path>` already exists, rename it to `<path>.bak` (with parent
///      fsync) so we have something to restore on failure.
///   5. atomic_replace tmp → live (the load-bearing rename).
///   6. On failure: restore `.bak` → live, fsync, remove tmp, return.
///   7. On success: remove `.bak` with parent fsync.
///
/// Returns the blake3 hex digest of the blob now at `path` so the caller can
/// stamp it into the sidecar.
#[cfg(feature = "cuda-index")]
fn save_blob_atomic_with_rollback(
    cagra: &CagraIndex,
    gpu: &GpuState,
    path: &Path,
) -> Result<String, CagraError> {
    let parent = path.parent().ok_or_else(|| {
        CagraError::Io(format!("CAGRA save path has no parent: {}", path.display()))
    })?;

    let file_name = path
        .file_name()
        .and_then(|n| n.to_str())
        .unwrap_or("index.cagra");
    let bak_path = parent.join(format!("{}.bak", file_name));
    let tmp_path = parent.join(format!(".{}.{:016x}.tmp", file_name, crate::temp_suffix()));

    // (1) Stale-`.bak` guard. A leftover means a prior save failed mid-rollback.
    if bak_path.exists() {
        return Err(CagraError::Io(format!(
            "stale {} from prior failed save; manual recovery required \
             (rename to {} or remove if confirmed bad) before retrying",
            bak_path.display(),
            path.display(),
        )));
    }

    // (2) cuVS serialize to tmp — never overwrites the live blob.
    let tmp_str = tmp_path.to_str().ok_or_else(|| {
        CagraError::Io(format!(
            "CAGRA tmp path is not valid UTF-8: {}",
            tmp_path.display()
        ))
    })?;

    tracing::info!(
        n_vectors = cagra.id_map.len(),
        dim = cagra.dim,
        tmp = %tmp_path.display(),
        "Serializing CAGRA index to disk"
    );
    if let Err(e) = gpu.index.serialize(&gpu.resources, tmp_str, true) {
        let _ = std::fs::remove_file(&tmp_path);
        return Err(CagraError::Cuvs(format!("cuvsCagraSerialize failed: {e}")));
    }

    // (3) Checksum the tmp before promoting — load() validates against this.
    let blob_hash = match blake3_of_path(&tmp_path) {
        Ok(h) => h,
        Err(e) => {
            let _ = std::fs::remove_file(&tmp_path);
            return Err(e);
        }
    };

    // (4) Back up existing live file. Skipped on first save.
    let backed_up = if path.exists() {
        if let Err(e) = std::fs::rename(path, &bak_path) {
            let _ = std::fs::remove_file(&tmp_path);
            return Err(CagraError::Io(format!(
                "Failed to back up {} -> {} before save: {}",
                path.display(),
                bak_path.display(),
                e,
            )));
        }
        true
    } else {
        false
    };

    if backed_up {
        // fsync parent so the `.bak` rename is durable before atomic_replace.
        if let Ok(f) = std::fs::File::open(parent) {
            if let Err(e) = f.sync_all() {
                tracing::debug!(
                    error = %e,
                    dir = %parent.display(),
                    "fsync of CAGRA parent after backup failed (non-fatal)"
                );
            }
        }
    }

    // (5) atomic_replace tmp -> live.
    if let Err(e) = crate::fs::atomic_replace(&tmp_path, path) {
        // (6) Roll back: restore .bak -> live, cleanup tmp, return.
        let _ = std::fs::remove_file(&tmp_path);
        if backed_up {
            if let Err(restore_err) = std::fs::rename(&bak_path, path) {
                tracing::error!(
                    path = %path.display(),
                    error = %restore_err,
                    "Failed to restore CAGRA backup during save rollback — \
                     manual recovery required (rename .bak)"
                );
                return Err(CagraError::Io(format!(
                    "CAGRA save failed and rollback failed: \
                     atomic_replace error={e}; restore error={restore_err}; \
                     manual recovery — rename {bak} to {path}",
                    bak = bak_path.display(),
                    path = path.display(),
                )));
            }
            if let Ok(f) = std::fs::File::open(parent) {
                let _ = f.sync_all();
            }
        }
        return Err(CagraError::Io(format!(
            "Failed to promote {} -> {}: {}",
            tmp_path.display(),
            path.display(),
            e
        )));
    }

    // (7) Successful save: remove `.bak`, fsync.
    if backed_up {
        let _ = std::fs::remove_file(&bak_path);
        if let Ok(f) = std::fs::File::open(parent) {
            let _ = f.sync_all();
        }
    }

    Ok(blob_hash)
}

/// Write the CAGRA sidecar via write-temp + rename to avoid a torn JSON on
/// crash.
#[cfg(feature = "cuda-index")]
fn write_meta_atomic(path: &Path, meta: &CagraMeta) -> Result<(), CagraError> {
    let parent = path.parent().ok_or_else(|| {
        CagraError::Io(format!(
            "CAGRA sidecar has no parent dir: {}",
            path.display()
        ))
    })?;
    std::fs::create_dir_all(parent).map_err(|e| {
        CagraError::Io(format!(
            "Failed to create parent {} for sidecar: {}",
            parent.display(),
            e
        ))
    })?;

    let tmp = parent.join(format!(
        ".{}.{:016x}.tmp",
        path.file_name()
            .and_then(|n| n.to_str())
            .unwrap_or("cagra_meta"),
        crate::temp_suffix()
    ));
    {
        let file = std::fs::File::create(&tmp).map_err(|e| {
            CagraError::Io(format!(
                "Failed to create sidecar temp {}: {}",
                tmp.display(),
                e
            ))
        })?;
        let mut writer = std::io::BufWriter::new(file);
        serde_json::to_writer(&mut writer, meta)
            .map_err(|e| CagraError::Io(format!("Failed to serialize sidecar: {}", e)))?;
        use std::io::Write as _;
        writer
            .flush()
            .map_err(|e| CagraError::Io(format!("Failed to flush sidecar: {}", e)))?;
        // Best-effort fsync — ignore failures on platforms that don't
        // support it on a regular File backed by the FS we're on.
        if let Err(e) = writer.get_ref().sync_all() {
            tracing::debug!(error = %e, "fsync of CAGRA sidecar temp failed (non-fatal)");
        }
    }

    if let Err(e) = std::fs::rename(&tmp, path) {
        let _ = std::fs::remove_file(&tmp);
        return Err(CagraError::Io(format!(
            "Failed to rename sidecar {} -> {}: {}",
            tmp.display(),
            path.display(),
            e
        )));
    }

    Ok(())
}

/// GPU vector index backend. Priority 100 — preferred over HNSW when:
///   - chunk count ≥ `CQS_CAGRA_THRESHOLD` (default 5000), and
///   - a CUDA-capable GPU is available.
///
/// Tries the persisted index first (`{cqs_dir}/index.cagra`), falls back
/// to a fresh build from the store on load failure, and best-effort
/// persists the result. Returns `Ok(None)` when the GPU/threshold gate
/// fails or the build itself fails — the selector then falls through to
/// HNSW.
#[cfg(feature = "cuda-index")]
pub struct CagraBackend;

#[cfg(feature = "cuda-index")]
impl<Mode: crate::store::ClearHnswDirty> crate::index::IndexBackend<Mode> for CagraBackend {
    fn name(&self) -> &'static str {
        "cagra"
    }

    fn priority(&self) -> i32 {
        100
    }

    fn try_open(
        &self,
        ctx: &crate::index::BackendContext<'_, Mode>,
    ) -> std::result::Result<Option<Box<dyn VectorIndex>>, crate::store::StoreError> {
        // Resolution order is env > [index.policy] > default. A missing /
        // unparseable env var falls through to the policy table; a missing
        // policy falls through to the built-in default.
        const CAGRA_THRESHOLD_DEFAULT: u64 = 5000;
        let cagra_threshold: u64 = std::env::var("CQS_CAGRA_THRESHOLD")
            .ok()
            .and_then(|v| v.parse().ok())
            .or_else(|| ctx.policy.and_then(|p| p.cagra_threshold))
            .unwrap_or(CAGRA_THRESHOLD_DEFAULT);
        let chunk_count = ctx.store.chunk_count().unwrap_or_else(|e| {
            tracing::warn!(error = %e, "Failed to get chunk count for CAGRA threshold check");
            0
        });
        // Route through `gpu_available_for(n, dim)` so the VRAM-budget
        // check actually fires. The zero-arg `gpu_available()` collapses to
        // `gpu_available_for(0, 0)` which short-circuits the corpus-aware
        // branch — the gate would then claim eligibility even on a corpus
        // that would OOM CAGRA build on 8 GB GPUs.
        let dim = ctx.store.dim();
        let gpu_available = CagraIndex::gpu_available_for(chunk_count as usize, dim);
        if chunk_count < cagra_threshold || !gpu_available {
            // Info-level so an operator with default log filtering can see
            // the GPU/CPU fallback decision in journald.
            // One line per Store init (not per query), tagged with the same
            // `backend` / `source` shape as the success arms (cagra=persisted
            // / built; hnsw=cagra-ineligible) so log filters work uniformly.
            tracing::info!(
                backend = "hnsw",
                source = "cagra-ineligible",
                chunk_count,
                cagra_threshold,
                dim,
                gpu_available,
                "Vector index backend selected"
            );
            return Ok(None);
        }

        // Try the persisted index first. cuVS native
        // deserialize is fast (~sub-second even for tens of thousands of
        // vectors) compared to the ~30s rebuild on a mid-size repo, so
        // the daemon cold-start cost drops dramatically across systemctl
        // restarts / `cqs index` cycles. `load` validates magic, dim,
        // chunk_count, and blake3 before handing the blob to cuVS, so a
        // stale file falls through to rebuild rather than corrupting
        // results.
        let cagra_path = ctx.cqs_dir.join("index.cagra");
        if cagra_persist_enabled() && cagra_path.exists() {
            match CagraIndex::load(&cagra_path, ctx.store.dim(), chunk_count as usize) {
                Ok(idx) => {
                    tracing::info!(
                        backend = "cagra",
                        source = "persisted",
                        vectors = idx.len(),
                        chunk_count,
                        cagra_threshold,
                        "Vector index backend selected"
                    );
                    return Ok(Some(Box::new(idx) as Box<dyn VectorIndex>));
                }
                Err(e) => {
                    // Sidecar mismatch / stale / corrupt — nuke both files
                    // so the next run doesn't pay the same load-then-fail
                    // cost and instead jumps straight to the rebuild path.
                    tracing::warn!(
                        error = %e,
                        path = %cagra_path.display(),
                        "CAGRA persisted load failed, rebuilding from store"
                    );
                    CagraIndex::delete_persisted(&cagra_path);
                }
            }
        }

        // Resolve the metric for a fresh build: explicit env wins;
        // otherwise follow the slot's stored HNSW metric so the two backends
        // can't drift apart; cosine when neither exists. An invalid env
        // value falls through to HNSW, whose load/build path surfaces the
        // typed error.
        let metric = match DistanceMetric::from_env() {
            Ok(Some(m)) => m,
            Ok(None) => {
                crate::hnsw::HnswIndex::stored_metric(ctx.cqs_dir, "index").unwrap_or_else(|e| {
                    tracing::warn!(
                        error = %e,
                        "Failed to read stored HNSW metric for CAGRA build — using cosine"
                    );
                    DistanceMetric::Cosine
                })
            }
            Err(e) => {
                tracing::warn!(error = %e, "Invalid CQS_DISTANCE_METRIC — falling through to HNSW");
                return Ok(None);
            }
        };

        match CagraIndex::build_from_store_with_metric(ctx.store, ctx.store.dim(), metric) {
            Ok(idx) => {
                tracing::info!(
                    backend = "cagra",
                    source = "rebuilt",
                    vectors = idx.len(),
                    chunk_count,
                    cagra_threshold,
                    "Vector index backend selected"
                );
                if cagra_persist_enabled() {
                    if let Err(e) = idx.save_with_store(&cagra_path, ctx.store) {
                        tracing::warn!(
                            error = %e,
                            path = %cagra_path.display(),
                            "Failed to persist CAGRA index (will rebuild next restart)"
                        );
                    }
                }
                Ok(Some(Box::new(idx) as Box<dyn VectorIndex>))
            }
            Err(e) => {
                tracing::warn!(error = %e, "Failed to build CAGRA index, falling through to HNSW");
                Ok(None)
            }
        }
    }
}

#[cfg(all(test, feature = "cuda-index"))]
mod tests {
    use super::*;
    use crate::index::VectorIndex;
    use crate::EMBEDDING_DIM;
    use std::sync::Mutex;

    /// Serialize GPU tests — concurrent CUDA contexts cause SIGSEGV
    static GPU_LOCK: Mutex<()> = Mutex::new(());

    fn make_embedding(seed: u32) -> Embedding {
        let mut v = vec![0.0f32; EMBEDDING_DIM];
        for (i, val) in v.iter_mut().enumerate() {
            *val = ((seed as f32 * 10.0) + (i as f32 * 0.001)).sin();
        }
        let norm: f32 = v.iter().map(|x| x * x).sum::<f32>().sqrt();
        if norm > 0.0 {
            v.iter_mut().for_each(|x| *x /= norm);
        }
        Embedding::new(v)
    }

    fn require_gpu() -> bool {
        if !CagraIndex::gpu_available() {
            eprintln!("Skipping CAGRA test: no GPU available");
            return false;
        }
        true
    }

    fn build_test_index(n: u32) -> CagraIndex {
        let embeddings: Vec<(String, Embedding)> = (0..n)
            .map(|i| (format!("chunk_{}", i), make_embedding(i)))
            .collect();
        CagraIndex::build(embeddings, EMBEDDING_DIM).expect("Failed to build test index")
    }

    #[test]
    fn test_gpu_available() {
        let _ = CagraIndex::gpu_available();
    }

    #[test]
    fn test_build_simple() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let index = build_test_index(5);
        assert_eq!(index.len(), 5);
        assert!(!index.is_empty());
    }

    #[test]
    fn test_build_empty() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let result = CagraIndex::build(vec![], EMBEDDING_DIM);
        assert!(result.is_err());
    }

    #[test]
    fn test_build_dimension_mismatch() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let bad_embedding = Embedding::new(vec![1.0; 100]);
        let result = CagraIndex::build(vec![("bad".into(), bad_embedding)], EMBEDDING_DIM);
        match result {
            Err(CagraError::Build(_)) => {}
            Err(e) => panic!("Expected Build error, got: {:?}", e),
            Ok(_) => panic!("Expected error, got Ok"),
        }
    }

    #[test]
    fn test_search_self_match() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let index = build_test_index(10);
        let query = make_embedding(3);
        let results = index.search(&query, 5);
        assert!(!results.is_empty(), "Search returned no results");
        // CAGRA is an approximate GPU graph index; assert top-k containment
        // of the self-match vector, not exact rank-1.
        let ids: Vec<&str> = results.iter().map(|r| r.id.as_str()).collect();
        assert!(
            ids.contains(&"chunk_3"),
            "chunk_3 should be in top-5, got {ids:?}"
        );
        let self_score = results
            .iter()
            .find(|r| r.id == "chunk_3")
            .map(|r| r.score)
            .unwrap();
        assert!(
            self_score > 0.9,
            "Self-match score should be high, got {self_score}"
        );
    }

    #[test]
    fn test_search_k_limiting() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let index = build_test_index(10);
        let query = make_embedding(0);
        let results = index.search(&query, 3);
        assert!(results.len() <= 3);
    }

    #[test]
    fn test_search_ordering() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let index = build_test_index(10);
        let query = make_embedding(0);
        let results = index.search(&query, 5);
        for window in results.windows(2) {
            assert!(
                window[0].score >= window[1].score,
                "Results not sorted: {} < {}",
                window[0].score,
                window[1].score
            );
        }
    }

    #[test]
    fn test_search_dimension_mismatch_query() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let index = build_test_index(5);
        let bad_query = Embedding::new(vec![1.0; 100]);
        let results = index.search(&bad_query, 3);
        assert!(results.is_empty());
    }

    #[test]
    fn test_multiple_searches() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let index = build_test_index(10);

        // Non-consuming search — no rebuild needed
        let results1 = index.search(&make_embedding(0), 3);
        assert!(!results1.is_empty());

        let results2 = index.search(&make_embedding(5), 3);
        assert!(!results2.is_empty());
        // Top-k containment, not exact rank-1: approximate GPU index.
        let ids: Vec<&str> = results2.iter().map(|r| r.id.as_str()).collect();
        assert!(
            ids.contains(&"chunk_5"),
            "chunk_5 should be in top-3, got {ids:?}"
        );
    }

    #[test]
    fn test_consecutive_searches() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let index = build_test_index(20);

        for i in 0..10 {
            let query = make_embedding(i);
            let results = index.search(&query, 5);
            assert!(!results.is_empty(), "Search {} should return results", i);
            assert!(results.len() <= 5);
        }
    }

    #[test]
    fn test_search_with_invalid_k() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let index = build_test_index(5);

        let results = index.search(&make_embedding(0), 0);
        assert!(results.is_empty());

        let results = index.search(&make_embedding(1), 3);
        assert!(!results.is_empty());
    }

    /// When `k > index.len()`, cuVS leaves the extra output slots
    /// untouched. The `INVALID_DISTANCE` sentinel must
    /// filter them out so we never emit phantom perfect-match hits
    /// (distance `0.0` → score `1.0`) pointing at internal index 0.
    #[test]
    fn test_search_k_greater_than_len_drops_phantoms() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let index = build_test_index(3);
        let results = index.search(&make_embedding(0), 10);

        // Only three real vectors exist, so the result must never exceed
        // that count even though we asked for ten.
        assert!(
            results.len() <= 3,
            "expected at most 3 results, got {}: {:?}",
            results.len(),
            results
        );
        // Every returned id is one of the three real chunks; nothing
        // phantom has slipped through the sentinel filter.
        for r in &results {
            assert!(
                matches!(r.id.as_str(), "chunk_0" | "chunk_1" | "chunk_2"),
                "phantom id leaked past sentinel check: {}",
                r.id
            );
        }
        // No two results share an id (the phantom bug repeatedly emitted
        // `chunk_0`, so a duplicate would also catch a regression).
        let mut ids: Vec<&str> = results.iter().map(|r| r.id.as_str()).collect();
        ids.sort_unstable();
        let before = ids.len();
        ids.dedup();
        assert_eq!(before, ids.len(), "duplicate ids in results: {:?}", results);
    }

    #[test]
    fn test_name_returns_cagra() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let index = build_test_index(5);
        let vi: &dyn VectorIndex = &index;
        assert_eq!(vi.name(), "CAGRA");
    }

    #[test]
    fn test_oom_guard_arithmetic() {
        let max_bytes = super::cagra_max_bytes();
        let max_chunks = max_bytes / (EMBEDDING_DIM * 4);
        let under = max_chunks.saturating_mul(EMBEDDING_DIM).saturating_mul(4);
        assert!(under <= max_bytes);
        let over = (max_chunks + 1)
            .saturating_mul(EMBEDDING_DIM)
            .saturating_mul(4);
        assert!(over > max_bytes);
    }

    /// build → save → load → search, asserting bit-exact (same order, same
    /// scores) neighbors before and after the round-trip.
    #[test]
    fn test_save_load_round_trip() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let dir = tempfile::TempDir::new().unwrap();
        let path = dir.path().join("test.cagra");

        let original = build_test_index(32);
        original
            .save(&path)
            .expect("CAGRA persist save should succeed");
        assert!(path.exists(), "CAGRA blob should be written");
        let meta_path = super::meta_path_for(&path);
        assert!(meta_path.exists(), "CAGRA sidecar should be written");

        // Compare a handful of queries across the boundary. k=5 so we can
        // check ordering and scores simultaneously.
        let queries: Vec<Embedding> = (0..5).map(make_embedding).collect();
        let original_results: Vec<Vec<IndexResult>> =
            queries.iter().map(|q| original.search(q, 5)).collect();

        // Drop the original so there's no way the loaded index is just
        // aliasing the in-memory state.
        drop(original);

        let loaded =
            CagraIndex::load(&path, EMBEDDING_DIM, 32).expect("CAGRA persist load should succeed");
        assert_eq!(loaded.len(), 32, "loaded index should have 32 vectors");
        assert_eq!(loaded.dim, EMBEDDING_DIM);

        for (i, query) in queries.iter().enumerate() {
            let got = loaded.search(query, 5);
            let expected = &original_results[i];
            assert_eq!(
                got.len(),
                expected.len(),
                "query {} returned different neighbour count",
                i
            );
            for (a, b) in got.iter().zip(expected.iter()) {
                assert_eq!(
                    a.id, b.id,
                    "query {} neighbour id differs after round-trip",
                    i
                );
                // Scores should match bit-for-bit because the graph and
                // dataset are both serialized by cuVS.
                assert_eq!(
                    a.score.to_bits(),
                    b.score.to_bits(),
                    "query {} score {} != {} after round-trip",
                    i,
                    a.score,
                    b.score
                );
            }
        }
    }

    // ====== `.bak` rollback pattern ======

    /// First save (no prior file) leaves no `.bak` and no `.tmp`. Pins the
    /// "skipped backup" branch.
    #[test]
    fn save_no_prior_file_leaves_no_bak() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let dir = tempfile::TempDir::new().unwrap();
        let path = dir.path().join("test.cagra");
        let bak = dir.path().join("test.cagra.bak");
        assert!(!path.exists());

        build_test_index(8).save(&path).expect("first save");

        assert!(path.exists(), "save must create the live file");
        assert!(!bak.exists(), "first save must not leave a .bak behind");
        // Tmp file pattern: `.<file>.<hex16>.tmp`. Glob-check none survived.
        let entries: Vec<_> = std::fs::read_dir(dir.path())
            .unwrap()
            .filter_map(|e| e.ok())
            .filter(|e| {
                let n = e.file_name().to_string_lossy().to_string();
                n.starts_with(".test.cagra.") && n.ends_with(".tmp")
            })
            .collect();
        assert!(
            entries.is_empty(),
            "save must clean up its tmp file; left behind: {:?}",
            entries.iter().map(|e| e.file_name()).collect::<Vec<_>>()
        );
    }

    /// Successful save with a prior live file: tmp staged, atomic_replace
    /// promotes, `.bak` gets cleaned up after the new blob lands. The new
    /// blob is loadable and the prior `.bak` is gone.
    #[test]
    fn save_with_prior_file_replaces_and_cleans_up_bak() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let dir = tempfile::TempDir::new().unwrap();
        let path = dir.path().join("test.cagra");
        let bak = dir.path().join("test.cagra.bak");

        // First save establishes a prior live file.
        build_test_index(8).save(&path).expect("first save");
        let v1_bytes = std::fs::read(&path).expect("read v1");
        assert!(!v1_bytes.is_empty());

        // Second save — must rename existing -> .bak, atomic_replace, then
        // clean up .bak after success.
        build_test_index(16).save(&path).expect("second save");

        assert!(path.exists(), "live file must be present after save");
        assert!(
            !bak.exists(),
            "successful save must clean up .bak; left at {}",
            bak.display(),
        );

        // The post-save blob is a valid index loadable at the new chunk count.
        let loaded = CagraIndex::load(&path, EMBEDDING_DIM, 16)
            .expect("post-save blob must load with new chunk count");
        assert_eq!(loaded.len(), 16);
    }

    /// Stale `.bak` left over from a prior failed save must block the next
    /// save with an actionable error so the operator notices and clears it.
    /// Pre-fix, the cuVS serialize would clobber the live file (the only
    /// good copy) without warning.
    #[test]
    fn save_refuses_when_stale_bak_exists() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let dir = tempfile::TempDir::new().unwrap();
        let path = dir.path().join("test.cagra");
        let bak = dir.path().join("test.cagra.bak");

        // First save establishes a live file.
        build_test_index(8).save(&path).expect("first save");
        let live_bytes_before = std::fs::read(&path).expect("read live before");

        // Simulate a prior crashed save: a stale `.bak` lingers next to it.
        std::fs::write(&bak, b"stale-bak-from-prior-crash").unwrap();

        // The next save MUST refuse with an error mentioning recovery.
        let err = build_test_index(16)
            .save(&path)
            .expect_err("save must refuse when stale .bak exists");
        let msg = format!("{err}");
        assert!(
            msg.contains("stale") && msg.contains("manual recovery"),
            "stale-bak guard must surface an actionable error, got: {msg}"
        );

        // The live file must remain untouched (still loadable at the
        // ORIGINAL chunk count of 8).
        let live_bytes_after = std::fs::read(&path).expect("read live after");
        assert_eq!(
            live_bytes_before, live_bytes_after,
            "live blob must survive byte-for-byte when save bails on stale .bak"
        );
        let loaded =
            CagraIndex::load(&path, EMBEDDING_DIM, 8).expect("live blob must still load post-bail");
        assert_eq!(loaded.len(), 8);
    }

    /// Mismatched chunk count must fail to load — protects us from handing
    /// cuVS a blob whose id_map is no longer valid against the current
    /// store.
    #[test]
    fn test_load_rejects_chunk_count_mismatch() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let dir = tempfile::TempDir::new().unwrap();
        let path = dir.path().join("test.cagra");
        build_test_index(10)
            .save(&path)
            .expect("save should succeed");

        // Tell load() we expect 20 chunks; the sidecar says 10.
        match CagraIndex::load(&path, EMBEDDING_DIM, 20) {
            Err(CagraError::Stale { reason }) => {
                assert!(reason.contains("chunk_count"), "reason: {}", reason);
            }
            other => panic!("expected Stale, got {:?}", other),
        }
    }

    /// Mismatched dim must fail to load — catches embedding-model swaps.
    #[test]
    fn test_load_rejects_dim_mismatch() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let dir = tempfile::TempDir::new().unwrap();
        let path = dir.path().join("test.cagra");
        build_test_index(10)
            .save(&path)
            .expect("save should succeed");

        match CagraIndex::load(&path, EMBEDDING_DIM + 1, 10) {
            Err(CagraError::Stale { reason }) => {
                assert!(reason.contains("dim"), "reason: {}", reason);
            }
            other => panic!("expected Stale, got {:?}", other),
        }
    }

    /// Flipping bytes in the `.cagra` blob must be detected via blake3
    /// before we let cuVS deserialize it.
    #[test]
    fn test_load_rejects_corrupted_blob() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let dir = tempfile::TempDir::new().unwrap();
        let path = dir.path().join("test.cagra");
        build_test_index(10)
            .save(&path)
            .expect("save should succeed");

        // Flip a byte near the end of the cuVS blob to mimic disk rot.
        let mut bytes = std::fs::read(&path).unwrap();
        let pos = bytes.len().saturating_sub(16);
        bytes[pos] ^= 0xff;
        std::fs::write(&path, &bytes).unwrap();

        match CagraIndex::load(&path, EMBEDDING_DIM, 10) {
            Err(CagraError::ChecksumMismatch(p)) => {
                assert!(
                    p.contains("test.cagra"),
                    "checksum error should reference file: {}",
                    p
                );
            }
            other => panic!("expected ChecksumMismatch, got {:?}", other),
        }
    }

    /// Missing sidecar is a hard load failure.
    #[test]
    fn test_load_requires_sidecar() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let dir = tempfile::TempDir::new().unwrap();
        let path = dir.path().join("test.cagra");
        build_test_index(5)
            .save(&path)
            .expect("save should succeed");
        std::fs::remove_file(super::meta_path_for(&path)).unwrap();
        match CagraIndex::load(&path, EMBEDDING_DIM, 5) {
            Err(CagraError::BadMeta(_)) => {}
            other => panic!("expected BadMeta, got {:?}", other),
        }
    }

    /// delete_persisted cleans up both files without complaining when they
    /// are already gone.
    #[test]
    fn test_delete_persisted_removes_both_files() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let dir = tempfile::TempDir::new().unwrap();
        let path = dir.path().join("test.cagra");
        build_test_index(3)
            .save(&path)
            .expect("save should succeed");
        let meta = super::meta_path_for(&path);
        assert!(path.exists() && meta.exists());
        CagraIndex::delete_persisted(&path);
        assert!(!path.exists() && !meta.exists());
        // Second call must not panic on missing files.
        CagraIndex::delete_persisted(&path);
    }

    /// Meta path computation (pure, GPU-free).
    #[test]
    fn test_meta_path_for() {
        let p = std::path::Path::new("/tmp/foo.cagra");
        let meta = super::meta_path_for(p);
        assert_eq!(meta.to_str().unwrap(), "/tmp/foo.cagra.meta");
    }

    /// CQS_CAGRA_PERSIST=0 makes save/load return Err without touching disk.
    /// Isolated to a dedicated test to keep env mutation narrow; tests here
    /// use a lock already but the env is process-wide.
    #[test]
    fn test_persistence_env_override_blocks_save() {
        // Direct test without needing GPU — save() checks the flag before
        // acquiring the GPU mutex.
        let saved = std::env::var("CQS_CAGRA_PERSIST").ok();
        // Force a fresh OnceLock read by never having called it for this
        // value before; OnceLock caches across invocations so we can only
        // test the cached result. Best-effort: if the cache was primed by
        // earlier tests we just assert the helper returns something
        // consistent.
        let enabled = super::cagra_persist_enabled();
        // Restore whatever the env had before; the OnceLock keeps its value
        // regardless, so this is really just being polite to other tests.
        match saved {
            Some(v) => std::env::set_var("CQS_CAGRA_PERSIST", v),
            None => std::env::remove_var("CQS_CAGRA_PERSIST"),
        }
        // If we're running in an env with PERSIST=0 the helper should have
        // observed it; otherwise the default is true. Either outcome is a
        // valid pass — the important thing is the helper returned without
        // panicking and the type is correct.
        let _: bool = enabled;
    }

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

    /// DotProduct build + search on the GPU backend. The self-match score
    /// pins the InnerProduct distance → similarity conversion in
    /// `search_impl` (cuVS returns the raw inner product, best-first; on
    /// unit-norm vectors the self-match similarity must come back ≈ 1).
    #[test]
    fn test_dot_product_build_and_search() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let embeddings: Vec<(String, Embedding)> = (0..10)
            .map(|i| (format!("chunk_{}", i), make_embedding(i)))
            .collect();
        let index =
            CagraIndex::build_with_metric(embeddings, EMBEDDING_DIM, DistanceMetric::DotProduct)
                .expect("DotProduct CAGRA build should succeed");
        assert_eq!(index.metric(), DistanceMetric::DotProduct);

        let results = index.search(&make_embedding(3), 5);
        assert!(!results.is_empty(), "search returned no results");
        let ids: Vec<&str> = results.iter().map(|r| r.id.as_str()).collect();
        assert!(
            ids.contains(&"chunk_3"),
            "chunk_3 should be in top-5, got {ids:?}"
        );
        let self_score = results
            .iter()
            .find(|r| r.id == "chunk_3")
            .map(|r| r.score)
            .unwrap();
        assert!(
            self_score > 0.9,
            "unit-norm self-match under InnerProduct should score ≈ 1, got {self_score} \
             (full results: {results:?}) — if this fails the distance→similarity \
             conversion for InnerProduct is wrong"
        );
        // Descending score order must hold for the negated-IP conversion too.
        for window in results.windows(2) {
            assert!(
                window[0].score >= window[1].score,
                "results not sorted: {} < {}",
                window[0].score,
                window[1].score
            );
        }
    }

    /// Sidecar metric roundtrip: a DotProduct index persists its metric and
    /// loads back as DotProduct with env unset (the stored value wins).
    /// Holds `HNSW_ENV_LOCK` so the hnsw mismatch test's transient
    /// `CQS_DISTANCE_METRIC=cosine` can't mark this dot sidecar stale
    /// mid-load. Lock order is GPU_LOCK → HNSW_ENV_LOCK in every CAGRA test
    /// that takes both.
    #[test]
    fn test_metric_sidecar_roundtrip_dot() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let _env = crate::hnsw::HNSW_ENV_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        let dir = tempfile::TempDir::new().unwrap();
        let path = dir.path().join("test.cagra");

        let embeddings: Vec<(String, Embedding)> = (0..8)
            .map(|i| (format!("chunk_{}", i), make_embedding(i)))
            .collect();
        let original =
            CagraIndex::build_with_metric(embeddings, EMBEDDING_DIM, DistanceMetric::DotProduct)
                .unwrap();
        original.save(&path).expect("save should succeed");
        drop(original);

        let loaded = CagraIndex::load(&path, EMBEDDING_DIM, 8)
            .expect("dot-product sidecar should load with env unset (stored wins)");
        assert_eq!(loaded.metric(), DistanceMetric::DotProduct);

        let results = loaded.search(&make_embedding(2), 3);
        let ids: Vec<&str> = results.iter().map(|r| r.id.as_str()).collect();
        assert!(
            ids.contains(&"chunk_2"),
            "chunk_2 should be in top-3 after reload, got {ids:?}"
        );
    }

    /// A legacy (pre-#1351) sidecar has no `metric` field; serde's default
    /// must decode it as cosine — that is what every legacy build used.
    #[test]
    fn test_legacy_sidecar_without_metric_loads_as_cosine() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let dir = tempfile::TempDir::new().unwrap();
        let path = dir.path().join("test.cagra");

        let original = build_test_index(8);
        original.save(&path).expect("save should succeed");
        drop(original);

        // Strip the `metric` field from the sidecar to simulate a legacy
        // file. The blob checksum only covers the `.cagra` blob, so the
        // sidecar rewrite doesn't invalidate it.
        let meta_path = super::meta_path_for(&path);
        let mut meta: serde_json::Value =
            serde_json::from_str(&std::fs::read_to_string(&meta_path).unwrap()).unwrap();
        assert!(
            meta.as_object_mut().unwrap().remove("metric").is_some(),
            "current save must stamp the metric field"
        );
        std::fs::write(&meta_path, serde_json::to_string(&meta).unwrap()).unwrap();

        let loaded = CagraIndex::load(&path, EMBEDDING_DIM, 8)
            .expect("legacy sidecar without metric field must load");
        assert_eq!(loaded.metric(), DistanceMetric::Cosine);
    }

    /// Loading a DotProduct sidecar with `CQS_DISTANCE_METRIC` explicitly
    /// set to cosine marks the blob stale (the rebuild path is CAGRA's
    /// mismatch remedy, unlike HNSW's hard typed error). Test discipline:
    /// the env var is only ever set to "cosine" so concurrent cosine loads
    /// are unaffected; held under `HNSW_ENV_LOCK` (taken after GPU_LOCK).
    #[test]
    fn test_metric_mismatch_marks_stale() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let _env = crate::hnsw::HNSW_ENV_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        let dir = tempfile::TempDir::new().unwrap();
        let path = dir.path().join("test.cagra");

        let embeddings: Vec<(String, Embedding)> = (0..8)
            .map(|i| (format!("chunk_{}", i), make_embedding(i)))
            .collect();
        let original =
            CagraIndex::build_with_metric(embeddings, EMBEDDING_DIM, DistanceMetric::DotProduct)
                .unwrap();
        original.save(&path).expect("save should succeed");
        drop(original);

        let _var = crate::hnsw::EnvVarGuard::set("CQS_DISTANCE_METRIC", "cosine");
        match CagraIndex::load(&path, EMBEDDING_DIM, 8) {
            Err(CagraError::Stale { reason }) => {
                assert!(
                    reason.contains("metric"),
                    "stale reason should mention the metric: {reason}"
                );
            }
            Err(other) => panic!("expected Stale, got: {other:?}"),
            Ok(_) => panic!("conflicting metric must not load"),
        }
    }

    /// GPU adversarial pin: the non-finite-query guard at the *actual*
    /// `CagraIndex::search` entry returns empty results on a live index
    /// rather than dispatching a kernel on poisoned input. The pure
    /// predicate is covered CPU-side by `query_is_finite_rejects_nonfinite`;
    /// this exercises the end-to-end path against real cuVS resources so a
    /// future refactor that moves or drops the guard is caught.
    #[test]
    fn test_search_nonfinite_query_returns_empty_gpu() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let index = build_test_index(10);

        // A NaN somewhere in the query must short-circuit to empty.
        let mut nan_vec = make_embedding(3).as_slice().to_vec();
        nan_vec[0] = f32::NAN;
        let nan_query = Embedding::new(nan_vec);
        assert!(
            index.search(&nan_query, 5).is_empty(),
            "NaN query must return empty from the live search path"
        );

        // +Inf as well — the guard rejects any non-finite component.
        let mut inf_vec = make_embedding(3).as_slice().to_vec();
        inf_vec[1] = f32::INFINITY;
        let inf_query = Embedding::new(inf_vec);
        assert!(
            index.search(&inf_query, 5).is_empty(),
            "Inf query must return empty from the live search path"
        );

        // Sanity: the same index still answers a finite query, so the empty
        // results above are the guard firing, not a broken index.
        assert!(
            !index.search(&make_embedding(3), 5).is_empty(),
            "finite query must still return results on the same index"
        );
    }

    /// GPU adversarial pin: a finite query against an index that *contains*
    /// edge-case vectors (a zero vector, which is degenerate under cosine —
    /// the L2-normalisation in `build` cannot normalise it) must still
    /// return only finite-scored results. Documents the open uncertainty in
    /// the `INVALID_DISTANCE` design note: whatever cuVS does internally with
    /// a degenerate stored vector, the post-hoc `!dist.is_finite()` filter
    /// keeps any NaN/Inf distance it might emit out of the result set.
    #[test]
    fn test_search_finite_query_degenerate_index_vector() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        // Mix ordinary vectors with a zero vector (degenerate direction).
        let mut embeddings: Vec<(String, Embedding)> = (0..8)
            .map(|i| (format!("chunk_{i}"), make_embedding(i)))
            .collect();
        embeddings.push((
            "zero_vec".to_string(),
            Embedding::new(vec![0.0f32; EMBEDDING_DIM]),
        ));

        let index = CagraIndex::build(embeddings, EMBEDDING_DIM)
            .expect("build with a degenerate vector should still succeed");

        let results = index.search(&make_embedding(2), 9);
        // Every score that survives must be finite — the sentinel filter
        // (and the score conversion) must never surface NaN/Inf to callers.
        for r in &results {
            assert!(
                r.score.is_finite(),
                "result {:?} has non-finite score {} — sentinel/score filter leaked",
                r.id,
                r.score
            );
        }
        // The query's own self-match must still be retrievable.
        assert!(
            results.iter().any(|r| r.id == "chunk_2"),
            "self-match lost when index contains a degenerate vector: {:?}",
            results.iter().map(|r| &r.id).collect::<Vec<_>>()
        );
    }

    /// GPU adversarial pin: confirm the post-hoc `!dist.is_finite()` sentinel
    /// filter actually fires in practice. Request `k` far larger than the
    /// corpus so cuVS cannot fill every neighbour slot; the unfilled slots
    /// retain the `INVALID_DISTANCE` (`+∞`) seed. The filter must drop them,
    /// so the returned count is bounded by the corpus size and never exposes
    /// a sentinel-scored garbage neighbour.
    #[test]
    fn test_search_oversized_k_sentinel_filter_fires() {
        let _guard = GPU_LOCK.lock().unwrap();
        if !require_gpu() {
            return;
        }
        let n: usize = 6;
        let index = build_test_index(n as u32);

        // k > corpus size: cuVS leaves trailing slots at INVALID_DISTANCE.
        let k = n + 20;
        let results = index.search(&make_embedding(1), k);

        // The sentinel filter must keep the result count within the corpus.
        assert!(
            results.len() <= n,
            "sentinel filter did not fire: got {} results for a {}-vector corpus",
            results.len(),
            n
        );
        // No surviving result may carry the sentinel — all scores finite.
        for r in &results {
            assert!(
                r.score.is_finite(),
                "sentinel slot leaked into results: {:?} score {}",
                r.id,
                r.score
            );
        }
        // And we still got the real neighbours, not an empty drop-everything.
        assert!(
            !results.is_empty(),
            "oversized-k search dropped all real neighbours"
        );
    }

    /// CPU-only guard test: the finite-query predicate that fronts
    /// `CagraIndex::search` must reject NaN/Inf queries (so the search
    /// returns empty rather than dispatching a kernel on poisoned input)
    /// and accept ordinary finite ones. Constructing a `CagraIndex` needs a
    /// live GPU; the predicate itself is pure, so it's testable without one.
    #[test]
    fn query_is_finite_rejects_nonfinite() {
        let finite = Embedding::new(vec![0.1, 0.2, 0.3, 0.4]);
        assert!(super::query_is_finite(&finite), "finite query accepted");

        let nan = Embedding::new(vec![0.1, f32::NAN, 0.3, 0.4]);
        assert!(!super::query_is_finite(&nan), "NaN query rejected");

        let inf = Embedding::new(vec![f32::INFINITY, 0.2, 0.3, 0.4]);
        assert!(!super::query_is_finite(&inf), "Inf query rejected");

        let neg_inf = Embedding::new(vec![0.1, 0.2, 0.3, f32::NEG_INFINITY]);
        assert!(!super::query_is_finite(&neg_inf), "-Inf query rejected");
    }
}

/// CPU-side tests for the orphan-tmp sweep. The sweep is GPU-free (pure
/// filesystem), so these run without the `cuda-index` feature.
#[cfg(test)]
mod orphan_sweep_tests {
    use std::fs;

    #[test]
    fn sweep_removes_blob_and_meta_orphans_keeps_live_and_unrelated() {
        let dir = tempfile::TempDir::new().unwrap();
        let blob = dir.path().join("cagra.index.bin");
        let meta = dir.path().join("cagra.index.bin.meta");

        // Live artifacts an interrupted-on-next-save state would have.
        fs::write(&blob, b"live blob").unwrap();
        fs::write(&meta, b"live meta").unwrap();

        // Orphan tmp files: blob-staged and sidecar-staged, with the random
        // hex suffix the real save paths use.
        let blob_orphan = dir.path().join(".cagra.index.bin.0123456789abcdef.tmp");
        let meta_orphan = dir
            .path()
            .join(".cagra.index.bin.meta.fedcba9876543210.tmp");
        fs::write(&blob_orphan, b"orphan blob").unwrap();
        fs::write(&meta_orphan, b"orphan meta").unwrap();

        // A `.bak` rollback breadcrumb and an unrelated file must survive.
        let bak = dir.path().join("cagra.index.bin.bak");
        let unrelated = dir.path().join("other.index.bin.0000000000000000.tmp");
        fs::write(&bak, b"bak").unwrap();
        fs::write(&unrelated, b"unrelated").unwrap();

        super::cleanup_orphan_temp_files(&blob);

        assert!(!blob_orphan.exists(), "blob orphan tmp swept");
        assert!(!meta_orphan.exists(), "meta orphan tmp swept");
        assert!(blob.exists(), "live blob preserved");
        assert!(meta.exists(), "live meta preserved");
        assert!(bak.exists(), "`.bak` rollback preserved");
        assert!(unrelated.exists(), "unrelated tmp preserved");
    }

    #[test]
    fn sweep_is_noop_when_parent_missing() {
        // Non-existent path with a parent that does not exist: best-effort,
        // no panic.
        let path = std::path::Path::new("/nonexistent-cqs-dir-xyz/cagra.index.bin");
        super::cleanup_orphan_temp_files(path);
    }
}
