precinct 0.9.1

//! Updatable, durable multi-segment region-ANN index via `segstore`.
//!
//! Enabled by the optional `store` feature. The base [`RegionIndex`] is
//! build-once; this wraps a corpus of axis-aligned boxes in a segstore
//! `SegmentedStore` so regions can be added and deleted incrementally with a
//! write-ahead log + checkpoint + compaction, and the index survives a restart.
//!
//! Like vicinity's store, this is *multi-segment*: a per-segment `RegionIndex`
//! is built from the live regions of each segment and searched, then the
//! per-segment top-k are merged. The cross-segment merge is exact *given* exact
//! per-segment top-k, but the underlying HNSW search is itself approximate, so
//! the merged result is approximate (as any HNSW result is). It is the deliberate
//! alternative to a single evolving graph; precinct builds on vicinity's HNSW, so
//! it inherits that family's choice.
//!
//! Each per-segment index is built over that segment's *live* regions and
//! **cached**, rebuilt only when the index is mutated (an add that seals a
//! segment, a delete, or a compaction), not on every query. The small unflushed
//! buffer is built per query.
//!
//! Specialized to [`AxisBox`]; a generic-over-[`crate::Region`] form is possible
//! once a second region type needs it.

use std::cell::RefCell;
use std::cmp::Ordering;
use std::collections::hash_map::Entry;
use std::collections::{HashMap, HashSet};
use std::io::Read;
use std::sync::Arc;

use durability::{Directory, PersistenceError, PersistenceResult};
use segstore::{DefaultStore, SegmentCatalog, SegmentedStore, SidecarEnvelope};

use crate::{AxisBox, IndexParams, Region, RegionIndex, SearchParams};

/// segstore payload: items are axis-aligned box regions, a segment is a batch of
/// source boxes (a per-segment `RegionIndex` is built + cached from the live ones).
type BoxBacking = DefaultStore<u32, AxisBox>;

/// Per-segment region indexes keyed by segstore's stable segment id. A sealed
/// add creates one new segment id, so cached indexes for existing segments are
/// reused instead of rebuilding the whole corpus on the next query.
struct Cache {
    by_segment_id: HashMap<u64, Option<RegionIndex<AxisBox>>>,
}

/// The `kind` tag for a persisted per-segment region-index sidecar.
const INDEX_KIND: &str = "region-hnsw";
const SIDECAR_MAGIC: &[u8; 8] = b"PRECIDX1";
const SIDECAR_VERSION: u32 = 2;

/// An updatable, durable multi-segment region-ANN index over axis-aligned boxes.
pub struct UpdatableIndex {
    inner: SegmentedStore<BoxBacking>,
    dim: usize,
    m: usize,
    m_max: usize,
    ef_construction: usize,
    sidecar_recipe: String,
    cache: RefCell<Cache>,
    /// Segment ids whose on-disk region-index sidecar was validated or written
    /// in this process, so checkpoint persistence stays O(new segments).
    persisted: RefCell<HashSet<u64>>,
}

/// A read-only checkpoint view that loads per-segment region-index sidecars
/// before falling back to source region segment payloads.
///
/// This is the restart/query path for larger stores whose built per-segment
/// `RegionIndex`es have already been persisted by [`UpdatableIndex::checkpoint`].
/// It opens the segstore manifest without decoding source segments, then loads
/// sidecars. A sidecar that contains a tombstoned id is rebuilt from that one
/// segment before search; the underlying HNSW searches are approximate, so
/// filtering deleted hits after a truncated graph search is not enough to
/// preserve recall.
pub struct SnapshotIndex {
    catalog: SegmentCatalog<u32>,
    dim: usize,
    m: usize,
    m_max: usize,
    ef_construction: usize,
    sidecar_recipe: String,
    cache: RefCell<HashMap<u64, Option<RegionIndex<AxisBox>>>>,
}

impl UpdatableIndex {
    /// Open (or recover) an index under `dir` for `dim`-dimensional regions, using
    /// `params` to build each segment's index. Up to `flush_threshold` regions are
    /// buffered before a new immutable segment is sealed.
    pub fn open(
        dir: Arc<dyn Directory>,
        flush_threshold: usize,
        dim: usize,
        params: IndexParams,
    ) -> PersistenceResult<Self> {
        Ok(Self {
            inner: SegmentedStore::open(dir, BoxBacking::new(), flush_threshold)?,
            dim,
            m: params.m,
            m_max: params.m_max,
            ef_construction: params.ef_construction,
            sidecar_recipe: Self::make_sidecar_recipe(dim, params),
            cache: RefCell::new(Cache {
                by_segment_id: HashMap::new(),
            }),
            persisted: RefCell::new(HashSet::new()),
        })
    }

    /// Add (or re-add) a box region by id. Returns an error if the region's
    /// dimension does not match the index, rather than silently dropping it from
    /// every per-segment rebuild.
    pub fn add(&mut self, id: u32, region: AxisBox) -> PersistenceResult<()> {
        if region.dim() != self.dim {
            return Err(PersistenceError::InvalidConfig(format!(
                "region dimension {} does not match index dimension {}",
                region.dim(),
                self.dim
            )));
        }
        // A sealed add introduces a new segment id; existing segment ids stay
        // stable, so the cache reuses them and builds only the new one.
        self.inner.add(id, region)?;
        Ok(())
    }

    /// Add (or re-add) many regions, syncing the write-ahead log once for the whole
    /// batch instead of once per region. This is the bulk-ingest path (the
    /// corpus-load phase): per-item WAL sync is the dominant cost on a real disk, so
    /// one sync per batch is several times faster than a loop of [`Self::add`].
    /// Every region's dimension is validated before any is ingested.
    pub fn extend(
        &mut self,
        regions: impl IntoIterator<Item = (u32, AxisBox)>,
    ) -> PersistenceResult<()> {
        let dim = self.dim;
        let validated: Result<Vec<(u32, AxisBox)>, PersistenceError> = regions
            .into_iter()
            .map(|(id, region)| {
                if region.dim() != dim {
                    Err(PersistenceError::InvalidConfig(format!(
                        "region dimension {} does not match index dimension {}",
                        region.dim(),
                        dim
                    )))
                } else {
                    Ok((id, region))
                }
            })
            .collect();
        self.inner.extend(validated?)?;
        Ok(())
    }

    /// Tombstone a region.
    pub fn delete(&mut self, id: u32) -> PersistenceResult<()> {
        self.inner.delete(id)?;
        // A tombstone only changes the live-set of the segment that holds `id`, so
        // invalidate just that segment's cached index -- not the whole cache.
        let mut cache = self.cache.borrow_mut();
        let ids = self.inner.segment_ids();
        for (seg_idx, seg) in self.inner.segments().iter().enumerate() {
            if seg.iter().any(|(sid, _)| *sid == id) {
                let seg_id = ids[seg_idx];
                cache.by_segment_id.remove(&seg_id);
                self.persisted.borrow_mut().remove(&seg_id);
                let _ = self
                    .inner
                    .dir()
                    .delete(&self.inner.index_name(seg_id, INDEX_KIND));
            }
        }
        Ok(())
    }

    /// Merge segments (dropping tombstoned regions) and persist a checkpoint.
    pub fn compact(&mut self) -> PersistenceResult<()> {
        self.inner.compact()?;
        self.prune_cache_to_current_segments();
        self.persist_new_segments();
        Ok(())
    }

    /// Persist a checkpoint without merging.
    pub fn checkpoint(&mut self) -> PersistenceResult<()> {
        self.inner.checkpoint()?;
        self.persist_new_segments();
        Ok(())
    }

    /// Run one round of size-tiered compaction, merging similarly-sized segments
    /// so the segment count stays bounded without a full [`compact`](Self::compact).
    pub fn compact_tiers(&mut self) -> PersistenceResult<()> {
        let stats = self.inner.compact_tiers()?;
        if stats.merges > 0 {
            self.prune_cache_to_current_segments();
            self.persist_new_segments();
        }
        Ok(())
    }

    /// Merge only the segments whose live ratio is below `min_live_ratio`,
    /// reclaiming tombstoned regions -- the cheap alternative to a full
    /// [`compact`](Self::compact) when a few segments are delete-heavy.
    pub fn reclaim(&mut self, min_live_ratio: f64) -> PersistenceResult<()> {
        let stats = self.inner.reclaim_tombstones(min_live_ratio)?;
        if stats.merges > 0 {
            self.prune_cache_to_current_segments();
            self.persist_new_segments();
        }
        Ok(())
    }

    /// Storage amplification: stored regions divided by live regions (`1.0` with
    /// no tombstones, higher as deletes accumulate).
    pub fn space_amplification(&self) -> Option<f64> {
        self.inner.space_amplification()
    }

    /// The `k` nearest regions to the query point, by point-to-region distance,
    /// over the live corpus. Returns `(region_id, distance)`.
    pub fn search(&self, query: &[f32], k: usize, params: SearchParams) -> Vec<(u32, f32)> {
        let SearchParams { ef, overretrieve } = params;
        let sp = || SearchParams { ef, overretrieve };
        Self::truncate_nearest(
            self.collect_from_segment_indexes(|idx| idx.search(query, k, sp()).unwrap_or_default()),
            k,
        )
    }

    /// Regions that contain `point`, unioned over live sealed segments and the
    /// unflushed buffer. Approximate: recall is bounded by `params`.
    pub fn containing(&self, point: &[f32], params: SearchParams) -> Vec<u32> {
        let SearchParams { ef, overretrieve } = params;
        let sp = || SearchParams { ef, overretrieve };
        Self::sort_dedup_ids(
            self.collect_from_segment_indexes(|idx| {
                idx.containing(point, sp()).unwrap_or_default()
            }),
        )
    }

    /// Regions that fully contain `query`, unioned over live segments.
    /// Approximate: recall is bounded by `params`.
    pub fn subsumers(&self, query: &AxisBox, params: SearchParams) -> Vec<u32> {
        let SearchParams { ef, overretrieve } = params;
        let sp = || SearchParams { ef, overretrieve };
        Self::sort_dedup_ids(
            self.collect_from_segment_indexes(|idx| idx.subsumers(query, sp()).unwrap_or_default()),
        )
    }

    /// Regions that softly subsume `query`, returned with probability highest
    /// first and unioned over live segments. Approximate: recall is bounded by
    /// `params`.
    pub fn subsumers_soft(
        &self,
        query: &AxisBox,
        min_prob: f32,
        params: SearchParams,
    ) -> Vec<(u32, f32)> {
        let SearchParams { ef, overretrieve } = params;
        let sp = || SearchParams { ef, overretrieve };
        let mut by_id: HashMap<u32, f32> = HashMap::new();
        for (id, p) in self.collect_from_segment_indexes(|idx| {
            idx.subsumers_soft(query, min_prob, sp())
                .unwrap_or_default()
        }) {
            by_id
                .entry(id)
                .and_modify(|existing| *existing = existing.max(p))
                .or_insert(p);
        }
        let mut out: Vec<(u32, f32)> = by_id.into_iter().collect();
        out.sort_unstable_by(|a, b| {
            b.1.partial_cmp(&a.1)
                .unwrap_or(Ordering::Equal)
                .then_with(|| a.0.cmp(&b.0))
        });
        out
    }

    /// Regions fully contained in `query`, unioned over live segments.
    /// Exhaustive within each segment.
    pub fn subsumees(&self, query: &AxisBox) -> Vec<u32> {
        Self::sort_dedup_ids(self.collect_from_segment_indexes(|idx| idx.subsumees(query)))
    }

    /// Regions that intersect `query`, unioned over live segments. Approximate:
    /// recall is bounded by `params`.
    pub fn overlapping(&self, query: &AxisBox, params: SearchParams) -> Vec<u32> {
        let SearchParams { ef, overretrieve } = params;
        let sp = || SearchParams { ef, overretrieve };
        Self::sort_dedup_ids(
            self.collect_from_segment_indexes(|idx| {
                idx.overlapping(query, sp()).unwrap_or_default()
            }),
        )
    }

    /// The `k` regions nearest to `query` by center distance.
    pub fn nearest_region(
        &self,
        query: &AxisBox,
        k: usize,
        params: SearchParams,
    ) -> Vec<(u32, f32)> {
        let SearchParams { ef, overretrieve } = params;
        let sp = || SearchParams { ef, overretrieve };
        Self::truncate_nearest(
            self.collect_from_segment_indexes(|idx| {
                idx.nearest_region(query, k, sp()).unwrap_or_default()
            }),
            k,
        )
    }

    /// Exhaustive containment query over live segments.
    pub fn containing_exhaustive(&self, point: &[f32]) -> Vec<u32> {
        Self::sort_dedup_ids(
            self.collect_from_segment_indexes(|idx| idx.containing_exhaustive(point)),
        )
    }

    /// Exhaustive subsumer query over live segments.
    pub fn subsumers_exhaustive(&self, query: &AxisBox) -> Vec<u32> {
        Self::sort_dedup_ids(
            self.collect_from_segment_indexes(|idx| idx.subsumers_exhaustive(query)),
        )
    }

    /// Exhaustive overlap query over live segments.
    pub fn overlapping_exhaustive(&self, query: &AxisBox) -> Vec<u32> {
        Self::sort_dedup_ids(
            self.collect_from_segment_indexes(|idx| idx.overlapping_exhaustive(query)),
        )
    }

    /// Exhaustive nearest-region search over live segments.
    pub fn search_exhaustive(&self, query: &[f32], k: usize) -> Vec<(u32, f32)> {
        Self::truncate_nearest(
            self.collect_from_segment_indexes(|idx| idx.search_exhaustive(query, k)),
            k,
        )
    }

    fn collect_from_segment_indexes<T>(
        &self,
        mut f: impl FnMut(&RegionIndex<AxisBox>) -> Vec<T>,
    ) -> Vec<T> {
        let mut out = Vec::new();
        {
            let segs = self.inner.segments();
            let mut cache = self.cache.borrow_mut();
            // Build only segments not already cached, loading a persisted sidecar
            // first when one matches the current recipe and live id set.
            let ids = self.inner.segment_ids();
            for (i, seg) in segs.iter().enumerate() {
                let seg_id = ids[i];
                let index = cache
                    .by_segment_id
                    .entry(seg_id)
                    .or_insert_with(|| self.build_or_load(&seg[..], seg_id));
                if let Some(idx) = index {
                    out.extend(f(idx));
                }
            }
        }
        let buffered = self.inner.buffer();
        if let Some(idx) = self.build_live_index(buffered) {
            out.extend(f(&idx));
        }
        out
    }

    fn sort_dedup_ids(mut ids: Vec<u32>) -> Vec<u32> {
        ids.sort_unstable();
        ids.dedup();
        ids
    }

    fn truncate_nearest(mut cand: Vec<(u32, f32)>, k: usize) -> Vec<(u32, f32)> {
        // Lower distance is nearer.
        cand.sort_by(|a, b| {
            a.1.partial_cmp(&b.1)
                .unwrap_or(Ordering::Equal)
                .then_with(|| a.0.cmp(&b.0))
        });
        cand.truncate(k);
        cand
    }

    fn prune_cache_to_current_segments(&self) {
        let current: HashSet<u64> = self.inner.segment_ids().iter().copied().collect();
        self.cache
            .borrow_mut()
            .by_segment_id
            .retain(|id, _| current.contains(id));
    }

    /// Build a per-segment `RegionIndex` over the live regions of `batch` (None if
    /// empty or the build fails).
    fn build_live_index(&self, batch: &[(u32, AxisBox)]) -> Option<RegionIndex<AxisBox>> {
        let params = IndexParams {
            m: self.m,
            m_max: self.m_max,
            ef_construction: self.ef_construction,
        };
        Self::build_live_index_from(self.dim, params, batch, &|id| self.inner.is_live(id))
    }

    fn build_live_index_from(
        dim: usize,
        params: IndexParams,
        batch: &[(u32, AxisBox)],
        live: &dyn Fn(&u32) -> bool,
    ) -> Option<RegionIndex<AxisBox>> {
        let mut idx = match RegionIndex::<AxisBox>::new(dim, params) {
            Ok(i) => i,
            Err(_) => return None,
        };
        let mut any = false;
        for (id, region) in batch {
            if live(id) && idx.add(*id, region.clone()).is_ok() {
                any = true;
            }
        }
        if !any || idx.build().is_err() {
            return None;
        }
        Some(idx)
    }

    /// Load segment `seg_id`'s persisted region index from its sidecar, or build
    /// it over the segment's live regions and persist it for the next restart.
    fn build_or_load(&self, seg: &[(u32, AxisBox)], seg_id: u64) -> Option<RegionIndex<AxisBox>> {
        if let Some(idx) = self.load_sidecar(seg, seg_id) {
            self.persisted.borrow_mut().insert(seg_id);
            return Some(idx);
        }
        let idx = self.build_live_index(seg)?;
        self.persist_sidecar(&idx, seg_id);
        Some(idx)
    }

    /// Load a sidecar only if its recipe matches and its ids match the segment's
    /// current live ids. A stale sidecar can never serve a tombstoned region.
    fn load_sidecar(&self, seg: &[(u32, AxisBox)], seg_id: u64) -> Option<RegionIndex<AxisBox>> {
        let name = self.inner.index_name(seg_id, INDEX_KIND);
        if !self.inner.dir().exists(&name) {
            return None;
        }
        let mut bytes = Vec::new();
        self.inner
            .dir()
            .open_file(&name)
            .ok()?
            .read_to_end(&mut bytes)
            .ok()?;
        let index_bytes = self.decode_sidecar(&bytes, seg_id)?;
        let idx = RegionIndex::from_postcard(index_bytes).ok()?;
        let mut live = HashSet::with_capacity(seg.len());
        for (id, _) in seg {
            if self.inner.is_live(id) {
                live.insert(*id);
            }
        }
        if idx.ids().len() == live.len() && idx.ids().iter().all(|id| live.contains(id)) {
            Some(idx)
        } else {
            None
        }
    }

    /// Persist a built per-segment region index as its sidecar. Best-effort: a
    /// failed write leaves the in-memory index usable and simply rebuilds later.
    fn persist_sidecar(&self, idx: &RegionIndex<AxisBox>, seg_id: u64) {
        if let Ok(index) = idx.to_postcard() {
            let Some(bytes) = self.encode_sidecar(&index, seg_id) else {
                return;
            };
            if self
                .inner
                .dir()
                .atomic_write(&self.inner.index_name(seg_id, INDEX_KIND), &bytes)
                .is_ok()
            {
                self.persisted.borrow_mut().insert(seg_id);
            }
        }
    }

    fn make_sidecar_recipe(dim: usize, params: IndexParams) -> String {
        format!(
            "precinct-store-region-hnsw-v1;\
             region=axis-box;dim={};m={};m_max={};ef_construction={};\
             center=vicinity-hnsw-l2;lift=power-distance-mips-l2;\
             codec=postcard-region-index-v1",
            dim, params.m, params.m_max, params.ef_construction
        )
    }

    fn encode_sidecar(&self, index: &[u8], seg_id: u64) -> Option<Vec<u8>> {
        Self::encode_sidecar_for_recipe(&self.sidecar_recipe, seg_id, index)
    }

    fn encode_sidecar_for_recipe(
        sidecar_recipe: &str,
        seg_id: u64,
        index: &[u8],
    ) -> Option<Vec<u8>> {
        SidecarEnvelope::encode(
            SIDECAR_MAGIC,
            SIDECAR_VERSION,
            seg_id,
            sidecar_recipe.as_bytes(),
            index,
        )
        .ok()
    }

    fn decode_sidecar<'a>(&self, bytes: &'a [u8], seg_id: u64) -> Option<&'a [u8]> {
        Self::decode_sidecar_for_recipe(&self.sidecar_recipe, seg_id, bytes)
    }

    fn decode_sidecar_for_recipe<'a>(
        sidecar_recipe: &str,
        seg_id: u64,
        bytes: &'a [u8],
    ) -> Option<&'a [u8]> {
        SidecarEnvelope::decode(
            SIDECAR_MAGIC,
            SIDECAR_VERSION,
            seg_id,
            sidecar_recipe.as_bytes(),
            bytes,
        )
        .ok()
    }

    /// Persist sidecars for sealed segments that lack a current one. This is
    /// incremental: already validated/written segment ids are skipped.
    fn persist_new_segments(&self) {
        let ids = self.inner.segment_ids();
        let id_set: HashSet<u64> = ids.iter().copied().collect();
        self.persisted.borrow_mut().retain(|id| id_set.contains(id));
        for (i, seg) in self.inner.segments().iter().enumerate() {
            let seg_id = ids[i];
            if self.persisted.borrow().contains(&seg_id) {
                continue;
            }
            if self.load_sidecar(&seg[..], seg_id).is_some() {
                self.persisted.borrow_mut().insert(seg_id);
                continue;
            }
            if let Some(idx) = self.build_live_index(&seg[..]) {
                self.persist_sidecar(&idx, seg_id);
            }
        }
    }
}

impl SnapshotIndex {
    /// Open the last checkpoint under `dir` as a read-only region-search
    /// snapshot.
    ///
    /// WAL records after the last checkpoint are intentionally not visible;
    /// checkpoint before opening a snapshot when newly added regions must be
    /// searchable through this path.
    pub fn open(
        dir: Arc<dyn Directory>,
        dim: usize,
        params: IndexParams,
    ) -> PersistenceResult<Self> {
        let m = params.m;
        let m_max = params.m_max;
        let ef_construction = params.ef_construction;
        Ok(Self {
            catalog: SegmentCatalog::open(dir)?,
            dim,
            sidecar_recipe: UpdatableIndex::make_sidecar_recipe(dim, params),
            m,
            m_max,
            ef_construction,
            cache: RefCell::new(HashMap::new()),
        })
    }

    /// Number of checkpointed immutable segments in this snapshot.
    pub fn segment_count(&self) -> usize {
        self.catalog.segment_count()
    }

    /// Number of tombstoned region ids in this snapshot.
    pub fn tombstone_count(&self) -> usize {
        self.catalog.tombstone_count()
    }

    /// The `k` nearest regions to the query point, by point-to-region distance.
    pub fn search(
        &self,
        query: &[f32],
        k: usize,
        params: SearchParams,
    ) -> PersistenceResult<Vec<(u32, f32)>> {
        let SearchParams { ef, overretrieve } = params;
        let sp = || SearchParams { ef, overretrieve };
        Ok(UpdatableIndex::truncate_nearest(
            self.collect_from_segment_indexes(|idx| {
                idx.search(query, k, sp()).unwrap_or_default()
            })?,
            k,
        ))
    }

    /// Regions that contain `point`, unioned over live checkpointed segments.
    pub fn containing(&self, point: &[f32], params: SearchParams) -> PersistenceResult<Vec<u32>> {
        let SearchParams { ef, overretrieve } = params;
        let sp = || SearchParams { ef, overretrieve };
        Ok(UpdatableIndex::sort_dedup_ids(
            self.collect_from_segment_indexes(|idx| {
                idx.containing(point, sp()).unwrap_or_default()
            })?,
        ))
    }

    /// Regions that fully contain `query`, unioned over live checkpointed
    /// segments.
    pub fn subsumers(&self, query: &AxisBox, params: SearchParams) -> PersistenceResult<Vec<u32>> {
        let SearchParams { ef, overretrieve } = params;
        let sp = || SearchParams { ef, overretrieve };
        Ok(UpdatableIndex::sort_dedup_ids(
            self.collect_from_segment_indexes(|idx| {
                idx.subsumers(query, sp()).unwrap_or_default()
            })?,
        ))
    }

    /// Regions that softly subsume `query`, returned with probability highest
    /// first.
    pub fn subsumers_soft(
        &self,
        query: &AxisBox,
        min_prob: f32,
        params: SearchParams,
    ) -> PersistenceResult<Vec<(u32, f32)>> {
        let SearchParams { ef, overretrieve } = params;
        let sp = || SearchParams { ef, overretrieve };
        let mut by_id: HashMap<u32, f32> = HashMap::new();
        for (id, p) in self.collect_from_segment_indexes(|idx| {
            idx.subsumers_soft(query, min_prob, sp())
                .unwrap_or_default()
        })? {
            by_id
                .entry(id)
                .and_modify(|existing| *existing = existing.max(p))
                .or_insert(p);
        }
        let mut out: Vec<(u32, f32)> = by_id.into_iter().collect();
        out.sort_unstable_by(|a, b| {
            b.1.partial_cmp(&a.1)
                .unwrap_or(Ordering::Equal)
                .then_with(|| a.0.cmp(&b.0))
        });
        Ok(out)
    }

    /// Regions fully contained in `query`, unioned over live checkpointed
    /// segments.
    pub fn subsumees(&self, query: &AxisBox) -> PersistenceResult<Vec<u32>> {
        Ok(UpdatableIndex::sort_dedup_ids(
            self.collect_from_segment_indexes(|idx| idx.subsumees(query))?,
        ))
    }

    /// Regions that intersect `query`, unioned over live checkpointed segments.
    pub fn overlapping(
        &self,
        query: &AxisBox,
        params: SearchParams,
    ) -> PersistenceResult<Vec<u32>> {
        let SearchParams { ef, overretrieve } = params;
        let sp = || SearchParams { ef, overretrieve };
        Ok(UpdatableIndex::sort_dedup_ids(
            self.collect_from_segment_indexes(|idx| {
                idx.overlapping(query, sp()).unwrap_or_default()
            })?,
        ))
    }

    /// The `k` regions nearest to `query` by center distance.
    pub fn nearest_region(
        &self,
        query: &AxisBox,
        k: usize,
        params: SearchParams,
    ) -> PersistenceResult<Vec<(u32, f32)>> {
        let SearchParams { ef, overretrieve } = params;
        let sp = || SearchParams { ef, overretrieve };
        Ok(UpdatableIndex::truncate_nearest(
            self.collect_from_segment_indexes(|idx| {
                idx.nearest_region(query, k, sp()).unwrap_or_default()
            })?,
            k,
        ))
    }

    /// Exhaustive containment query over live checkpointed segments.
    pub fn containing_exhaustive(&self, point: &[f32]) -> PersistenceResult<Vec<u32>> {
        Ok(UpdatableIndex::sort_dedup_ids(
            self.collect_from_segment_indexes(|idx| idx.containing_exhaustive(point))?,
        ))
    }

    /// Exhaustive subsumer query over live checkpointed segments.
    pub fn subsumers_exhaustive(&self, query: &AxisBox) -> PersistenceResult<Vec<u32>> {
        Ok(UpdatableIndex::sort_dedup_ids(
            self.collect_from_segment_indexes(|idx| idx.subsumers_exhaustive(query))?,
        ))
    }

    /// Exhaustive overlap query over live checkpointed segments.
    pub fn overlapping_exhaustive(&self, query: &AxisBox) -> PersistenceResult<Vec<u32>> {
        Ok(UpdatableIndex::sort_dedup_ids(
            self.collect_from_segment_indexes(|idx| idx.overlapping_exhaustive(query))?,
        ))
    }

    /// Exhaustive nearest-region search over live checkpointed segments.
    pub fn search_exhaustive(&self, query: &[f32], k: usize) -> PersistenceResult<Vec<(u32, f32)>> {
        Ok(UpdatableIndex::truncate_nearest(
            self.collect_from_segment_indexes(|idx| idx.search_exhaustive(query, k))?,
            k,
        ))
    }

    fn collect_from_segment_indexes<T>(
        &self,
        mut f: impl FnMut(&RegionIndex<AxisBox>) -> Vec<T>,
    ) -> PersistenceResult<Vec<T>> {
        let mut out = Vec::new();
        let mut cache = self.cache.borrow_mut();
        let current: HashSet<u64> = self.catalog.segment_ids().iter().copied().collect();
        cache.retain(|seg_id, _| current.contains(seg_id));
        for &seg_id in self.catalog.segment_ids() {
            if let Entry::Vacant(entry) = cache.entry(seg_id) {
                let index = self.build_or_load(seg_id)?;
                entry.insert(index);
            }
            if let Some(Some(idx)) = cache.get(&seg_id) {
                out.extend(f(idx));
            }
        }
        Ok(out)
    }

    fn build_or_load(&self, seg_id: u64) -> PersistenceResult<Option<RegionIndex<AxisBox>>> {
        if let Some(index) = self.load_sidecar(seg_id) {
            return Ok(Some(index));
        }
        let segment: Vec<(u32, AxisBox)> = self.catalog.read_segment(seg_id)?;
        let params = IndexParams {
            m: self.m,
            m_max: self.m_max,
            ef_construction: self.ef_construction,
        };
        let index = UpdatableIndex::build_live_index_from(self.dim, params, &segment, &|id| {
            self.catalog.is_live(id)
        });
        if let Some(index) = &index {
            self.persist_sidecar(index, seg_id);
        }
        Ok(index)
    }

    fn load_sidecar(&self, seg_id: u64) -> Option<RegionIndex<AxisBox>> {
        let name = self.catalog.index_name(seg_id, INDEX_KIND);
        if !self.catalog.dir().exists(&name) {
            return None;
        }
        let mut bytes = Vec::new();
        self.catalog
            .dir()
            .open_file(&name)
            .ok()?
            .read_to_end(&mut bytes)
            .ok()?;
        let index_bytes =
            UpdatableIndex::decode_sidecar_for_recipe(&self.sidecar_recipe, seg_id, &bytes)?;
        let idx = RegionIndex::from_postcard(index_bytes).ok()?;
        if idx.ids().iter().all(|id| self.catalog.is_live(id)) {
            Some(idx)
        } else {
            None
        }
    }

    fn persist_sidecar(&self, idx: &RegionIndex<AxisBox>, seg_id: u64) {
        if let Ok(index) = idx.to_postcard() {
            let Some(bytes) =
                UpdatableIndex::encode_sidecar_for_recipe(&self.sidecar_recipe, seg_id, &index)
            else {
                return;
            };
            let _ = self
                .catalog
                .dir()
                .atomic_write(&self.catalog.index_name(seg_id, INDEX_KIND), &bytes);
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use durability::MemoryDirectory;
    use std::io::{Read, Write};
    use std::path::PathBuf;

    struct RecordingDirectory {
        inner: Arc<dyn Directory>,
        opened: Arc<std::sync::Mutex<Vec<String>>>,
    }

    impl RecordingDirectory {
        fn wrap(
            inner: Arc<dyn Directory>,
        ) -> (Arc<dyn Directory>, Arc<std::sync::Mutex<Vec<String>>>) {
            let opened = Arc::new(std::sync::Mutex::new(Vec::new()));
            (
                Arc::new(Self {
                    inner,
                    opened: opened.clone(),
                }),
                opened,
            )
        }
    }

    impl Directory for RecordingDirectory {
        fn create_file(&self, path: &str) -> PersistenceResult<Box<dyn Write + Send>> {
            self.inner.create_file(path)
        }

        fn open_file(&self, path: &str) -> PersistenceResult<Box<dyn Read + Send>> {
            if let Ok(mut opened) = self.opened.lock() {
                opened.push(path.to_string());
            }
            self.inner.open_file(path)
        }

        fn exists(&self, path: &str) -> bool {
            self.inner.exists(path)
        }

        fn delete(&self, path: &str) -> PersistenceResult<()> {
            self.inner.delete(path)
        }

        fn atomic_rename(&self, from: &str, to: &str) -> PersistenceResult<()> {
            self.inner.atomic_rename(from, to)
        }

        fn create_dir_all(&self, path: &str) -> PersistenceResult<()> {
            self.inner.create_dir_all(path)
        }

        fn list_dir(&self, path: &str) -> PersistenceResult<Vec<String>> {
            self.inner.list_dir(path)
        }

        fn append_file(&self, path: &str) -> PersistenceResult<Box<dyn Write + Send>> {
            self.inner.append_file(path)
        }

        fn atomic_write(&self, path: &str, data: &[u8]) -> PersistenceResult<()> {
            self.inner.atomic_write(path, data)
        }

        fn file_path(&self, path: &str) -> Option<PathBuf> {
            self.inner.file_path(path)
        }
    }

    fn b(lo: f32, hi: f32) -> AxisBox {
        AxisBox::new(vec![lo, lo], vec![hi, hi])
    }

    fn read_file(dir: &Arc<dyn Directory>, name: &str) -> Vec<u8> {
        let mut bytes = Vec::new();
        dir.open_file(name)
            .unwrap()
            .read_to_end(&mut bytes)
            .unwrap();
        bytes
    }

    fn checkpointed_store(dir: Arc<dyn Directory>, params: IndexParams) -> (String, Vec<u8>) {
        let mut store = UpdatableIndex::open(dir, 4, 2, params).unwrap();
        for i in 0..12u32 {
            let lo = i as f32 * 0.25;
            store.add(i, b(lo, lo + 0.5)).unwrap();
        }
        store.checkpoint().unwrap();
        let seg_id = store.inner.segment_ids()[0];
        let name = store.inner.index_name(seg_id, INDEX_KIND);
        let bytes = read_file(store.inner.dir(), &name);
        (name, bytes)
    }

    #[test]
    fn add_delete_compact_recover_through_real_region_index() {
        let dir = MemoryDirectory::arc();
        {
            let mut store =
                UpdatableIndex::open(dir.clone(), 2, 2, IndexParams::default()).unwrap();
            store.add(0, b(0.0, 1.0)).unwrap(); // near the origin
            store.add(1, b(5.0, 6.0)).unwrap(); // flush
            store.add(2, b(10.0, 11.0)).unwrap(); // far; buffered

            // A point inside box 0 is nearest to region 0.
            let top: Vec<u32> = store
                .search(&[0.5, 0.5], 1, SearchParams::default())
                .into_iter()
                .map(|(id, _)| id)
                .collect();
            assert_eq!(top, vec![0], "point inside box 0 retrieves region 0");
            // Second query (no mutation) must use the cache and stay correct.
            let again: Vec<u32> = store
                .search(&[0.5, 0.5], 1, SearchParams::default())
                .into_iter()
                .map(|(id, _)| id)
                .collect();
            assert_eq!(again, vec![0], "cached query is stable");

            store.delete(0).unwrap();
            let top: Vec<u32> = store
                .search(&[0.5, 0.5], 1, SearchParams::default())
                .into_iter()
                .map(|(id, _)| id)
                .collect();
            assert_eq!(top, vec![1], "after deleting 0, nearest region is 1");

            store.compact().unwrap();
            assert_eq!(
                store
                    .search(&[0.5, 0.5], 1, SearchParams::default())
                    .first()
                    .map(|(id, _)| *id),
                Some(1)
            );
        }
        let store = UpdatableIndex::open(dir, 2, 2, IndexParams::default()).unwrap();
        let top: Vec<u32> = store
            .search(&[0.5, 0.5], 1, SearchParams::default())
            .into_iter()
            .map(|(id, _)| id)
            .collect();
        assert_eq!(top, vec![1], "recovery preserves the search");
    }

    #[test]
    fn store_region_queries_cover_segments_and_buffer() {
        let dir = MemoryDirectory::arc();
        let mut store = UpdatableIndex::open(dir, 2, 2, IndexParams::default()).unwrap();
        store
            .add(0, AxisBox::new(vec![0.0, 0.0], vec![10.0, 10.0]))
            .unwrap();
        store
            .add(1, AxisBox::new(vec![1.0, 1.0], vec![2.0, 2.0]))
            .unwrap();
        store
            .add(2, AxisBox::new(vec![8.0, 8.0], vec![12.0, 12.0]))
            .unwrap();
        store
            .add(3, AxisBox::new(vec![20.0, 20.0], vec![21.0, 21.0]))
            .unwrap();
        store
            .add(4, AxisBox::new(vec![4.0, 4.0], vec![6.0, 6.0]))
            .unwrap();

        let params = || SearchParams {
            ef: 100,
            overretrieve: 100,
        };
        let inner = AxisBox::new(vec![1.25, 1.25], vec![1.75, 1.75]);
        let overlap = AxisBox::new(vec![9.0, 9.0], vec![11.0, 11.0]);

        assert_eq!(store.containing(&[5.0, 5.0], params()), vec![0, 4]);
        assert_eq!(store.subsumers(&inner, params()), vec![0, 1]);
        assert_eq!(store.subsumees(&b(0.0, 10.0)), vec![0, 1, 4]);
        assert_eq!(store.overlapping(&overlap, params()), vec![0, 2]);
        assert_eq!(store.containing_exhaustive(&[5.0, 5.0]), vec![0, 4]);
        assert_eq!(store.subsumers_exhaustive(&inner), vec![0, 1]);
        assert_eq!(store.overlapping_exhaustive(&overlap), vec![0, 2]);

        let soft = store.subsumers_soft(&overlap, 0.2, params());
        assert_eq!(
            soft.iter().map(|(id, _)| *id).collect::<Vec<_>>(),
            vec![2, 0]
        );
        assert!((soft[0].1 - 1.0).abs() < 1e-5);
        assert!((soft[1].1 - 0.25).abs() < 1e-5);

        let nearest_region = store.nearest_region(&overlap, 1, params());
        assert_eq!(nearest_region[0].0, 2);

        let exact = store.search_exhaustive(&[5.0, 5.0], 2);
        assert_eq!(
            exact.iter().map(|(id, _)| *id).collect::<Vec<_>>(),
            vec![0, 4]
        );
    }

    #[test]
    fn checkpoint_persists_sidecars_and_reopen_loads_them() {
        let dir = MemoryDirectory::arc();
        {
            let mut store =
                UpdatableIndex::open(dir.clone(), 4, 2, IndexParams::default()).unwrap();
            for i in 0..12u32 {
                let lo = i as f32 * 0.25;
                store.add(i, b(lo, lo + 0.5)).unwrap();
            }
            store.checkpoint().unwrap();

            let ids: Vec<u64> = store.inner.segment_ids().to_vec();
            assert!(
                !ids.is_empty(),
                "12 adds at flush 4 seal at least one segment"
            );
            for id in &ids {
                assert!(
                    store
                        .inner
                        .dir()
                        .exists(&store.inner.index_name(*id, INDEX_KIND)),
                    "segment {id} must have a persisted sidecar after checkpoint"
                );
            }
        }

        let store = UpdatableIndex::open(dir, 4, 2, IndexParams::default()).unwrap();
        assert!(
            !store
                .search(&[0.3, 0.3], 1, SearchParams::default())
                .is_empty(),
            "search over loaded sidecars returns results"
        );
    }

    #[test]
    fn snapshot_index_queries_sidecars_without_opening_segment_payloads() {
        let dir = MemoryDirectory::arc();
        {
            let mut store =
                UpdatableIndex::open(dir.clone(), 2, 2, IndexParams::default()).unwrap();
            store
                .add(0, AxisBox::new(vec![0.0, 0.0], vec![10.0, 10.0]))
                .unwrap();
            store
                .add(1, AxisBox::new(vec![1.0, 1.0], vec![2.0, 2.0]))
                .unwrap();
            store
                .add(2, AxisBox::new(vec![8.0, 8.0], vec![12.0, 12.0]))
                .unwrap();
            store
                .add(3, AxisBox::new(vec![20.0, 20.0], vec![21.0, 21.0]))
                .unwrap();
            store.checkpoint().unwrap();
        }

        let (watched, opened) = RecordingDirectory::wrap(dir);
        let snapshot = SnapshotIndex::open(watched, 2, IndexParams::default()).unwrap();
        assert_eq!(snapshot.segment_count(), 2);
        assert_eq!(snapshot.tombstone_count(), 0);

        let params = || SearchParams {
            ef: 100,
            overretrieve: 100,
        };
        let inner = AxisBox::new(vec![1.25, 1.25], vec![1.75, 1.75]);
        let overlap = AxisBox::new(vec![9.0, 9.0], vec![11.0, 11.0]);

        assert_eq!(snapshot.containing(&[5.0, 5.0], params()).unwrap(), vec![0]);
        assert_eq!(snapshot.subsumers(&inner, params()).unwrap(), vec![0, 1]);
        assert_eq!(snapshot.subsumees(&b(0.0, 10.0)).unwrap(), vec![0, 1]);
        assert_eq!(
            snapshot.overlapping(&overlap, params()).unwrap(),
            vec![0, 2]
        );
        assert_eq!(
            snapshot.containing_exhaustive(&[5.0, 5.0]).unwrap(),
            vec![0]
        );
        assert_eq!(snapshot.subsumers_exhaustive(&inner).unwrap(), vec![0, 1]);
        assert_eq!(
            snapshot.overlapping_exhaustive(&overlap).unwrap(),
            vec![0, 2]
        );
        assert_eq!(
            snapshot.nearest_region(&overlap, 1, params()).unwrap()[0].0,
            2
        );
        assert_eq!(snapshot.search(&[5.0, 5.0], 1, params()).unwrap()[0].0, 0);
        assert_eq!(
            snapshot
                .search_exhaustive(&[5.0, 5.0], 2)
                .unwrap()
                .into_iter()
                .map(|(id, _)| id)
                .collect::<Vec<_>>(),
            vec![0, 1]
        );
        let soft = snapshot.subsumers_soft(&overlap, 0.2, params()).unwrap();
        assert_eq!(
            soft.iter().map(|(id, _)| *id).collect::<Vec<_>>(),
            vec![2, 0]
        );

        let opened = opened.lock().unwrap().clone();
        assert!(
            opened.iter().any(|path| path.starts_with("segstore.idx.")),
            "snapshot should open persisted sidecars: {opened:?}"
        );
        assert!(
            !opened.iter().any(|path| path.starts_with("segstore.seg.")),
            "valid sidecars should avoid source segment payload reads: {opened:?}"
        );
    }

    #[test]
    fn snapshot_index_rebuilds_tombstoned_region_sidecar_before_search() {
        let dir = MemoryDirectory::arc();
        let (name, stale_sidecar) = checkpointed_store(dir.clone(), IndexParams::default());
        {
            let mut store =
                UpdatableIndex::open(dir.clone(), 4, 2, IndexParams::default()).unwrap();
            store.delete(0).unwrap();
            store.checkpoint().unwrap();
            store
                .inner
                .dir()
                .atomic_write(&name, &stale_sidecar)
                .unwrap();
        }

        let (watched, opened) = RecordingDirectory::wrap(dir);
        let snapshot = SnapshotIndex::open(watched, 2, IndexParams::default()).unwrap();
        assert_eq!(snapshot.tombstone_count(), 1);
        let hits = snapshot
            .search(&[0.3, 0.3], 3, SearchParams::default())
            .unwrap();
        assert!(
            !hits.iter().any(|(id, _)| *id == 0),
            "deleted id must not be served by a stale region sidecar"
        );
        assert!(!hits.is_empty(), "rebuilt segment should keep live hits");

        let opened = opened.lock().unwrap().clone();
        assert!(
            opened.iter().any(|path| path.starts_with("segstore.idx.")),
            "snapshot should inspect the stale sidecar first: {opened:?}"
        );
        assert!(
            opened.iter().any(|path| path.starts_with("segstore.seg.")),
            "stale region sidecars should be rebuilt from the source segment: {opened:?}"
        );
    }

    #[test]
    fn snapshot_index_rebuilds_missing_sidecar_from_one_segment() {
        let dir = MemoryDirectory::arc();
        let (name, _) = checkpointed_store(dir.clone(), IndexParams::default());
        dir.delete(&name).unwrap();

        let (watched, opened) = RecordingDirectory::wrap(dir.clone());
        let snapshot = SnapshotIndex::open(watched, 2, IndexParams::default()).unwrap();
        let hits = snapshot
            .search(&[0.3, 0.3], 3, SearchParams::default())
            .unwrap();
        assert!(
            !hits.is_empty(),
            "missing sidecar should rebuild enough index to search"
        );
        assert!(
            dir.exists(&name),
            "snapshot fallback should persist the rebuilt sidecar"
        );

        let opened = opened.lock().unwrap().clone();
        assert!(
            opened.iter().any(|path| path.starts_with("segstore.seg.")),
            "missing sidecar should fall back to one source segment read: {opened:?}"
        );
    }

    #[test]
    fn snapshot_index_rebuilds_sidecar_with_wrong_segment_id() {
        let dir = MemoryDirectory::arc();
        {
            let mut store =
                UpdatableIndex::open(dir.clone(), 2, 2, IndexParams::default()).unwrap();
            store.add(0, b(0.0, 1.0)).unwrap();
            store.add(1, b(0.2, 1.2)).unwrap();
            store.add(2, b(5.0, 6.0)).unwrap();
            store.add(3, b(8.0, 9.0)).unwrap();
            store.checkpoint().unwrap();

            let ids = store.inner.segment_ids();
            assert_eq!(ids.len(), 2, "test setup should create two segments");
            let first = read_file(
                store.inner.dir(),
                &store.inner.index_name(ids[0], INDEX_KIND),
            );
            store
                .inner
                .dir()
                .atomic_write(&store.inner.index_name(ids[1], INDEX_KIND), &first)
                .unwrap();
        }

        let (watched, opened) = RecordingDirectory::wrap(dir);
        let snapshot = SnapshotIndex::open(watched, 2, IndexParams::default()).unwrap();
        let hits = snapshot
            .search(&[5.5, 5.5], 2, SearchParams::default())
            .unwrap();
        assert!(
            hits.iter().any(|(id, _)| *id == 2),
            "segment 1 should be rebuilt and searched after rejecting the copied sidecar: {hits:?}"
        );

        let opened = opened.lock().unwrap().clone();
        assert!(
            opened.iter().any(|path| path == "segstore.seg.1"),
            "wrong-segment sidecar should fall back to that source segment: {opened:?}"
        );
    }

    #[test]
    fn compact_persists_sidecar_and_prunes_cached_indexes() {
        let dir = MemoryDirectory::arc();
        let mut store = UpdatableIndex::open(dir, 2, 2, IndexParams::default()).unwrap();
        store.add(0, b(0.0, 1.0)).unwrap();
        store.add(1, b(0.2, 1.2)).unwrap();
        store.add(2, b(5.0, 6.0)).unwrap();
        store.add(3, b(8.0, 9.0)).unwrap();

        let before_ids = store.inner.segment_ids().to_vec();
        assert!(
            before_ids.len() >= 2,
            "test setup should create multiple sealed segments"
        );
        let _ = store.search(&[0.5, 0.5], 2, SearchParams::default());
        assert_eq!(
            store.cache.borrow().by_segment_id.len(),
            before_ids.len(),
            "warm query should cache each sealed segment"
        );

        store.compact().unwrap();

        let after_ids = store.inner.segment_ids().to_vec();
        assert_eq!(
            after_ids.len(),
            1,
            "compact should merge the sealed segments"
        );
        assert!(
            store
                .inner
                .dir()
                .exists(&store.inner.index_name(after_ids[0], INDEX_KIND)),
            "merged segment should have a sidecar immediately after compact"
        );
        assert!(
            store
                .cache
                .borrow()
                .by_segment_id
                .keys()
                .all(|id| after_ids.contains(id)),
            "cache should not retain indexes for compacted-away segment ids"
        );
    }

    #[test]
    fn region_sidecar_recipe_mismatch_rebuilds() {
        let dir = MemoryDirectory::arc();
        let original = IndexParams::default();
        let (name, before) = checkpointed_store(dir.clone(), original);
        assert_eq!(
            &before[..SIDECAR_MAGIC.len()],
            SIDECAR_MAGIC,
            "new sidecars carry the precinct region-index envelope"
        );

        let changed = IndexParams {
            m: 8,
            m_max: 16,
            ..IndexParams::default()
        };
        let store = UpdatableIndex::open(dir.clone(), 4, 2, changed).unwrap();
        let seg_id = store.inner.segment_ids()[0];
        assert!(
            store
                .load_sidecar(&store.inner.segments()[0][..], seg_id)
                .is_none(),
            "sidecar built with default HNSW params must not load under changed params"
        );
        assert!(
            !store
                .search(&[0.3, 0.3], 1, SearchParams::default())
                .is_empty(),
            "mismatched sidecar falls back to rebuild"
        );

        let after = read_file(store.inner.dir(), &name);
        assert_ne!(before, after, "rebuild overwrites the stale-recipe sidecar");
        assert!(
            store
                .load_sidecar(&store.inner.segments()[0][..], seg_id)
                .is_some(),
            "rebuilt sidecar now matches the current recipe"
        );
    }

    #[test]
    fn region_sidecar_envelope_rejects_corrupt_headers() {
        let store =
            UpdatableIndex::open(MemoryDirectory::arc(), 4, 2, IndexParams::default()).unwrap();
        let index = b"index-bytes";
        let seg_id = 7;
        let bytes = store.encode_sidecar(index, seg_id).unwrap();
        assert_eq!(store.decode_sidecar(&bytes, seg_id), Some(index.as_slice()));

        assert!(store.decode_sidecar(&bytes[..8], seg_id).is_none());

        let mut bad_magic = bytes.clone();
        bad_magic[0] ^= 0xFF;
        assert!(store.decode_sidecar(&bad_magic, seg_id).is_none());

        let mut bad_version = bytes.clone();
        bad_version[8..12].copy_from_slice(&(SIDECAR_VERSION + 1).to_le_bytes());
        assert!(store.decode_sidecar(&bad_version, seg_id).is_none());

        assert!(
            store.decode_sidecar(&bytes, seg_id + 1).is_none(),
            "sidecar must not load for a different segment id"
        );

        let mut bad_recipe_len = bytes.clone();
        bad_recipe_len[20..24].copy_from_slice(&u32::MAX.to_le_bytes());
        assert!(store.decode_sidecar(&bad_recipe_len, seg_id).is_none());

        let mut bad_recipe = bytes.clone();
        bad_recipe[24] ^= 0x01;
        assert!(store.decode_sidecar(&bad_recipe, seg_id).is_none());
    }

    #[test]
    fn region_sidecar_invalid_payload_rebuilds() {
        let dir = MemoryDirectory::arc();
        let (name, _) = checkpointed_store(dir.clone(), IndexParams::default());
        {
            let store = UpdatableIndex::open(dir.clone(), 4, 2, IndexParams::default()).unwrap();
            let seg_id = store.inner.segment_ids()[0];
            let corrupt = store
                .encode_sidecar(b"not-a-postcard-region-index", seg_id)
                .unwrap();
            store.inner.dir().atomic_write(&name, &corrupt).unwrap();
        }

        let store = UpdatableIndex::open(dir.clone(), 4, 2, IndexParams::default()).unwrap();
        let seg_id = store.inner.segment_ids()[0];
        assert!(
            store
                .load_sidecar(&store.inner.segments()[0][..], seg_id)
                .is_none(),
            "valid envelope with invalid region-index bytes is rejected"
        );
        assert!(
            !store
                .search(&[0.3, 0.3], 1, SearchParams::default())
                .is_empty(),
            "invalid payload falls back to rebuild"
        );
        assert!(
            store
                .load_sidecar(&store.inner.segments()[0][..], seg_id)
                .is_some(),
            "rebuilt sidecar loads after the fallback"
        );
    }

    #[test]
    fn deleted_id_does_not_resurface_through_a_sidecar() {
        let dir = MemoryDirectory::arc();
        {
            let mut store =
                UpdatableIndex::open(dir.clone(), 2, 2, IndexParams::default()).unwrap();
            store.add(0, b(0.0, 1.0)).unwrap();
            store.add(1, b(0.2, 1.2)).unwrap();
            store.add(2, b(5.0, 6.0)).unwrap();
            store.checkpoint().unwrap();
            store.delete(0).unwrap();
            store.checkpoint().unwrap();
        }

        let store = UpdatableIndex::open(dir, 2, 2, IndexParams::default()).unwrap();
        let top: Vec<u32> = store
            .search(&[0.5, 0.5], 3, SearchParams::default())
            .into_iter()
            .map(|(id, _)| id)
            .collect();
        assert!(
            !top.contains(&0),
            "deleted id 0 must not resurface from a persisted sidecar"
        );
        assert!(
            top.contains(&1),
            "nearest live box should remain searchable"
        );
    }

    #[test]
    fn checkpoint_after_replayed_delete_rewrites_stale_sidecar() {
        let dir = MemoryDirectory::arc();
        let (name, stale_bytes) = {
            let mut store =
                UpdatableIndex::open(dir.clone(), 2, 2, IndexParams::default()).unwrap();
            store.add(0, b(0.0, 1.0)).unwrap();
            store.add(1, b(0.2, 1.2)).unwrap();
            store.add(2, b(5.0, 6.0)).unwrap();
            store.checkpoint().unwrap();

            let seg_id = store.inner.segment_ids()[0];
            let name = store.inner.index_name(seg_id, INDEX_KIND);
            let bytes = read_file(store.inner.dir(), &name);

            // Simulate a crash after the delete is durably logged but before
            // `UpdatableIndex::delete` removes the now-stale sidecar.
            store.inner.delete(0).unwrap();
            (name, bytes)
        };

        let mut store = UpdatableIndex::open(dir.clone(), 2, 2, IndexParams::default()).unwrap();
        let seg_id = store.inner.segment_ids()[0];
        assert!(
            store
                .load_sidecar(&store.inner.segments()[0][..], seg_id)
                .is_none(),
            "replayed tombstone must make the old sidecar stale"
        );

        store.checkpoint().unwrap();

        let rewritten = read_file(&dir, &name);
        assert_ne!(
            rewritten, stale_bytes,
            "checkpoint should rewrite stale sidecars even before search"
        );
        let idx = store
            .load_sidecar(&store.inner.segments()[0][..], seg_id)
            .expect("rewritten sidecar should be valid");
        assert!(
            !idx.ids().contains(&0),
            "rewritten sidecar must exclude the replayed delete"
        );
        assert!(
            idx.ids().contains(&1),
            "rewritten sidecar should keep live ids from the segment"
        );
    }
}