semlith 0.41.0

//! The vector side of a store: read from disk only when something needs it,
//! and in pieces small enough to put down again.
//!
//! A store's quantized vectors are the largest thing it owns, and turbovec
//! keeps two copies of them resident once a search has warmed the index: the
//! packed codes it loaded, and the repacked block layout its SIMD kernel scans.
//! That cost is worth paying to answer a query and worth nothing at all to
//! answer `stats`, list files, or sit in an agent's MCP server waiting to be
//! asked something. So nothing here touches the disk until a caller asks for
//! something only the vectors can answer.
//!
//! # Two layouts
//!
//! [`Single`] is one `index.tv` holding everything, which is what every store
//! written before 0.7.0 has and keeps. Opening one is unchanged, searching one
//! is unchanged, and this release does not migrate it: the vectors in it are
//! quantized and cannot be split back out, so the only honest way onto the new
//! layout is to index again.
//!
//! [`Sharded`] is a directory of fixed-size shards, which is what a store
//! created by 0.7.0 gets, and what `format_version` 2 records. Three things
//! follow from it and none of them are possible with one file:
//!
//! - Searching holds a bounded number of shards, not the whole corpus.
//! - Changing one file rewrites one shard, not the entire index — which is what
//!   `semlith watch` does on every save of every watched file.
//! - A long index run can make its work durable as it goes, so an interruption
//!   costs the last shard rather than the whole run.
//!
//! # The shard directory is the manifest
//!
//! A shard is named for the first chunk id it holds — `index/0000000000000001.tvim`
//! — zero-padded so that sorted by name is sorted by id. Reading the directory
//! is therefore reading the map: shard boundaries are the names, and the shard
//! an id belongs to is the last one whose name does not exceed it. Nothing else
//! records the layout, so nothing else can disagree with it.
//!
//! This works because chunk ids only ever ascend. `chunks.id` is
//! `AUTOINCREMENT` precisely so that SQLite cannot hand a deleted row's id to a
//! new chunk, which would put one id inside two shards' ranges at once.

use anyhow::{Context, Result, bail};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use turbovec::IdMapIndex;

/// Vectors per shard.
///
/// A shard is the unit of everything here: what a search loads, what a save
/// rewrites, and what an interrupted index run keeps. 65536 vectors of a
/// 384-dimensional model at 4 bits is around 12 MB of packed codes, doubled by
/// turbovec's repacked search copy — small enough that a handful fit in a
/// modest budget, large enough that a corpus of a few hundred thousand chunks
/// is a handful and not a thousand.
pub const SHARD_VECTORS: usize = 65536;

/// Override for [`SHARD_VECTORS`].
///
/// This exists so the tests can build a multi-shard store without embedding
/// hundreds of thousands of chunks, and so the size can be measured rather than
/// argued about. It is not part of the documented environment.
const SHARD_VECTORS_ENV: &str = "SEMLITH_SHARD_VECTORS";

/// Where a sharded store keeps its shards, under the store directory.
const SHARD_DIR: &str = "index";

const SHARD_EXT: &str = "tvim";

fn shard_capacity() -> usize {
    std::env::var(SHARD_VECTORS_ENV)
        .ok()
        .and_then(|v| v.parse().ok())
        .filter(|n| *n > 0)
        .unwrap_or(SHARD_VECTORS)
}

/// How much of a store's vectors may be resident at once, in megabytes.
///
/// 512 MB is chosen for the machines this runs on: comfortable on the 16 GB
/// laptop most of this is developed against, and survivable on the 8 GB one
/// product.yaml names. It bounds the vectors only — the embedding model, SQLite
/// and the process itself are on top of it.
pub const INDEX_MEMORY_MB: usize = 512;

/// Environment override for [`INDEX_MEMORY_MB`], in megabytes.
pub const INDEX_MEMORY_ENV: &str = "SEMLITH_INDEX_MEMORY";

/// The resident budget in force, in megabytes.
pub fn budget_mb() -> usize {
    index_budget_bytes() / 1024 / 1024
}

/// The `MiB per store` the daemon has in force, or 0 outside one.
static BUDGET_IN_FORCE: AtomicUsize = AtomicUsize::new(0);

/// Moves every time the value in force changes, so an open index can tell
/// cheaply whether it has to re-derive its shard allowance.
static BUDGET_GENERATION: AtomicU64 = AtomicU64::new(0);

/// Set the budget every open index lives within. Applied by each index the
/// next time it is asked, and at once by the readers the settings route holds.
pub fn set_budget_mb(mb: usize) {
    if BUDGET_IN_FORCE.swap(mb, Ordering::Relaxed) != mb {
        BUDGET_GENERATION.fetch_add(1, Ordering::Relaxed);
    }
}

pub fn budget_generation() -> u64 {
    BUDGET_GENERATION.load(Ordering::Relaxed)
}

fn index_budget_bytes() -> usize {
    // The environment first, as before: an exported variable is the machine's
    // owner speaking and outranks what was typed into a page.
    std::env::var(INDEX_MEMORY_ENV)
        .ok()
        .and_then(|v| v.parse::<usize>().ok())
        .filter(|n| *n > 0)
        .or_else(|| Some(BUDGET_IN_FORCE.load(Ordering::Relaxed)).filter(|n| *n > 0))
        .unwrap_or(INDEX_MEMORY_MB)
        * 1024
        * 1024
}

/// What `n` vectors cost in memory once turbovec has them ready to search.
///
/// Counted from what the crate actually holds, not from the file size: the
/// packed codes it loaded, the repacked block layout its SIMD kernel scans
/// (`prepare` builds a second copy), a scale per vector, the slot-to-id table,
/// and the hash map back the other way. Those last three are why a 4-bit
/// 384-dimensional vector costs well over the 192 bytes its codes suggest.
pub fn resident_bytes(n: usize, dim: usize, bit_width: usize) -> usize {
    let codes = dim * bit_width / 8;
    // codes + repacked copy + f32 scale + u64 slot_to_id + a hashbrown entry
    n * (2 * codes + 4 + 8 + 24)
}

/// What the vector index is allowed to look at for one query.
#[derive(Clone)]
pub enum Allowlist {
    /// No filter, or one that selects everything the index holds.
    All,
    /// The filter selects nothing. The query is over before it starts.
    Empty,
    /// Ascending ids, so a sharded index takes each shard's part by binary
    /// search instead of testing every id against every shard on every query.
    /// Built with [`Allowlist::subset`].
    Subset(Vec<u64>),
}

impl Allowlist {
    pub fn subset(mut ids: Vec<u64>) -> Self {
        ids.sort_unstable();
        ids.dedup();
        Allowlist::Subset(ids)
    }
}

/// A store's vectors, in whichever layout the store was written with.
pub enum VectorIndex {
    /// One file. Every store written before 0.7.0. Boxed because a whole
    /// turbovec index sits inside it and the sharded variant is a handful of
    /// pointers.
    Single(Box<Single>),
    /// A directory of shards. Every store created by 0.7.0 and later.
    Sharded(Sharded),
}

impl VectorIndex {
    /// Name the index without reading it. The only I/O is listing the shard
    /// directory, which is what tells the two layouts apart.
    ///
    /// What is on disk decides, not what the store says about itself: shards
    /// present means shards, an `index.tv` means one file, and `prefer_sharded`
    /// — the store's `format_version` — only settles the case where a store has
    /// no vectors yet and one of the two is about to be written. The format key
    /// is how a store warns an older binary off; the directory is how this one
    /// knows what it is reading.
    pub fn open(dir: &Path, dim: usize, bit_width: usize, prefer_sharded: bool) -> Result<Self> {
        let sharded = Sharded::open(dir, dim, bit_width)?;
        if !sharded.shards.is_empty() {
            return Ok(VectorIndex::Sharded(sharded));
        }
        let single = Single::open(dir, dim, bit_width);
        if single.path.exists() || !prefer_sharded {
            return Ok(VectorIndex::Single(Box::new(single)));
        }
        Ok(VectorIndex::Sharded(sharded))
    }

    /// Is there an index on disk at all?
    pub fn exists(&self) -> bool {
        match self {
            VectorIndex::Single(s) => s.path.exists(),
            VectorIndex::Sharded(s) => !s.shards.is_empty(),
        }
    }

    /// True while any vectors are in memory.
    pub fn is_resident(&self) -> bool {
        match self {
            VectorIndex::Single(s) => s.resident.is_some(),
            VectorIndex::Sharded(s) => s.shards.iter().any(|s| s.resident.is_some()),
        }
    }

    /// How many vectors are resident, or `None` when nothing is loaded.
    ///
    /// A partially loaded sharded index answers `None` rather than a number
    /// covering some of itself: a count that means "the shards I happen to have
    /// read" is worse than no count, and the caller has SQLite for the honest
    /// one.
    pub fn resident_len(&self) -> Option<usize> {
        match self {
            VectorIndex::Single(s) => s.resident.as_ref().map(|i| i.len()),
            VectorIndex::Sharded(s) => s.fully_resident_len(),
        }
    }

    /// Forget what is loaded, so the next use reads from disk again.
    pub fn evict(&mut self) {
        match self {
            VectorIndex::Single(s) => s.resident = None,
            VectorIndex::Sharded(s) => s.evict_all(),
        }
    }

    /// Load and pay turbovec's repack cost now, so the first query is not
    /// slower than the rest. A sharded index warms only what its budget allows.
    pub fn prepare(&mut self) -> Result<()> {
        match self {
            VectorIndex::Single(s) => {
                s.get()?.prepare();
            }
            VectorIndex::Sharded(s) => s.prepare()?,
        }
        Ok(())
    }

    pub fn add(&mut self, vectors: &[f32], ids: &[u64]) -> Result<()> {
        match self {
            VectorIndex::Single(s) => s
                .get()?
                .add_with_ids(vectors, ids)
                .map_err(|e| anyhow::anyhow!("{e:?}")),
            VectorIndex::Sharded(s) => s.add(vectors, ids),
        }
    }

    pub fn remove(&mut self, id: u64) -> Result<()> {
        match self {
            VectorIndex::Single(s) => {
                s.get()?.remove(id);
            }
            VectorIndex::Sharded(s) => s.remove(id)?,
        }
        Ok(())
    }

    pub fn contains(&mut self, id: u64) -> Result<bool> {
        match self {
            VectorIndex::Single(s) => Ok(s.get()?.contains(id)),
            VectorIndex::Sharded(s) => s.contains(id),
        }
    }

    /// `(scores, ids)` for the best `depth` vectors `allowlist` permits.
    pub fn search(
        &mut self,
        vector: &[f32],
        depth: usize,
        allowlist: &Allowlist,
    ) -> Result<(Vec<f32>, Vec<u64>)> {
        if matches!(allowlist, Allowlist::Empty) || depth == 0 {
            return Ok((Vec::new(), Vec::new()));
        }
        match self {
            VectorIndex::Single(s) => {
                let index = s.get()?;
                if index.is_empty() {
                    return Ok((Vec::new(), Vec::new()));
                }
                let list = match allowlist {
                    Allowlist::Subset(ids) => Some(ids.as_slice()),
                    _ => None,
                };
                index
                    .search_with_allowlist(vector, depth, list)
                    .map_err(|e| anyhow::anyhow!("searching the vector index: {e}"))
            }
            VectorIndex::Sharded(s) => s.search(vector, depth, allowlist),
        }
    }

    /// Make the index durable. Only what changed is written.
    pub fn save(&mut self) -> Result<()> {
        match self {
            VectorIndex::Single(s) => s.save(),
            VectorIndex::Sharded(s) => s.save(),
        }
    }

    /// How many shards may be resident at once, or `None` for a single-file
    /// index, which has no choice about it.
    pub fn max_resident(&self) -> Option<usize> {
        match self {
            VectorIndex::Single(_) => None,
            VectorIndex::Sharded(s) => Some(s.max_resident()),
        }
    }

    /// Re-derive how many shards may be resident from the budget in force,
    /// and put down whatever is now over it. A shard with unwritten changes is
    /// kept, as it always is.
    pub fn apply_budget(&mut self) {
        if let VectorIndex::Sharded(s) = self {
            let full = resident_bytes(s.capacity, s.dim, s.bit_width).max(1);
            s.max_resident = (index_budget_bytes() / full).max(1);
            s.shed();
        }
    }

    /// How many shards are in memory now.
    pub fn resident_shards(&self) -> usize {
        match self {
            VectorIndex::Single(s) => usize::from(s.resident.is_some()),
            VectorIndex::Sharded(s) => s.shards.iter().filter(|s| s.resident.is_some()).count(),
        }
    }

    /// Shards put down to stay inside the budget since this store was opened.
    pub fn evictions(&self) -> u64 {
        match self {
            VectorIndex::Single(_) => 0,
            VectorIndex::Sharded(s) => s.evictions(),
        }
    }

    /// Vectors per shard, or `None` for a single-file index.
    pub fn capacity(&self) -> Option<usize> {
        match self {
            VectorIndex::Single(_) => None,
            VectorIndex::Sharded(s) => Some(s.capacity),
        }
    }

    /// A test's small shards, so a multi-shard store takes a dozen vectors.
    #[cfg(test)]
    pub(crate) fn set_capacity(&mut self, capacity: usize) {
        if let VectorIndex::Sharded(s) = self {
            s.capacity = capacity;
        }
    }

    /// How many shards the store holds. One, for a single-file index.
    pub fn shards(&self) -> usize {
        match self {
            VectorIndex::Single(s) => usize::from(s.path.exists()),
            VectorIndex::Sharded(s) => s.shards.len(),
        }
    }

    /// Remove the leavings of a run killed mid-save. The caller holds the store
    /// lock, so there is no live writer whose half-written index this could be.
    pub fn clean(&self) {
        match self {
            VectorIndex::Single(s) => {
                let _ = std::fs::remove_file(s.path.with_extension("tv.tmp"));
            }
            VectorIndex::Sharded(s) => s.clean(),
        }
    }
}

/// One `index.tv` holding every vector — the layout of every store written
/// before 0.7.0, kept exactly as it was.
pub struct Single {
    path: PathBuf,
    dim: usize,
    bit_width: usize,
    resident: Option<IdMapIndex>,
}

impl Single {
    fn open(dir: &Path, dim: usize, bit_width: usize) -> Self {
        Self {
            path: dir.join("index.tv"),
            dim,
            bit_width,
            resident: None,
        }
    }

    /// The resident index, reading the file on first use.
    fn get(&mut self) -> Result<&mut IdMapIndex> {
        if self.resident.is_none() {
            self.resident = Some(load_or_new(&self.path, self.dim, self.bit_width)?);
        }
        Ok(self.resident.as_mut().unwrap())
    }

    fn save(&mut self) -> Result<()> {
        let tmp = self.path.with_extension("tv.tmp");
        let path = self.path.clone();
        self.get()?
            .write(&tmp)
            .with_context(|| format!("writing {}", tmp.display()))?;
        std::fs::rename(&tmp, &path)?;
        crate::home::tighten_file(&path);
        Ok(())
    }
}

/// One shard: a slice of the id space, and whatever of it is resident.
struct Shard {
    /// First chunk id this shard may hold. Its range ends where the next
    /// shard's begins; the last shard's range is open.
    start_id: u64,
    path: PathBuf,
    resident: Option<IdMapIndex>,
    /// Changed since it was last written.
    dirty: bool,
    /// Reading of the index's clock when this shard was last touched. The
    /// least of these is what gets put down when the budget is reached.
    last_used: u64,
}

/// A directory of shards — the layout of a store created by 0.7.0.
pub struct Sharded {
    dir: PathBuf,
    dim: usize,
    bit_width: usize,
    capacity: usize,
    /// Ascending by `start_id`, which is also ascending by file name.
    shards: Vec<Shard>,
    /// How many shards may be resident at once. Derived from the memory budget
    /// and what a full shard costs, and never less than one — a store must be
    /// searchable on any budget, even a silly one.
    max_resident: usize,
    /// Ticks on every shard access, so the smallest `last_used` is the coldest
    /// shard. A counter rather than a clock: it cannot go backwards, and two
    /// accesses in the same microsecond still order.
    clock: u64,
    /// Shards put down to stay inside the budget. Worth reporting: it is the
    /// signal that a store has outgrown its budget and every query is paying to
    /// read shards back.
    evictions: u64,
}

impl Sharded {
    fn open(store_dir: &Path, dim: usize, bit_width: usize) -> Result<Self> {
        let mut dir = store_dir.join(SHARD_DIR);
        // Between the two renames of a compaction's swap, or after a crash
        // there, the complete new set is the only one on disk. A reader takes
        // it; the next writer's `recover` gives it its proper name.
        if !dir.is_dir() && store_dir.join(COMPACT_DIR).is_dir() {
            dir = store_dir.join(COMPACT_DIR);
        }
        let mut shards = Vec::new();
        if dir.is_dir() {
            for entry in std::fs::read_dir(&dir)
                .with_context(|| format!("reading {}", dir.display()))?
                .flatten()
            {
                let path = entry.path();
                if path.extension().and_then(|e| e.to_str()) != Some(SHARD_EXT) {
                    continue;
                }
                let Some(start_id) = path
                    .file_stem()
                    .and_then(|s| s.to_str())
                    .and_then(|s| s.parse::<u64>().ok())
                else {
                    // Not one of ours. Leaving it alone is the only safe thing
                    // to do with a file we cannot account for.
                    continue;
                };
                shards.push(Shard {
                    start_id,
                    path,
                    resident: None,
                    dirty: false,
                    last_used: 0,
                });
            }
        }
        shards.sort_by_key(|s| s.start_id);
        let capacity = shard_capacity();
        let full = resident_bytes(capacity, dim, bit_width).max(1);
        Ok(Self {
            dir,
            dim,
            bit_width,
            capacity,
            shards,
            max_resident: (index_budget_bytes() / full).max(1),
            clock: 0,
            evictions: 0,
        })
    }

    /// How many shards this store may hold in memory at once.
    pub fn max_resident(&self) -> usize {
        self.max_resident
    }

    /// Shards put down to stay inside the budget since this store was opened.
    pub fn evictions(&self) -> u64 {
        self.evictions
    }

    fn shard_path(&self, start_id: u64) -> PathBuf {
        self.dir.join(format!("{start_id:016}.{SHARD_EXT}"))
    }

    /// The shard whose range covers `id`, if any shard does.
    fn locate(&self, id: u64) -> Option<usize> {
        match self.shards.partition_point(|s| s.start_id <= id) {
            0 => None,
            n => Some(n - 1),
        }
    }

    fn load(&mut self, at: usize) -> Result<&mut IdMapIndex> {
        self.clock += 1;
        self.shards[at].last_used = self.clock;
        if self.shards[at].resident.is_none() {
            self.make_room(at);
            let loaded = load_or_new(&self.shards[at].path, self.dim, self.bit_width)?;
            self.shards[at].resident = Some(loaded);
        }
        Ok(self.shards[at].resident.as_mut().unwrap())
    }

    /// Put down the coldest shards until one more fits inside the budget.
    ///
    /// A shard with unwritten changes is never a candidate: its vectors exist
    /// nowhere else yet, so dropping it would lose them. That is why an index
    /// run saves as it goes rather than holding every shard it has touched.
    fn make_room(&mut self, keep: usize) {
        loop {
            let resident: Vec<usize> = (0..self.shards.len())
                .filter(|i| self.shards[*i].resident.is_some())
                .collect();
            if resident.len() < self.max_resident {
                return;
            }
            let Some(&coldest) = resident
                .iter()
                .filter(|i| **i != keep && !self.shards[**i].dirty)
                .min_by_key(|i| self.shards[**i].last_used)
            else {
                // Everything resident is dirty or is the shard being asked for.
                // Exceeding the budget beats losing vectors or refusing to
                // answer; the count says it happened.
                return;
            };
            self.shards[coldest].resident = None;
            self.evictions += 1;
        }
    }

    /// Put down the coldest clean shards until no more than the allowance
    /// is resident.
    fn shed(&mut self) {
        loop {
            let resident: Vec<usize> = (0..self.shards.len())
                .filter(|i| self.shards[*i].resident.is_some())
                .collect();
            if resident.len() <= self.max_resident {
                return;
            }
            let Some(&coldest) = resident
                .iter()
                .filter(|i| !self.shards[**i].dirty)
                .min_by_key(|i| self.shards[**i].last_used)
            else {
                return;
            };
            self.shards[coldest].resident = None;
            self.evictions += 1;
        }
    }

    fn evict_all(&mut self) {
        for shard in &mut self.shards {
            // A shard with unwritten changes is the one thing that must not be
            // dropped: its vectors exist nowhere else.
            if !shard.dirty {
                shard.resident = None;
            }
        }
    }

    /// Resident count, but only when every shard is loaded — see
    /// [`VectorIndex::resident_len`].
    fn fully_resident_len(&self) -> Option<usize> {
        if self.shards.is_empty() || self.shards.iter().any(|s| s.resident.is_none()) {
            return None;
        }
        Some(
            self.shards
                .iter()
                .map(|s| s.resident.as_ref().map_or(0, |i| i.len()))
                .sum(),
        )
    }

    /// Warm what the budget allows, newest first.
    ///
    /// Warming every shard of a store larger than its budget would evict as
    /// fast as it loaded and leave the same shards cold, having read the whole
    /// corpus to achieve it.
    fn prepare(&mut self) -> Result<()> {
        let warm = self.max_resident.min(self.shards.len());
        for at in (self.shards.len() - warm)..self.shards.len() {
            self.load(at)?.prepare();
        }
        Ok(())
    }

    /// Append vectors, opening a new shard whenever the last one is full.
    ///
    /// Ids ascend, so appending always means the last shard. A batch that
    /// straddles a boundary is split rather than allowed to overflow: the
    /// capacity is what every memory claim downstream rests on.
    fn add(&mut self, vectors: &[f32], ids: &[u64]) -> Result<()> {
        if ids.is_empty() {
            return Ok(());
        }
        let mut offset = 0;
        while offset < ids.len() {
            let at = self.open_shard(ids[offset])?;
            let room = self.capacity.saturating_sub(self.load(at)?.len());
            let take = room.min(ids.len() - offset);
            let dim = self.dim;
            self.load(at)?
                .add_with_ids(
                    &vectors[offset * dim..(offset + take) * dim],
                    &ids[offset..offset + take],
                )
                .map_err(|e| anyhow::anyhow!("{e:?}"))?;
            self.shards[at].dirty = true;
            offset += take;
        }
        Ok(())
    }

    /// The shard the next id goes into, creating one when the last is full or
    /// there is none yet.
    fn open_shard(&mut self, next_id: u64) -> Result<usize> {
        if let Some(last) = self.shards.len().checked_sub(1)
            && self.load(last)?.len() < self.capacity
        {
            return Ok(last);
        }
        let path = self.shard_path(next_id);
        self.clock += 1;
        self.make_room(self.shards.len());
        self.shards.push(Shard {
            start_id: next_id,
            path,
            resident: Some(
                IdMapIndex::new(self.dim, self.bit_width).map_err(|e| anyhow::anyhow!("{e:?}"))?,
            ),
            dirty: true,
            last_used: self.clock,
        });
        Ok(self.shards.len() - 1)
    }

    fn remove(&mut self, id: u64) -> Result<()> {
        let Some(at) = self.locate(id) else {
            return Ok(());
        };
        if self.load(at)?.remove(id) {
            self.shards[at].dirty = true;
        }
        Ok(())
    }

    fn contains(&mut self, id: u64) -> Result<bool> {
        match self.locate(id) {
            Some(at) => Ok(self.load(at)?.contains(id)),
            None => Ok(false),
        }
    }

    /// Search every shard and merge the results into one ranking.
    ///
    /// The quantizer is data-oblivious, so a score from one shard means what it
    /// means in another and the merge is a sort rather than a normalisation.
    fn search(
        &mut self,
        vector: &[f32],
        depth: usize,
        allowlist: &Allowlist,
    ) -> Result<(Vec<f32>, Vec<u64>)> {
        let mut merged: Vec<(f32, u64)> = Vec::new();
        for at in 0..self.shards.len() {
            let start = self.shards[at].start_id;
            let end = self.shards.get(at + 1).map(|s| s.start_id);
            // Narrow the allowlist to this shard before loading it: a shard no
            // permitted id falls inside is a shard the query never reads.
            let subset = match allowlist {
                Allowlist::Empty => return Ok((Vec::new(), Vec::new())),
                Allowlist::All => None,
                Allowlist::Subset(ids) => {
                    // `ids` ascend, so this shard's part is one contiguous run.
                    let from = ids.partition_point(|id| *id < start);
                    let to = end.map_or(ids.len(), |e| ids.partition_point(|id| *id < e));
                    if from >= to {
                        continue;
                    }
                    Some(ids[from..to].to_vec())
                }
            };

            let index = self.load(at)?;
            if index.is_empty() {
                continue;
            }
            // A shard can be missing an id its range covers — a chunk deleted
            // since, or one whose vectors a killed run never made durable.
            // turbovec 1.0 reports that as an error rather than panicking, but
            // an allowlist entry the shard does not hold is normal here and
            // not something to fail a search over, so it is filtered out.
            let subset = subset.map(|ids| {
                ids.into_iter()
                    .filter(|id| index.contains(*id))
                    .collect::<Vec<u64>>()
            });
            if subset.as_ref().is_some_and(|ids| ids.is_empty()) {
                continue;
            }
            let (scores, ids) = index
                .search_with_allowlist(vector, depth, subset.as_deref())
                .map_err(|e| anyhow::anyhow!("searching a vector shard: {e}"))?;
            merged.extend(scores.into_iter().zip(ids));
        }

        merged.sort_by(|a, b| b.0.total_cmp(&a.0));
        merged.truncate(depth);
        Ok(merged.into_iter().unzip())
    }

    fn save(&mut self) -> Result<()> {
        if !self.shards.iter().any(|s| s.dirty) {
            return Ok(());
        }
        // Created 0700, not narrowed before it exists (audit S24).
        crate::home::secure_dir(&self.dir)?;
        for shard in &mut self.shards {
            if !shard.dirty {
                continue;
            }
            let Some(index) = shard.resident.as_ref() else {
                continue;
            };
            let tmp = shard.path.with_extension("tmp");
            index
                .write(&tmp)
                .with_context(|| format!("writing {}", tmp.display()))?;
            std::fs::rename(&tmp, &shard.path)?;
            // A shard holds the vectors of the corpus, so it is worth what the
            // corpus is worth. Applied after the rename, because the mode
            // travels with the file rather than with the name.
            crate::home::tighten_file(&shard.path);
            shard.dirty = false;
        }
        Ok(())
    }

    fn clean(&self) {
        let Ok(entries) = std::fs::read_dir(&self.dir) else {
            return;
        };
        for entry in entries.flatten() {
            if entry.path().extension().and_then(|e| e.to_str()) == Some("tmp") {
                let _ = std::fs::remove_file(entry.path());
            }
        }
    }
}

/// Where a compaction builds a store's replacement shards, beside the live set.
///
/// Not a name [`Sharded::open`] lists, so a half-built set is never searched;
/// it becomes `index/` by one rename once every shard in it is written.
const COMPACT_DIR: &str = "index.compact";

/// Where the shards a compaction replaced wait for removal, for the moment
/// between the two renames of the swap.
const REPLACED_DIR: &str = "index.old";

/// The shard files a store holds on disk, ascending by first id, or its single
/// `index.tv` for a store written before 0.7.0. Read nothing but the listing.
pub fn vector_files(store_dir: &Path) -> Result<Vec<PathBuf>> {
    let dir = store_dir.join(SHARD_DIR);
    let mut named: Vec<(u64, PathBuf)> = Vec::new();
    if dir.is_dir() {
        for entry in std::fs::read_dir(&dir)
            .with_context(|| format!("reading {}", dir.display()))?
            .flatten()
        {
            let path = entry.path();
            if path.extension().and_then(|e| e.to_str()) != Some(SHARD_EXT) {
                continue;
            }
            if let Some(start) = path
                .file_stem()
                .and_then(|s| s.to_str())
                .and_then(|s| s.parse::<u64>().ok())
            {
                named.push((start, path));
            }
        }
    }
    named.sort_by_key(|(start, _)| *start);
    let mut files: Vec<PathBuf> = named.into_iter().map(|(_, p)| p).collect();
    let single = store_dir.join("index.tv");
    if files.is_empty() && single.exists() {
        files.push(single);
    }
    Ok(files)
}

/// What a store's vector files take on disk.
pub fn vector_bytes(store_dir: &Path) -> u64 {
    vector_files(store_dir)
        .unwrap_or_default()
        .iter()
        .filter_map(|p| std::fs::metadata(p).ok())
        .map(|m| m.len())
        .sum()
}

/// Every id a store's vectors hold, ascending, read one file at a time so a
/// store larger than its budget is never resident whole.
pub fn ids_on_disk(store_dir: &Path, dim: usize, bit_width: usize) -> Result<Vec<u64>> {
    let mut ids = Vec::new();
    for path in vector_files(store_dir)? {
        ids.extend(load_or_new(&path, dim, bit_width)?.iter_ids());
    }
    ids.sort_unstable();
    Ok(ids)
}

/// Build a packed shard set from `exact`'s records into [`COMPACT_DIR`].
///
/// Every shard but the last is full, and each is named for its first id, so
/// the set keeps the one rule the layout rests on: ranges ascend and new ids
/// land in the last shard. The codes are turbovec's encoding of the same f32
/// values the index was given when they were embedded, and semlith never
/// calibrates, so every vector's codes are the bytes it had before.
///
/// Returns how many shards were written. Nothing is swapped in here.
pub fn build_compacted(
    store_dir: &Path,
    exact: &Exact,
    dim: usize,
    bit_width: usize,
    capacity: usize,
) -> Result<usize> {
    let dir = store_dir.join(COMPACT_DIR);
    if dir.exists() {
        std::fs::remove_dir_all(&dir)?;
    }
    std::fs::create_dir_all(&dir).with_context(|| format!("creating {}", dir.display()))?;
    crate::home::tighten_dir(&dir);
    let mut written = 0;
    let mut shard: Option<(u64, IdMapIndex)> = None;
    let mut ids: Vec<u64> = Vec::new();
    let mut values: Vec<f32> = Vec::new();

    let write = |start: u64, index: &IdMapIndex| -> Result<()> {
        let path = dir.join(format!("{start:016}.{SHARD_EXT}"));
        index
            .write(&path)
            .with_context(|| format!("writing {}", path.display()))?;
        crate::home::tighten_file(&path);
        Ok(())
    };
    // Batched into the index rather than added a vector at a time: the codes
    // are the same however the rows are batched, and the encoder is not.
    let flush =
        |shard: &mut Option<(u64, IdMapIndex)>, ids: &mut Vec<u64>, values: &mut Vec<f32>| {
            if let Some((_, index)) = shard.as_mut()
                && !ids.is_empty()
            {
                index
                    .add_with_ids(values, ids)
                    .map_err(|e| anyhow::anyhow!("{e:?}"))?;
            }
            ids.clear();
            values.clear();
            anyhow::Ok(())
        };

    exact.for_each(|id, vector| {
        if shard.is_none() {
            shard = Some((
                id,
                IdMapIndex::new(dim, bit_width).map_err(|e| anyhow::anyhow!("{e:?}"))?,
            ));
        }
        ids.push(id);
        values.extend_from_slice(vector);
        let held = shard.as_ref().map_or(0, |(_, i)| i.len()) + ids.len();
        if held == capacity {
            flush(&mut shard, &mut ids, &mut values)?;
            let (start, index) = shard.take().unwrap();
            write(start, &index)?;
            written += 1;
        } else if ids.len() >= 4096 {
            flush(&mut shard, &mut ids, &mut values)?;
        }
        Ok(())
    })?;
    flush(&mut shard, &mut ids, &mut values)?;
    if let Some((start, index)) = shard.take() {
        write(start, &index)?;
        written += 1;
    }
    Ok(written)
}

/// Put the shard set [`build_compacted`] wrote in place of the live one.
///
/// Two renames: the live set aside, the new set in. A reader that lists the
/// directory between them sees no shards, which is why the caller brackets
/// this with the store's swap marker and generation, and why [`recover`]
/// finishes a swap a crash interrupted.
pub fn swap_compacted(store_dir: &Path) -> Result<()> {
    let live = store_dir.join(SHARD_DIR);
    let built = store_dir.join(COMPACT_DIR);
    let replaced = store_dir.join(REPLACED_DIR);
    if replaced.exists() {
        std::fs::remove_dir_all(&replaced)?;
    }
    if live.exists() {
        rename_retrying(&live, &replaced)?;
    }
    if let Err(e) = rename_retrying(&built, &live) {
        // Put the live set back rather than leave the store with none.
        let _ = rename_retrying(&replaced, &live);
        return Err(e);
    }
    let _ = std::fs::remove_dir_all(&replaced);
    Ok(())
}

/// A directory rename, retried for a moment: on Windows a reader that has a
/// shard file open for the length of one load makes the rename fail, and it
/// lets go on its own.
fn rename_retrying(from: &Path, to: &Path) -> Result<()> {
    let mut tries = 0;
    loop {
        match std::fs::rename(from, to) {
            Ok(()) => return Ok(()),
            Err(_) if tries < 20 => {
                tries += 1;
                std::thread::sleep(std::time::Duration::from_millis(50));
            }
            Err(e) => {
                return Err(anyhow::Error::new(e).context(format!(
                    "moving {} to {}",
                    from.display(),
                    to.display()
                )));
            }
        }
    }
}

/// Throw away what a compaction built and never swapped in.
pub fn discard_compacted(store_dir: &Path) -> Result<()> {
    let built = store_dir.join(COMPACT_DIR);
    if built.exists() {
        std::fs::remove_dir_all(&built)?;
    }
    let _ = std::fs::remove_file(store_dir.join(EXACT_COMPACT));
    Ok(())
}

/// Finish or undo what a compaction killed part-way left behind. The caller
/// holds the store lock, so no compaction is still running.
///
/// A built set with no live set beside it was killed between the two renames
/// of the swap, after every shard in it was written, so it is the store's
/// vectors and goes in. A built set beside a live one was never swapped and is
/// thrown away. Returns whether the shard directory changed, so the caller
/// knows to re-open its index.
pub fn recover(store_dir: &Path) -> Result<bool> {
    let live = store_dir.join(SHARD_DIR);
    let built = store_dir.join(COMPACT_DIR);
    let replaced = store_dir.join(REPLACED_DIR);
    let mut changed = false;
    if built.exists() {
        if live.exists() {
            std::fs::remove_dir_all(&built)?;
        } else {
            std::fs::rename(&built, &live)?;
            changed = true;
        }
    }
    if replaced.exists() {
        std::fs::remove_dir_all(&replaced)?;
    }
    let _ = std::fs::remove_file(store_dir.join(EXACT_COMPACT));
    Ok(changed)
}

/// Read an index file, or make an empty one when there is nothing there yet.
fn load_or_new(path: &Path, dim: usize, bit_width: usize) -> Result<IdMapIndex> {
    if !path.exists() {
        return IdMapIndex::new(dim, bit_width).map_err(|e| anyhow::anyhow!("{e:?}"));
    }
    let index = IdMapIndex::load(path).with_context(|| format!("loading {}", path.display()))?;
    if let Some(d) = index.dim_opt()
        && d != dim
    {
        bail!("index is {d}-dimensional but the store's model produces {dim}; store is corrupt");
    }
    Ok(index)
}

/// The full-precision copy of what the codes approximate.
///
/// turbovec keeps 4-bit codes and a repacked copy of them, and nothing in the
/// crate reconstructs a vector from either -- which is why 0.22.0 specified
/// full-precision rescoring, could not build it, and shipped the gap as a named
/// limitation. This is the missing half: every vector the index is given is
/// also appended here at f32, and the query path reads back only the handful it
/// is about to reorder.
///
/// One file of fixed-size records, `[id: u64 little-endian][dim f32 little-
/// endian]`, appended in the order the index sees them. Chunk ids come from an
/// `AUTOINCREMENT` column that SQLite never reissues, so the ids in the file
/// ascend, and a record is found by binary search over the record count rather
/// than by an index that would have to be kept in step with two other things.
///
/// A record whose chunk has since been deleted stays in the file. It costs its
/// bytes until the next full pass rewrites the store, and it can never be
/// returned: the only ids ever looked up are candidates the live index just
/// produced.
///
/// Every read verifies the id in the record it landed on. If a file's ids are
/// ever not ascending, the search misses and the caller keeps the code-based
/// score -- the failure is "no rescoring", never a vector belonging to another
/// chunk.
pub struct Exact {
    path: PathBuf,
    dim: usize,
}

/// Where a compaction writes the rewritten sidecar before it is renamed in.
const EXACT_COMPACT: &str = "exact.f32.compact";

/// What [`Exact::build_compacted`] kept and dropped.
#[derive(Debug, Default, Clone, Copy)]
pub struct ExactRewrite {
    pub kept: u64,
    pub dropped: u64,
    /// False when the file's ids did not ascend, which is the one layout a
    /// packed shard set cannot be built from.
    pub ascending: bool,
}

impl Exact {
    /// The sidecar beside a store's vectors, whether or not it exists yet.
    pub fn open(dir: &Path, dim: usize) -> Self {
        Self {
            path: dir.join("exact.f32"),
            dim,
        }
    }

    /// Does this store carry one? A store written before 0.23.0 does not, and
    /// is simply never rescored until its next full index pass.
    pub fn exists(&self) -> bool {
        self.path.exists()
    }

    pub fn bytes(&self) -> u64 {
        std::fs::metadata(&self.path).map(|m| m.len()).unwrap_or(0)
    }

    fn record_size(&self) -> usize {
        8 + self.dim * 4
    }

    /// What one vector costs in the sidecar, in bytes.
    pub fn record_bytes(&self) -> u64 {
        self.record_size() as u64
    }

    /// Append the vectors the index was just given, in the same order.
    ///
    /// `vectors` is flat, `dim` values per id, exactly as [`VectorIndex::add`]
    /// takes it.
    pub fn append(&self, vectors: &[f32], ids: &[u64]) -> Result<()> {
        if ids.is_empty() {
            return Ok(());
        }
        anyhow::ensure!(
            vectors.len() == ids.len() * self.dim,
            "{} values for {} vectors of {} dimensions",
            vectors.len(),
            ids.len(),
            self.dim
        );
        if let Some(parent) = self.path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        // The raw vectors are the corpus's meaning in numbers: 0600 wherever
        // the store lives (audit S24).
        let mut options = std::fs::OpenOptions::new();
        options.create(true).append(true);
        #[cfg(unix)]
        {
            use std::os::unix::fs::OpenOptionsExt;
            options.mode(crate::home::FILE_MODE);
        }
        let mut out = std::io::BufWriter::new(options.open(&self.path)?);
        crate::home::tighten_file(&self.path);
        use std::io::Write;
        for (slot, id) in ids.iter().enumerate() {
            out.write_all(&id.to_le_bytes())?;
            for value in &vectors[slot * self.dim..(slot + 1) * self.dim] {
                out.write_all(&value.to_le_bytes())?;
            }
        }
        out.flush()?;
        Ok(())
    }

    /// The vector stored for `id`, or `None` if this store has no sidecar, the
    /// id is not in it, or the file's ids are not ascending.
    pub fn get(&self, id: u64) -> Option<Vec<f32>> {
        let mut file = std::fs::File::open(&self.path).ok()?;
        let at = self.find(&mut file, id)?;
        self.read_at(&mut file, at).map(|(_, vector)| vector)
    }

    /// Overwrite the vectors stored for `ids` with zeros, keeping each record
    /// and its id where it is.
    ///
    /// A deleted chunk's record otherwise stays until a full pass rewrites the
    /// file, which is harmless for a chunk that was only replaced and is not
    /// for one refused because it held a secret: its vector is derived from
    /// that text. Zeros keep the file's layout and its binary search intact.
    pub fn zero(&self, ids: &[u64]) -> Result<()> {
        use std::io::{Seek, SeekFrom, Write};
        let Ok(mut file) = std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .open(&self.path)
        else {
            return Ok(());
        };
        let zeros = vec![0u8; self.dim * 4];
        for id in ids {
            if let Some(at) = self.find(&mut file, *id) {
                file.seek(SeekFrom::Start(at * self.record_size() as u64 + 8))?;
                file.write_all(&zeros)?;
            }
        }
        file.sync_data()?;
        Ok(())
    }

    /// The record number holding `id`, by binary search over the ascending ids.
    fn find(&self, file: &mut std::fs::File, id: u64) -> Option<u64> {
        let size = self.record_size() as u64;
        let records = file.metadata().ok()?.len() / size;
        if records == 0 {
            return None;
        }
        let (mut low, mut high) = (0u64, records - 1);
        while low <= high {
            let mid = low + (high - low) / 2;
            let (found, _) = self.read_at(file, mid)?;
            match found.cmp(&id) {
                std::cmp::Ordering::Equal => return Some(mid),
                std::cmp::Ordering::Less => low = mid + 1,
                std::cmp::Ordering::Greater => {
                    if mid == 0 {
                        return None;
                    }
                    high = mid - 1;
                }
            }
        }
        None
    }

    /// Every record in file order, streamed: `f(id, vector)`.
    pub fn for_each(&self, mut f: impl FnMut(u64, &[f32]) -> Result<()>) -> Result<()> {
        use std::io::Read;
        let Ok(file) = std::fs::File::open(&self.path) else {
            return Ok(());
        };
        let mut reader = std::io::BufReader::with_capacity(1 << 20, file);
        let mut buf = vec![0u8; self.record_size()];
        let mut vector = vec![0f32; self.dim];
        loop {
            match reader.read_exact(&mut buf) {
                Ok(()) => {}
                Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(()),
                Err(e) => return Err(e.into()),
            }
            let id = u64::from_le_bytes(buf[..8].try_into()?);
            for (slot, b) in buf[8..].as_chunks::<4>().0.iter().enumerate() {
                vector[slot] = f32::from_le_bytes(*b);
            }
            f(id, &vector)?;
        }
    }

    /// Rewrite the sidecar to the records whose id is in `keep` (ascending),
    /// beside the live file. Nothing replaces the live file here; see
    /// [`Exact::swap_compacted`]. Each kept record is copied byte for byte.
    pub fn build_compacted(&self, keep: &[u64]) -> Result<ExactRewrite> {
        use std::io::{Read, Write};
        let mut out = ExactRewrite {
            ascending: true,
            ..Default::default()
        };
        let Ok(file) = std::fs::File::open(&self.path) else {
            return Ok(out);
        };
        let target = self.path.with_file_name(EXACT_COMPACT);
        let mut writer = std::io::BufWriter::new(std::fs::File::create(&target)?);
        let mut reader = std::io::BufReader::with_capacity(1 << 20, file);
        let mut buf = vec![0u8; self.record_size()];
        let mut last: Option<u64> = None;
        loop {
            match reader.read_exact(&mut buf) {
                Ok(()) => {}
                Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => break,
                Err(e) => return Err(e.into()),
            }
            let id = u64::from_le_bytes(buf[..8].try_into()?);
            if last.is_some_and(|l| id <= l) {
                out.ascending = false;
            }
            last = Some(id);
            if keep.binary_search(&id).is_ok() {
                writer.write_all(&buf)?;
                out.kept += 1;
            } else {
                out.dropped += 1;
            }
        }
        writer.flush()?;
        writer.get_ref().sync_all()?;
        crate::home::tighten_file(&target);
        Ok(out)
    }

    /// The sidecar [`Exact::build_compacted`] wrote, as a view to read before
    /// it replaces the live one.
    pub fn compacted(&self) -> Exact {
        Exact {
            path: self.path.with_file_name(EXACT_COMPACT),
            dim: self.dim,
        }
    }

    /// Put the rewritten sidecar in place of the live one: one rename, so a
    /// reader opens either the old file or the new one, never half of each.
    pub fn swap_compacted(&self) -> Result<()> {
        std::fs::rename(self.path.with_file_name(EXACT_COMPACT), &self.path)?;
        crate::home::tighten_file(&self.path);
        Ok(())
    }

    fn read_at(&self, file: &mut std::fs::File, record: u64) -> Option<(u64, Vec<f32>)> {
        use std::io::{Read, Seek, SeekFrom};
        file.seek(SeekFrom::Start(record * self.record_size() as u64))
            .ok()?;
        let mut buf = vec![0u8; self.record_size()];
        file.read_exact(&mut buf).ok()?;
        let id = u64::from_le_bytes(buf[..8].try_into().ok()?);
        let vector = buf[8..]
            .as_chunks::<4>()
            .0
            .iter()
            .map(|b| f32::from_le_bytes(*b))
            .collect();
        Some((id, vector))
    }
}

/// Cosine similarity, on vectors the embedder already normalised.
///
/// The norms are computed anyway rather than assumed: a vector read back from
/// disk is data, and a model that ever stops normalising would otherwise turn
/// into a silent ranking change rather than a visible one.
pub fn cosine(a: &[f32], b: &[f32]) -> f32 {
    if a.len() != b.len() {
        return 0.0;
    }
    let mut dot = 0.0f32;
    let mut na = 0.0f32;
    let mut nb = 0.0f32;
    for (x, y) in a.iter().zip(b) {
        dot += x * y;
        na += x * x;
        nb += y * y;
    }
    if na == 0.0 || nb == 0.0 {
        return 0.0;
    }
    dot / (na.sqrt() * nb.sqrt())
}

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

    fn scratch(name: &str) -> PathBuf {
        let dir = std::env::temp_dir().join(format!("semlith-index-{name}-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        dir
    }

    /// Opening must not read the file. Proved with a file that cannot be read:
    /// if `open` touched it, `open` would be the thing that failed.
    #[test]
    fn opening_reads_nothing() {
        let dir = scratch("lazy");
        std::fs::write(dir.join("index.tv"), b"not a turbovec file at all").unwrap();

        let mut index = VectorIndex::open(&dir, 384, 4, false).unwrap();
        assert!(index.exists(), "the file is there to be read");
        assert!(!index.is_resident(), "opening loaded it anyway");
        assert_eq!(index.resident_len(), None);

        // And the read, when it finally happens, is the thing that fails.
        assert!(
            index.contains(1).is_err(),
            "a file that is not an index was accepted as one"
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// A store with no index yet answers a search rather than failing.
    #[test]
    fn an_absent_index_is_an_empty_one() {
        let dir = scratch("absent");
        let mut index = VectorIndex::open(&dir, 8, 4, false).unwrap();
        assert!(!index.exists());

        let (scores, ids) = index.search(&[0.0; 8], 5, &Allowlist::All).unwrap();
        assert!(scores.is_empty() && ids.is_empty());
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// Unit vectors pointing at one axis each, so the nearest neighbour of axis
    /// `i` is the vector on axis `i` and the ranking is decidable by hand.
    fn axis(dim: usize, i: usize) -> Vec<f32> {
        let mut v = vec![0.0; dim];
        v[i % dim] = 1.0;
        v
    }

    /// The release's structural claim, at a size a test can hold: vectors land
    /// in shards of a fixed size, the shards are files named for their first
    /// id, and a search over them finds what a search over one file would.
    #[test]
    fn vectors_land_in_shards_and_the_shards_are_searchable() {
        let dir = scratch("shards");
        let dim = 8;
        let mut index = VectorIndex::open(&dir, dim, 4, true).unwrap();
        let VectorIndex::Sharded(ref mut sharded) = index else {
            panic!("asked for shards, got one file")
        };
        sharded.capacity = 4;

        // Ten vectors, added in batches of three, so batches straddle shard
        // boundaries the way a real embedding batch does.
        for batch in 0..4 {
            let ids: Vec<u64> = (0..3)
                .map(|i| batch * 3 + i + 1)
                .filter(|id| *id <= 10)
                .collect();
            if ids.is_empty() {
                continue;
            }
            let flat: Vec<f32> = ids
                .iter()
                .flat_map(|id| axis(dim, *id as usize - 1))
                .collect();
            index.add(&flat, &ids).unwrap();
        }
        index.save().unwrap();

        assert_eq!(index.shards(), 3, "10 vectors at 4 per shard is 3 shards");
        let names: Vec<String> = std::fs::read_dir(dir.join(SHARD_DIR))
            .unwrap()
            .flatten()
            .map(|e| e.file_name().to_string_lossy().into_owned())
            .filter(|n| n.ends_with(SHARD_EXT))
            .collect();
        assert!(
            names.contains(&format!("{:016}.{SHARD_EXT}", 1u64)),
            "a shard is named for its first id: {names:?}"
        );

        // Reopened from the directory alone — nothing else records the layout.
        let mut reopened = VectorIndex::open(&dir, dim, 4, true).unwrap();
        assert_eq!(reopened.shards(), 3);
        for id in 1..=10u64 {
            assert!(reopened.contains(id).unwrap(), "id {id} went missing");
            let (_, hits) = reopened
                .search(&axis(dim, id as usize - 1), 1, &Allowlist::All)
                .unwrap();
            // Axes repeat every `dim`, so the best hit is one of the ids on
            // that axis rather than a specific one.
            assert!(
                hits.iter()
                    .all(|h| (h - 1) % dim as u64 == (id - 1) % dim as u64),
                "id {id} ranked {hits:?}, which is not on its axis"
            );
        }

        // A filter narrows to one shard's worth without reading past it.
        let (_, hits) = reopened
            .search(&axis(dim, 0), 5, &Allowlist::Subset(vec![9]))
            .unwrap();
        assert_eq!(hits, vec![9], "an allowlist of one returned {hits:?}");

        let _ = std::fs::remove_dir_all(&dir);
    }

    /// A store larger than its budget must still answer, holding only what the
    /// budget allows while it does.
    #[test]
    fn a_budget_of_one_shard_still_searches_all_of_them() {
        let dir = scratch("budget");
        let dim = 8;
        let mut index = VectorIndex::open(&dir, dim, 4, true).unwrap();
        let VectorIndex::Sharded(ref mut sharded) = index else {
            panic!("asked for shards")
        };
        sharded.capacity = 4;

        let ids: Vec<u64> = (1..=12).collect();
        let flat: Vec<f32> = ids
            .iter()
            .flat_map(|id| axis(dim, *id as usize - 1))
            .collect();
        index.add(&flat, &ids).unwrap();
        index.save().unwrap();
        assert_eq!(index.shards(), 3);

        let mut reopened = VectorIndex::open(&dir, dim, 4, true).unwrap();
        let VectorIndex::Sharded(ref mut sharded) = reopened else {
            panic!("shards on disk must reopen as shards")
        };
        sharded.max_resident = 1;

        // A search still reaches every shard...
        let (_, hits) = reopened.search(&axis(dim, 2), 12, &Allowlist::All).unwrap();
        assert!(
            hits.len() >= 3,
            "a budget of one shard returned {hits:?}, so shards were skipped rather than reloaded"
        );
        // ...and it did so by putting shards down again.
        assert!(
            reopened.evictions() >= 2,
            "three shards were searched under a one-shard budget without a single eviction"
        );
        let VectorIndex::Sharded(ref s) = reopened else {
            unreachable!()
        };
        assert_eq!(
            s.shards.iter().filter(|s| s.resident.is_some()).count(),
            1,
            "more than the budget stayed resident"
        );

        let _ = std::fs::remove_dir_all(&dir);
    }

    /// Removing a vector must touch its shard and no other file on disk.
    #[test]
    fn removing_rewrites_only_the_shard_holding_it() {
        let dir = scratch("dirty");
        let dim = 8;
        let mut index = VectorIndex::open(&dir, dim, 4, true).unwrap();
        let VectorIndex::Sharded(ref mut sharded) = index else {
            panic!("asked for shards")
        };
        sharded.capacity = 4;

        let ids: Vec<u64> = (1..=8).collect();
        let flat: Vec<f32> = ids
            .iter()
            .flat_map(|id| axis(dim, *id as usize - 1))
            .collect();
        index.add(&flat, &ids).unwrap();
        index.save().unwrap();
        assert_eq!(index.shards(), 2);

        let stamp = |name: u64| {
            std::fs::metadata(dir.join(SHARD_DIR).join(format!("{name:016}.{SHARD_EXT}")))
                .unwrap()
                .modified()
                .unwrap()
        };
        let first_before = stamp(1);
        let second_before = stamp(5);

        // Slept, because a rewrite within the same filesystem timestamp tick
        // would be indistinguishable from no rewrite at all.
        std::thread::sleep(std::time::Duration::from_millis(1100));
        let mut reopened = VectorIndex::open(&dir, dim, 4, true).unwrap();
        reopened.remove(6).unwrap();
        reopened.save().unwrap();

        assert_eq!(stamp(1), first_before, "an untouched shard was rewritten");
        assert!(
            stamp(5) > second_before,
            "the touched shard was not written"
        );
        assert!(!reopened.contains(6).unwrap());
        assert!(reopened.contains(5).unwrap());

        let _ = std::fs::remove_dir_all(&dir);
    }
}