lucisearch 0.8.0

//! HNSW (Hierarchical Navigable Small World) graph for approximate nearest
//! neighbor search.
//!
//! See [[hierarchical-navigable-small-world]] and [[architecture-overview#Step 2]].

use std::cmp::Ordering;
use std::collections::{BinaryHeap, HashSet};
use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};

use crate::core::LuciError;
use crate::mapping::QuantizationType;
use parking_lot::RwLock;
use rayon::ThreadPoolBuilder;
use rayon::current_num_threads;
use rayon::prelude::*;

use super::{DistanceMetric, normalize_in_place};

/// Sentinel ordinal in the packed [`HnswBuilder::entry`] word meaning
/// "no entry point yet". `u32::MAX` is never a real ordinal (ordinals are
/// dense from 0), so it unambiguously marks the un-bootstrapped state.
/// See [[optimization-concurrent-hnsw-insert]].
const ENTRY_SENTINEL: u32 = u32::MAX;

/// Pack `(entry_point, max_level)` into one `u64` so a concurrent descent
/// reads a consistent pair in a single atomic load — Lucene's
/// `AtomicReference<EntryNode>` as a packed word. High 32 bits = entry
/// ordinal, low 32 = max level.
fn pack_entry(entry_point: u32, max_level: u32) -> u64 {
    ((entry_point as u64) << 32) | (max_level as u64)
}

/// Inverse of [`pack_entry`] — returns `(entry_point, max_level)`.
fn unpack_entry(packed: u64) -> (u32, u32) {
    ((packed >> 32) as u32, packed as u32)
}

/// Thread budget for the parallel connect phase
/// ([[optimization-concurrent-hnsw-insert]]). `Ambient` uses rayon's
/// global pool (the production default); `Fixed(n)` runs in a scoped
/// `n`-thread pool. `Fixed(1)` takes the sequential, bit-identical path
/// (the deterministic test/profile contract).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum BuildThreads {
    Ambient,
    Fixed(usize),
}

/// Magic prefix for v2+ HNSW segments — `b"LHNS"`.
///
/// v1 segments wrote `dims: u32` as their first 4 bytes, so any v1
/// segment whose dims happen to equal `0x534E484C` (1 397 772 876)
/// would collide. No real HNSW segment carries 1.4B dimensions, so the
/// detection is unambiguous. See [[optimize-cosine-norm-precompute]].
pub const HNSW_FORMAT_MAGIC: [u8; 4] = *b"LHNS";

/// Current on-disk HNSW format version. Bumped when the segment bytes
/// change in a way readers must observe. v2 added the magic prefix and
/// the cosine kernel's "vectors are unit-length on disk" invariant.
pub const HNSW_FORMAT_VERSION: u8 = 2;

/// On-disk HNSW format version. v1 is the legacy format with no magic
/// prefix; v2 added [[HNSW_FORMAT_MAGIC]] and the cosine
/// normalize-at-insert invariant.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum HnswFormatVersion {
    V1,
    V2,
}

/// Parsed HNSW header fields plus a byte cursor pointing at the next
/// section (the flat vector array).
#[derive(Clone, Copy, Debug)]
pub struct HnswHeader {
    pub version: HnswFormatVersion,
    pub dims: usize,
    pub m: usize,
    pub metric: DistanceMetric,
    pub num_vectors: usize,
    pub entry_point: Option<u32>,
    pub max_level: usize,
    /// Byte offset of the section immediately after the header, used by
    /// every caller to position its own parse cursor.
    pub vectors_offset: usize,
}

/// Read exactly `n` bytes at `*pos`, advancing `*pos`. Returns
/// [`LuciError::IndexCorrupted`] instead of slice-panicking when the blob
/// is truncated, so a malformed `.luci` vector blob fails loudly rather
/// than out-of-bounds. Per [[cross-platform-portability]] a valid file is
/// never truncated, so this rejects only genuinely-corrupt input. Shared
/// with [`super::global`]'s `load_field`, the other vector-index reader.
pub(crate) fn take_bytes<'a>(
    data: &'a [u8],
    pos: &mut usize,
    n: usize,
) -> Result<&'a [u8], LuciError> {
    let start = *pos;
    let end = start
        .checked_add(n)
        .filter(|&e| e <= data.len())
        .ok_or_else(|| {
            LuciError::IndexCorrupted(format!(
                "vector index blob truncated: need {n} bytes at offset {start}, have {} total",
                data.len()
            ))
        })?;
    *pos = end;
    Ok(&data[start..end])
}

pub(crate) fn read_u32(data: &[u8], pos: &mut usize) -> Result<u32, LuciError> {
    Ok(u32::from_le_bytes(
        take_bytes(data, pos, 4)?.try_into().unwrap(),
    ))
}

pub(crate) fn read_u64(data: &[u8], pos: &mut usize) -> Result<u64, LuciError> {
    Ok(u64::from_le_bytes(
        take_bytes(data, pos, 8)?.try_into().unwrap(),
    ))
}

fn read_u8(data: &[u8], pos: &mut usize) -> Result<u8, LuciError> {
    Ok(take_bytes(data, pos, 1)?[0])
}

fn read_f32(data: &[u8], pos: &mut usize) -> Result<f32, LuciError> {
    Ok(f32::from_le_bytes(
        take_bytes(data, pos, 4)?.try_into().unwrap(),
    ))
}

/// Validate a length prefix before it is used to pre-allocate, so a corrupt
/// count cannot drive a huge speculative `Vec::with_capacity` on malformed
/// input. Each element consumes at least `min_elem_bytes` downstream, so a
/// count exceeding `remaining / min_elem_bytes` cannot be satisfied by the
/// blob and is rejected as corruption. Shared with [`super::global`]'s
/// `load_field`.
pub(crate) fn checked_len(
    count: usize,
    min_elem_bytes: usize,
    data: &[u8],
    pos: usize,
) -> Result<usize, LuciError> {
    let remaining = data.len().saturating_sub(pos);
    if min_elem_bytes != 0 && count > remaining / min_elem_bytes {
        return Err(LuciError::IndexCorrupted(format!(
            "vector index blob declares {count} elements (≥{min_elem_bytes} B each) \
             but only {remaining} bytes remain"
        )));
    }
    Ok(count)
}

/// Parse the HNSW segment header.
///
/// Dispatches on the 4-byte magic prefix: v2 segments start with `LHNS`
/// followed by a version byte; everything else is treated as a v1
/// segment. v1 cosine segments are rejected with a migration error —
/// they stored raw vectors and the v2 kernel requires unit-length
/// vectors on disk. v1 L2 / DotProduct segments load unchanged.
pub fn read_header(data: &[u8]) -> Result<HnswHeader, LuciError> {
    let (version, mut pos) = if data.len() >= 5 && data[0..4] == HNSW_FORMAT_MAGIC {
        let v = data[4];
        if v != HNSW_FORMAT_VERSION {
            return Err(LuciError::SegmentFormatUnknown(format!(
                "unknown HNSW format version: {v}",
            )));
        }
        (HnswFormatVersion::V2, 5_usize)
    } else {
        (HnswFormatVersion::V1, 0_usize)
    };

    let dims = read_u32(data, &mut pos)? as usize;
    let m = read_u32(data, &mut pos)? as usize;
    let metric = DistanceMetric::from_byte(read_u8(data, &mut pos)?);

    if version == HnswFormatVersion::V1 && metric == DistanceMetric::Cosine {
        return Err(LuciError::SegmentFormatMigrationRequired(
            "cosine HNSW segment was built with Luci ≤ 0.7.1 which stored \
             raw vectors. The v0.7.2+ kernel requires unit-length vectors \
             on disk. Re-index this collection."
                .into(),
        ));
    }

    let num_vectors = read_u32(data, &mut pos)? as usize;
    let ep = read_u32(data, &mut pos)?;
    let entry_point = if ep == u32::MAX { None } else { Some(ep) };
    let max_level = read_u32(data, &mut pos)? as usize;

    Ok(HnswHeader {
        version,
        dims,
        m,
        metric,
        num_vectors,
        entry_point,
        max_level,
        vectors_offset: pos,
    })
}

/// HNSW index parameters.
#[derive(Clone, Debug)]
pub struct HnswParams {
    pub dims: usize,
    pub m: usize,               // max connections per node per layer
    pub ef_construction: usize, // beam width during construction
    pub metric: DistanceMetric,
    /// How stored vectors are quantized for fast distance computation.
    /// Validated at [`HnswBuilder::new`]; only [`QuantizationType::None`]
    /// and [`QuantizationType::Int8`] are accepted today. The mapping
    /// parser rejects unimplemented values, so this is a defense-in-depth
    /// check rather than a routing fork.
    pub quantization: QuantizationType,
}

impl Default for HnswParams {
    fn default() -> Self {
        Self {
            dims: 128,
            m: 16,
            ef_construction: 100,
            metric: DistanceMetric::Cosine,
            quantization: QuantizationType::DEFAULT,
        }
    }
}

/// A scored neighbor candidate (lower distance = better).
#[derive(Clone, Copy)]
struct Candidate {
    id: u32,
    dist: f32,
}

impl PartialEq for Candidate {
    fn eq(&self, o: &Self) -> bool {
        self.dist == o.dist
    }
}
impl Eq for Candidate {}
impl PartialOrd for Candidate {
    fn partial_cmp(&self, o: &Self) -> Option<Ordering> {
        Some(self.cmp(o))
    }
}
// Min-heap: lower distance first
impl Ord for Candidate {
    fn cmp(&self, o: &Self) -> Ordering {
        o.dist.partial_cmp(&self.dist).unwrap_or(Ordering::Equal)
    }
}

/// Max-heap version for finding furthest candidates.
struct FurthestCandidate(Candidate);
impl PartialEq for FurthestCandidate {
    fn eq(&self, o: &Self) -> bool {
        self.0.dist == o.0.dist
    }
}
impl Eq for FurthestCandidate {}
impl PartialOrd for FurthestCandidate {
    fn partial_cmp(&self, o: &Self) -> Option<Ordering> {
        Some(self.cmp(o))
    }
}
impl Ord for FurthestCandidate {
    fn cmp(&self, o: &Self) -> Ordering {
        self.0
            .dist
            .partial_cmp(&o.0.dist)
            .unwrap_or(Ordering::Equal)
    }
}

/// Node in the HNSW graph.
#[derive(Clone)]
pub(crate) struct Node {
    /// Connections per layer: layer_index → vec of neighbor ids
    neighbors: Vec<Vec<u32>>,
    level: usize,
}

/// HNSW graph builder.
///
/// `nodes` holds each node's neighbor lists behind a per-node
/// `parking_lot::RwLock` (data-in-lock, qdrant's shape) so the parallel
/// connect phase ([[optimization-concurrent-hnsw-insert]]) can mutate
/// edges through `&self` while distance reads hit the separate,
/// connect-phase-immutable `vectors`. `entry` packs
/// `(entry_point, max_level)` into one atomic for a consistent
/// single-load descent. `Node` is reused as the in-lock payload — it is
/// structurally the design's `NodeNeighbors`, so reusing it keeps the
/// `build()` conversion a plain `into_inner()`. The manual `Clone`
/// (below) snapshots each lock; `#[derive(Clone)]` cannot, since neither
/// `RwLock` nor `AtomicU64` is `Clone`.
pub struct HnswBuilder {
    params: HnswParams,
    /// Connect-phase-immutable, so distance reads are race-free without
    /// locks; only neighbor lists and `entry` mutate concurrently.
    vectors: Vec<Vec<f32>>,
    /// Per-node neighbor lists, each behind its own lock.
    nodes: Vec<RwLock<Node>>,
    /// Packed `(entry_point:u32, max_level:u32)`; `entry_point ==
    /// ENTRY_SENTINEL` means un-bootstrapped.
    entry: AtomicU64,
    /// `ceil(len/64)` words; bit `ord` set (Release) ⇒ node `ord` is
    /// fully linked at **every** level and safe to traverse. Required for
    /// correctness, not just safety: it closes the inter-level race a
    /// per-node `RwLock` alone cannot (a traverser must never route
    /// through a node mid-multi-level-build). See
    /// [[optimization-concurrent-hnsw-insert]].
    ready: Vec<AtomicU64>,
    /// Pending-tail cursor: nodes `[0, connected_count)` are linked;
    /// `[connected_count, vectors.len())` are stored-but-unlinked and are
    /// linked by the next `connect_pending`.
    connected_count: usize,
    level_mult: f64,
    rng_state: u64,
}

impl Clone for HnswBuilder {
    /// Snapshots each node out of its lock and copies the packed entry.
    /// The sole load-bearing caller is `get_or_build_index`
    /// (`builder.clone().build()`), where the locks are uncontended.
    fn clone(&self) -> Self {
        Self {
            params: self.params.clone(),
            vectors: self.vectors.clone(),
            nodes: self
                .nodes
                .iter()
                .map(|n| RwLock::new(n.read().clone()))
                .collect(),
            entry: AtomicU64::new(self.entry.load(AtomicOrdering::Relaxed)),
            ready: self
                .ready
                .iter()
                .map(|w| AtomicU64::new(w.load(AtomicOrdering::Relaxed)))
                .collect(),
            connected_count: self.connected_count,
            level_mult: self.level_mult,
            rng_state: self.rng_state,
        }
    }
}

impl HnswBuilder {
    /// # Panics
    ///
    /// Panics if `params.quantization` is a recognized but unimplemented
    /// variant ([`QuantizationType::Int4`] or [`QuantizationType::Bbq`]).
    /// The mapping parser rejects these values at index creation time, so
    /// reaching this panic indicates an upstream wiring bug — not user
    /// input. The panic is preferred over silently substituting a
    /// different quantization; see [[code-must-not-lie]].
    pub fn new(params: HnswParams) -> Self {
        match params.quantization {
            QuantizationType::None | QuantizationType::Int8 => {}
            unsupported @ (QuantizationType::Int4 | QuantizationType::Bbq) => {
                panic!(
                    "HnswBuilder constructed with unimplemented quantization \
                     type {unsupported:?}; the mapping parser should have \
                     rejected this at index creation. This is an internal \
                     wiring bug, not user input."
                );
            }
        }
        let level_mult = 1.0 / (params.m as f64).ln();
        Self {
            params,
            vectors: Vec::new(),
            nodes: Vec::new(),
            entry: AtomicU64::new(pack_entry(ENTRY_SENTINEL, 0)),
            ready: Vec::new(),
            connected_count: 0,
            level_mult,
            rng_state: 42,
        }
    }

    /// Construct a builder pre-sized for a known final ordinal range,
    /// intended for the merge path ([[optimize-hnsw-merge-stitching]]).
    /// `capacity` is the final merged-segment doc count; the builder
    /// reserves `vectors[]` and `nodes[]` of that size with empty
    /// placeholders.
    ///
    /// Callers must populate every slot either by
    /// [`Self::seed_from_graph`] (bulk-copy from a source segment's
    /// graph) or [`Self::add_vector_at_ordinal`]. Slots left empty at
    /// [`Self::build`] time are still part of `vectors.len()` and will
    /// produce a degenerate graph — that's a caller bug to surface
    /// during testing.
    pub fn with_capacity_for_merge(params: HnswParams, capacity: usize) -> Self {
        match params.quantization {
            QuantizationType::None | QuantizationType::Int8 => {}
            unsupported @ (QuantizationType::Int4 | QuantizationType::Bbq) => {
                panic!(
                    "HnswBuilder::with_capacity_for_merge constructed with \
                     unimplemented quantization type {unsupported:?}; the \
                     mapping parser should have rejected this at index \
                     creation. This is an internal wiring bug, not user \
                     input."
                );
            }
        }
        let level_mult = 1.0 / (params.m as f64).ln();
        let mut vectors = Vec::with_capacity(capacity);
        let mut nodes = Vec::with_capacity(capacity);
        for _ in 0..capacity {
            vectors.push(Vec::new());
            nodes.push(RwLock::new(Node {
                neighbors: Vec::new(),
                level: 0,
            }));
        }
        let mut ready = Vec::with_capacity(capacity.div_ceil(64));
        for _ in 0..capacity.div_ceil(64) {
            ready.push(AtomicU64::new(0));
        }
        Self {
            params,
            vectors,
            nodes,
            entry: AtomicU64::new(pack_entry(ENTRY_SENTINEL, 0)),
            ready,
            connected_count: 0,
            level_mult,
            rng_state: 42,
        }
    }

    /// Seed the merge-time builder with a source segment's graph,
    /// remapping its ordinals into the merged segment's ordinal space.
    ///
    /// `seed`: the source segment's parsed graph (levels + neighbors).
    /// `hnsw_bytes`: the source segment's HNSW byte range, starting at
    ///   the magic prefix / header. The vector data is read out of
    ///   `hnsw_bytes[seed.vector_data_offset..]` four bytes per `f32`
    ///   (little-endian, matches the on-disk format).
    /// `ord_map`: function mapping source ordinals to merged ordinals.
    ///   Must return a valid merged ordinal for every source ordinal in
    ///   `0..seed.num_vectors`.
    ///
    /// After this returns, `self.entry_point` and `self.max_level`
    /// reflect the seed graph; non-seed slots stay empty until a
    /// caller fills them via [`Self::add_vector_at_ordinal`].
    ///
    /// **Cosine note:** the source segment's stored vectors are already
    /// unit-length under v0.7.2's invariant
    /// ([[optimize-cosine-norm-precompute]]), so they go in without
    /// re-normalization. v1 cosine segments cannot reach this path —
    /// [[read_header]] rejects them at open time.
    pub fn seed_from_graph<F>(&mut self, seed: &ParsedGraph, hnsw_bytes: &[u8], ord_map: F)
    where
        F: Fn(u32) -> u32,
    {
        let dims = self.params.dims;
        debug_assert_eq!(seed.params.dims, dims);
        debug_assert_eq!(seed.params.metric, self.params.metric);
        debug_assert_eq!(seed.params.m, self.params.m);

        for src_ord in 0..seed.num_vectors as u32 {
            let merged_ord = ord_map(src_ord);
            let start = seed.vector_data_offset + src_ord as usize * dims * 4;
            let mut v = Vec::with_capacity(dims);
            for d in 0..dims {
                let off = start + d * 4;
                v.push(f32::from_le_bytes(
                    hnsw_bytes[off..off + 4].try_into().unwrap(),
                ));
            }
            self.vectors[merged_ord as usize] = v;

            let src_node = &seed.nodes[src_ord as usize];
            let neighbors: Vec<Vec<u32>> = src_node
                .neighbors
                .iter()
                .map(|layer| layer.iter().copied().map(&ord_map).collect())
                .collect();
            self.nodes[merged_ord as usize] = RwLock::new(Node {
                neighbors,
                level: src_node.level,
            });
        }

        let ep = seed.entry_point.map(&ord_map).unwrap_or(ENTRY_SENTINEL);
        self.entry.store(
            pack_entry(ep, seed.max_level as u32),
            AtomicOrdering::Relaxed,
        );
    }

    fn next_rand(&mut self) -> f64 {
        // Simple xorshift64 for deterministic tests
        self.rng_state ^= self.rng_state << 13;
        self.rng_state ^= self.rng_state >> 7;
        self.rng_state ^= self.rng_state << 17;
        (self.rng_state as f64) / (u64::MAX as f64)
    }

    fn random_level(&mut self) -> usize {
        let r = self.next_rand().max(1e-10);
        (-r.ln() * self.level_mult).floor() as usize
    }

    /// Store a vector **without linking it into the graph** — the defer
    /// half of the [[declare-intent-defer-execution]] write model. The
    /// vector is assigned ordinal `self.vectors.len()`, normalized
    /// (cosine), given a drawn level and empty per-level neighbor lists,
    /// and its `ready` bit left unset. The node joins the pending tail
    /// `[connected_count, vectors.len())` and is linked by a later
    /// [`Self::connect_pending`]. Cheap and sequential.
    ///
    /// For [`DistanceMetric::Cosine`], the input is normalized to unit
    /// length in place so the kernel can run as a single-pass dot product
    /// (see [[optimize-cosine-norm-precompute]]). Zero / non-finite
    /// vectors are rejected with `LuciError::InvalidQuery` rather than
    /// silently embedded — see [[code-must-not-lie]].
    pub fn store_vector(&mut self, mut vector: Vec<f32>) -> Result<(), LuciError> {
        debug_assert_eq!(vector.len(), self.params.dims);
        if self.params.metric == DistanceMetric::Cosine {
            normalize_in_place(&mut vector)?;
        }
        let ord = self.vectors.len();
        let level = self.random_level();

        self.vectors.push(vector);
        let mut neighbors = Vec::with_capacity(level + 1);
        for _ in 0..=level {
            neighbors.push(Vec::new());
        }
        self.nodes.push(RwLock::new(Node { neighbors, level }));
        // Grow the ready bitset to cover the new ord (bit starts unset).
        if ord / 64 >= self.ready.len() {
            self.ready.push(AtomicU64::new(0));
        }
        Ok(())
    }

    /// Store **and** link a vector in one call (sequential). Convenience
    /// for tests and one-shot builders; equivalent to `store_vector`
    /// followed by an immediate sequential connect, advancing the
    /// connected-prefix cursor. The deferred write path uses
    /// `store_vector` + `connect_pending` instead.
    pub fn add_vector(&mut self, vector: Vec<f32>) -> Result<(), LuciError> {
        let id = self.vectors.len() as u32;
        self.store_vector(vector)?;
        let level = self.nodes[id as usize].read().level;
        self.connect_node(id, level);
        self.set_ready(id);
        self.connected_count = self.vectors.len();
        Ok(())
    }

    /// Add a vector at a specific pre-reserved ordinal. Used by the merge
    /// path ([[optimize-hnsw-merge-stitching]]) which knows the final
    /// merged-segment doc count up front and reserves slots via
    /// [`Self::with_capacity_for_merge`]. The slot at `ord` must already
    /// exist (otherwise the builder would have been a [`Self::new`]
    /// instance and the caller should use [`Self::add_vector`] instead).
    ///
    /// Cosine normalize / zero-vector rejection happens here identically
    /// to `add_vector`. The same insertion routine runs once the slot is
    /// populated — search_layer against the partially-filled graph,
    /// neighbor heuristic, back-links — so the resulting edges are
    /// indistinguishable from a from-scratch build modulo level-sampling
    /// RNG state.
    pub fn add_vector_at_ordinal(
        &mut self,
        ord: u32,
        mut vector: Vec<f32>,
    ) -> Result<(), LuciError> {
        debug_assert_eq!(vector.len(), self.params.dims);
        debug_assert!((ord as usize) < self.vectors.len());
        debug_assert!(
            self.vectors[ord as usize].is_empty(),
            "add_vector_at_ordinal called on already-filled ordinal {ord}",
        );
        if self.params.metric == DistanceMetric::Cosine {
            normalize_in_place(&mut vector)?;
        }
        let level = self.random_level();

        self.vectors[ord as usize] = vector;
        let mut neighbors = Vec::with_capacity(level + 1);
        for _ in 0..=level {
            neighbors.push(Vec::new());
        }
        self.nodes[ord as usize] = RwLock::new(Node { neighbors, level });

        self.connect_node(ord, level);
        Ok(())
    }

    /// Search-layer + neighbor-select + back-link insertion routine.
    /// Shared by [`Self::add_vector`] (append) and
    /// [`Self::add_vector_at_ordinal`] (pre-reserved slot). At call
    /// time the slot for `id` must already hold its vector and an
    /// empty-neighbors `Node { level }`. Updates `entry_point` /
    /// `max_level` if this is the first node or if `level` exceeds
    /// the current max.
    fn connect_node(&mut self, id: u32, level: usize) {
        let (ep0, max_l0) = unpack_entry(self.entry.load(AtomicOrdering::Relaxed));
        if ep0 == ENTRY_SENTINEL {
            // First node bootstraps the entry; it has no neighbors to link.
            self.entry
                .store(pack_entry(id, level as u32), AtomicOrdering::Relaxed);
            return;
        }
        // max_level cannot change during this call — only the promotion
        // below mutates it, so the start-of-call snapshot is exact (the
        // basis for `Fixed(1)` bit-identity with the old `self.max_level`).
        let max_level = max_l0 as usize;

        // Greedy search from top to the insertion level + 1
        let mut current = ep0;
        for lev in (level + 1..=max_level).rev() {
            current = self.greedy_closest(current, id, lev);
        }

        // Insert at each level from min(level, max_level) down to 0
        let insert_from = level.min(max_level);
        let mut ep_for_level = current;

        for lev in (0..=insert_from).rev() {
            let candidates = self.search_layer(id, ep_for_level, self.params.ef_construction, lev);
            let neighbors_to_connect =
                self.select_neighbors_heuristic(&candidates, self.m_max(lev));

            // Set this node's neighbors at this level (under its own lock).
            {
                let mut node = self.nodes[id as usize].write();
                if lev < node.neighbors.len() {
                    node.neighbors[lev] = neighbors_to_connect.iter().map(|c| c.id).collect();
                }
            }

            // Add back-links — one neighbor lock held at a time.
            for &n in &neighbors_to_connect {
                let mut node = self.nodes[n.id as usize].write();
                // Bounds guard (mirrors the original): n must exist at this
                // level, else per-level index is OOB.
                if lev < node.neighbors.len() {
                    node.neighbors[lev].push(id);
                    // Prune when over the layer-appropriate cap.
                    // M_max0 = 2*M at layer 0, M_max = M elsewhere.
                    if node.neighbors[lev].len() > self.m_max(lev) {
                        self.prune_connections_in(&mut node, n.id, lev);
                    }
                }
            }

            if !candidates.is_empty() {
                ep_for_level = candidates[0].id;
            }
        }

        // Promote to entry only on a strictly taller level (hnswlib's
        // "only promote when taller"; strict `>` preserves bit-identity).
        if level as u32 > max_l0 {
            self.entry
                .store(pack_entry(id, level as u32), AtomicOrdering::Relaxed);
        }
    }

    fn dist(&self, a: u32, b: u32) -> f32 {
        super::distance(
            &self.vectors[a as usize],
            &self.vectors[b as usize],
            self.params.metric,
        )
    }

    fn dist_to_vec(&self, a: u32, query: &[f32]) -> f32 {
        super::distance(&self.vectors[a as usize], query, self.params.metric)
    }

    fn greedy_closest(&self, start: u32, target: u32, level: usize) -> u32 {
        let mut current = start;
        let mut current_dist = self.dist(current, target);
        loop {
            let mut changed = false;
            // Read-lock `current` across the scan: `dist` reads `vectors`,
            // never `nodes`, so no second node lock is taken (deadlock-free)
            // and no neighbor copy is needed on this sequential path.
            let node = self.nodes[current as usize].read();
            if level < node.neighbors.len() {
                for &neighbor in &node.neighbors[level] {
                    let d = self.dist(neighbor, target);
                    if d < current_dist {
                        current = neighbor;
                        current_dist = d;
                        changed = true;
                    }
                }
            }
            drop(node);
            if !changed {
                break;
            }
        }
        current
    }

    fn search_layer(&self, query_id: u32, entry: u32, ef: usize, level: usize) -> Vec<Candidate> {
        self.search_layer_vec(&self.vectors[query_id as usize], entry, ef, level, None)
    }

    fn search_layer_vec(
        &self,
        query: &[f32],
        entry: u32,
        ef: usize,
        level: usize,
        filter: Option<&roaring::RoaringBitmap>,
    ) -> Vec<Candidate> {
        let mut visited = HashSet::new();
        let mut candidates = BinaryHeap::new(); // min-heap (closest first)
        let mut results = BinaryHeap::new(); // max-heap (furthest first)

        let d = self.dist_to_vec(entry, query);
        visited.insert(entry);
        candidates.push(Candidate { id: entry, dist: d });
        if filter.is_none() || filter.unwrap().contains(entry) {
            results.push(FurthestCandidate(Candidate { id: entry, dist: d }));
        }

        while let Some(c) = candidates.pop() {
            let furthest_dist = results.peek().map(|f| f.0.dist).unwrap_or(f32::MAX);
            if c.dist > furthest_dist && results.len() >= ef {
                break;
            }

            let node = self.nodes[c.id as usize].read();
            if level < node.neighbors.len() {
                for &neighbor in &node.neighbors[level] {
                    if visited.insert(neighbor) {
                        let d = self.dist_to_vec(neighbor, query);
                        let furthest_dist = results.peek().map(|f| f.0.dist).unwrap_or(f32::MAX);
                        if d < furthest_dist || results.len() < ef {
                            candidates.push(Candidate {
                                id: neighbor,
                                dist: d,
                            });
                            if filter.is_none() || filter.unwrap().contains(neighbor) {
                                results.push(FurthestCandidate(Candidate {
                                    id: neighbor,
                                    dist: d,
                                }));
                                if results.len() > ef {
                                    results.pop();
                                }
                            }
                        }
                    }
                }
            }
        }

        results.into_sorted_vec().into_iter().map(|f| f.0).collect()
    }

    /// Layer-appropriate neighbor cap. M_max0 = 2*M at layer 0,
    /// M_max = M at higher layers. Matches Lucene's
    /// `Lucene99HnswVectorsFormat`, hnswlib's `maxM0_ = 2 * M_`,
    /// faiss's `nb_neighbors(0) = M*2`, and pgvector's
    /// `HnswGetMaxConnections`. See [[fix-hnsw-neighbor-heuristic]].
    fn m_max(&self, level: usize) -> usize {
        if level == 0 {
            self.params.m * 2
        } else {
            self.params.m
        }
    }

    /// Algorithm 4 from Malkov & Yashunin 2018 — neighbor selection
    /// with diversity. Without the diversity check, the resulting
    /// graph clusters neighbors in one direction and the search has
    /// to traverse many extra nodes to reach other parts of the
    /// vector space. See [[fix-hnsw-neighbor-heuristic]].
    ///
    /// Matches hnswlib's `getNeighborsByHeuristic2` strictness: a
    /// candidate `c` is kept when `dist(c, s) >= c.dist_to_query`
    /// for every already-selected neighbor `s`. The `>=` (not `>`)
    /// handles exact ties — matters for diff-testing graphs against
    /// hnswlib even though cosine on real corpora rarely produces
    /// ties.
    fn select_neighbors_heuristic(&self, candidates: &[Candidate], m: usize) -> Vec<Candidate> {
        let mut working: Vec<Candidate> = candidates.to_vec();
        working.sort_by(|a, b| a.dist.partial_cmp(&b.dist).unwrap_or(Ordering::Equal));

        let mut selected: Vec<Candidate> = Vec::with_capacity(m);
        for c in working {
            if selected.len() == m {
                break;
            }
            // c.dist holds dist-to-query. Keep c only if it is closer
            // to the query than to any already-selected neighbor.
            let diverse = selected.iter().all(|s| self.dist(c.id, s.id) >= c.dist);
            if diverse {
                selected.push(c);
            }
        }
        selected
    }

    /// Re-prune `node`'s level-`level` neighbor list down to the layer
    /// cap, operating on an already-held write guard so no second node
    /// lock is taken (deadlock-free). Reads only this node's own list
    /// plus the connect-phase-immutable `vectors`. Replaces the old
    /// `prune_connections(&mut self, ...)` that re-indexed `self.nodes`,
    /// which would self-deadlock once `nodes` is `Vec<RwLock<Node>>`.
    fn prune_connections_in(&self, node: &mut Node, node_id: u32, level: usize) {
        let candidates: Vec<Candidate> = node.neighbors[level]
            .iter()
            .map(|&n| Candidate {
                id: n,
                dist: self.dist(node_id, n),
            })
            .collect();
        let kept = self.select_neighbors_heuristic(&candidates, self.m_max(level));
        node.neighbors[level] = kept.iter().map(|c| c.id).collect();
    }

    /// Mark node `ord` fully linked (Release) so traversers may route
    /// through it. Paired with [`Self::is_ready`]'s Acquire load.
    fn set_ready(&self, ord: u32) {
        let word = (ord / 64) as usize;
        let bit = ord % 64;
        self.ready[word].fetch_or(1u64 << bit, AtomicOrdering::Release);
    }

    /// True iff node `ord` is fully linked at every level (Acquire).
    fn is_ready(&self, ord: u32) -> bool {
        let word = (ord / 64) as usize;
        let bit = ord % 64;
        (self.ready[word].load(AtomicOrdering::Acquire) >> bit) & 1 == 1
    }

    /// Link every stored-but-unlinked node — the pending tail
    /// `[connected_count, vectors.len())` — into the graph, then advance
    /// the cursor. A no-op when the tail is empty. `Fixed(1)` runs the
    /// sequential, bit-identical path; `Ambient` / `Fixed(n>1)` run the
    /// `par_iter` over [`Self::connect_node_locked`]. See
    /// [[optimization-concurrent-hnsw-insert]] §Write model.
    pub fn connect_pending(&mut self, threads: BuildThreads) {
        let start = self.connected_count;
        let end = self.vectors.len();
        if start >= end {
            return;
        }
        match threads {
            BuildThreads::Fixed(1) => self.connect_tail_sequential(start, end),
            BuildThreads::Ambient => {
                // A single-threaded ambient pool — a 1-core machine, a build
                // pinned via `set_num_threads(1)` / `RAYON_NUM_THREADS=1`, or
                // a 1-thread scoped pool — has no concurrency to exploit, so
                // run the deterministic, byte-identical sequential path
                // instead of the lock/atomic machinery. This makes 1-thread
                // builds reproducible (and keeps the recall-regression guard
                // deterministic) at zero cost to the multi-core default — the
                // `Fixed(1)` bit-identical contract from
                // [[optimization-concurrent-hnsw-insert]] §Determinism.
                if current_num_threads() <= 1 {
                    self.connect_tail_sequential(start, end);
                } else {
                    self.connect_tail_parallel(start, end);
                }
            }
            BuildThreads::Fixed(n) => {
                let pool = ThreadPoolBuilder::new()
                    .num_threads(n)
                    .build()
                    .expect("failed to build HNSW connect thread pool");
                pool.install(|| self.connect_tail_parallel(start, end));
            }
        }
        self.connected_count = end;
    }

    /// Sequential connect of the tail `[start, end)` — the deterministic
    /// path that is byte-identical to building via [`Self::add_vector`]
    /// (proven by `connect_pending_fixed1_matches_add_vector_bytes`). Used
    /// for `Fixed(1)` and for a single-threaded `Ambient` pool.
    fn connect_tail_sequential(&mut self, start: usize, end: usize) {
        for ord in start..end {
            let level = self.nodes[ord].read().level;
            self.connect_node(ord as u32, level);
            self.set_ready(ord as u32);
        }
    }

    /// `par_iter` the tail through [`Self::connect_node_locked`]. Borrows
    /// `&self`; neighbor state mutates through the per-node locks and the
    /// packed `entry` atomic, so no `&mut` crosses threads. Runs on
    /// whatever rayon pool the caller installed.
    fn connect_tail_parallel(&self, start: usize, end: usize) {
        (start..end).into_par_iter().for_each(|ord| {
            let level = self.nodes[ord].read().level;
            self.connect_node_locked(ord as u32, level);
        });
    }

    /// Parallel insert of an already-stored node `ord` at its drawn
    /// `level`. Mutates only interior locks and the `entry` / `ready`
    /// atomics through `&self`, so many run concurrently under
    /// `connect_tail_parallel`. Mirrors the sequential
    /// [`Self::connect_node`] but: (a) reads gate on `ready` and
    /// snapshot-under-lock, (b) entry bootstrap / promotion is CAS,
    /// (c) `ready` is published last. See
    /// [[optimization-concurrent-hnsw-insert]] §connect_node_locked. The
    /// entry CAS and the `ready` gate are modeled exhaustively in
    /// `tests/loom_hnsw.rs` — keep the two in sync.
    fn connect_node_locked(&self, ord: u32, level: usize) {
        // `max_level_u` is a point-in-time snapshot of the entry word; a
        // concurrent promotion can make it stale, so the descent may start
        // one layer low. Intentional (hnswlib's `maxlevelcopy`) — the node
        // still links at every one of its own layers, so don't "fix" this
        // into a lock.
        let (mut ep, mut max_level_u) = unpack_entry(self.entry.load(AtomicOrdering::Acquire));
        if ep == ENTRY_SENTINEL {
            // First node bootstraps the entry; it has no neighbors. Mark
            // ready BEFORE the CAS so any thread that reads `ord` as the
            // entry and gates on `ready[ord]` finds it set. A thread that
            // *loses* the bootstrap race has now published `ready[ord]` with
            // zero edges and falls through to link below; that is safe — no
            // node points to `ord` until it builds its own back-links, its
            // slot and vector were filled by `store_vector`, and every read
            // is lock-protected, so a traverser reaching `ord` in that window
            // sees an empty (dead-end) list, never UB. Recall wobble is
            // construction-time only.
            self.set_ready(ord);
            match self.entry.compare_exchange(
                pack_entry(ENTRY_SENTINEL, 0),
                pack_entry(ord, level as u32),
                AtomicOrdering::AcqRel,
                AtomicOrdering::Acquire,
            ) {
                Ok(_) => return,
                Err(actual) => {
                    // Lost the bootstrap race; descend from the winner,
                    // whose (ep, max_level) pair is read as one word.
                    let (a_ep, a_max) = unpack_entry(actual);
                    ep = a_ep;
                    max_level_u = a_max;
                }
            }
        }
        let max_level = max_level_u as usize;

        // Greedy descent from max_level down to level+1 (reads only).
        for lev in (level + 1..=max_level).rev() {
            ep = self.greedy_closest_ready(ep, ord, lev);
        }

        let insert_from = level.min(max_level);
        let mut ep_for_level = ep;
        for lev in (0..=insert_from).rev() {
            let candidates =
                self.search_layer_ready(ord, ep_for_level, self.params.ef_construction, lev);
            let selected = self.select_neighbors_heuristic(&candidates, self.m_max(lev));

            // Write own list under ord's lock.
            {
                let mut node = self.nodes[ord as usize].write();
                if lev < node.neighbors.len() {
                    node.neighbors[lev] = selected.iter().map(|c| c.id).collect();
                }
            }
            // Back-links — one neighbor lock held at a time (no nested
            // locks ⇒ no lock-ordering deadlock).
            for &n in &selected {
                let mut node = self.nodes[n.id as usize].write();
                if lev < node.neighbors.len() {
                    node.neighbors[lev].push(ord);
                    if node.neighbors[lev].len() > self.m_max(lev) {
                        self.prune_connections_in(&mut node, n.id, lev);
                    }
                }
            }
            if !candidates.is_empty() {
                ep_for_level = candidates[0].id;
            }
        }

        // Publish: node fully linked at every level, now traversable.
        self.set_ready(ord);

        // Promote to entry only on a strictly taller level (CAS-loop;
        // strict `>` matches the sequential path's promotion rule).
        loop {
            let cur = self.entry.load(AtomicOrdering::Acquire);
            let (_, cur_max) = unpack_entry(cur);
            if (level as u32) <= cur_max {
                break;
            }
            if self
                .entry
                .compare_exchange(
                    cur,
                    pack_entry(ord, level as u32),
                    AtomicOrdering::AcqRel,
                    AtomicOrdering::Acquire,
                )
                .is_ok()
            {
                break;
            }
        }
    }

    /// `greedy_closest` for the parallel path: skips candidates whose
    /// `ready` bit is unset (a half-linked node is never traversed) and
    /// snapshots each node's neighbor ids under a short read lock before
    /// computing `dist` outside the lock (qdrant pattern — never hold the
    /// lock across `dist`, never block a back-linker).
    fn greedy_closest_ready(&self, start: u32, target: u32, level: usize) -> u32 {
        let mut current = start;
        let mut current_dist = self.dist(current, target);
        loop {
            let neighbors: Vec<u32> = {
                let node = self.nodes[current as usize].read();
                if level < node.neighbors.len() {
                    node.neighbors[level].clone()
                } else {
                    Vec::new()
                }
            };
            let mut changed = false;
            for neighbor in neighbors {
                if !self.is_ready(neighbor) {
                    continue;
                }
                let d = self.dist(neighbor, target);
                if d < current_dist {
                    current = neighbor;
                    current_dist = d;
                    changed = true;
                }
            }
            if !changed {
                break;
            }
        }
        current
    }

    /// `search_layer` for the parallel path: ready-gated, snapshot-under-
    /// lock (see [`Self::greedy_closest_ready`]). No `filter` arg — the
    /// builder never filters at construction time.
    fn search_layer_ready(
        &self,
        query_id: u32,
        entry: u32,
        ef: usize,
        level: usize,
    ) -> Vec<Candidate> {
        let query = self.vectors[query_id as usize].as_slice();
        let mut visited = HashSet::new();
        let mut candidates = BinaryHeap::new();
        let mut results = BinaryHeap::new();

        let d = self.dist_to_vec(entry, query);
        visited.insert(entry);
        candidates.push(Candidate { id: entry, dist: d });
        results.push(FurthestCandidate(Candidate { id: entry, dist: d }));

        while let Some(c) = candidates.pop() {
            let furthest_dist = results.peek().map(|f| f.0.dist).unwrap_or(f32::MAX);
            if c.dist > furthest_dist && results.len() >= ef {
                break;
            }
            let neighbors: Vec<u32> = {
                let node = self.nodes[c.id as usize].read();
                if level < node.neighbors.len() {
                    node.neighbors[level].clone()
                } else {
                    Vec::new()
                }
            };
            for neighbor in neighbors {
                if !self.is_ready(neighbor) {
                    continue;
                }
                if visited.insert(neighbor) {
                    let d = self.dist_to_vec(neighbor, query);
                    let furthest_dist = results.peek().map(|f| f.0.dist).unwrap_or(f32::MAX);
                    if d < furthest_dist || results.len() < ef {
                        candidates.push(Candidate {
                            id: neighbor,
                            dist: d,
                        });
                        results.push(FurthestCandidate(Candidate {
                            id: neighbor,
                            dist: d,
                        }));
                        if results.len() > ef {
                            results.pop();
                        }
                    }
                }
            }
        }
        results.into_sorted_vec().into_iter().map(|f| f.0).collect()
    }

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

    /// True iff there are stored-but-unlinked nodes (a non-empty pending
    /// tail). `get_or_build_index` asserts this is false before building
    /// the read-side index — a true value at persist means a
    /// `connect_pending` trigger was missed (the disconnected-graph bug).
    /// See [[optimization-concurrent-hnsw-insert]] §Conversion.
    pub fn has_pending_tail(&self) -> bool {
        self.connected_count < self.vectors.len()
    }

    /// Configured params (dims, M, metric, quantization). Exposed for
    /// the merge path so it can size a new builder identically without
    /// re-deriving from the schema.
    pub fn params(&self) -> HnswParams {
        self.params.clone()
    }

    /// Reconstruct a builder from a previously-built [`HnswIndex`].
    ///
    /// Used by the global-HNSW reopen path: persisted graphs come back
    /// as an [`HnswIndex`], but the writer needs an `HnswBuilder` to
    /// accept further inserts. `level_mult` and `rng_state` are
    /// re-derived; HNSW is approximate, so resuming with a fresh RNG
    /// state changes which levels new inserts land on but not
    /// correctness.
    pub fn from_index(index: HnswIndex) -> Self {
        let level_mult = 1.0 / (index.params.m as f64).ln();
        let ep = index.entry_point.unwrap_or(ENTRY_SENTINEL);
        let entry = AtomicU64::new(pack_entry(ep, index.max_level as u32));
        let n = index.vectors.len();
        // A reloaded graph is fully connected: every node is `ready` and
        // the pending tail is empty. Without all-ready, a subsequent
        // `connect_pending` would skip the loaded nodes as un-traversable
        // and link the new tail into an effectively empty graph. Trailing
        // bits past `n` are set but never indexed (harmless).
        let mut ready = Vec::with_capacity(n.div_ceil(64));
        for _ in 0..n.div_ceil(64) {
            ready.push(AtomicU64::new(u64::MAX));
        }
        Self {
            params: index.params,
            vectors: index.vectors,
            nodes: index.nodes.into_iter().map(RwLock::new).collect(),
            entry,
            ready,
            connected_count: n,
            level_mult,
            rng_state: 42,
        }
    }

    /// Build the final index. Quantization is applied per the
    /// `quantization` field on [`HnswParams`].
    pub fn build(self) -> HnswIndex {
        let quantized = match self.params.quantization {
            QuantizationType::None => None,
            QuantizationType::Int8 if !self.vectors.is_empty() => Some(
                super::quantize::QuantizedVectors::quantize(&self.vectors, self.params.metric),
            ),
            QuantizationType::Int8 => None,
            // `new()` rejects these variants. Reaching this arm means
            // someone bypassed the constructor invariant.
            unsupported @ (QuantizationType::Int4 | QuantizationType::Bbq) => {
                panic!(
                    "HnswBuilder::build reached with unimplemented \
                     quantization {unsupported:?}; the constructor was \
                     supposed to reject this."
                );
            }
        };

        let (ep, max_level) = unpack_entry(self.entry.load(AtomicOrdering::Relaxed));
        let entry_point = if ep == ENTRY_SENTINEL { None } else { Some(ep) };

        HnswIndex {
            params: self.params,
            vectors: self.vectors,
            nodes: self
                .nodes
                .into_iter()
                .map(|lock| lock.into_inner())
                .collect(),
            entry_point,
            max_level: max_level as usize,
            quantized,
        }
    }
}

/// Immutable HNSW index for search.
pub struct HnswIndex {
    params: HnswParams,
    vectors: Vec<Vec<f32>>,
    nodes: Vec<Node>,
    entry_point: Option<u32>,
    max_level: usize,
    /// Int8 quantized vectors for fast approximate distance.
    quantized: Option<super::quantize::QuantizedVectors>,
}

impl HnswIndex {
    /// Search for k nearest neighbors.
    pub fn search(&self, query: &[f32], k: usize, ef: usize) -> Result<Vec<(u32, f32)>, LuciError> {
        self.search_filtered(query, k, ef, None)
    }

    /// Search with optional pre-filter bitmap.
    ///
    /// For [`DistanceMetric::Cosine`] the query is normalized once at entry
    /// so every kernel call sees unit-length inputs. Zero / non-finite
    /// query vectors return `LuciError::InvalidQuery` rather than producing
    /// garbage scores.
    pub fn search_filtered(
        &self,
        query: &[f32],
        k: usize,
        ef: usize,
        filter: Option<&roaring::RoaringBitmap>,
    ) -> Result<Vec<(u32, f32)>, LuciError> {
        if self.entry_point.is_none() {
            return Ok(Vec::new());
        }

        // Normalize query once for cosine; the inner kernel runs as a
        // pure dot product under the unit-length invariant.
        let query_owned: Vec<f32> = if self.params.metric == DistanceMetric::Cosine {
            let mut q = query.to_vec();
            normalize_in_place(&mut q)?;
            q
        } else {
            query.to_vec()
        };
        let query = &query_owned[..];

        // Brute-force fallback for very selective filters
        if let Some(bm) = filter {
            if (bm.len() as f64) < (self.vectors.len() as f64 * 0.01) {
                return Ok(self.brute_force_search(query, k, bm));
            }
        }

        let ep = self.entry_point.unwrap();
        let ef_actual = ef.max(k);

        // Greedy descend from top to layer 1
        let mut current = ep;
        for lev in (1..=self.max_level).rev() {
            current = self.greedy_closest_vec(current, query, lev);
        }

        // Search layer 0 with beam (uses quantized distance if available)
        let candidates = self.search_layer_0(query, current, ef_actual, filter);

        // Rerank with exact float32 distances for final ordering
        let mut results: Vec<(u32, f32)> = if self.quantized.is_some() {
            candidates
                .into_iter()
                .map(|c| (c.id, self.exact_dist(c.id, query)))
                .collect()
        } else {
            candidates.into_iter().map(|c| (c.id, c.dist)).collect()
        };
        results.sort_by(|a, b| a.1.partial_cmp(&b.1).unwrap_or(Ordering::Equal));
        results.truncate(k);
        Ok(results)
    }

    fn greedy_closest_vec(&self, start: u32, query: &[f32], level: usize) -> u32 {
        let mut current = start;
        let mut current_dist = self.quantized_dist(current, query);
        loop {
            let mut changed = false;
            if level < self.nodes[current as usize].neighbors.len() {
                for &neighbor in &self.nodes[current as usize].neighbors[level] {
                    let d = self.quantized_dist(neighbor, query);
                    if d < current_dist {
                        current = neighbor;
                        current_dist = d;
                        changed = true;
                    }
                }
            }
            if !changed {
                break;
            }
        }
        current
    }

    fn search_layer_0(
        &self,
        query: &[f32],
        entry: u32,
        ef: usize,
        filter: Option<&roaring::RoaringBitmap>,
    ) -> Vec<Candidate> {
        let mut visited = HashSet::new();
        let mut candidates = BinaryHeap::new();
        let mut results = BinaryHeap::new();

        let d = self.quantized_dist(entry, query);
        visited.insert(entry);
        candidates.push(Candidate { id: entry, dist: d });
        if filter.is_none() || filter.unwrap().contains(entry) {
            results.push(FurthestCandidate(Candidate { id: entry, dist: d }));
        }

        while let Some(c) = candidates.pop() {
            let furthest_dist = results.peek().map(|f| f.0.dist).unwrap_or(f32::MAX);
            if c.dist > furthest_dist && results.len() >= ef {
                break;
            }

            if !self.nodes[c.id as usize].neighbors.is_empty() {
                for &neighbor in &self.nodes[c.id as usize].neighbors[0] {
                    if visited.insert(neighbor) {
                        let d = self.quantized_dist(neighbor, query);
                        let furthest_dist = results.peek().map(|f| f.0.dist).unwrap_or(f32::MAX);
                        if d < furthest_dist || results.len() < ef {
                            candidates.push(Candidate {
                                id: neighbor,
                                dist: d,
                            });
                            if filter.is_none() || filter.unwrap().contains(neighbor) {
                                results.push(FurthestCandidate(Candidate {
                                    id: neighbor,
                                    dist: d,
                                }));
                                if results.len() > ef {
                                    results.pop();
                                }
                            }
                        }
                    }
                }
            }
        }

        let mut result: Vec<Candidate> = results.into_iter().map(|f| f.0).collect();
        result.sort_by(|a, b| a.dist.partial_cmp(&b.dist).unwrap_or(Ordering::Equal));
        result
    }

    fn brute_force_search(
        &self,
        query: &[f32],
        k: usize,
        filter: &roaring::RoaringBitmap,
    ) -> Vec<(u32, f32)> {
        let mut results: Vec<(u32, f32)> = filter
            .iter()
            .filter(|&id| (id as usize) < self.vectors.len())
            .map(|id| (id, self.quantized_dist(id, query)))
            .collect();
        results.sort_by(|a, b| a.1.partial_cmp(&b.1).unwrap_or(Ordering::Equal));
        results.truncate(k);
        results
    }

    /// Distance between stored vector `a` and query. Uses quantized
    /// distance when available, falling back to float32.
    #[inline]
    fn quantized_dist(&self, a: u32, query: &[f32]) -> f32 {
        if let Some(ref qv) = self.quantized {
            qv.asymmetric_distance(a as usize, query)
        } else {
            super::distance(&self.vectors[a as usize], query, self.params.metric)
        }
    }

    /// Exact float32 distance (for reranking after quantized search).
    fn exact_dist(&self, a: u32, query: &[f32]) -> f32 {
        super::distance(&self.vectors[a as usize], query, self.params.metric)
    }

    pub fn len(&self) -> usize {
        self.vectors.len()
    }
    pub fn is_empty(&self) -> bool {
        self.vectors.is_empty()
    }
    pub fn dims(&self) -> usize {
        self.params.dims
    }

    /// Get a vector by ID.
    pub fn vector(&self, id: u32) -> &[f32] {
        &self.vectors[id as usize]
    }

    /// Serialize to bytes.
    ///
    /// Writes the v2 format unconditionally — magic prefix
    /// [`HNSW_FORMAT_MAGIC`] followed by [`HNSW_FORMAT_VERSION`], then
    /// the header fields readers expect. v1 segments (without the
    /// magic prefix) are read-only legacy; the builder never writes
    /// them. See [[optimize-cosine-norm-precompute]].
    pub fn to_bytes(&self) -> Vec<u8> {
        let mut buf = Vec::new();
        // v2 format prefix
        buf.extend_from_slice(&HNSW_FORMAT_MAGIC);
        buf.push(HNSW_FORMAT_VERSION);
        // Header
        buf.extend_from_slice(&(self.params.dims as u32).to_le_bytes());
        buf.extend_from_slice(&(self.params.m as u32).to_le_bytes());
        buf.push(self.params.metric as u8);
        buf.extend_from_slice(&(self.vectors.len() as u32).to_le_bytes());
        buf.extend_from_slice(&self.entry_point.unwrap_or(u32::MAX).to_le_bytes());
        buf.extend_from_slice(&(self.max_level as u32).to_le_bytes());

        // Vectors: flat f32 array
        for v in &self.vectors {
            for &f in v {
                buf.extend_from_slice(&f.to_le_bytes());
            }
        }

        // Nodes: level + per-level neighbor lists
        for node in &self.nodes {
            buf.extend_from_slice(&(node.level as u32).to_le_bytes());
            buf.extend_from_slice(&(node.neighbors.len() as u32).to_le_bytes());
            for layer in &node.neighbors {
                buf.extend_from_slice(&(layer.len() as u32).to_le_bytes());
                for &n in layer {
                    buf.extend_from_slice(&n.to_le_bytes());
                }
            }
        }

        // Quantized vectors (if present)
        if let Some(ref qv) = self.quantized {
            buf.push(1u8); // has quantized
            let qbytes = qv.to_bytes();
            buf.extend_from_slice(&(qbytes.len() as u32).to_le_bytes());
            buf.extend_from_slice(&qbytes);
        } else {
            buf.push(0u8); // no quantized
        }

        buf
    }

    /// Deserialize from bytes.
    ///
    /// Returns `Err(SegmentFormatMigrationRequired)` for v0.7.1 cosine
    /// segments (which stored un-normalized vectors and can't be read
    /// under the v2 unit-length invariant) and
    /// `Err(SegmentFormatUnknown)` for future versions an older binary
    /// can't parse. See [[optimize-cosine-norm-precompute]] §Migration.
    pub fn from_bytes(data: &[u8]) -> Result<Self, LuciError> {
        let header = read_header(data)?;
        let dims = header.dims;
        let m = header.m;
        let metric = header.metric;
        let num_vectors = header.num_vectors;
        let entry_point = header.entry_point;
        let max_level = header.max_level;
        let mut pos = header.vectors_offset;

        // Vectors. Guard `num_vectors` before reserving: each index costs at
        // least its vector bytes (`dims × 4`) plus its node entry (≥8 B), so a
        // corrupt count can't drive a giant speculative allocation.
        checked_len(
            num_vectors,
            dims.saturating_mul(4).saturating_add(8),
            data,
            pos,
        )?;
        let mut vectors = Vec::with_capacity(num_vectors);
        for _ in 0..num_vectors {
            let mut v = Vec::with_capacity(dims);
            for _ in 0..dims {
                v.push(read_f32(data, &mut pos)?);
            }
            vectors.push(v);
        }

        // Nodes
        let mut nodes = Vec::with_capacity(num_vectors);
        for _ in 0..num_vectors {
            let level = read_u32(data, &mut pos)? as usize;
            let num_layers = read_u32(data, &mut pos)? as usize;
            let mut neighbors = Vec::with_capacity(checked_len(num_layers, 4, data, pos)?);
            for _ in 0..num_layers {
                let num_neighbors = read_u32(data, &mut pos)? as usize;
                let mut layer = Vec::with_capacity(checked_len(num_neighbors, 4, data, pos)?);
                for _ in 0..num_neighbors {
                    layer.push(read_u32(data, &mut pos)?);
                }
                neighbors.push(layer);
            }
            nodes.push(Node { neighbors, level });
        }

        // Quantized vectors. The on-disk format records a flag byte at
        // `pos` indicating whether quantized data follows:
        //   - `1` → a quantized blob follows; deserialise it.
        //   - `0` → no quantized data; honor it (this is what the user
        //          asked for when the mapping said `"quantization": "none"`).
        //   - missing (`pos == data.len()`) → v1 legacy file that predates
        //          the flag. v1 never had quantization, so treat as None.
        //
        // The previous else branch synthesised int8 quantized vectors when
        // the file flag was 0 *or* missing, overriding the user's mapping
        // and routing query-time beam search through int8 scalar instead
        // of FP32. That was a [[code-must-not-lie]] violation; see
        // [[hnsw-query-path-allocation-overhead]] for the diagnostic.
        let quantized = if pos < data.len() && data[pos] == 1 {
            pos += 1;
            let qlen = read_u32(data, &mut pos)? as usize;
            let qbytes = take_bytes(data, &mut pos, qlen)?;
            Some(super::quantize::QuantizedVectors::from_bytes(qbytes))
        } else {
            None
        };

        let quantization = if quantized.is_some() {
            QuantizationType::Int8
        } else {
            QuantizationType::None
        };

        // Validate the graph invariants the search path assumes, so a
        // structurally-parseable but corrupt graph errors here instead of
        // indexing `self.vectors[id]` / `self.nodes[id]` out of bounds during
        // a query. This is the failure the visited-pool design flagged as its
        // concrete panic trigger — see [[optimization-hnsw-visited-bitset]].
        if let Some(ep) = entry_point {
            if ep as usize >= num_vectors {
                return Err(LuciError::IndexCorrupted(format!(
                    "HNSW entry_point {ep} out of range (num_vectors {num_vectors})"
                )));
            }
        }
        for (node_id, node) in nodes.iter().enumerate() {
            for layer in &node.neighbors {
                for &nid in layer {
                    if nid as usize >= num_vectors {
                        return Err(LuciError::IndexCorrupted(format!(
                            "HNSW node {node_id} neighbour id {nid} out of range \
                             (num_vectors {num_vectors})"
                        )));
                    }
                }
            }
        }

        Ok(Self {
            params: HnswParams {
                dims,
                m,
                ef_construction: 100,
                metric,
                quantization,
            },
            vectors,
            nodes,
            entry_point,
            max_level,
            quantized,
        })
    }
}

// --- Zero-copy HNSW searcher for scalable segment access ---

/// Parsed graph structure (nodes + neighbors). This is the only part
/// that needs heap allocation. Cached in SegmentReader.
pub struct ParsedGraph {
    pub params: HnswParams,
    pub num_vectors: usize,
    pub entry_point: Option<u32>,
    pub max_level: usize,
    pub(crate) nodes: Vec<Node>,
    /// Byte offset where vector data starts in the segment slice.
    pub vector_data_offset: usize,
    /// Byte offset where quantized data starts (if present).
    pub quantized_offset: Option<usize>,
    /// Length of quantized data block.
    pub quantized_len: usize,
}

impl ParsedGraph {
    /// Parse the graph structure from HNSW bytes. Records byte offsets
    /// for vector and quantized data but does NOT copy them.
    pub fn parse(data: &[u8]) -> Result<Self, LuciError> {
        let header = read_header(data)?;
        let dims = header.dims;
        let num_vectors = header.num_vectors;
        let mut pos = header.vectors_offset;

        // Record vector data offset, skip past vector bytes
        let vector_data_offset = pos;
        pos += num_vectors * dims * 4; // skip float32 vectors

        // Parse nodes (graph structure — this we DO parse into memory)
        let mut nodes = Vec::with_capacity(num_vectors);
        for _ in 0..num_vectors {
            let level = u32::from_le_bytes(data[pos..pos + 4].try_into().unwrap()) as usize;
            pos += 4;
            let num_layers = u32::from_le_bytes(data[pos..pos + 4].try_into().unwrap()) as usize;
            pos += 4;
            let mut neighbors = Vec::with_capacity(num_layers);
            for _ in 0..num_layers {
                let num_neighbors =
                    u32::from_le_bytes(data[pos..pos + 4].try_into().unwrap()) as usize;
                pos += 4;
                let mut layer = Vec::with_capacity(num_neighbors);
                for _ in 0..num_neighbors {
                    layer.push(u32::from_le_bytes(data[pos..pos + 4].try_into().unwrap()));
                    pos += 4;
                }
                neighbors.push(layer);
            }
            nodes.push(Node { neighbors, level });
        }

        // Record quantized data offset
        let (quantized_offset, quantized_len) = if pos < data.len() && data[pos] == 1 {
            pos += 1;
            let qlen = u32::from_le_bytes(data[pos..pos + 4].try_into().unwrap()) as usize;
            pos += 4;
            (Some(pos), qlen)
        } else {
            (None, 0)
        };

        // Derive quantization from whether the segment carries a quantized
        // blob — see HnswIndex::from_bytes for the rationale.
        let quantization = if quantized_offset.is_some() {
            QuantizationType::Int8
        } else {
            QuantizationType::None
        };

        Ok(Self {
            params: HnswParams {
                dims,
                m: header.m,
                ef_construction: 100,
                metric: header.metric,
                quantization,
            },
            num_vectors,
            entry_point: header.entry_point,
            max_level: header.max_level,
            nodes,
            vector_data_offset,
            quantized_offset,
            quantized_len,
        })
    }
}

/// Zero-copy HNSW searcher that borrows segment data for vectors and
/// quantized data. Only the graph structure (via `ParsedGraph`) is
/// heap-allocated.
pub struct HnswSearcher<'a> {
    graph: &'a ParsedGraph,
    /// Raw HNSW bytes (vectors + nodes + quantized, borrowed from segment).
    data: &'a [u8],
    /// Parsed quantized calibration (mins, scales, norms) — small, cached.
    quantized_cal: Option<QuantizedCalibration<'a>>,
}

/// Quantized calibration data (borrowed from segment bytes).
struct QuantizedCalibration<'a> {
    dims: usize,
    _num_vectors: usize,
    _metric: DistanceMetric,
    mins: &'a [u8],   // dims × 4 bytes (f32 le)
    scales: &'a [u8], // dims × 4 bytes (f32 le)
    norms: &'a [u8],  // num_vectors × 4 bytes (f32 le)
    data: &'a [u8],   // num_vectors × dims bytes (u8 quantized values)
}

impl<'a> HnswSearcher<'a> {
    /// Create a searcher borrowing segment data and a parsed graph.
    pub fn new(data: &'a [u8], graph: &'a ParsedGraph) -> Self {
        let quantized_cal = graph.quantized_offset.map(|qoff| {
            let qdata = &data[qoff..qoff + graph.quantized_len];
            // Parse quantized calibration structure (borrow slices)
            let dims = graph.params.dims;
            let num_vectors = graph.num_vectors;
            let mut p = 0;
            // Skip dims, num_vectors, metric (already in graph.params)
            p += 4 + 4 + 1; // dims u32 + num_vectors u32 + metric u8
            let mins = &qdata[p..p + dims * 4];
            p += dims * 4;
            let scales = &qdata[p..p + dims * 4];
            p += dims * 4;
            let norms = &qdata[p..p + num_vectors * 4];
            p += num_vectors * 4;
            let vdata = &qdata[p..p + num_vectors * dims];
            QuantizedCalibration {
                dims,
                _num_vectors: num_vectors,
                _metric: graph.params.metric,
                mins,
                scales,
                norms,
                data: vdata,
            }
        });

        Self {
            graph,
            data,
            quantized_cal,
        }
    }

    /// Search for k nearest neighbors.
    pub fn search(&self, query: &[f32], k: usize, ef: usize) -> Result<Vec<(u32, f32)>, LuciError> {
        self.search_filtered(query, k, ef, None)
    }

    pub fn search_filtered(
        &self,
        query: &[f32],
        k: usize,
        ef: usize,
        filter: Option<&roaring::RoaringBitmap>,
    ) -> Result<Vec<(u32, f32)>, LuciError> {
        if self.graph.entry_point.is_none() {
            return Ok(Vec::new());
        }

        // Normalize the query once at entry for cosine. v2 segments store
        // unit-length vectors, so the kernel runs as a pure dot product.
        let query_owned: Vec<f32> = if self.graph.params.metric == DistanceMetric::Cosine {
            let mut q = query.to_vec();
            normalize_in_place(&mut q)?;
            q
        } else {
            query.to_vec()
        };
        let query = &query_owned[..];

        if let Some(bm) = filter {
            if (bm.len() as f64) < (self.graph.num_vectors as f64 * 0.01) {
                return Ok(self.brute_force_search(query, k, bm));
            }
        }

        let ep = self.graph.entry_point.unwrap();
        let ef_actual = ef.max(k);

        let mut current = ep;
        for lev in (1..=self.graph.max_level).rev() {
            current = self.greedy_closest(current, query, lev);
        }

        let candidates = self.search_layer_0(query, current, ef_actual, filter);

        // Rerank with exact float32 distance
        let mut results: Vec<(u32, f32)> = candidates
            .into_iter()
            .map(|c| (c.id, self.exact_dist(c.id, query)))
            .collect();
        results.sort_by(|a, b| a.1.partial_cmp(&b.1).unwrap_or(Ordering::Equal));
        results.truncate(k);
        Ok(results)
    }

    /// Read a float32 value from the vector data region.
    #[inline]
    fn read_f32(&self, byte_offset: usize) -> f32 {
        f32::from_le_bytes(self.data[byte_offset..byte_offset + 4].try_into().unwrap())
    }

    /// Exact float32 distance (for reranking).
    fn exact_dist(&self, idx: u32, query: &[f32]) -> f32 {
        let dims = self.graph.params.dims;
        let base = self.graph.vector_data_offset + (idx as usize) * dims * 4;
        let mut vec = Vec::with_capacity(dims);
        for d in 0..dims {
            vec.push(self.read_f32(base + d * 4));
        }
        super::distance(&vec, query, self.graph.params.metric)
    }

    /// Fast approximate distance (quantized if available, else float32).
    #[inline]
    fn approx_dist(&self, idx: u32, query: &[f32]) -> f32 {
        if let Some(ref cal) = self.quantized_cal {
            self.quantized_cosine_dist(cal, idx as usize, query)
        } else {
            self.exact_dist(idx, query)
        }
    }

    fn quantized_cosine_dist(&self, cal: &QuantizedCalibration, idx: usize, query: &[f32]) -> f32 {
        let dims = cal.dims;
        let qvec = &cal.data[idx * dims..(idx + 1) * dims];
        let mut dot = 0.0f32;
        for d in 0..dims {
            let min = f32::from_le_bytes(cal.mins[d * 4..d * 4 + 4].try_into().unwrap());
            let scale = f32::from_le_bytes(cal.scales[d * 4..d * 4 + 4].try_into().unwrap());
            let dequant = min + qvec[d] as f32 * scale;
            dot += dequant * query[d];
        }
        let norm_offset = idx * 4;
        let stored_norm =
            f32::from_le_bytes(cal.norms[norm_offset..norm_offset + 4].try_into().unwrap());
        // Query is unit-length by the v0.7.2 invariant; only the
        // dequantized stored_norm carries quantization drift.
        if stored_norm == 0.0 {
            1.0
        } else {
            1.0 - dot / stored_norm
        }
    }

    fn greedy_closest(&self, start: u32, query: &[f32], level: usize) -> u32 {
        let mut current = start;
        let mut current_dist = self.approx_dist(current, query);
        loop {
            let mut changed = false;
            if level < self.graph.nodes[current as usize].neighbors.len() {
                for &neighbor in &self.graph.nodes[current as usize].neighbors[level] {
                    let d = self.approx_dist(neighbor, query);
                    if d < current_dist {
                        current = neighbor;
                        current_dist = d;
                        changed = true;
                    }
                }
            }
            if !changed {
                break;
            }
        }
        current
    }

    fn search_layer_0(
        &self,
        query: &[f32],
        entry: u32,
        ef: usize,
        filter: Option<&roaring::RoaringBitmap>,
    ) -> Vec<Candidate> {
        let mut visited = HashSet::new();
        let mut candidates = BinaryHeap::new();
        let mut results = BinaryHeap::new();

        let d = self.approx_dist(entry, query);
        visited.insert(entry);
        candidates.push(Candidate { id: entry, dist: d });
        if filter.is_none() || filter.unwrap().contains(entry) {
            results.push(FurthestCandidate(Candidate { id: entry, dist: d }));
        }

        while let Some(c) = candidates.pop() {
            let furthest_dist = results.peek().map(|f| f.0.dist).unwrap_or(f32::MAX);
            if c.dist > furthest_dist && results.len() >= ef {
                break;
            }

            if !self.graph.nodes[c.id as usize].neighbors.is_empty() {
                for &neighbor in &self.graph.nodes[c.id as usize].neighbors[0] {
                    if visited.insert(neighbor) {
                        let d = self.approx_dist(neighbor, query);
                        let furthest_dist = results.peek().map(|f| f.0.dist).unwrap_or(f32::MAX);
                        if d < furthest_dist || results.len() < ef {
                            candidates.push(Candidate {
                                id: neighbor,
                                dist: d,
                            });
                            if filter.is_none() || filter.unwrap().contains(neighbor) {
                                results.push(FurthestCandidate(Candidate {
                                    id: neighbor,
                                    dist: d,
                                }));
                                if results.len() > ef {
                                    results.pop();
                                }
                            }
                        }
                    }
                }
            }
        }

        let mut result: Vec<Candidate> = results.into_iter().map(|f| f.0).collect();
        result.sort_by(|a, b| a.dist.partial_cmp(&b.dist).unwrap_or(Ordering::Equal));
        result
    }

    fn brute_force_search(
        &self,
        query: &[f32],
        k: usize,
        filter: &roaring::RoaringBitmap,
    ) -> Vec<(u32, f32)> {
        let mut results: Vec<(u32, f32)> = filter
            .iter()
            .filter(|&id| (id as usize) < self.graph.num_vectors)
            .map(|id| (id, self.exact_dist(id, query)))
            .collect();
        results.sort_by(|a, b| a.1.partial_cmp(&b.1).unwrap_or(Ordering::Equal));
        results.truncate(k);
        results
    }
}

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

    fn make_params(dims: usize) -> HnswParams {
        HnswParams {
            dims,
            m: 8,
            ef_construction: 50,
            metric: DistanceMetric::L2,
            quantization: QuantizationType::DEFAULT,
        }
    }

    #[test]
    fn build_and_search_small() {
        let mut builder = HnswBuilder::new(make_params(2));
        // 4 points in a square
        builder.add_vector(vec![0.0, 0.0]).unwrap(); // 0
        builder.add_vector(vec![1.0, 0.0]).unwrap(); // 1
        builder.add_vector(vec![0.0, 1.0]).unwrap(); // 2
        builder.add_vector(vec![1.0, 1.0]).unwrap(); // 3

        let index = builder.build();
        let results = index.search(&[0.1, 0.1], 2, 10).unwrap();
        assert_eq!(results.len(), 2);
        assert_eq!(results[0].0, 0); // closest to (0,0)
    }

    #[test]
    fn search_returns_k_results() {
        let mut builder = HnswBuilder::new(make_params(3));
        for i in 0..20 {
            builder.add_vector(vec![i as f32, 0.0, 0.0]).unwrap();
        }
        let index = builder.build();
        let results = index.search(&[5.0, 0.0, 0.0], 5, 20).unwrap();
        assert_eq!(results.len(), 5);
        // Should include point 5 (exact match)
        assert!(results.iter().any(|(id, _)| *id == 5));
    }

    #[test]
    fn recall_on_random_data() {
        let dims = 16;
        let n = 200;
        let mut builder = HnswBuilder::new(HnswParams {
            dims,
            m: 12,
            ef_construction: 50,
            metric: DistanceMetric::L2,
            quantization: QuantizationType::DEFAULT,
        });

        // Deterministic "random" vectors
        let mut rng: u64 = 12345;
        let mut vectors = Vec::new();
        for _ in 0..n {
            let mut v = Vec::with_capacity(dims);
            for _ in 0..dims {
                rng ^= rng << 13;
                rng ^= rng >> 7;
                rng ^= rng << 17;
                v.push((rng as f32 / u64::MAX as f32) * 2.0 - 1.0);
            }
            vectors.push(v);
        }

        for v in &vectors {
            builder.add_vector(v.clone()).unwrap();
        }
        let index = builder.build();

        // Query with the first vector — it should find itself
        let results = index.search(&vectors[0], 1, 50).unwrap();
        assert_eq!(results[0].0, 0);

        // Compute recall@10 against brute-force
        let query = &vectors[42];
        let hnsw_results = index.search(query, 10, 50).unwrap();

        // Brute-force top-10
        let mut brute: Vec<(u32, f32)> = (0..n as u32)
            .map(|i| {
                (
                    i,
                    super::super::distance(&vectors[i as usize], query, DistanceMetric::L2),
                )
            })
            .collect();
        brute.sort_by(|a, b| a.1.partial_cmp(&b.1).unwrap());
        let brute_top10: HashSet<u32> = brute[..10].iter().map(|x| x.0).collect();
        let hnsw_top10: HashSet<u32> = hnsw_results.iter().map(|x| x.0).collect();

        let recall = brute_top10.intersection(&hnsw_top10).count() as f64 / 10.0;
        assert!(recall >= 0.8, "recall@10 = {recall}, expected >= 0.8");
    }

    /// Deterministic pseudo-random vectors for the parallel-build tests.
    fn gen_vectors(n: usize, dims: usize, seed: u64) -> Vec<Vec<f32>> {
        let mut rng = seed | 1;
        let mut vectors = Vec::with_capacity(n);
        for _ in 0..n {
            let mut v = Vec::with_capacity(dims);
            for _ in 0..dims {
                rng ^= rng << 13;
                rng ^= rng >> 7;
                rng ^= rng << 17;
                v.push((rng as f32 / u64::MAX as f32) * 2.0 - 1.0);
            }
            vectors.push(v);
        }
        vectors
    }

    fn mean_recall_at_10(index: &HnswIndex, vectors: &[Vec<f32>], queries: &[usize]) -> f64 {
        let n = vectors.len();
        let mut total = 0.0;
        for &qi in queries {
            let query = &vectors[qi];
            let hnsw: HashSet<u32> = index
                .search(query, 10, 64)
                .unwrap()
                .iter()
                .map(|x| x.0)
                .collect();
            let mut brute: Vec<(u32, f32)> = (0..n as u32)
                .map(|i| {
                    (
                        i,
                        super::super::distance(&vectors[i as usize], query, DistanceMetric::L2),
                    )
                })
                .collect();
            brute.sort_by(|a, b| a.1.partial_cmp(&b.1).unwrap());
            let truth: HashSet<u32> = brute[..10].iter().map(|x| x.0).collect();
            total += truth.intersection(&hnsw).count() as f64 / 10.0;
        }
        total / queries.len() as f64
    }

    /// `store_vector` + `connect_pending(Fixed(1))` must produce a
    /// byte-identical graph to the immediate `add_vector` path — the
    /// deferred write model's determinism contract at unit scale. (The
    /// full golden-bytes test against `main`'s serializer is the
    /// integration suite's job.)
    #[test]
    fn connect_pending_fixed1_matches_add_vector_bytes() {
        let dims = 16;
        let vectors = gen_vectors(300, dims, 0x00AB_CDEF);

        let mut immediate = HnswBuilder::new(make_params(dims));
        for v in &vectors {
            immediate.add_vector(v.clone()).unwrap();
        }
        let bytes_immediate = immediate.build().to_bytes();

        let mut deferred = HnswBuilder::new(make_params(dims));
        for v in &vectors {
            deferred.store_vector(v.clone()).unwrap();
        }
        deferred.connect_pending(BuildThreads::Fixed(1));
        let bytes_deferred = deferred.build().to_bytes();

        assert_eq!(
            bytes_immediate, bytes_deferred,
            "Fixed(1) deferred build must be byte-identical to immediate add_vector",
        );
    }

    /// Parallel `connect_pending` must produce graph quality on par with
    /// the sequential build (qdrant's results-not-topology invariant). A
    /// broken parallel link (e.g. a wrong `ready` gate) craters recall to
    /// near zero, which the band below catches.
    #[test]
    fn parallel_build_recall_matches_sequential() {
        let dims = 16;
        let vectors = gen_vectors(800, dims, 0x1234_5678);
        let queries: Vec<usize> = (0..40).map(|i| i * 17 % vectors.len()).collect();

        let mut seq = HnswBuilder::new(make_params(dims));
        for v in &vectors {
            seq.add_vector(v.clone()).unwrap();
        }
        let recall_seq = mean_recall_at_10(&seq.build(), &vectors, &queries);

        for threads in [
            BuildThreads::Fixed(1),
            BuildThreads::Fixed(8),
            BuildThreads::Ambient,
        ] {
            let mut par = HnswBuilder::new(make_params(dims));
            for v in &vectors {
                par.store_vector(v.clone()).unwrap();
            }
            par.connect_pending(threads);
            let recall_par = mean_recall_at_10(&par.build(), &vectors, &queries);
            assert!(
                recall_par >= recall_seq - 0.10,
                "{threads:?}: parallel recall {recall_par:.3} should track sequential \
                 {recall_seq:.3} (within 0.10)",
            );
        }
    }

    /// Adversarial corpus with many exact-duplicate vectors maximizes
    /// shared back-link targets (lock contention). The build must
    /// complete (no deadlock) at a high thread count and still answer.
    #[test]
    fn parallel_build_no_deadlock_on_duplicates() {
        let dims = 8;
        let distinct = gen_vectors(10, dims, 0x0000_9999);
        let mut builder = HnswBuilder::new(make_params(dims));
        for i in 0..2000 {
            builder
                .store_vector(distinct[i % distinct.len()].clone())
                .unwrap();
        }
        builder.connect_pending(BuildThreads::Fixed(16));
        let index = builder.build();
        let results = index.search(&distinct[0], 5, 32).unwrap();
        assert!(
            !results.is_empty(),
            "duplicate-heavy graph should still answer queries",
        );
    }

    #[test]
    fn filtered_search() {
        let mut builder = HnswBuilder::new(make_params(2));
        for i in 0..10 {
            builder.add_vector(vec![i as f32, 0.0]).unwrap();
        }
        let index = builder.build();

        // Only allow even IDs
        let mut filter = roaring::RoaringBitmap::new();
        for i in (0..10).step_by(2) {
            filter.insert(i);
        }

        let results = index
            .search_filtered(&[3.0, 0.0], 3, 20, Some(&filter))
            .unwrap();
        // All results should have even IDs
        for (id, _) in &results {
            assert!(id % 2 == 0, "filtered result should be even, got {id}");
        }
        // Closest even to 3.0 is 2 or 4
        assert!(results[0].0 == 2 || results[0].0 == 4);
    }

    #[test]
    fn serialization_round_trip() {
        let mut builder = HnswBuilder::new(make_params(3));
        builder.add_vector(vec![1.0, 2.0, 3.0]).unwrap();
        builder.add_vector(vec![4.0, 5.0, 6.0]).unwrap();
        builder.add_vector(vec![7.0, 8.0, 9.0]).unwrap();
        let index = builder.build();

        let bytes = index.to_bytes();
        let restored = HnswIndex::from_bytes(&bytes).unwrap();

        assert_eq!(restored.len(), 3);
        assert_eq!(restored.dims(), 3);

        let r1 = index.search(&[1.0, 2.0, 3.0], 1, 10).unwrap();
        let r2 = restored.search(&[1.0, 2.0, 3.0], 1, 10).unwrap();
        assert_eq!(r1[0].0, r2[0].0);
    }

    /// `from_bytes` must reject a malformed blob with an explicit
    /// [`LuciError::IndexCorrupted`] rather than panicking out of bounds.
    ///
    /// Regression for the deserializer-robustness gap surfaced by the
    /// visited-pool review: truncated blobs slice-panicked, and an
    /// out-of-range neighbour / entry-point id sailed through deserialization
    /// to an out-of-bounds panic during search. See
    /// [[optimization-hnsw-visited-bitset]].
    #[test]
    fn from_bytes_rejects_corrupt_blob() {
        let mut builder = HnswBuilder::new(make_params(3));
        builder.add_vector(vec![1.0, 2.0, 3.0]).unwrap();
        builder.add_vector(vec![4.0, 5.0, 6.0]).unwrap();
        builder.add_vector(vec![7.0, 8.0, 9.0]).unwrap();
        let valid = builder.build().to_bytes();
        assert!(
            HnswIndex::from_bytes(&valid).is_ok(),
            "valid blob must load"
        );

        // Truncation anywhere in the required header/vectors/nodes region must
        // error, not panic (exercises the bounded readers).
        for cut in [
            1usize,
            10,
            20,
            valid.len() / 2,
            valid.len().saturating_sub(6),
        ] {
            assert!(
                HnswIndex::from_bytes(&valid[..cut]).is_err(),
                "truncated-to-{cut} blob must be rejected, not panic"
            );
        }

        // Out-of-range entry_point (v2 header bytes [18..22]) must be rejected
        // before it can index self.vectors[ep] during search.
        let mut bad_ep = valid.clone();
        bad_ep[18..22].copy_from_slice(&9999u32.to_le_bytes());
        assert!(
            matches!(
                HnswIndex::from_bytes(&bad_ep),
                Err(LuciError::IndexCorrupted(_))
            ),
            "out-of-range entry_point must be IndexCorrupted"
        );

        // Out-of-range neighbour id must likewise be rejected. Hand-build a
        // structurally-valid v2 blob (1 dim, 2 vectors) whose node 0 points at
        // a neighbour past num_vectors, reusing the real metric byte.
        let metric_byte = valid[13];
        let mut blob = Vec::new();
        blob.extend_from_slice(&HNSW_FORMAT_MAGIC);
        blob.push(HNSW_FORMAT_VERSION);
        blob.extend_from_slice(&1u32.to_le_bytes()); // dims
        blob.extend_from_slice(&1u32.to_le_bytes()); // m
        blob.push(metric_byte);
        blob.extend_from_slice(&2u32.to_le_bytes()); // num_vectors
        blob.extend_from_slice(&0u32.to_le_bytes()); // entry_point = 0 (valid)
        blob.extend_from_slice(&0u32.to_le_bytes()); // max_level
        blob.extend_from_slice(&0.0f32.to_le_bytes()); // vector 0
        blob.extend_from_slice(&1.0f32.to_le_bytes()); // vector 1
        // node 0: level 0, 1 layer, 1 neighbour pointing out of range
        blob.extend_from_slice(&0u32.to_le_bytes()); // level
        blob.extend_from_slice(&1u32.to_le_bytes()); // num_layers
        blob.extend_from_slice(&1u32.to_le_bytes()); // num_neighbors
        blob.extend_from_slice(&99u32.to_le_bytes()); // neighbour 99 >= 2 (BAD)
        // node 1: level 0, 1 layer, 1 valid neighbour
        blob.extend_from_slice(&0u32.to_le_bytes());
        blob.extend_from_slice(&1u32.to_le_bytes());
        blob.extend_from_slice(&1u32.to_le_bytes());
        blob.extend_from_slice(&0u32.to_le_bytes()); // neighbour 0 (valid)
        blob.push(0); // quantized flag: none
        assert!(
            matches!(
                HnswIndex::from_bytes(&blob),
                Err(LuciError::IndexCorrupted(_))
            ),
            "out-of-range neighbour id must be IndexCorrupted"
        );
    }

    /// `from_bytes` must honor `"quantization": "none"`.
    ///
    /// Regression test for the auto-quantize bug at the original
    /// `hnsw.rs:1006-1011`: when the on-disk format records "no
    /// quantized data" (flag byte 0), `from_bytes` was synthesising
    /// int8 vectors anyway. That routed query-time beam search through
    /// scalar int8 dequant even though the user mapping said `"none"`.
    /// See [[hnsw-query-path-allocation-overhead]] and
    /// [[code-must-not-lie]].
    #[test]
    fn from_bytes_honors_quantization_none() {
        let mut builder = HnswBuilder::new(HnswParams {
            dims: 4,
            m: 16,
            ef_construction: 100,
            metric: DistanceMetric::L2,
            quantization: QuantizationType::None,
        });
        builder.add_vector(vec![1.0, 2.0, 3.0, 4.0]).unwrap();
        builder.add_vector(vec![5.0, 6.0, 7.0, 8.0]).unwrap();
        let index = builder.build();
        assert!(
            index.quantized.is_none(),
            "build must produce quantized = None when mapping says None",
        );

        let bytes = index.to_bytes();
        let restored = HnswIndex::from_bytes(&bytes).unwrap();

        assert!(
            restored.quantized.is_none(),
            "from_bytes must honor on-disk `no quantized data` flag; \
             auto-synthesising int8 overrides the user's mapping",
        );
        assert_eq!(restored.params.quantization, QuantizationType::None);
    }

    /// Sanity: the explicit-int8 round-trip still works after the
    /// None-honoring fix.
    #[test]
    fn from_bytes_preserves_quantization_int8() {
        let mut builder = HnswBuilder::new(HnswParams {
            dims: 4,
            m: 16,
            ef_construction: 100,
            metric: DistanceMetric::L2,
            quantization: QuantizationType::Int8,
        });
        builder.add_vector(vec![1.0, 2.0, 3.0, 4.0]).unwrap();
        builder.add_vector(vec![5.0, 6.0, 7.0, 8.0]).unwrap();
        let index = builder.build();
        assert!(
            index.quantized.is_some(),
            "build must produce quantized = Some when mapping says Int8",
        );

        let bytes = index.to_bytes();
        let restored = HnswIndex::from_bytes(&bytes).unwrap();

        assert!(restored.quantized.is_some());
        assert_eq!(restored.params.quantization, QuantizationType::Int8);
    }

    #[test]
    fn empty_index() {
        let builder = HnswBuilder::new(make_params(2));
        let index = builder.build();
        let results = index.search(&[0.0, 0.0], 5, 10).unwrap();
        assert!(results.is_empty());
    }

    #[test]
    fn single_vector() {
        let mut builder = HnswBuilder::new(make_params(2));
        builder.add_vector(vec![1.0, 1.0]).unwrap();
        let index = builder.build();
        let results = index.search(&[0.0, 0.0], 1, 10).unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].0, 0);
    }

    #[test]
    fn cosine_metric() {
        let params = HnswParams {
            dims: 3,
            m: 8,
            ef_construction: 50,
            metric: DistanceMetric::Cosine,
            quantization: QuantizationType::DEFAULT,
        };
        let mut builder = HnswBuilder::new(params);
        builder.add_vector(vec![1.0, 0.0, 0.0]).unwrap(); // 0: x-axis
        builder.add_vector(vec![0.0, 1.0, 0.0]).unwrap(); // 1: y-axis
        builder.add_vector(vec![0.9, 0.1, 0.0]).unwrap(); // 2: near x-axis

        let index = builder.build();
        let results = index.search(&[1.0, 0.0, 0.0], 2, 10).unwrap();
        // Closest by cosine to x-axis: 0 (identical) then 2 (near x-axis)
        assert_eq!(results[0].0, 0);
        assert_eq!(results[1].0, 2);
    }

    #[test]
    #[should_panic(expected = "unimplemented quantization")]
    fn hnsw_builder_panics_on_int4_quantization() {
        // Defense in depth: even if Int4 somehow reaches the builder
        // (bypassing the mapping parser), the builder must refuse rather
        // than silently substitute Int8. See [[code-must-not-lie]].
        HnswBuilder::new(HnswParams {
            dims: 4,
            m: 8,
            ef_construction: 50,
            metric: DistanceMetric::Cosine,
            quantization: QuantizationType::Int4,
        });
    }

    #[test]
    #[should_panic(expected = "unimplemented quantization")]
    fn hnsw_builder_panics_on_bbq_quantization() {
        HnswBuilder::new(HnswParams {
            dims: 4,
            m: 8,
            ef_construction: 50,
            metric: DistanceMetric::Cosine,
            quantization: QuantizationType::Bbq,
        });
    }

    /// Cosine builder normalizes each insert to unit length. Stored vectors
    /// must have norm == 1 (within f32 rounding) — that's the invariant
    /// the v2 kernel relies on to skip the per-call norm passes. See
    /// [[optimize-cosine-norm-precompute]].
    #[test]
    fn builder_normalizes_input_on_cosine() {
        let mut builder = HnswBuilder::new(HnswParams {
            dims: 3,
            m: 8,
            ef_construction: 50,
            metric: DistanceMetric::Cosine,
            quantization: QuantizationType::None,
        });
        builder.add_vector(vec![3.0, 0.0, 4.0]).unwrap(); // norm = 5
        let v = &builder.vectors[0];
        let norm_sq: f32 = v.iter().map(|x| x * x).sum();
        assert!(
            (norm_sq - 1.0).abs() < 1e-5,
            "stored vector must be unit-length, got norm_sq = {norm_sq}",
        );
        assert!((v[0] - 0.6).abs() < 1e-5);
        assert!((v[2] - 0.8).abs() < 1e-5);
    }

    /// Zero-vector insertion on Cosine must fail loudly rather than embed
    /// a vector whose cosine score is undefined. See [[code-must-not-lie]].
    #[test]
    fn builder_rejects_zero_vector_with_cosine() {
        let mut builder = HnswBuilder::new(HnswParams {
            dims: 3,
            m: 8,
            ef_construction: 50,
            metric: DistanceMetric::Cosine,
            quantization: QuantizationType::None,
        });
        let err = builder.add_vector(vec![0.0, 0.0, 0.0]).unwrap_err();
        assert!(
            matches!(err, LuciError::InvalidQuery(_)),
            "expected InvalidQuery, got {err:?}",
        );
    }

    /// DotProduct / L2 builders pass raw vectors through, including the
    /// all-zero vector — the cosine-specific normalization gate must not
    /// kick in.
    #[test]
    fn builder_accepts_zero_vector_with_dot_product() {
        let mut builder = HnswBuilder::new(HnswParams {
            dims: 3,
            m: 8,
            ef_construction: 50,
            metric: DistanceMetric::DotProduct,
            quantization: QuantizationType::None,
        });
        builder.add_vector(vec![0.0, 0.0, 0.0]).unwrap();
        assert_eq!(builder.vectors[0], vec![0.0, 0.0, 0.0]);
    }

    /// Cosine bulk insert must abort on the first zero vector so the
    /// caller learns about the bad input — silently inserting only the
    /// good ones would drop data.
    #[test]
    fn bulk_aborts_on_zero_vector() {
        let mut builder = HnswBuilder::new(HnswParams {
            dims: 3,
            m: 8,
            ef_construction: 50,
            metric: DistanceMetric::Cosine,
            quantization: QuantizationType::None,
        });
        builder.add_vector(vec![1.0, 0.0, 0.0]).unwrap();
        let err = builder.add_vector(vec![0.0, 0.0, 0.0]).unwrap_err();
        assert!(matches!(err, LuciError::InvalidQuery(_)));
        // No third insert should run.
        assert_eq!(builder.vectors.len(), 1);
    }

    #[test]
    fn hnsw_builder_accepts_none_quantization() {
        let mut builder = HnswBuilder::new(HnswParams {
            dims: 3,
            m: 8,
            ef_construction: 50,
            metric: DistanceMetric::Cosine,
            quantization: QuantizationType::None,
        });
        builder.add_vector(vec![1.0, 0.0, 0.0]).unwrap();
        let index = builder.build();
        // No quantized blob materialized when QuantizationType::None.
        assert!(index.quantized.is_none());
    }

    #[test]
    fn m_max_is_2m_at_layer_0() {
        // M_max0 = 2*M at layer 0, M_max = M at higher layers.
        // See [[fix-hnsw-neighbor-heuristic]].
        let builder = HnswBuilder::new(HnswParams {
            dims: 2,
            m: 8,
            ef_construction: 50,
            metric: DistanceMetric::L2,
            quantization: QuantizationType::None,
        });
        assert_eq!(builder.m_max(0), 16);
        assert_eq!(builder.m_max(1), 8);
        assert_eq!(builder.m_max(5), 8);
    }

    #[test]
    fn select_neighbors_heuristic_rejects_clustered_candidate() {
        // Direction-regression test. Configuration where simple
        // (m-closest) selection would pick A and B (both close to
        // query, in the same direction), and the diversity-aware
        // heuristic should pick A and C (B rejected as redundant,
        // C kept because it's in a different direction).
        //
        // Query is conceptually the origin. Vectors live in 2D:
        //   id=0 — query anchor (only used to provide a builder slot)
        //   id=1 — A at (1.0, 0.0)
        //   id=2 — B at (1.0, 0.05) — very close to A in space
        //   id=3 — C at (0.0, 1.05) — different direction, slightly farther
        let mut builder = HnswBuilder::new(HnswParams {
            dims: 2,
            m: 2,
            ef_construction: 10,
            metric: DistanceMetric::L2,
            quantization: QuantizationType::None,
        });
        builder.add_vector(vec![0.0, 0.0]).unwrap(); // 0 — query anchor
        builder.add_vector(vec![1.0, 0.0]).unwrap(); // 1 — A
        builder.add_vector(vec![1.0, 0.05]).unwrap(); // 2 — B (clustered with A)
        builder.add_vector(vec![0.0, 1.05]).unwrap(); // 3 — C (diverse from A)

        // Synthetic candidate list: dist field is dist-to-query.
        let candidates = vec![
            Candidate { id: 1, dist: 1.0 }, // A
            Candidate {
                id: 2,
                dist: 1.00125,
            }, // B (sqrt(1^2 + 0.05^2))
            Candidate { id: 3, dist: 1.05 }, // C
        ];

        let selected = builder.select_neighbors_heuristic(&candidates, 2);
        let ids: HashSet<u32> = selected.iter().map(|c| c.id).collect();

        assert_eq!(selected.len(), 2);
        assert!(
            ids.contains(&1),
            "Expected A (id 1) selected, got {:?}",
            ids
        );
        assert!(
            ids.contains(&3),
            "Expected C (id 3) selected (diverse direction), got {:?}",
            ids
        );
        assert!(
            !ids.contains(&2),
            "Expected B (id 2) rejected (too close to A), got {:?}",
            ids
        );
    }

    #[test]
    fn select_neighbors_heuristic_satisfies_diversity_invariant() {
        // For any output of the heuristic with m candidates selected,
        // every later-selected neighbor must be at least as close to
        // the query as it is to every earlier-selected neighbor.
        // This is the load-bearing invariant of Algorithm 4 —
        // robust to tie-breaking shifts that a literal-output test
        // would break on.
        let dims = 4;
        let n = 30;
        let mut builder = HnswBuilder::new(HnswParams {
            dims,
            m: 8,
            ef_construction: 50,
            metric: DistanceMetric::L2,
            quantization: QuantizationType::None,
        });

        // Deterministic synthetic vectors.
        let mut rng: u64 = 12345;
        for _ in 0..n {
            let mut v = Vec::with_capacity(dims);
            for _ in 0..dims {
                rng ^= rng << 13;
                rng ^= rng >> 7;
                rng ^= rng << 17;
                v.push((rng as f32 / u64::MAX as f32) * 2.0 - 1.0);
            }
            builder.add_vector(v).unwrap();
        }

        // Treat node 0 as the conceptual query. Build candidates
        // from all other nodes with dist field = dist(other, 0).
        let candidates: Vec<Candidate> = (1..n as u32)
            .map(|id| Candidate {
                id,
                dist: builder.dist(id, 0),
            })
            .collect();

        let selected = builder.select_neighbors_heuristic(&candidates, 8);
        assert!(!selected.is_empty(), "heuristic returned empty selection");

        for (i, s_i) in selected.iter().enumerate() {
            for s_k in &selected[..i] {
                let d_ik = builder.dist(s_i.id, s_k.id);
                assert!(
                    d_ik >= s_i.dist,
                    "Diversity invariant violated: selected[{i}] (id={}, dist_to_query={}) \
                     has dist to earlier-selected (id={}) of {}; expected >= {}",
                    s_i.id,
                    s_i.dist,
                    s_k.id,
                    d_ik,
                    s_i.dist
                );
            }
        }
    }

    /// `with_capacity_for_merge` pre-sizes vectors and nodes and
    /// `add_vector_at_ordinal` fills slot N regardless of insertion
    /// order. Built graph is searchable.
    #[test]
    fn add_vector_at_ordinal_fills_reserved_slot() {
        let params = make_params(3);
        let mut builder = HnswBuilder::with_capacity_for_merge(params, 5);
        // Insert ordinals out of order to exercise the slot semantic
        builder
            .add_vector_at_ordinal(2, vec![0.0, 0.0, 1.0])
            .unwrap();
        builder
            .add_vector_at_ordinal(0, vec![1.0, 0.0, 0.0])
            .unwrap();
        builder
            .add_vector_at_ordinal(4, vec![0.5, 0.5, 0.0])
            .unwrap();
        builder
            .add_vector_at_ordinal(1, vec![0.0, 1.0, 0.0])
            .unwrap();
        builder
            .add_vector_at_ordinal(3, vec![0.7, 0.0, 0.7])
            .unwrap();

        assert_eq!(builder.vectors[0], vec![1.0, 0.0, 0.0]);
        assert_eq!(builder.vectors[2], vec![0.0, 0.0, 1.0]);
        assert_eq!(builder.vectors[4], vec![0.5, 0.5, 0.0]);

        let index = builder.build();
        let results = index.search(&[1.0, 0.0, 0.0], 1, 10).unwrap();
        assert_eq!(results[0].0, 0, "x-axis vector should match query closest");
    }

    /// `seed_from_graph` copies a built graph's topology into a
    /// pre-sized merge builder under an ordinal remapping. Search
    /// against the seeded builder returns the same top-1 as the
    /// original, modulo the remap.
    #[test]
    fn seed_from_graph_round_trips_topology() {
        // Build a small source graph.
        let dims = 3;
        let mut src = HnswBuilder::new(make_params(dims));
        src.add_vector(vec![1.0, 0.0, 0.0]).unwrap();
        src.add_vector(vec![0.0, 1.0, 0.0]).unwrap();
        src.add_vector(vec![0.0, 0.0, 1.0]).unwrap();
        src.add_vector(vec![0.7, 0.0, 0.7]).unwrap();
        let src_index = src.build();
        let src_bytes = src_index.to_bytes();

        // Re-open the source as a parsed graph, then seed-fill a new
        // builder under an identity remap (no actual remapping).
        let graph = ParsedGraph::parse(&src_bytes).unwrap();
        let mut merged = HnswBuilder::with_capacity_for_merge(make_params(dims), 4);
        merged.seed_from_graph(&graph, &src_bytes, |src_ord| src_ord);

        // Entry point and max level survive (packed into `entry` now).
        let (merged_ep, merged_max) = unpack_entry(merged.entry.load(AtomicOrdering::Relaxed));
        assert_eq!(merged_ep, graph.entry_point.unwrap_or(ENTRY_SENTINEL));
        assert_eq!(merged_max as usize, graph.max_level);
        // All four vector slots are populated
        for ord in 0..4 {
            assert_eq!(
                merged.vectors[ord].len(),
                dims,
                "ordinal {ord} should be seeded with a {dims}-dim vector",
            );
        }

        // Search against the rebuilt graph
        let merged_index = merged.build();
        let results = merged_index.search(&[1.0, 0.0, 0.0], 1, 10).unwrap();
        assert_eq!(results[0].0, 0);
    }
}