khive_runtime/

retrieval.rs

//! Retrieval operations: local embedding generation and hybrid search with RRF fusion.

use std::collections::{HashMap, HashSet};

use lattice_embed::{EmbeddingModel, MAX_TEXT_BYTES};
use uuid::Uuid;

use crate::config::{parse_embedding_model_alias, sanitize_key};
use crate::curation::note_fts_document;
use crate::embedder_registry::with_embedding_admission;
use crate::error::{RuntimeError, RuntimeResult};
use crate::runtime::{KhiveRuntime, NamespaceToken};
use khive_retrieval::hybrid::{combine_leg_first_appearance, fuse_labelled, HitLabel};
use khive_score::DeterministicScore;
use khive_storage::types::{
    PageRequest, TextFilter, TextQueryMode, TextSearchHit, TextSearchRequest, VectorRecord,
    VectorSearchHit, VectorSearchRequest,
};
use khive_storage::ContentRef;
use khive_storage::EntityFilter;
use khive_types::SubstrateKind;

pub use khive_retrieval::{
    HybridSearchOutcome, RankScoreKind, SearchHit, SearchSignals, SearchSource,
};

/// Bounds provider input and per-page outcome memory while amortizing model setup.
pub(crate) const EMBEDDING_BATCH_PAGE_SIZE: usize = 256;

// Fault-injection flag for backfill reader errors (test / `fault-injection` builds only).
#[cfg(any(test, feature = "fault-injection"))]
std::thread_local! {
    static BACKFILL_READER_FAIL: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
}

/// Arm the backfill reader fault injection: the next `backfill_missing_embeddings`
/// call substitutes a `StorageError::Pool` for `sql.reader().await`'s result, then
/// resets the flag. The injected error passes through the same
/// `map_err(RuntimeError::Storage)?` path as a real reader failure, exercising the
/// fail-closed guard rather than bypassing it.
#[cfg(any(test, feature = "fault-injection"))]
pub fn arm_backfill_reader_fail() {
    BACKFILL_READER_FAIL.with(|c| c.set(true));
}

/// RRF constant. Controls how strongly top ranks dominate.
///
/// The paper's k=60 over-compresses scores at KG scale (tens–thousands of
/// entities): rank 1 ≈ 0.016, rank 10 ≈ 0.014, spread ≈ 0.002. k=10 gives
/// rank 1 ≈ 0.091, rank 10 ≈ 0.050, spread ≈ 0.041 — 20× better discrimination,
/// which dedup-before-create needs at graph sizes of 50–2700 entities.
const RRF_K: usize = 10;

/// Candidates pulled per path before fusion. Higher = better recall, more work.
const CANDIDATE_MULTIPLIER: u32 = 4;

/// Advisory emitted by write verbs when only the embedding input was bounded.
pub const EMBEDDING_INPUT_TRUNCATED_WARNING: &str =
    "embedding input was truncated to the embedder maximum; full content was stored unchanged";

/// Outcome for one document embedding. The exact-input fingerprint is present
/// only when the resolved provider has the runtime-audited preparation path.
#[derive(Clone, Debug)]
pub struct DocumentEmbeddingOutcome {
    pub model_name: String,
    pub vector: Vec<f32>,
    /// Digest of the exact prepared input, or `None` for an unaudited provider.
    pub prepared_text_fingerprint: Option<ContentRef>,
    pub source_bytes: usize,
    pub embedded_bytes: usize,
    pub truncated: bool,
}

/// Aggregate truncation accounting for one logical write or reindex batch.
#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
pub struct EmbeddingTruncationReport {
    pub truncated: u64,
    pub discarded_bytes: u64,
}

impl EmbeddingTruncationReport {
    pub fn observe(&mut self, outcome: &DocumentEmbeddingOutcome) {
        if outcome.truncated {
            self.truncated += 1;
            self.discarded_bytes +=
                outcome.source_bytes.saturating_sub(outcome.embedded_bytes) as u64;
        }
    }

    #[must_use]
    pub const fn any_truncated(&self) -> bool {
        self.truncated > 0
    }

    /// Add another batch's counters without wrapping on overflow.
    pub fn merge(&mut self, other: Self) {
        self.truncated = self.truncated.saturating_add(other.truncated);
        self.discarded_bytes = self.discarded_bytes.saturating_add(other.discarded_bytes);
    }
}

/// Maximum document bytes accepted before the embedding service adds its model prefix.
pub fn document_embedding_budget(model_name: &str) -> usize {
    parse_embedding_model_alias(model_name)
        .and_then(|model| model.document_instruction())
        .map_or(MAX_TEXT_BYTES, |prefix| {
            MAX_TEXT_BYTES.saturating_sub(prefix.len())
        })
}

/// Return the longest UTF-8-safe prefix of `text` within `max_bytes` and whether it was shortened.
pub fn bounded_embedding_input(text: &str, max_bytes: usize) -> (&str, bool) {
    if text.len() <= max_bytes {
        return (text, false);
    }

    let end = text
        .char_indices()
        .map(|(index, _)| index)
        .take_while(|index| *index <= max_bytes)
        .last()
        .unwrap_or(0);
    (&text[..end], true)
}

fn prepared_document_fingerprint(text: &str, model: EmbeddingModel) -> ContentRef {
    match model.document_instruction() {
        Some(prefix) => VectorRecord::fingerprint_text(&format!("{prefix}{text}")),
        None => VectorRecord::fingerprint_text(text),
    }
}

impl KhiveRuntime {
    fn require_default_embedder(&self) -> RuntimeResult<&str> {
        let model_name = self.default_embedder_name();
        if model_name.is_empty() {
            return Err(RuntimeError::Unconfigured("embedding_model".into()));
        }
        Ok(model_name)
    }

    /// Generate an embedding vector for `text` using the configured default model.
    ///
    /// First call lazily loads model weights (cold start cost). Subsequent calls reuse them.
    /// Returns `Unconfigured("embedding_model")` if no model is configured.
    pub async fn embed(&self, text: &str) -> RuntimeResult<Vec<f32>> {
        let model_name = self.require_default_embedder()?;
        self.embed_with_model(model_name, text).await
    }

    /// Generate an embedding vector for `text` using the named model.
    ///
    /// Accepts both built-in lattice model names/aliases and custom provider
    /// names registered via [`KhiveRuntime::register_embedder`]. For lattice
    /// models the resolved `EmbeddingModel` enum is forwarded to `embed_one`
    /// so the service can select the correct model variant. For custom
    /// providers, `embed_one` is called with `EmbeddingModel::default()`
    /// because custom services are expected to ignore the enum argument (they
    /// own a single model implicitly).
    ///
    /// Applies no instruction prefix (generic role). Use
    /// [`Self::embed_document_with_model_outcome`] / [`Self::embed_query_with_model`] for
    /// instruction-tuned models where the asymmetric prefix matters.
    ///
    /// Returns `UnknownModel` if `model_name` is not in the embedder registry.
    pub async fn embed_with_model(&self, model_name: &str, text: &str) -> RuntimeResult<Vec<f32>> {
        let model = parse_embedding_model_alias(model_name);
        let service = self.embedder(model_name).await?;
        let emb_model = model.unwrap_or_default();
        // Issued-at-dispatch: count before the provider await so a call that
        // was handed to the provider is counted even if this task is aborted
        // while parked on the await (drain_embed_join_set cancellation path).
        crate::usage::count(crate::usage::UsageUnit::EmbedCalls, 1);
        let out = with_embedding_admission(service.embed_one(text, emb_model)).await;
        out
    }

    /// Embed a document/passage for indexing using the named model.
    ///
    /// Applies `EmbeddingService::embed_passage`, which prepends the model's
    /// `document_instruction()` prefix when defined (e.g. `"passage: "` for
    /// multilingual-e5). For models with no document prefix (MiniLM, BGE) this
    /// is identical to [`Self::embed_with_model`].
    ///
    /// Use this for all index/store/backfill paths so that instruction-tuned
    /// models produce passage-side vectors.
    ///
    /// **Reindex caveat**: switching from an unprefixed model (or a model with no
    /// `document_instruction`) to an instruction-tuned model changes the vector
    /// representation. Vectors stored under the old scheme are not comparable to
    /// newly prefixed vectors. Operators must trigger a full reindex
    /// (`knowledge.index(rebuild_ann=true)` / `kkernel reindex`) after changing
    /// the embedding model config.
    ///
    /// Returns `UnknownModel` if `model_name` is not registered.
    pub async fn embed_document_with_model_outcome(
        &self,
        model_name: &str,
        text: &str,
    ) -> RuntimeResult<DocumentEmbeddingOutcome> {
        self.embed_document_with_model_outcome_inner(None, model_name, text)
            .await
    }

    pub(crate) async fn embed_document_with_model_outcome_for_token(
        &self,
        token: &NamespaceToken,
        model_name: &str,
        text: &str,
    ) -> RuntimeResult<DocumentEmbeddingOutcome> {
        self.embed_document_with_model_outcome_inner(Some(token), model_name, text)
            .await
    }

    async fn embed_document_with_model_outcome_inner(
        &self,
        token: Option<&NamespaceToken>,
        model_name: &str,
        text: &str,
    ) -> RuntimeResult<DocumentEmbeddingOutcome> {
        let model = parse_embedding_model_alias(model_name);
        let (service, audited_document_preparation) = self
            .embedder_with_input_attestation(model_name, token)
            .await?;
        let emb_model = model.unwrap_or_default();
        let source_bytes = text.len();
        let (text, truncated) =
            bounded_embedding_input(text, document_embedding_budget(model_name));
        let embedded_bytes = text.len();
        // Issued-at-dispatch: counted before the await — see embed_with_model.
        crate::usage::count(crate::usage::UsageUnit::EmbedCalls, 1);
        let embeddings =
            with_embedding_admission(service.embed_passage(&[text.to_string()], emb_model)).await;
        let mut vectors = embeddings?;
        if vectors.len() != 1 {
            return Err(RuntimeError::Internal(format!(
                "embed_passage returned {} vectors for 1 input",
                vectors.len()
            )));
        }
        let out = vectors.pop().expect("checked len == 1 above");
        Ok(DocumentEmbeddingOutcome {
            model_name: model_name.to_owned(),
            vector: out,
            prepared_text_fingerprint: audited_document_preparation
                .then(|| prepared_document_fingerprint(text, emb_model)),
            source_bytes,
            embedded_bytes,
            truncated,
        })
    }

    /// Embed a query string for retrieval using the named model.
    ///
    /// Applies the pre-0.6 query-role behavior. E5 and Qwen models retain
    /// their query instructions, while BGE and custom providers receive the
    /// original unprefixed query text.
    ///
    /// Use this for all search/recall/suggest query embedding paths so that
    /// instruction-tuned models land in the correct side of their retrieval
    /// space.
    ///
    /// Returns `UnknownModel` if `model_name` is not registered.
    pub async fn embed_query_with_model(
        &self,
        model_name: &str,
        text: &str,
    ) -> RuntimeResult<Vec<f32>> {
        self.embed_query_with_model_inner(None, model_name, text)
            .await
    }

    pub(crate) async fn embed_query_with_model_for_token(
        &self,
        token: &NamespaceToken,
        model_name: &str,
        text: &str,
    ) -> RuntimeResult<Vec<f32>> {
        self.embed_query_with_model_inner(Some(token), model_name, text)
            .await
    }

    async fn embed_query_with_model_inner(
        &self,
        token: Option<&NamespaceToken>,
        model_name: &str,
        text: &str,
    ) -> RuntimeResult<Vec<f32>> {
        let model = parse_embedding_model_alias(model_name);
        let service = match token {
            Some(token) => self.embedder_with_token(token, model_name).await?,
            None => self.embedder(model_name).await?,
        };
        let texts = [text.to_string()];
        let emb_model = model.unwrap_or_default();
        // Issued-at-dispatch: counted before the await — see embed_with_model.
        crate::usage::count(crate::usage::UsageUnit::EmbedCalls, 1);
        let embeddings = match emb_model {
            EmbeddingModel::BgeSmallEnV15
            | EmbeddingModel::BgeBaseEnV15
            | EmbeddingModel::BgeLargeEnV15 => {
                with_embedding_admission(service.embed(&texts, emb_model)).await
            }
            _ => with_embedding_admission(service.embed_query(&texts, emb_model)).await,
        };
        let out = embeddings?
            .into_iter()
            .next()
            .ok_or_else(|| RuntimeError::Internal("embed_query returned empty vec".into()))?;
        Ok(out)
    }

    /// Embed a document for indexing using the configured default model.
    ///
    /// Delegates to [`Self::embed_document_outcome`]. Use for entity/note
    /// create and reindex paths.
    ///
    /// Returns `Unconfigured("embedding_model")` if no model is configured.
    /// Returns an error if the input is bounded; use the outcome method to
    /// inspect the vector together with the truncation metadata.
    pub async fn embed_document(&self, text: &str) -> RuntimeResult<Vec<f32>> {
        let outcome = self.embed_document_outcome(text).await?;
        if outcome.truncated {
            return Err(RuntimeError::InvalidInput(format!(
                "embedding input truncated from {} to {} bytes; use embed_document_outcome to inspect the bounded vector",
                outcome.source_bytes, outcome.embedded_bytes
            )));
        }
        Ok(outcome.vector)
    }

    /// Embed a document with the default model and retain input-bounding metadata.
    pub async fn embed_document_outcome(
        &self,
        text: &str,
    ) -> RuntimeResult<DocumentEmbeddingOutcome> {
        let model_name = self.require_default_embedder()?;
        self.embed_document_with_model_outcome(model_name, text)
            .await
    }

    /// Embed a query for retrieval using the configured default model.
    ///
    /// Delegates to [`Self::embed_query_with_model`]. Use for vector search and
    /// hybrid search query paths.
    ///
    /// Returns `Unconfigured("embedding_model")` if no model is configured.
    pub async fn embed_query(&self, text: &str) -> RuntimeResult<Vec<f32>> {
        let model_name = self.require_default_embedder()?;
        self.embed_query_with_model(model_name, text).await
    }

    async fn embed_query_for_token(
        &self,
        token: &NamespaceToken,
        text: &str,
    ) -> RuntimeResult<Vec<f32>> {
        let model_name = self.require_default_embedder()?;
        self.embed_query_with_model_for_token(token, model_name, text)
            .await
    }

    /// Generate embeddings for multiple texts in one call using the configured default model.
    ///
    /// Delegates to the cached `EmbeddingService::embed`, so repeated texts within
    /// and across calls benefit from the runtime-level LRU cache.
    ///
    /// Returns an empty vec for empty input without hitting the embedding service.
    /// Returns `Unconfigured("embedding_model")` if no model is configured.
    pub async fn embed_batch(&self, texts: &[String]) -> RuntimeResult<Vec<Vec<f32>>> {
        if texts.is_empty() {
            return Ok(vec![]);
        }
        let model_name = self.require_default_embedder()?;
        self.embed_batch_with_model(model_name, texts).await
    }

    /// Generate embeddings for multiple texts using the named model.
    ///
    /// Accepts lattice model names/aliases and custom provider names.
    /// Returns `UnknownModel` if `model_name` is not in the embedder registry.
    pub async fn embed_batch_with_model(
        &self,
        model_name: &str,
        texts: &[String],
    ) -> RuntimeResult<Vec<Vec<f32>>> {
        if texts.is_empty() {
            return Ok(vec![]);
        }
        let model = parse_embedding_model_alias(model_name);
        let service = self.embedder(model_name).await?;
        let emb_model = model.unwrap_or_default();
        let out = with_embedding_admission(service.embed(texts, emb_model)).await;
        crate::usage::count(crate::usage::UsageUnit::EmbedCalls, texts.len() as u64);
        out
    }

    /// Embed a batch of documents for indexing using the named model.
    ///
    /// Applies `EmbeddingService::embed_passage`. Use for all bulk
    /// index/backfill/reindex operations to apply the passage-side prefix.
    /// A mixed batch is bounded into one ordered owned batch so truncation never
    /// fragments one provider batch into sequential singleton inference calls.
    ///
    /// **Reindex caveat**: see [`Self::embed_document_with_model_outcome`] — the same
    /// incomparability applies to batch-indexed vectors when switching models.
    ///
    /// Returns `UnknownModel` if `model_name` is not registered.
    /// Returns an error when any input is bounded; use the outcomes method to
    /// retain the bounded vectors and per-document byte counts.
    pub async fn embed_document_batch_with_model(
        &self,
        model_name: &str,
        texts: &[String],
    ) -> RuntimeResult<Vec<Vec<f32>>> {
        if texts.is_empty() {
            return Ok(vec![]);
        }
        let outcomes = self
            .embed_document_batch_with_model_outcomes(model_name, texts)
            .await?;
        let mut report = EmbeddingTruncationReport::default();
        for outcome in &outcomes {
            report.observe(outcome);
        }
        if report.any_truncated() {
            return Err(RuntimeError::InvalidInput(format!(
                "embedding input truncated for {} documents ({} discarded bytes); use embed_document_batch_with_model_outcomes to inspect the bounded vectors",
                report.truncated, report.discarded_bytes
            )));
        }
        Ok(outcomes.into_iter().map(|outcome| outcome.vector).collect())
    }

    pub async fn embed_document_batch_with_model_outcomes(
        &self,
        model_name: &str,
        texts: &[String],
    ) -> RuntimeResult<Vec<DocumentEmbeddingOutcome>> {
        self.embed_document_batch_with_model_outcomes_inner(None, model_name, texts)
            .await
    }

    pub(crate) async fn embed_document_batch_with_model_outcomes_for_token(
        &self,
        token: &NamespaceToken,
        model_name: &str,
        texts: &[String],
    ) -> RuntimeResult<Vec<DocumentEmbeddingOutcome>> {
        self.embed_document_batch_with_model_outcomes_inner(Some(token), model_name, texts)
            .await
    }

    async fn embed_document_batch_with_model_outcomes_inner(
        &self,
        token: Option<&NamespaceToken>,
        model_name: &str,
        texts: &[String],
    ) -> RuntimeResult<Vec<DocumentEmbeddingOutcome>> {
        if texts.is_empty() {
            return Ok(vec![]);
        }
        let model = parse_embedding_model_alias(model_name);
        let (service, audited_document_preparation) = self
            .embedder_with_input_attestation(model_name, token)
            .await?;
        let emb_model = model.unwrap_or_default();
        let budget = document_embedding_budget(model_name);
        if token.is_some() {
            // Atomic preparation records issued work even if its task is cancelled.
            crate::usage::count(crate::usage::UsageUnit::EmbedCalls, texts.len() as u64);
        }
        let out = if texts.iter().all(|text| text.len() <= budget) {
            with_embedding_admission(service.embed_passage(texts, emb_model)).await
        } else {
            let bounded_texts: Vec<String> = texts
                .iter()
                .map(|text| bounded_embedding_input(text, budget).0.to_owned())
                .collect();
            with_embedding_admission(service.embed_passage(&bounded_texts, emb_model)).await
        };
        if token.is_none() {
            crate::usage::count(crate::usage::UsageUnit::EmbedCalls, texts.len() as u64);
        }
        let vectors = out?;
        if vectors.len() != texts.len() {
            return Err(RuntimeError::Internal(format!(
                "embed_passage returned {} vectors for {} inputs",
                vectors.len(),
                texts.len()
            )));
        }
        Ok(texts
            .iter()
            .zip(vectors)
            .map(|(text, vector)| {
                let (bounded, truncated) = bounded_embedding_input(text, budget);
                DocumentEmbeddingOutcome {
                    model_name: model_name.to_owned(),
                    vector,
                    prepared_text_fingerprint: audited_document_preparation
                        .then(|| prepared_document_fingerprint(bounded, emb_model)),
                    source_bytes: text.len(),
                    embedded_bytes: bounded.len(),
                    truncated,
                }
            })
            .collect())
    }

    /// Embed a batch of documents for indexing using the configured default model.
    ///
    /// Convenience delegate to [`Self::embed_document_batch_with_model`]. Use for
    /// bulk knowledge-atom and section indexing paths.
    ///
    /// Returns `Unconfigured("embedding_model")` if no model is configured.
    /// Returns an error when any input is bounded; use
    /// [`Self::embed_document_batch_outcomes`] to retain the bounded vectors.
    pub async fn embed_document_batch(&self, texts: &[String]) -> RuntimeResult<Vec<Vec<f32>>> {
        if texts.is_empty() {
            return Ok(vec![]);
        }
        let model_name = self.require_default_embedder()?;
        self.embed_document_batch_with_model(model_name, texts)
            .await
    }

    /// Embed documents with the default model and retain actual input-bounding metadata.
    pub async fn embed_document_batch_outcomes(
        &self,
        texts: &[String],
    ) -> RuntimeResult<Vec<DocumentEmbeddingOutcome>> {
        if texts.is_empty() {
            return Ok(vec![]);
        }
        let model_name = self.require_default_embedder()?;
        self.embed_document_batch_with_model_outcomes(model_name, texts)
            .await
    }

    /// Embed a batch of queries for retrieval using the named model.
    ///
    /// Applies the same pre-0.6 query-role compatibility behavior as
    /// [`Self::embed_query_with_model`].
    ///
    /// Returns `UnknownModel` if `model_name` is not registered.
    pub async fn embed_query_batch_with_model(
        &self,
        model_name: &str,
        texts: &[String],
    ) -> RuntimeResult<Vec<Vec<f32>>> {
        if texts.is_empty() {
            return Ok(vec![]);
        }
        let model = parse_embedding_model_alias(model_name);
        let service = self.embedder(model_name).await?;
        let emb_model = model.unwrap_or_default();
        let out = match emb_model {
            EmbeddingModel::BgeSmallEnV15
            | EmbeddingModel::BgeBaseEnV15
            | EmbeddingModel::BgeLargeEnV15 => {
                with_embedding_admission(service.embed(texts, emb_model)).await
            }
            _ => with_embedding_admission(service.embed_query(texts, emb_model)).await,
        };
        crate::usage::count(crate::usage::UsageUnit::EmbedCalls, texts.len() as u64);
        out
    }

    /// Search vectors using either a caller-provided embedding or query text.
    ///
    /// Existing callers pass `query_embedding: Some(vec)` to avoid re-embedding.
    /// Text callers pass `query_embedding: None, query_text: Some(...)` and the
    /// runtime embeds internally.
    pub async fn vector_search(
        &self,
        token: &NamespaceToken,
        query_embedding: Option<Vec<f32>>,
        query_text: Option<&str>,
        top_k: u32,
        kind: Option<SubstrateKind>,
    ) -> RuntimeResult<Vec<VectorSearchHit>> {
        let embedding = match query_embedding {
            Some(vec) => vec,
            None => {
                let text = query_text.ok_or_else(|| {
                    RuntimeError::InvalidInput(
                        "vector search requires query_embedding or query_text".into(),
                    )
                })?;
                if text.trim().is_empty() {
                    return Err(RuntimeError::InvalidInput(
                        "query_text must not be empty".into(),
                    ));
                }
                self.embed_query_for_token(token, text).await?
            }
        };

        let ns = token.namespace().as_str().to_owned();
        let hits = self
            .vectors(token)?
            .search(VectorSearchRequest {
                query_vectors: vec![embedding],
                top_k,
                namespace: Some(ns),
                kind,
                embedding_model: None,
                filter: None,
                backend_hints: None,
            })
            .await;
        crate::usage::count(crate::usage::UsageUnit::VectorPasses, 1);
        hits.map_err(RuntimeError::from)
    }

    /// The note-search vector leg uses the pack-owned graph when that model
    /// has an installed, consumer-protected bridge. Only a missing graph for
    /// this consumer takes the existing exact sqlite-vec route.
    pub(crate) async fn note_search_vector_search(
        &self,
        token: &NamespaceToken,
        query_embedding: Option<Vec<f32>>,
        query_text: &str,
        top_k: u32,
    ) -> RuntimeResult<Vec<VectorSearchHit>> {
        let embedding = match query_embedding {
            Some(embedding) => embedding,
            None => self.embed_query_for_token(token, query_text).await?,
        };
        let model = self.default_embedder_name();
        if !model.is_empty() {
            if let Some(provider) = self.note_search_ann_provider()? {
                if let Some(hits) = provider.search(token, model, &embedding, top_k).await? {
                    crate::note_search_ann::record_ann_route();
                    crate::usage::count(crate::usage::UsageUnit::VectorPasses, 1);
                    return Ok(hits);
                }
            }
        }
        crate::note_search_ann::record_fallback_route();
        self.vector_search(
            token,
            Some(embedding),
            None,
            top_k,
            Some(SubstrateKind::Note),
        )
        .await
    }

    /// Hybrid search: text (FTS5) + vector retrieval fused via Reciprocal Rank Fusion.
    ///
    /// - Always performs text search over `query_text`.
    /// - If `query_vector` is `Some`, also performs vector search and fuses both lists.
    /// - If `None`, returns text-only results — no vector store needed.
    /// - If `entity_kind` is `Some`, the alive-set query filters to that kind.
    ///   The text/vector candidate pools are unfiltered up front; the kind
    ///   filter applies at the alive-check stage where we already fetch each
    ///   candidate to confirm it isn't soft-deleted.
    /// - `tags_any`: when non-empty, only entities that have at least one of these
    ///   tags (case-insensitive) survive the alive-set intersection. Applied BEFORE
    ///   truncation so matches ranked beyond `limit` in the raw fusion are not lost.
    /// - `properties_filter`: when `Some`, only entities whose properties are a
    ///   superset of the given JSON object survive. Applied BEFORE truncation.
    ///
    /// `limit` caps the final returned list; internally pulls `limit * 4` candidates per path.
    ///
    /// # Cross-namespace visibility (entity search — text leg: visible set; vector leg: primary)
    ///
    /// The two legs of entity search scope namespaces differently.
    ///
    /// **Text leg.** Entity full-text search is one shared table (`fts_entities`)
    /// with a `namespace` column. The token's whole visible set (`visible_ns`) is
    /// forwarded in `TextFilter.namespaces`, which the store applies as a
    /// `namespace IN (...)` predicate, so one query returns text hits from every
    /// visible namespace.
    ///
    /// **Vector leg.** The vector leg runs only when a query vector is supplied or
    /// an embedding model is configured. It is an exact sqlite-vec search
    /// (brute-force cosine, as in `knn`); no ANN index is involved. It issues a
    /// single request scoped to the primary namespace, because a
    /// `VectorSearchRequest` carries one namespace. Searching the visible set
    /// would take one request per namespace and a merge of the per-namespace
    /// lists; that fanout is not implemented for entities, so entity vector hits
    /// come from the primary namespace only.
    ///
    /// The fused candidates are then checked against the entity store with the
    /// visible set, so an entity from an extra visible namespace is returned when
    /// the text leg matched it, and is not returned on a vector-only match.
    ///
    /// This differs from `memory.recall`, whose vector leg already searches every
    /// visible namespace.
    ///
    /// Callers can also read any visible entity directly via `get_entity` /
    /// `resolve`.
    #[allow(clippy::too_many_arguments)]
    pub async fn hybrid_search(
        &self,
        token: &NamespaceToken,
        query_text: &str,
        query_vector: Option<Vec<f32>>,
        limit: u32,
        entity_kind: Option<&str>,
        entity_type: Option<&str>,
        tags_any: &[String],
        properties_filter: Option<&serde_json::Value>,
    ) -> RuntimeResult<Vec<SearchHit>> {
        self.hybrid_search_with_text_mode(
            token,
            query_text,
            query_vector,
            limit,
            entity_kind,
            entity_type,
            tags_any,
            properties_filter,
            TextQueryMode::Plain,
        )
        .await
    }

    /// Hybrid search with an explicit lexical mode for the text arm.
    #[allow(clippy::too_many_arguments)]
    pub async fn hybrid_search_with_text_mode(
        &self,
        token: &NamespaceToken,
        query_text: &str,
        query_vector: Option<Vec<f32>>,
        limit: u32,
        entity_kind: Option<&str>,
        entity_type: Option<&str>,
        tags_any: &[String],
        properties_filter: Option<&serde_json::Value>,
        text_mode: TextQueryMode,
    ) -> RuntimeResult<Vec<SearchHit>> {
        let (hits, _vector_error) = self
            .hybrid_search_inner(
                token,
                query_text,
                query_vector,
                limit,
                entity_kind,
                entity_type,
                tags_any,
                properties_filter,
                text_mode,
                None,
                false,
                None,
            )
            .await?;
        Ok(hits)
    }

    /// Hybrid search over several entity kinds that issues the vector query once.
    ///
    /// Returns one list per entry of `entity_kinds`, in that order. Each list is what
    /// [`Self::hybrid_search`] returns for that kind with the same `limit`: the text stage
    /// is filtered to the kind and keeps its own `limit * 4` budget, and fusion, the kind
    /// filter and the cut to `limit` run per kind. Only the vector stage is shared, because
    /// it takes no entity kind: the single query is the one every per-kind search would have
    /// issued. When `query_vector` is `None` and an embedding model is configured, the query
    /// text is embedded once instead of once per kind. No entity-type, tag or property
    /// filter is set. An empty `entity_kinds` returns no lists and runs no query.
    pub async fn hybrid_search_each_kind(
        &self,
        token: &NamespaceToken,
        query_text: &str,
        query_vector: Option<Vec<f32>>,
        limit: u32,
        entity_kinds: &[&str],
    ) -> RuntimeResult<Vec<Vec<SearchHit>>> {
        if entity_kinds.is_empty() {
            return Ok(Vec::new());
        }
        let candidates = limit.saturating_mul(CANDIDATE_MULTIPLIER).max(limit);
        let (vector_hits, _vector_error) = self
            .hybrid_vector_stage(token, query_text, query_vector, candidates, None, false)
            .await?;
        let mut per_kind = Vec::with_capacity(entity_kinds.len());
        for &kind in entity_kinds {
            let (hits, _vector_error) = self
                .hybrid_search_inner(
                    token,
                    query_text,
                    None,
                    limit,
                    Some(kind),
                    None,
                    &[],
                    None,
                    TextQueryMode::Plain,
                    None,
                    false,
                    Some(vector_hits.clone()),
                )
                .await?;
            per_kind.push(hits);
        }
        Ok(per_kind)
    }

    /// `vector_similarity_floor` is a cosine-similarity value in `[-1.0,
    /// 1.0]`, matching both the canonical vector-store score contract and the
    /// scale documented on the `resolve` verb and on
    /// `SEARCH_VECTOR_SIMILARITY_FLOOR`.
    #[allow(clippy::too_many_arguments)]
    pub(crate) async fn hybrid_search_with_vector_similarity_floor(
        &self,
        token: &NamespaceToken,
        query_text: &str,
        query_vector: Option<Vec<f32>>,
        limit: u32,
        entity_kind: Option<&str>,
        entity_type: Option<&str>,
        tags_any: &[String],
        properties_filter: Option<&serde_json::Value>,
        vector_similarity_floor: f64,
    ) -> RuntimeResult<Vec<SearchHit>> {
        let (hits, _vector_error) = self
            .hybrid_search_inner(
                token,
                query_text,
                query_vector,
                limit,
                entity_kind,
                entity_type,
                tags_any,
                properties_filter,
                TextQueryMode::Plain,
                Some(vector_similarity_floor),
                false,
                None,
            )
            .await?;
        Ok(hits)
    }

    /// Coordinator fan-out variant of [`Self::hybrid_search`]: the text arm
    /// still fails loud (propagated through `hybrid_search_inner`'s
    /// `tolerate_vector_error=false` semantics for that leg), but a
    /// vector-arm failure after a successful text leg is captured instead of
    /// discarding the text hits. Reserved for
    /// `SubstrateCoordinator::fan_out_search_with_visibility` — every other
    /// caller keeps the fail-loud [`Self::hybrid_search`] contract, so a
    /// single backend's vector-store outage does not misreport that
    /// backend's text arm as failed too.
    #[allow(clippy::too_many_arguments)]
    pub async fn hybrid_search_outcome(
        &self,
        token: &NamespaceToken,
        query_text: &str,
        limit: u32,
        entity_kind: Option<&str>,
        entity_type: Option<&str>,
        tags_any: &[String],
        properties_filter: Option<&serde_json::Value>,
    ) -> RuntimeResult<HybridSearchOutcome> {
        self.hybrid_search_outcome_with_text_mode(
            token,
            query_text,
            limit,
            entity_kind,
            entity_type,
            tags_any,
            properties_filter,
            TextQueryMode::Plain,
        )
        .await
    }

    /// Coordinator variant with an explicit lexical mode for the text arm.
    #[allow(clippy::too_many_arguments)]
    pub async fn hybrid_search_outcome_with_text_mode(
        &self,
        token: &NamespaceToken,
        query_text: &str,
        limit: u32,
        entity_kind: Option<&str>,
        entity_type: Option<&str>,
        tags_any: &[String],
        properties_filter: Option<&serde_json::Value>,
        text_mode: TextQueryMode,
    ) -> RuntimeResult<HybridSearchOutcome> {
        let (hits, vector_error) = self
            .hybrid_search_inner(
                token,
                query_text,
                None,
                limit,
                entity_kind,
                entity_type,
                tags_any,
                properties_filter,
                text_mode,
                None,
                true,
                None,
            )
            .await?;
        Ok(HybridSearchOutcome { hits, vector_error })
    }

    /// `vector_pool`, when `Some`, supplies the vector stage's hits and the stage does not
    /// run; `None` runs it. The stage takes no entity kind, so one pool serves every kind.
    #[allow(clippy::too_many_arguments)]
    async fn hybrid_search_inner(
        &self,
        token: &NamespaceToken,
        query_text: &str,
        query_vector: Option<Vec<f32>>,
        limit: u32,
        entity_kind: Option<&str>,
        entity_type: Option<&str>,
        tags_any: &[String],
        properties_filter: Option<&serde_json::Value>,
        text_mode: TextQueryMode,
        vector_similarity_floor: Option<f64>,
        tolerate_vector_error: bool,
        vector_pool: Option<Vec<VectorSearchHit>>,
    ) -> RuntimeResult<(Vec<SearchHit>, Option<String>)> {
        let candidates = limit.saturating_mul(CANDIDATE_MULTIPLIER).max(limit);

        let visible_ns: Vec<String> = token
            .visible_namespaces()
            .iter()
            .map(|ns| ns.as_str().to_owned())
            .collect();
        // sanitize_fts5_query strips known-unsafe FTS5 metacharacters up front, but if
        // the lexical leg still errors at runtime on residual punctuation the sanitizer
        // doesn't strip, this fails loud instead of degrading to vector-only fusion.
        let text_store = self.text(token)?;
        let text_fut = text_store.search(TextSearchRequest {
            query: query_text.to_string(),
            mode: text_mode,
            filter: Some(TextFilter {
                namespaces: visible_ns.clone(),
                // Push the entity-kind filter into the FTS query. Without it the
                // text arm returns the top `candidates` rows across EVERY entity
                // kind in the namespace and the kind is applied only afterwards,
                // so when one kind dominates the lexical ranking a search for a
                // rarer kind gets back fewer rows than exist, or none. The
                // `EntityFilter.kinds` check below stays as the backstop.
                record_kinds: entity_kind
                    .map(|kind| vec![kind.to_string()])
                    .unwrap_or_default(),
                ..TextFilter::default()
            }),
            top_k: candidates,
            snippet_chars: 200,
        });
        let text_fut = crate::stage_seam::text_stage(text_fut);
        // The stages read nothing from each other, so they run together; a text error wins.
        let vector_fut = async {
            match vector_pool {
                Some(pool) => Ok((pool, None)),
                None => {
                    self.hybrid_vector_stage(
                        token,
                        query_text,
                        query_vector,
                        candidates,
                        vector_similarity_floor,
                        tolerate_vector_error,
                    )
                    .await
                }
            }
        };
        let (text_search_result, vector_result) = tokio::join!(text_fut, vector_fut);
        // FtsPasses is counted inside the store's `search()` (khive-db
        // stores/text.rs), only once a real FTS5 statement is prepared —
        // an empty/fully-sanitized query short-circuits there before any
        // statement exists and must not count.
        let text_hits = crate::error::fts_text_leg_or_err(
            text_search_result.map_err(RuntimeError::from),
            "hybrid_search",
            query_text,
        )?;
        let (vector_hits, vector_error) = vector_result?;

        // Each arm fetched `candidates` independently, so their union can contain
        // twice that many distinct IDs. Keep the complete fetched pool through
        // ranking and the alive/kind/tag/property filters below; reusing one arm's
        // cap here lets stale or filtered hits hide live candidates from the other.
        let fusion_limit = text_hits.len().saturating_add(vector_hits.len());
        let mut fused = rrf_fuse(text_hits, vector_hits, fusion_limit, query_text);

        // tags_any has a SQL column and is pushed into query_entities; properties
        // filtering has no SQL column and is applied in Rust below on the fetched records.
        if !fused.is_empty() {
            let candidate_ids: Vec<Uuid> = fused.iter().map(|h| h.entity_id).collect();
            let alive_page = self
                .entities(token)?
                .query_entities(
                    token.namespace().as_str(),
                    EntityFilter {
                        ids: candidate_ids,
                        kinds: entity_kind.map(|k| vec![k.to_string()]).unwrap_or_default(),
                        entity_types: entity_type.map(|t| vec![t.to_string()]).unwrap_or_default(),
                        namespaces: visible_ns,
                        tags_any: tags_any.to_vec(),
                        ..EntityFilter::default()
                    },
                    PageRequest {
                        offset: 0,
                        limit: u32::try_from(fused.len()).unwrap_or(u32::MAX),
                    },
                )
                .await?;
            let mut entity_meta: HashMap<Uuid, (String, Option<String>)> = HashMap::new();
            let mut alive: HashSet<Uuid> = HashSet::new();
            for e in alive_page.items {
                // Drop non-matching candidates here, before the alive set is built,
                // so they're excluded ahead of truncation.
                if let Some(pf) = properties_filter {
                    if !properties_match(e.properties.as_ref(), pf) {
                        continue;
                    }
                }
                alive.insert(e.id);
                entity_meta.insert(e.id, (e.name, e.description));
            }

            fused.retain(|h| alive.contains(&h.entity_id));

            // Enrich vector-only hits (title/snippet == None) from entity record.
            for hit in &mut fused {
                if let Some((name, description)) = entity_meta.get(&hit.entity_id) {
                    if hit.title.is_none() {
                        hit.title = Some(name.clone());
                    }
                    if hit.snippet.is_none() {
                        hit.snippet = description.clone();
                    }
                }
            }
        }

        fused.truncate(limit as usize);
        Ok((fused, vector_error))
    }

    /// The vector stage of hybrid entity search: one KNN query over the entity vectors of
    /// the primary namespace. It does not depend on an entity kind, which only the text
    /// stage and the filter after fusion apply.
    async fn hybrid_vector_stage(
        &self,
        token: &NamespaceToken,
        query_text: &str,
        query_vector: Option<Vec<f32>>,
        candidates: u32,
        vector_similarity_floor: Option<f64>,
        tolerate_vector_error: bool,
    ) -> RuntimeResult<(Vec<VectorSearchHit>, Option<String>)> {
        let mut vector_error: Option<String> = None;
        let mut vector_hits = if query_vector.is_some() || self.config().embedding_model.is_some() {
            match self
                .vector_search(
                    token,
                    query_vector,
                    Some(query_text),
                    candidates,
                    Some(SubstrateKind::Entity),
                )
                .await
            {
                Ok(hits) => hits,
                Err(e) if tolerate_vector_error => {
                    vector_error = Some(e.to_string());
                    Vec::new()
                }
                Err(e) => return Err(e),
            }
        } else {
            Vec::new()
        };
        if let Some(cosine_floor) = vector_similarity_floor {
            // Vector store scores use canonical cosine similarity (`1 - distance`),
            // so the caller's raw-cosine floor is already on the comparison scale.
            let score_floor = DeterministicScore::from_f64(cosine_floor);
            vector_hits.retain(|hit| hit.score >= score_floor);
        }
        Ok((vector_hits, vector_error))
    }

    /// Exact KNN over the full namespace's vector store.
    ///
    /// sqlite-vec uses brute-force cosine — results are exact, not approximate.
    /// Cost is O(N · D) per query. For small-to-medium namespaces (~hundreds of
    /// thousands of vectors) this is well within latency budgets.
    pub async fn knn(
        &self,
        token: &NamespaceToken,
        query_vector: Vec<f32>,
        top_k: u32,
    ) -> RuntimeResult<Vec<VectorSearchHit>> {
        let ns = token.namespace().as_str().to_owned();
        Ok(self
            .vectors(token)?
            .search(VectorSearchRequest {
                query_vectors: vec![query_vector],
                top_k,
                namespace: Some(ns),
                kind: Some(SubstrateKind::Entity),
                embedding_model: None,
                filter: None,
                backend_hints: None,
            })
            .await?)
    }

    /// Exact KNN restricted to a candidate set.
    ///
    /// Useful for reranking the top-N results from `hybrid_search` (or any other
    /// retrieval path) with exact cosine similarity against a query vector.
    /// Returns hits sorted by similarity (highest first), truncated to `top_k`.
    pub async fn rerank(
        &self,
        token: &NamespaceToken,
        query_vector: &[f32],
        candidate_ids: &[Uuid],
        top_k: u32,
    ) -> RuntimeResult<Vec<VectorSearchHit>> {
        let candidate_set: HashSet<Uuid> = candidate_ids.iter().copied().collect();
        let ns = token.namespace().as_str().to_owned();
        let all_hits = self
            .vectors(token)?
            .search(VectorSearchRequest {
                query_vectors: vec![query_vector.to_vec()],
                top_k: candidate_ids.len() as u32,
                namespace: Some(ns),
                kind: Some(SubstrateKind::Entity),
                embedding_model: None,
                filter: None,
                backend_hints: None,
            })
            .await?;
        let mut hits: Vec<VectorSearchHit> = all_hits
            .into_iter()
            .filter(|h| candidate_set.contains(&h.subject_id))
            .collect();
        hits.sort_by_key(|hit| std::cmp::Reverse(hit.score));
        hits.truncate(top_k as usize);
        Ok(hits)
    }

    async fn embed_backfill_page(
        &self,
        token: &NamespaceToken,
        model_name: &str,
        inputs: &[(Uuid, String)],
    ) -> Vec<Option<DocumentEmbeddingOutcome>> {
        let texts: Vec<String> = inputs.iter().map(|(_, text)| text.clone()).collect();
        match self
            .embed_document_batch_with_model_outcomes_for_token(token, model_name, &texts)
            .await
        {
            Ok(outcomes) => outcomes.into_iter().map(Some).collect(),
            Err(error) => {
                tracing::warn!(
                    model = %model_name,
                    error = %error,
                    "backfill_missing_embeddings: batch embed failed; retrying records individually"
                );
                let mut outcomes = Vec::with_capacity(inputs.len());
                for (id, text) in inputs {
                    match self
                        .embed_document_with_model_outcome_for_token(token, model_name, text)
                        .await
                    {
                        Ok(outcome) => outcomes.push(Some(outcome)),
                        Err(error) => {
                            tracing::warn!(
                                id = %id,
                                model = %model_name,
                                error = %error,
                                "backfill_missing_embeddings: record embed failed"
                            );
                            outcomes.push(None);
                        }
                    }
                }
                outcomes
            }
        }
    }

    /// Backfill vector and FTS index entries for entities and notes that are missing them.
    ///
    /// Intended to run once at startup as a background task (warm-up sequence steps 2–4).
    /// Queries the SQL substrate for entity bodies and note contents that have no
    /// corresponding entry in an eligible embedding model's vector store, then
    /// embeds and inserts them. FTS entries missing for notes are also repopulated.
    ///
    /// The operation is best-effort: individual embed/insert failures are logged and
    /// skipped rather than aborting the whole backfill. If no embedding models are
    /// registered, returns immediately with 0.
    ///
    /// Returns the total number of records backfilled across all models.
    pub async fn backfill_missing_embeddings(&self, token: &NamespaceToken) -> RuntimeResult<u64> {
        use khive_storage::types::{SqlRow, SqlStatement, SqlValue};

        let model_names = self.registered_embedding_model_names();
        if model_names.is_empty() {
            tracing::debug!(
                "backfill_missing_embeddings: no embedding models registered, skipping"
            );
            return Ok(0);
        }

        let ns = token.namespace().as_str().to_string();
        let mut total_backfilled = 0u64;

        for model_name in &model_names {
            let mut model_truncation = EmbeddingTruncationReport::default();
            match self.vectors_for_model(token, model_name) {
                Ok(_) => {}
                Err(error) => {
                    tracing::warn!(model = %model_name, error = %error,
                        "backfill_missing_embeddings: vector store unavailable");
                    continue;
                }
            };
            // Must match vec_model_key's naming logic.
            let vec_table = format!("vec_{}", sanitize_key(model_name));

            // --- Entities: embed the canonical body where no vector exists ---
            // Keyset pagination advances past failed or ineligible records too;
            // they must not keep the first page full forever.
            const PAGE_SIZE: usize = EMBEDDING_BATCH_PAGE_SIZE;
            let mut entity_total = 0usize;
            let mut entity_cursor = String::new();
            loop {
                let entity_sql = SqlStatement {
                    sql: format!(
                        "SELECT id FROM entities \
                         WHERE namespace = ?1 AND deleted_at IS NULL AND id > ?3 \
                         AND id NOT IN (\
                             SELECT subject_id FROM {vec_table} \
                             WHERE namespace = ?1 AND embedding_model = ?2 \
                         ) ORDER BY id LIMIT {PAGE_SIZE}"
                    ),
                    params: vec![
                        SqlValue::Text(ns.clone()),
                        SqlValue::Text(model_name.clone()),
                        SqlValue::Text(entity_cursor.clone()),
                    ],
                    label: Some("backfill_entities".into()),
                };

                let entity_rows: Vec<SqlRow> = {
                    let sql = self.sql();
                    let reader_result = sql.reader().await;
                    #[cfg(any(test, feature = "fault-injection"))]
                    let reader_result = if BACKFILL_READER_FAIL.with(|c| c.get()) {
                        BACKFILL_READER_FAIL.with(|c| c.set(false));
                        Err(khive_storage::StorageError::Pool {
                            operation: "reader".into(),
                            message: "injected failure".into(),
                        })
                    } else {
                        reader_result
                    };
                    let mut reader = reader_result.map_err(RuntimeError::Storage)?;
                    reader
                        .query_all(entity_sql)
                        .await
                        .map_err(RuntimeError::Storage)?
                };

                let batch_len = entity_rows.len();
                entity_total += batch_len;
                if batch_len == 0 {
                    break;
                }
                entity_cursor = match entity_rows.last().and_then(|row| row.columns.first()) {
                    Some(column) => match &column.value {
                        SqlValue::Text(id) => id.clone(),
                        _ => {
                            return Err(RuntimeError::Internal(
                                "backfill entity ID is not text".into(),
                            ))
                        }
                    },
                    None => {
                        return Err(RuntimeError::Internal(
                            "backfill entity page is empty".into(),
                        ))
                    }
                };

                let entity_store = self.entities(token)?;
                let mut entities = Vec::with_capacity(batch_len);
                let mut inputs = Vec::with_capacity(batch_len);
                for row in &entity_rows {
                    let Some(SqlValue::Text(id)) = row.columns.first().map(|column| &column.value)
                    else {
                        continue;
                    };
                    let Ok(id) = id.parse::<Uuid>() else { continue };
                    let Some(entity) = entity_store.get_entity(id).await? else {
                        continue;
                    };
                    if entity.namespace != ns || entity.deleted_at.is_some() {
                        continue;
                    }
                    let text = crate::curation::entity_embedding_text(&entity);
                    if text.trim().is_empty() {
                        continue;
                    }
                    inputs.push((id, text));
                    entities.push(entity);
                }
                let outcomes = self.embed_backfill_page(token, model_name, &inputs).await;
                for (entity, outcome) in entities.into_iter().zip(outcomes) {
                    let Some(outcome) = outcome else { continue };
                    model_truncation.observe(&outcome);
                    match self
                        .publish_entity_vector_revision(token, &entity, model_name, &outcome.vector)
                        .await
                    {
                        Ok(true) => total_backfilled += 1,
                        Ok(false) => {}
                        Err(error) => tracing::warn!(
                            id = %entity.id, model = %model_name, error = %error,
                            "backfill_missing_embeddings: entity vector insert failed"
                        ),
                    }
                }

                if batch_len < PAGE_SIZE {
                    break;
                }
            }

            // --- Notes: embed content where no vector entry exists ---
            let text_store = self.text_for_notes(token).ok();
            let note_store = self.notes(token).ok();
            let mut note_total = 0usize;
            // Excluded kinds remain absent from this model's vector table.
            // Advancing by id prevents a full page of them from repeating.
            let mut note_cursor = String::new();
            loop {
                // Only the id is selected here; the full Note is fetched below so
                // note_fts_document gets all fields and stays parity-correct.
                let note_sql = SqlStatement {
                    sql: format!(
                        "SELECT id FROM notes \
                         WHERE namespace = ?1 AND deleted_at IS NULL AND id > ?3 \
                         AND id NOT IN (\
                             SELECT subject_id FROM {vec_table} \
                             WHERE namespace = ?1 AND embedding_model = ?2 \
                         ) ORDER BY id LIMIT {PAGE_SIZE}"
                    ),
                    params: vec![
                        SqlValue::Text(ns.clone()),
                        SqlValue::Text(model_name.clone()),
                        SqlValue::Text(note_cursor.clone()),
                    ],
                    label: Some("backfill_notes".into()),
                };

                let note_rows: Vec<SqlRow> = {
                    let sql = self.sql();
                    let reader_result = sql.reader().await;
                    #[cfg(any(test, feature = "fault-injection"))]
                    let reader_result = if BACKFILL_READER_FAIL.with(|c| c.get()) {
                        BACKFILL_READER_FAIL.with(|c| c.set(false));
                        Err(khive_storage::StorageError::Pool {
                            operation: "reader".into(),
                            message: "injected failure".into(),
                        })
                    } else {
                        reader_result
                    };
                    let mut reader = reader_result.map_err(RuntimeError::Storage)?;
                    reader
                        .query_all(note_sql)
                        .await
                        .map_err(RuntimeError::Storage)?
                };

                let batch_len = note_rows.len();
                note_total += batch_len;
                if batch_len == 0 {
                    break;
                }
                note_cursor = match note_rows.last().and_then(|row| row.columns.first()) {
                    Some(column) => match &column.value {
                        SqlValue::Text(id) => id.clone(),
                        _ => {
                            return Err(RuntimeError::Internal(
                                "backfill note ID is not text".into(),
                            ))
                        }
                    },
                    None => {
                        return Err(RuntimeError::Internal("backfill note page is empty".into()))
                    }
                };

                let mut notes_for_batch = Vec::with_capacity(batch_len);
                let mut inputs = Vec::with_capacity(batch_len);
                for row in &note_rows {
                    let id_str = row.columns.first().and_then(|c| {
                        if let SqlValue::Text(s) = &c.value {
                            Some(s.clone())
                        } else {
                            None
                        }
                    });

                    let Some(id_str) = id_str else {
                        continue;
                    };
                    let Ok(id) = id_str.parse::<Uuid>() else {
                        continue;
                    };

                    let note = match &note_store {
                        Some(store) => match store.get_note(id).await {
                            Ok(Some(n)) => n,
                            _ => continue,
                        },
                        None => continue,
                    };

                    if note.content.trim().is_empty() {
                        continue;
                    }

                    // Repopulate FTS entry using the shared constructor (first model only
                    // to avoid N identical overwrites per note).
                    if model_names.first().map(|n| n.as_str()) == Some(model_name.as_str()) {
                        if let Some(ref ts) = text_store {
                            if let Err(e) = ts.upsert_document(note_fts_document(&note)).await {
                                tracing::warn!(id = %id, error = %e,
                                    "backfill_missing_embeddings: note FTS upsert failed");
                            }
                        }
                    }

                    if !self
                        .embedding_models_for_note_kind(&note.kind)
                        .contains(model_name)
                    {
                        continue;
                    }

                    inputs.push((id, note.content.clone()));
                    notes_for_batch.push(note);
                }

                let outcomes = self.embed_backfill_page(token, model_name, &inputs).await;
                for (note, outcome) in notes_for_batch.into_iter().zip(outcomes) {
                    let Some(outcome) = outcome else { continue };
                    model_truncation.observe(&outcome);
                    match self
                        .publish_note_vector_revision(token, &note, model_name, &outcome.vector)
                        .await
                    {
                        Ok(true) => total_backfilled += 1,
                        Ok(false) => {}
                        Err(error) => tracing::warn!(
                            id = %note.id, model = %model_name, error = %error,
                            "backfill_missing_embeddings: note vector insert failed"
                        ),
                    }
                }

                if batch_len < PAGE_SIZE {
                    break;
                }
            }

            tracing::info!(
                model = %model_name,
                namespace = %ns,
                entities = entity_total,
                notes = note_total,
                truncated = model_truncation.truncated,
                discarded_bytes = model_truncation.discarded_bytes,
                "backfill_missing_embeddings: model pass complete"
            );
        }

        tracing::info!(
            namespace = %ns,
            total_backfilled = total_backfilled,
            "backfill_missing_embeddings: finished"
        );

        Ok(total_backfilled)
    }

    /// Sweep orphaned vector entries for all registered embedding models.
    ///
    /// A vector entry is orphaned when its `subject_id` no longer exists as a
    /// live row in the entity, note, or knowledge-atom tables (i.e. either the
    /// row is absent or has `deleted_at IS NOT NULL`). Orphaned entries
    /// accumulate after hard-deletes because the vector store and SQL
    /// substrate are decoupled.
    ///
    /// Iterates over every registered embedding model and calls
    /// [`khive_storage::VectorStore::orphan_sweep`] for the token's namespace. Models whose
    /// backend returns [`khive_storage::StorageError::Unsupported`] are skipped without error —
    /// this preserves forward-compat when a newly registered model does not yet
    /// implement sweep.
    ///
    /// Returns the total number of vector rows deleted across all models.
    pub async fn sweep_orphan_vectors(
        &self,
        token: &NamespaceToken,
        max_delete_per_model: u32,
        dry_run: bool,
    ) -> RuntimeResult<u64> {
        use khive_storage::types::OrphanSweepConfig;
        use khive_storage::StorageError;

        let model_names = self.registered_embedding_model_names();
        if model_names.is_empty() {
            tracing::debug!("sweep_orphan_vectors: no embedding models registered, skipping");
            return Ok(0);
        }

        let ns = token.namespace().as_str().to_string();
        let mut total_deleted = 0u64;

        for model_name in &model_names {
            let store = match self.vectors_for_model(token, model_name) {
                Ok(s) => s,
                Err(e) => {
                    tracing::warn!(
                        model = %model_name,
                        error = %e,
                        "sweep_orphan_vectors: failed to get vector store, skipping model"
                    );
                    continue;
                }
            };

            let caps = store.capabilities();
            if !caps.supports_orphan_sweep {
                tracing::debug!(
                    model = %model_name,
                    "sweep_orphan_vectors: backend does not support orphan sweep, skipping"
                );
                continue;
            }

            let config = OrphanSweepConfig {
                subject_id_allowlist: None,
                namespaces: vec![ns.clone()],
                substrate_kinds: vec![],
                max_delete: max_delete_per_model,
                dry_run,
            };

            match store.orphan_sweep(&config).await {
                Ok(result) => {
                    tracing::info!(
                        model = %model_name,
                        namespace = %ns,
                        scanned = result.scanned,
                        deleted = result.deleted,
                        would_delete = result.would_delete,
                        dry_run = dry_run,
                        "sweep_orphan_vectors: sweep complete"
                    );
                    total_deleted += result.deleted;
                }
                Err(StorageError::Unsupported { .. }) => {
                    tracing::debug!(
                        model = %model_name,
                        "sweep_orphan_vectors: backend returned Unsupported, skipping"
                    );
                }
                Err(e) => {
                    tracing::warn!(
                        model = %model_name,
                        error = %e,
                        "sweep_orphan_vectors: sweep failed, continuing with other models"
                    );
                }
            }
        }

        tracing::info!(
            namespace = %ns,
            total_deleted = total_deleted,
            dry_run = dry_run,
            "sweep_orphan_vectors: finished"
        );

        Ok(total_deleted)
    }
}

/// Returns `true` when `properties` is a superset of all key-value pairs in `filter`.
///
/// Mirrors the semantics of `khive_pack_kg::handlers::common::props_match` so that the
/// storage-leg predicate is identical to the handler-side post-filter.
pub(crate) fn properties_match(
    properties: Option<&serde_json::Value>,
    filter: &serde_json::Value,
) -> bool {
    let required = match filter.as_object() {
        Some(obj) if !obj.is_empty() => obj,
        _ => return true,
    };
    let actual = match properties.and_then(serde_json::Value::as_object) {
        Some(obj) => obj,
        None => return false,
    };
    required
        .iter()
        .all(|(k, v)| actual.get(k).is_some_and(|av| av == v))
}

/// Score bonus applied when an entity's title is an exact case-insensitive match for
/// the query. Dominates RRF scores (~0.09–0.18 range with k=10) so that an exact
/// name match always ranks above any partial or semantic match.
const EXACT_MATCH_BOOST: f64 = 0.5;

/// Fuse text + vector hits with Reciprocal Rank Fusion (k=10).
///
/// Scoring and the merge of each id's labels come from the shared labelled fusion in
/// `khive-retrieval`. Entity search keeps its own k=10 and exact-match boosting here.
/// Hits in both lists get RRF scores summed. If `query_text` exactly matches
/// (case-insensitive) the title a fused hit carries from the text hits, a bonus of
/// `EXACT_MATCH_BOOST` is added to ensure exact-name matches dominate.
/// Sort by fused score, take top-`limit`.
fn rrf_fuse(
    text_hits: Vec<TextSearchHit>,
    vector_hits: Vec<VectorSearchHit>,
    limit: usize,
    query_text: &str,
) -> Vec<SearchHit> {
    let mut text_arm = Vec::new();
    for (rank, hit) in text_hits.into_iter().enumerate() {
        let label = HitLabel {
            rank,
            signals: SearchSignals {
                vector_similarity: None,
                keyword_score: Some(hit.score),
            },
            source: SearchSource::Text,
            title: hit.title,
            snippet: hit.snippet,
        };
        text_arm.push((hit.subject_id, label));
    }
    let mut vector_arm = Vec::new();
    for (rank, hit) in vector_hits.into_iter().enumerate() {
        let label = HitLabel {
            rank,
            signals: SearchSignals {
                vector_similarity: Some(hit.score),
                keyword_score: None,
            },
            source: SearchSource::Vector,
            title: None,
            snippet: None,
        };
        vector_arm.push((hit.subject_id, label));
    }

    let fused = fuse_labelled(
        vec![text_arm, vector_arm],
        RRF_K,
        combine_leg_first_appearance,
    );

    // The bonus joins the full fused order, so it decides the sort and the cut below.
    let query_lower = query_text.to_lowercase();
    let boost = DeterministicScore::from_f64(EXACT_MATCH_BOOST);
    let mut hits = Vec::with_capacity(fused.len());
    for (entity_id, score, label) in fused {
        let title_lower = label.title.as_deref().map(str::to_lowercase);
        let exact = title_lower.as_deref() == Some(query_lower.as_str());
        let score = if exact { score + boost } else { score };
        hits.push(SearchHit {
            entity_id,
            score,
            rank_score_kind: RankScoreKind::Rrf,
            signals: label.signals,
            source: label.source,
            title: label.title,
            snippet: label.snippet,
        });
    }

    hits.sort_by(|a, b| b.score.cmp(&a.score).then(a.entity_id.cmp(&b.entity_id)));
    hits.truncate(limit);
    hits
}

#[cfg(test)]
mod rrf_fuse_label_tests;

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Arc;

    use crate::runtime::{KhiveRuntime, NamespaceToken, RuntimeConfig};
    use khive_score::rrf_score;
    use khive_storage::types::{TextSearchHit, VectorSearchHit};
    use khive_types::namespace::Namespace;
    use lattice_embed::{EmbedError, EmbeddingModel};

    /// An `EmbeddingService` that always fails — used to drive a real
    /// vector-arm failure (as opposed to an `Unconfigured` short-circuit)
    /// through `embed_query_for_token` without loading actual model weights.
    struct FailingEmbeddingService;

    #[async_trait::async_trait]
    impl EmbeddingService for FailingEmbeddingService {
        async fn embed(
            &self,
            _texts: &[String],
            _model: EmbeddingModel,
        ) -> Result<Vec<Vec<f32>>, EmbedError> {
            Err(EmbedError::ModelInitialization(
                "injected vector-arm failure".to_string(),
            ))
        }

        fn supports_model(&self, _model: EmbeddingModel) -> bool {
            true
        }

        fn name(&self) -> &'static str {
            "hybrid-search-test-failing-embedding"
        }
    }

    struct FailingEmbedderProvider {
        name: String,
        dimensions: usize,
    }

    #[async_trait::async_trait]
    impl EmbedderProvider for FailingEmbedderProvider {
        fn name(&self) -> &str {
            &self.name
        }

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

        async fn build(&self) -> RuntimeResult<Arc<dyn EmbeddingService>> {
            Ok(Arc::new(FailingEmbeddingService))
        }
    }

    /// Swap the runtime's registered embedder for the always-failing one,
    /// re-keyed under the same model name so `hybrid_search`'s vector leg
    /// (which resolves the embedder by the runtime's configured model name)
    /// picks it up. `EmbedderRegistry::register` overwrites by name and
    /// resets the provider's build cache, so this takes effect on the next
    /// embed call without needing a fresh runtime.
    fn break_vector_arm(runtime: &KhiveRuntime) {
        let model = EmbeddingModel::AllMiniLmL6V2;
        runtime.register_embedder(FailingEmbedderProvider {
            name: model.to_string(),
            dimensions: model.dimensions(),
        });
    }

    /// An `EmbeddingService` that always succeeds with a fixed vector — used
    /// to exercise a genuinely healthy vector leg without loading real model
    /// weights.
    struct ConstantEmbeddingService {
        dimensions: usize,
    }

    #[async_trait::async_trait]
    impl EmbeddingService for ConstantEmbeddingService {
        async fn embed(
            &self,
            texts: &[String],
            _model: EmbeddingModel,
        ) -> Result<Vec<Vec<f32>>, EmbedError> {
            Ok(texts.iter().map(|_| vec![1.0; self.dimensions]).collect())
        }

        fn supports_model(&self, _model: EmbeddingModel) -> bool {
            true
        }

        fn name(&self) -> &'static str {
            "hybrid-search-test-constant-embedding"
        }
    }

    struct ConstantEmbedderProvider {
        name: String,
        dimensions: usize,
    }

    #[async_trait::async_trait]
    impl EmbedderProvider for ConstantEmbedderProvider {
        fn name(&self) -> &str {
            &self.name
        }

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

        async fn build(&self) -> RuntimeResult<Arc<dyn EmbeddingService>> {
            Ok(Arc::new(ConstantEmbeddingService {
                dimensions: self.dimensions,
            }))
        }
    }

    /// A runtime configured with a healthy (constant, non-failing) embedder —
    /// the vector leg genuinely runs and succeeds.
    fn runtime_with_constant_embeddings() -> KhiveRuntime {
        let model = EmbeddingModel::AllMiniLmL6V2;
        let runtime = KhiveRuntime::new(RuntimeConfig {
            db_path: None,
            embedding_model: Some(model),
            packs: vec!["kg".to_string()],
            ..RuntimeConfig::no_embeddings()
        })
        .expect("in-memory runtime");
        runtime.register_embedder(ConstantEmbedderProvider {
            name: model.to_string(),
            dimensions: model.dimensions(),
        });
        runtime
    }

    #[test]
    fn bounded_embedding_input_reserves_prefix_and_preserves_utf8() {
        assert_eq!(
            document_embedding_budget("multilingual-e5-base"),
            MAX_TEXT_BYTES - "passage: ".len()
        );

        let input = format!("{}\u{1f980}tail", "a".repeat(MAX_TEXT_BYTES - 1));
        let (bounded, truncated) = bounded_embedding_input(&input, MAX_TEXT_BYTES);
        assert!(truncated);
        assert_eq!(bounded.len(), MAX_TEXT_BYTES - 1);
        assert!(bounded.is_char_boundary(bounded.len()));
        assert!(!bounded.contains('\u{1f980}'));
    }

    #[test]
    fn bounded_embedding_input_leaves_normal_text_unchanged() {
        let input = "normal byte-identical embedding input";
        let (bounded, truncated) = bounded_embedding_input(input, MAX_TEXT_BYTES);
        assert!(!truncated);
        assert_eq!(bounded, input);
        assert_eq!(bounded.as_ptr(), input.as_ptr());
    }

    fn text_hit(id: Uuid, rank: u32, title: &str) -> TextSearchHit {
        TextSearchHit {
            subject_id: id,
            score: DeterministicScore::from_f64(1.0),
            rank,
            title: Some(title.to_string()),
            snippet: Some("...".to_string()),
        }
    }

    fn vector_hit(id: Uuid, rank: u32) -> VectorSearchHit {
        VectorSearchHit {
            subject_id: id,
            score: DeterministicScore::from_f64(0.9),
            rank,
        }
    }

    #[test]
    fn rrf_evidence_keeps_absence_distinct_from_measured_zero() {
        let text_id = Uuid::from_u128(1);
        let vector_id = Uuid::from_u128(2);
        let both_id = Uuid::from_u128(3);
        let quarter = DeterministicScore::from_raw(1_i64 << 30);
        let half = DeterministicScore::from_raw(1_i64 << 31);
        let text = vec![
            TextSearchHit {
                score: DeterministicScore::ZERO,
                ..text_hit(text_id, 1, "text")
            },
            TextSearchHit {
                score: quarter,
                ..text_hit(both_id, 2, "both")
            },
            text_hit(both_id, 3, "duplicate"),
        ];
        let vector = vec![
            VectorSearchHit {
                score: DeterministicScore::ZERO,
                ..vector_hit(vector_id, 1)
            },
            VectorSearchHit {
                score: half,
                ..vector_hit(both_id, 2)
            },
            vector_hit(both_id, 3),
        ];
        let hits = rrf_fuse(text, vector, 10, "unmatched");
        assert_eq!(hits.len(), 3);
        for (id, source, signals) in [
            (
                text_id,
                SearchSource::Text,
                SearchSignals {
                    vector_similarity: None,
                    keyword_score: Some(DeterministicScore::ZERO),
                },
            ),
            (
                vector_id,
                SearchSource::Vector,
                SearchSignals {
                    vector_similarity: Some(DeterministicScore::ZERO),
                    keyword_score: None,
                },
            ),
            (
                both_id,
                SearchSource::Both,
                SearchSignals {
                    vector_similarity: Some(half),
                    keyword_score: Some(quarter),
                },
            ),
        ] {
            let hit = hits.iter().find(|hit| hit.entity_id == id).unwrap();
            assert_eq!(hit.rank_score_kind, RankScoreKind::Rrf);
            assert_eq!(hit.source, source);
            assert_eq!(hit.signals, signals);
        }
    }

    #[test]
    fn rrf_evidence_golden_preserves_true_ties_across_permutations() {
        let a = Uuid::from_u128(1);
        let b = Uuid::from_u128(2);
        let expected = vec![
            (
                a,
                748_365_513,
                RankScoreKind::Rrf,
                SearchSignals {
                    vector_similarity: Some(DeterministicScore::from_raw(1_i64 << 31)),
                    keyword_score: Some(DeterministicScore::from_raw(1_i64 << 30)),
                },
            ),
            (
                b,
                748_365_513,
                RankScoreKind::Rrf,
                SearchSignals {
                    vector_similarity: Some(DeterministicScore::from_raw(1_i64 << 32)),
                    keyword_score: Some(DeterministicScore::from_raw(3_i64 << 30)),
                },
            ),
        ];
        // Swapping both arms preserves each ID's rank-1 plus rank-2 total.
        for (text_ids, vector_ids) in [([a, b], [b, a]), ([b, a], [a, b])] {
            for _ in 0..4 {
                let text = text_ids
                    .into_iter()
                    .enumerate()
                    .map(|(rank, id)| TextSearchHit {
                        score: DeterministicScore::from_raw(
                            if id == a { 1_i64 } else { 3_i64 } << 30,
                        ),
                        ..text_hit(id, rank as u32 + 1, "candidate")
                    })
                    .collect();
                let vector = vector_ids
                    .into_iter()
                    .enumerate()
                    .map(|(rank, id)| VectorSearchHit {
                        score: DeterministicScore::from_raw(
                            if id == a { 1_i64 } else { 2_i64 } << 31,
                        ),
                        ..vector_hit(id, rank as u32 + 1)
                    })
                    .collect();
                let hits = rrf_fuse(text, vector, 10, "unmatched");
                assert_eq!(hits.len(), 2);
                assert_eq!(hits[0].score, hits[1].score);
                assert!(hits.iter().all(|hit| hit.source == SearchSource::Both));
                let snapshot: Vec<_> = hits
                    .iter()
                    .map(|hit| {
                        (
                            hit.entity_id,
                            hit.score.to_raw(),
                            hit.rank_score_kind,
                            hit.signals,
                        )
                    })
                    .collect();
                assert_eq!(snapshot, expected);
            }
        }
    }

    #[test]
    fn rrf_fuse_text_only() {
        let a = Uuid::new_v4();
        let b = Uuid::new_v4();
        let text = vec![text_hit(a, 1, "A"), text_hit(b, 2, "B")];
        let hits = rrf_fuse(text, vec![], 10, "query");
        assert_eq!(hits.len(), 2);
        assert_eq!(hits[0].entity_id, a);
        assert_eq!(hits[0].source, SearchSource::Text);
        assert_eq!(hits[0].title.as_deref(), Some("A"));
    }

    #[test]
    fn rrf_fuse_vector_only() {
        let a = Uuid::new_v4();
        let hits = rrf_fuse(vec![], vec![vector_hit(a, 1)], 10, "query");
        assert_eq!(hits.len(), 1);
        assert_eq!(hits[0].source, SearchSource::Vector);
        assert!(hits[0].title.is_none());
    }

    #[test]
    fn rrf_fuse_marks_both_when_in_both_lists() {
        let id = Uuid::new_v4();
        let text = vec![text_hit(id, 1, "A")];
        let vec = vec![vector_hit(id, 1)];
        let hits = rrf_fuse(text, vec, 10, "query");
        assert_eq!(hits.len(), 1);
        assert_eq!(hits[0].source, SearchSource::Both);
    }

    #[test]
    fn rrf_fuse_preserves_unique_leg_scores_exactly() {
        let text_only = Uuid::new_v4();
        let both = Uuid::new_v4();
        let vector_only = Uuid::new_v4();
        let text = vec![text_hit(text_only, 1, "A"), text_hit(both, 2, "B")];
        let vector = vec![vector_hit(both, 1), vector_hit(vector_only, 2)];

        let hits = rrf_fuse(text, vector, 10, "query");
        let score_for = |id| {
            hits.iter()
                .find(|hit| hit.entity_id == id)
                .expect("expected fused hit")
                .score
        };

        assert_eq!(score_for(text_only), rrf_score(1, RRF_K));
        assert_eq!(score_for(both), rrf_score(2, RRF_K) + rrf_score(1, RRF_K));
        assert_eq!(score_for(vector_only), rrf_score(2, RRF_K));
    }

    #[test]
    fn rrf_fuse_counts_duplicate_once_per_leg() {
        let id = Uuid::new_v4();
        let text = vec![text_hit(id, 1, "A"), text_hit(id, 2, "A duplicate")];
        let vector = vec![vector_hit(id, 1), vector_hit(id, 2)];

        let hits = rrf_fuse(text, vector, 10, "query");

        assert_eq!(hits.len(), 1);
        assert_eq!(hits[0].source, SearchSource::Both);
        assert_eq!(hits[0].score, rrf_score(1, RRF_K) + rrf_score(1, RRF_K));
    }

    #[test]
    fn rrf_fuse_respects_limit() {
        let hits: Vec<TextSearchHit> = (0..20)
            .map(|i| text_hit(Uuid::new_v4(), i + 1, "x"))
            .collect();
        let fused = rrf_fuse(hits, vec![], 5, "query");
        assert_eq!(fused.len(), 5);
    }

    #[test]
    fn rrf_fuse_orders_higher_score_first() {
        // Same UUID in both lists at rank 1 → score 2/(10+1). Different UUIDs → 1/(10+1) each.
        let a = Uuid::new_v4();
        let b = Uuid::new_v4();
        let text = vec![text_hit(a, 1, "A")];
        let vec = vec![vector_hit(a, 1), vector_hit(b, 2)];
        let hits = rrf_fuse(text, vec, 10, "query");
        assert_eq!(hits[0].entity_id, a);
        assert_eq!(hits[0].source, SearchSource::Both);
        assert!(hits[0].score > hits[1].score);
    }

    #[test]
    fn rrf_fuse_k10_score_spread_exceeds_threshold() {
        // With k=10: rank 1 → 1/11 ≈ 0.0909, rank 10 → 1/20 = 0.0500.
        // Spread ≈ 0.041, well above the 0.03 minimum required for reliable dedup.
        let ids: Vec<Uuid> = (0..10).map(|_| Uuid::new_v4()).collect();
        let text: Vec<TextSearchHit> = ids
            .iter()
            .enumerate()
            .map(|(i, &id)| text_hit(id, (i + 1) as u32, "x"))
            .collect();
        let hits = rrf_fuse(text, vec![], 10, "query");
        assert_eq!(hits.len(), 10);
        let top_score = hits[0].score.to_f64();
        let bottom_score = hits[9].score.to_f64();
        let spread = top_score - bottom_score;
        assert!(
            spread >= 0.03,
            "score spread {spread:.4} between rank 1 and rank 10 must be ≥ 0.03 (was {spread:.4})"
        );
    }

    #[test]
    fn rrf_fuse_exact_match_boost_elevates_score() {
        // An entity whose title exactly matches the query should receive a score
        // significantly above a non-matching entity ranked first by text search.
        let exact_id = Uuid::new_v4();
        let other_id = Uuid::new_v4();
        // other_id ranks 1 in text, exact_id ranks 2 — but exact_id matches query.
        let text = vec![
            text_hit(other_id, 1, "something else"),
            text_hit(exact_id, 2, "FlashAttention"),
        ];
        let hits = rrf_fuse(text, vec![], 10, "flashattention");
        assert_eq!(hits.len(), 2);
        assert_eq!(
            hits[0].entity_id, exact_id,
            "exact match must rank first despite being rank-2 in raw text search"
        );
    }

    // ---- embed_batch tests ----

    #[test]
    fn embed_batch_unconfigured_on_memory_runtime() {
        // KhiveRuntime::memory() has no embedding model — embed_batch returns Unconfigured.
        let rt = KhiveRuntime::memory().unwrap();
        let result = tokio::runtime::Runtime::new()
            .unwrap()
            .block_on(rt.embed_batch(&[]));
        // Empty slice short-circuits before hitting the model check.
        assert!(result.is_ok());
        assert!(result.unwrap().is_empty());
    }

    #[test]
    fn embed_batch_empty_input_returns_empty_vec() {
        // No model needed — empty slice is handled before the embedder is touched.
        let rt = KhiveRuntime::memory().unwrap();
        let result = tokio::runtime::Runtime::new()
            .unwrap()
            .block_on(rt.embed_batch(&[]));
        assert_eq!(result.unwrap(), Vec::<Vec<f32>>::new());
    }

    #[test]
    fn embed_batch_no_model_non_empty_returns_unconfigured() {
        let rt = KhiveRuntime::memory().unwrap();
        let texts = vec!["hello".to_string()];
        let result = tokio::runtime::Runtime::new()
            .unwrap()
            .block_on(rt.embed_batch(&texts));
        match result {
            Err(crate::RuntimeError::Unconfigured(s)) => assert_eq!(s, "embedding_model"),
            Err(other) => panic!("expected Unconfigured, got {:?}", other),
            Ok(_) => panic!("expected Err, got Ok"),
        }
    }

    #[test]
    #[ignore = "loads ~80 MB model; run with --include-ignored"]
    fn embed_batch_count_matches_input() {
        let config = RuntimeConfig {
            db_path: None,
            default_namespace: Namespace::parse("test").unwrap(),
            embedding_model: Some(EmbeddingModel::AllMiniLmL6V2),
            packs: vec!["kg".to_string()],
            ..RuntimeConfig::default()
        };
        let rt = KhiveRuntime::new(config).unwrap();
        let texts: Vec<String> = vec!["foo".to_string(), "bar".to_string(), "baz".to_string()];
        let result = tokio::runtime::Runtime::new()
            .unwrap()
            .block_on(rt.embed_batch(&texts));
        let embeddings = result.unwrap();
        assert_eq!(embeddings.len(), texts.len());
    }

    #[test]
    fn vector_search_requires_embedding_or_text() {
        let rt = KhiveRuntime::memory().unwrap();
        let tok = NamespaceToken::local();
        let result = tokio::runtime::Runtime::new()
            .unwrap()
            .block_on(rt.vector_search(&tok, None, None, 10, Some(SubstrateKind::Entity)));
        match result {
            Err(crate::RuntimeError::InvalidInput(msg)) => {
                assert!(msg.contains("query_embedding or query_text"), "msg: {msg}");
            }
            other => panic!("expected InvalidInput, got {other:?}"),
        }
    }

    #[test]
    fn vector_search_text_without_model_returns_unconfigured() {
        let rt = KhiveRuntime::memory().unwrap();
        let tok = NamespaceToken::local();
        let result = tokio::runtime::Runtime::new()
            .unwrap()
            .block_on(rt.vector_search(
                &tok,
                None,
                Some("attention"),
                10,
                Some(SubstrateKind::Entity),
            ));
        match result {
            Err(crate::RuntimeError::Unconfigured(s)) => assert_eq!(s, "embedding_model"),
            other => panic!("expected Unconfigured, got {other:?}"),
        }
    }

    #[test]
    #[ignore = "loads ~80 MB model; run with --include-ignored"]
    fn embed_batch_vectors_have_expected_dimensions() {
        let model = EmbeddingModel::AllMiniLmL6V2;
        let config = RuntimeConfig {
            db_path: None,
            default_namespace: Namespace::parse("test").unwrap(),
            embedding_model: Some(model),
            packs: vec!["kg".to_string()],
            ..RuntimeConfig::default()
        };
        let rt = KhiveRuntime::new(config).unwrap();
        let texts = vec!["hello world".to_string()];
        let result = tokio::runtime::Runtime::new()
            .unwrap()
            .block_on(rt.embed_batch(&texts));
        let embeddings = result.unwrap();
        assert_eq!(embeddings[0].len(), model.dimensions());
    }

    // ---- hybrid_search_outcome: vector-arm failure must not discard text hits ----

    /// Baseline (red without the fix): the fail-loud `hybrid_search` must
    /// still propagate a vector-arm failure as a whole-call error, even
    /// though the text leg found a match — proving `break_vector_arm`
    /// genuinely exercises the vector leg and that `hybrid_search_outcome`'s
    /// tolerance below is a real behavioral difference, not a no-op.
    #[tokio::test]
    async fn hybrid_search_still_fails_loud_on_vector_arm_error() {
        let rt = runtime_with_constant_embeddings();
        let tok = NamespaceToken::local();
        rt.create_entity(
            &tok,
            "concept",
            None,
            "FlashAttention",
            Some("IO-aware exact attention using tiling"),
            None,
            vec![],
        )
        .await
        .unwrap();
        break_vector_arm(&rt);

        let result = rt
            .hybrid_search(&tok, "FlashAttention", None, 10, None, None, &[], None)
            .await;

        assert!(
            result.is_err(),
            "the fail-loud entry point must still propagate a vector-arm failure, got {result:?}"
        );
    }

    /// Green: the coordinator-only outcome variant preserves the text hit and
    /// reports the vector failure separately instead of discarding the whole
    /// call.
    #[tokio::test]
    async fn hybrid_search_outcome_preserves_text_hits_on_vector_arm_error() {
        let rt = runtime_with_constant_embeddings();
        let tok = NamespaceToken::local();
        rt.create_entity(
            &tok,
            "concept",
            None,
            "FlashAttention",
            Some("IO-aware exact attention using tiling"),
            None,
            vec![],
        )
        .await
        .unwrap();
        break_vector_arm(&rt);

        let outcome = rt
            .hybrid_search_outcome(&tok, "FlashAttention", 10, None, None, &[], None)
            .await
            .expect("text leg must still succeed");

        assert!(
            !outcome.hits.is_empty(),
            "text arm's hit must survive a vector-arm failure"
        );
        assert!(
            outcome.hits[0]
                .title
                .as_deref()
                .unwrap_or_default()
                .contains("FlashAttention"),
            "surviving hit must be the text match"
        );
        let vector_error = outcome
            .vector_error
            .expect("vector arm failure must be reported");
        assert!(
            vector_error.contains("injected vector-arm failure"),
            "vector_error must carry the underlying cause, got {vector_error:?}"
        );
    }

    /// A healthy vector arm still leaves `vector_error` `None` — the outcome
    /// variant does not manufacture a failure where there is none.
    #[tokio::test]
    async fn hybrid_search_outcome_has_no_vector_error_when_vector_arm_healthy() {
        let rt = runtime_with_constant_embeddings();
        let tok = NamespaceToken::local();
        rt.create_entity(
            &tok,
            "concept",
            None,
            "FlashAttention",
            Some("IO-aware exact attention using tiling"),
            None,
            vec![],
        )
        .await
        .unwrap();

        let outcome = rt
            .hybrid_search_outcome(&tok, "FlashAttention", 10, None, None, &[], None)
            .await
            .expect("hybrid search must succeed");

        assert!(!outcome.hits.is_empty(), "should find the entity");
        assert!(
            outcome.vector_error.is_none(),
            "a healthy vector arm must not report an error"
        );
    }

    // ---- hybrid_search enrichment ----

    #[tokio::test]
    async fn hybrid_search_entity_hit_has_title() {
        let rt = KhiveRuntime::memory().unwrap();
        let tok = NamespaceToken::local();
        rt.create_entity(
            &tok,
            "concept",
            None,
            "FlashAttention",
            Some("IO-aware exact attention using tiling"),
            None,
            vec![],
        )
        .await
        .unwrap();

        let hits = rt
            .hybrid_search(&tok, "FlashAttention", None, 10, None, None, &[], None)
            .await
            .unwrap();

        assert!(!hits.is_empty(), "should find the entity");
        let hit = &hits[0];
        assert!(hit.title.is_some(), "title must be populated");
        assert!(
            hit.title.as_deref().unwrap().contains("FlashAttention"),
            "title must contain entity name"
        );
    }

    /// `hybrid_search` must not hard-fail on a query containing FTS5 metacharacters
    /// like `$` (e.g. the DSL doc query `$prev.id`). `sanitize_fts5_query` (khive-db)
    /// strips `$`, so this exercises the sanitizer path and takes the `Ok` arm; see
    /// `hybrid_search_with_residual_fts5_char_now_sanitized` below for the character
    /// class #916 closed (previously the fail-loud arm, prior to #916).
    #[tokio::test]
    async fn hybrid_search_with_dollar_sign_query_does_not_error() {
        let rt = KhiveRuntime::memory().unwrap();
        let tok = NamespaceToken::local();
        rt.create_entity(
            &tok,
            "concept",
            None,
            "DSL docs",
            Some("use $prev.id to chain calls"),
            None,
            vec![],
        )
        .await
        .unwrap();

        let result = rt
            .hybrid_search(&tok, "$prev.id", None, 10, None, None, &[], None)
            .await;

        assert!(
            result.is_ok(),
            "#388 hybrid_search must not hard-fail on a '$'-bearing query, got: {:?}",
            result.err()
        );
    }

    /// #916: `@` was previously NOT stripped by `sanitize_fts5_query` and SQLite
    /// FTS5's bareword parser rejected it unconditionally, so this query used to
    /// reach the runtime-level fail-loud arm (`RuntimeError::InvalidInput`, #569).
    /// `sanitize_fts5_token_group`'s bareword-safety gate now recognizes `@` (and
    /// every other ASCII punctuation character not already handled) as unsafe for
    /// an unquoted bareword position and routes it through the quoted-phrase
    /// alternative instead, which FTS5 accepts literally, so the query
    /// now succeeds and finds the seeded content rather than erroring.
    #[tokio::test]
    async fn hybrid_search_with_residual_fts5_char_now_sanitized() {
        let rt = KhiveRuntime::memory().unwrap();
        let tok = NamespaceToken::local();
        rt.create_entity(
            &tok,
            "concept",
            None,
            "DSL docs",
            Some("use foo@bar to chain calls"),
            None,
            vec![],
        )
        .await
        .unwrap();

        let result = rt
            .hybrid_search(&tok, "foo@bar", None, 10, None, None, &[], None)
            .await;

        let hits = result.unwrap_or_else(|e| {
            panic!("#916 hybrid_search must not fail on an '@'-bearing query, got: {e:?}")
        });
        assert!(
            !hits.is_empty(),
            "#916 '@'-bearing query must still find the seeded 'foo@bar' content via the \
             quoted-phrase alternative"
        );
    }

    /// #916 end-to-end regression, using the exact character classes from the
    /// issue's live-log evidence (`#682 Stage 2`, `Min-K%Prob`, `B=128`):
    /// `hybrid_search`'s FTS leg must not lose its lexical signal to a parser
    /// syntax error on `#`, `%`, or `=`. Each query below must both succeed
    /// and actually surface a `Text`/`Both`-sourced hit, proving the FTS leg
    /// contributed, not just that the vector leg papered over a degraded
    /// text leg.
    #[tokio::test]
    async fn hybrid_search_with_916_issue_characters_finds_text_leg_hits() {
        let rt = KhiveRuntime::memory().unwrap();
        let tok = NamespaceToken::local();

        rt.create_entity(
            &tok,
            "concept",
            None,
            "issue tracker",
            Some("tracking #682 Stage 2: MoE expert-cache prefetch work"),
            None,
            vec![],
        )
        .await
        .unwrap();
        rt.create_entity(
            &tok,
            "concept",
            None,
            "benchmark notes",
            Some("chunkwise B=128 traffic arithmetic simdgroup_matrix DPLR"),
            None,
            vec![],
        )
        .await
        .unwrap();
        rt.create_entity(
            &tok,
            "concept",
            None,
            "sampling notes",
            Some("evaluated with the Min-K%Prob membership inference method"),
            None,
            vec![],
        )
        .await
        .unwrap();

        for query in ["#682 Stage 2", "B=128", "Min-K%Prob"] {
            let result = rt
                .hybrid_search(&tok, query, None, 10, None, None, &[], None)
                .await;
            let hits = result.unwrap_or_else(|e| {
                panic!("#916 hybrid_search must not fail on query {query:?}, got: {e:?}")
            });
            assert!(
                hits.iter()
                    .any(|h| matches!(h.source, SearchSource::Text | SearchSource::Both)),
                "#916 query {query:?} must surface a Text/Both-sourced hit \
                 (the FTS leg must contribute, not just the vector leg); got {hits:?}"
            );
        }
    }

    // ---- predicate pushdown before truncation ----

    /// Entity-branch tag-filter regression.
    ///
    /// Scenario: `limit=1`, tag_filter=["target-tag"]. Two entities are inserted:
    ///   - "decoy_alpha_beta_gamma": many query tokens → ranks 1 in FTS (dominates).
    ///     Does NOT have "target-tag".
    ///   - "alpha_beta_gamma target": fewer query tokens → ranks 2 in FTS.
    ///     HAS "target-tag".
    ///
    /// Without predicate pushdown, `fused.truncate(1)` keeps only the decoy and the
    /// tag-matching entity is invisible: this requires `tags_any` to be passed into
    /// `query_entities`'s `EntityFilter` so the decoy is excluded before truncation.
    #[tokio::test]
    async fn hybrid_search_tag_filter_pushed_before_truncation() {
        let rt = KhiveRuntime::memory().unwrap();
        let tok = NamespaceToken::local();

        // Decoy: high-ranking FTS hit (content repeats query words), no target tag.
        rt.create_entity(
            &tok,
            "concept",
            None,
            "alpha beta gamma decoy alpha beta gamma",
            Some("alpha beta gamma decoy description alpha beta gamma"),
            None,
            vec!["other-tag".to_string()],
        )
        .await
        .unwrap();

        // Target: lower-ranking FTS hit, has the tag we filter on.
        let target = rt
            .create_entity(
                &tok,
                "concept",
                None,
                "alpha beta gamma target",
                Some("alpha beta gamma target description"),
                None,
                vec!["target-tag".to_string()],
            )
            .await
            .unwrap();

        // With limit=1 and tag_filter, the fix must return the target entity despite
        // the decoy ranking higher. Without pushdown, the decoy occupies the single
        // slot and the target is silently dropped.
        let hits = rt
            .hybrid_search(
                &tok,
                "alpha beta gamma",
                None,
                1,
                None,
                None,
                &["target-tag".to_string()],
                None,
            )
            .await
            .unwrap();

        assert_eq!(
            hits.len(),
            1,
            "exactly one hit expected (the tag-matching entity)"
        );
        assert_eq!(
            hits[0].entity_id, target.id,
            "the tag-filtered entity must be returned even when ranked below limit in raw fusion"
        );
    }

    /// Entity-branch properties-filter regression (analogous to the tag-filter test above).
    ///
    /// Scenario: `limit=1`, properties_filter={{"domain": "target"}}. Two entities:
    ///   - decoy: high FTS rank, properties {{"domain": "other"}}.
    ///   - target: lower FTS rank, properties {{"domain": "target"}}.
    ///
    /// Without pushdown: decoy fills the slot, target is dropped. With pushdown:
    /// only the target survives the properties filter before truncation.
    #[tokio::test]
    async fn hybrid_search_props_filter_pushed_before_truncation() {
        let rt = KhiveRuntime::memory().unwrap();
        let tok = NamespaceToken::local();

        rt.create_entity(
            &tok,
            "concept",
            None,
            "delta epsilon zeta decoy delta epsilon zeta",
            Some("delta epsilon zeta decoy description delta epsilon zeta"),
            Some(serde_json::json!({"domain": "other"})),
            vec![],
        )
        .await
        .unwrap();

        let target = rt
            .create_entity(
                &tok,
                "concept",
                None,
                "delta epsilon zeta target",
                Some("delta epsilon zeta target description"),
                Some(serde_json::json!({"domain": "target"})),
                vec![],
            )
            .await
            .unwrap();

        let filter = serde_json::json!({"domain": "target"});
        let hits = rt
            .hybrid_search(
                &tok,
                "delta epsilon zeta",
                None,
                1,
                None,
                None,
                &[],
                Some(&filter),
            )
            .await
            .unwrap();

        assert_eq!(hits.len(), 1, "exactly one hit expected (properties match)");
        assert_eq!(
            hits[0].entity_id, target.id,
            "the properties-filtered entity must be returned even when ranked below limit"
        );
    }

    /// Entity-branch kind-filter regression.
    ///
    /// Scenario: `limit=1`, so the text arm fetches 4 candidates, and
    /// `entity_kind="document"`. Twelve `concept` entities repeat the query term and
    /// outrank the one `document` entity that mentions it once in a long description.
    ///
    /// When the kind is applied only after the text arm has picked its candidates,
    /// the filter sees four concepts and the document is never returned. The kind
    /// has to reach the text query so the candidate budget is spent on documents.
    #[tokio::test]
    async fn hybrid_search_entity_kind_filter_pushed_into_text_arm() {
        let rt = KhiveRuntime::memory().unwrap();
        let tok = NamespaceToken::local();

        for i in 0..12 {
            rt.create_entity(
                &tok,
                "concept",
                None,
                &format!("quillfeather decoy {i}"),
                Some("quillfeather quillfeather quillfeather"),
                None,
                vec![],
            )
            .await
            .unwrap();
        }

        let description = format!(
            "A long administrative description that mentions quillfeather once. {}",
            "Unrelated filing, scheduling and review words. ".repeat(20)
        );
        let target = rt
            .create_entity(
                &tok,
                "document",
                None,
                "Archive filing report",
                Some(description.as_str()),
                None,
                vec![],
            )
            .await
            .unwrap();

        // Premise: without a kind filter the target ranks behind more entities than
        // the 4 candidates the filtered `limit=1` search requests from the text arm.
        let unfiltered = rt
            .hybrid_search(&tok, "quillfeather", None, 13, None, None, &[], None)
            .await
            .unwrap();
        let target_rank = unfiltered
            .iter()
            .position(|hit| hit.entity_id == target.id)
            .expect("the document is found without a kind filter");
        assert!(
            target_rank >= 4,
            "the document must rank behind the candidate window, got rank {target_rank}"
        );

        let hits = rt
            .hybrid_search(
                &tok,
                "quillfeather",
                None,
                1,
                Some("document"),
                None,
                &[],
                None,
            )
            .await
            .unwrap();

        assert_eq!(
            hits.len(),
            1,
            "the kind-filtered search must return the matching document"
        );
        assert_eq!(
            hits[0].entity_id, target.id,
            "the returned hit must be the document, not a concept"
        );
    }

    /// `hybrid_search_each_kind` gives each kind the list `hybrid_search` returns for it.
    ///
    /// Scenario: twenty `concept` entities repeat the query term and outrank three
    /// `document` entities that mention it once, with `limit=2`. The text arm fetches 8
    /// candidates per kind, so the concepts alone overflow it and the cut to `limit`
    /// binds. A text budget shared by the two kinds would leave the documents none; each
    /// kind has to keep its own, so both lists equal the per-kind search.
    #[tokio::test]
    async fn hybrid_search_each_kind_matches_per_kind_search_when_one_kind_dominates() {
        let rt = KhiveRuntime::memory().unwrap();
        let tok = NamespaceToken::local();

        for i in 0..20 {
            rt.create_entity(
                &tok,
                "concept",
                None,
                &format!("quillfeather decoy {i}"),
                Some("quillfeather quillfeather quillfeather"),
                None,
                vec![],
            )
            .await
            .unwrap();
        }
        for i in 0..3 {
            let description = format!(
                "A long administrative description {i} that mentions quillfeather once. {}",
                "Unrelated filing, scheduling and review words. ".repeat(20)
            );
            rt.create_entity(
                &tok,
                "document",
                None,
                &format!("Archive filing report {i}"),
                Some(description.as_str()),
                None,
                vec![],
            )
            .await
            .unwrap();
        }

        let kinds = ["concept", "document"];
        let mut expected: Vec<Vec<Uuid>> = Vec::new();
        for kind in kinds {
            let hits = rt
                .hybrid_search(&tok, "quillfeather", None, 2, Some(kind), None, &[], None)
                .await
                .unwrap();
            expected.push(hits.iter().map(|hit| hit.entity_id).collect());
        }
        assert_eq!(
            expected[0].len(),
            2,
            "premise: the concept list is cut to limit"
        );
        assert_eq!(
            expected[1].len(),
            2,
            "premise: the document list is not starved"
        );

        let per_kind = rt
            .hybrid_search_each_kind(&tok, "quillfeather", None, 2, &kinds)
            .await
            .unwrap();
        let mut actual: Vec<Vec<Uuid>> = Vec::new();
        for hits in &per_kind {
            actual.push(hits.iter().map(|hit| hit.entity_id).collect());
        }
        assert_eq!(
            actual, expected,
            "each kind must get the ids and order its own per-kind search returns"
        );
    }

    // ---- embed intent tests ----

    struct CapturingEmbeddingService {
        captured: std::sync::Arc<std::sync::Mutex<Vec<Vec<String>>>>,
    }

    #[async_trait::async_trait]
    impl EmbeddingService for CapturingEmbeddingService {
        async fn embed(
            &self,
            texts: &[String],
            _model: EmbeddingModel,
        ) -> std::result::Result<Vec<Vec<f32>>, lattice_embed::EmbedError> {
            self.captured.lock().unwrap().push(texts.to_vec());
            Ok(texts.iter().map(|_| vec![1.0]).collect())
        }

        fn supports_model(&self, _model: EmbeddingModel) -> bool {
            true
        }

        fn name(&self) -> &'static str {
            "capturing-embedding-service"
        }
    }

    struct CapturingEmbedderProvider {
        name: String,
        captured: std::sync::Arc<std::sync::Mutex<Vec<Vec<String>>>>,
    }

    struct RewritingPassageService {
        captured: std::sync::Arc<std::sync::Mutex<Vec<Vec<String>>>>,
    }

    #[async_trait::async_trait]
    impl EmbeddingService for RewritingPassageService {
        async fn embed(
            &self,
            texts: &[String],
            _model: EmbeddingModel,
        ) -> std::result::Result<Vec<Vec<f32>>, lattice_embed::EmbedError> {
            self.captured.lock().unwrap().push(texts.to_vec());
            Ok(texts.iter().map(|_| vec![1.0]).collect())
        }

        async fn embed_passage(
            &self,
            texts: &[String],
            model: EmbeddingModel,
        ) -> std::result::Result<Vec<Vec<f32>>, lattice_embed::EmbedError> {
            let prepared: Vec<String> = texts.iter().map(|text| format!("custom:{text}")).collect();
            self.embed(&prepared, model).await
        }

        fn supports_model(&self, _model: EmbeddingModel) -> bool {
            true
        }

        fn name(&self) -> &'static str {
            "rewriting-passage-service"
        }
    }

    struct RewritingPassageProvider {
        name: String,
        captured: std::sync::Arc<std::sync::Mutex<Vec<Vec<String>>>>,
    }

    #[async_trait::async_trait]
    impl EmbedderProvider for RewritingPassageProvider {
        fn name(&self) -> &str {
            &self.name
        }

        fn dimensions(&self) -> usize {
            1
        }

        async fn build(&self) -> crate::error::RuntimeResult<std::sync::Arc<dyn EmbeddingService>> {
            Ok(std::sync::Arc::new(RewritingPassageService {
                captured: std::sync::Arc::clone(&self.captured),
            }))
        }
    }

    struct WrongCardinalityEmbeddingService;

    #[async_trait::async_trait]
    impl EmbeddingService for WrongCardinalityEmbeddingService {
        async fn embed(
            &self,
            texts: &[String],
            _model: EmbeddingModel,
        ) -> std::result::Result<Vec<Vec<f32>>, lattice_embed::EmbedError> {
            Ok(texts
                .iter()
                .take(texts.len().saturating_sub(1))
                .map(|_| vec![1.0])
                .collect())
        }

        fn supports_model(&self, _model: EmbeddingModel) -> bool {
            true
        }

        fn name(&self) -> &'static str {
            "wrong-cardinality-embedding-service"
        }
    }

    struct WrongCardinalityEmbedderProvider;

    #[async_trait::async_trait]
    impl EmbedderProvider for WrongCardinalityEmbedderProvider {
        fn name(&self) -> &str {
            "wrong-cardinality-embedding-service"
        }

        fn dimensions(&self) -> usize {
            1
        }

        async fn build(&self) -> crate::error::RuntimeResult<std::sync::Arc<dyn EmbeddingService>> {
            Ok(std::sync::Arc::new(WrongCardinalityEmbeddingService))
        }
    }

    struct SurplusCardinalityEmbeddingService;

    #[async_trait::async_trait]
    impl EmbeddingService for SurplusCardinalityEmbeddingService {
        async fn embed(
            &self,
            texts: &[String],
            _model: EmbeddingModel,
        ) -> std::result::Result<Vec<Vec<f32>>, lattice_embed::EmbedError> {
            let mut vectors: Vec<Vec<f32>> = texts.iter().map(|_| vec![1.0]).collect();
            vectors.push(vec![1.0]);
            Ok(vectors)
        }

        fn supports_model(&self, _model: EmbeddingModel) -> bool {
            true
        }

        fn name(&self) -> &'static str {
            "surplus-cardinality-embedding-service"
        }
    }

    struct SurplusCardinalityEmbedderProvider;

    #[async_trait::async_trait]
    impl EmbedderProvider for SurplusCardinalityEmbedderProvider {
        fn name(&self) -> &str {
            "surplus-cardinality-embedding-service"
        }

        fn dimensions(&self) -> usize {
            1
        }

        async fn build(&self) -> crate::error::RuntimeResult<std::sync::Arc<dyn EmbeddingService>> {
            Ok(std::sync::Arc::new(SurplusCardinalityEmbeddingService))
        }
    }

    #[async_trait::async_trait]
    impl EmbedderProvider for CapturingEmbedderProvider {
        fn name(&self) -> &str {
            &self.name
        }

        fn dimensions(&self) -> usize {
            1
        }

        async fn build(&self) -> crate::error::RuntimeResult<std::sync::Arc<dyn EmbeddingService>> {
            Ok(std::sync::Arc::new(CapturingEmbeddingService {
                captured: std::sync::Arc::clone(&self.captured),
            }))
        }
    }

    fn runtime_with_capturing_embedder(
        model: EmbeddingModel,
    ) -> (
        KhiveRuntime,
        std::sync::Arc<std::sync::Mutex<Vec<Vec<String>>>>,
    ) {
        let runtime = KhiveRuntime::memory().unwrap();
        let captured = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
        runtime.register_embedder(CapturingEmbedderProvider {
            name: model.to_string(),
            captured: std::sync::Arc::clone(&captured),
        });
        (runtime, captured)
    }

    #[tokio::test]
    async fn bge_query_paths_pass_raw_unprefixed_text() {
        const BGE_QUERY_INSTRUCTION: &str =
            "Represent this sentence for searching relevant passages: ";
        let single = "single raw query";
        let batch = vec![
            "first raw query".to_string(),
            "second raw query".to_string(),
        ];

        for model in [
            EmbeddingModel::BgeSmallEnV15,
            EmbeddingModel::BgeBaseEnV15,
            EmbeddingModel::BgeLargeEnV15,
        ] {
            let (runtime, captured) = runtime_with_capturing_embedder(model);
            runtime
                .embed_query_with_model(&model.to_string(), single)
                .await
                .unwrap();
            runtime
                .embed_query_batch_with_model(&model.to_string(), &batch)
                .await
                .unwrap();

            let calls = captured.lock().unwrap().clone();
            assert_eq!(
                calls,
                vec![vec![single.to_string()], batch.clone()],
                "{model} must receive raw query text through single and batch paths"
            );
            assert!(
                calls
                    .iter()
                    .flatten()
                    .all(|text| !text.contains(BGE_QUERY_INSTRUCTION)),
                "{model} must not receive the BGE retrieval instruction"
            );
        }
    }

    #[tokio::test]
    async fn e5_query_paths_apply_query_prefix() {
        let model = EmbeddingModel::MultilingualE5Small;
        let single = "single raw query";
        let batch = vec![
            "first raw query".to_string(),
            "second raw query".to_string(),
        ];
        let (runtime, captured) = runtime_with_capturing_embedder(model);

        runtime
            .embed_query_with_model(&model.to_string(), single)
            .await
            .unwrap();
        runtime
            .embed_query_batch_with_model(&model.to_string(), &batch)
            .await
            .unwrap();

        assert_eq!(
            captured.lock().unwrap().as_slice(),
            [
                vec!["query: single raw query".to_string()],
                vec![
                    "query: first raw query".to_string(),
                    "query: second raw query".to_string(),
                ],
            ],
            "E5 must receive its query prefix through single and batch paths"
        );
    }

    #[tokio::test]
    async fn mixed_document_batch_stays_one_ordered_provider_call() {
        let model = EmbeddingModel::AllMiniLmL6V2;
        let (runtime, captured) = runtime_with_capturing_embedder(model);
        let texts = vec![
            "first normal document".to_string(),
            "x".repeat(MAX_TEXT_BYTES + 1),
            "second normal document".to_string(),
        ];

        let outcomes = runtime
            .embed_document_batch_with_model_outcomes(&model.to_string(), &texts)
            .await
            .expect("mixed batch must embed");
        assert_eq!(outcomes.len(), texts.len());
        assert!(!outcomes[0].truncated);
        assert_eq!(outcomes[0].source_bytes, outcomes[0].embedded_bytes);
        assert!(outcomes[1].truncated);
        assert_eq!(outcomes[1].source_bytes, MAX_TEXT_BYTES + 1);
        assert_eq!(outcomes[1].embedded_bytes, MAX_TEXT_BYTES);
        assert!(!outcomes[2].truncated);
        assert_eq!(
            captured.lock().unwrap().as_slice(),
            [vec![
                texts[0].clone(),
                "x".repeat(MAX_TEXT_BYTES),
                texts[2].clone(),
            ]]
        );
    }

    #[tokio::test]
    async fn canonical_name_override_has_no_exact_input_fingerprint() {
        let model = EmbeddingModel::MultilingualE5Small;
        let model_name = model.to_string();
        let (runtime, captured) = runtime_with_capturing_embedder(model);
        let input = "x".repeat(document_embedding_budget(&model_name) + 3);
        let outcomes = runtime
            .embed_document_batch_with_model_outcomes(&model_name, std::slice::from_ref(&input))
            .await
            .unwrap();
        let calls = captured.lock().unwrap();
        let actual_input = &calls[0][0];
        assert!(actual_input.starts_with("passage: "));
        assert_eq!(actual_input.len(), MAX_TEXT_BYTES);
        assert!(outcomes[0].truncated);
        assert_eq!(outcomes[0].prepared_text_fingerprint, None);
        assert_eq!(
            prepared_document_fingerprint(
                bounded_embedding_input(&input, document_embedding_budget(&model_name)).0,
                model,
            ),
            VectorRecord::fingerprint_text(actual_input),
            "the audited lattice path hashes the bounded text plus passage prefix"
        );
    }

    #[tokio::test]
    async fn ordinary_custom_document_provider_has_no_exact_input_fingerprint() {
        let runtime = KhiveRuntime::memory().unwrap();
        let captured = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
        runtime.register_embedder(CapturingEmbedderProvider {
            name: "custom-document-provider".to_string(),
            captured: std::sync::Arc::clone(&captured),
        });

        let outcome = runtime
            .embed_document_with_model_outcome("custom-document-provider", "document")
            .await
            .expect("custom provider must embed");
        assert_eq!(outcome.prepared_text_fingerprint, None);
        assert_eq!(captured.lock().unwrap().len(), 1);
    }

    #[tokio::test]
    async fn custom_passage_override_under_builtin_name_cannot_claim_exact_input() {
        let runtime = KhiveRuntime::memory().unwrap();
        let captured = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
        let model = EmbeddingModel::MultilingualE5Small;
        let model_name = model.to_string();
        runtime.register_embedder(RewritingPassageProvider {
            name: model_name.clone(),
            captured: std::sync::Arc::clone(&captured),
        });

        let outcome = runtime
            .embed_document_with_model_outcome(&model_name, "document")
            .await
            .expect("custom override must embed");
        assert_eq!(outcome.prepared_text_fingerprint, None);
        assert_eq!(
            captured.lock().unwrap()[0],
            vec!["custom:document".to_string()]
        );
        assert_ne!(
            prepared_document_fingerprint("document", model),
            VectorRecord::fingerprint_text("custom:document")
        );
    }

    #[tokio::test]
    async fn truncated_document_batch_rejects_provider_cardinality_mismatch() {
        let runtime = KhiveRuntime::memory().unwrap();
        runtime.register_embedder(WrongCardinalityEmbedderProvider);
        let model_name = "wrong-cardinality-embedding-service";
        let texts = vec!["x".repeat(MAX_TEXT_BYTES + 1), "normal".to_string()];

        let error = runtime
            .embed_document_batch_with_model_outcomes(model_name, &texts)
            .await
            .expect_err("provider cardinality mismatch must fail the whole batch");

        assert!(
            error
                .to_string()
                .contains("embed_passage returned 1 vectors for 2 inputs"),
            "unexpected error: {error}"
        );
    }

    #[tokio::test]
    async fn singleton_document_embed_rejects_zero_vectors() {
        let runtime = KhiveRuntime::memory().unwrap();
        runtime.register_embedder(WrongCardinalityEmbedderProvider);
        let model_name = "wrong-cardinality-embedding-service";

        let error = runtime
            .embed_document_with_model_outcome(model_name, "single document")
            .await
            .expect_err("provider returning zero vectors must fail closed");

        assert!(
            error
                .to_string()
                .contains("embed_passage returned 0 vectors for 1 input"),
            "unexpected error: {error}"
        );
    }

    #[tokio::test]
    async fn singleton_document_embed_rejects_surplus_vectors() {
        let runtime = KhiveRuntime::memory().unwrap();
        runtime.register_embedder(SurplusCardinalityEmbedderProvider);
        let model_name = "surplus-cardinality-embedding-service";

        let error = runtime
            .embed_document_with_model_outcome(model_name, "single document")
            .await
            .expect_err("provider returning surplus vectors must fail closed");

        assert!(
            error
                .to_string()
                .contains("embed_passage returned 2 vectors for 1 input"),
            "unexpected error: {error}"
        );
    }

    #[test]
    #[ignore = "loads ~80 MB model; run with --include-ignored"]
    fn minilm_document_and_query_embed_are_identical_no_prefix_model() {
        // MiniLM has no instruction prefixes; document and query paths must
        // produce byte-identical vectors so that existing stored vectors remain
        // comparable after this change.
        let model = EmbeddingModel::AllMiniLmL6V2;
        let config = RuntimeConfig {
            db_path: None,
            default_namespace: Namespace::parse("test").unwrap(),
            embedding_model: Some(model),
            packs: vec!["kg".to_string()],
            ..RuntimeConfig::default()
        };
        let rt = KhiveRuntime::new(config).unwrap();
        let text = "attention is all you need".to_string();
        let rt_ref = &rt;
        let (doc_emb, query_emb) = tokio::runtime::Runtime::new().unwrap().block_on(async {
            let d = rt_ref
                .embed_document_with_model_outcome(&model.to_string(), &text)
                .await
                .unwrap()
                .vector;
            let q = rt_ref
                .embed_query_with_model(&model.to_string(), &text)
                .await
                .unwrap();
            (d, q)
        });
        assert_eq!(
            doc_emb, query_emb,
            "MiniLM has no instruction prefix: document and query embeds must be identical"
        );
    }

    #[test]
    #[ignore = "loads multilingual-e5-small (~90 MB); run with --include-ignored"]
    fn e5_document_and_query_embed_differ_instruction_tuned_model() {
        // multilingual-e5 prepends "passage: " for documents and "query: " for
        // queries. The same raw text must produce different embeddings when the
        // correct prefixes are applied, confirming the asymmetric-retrieval
        // capability is now exercised.
        let model = EmbeddingModel::MultilingualE5Small;
        let config = RuntimeConfig {
            db_path: None,
            default_namespace: Namespace::parse("test").unwrap(),
            embedding_model: Some(model),
            packs: vec!["kg".to_string()],
            ..RuntimeConfig::default()
        };
        let rt = KhiveRuntime::new(config).unwrap();
        let text = "attention is all you need".to_string();
        let rt_ref = &rt;
        let (doc_emb, query_emb) = tokio::runtime::Runtime::new().unwrap().block_on(async {
            let d = rt_ref
                .embed_document_with_model_outcome(&model.to_string(), &text)
                .await
                .unwrap()
                .vector;
            let q = rt_ref
                .embed_query_with_model(&model.to_string(), &text)
                .await
                .unwrap();
            (d, q)
        });
        assert_ne!(
            doc_emb, query_emb,
            "multilingual-e5-small uses asymmetric prefixes: document ('passage: ') \
             and query ('query: ') embeds of the same text must differ"
        );
    }

    // ---- backfill reader error must be propagated, not swallowed ----

    use crate::embedder_registry::EmbedderProvider;
    use lattice_embed::EmbeddingService;

    /// A stub embedder that never actually loads weights — used to satisfy the
    /// `registered_embedding_model_names` check inside `backfill_missing_embeddings`
    /// without triggering a real model load. The test fault-injects a reader error
    /// before any embedding call is made, so `embed()` is never reached.
    struct StubEmbedderProvider;

    #[async_trait::async_trait]
    impl EmbedderProvider for StubEmbedderProvider {
        fn name(&self) -> &str {
            "stub-model-m07"
        }

        fn dimensions(&self) -> usize {
            4
        }

        async fn build(&self) -> crate::error::RuntimeResult<std::sync::Arc<dyn EmbeddingService>> {
            struct StubSvc;
            #[async_trait::async_trait]
            impl EmbeddingService for StubSvc {
                async fn embed(
                    &self,
                    _texts: &[String],
                    _model: lattice_embed::EmbeddingModel,
                ) -> std::result::Result<Vec<Vec<f32>>, lattice_embed::EmbedError> {
                    Ok(vec![])
                }

                fn supports_model(&self, _model: lattice_embed::EmbeddingModel) -> bool {
                    true
                }

                fn name(&self) -> &'static str {
                    "stub-svc-m07"
                }
            }
            Ok(std::sync::Arc::new(StubSvc))
        }
    }

    /// `backfill_missing_embeddings` must propagate a reader error rather than
    /// treating it as "zero rows to embed" (a silent `Err(_) => vec![]` would
    /// return `Ok(0)` and skip all embeddings without any signal).
    ///
    /// The fault injection substitutes a `StorageError::Pool` for the result of
    /// `sql.reader().await`, exercising the `map_err(RuntimeError::Storage)?` path;
    /// falling back to `unwrap_or_default()` there would swallow the injected error.
    #[tokio::test]
    async fn backfill_reader_error_is_propagated_not_swallowed() {
        let rt = KhiveRuntime::memory().unwrap();
        rt.register_embedder(StubEmbedderProvider);
        let tok = NamespaceToken::local();

        // Arm the fault injection: the next backfill call will substitute a
        // StorageError at the sql.reader().await boundary, then reset.
        super::arm_backfill_reader_fail();

        let result = rt.backfill_missing_embeddings(&tok).await;
        assert!(
            result.is_err(),
            "backfill_missing_embeddings must propagate the reader error (got Ok instead)"
        );
        let err_msg = result.unwrap_err().to_string();
        assert!(
            err_msg.contains("injected failure"),
            "error must originate from the injected reader failure, got: {err_msg}"
        );
    }

    #[tokio::test]
    async fn backfill_skips_a_full_page_of_excluded_messages_and_reaches_eligible_tail() {
        use crate::{NoteEmbeddingPolicy, NoteEmbeddingPolicySpec};
        use khive_storage::note::Note;

        const PAGE: u128 = EMBEDDING_BATCH_PAGE_SIZE as u128;
        let primary = EmbeddingModel::AllMiniLmL6V2;
        let primary_name = primary.to_string();
        let secondary_name = "zz-backfill-secondary";
        let rt = KhiveRuntime::new(RuntimeConfig {
            db_path: None,
            embedding_model: Some(primary),
            packs: vec![],
            ..RuntimeConfig::no_embeddings()
        })
        .unwrap();
        rt.register_embedder(ConstantEmbedderProvider {
            name: primary_name.clone(),
            dimensions: primary.dimensions(),
        });
        rt.register_embedder(ConstantEmbedderProvider {
            name: secondary_name.into(),
            dimensions: 4,
        });
        rt.install_note_embedding_policies(&[NoteEmbeddingPolicySpec {
            kind: "message",
            policy: NoteEmbeddingPolicy::DefaultModel,
        }]);
        let tok = NamespaceToken::local();
        let primary_store = rt.vectors_for_model(&tok, &primary_name).unwrap();
        let secondary_store = rt.vectors_for_model(&tok, secondary_name).unwrap();
        let notes = rt.notes(&tok).unwrap();
        let mut seeded = Vec::new();
        for ordinal in 1..=PAGE + 1 {
            let mut message = Note::new("local", "message", "excluded from secondary");
            message.id = Uuid::from_u128(ordinal);
            seeded.push(message);
        }
        let mut ordinary = Note::new("local", "observation", "eligible after full page");
        ordinary.id = Uuid::from_u128(u128::MAX);
        seeded.push(ordinary.clone());
        notes.upsert_notes(seeded).await.unwrap();

        let backfilled = tokio::time::timeout(
            std::time::Duration::from_secs(60),
            rt.backfill_missing_embeddings(&tok),
        )
        .await
        .expect("backfill must advance past a full excluded page")
        .unwrap();
        assert_eq!(backfilled, (PAGE + 2) as u64 + 1);
        assert_eq!(primary_store.count().await.unwrap(), (PAGE + 2) as u64);
        assert_eq!(secondary_store.count().await.unwrap(), 1);
        let excluded_document = rt
            .text_for_notes(&tok)
            .unwrap()
            .get_document("local", Uuid::from_u128(1))
            .await
            .unwrap()
            .expect("first-model pass must index an excluded message");
        assert_eq!(excluded_document.body, "excluded from secondary");
        assert!(
            rt.text_for_notes(&tok)
                .unwrap()
                .get_document("local", ordinary.id)
                .await
                .unwrap()
                .is_some(),
            "the first-model pass must still repopulate FTS"
        );
    }

    const BACKFILL_BATCH_MODEL: &str = "backfill-batch-model";

    struct CountingBackfillService {
        calls: Arc<AtomicUsize>,
        reject_poison: bool,
    }

    #[async_trait::async_trait]
    impl EmbeddingService for CountingBackfillService {
        async fn embed(
            &self,
            texts: &[String],
            _model: EmbeddingModel,
        ) -> std::result::Result<Vec<Vec<f32>>, EmbedError> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            if self.reject_poison
                && (texts.len() > 1 || texts.iter().any(|text| text.contains("poison")))
            {
                return Err(EmbedError::InferenceFailed("poison input".into()));
            }
            Ok(texts
                .iter()
                .map(|text| {
                    vec![
                        text.len() as f32,
                        text.bytes().map(u32::from).sum::<u32>() as f32,
                        text.as_bytes().first().copied().unwrap_or_default() as f32,
                        text.as_bytes().last().copied().unwrap_or_default() as f32,
                    ]
                })
                .collect())
        }

        fn supports_model(&self, _model: EmbeddingModel) -> bool {
            true
        }

        fn name(&self) -> &'static str {
            "backfill-batch-counting-service"
        }
    }

    struct CountingBackfillProvider {
        calls: Arc<AtomicUsize>,
        reject_poison: bool,
    }

    #[async_trait::async_trait]
    impl EmbedderProvider for CountingBackfillProvider {
        fn name(&self) -> &str {
            BACKFILL_BATCH_MODEL
        }

        fn dimensions(&self) -> usize {
            4
        }

        async fn build(&self) -> crate::error::RuntimeResult<Arc<dyn EmbeddingService>> {
            Ok(Arc::new(CountingBackfillService {
                calls: Arc::clone(&self.calls),
                reject_poison: self.reject_poison,
            }))
        }
    }

    #[tokio::test]
    async fn backfill_batches_provider_calls_and_matches_single_record_vectors() {
        let token = NamespaceToken::local();
        let runtime = KhiveRuntime::memory().unwrap();
        let mut entity_ids = Vec::new();
        for index in 0..257 {
            let entity = runtime
                .create_entity(
                    &token,
                    "concept",
                    None,
                    &format!("Backfill entity {index}"),
                    (index != 0).then_some("body"),
                    None,
                    vec![],
                )
                .await
                .unwrap();
            entity_ids.push(entity.id);
        }
        let mut note_ids = Vec::new();
        for index in 0..3 {
            let note = runtime
                .create_note(
                    &token,
                    "observation",
                    None,
                    &format!("Backfill note {index}"),
                    None,
                    None,
                    vec![],
                )
                .await
                .unwrap();
            note_ids.push(note.id);
        }
        let calls = Arc::new(AtomicUsize::new(0));
        runtime.register_embedder(CountingBackfillProvider {
            calls: Arc::clone(&calls),
            reject_poison: false,
        });
        let usage = crate::usage::UsageContext::new();
        let backfilled =
            crate::usage::scope(usage.clone(), runtime.backfill_missing_embeddings(&token))
                .await
                .unwrap();
        assert_eq!(backfilled, 260);
        assert_eq!(
            calls.load(Ordering::SeqCst),
            3,
            "two entity pages and one note page"
        );
        assert_eq!(
            usage.snapshot()["embed_calls"],
            260,
            "ADR-103 counts texts, not provider calls"
        );

        let singles = KhiveRuntime::memory().unwrap();
        let single_calls = Arc::new(AtomicUsize::new(0));
        singles.register_embedder(CountingBackfillProvider {
            calls: Arc::clone(&single_calls),
            reject_poison: false,
        });
        for id in &entity_ids {
            let entity = runtime.get_entity(&token, *id).await.unwrap();
            singles
                .entities(&token)
                .unwrap()
                .upsert_entity(entity.clone())
                .await
                .unwrap();
            singles.reindex_entity(&token, &entity).await.unwrap();
        }
        for id in &note_ids {
            let note = runtime
                .notes(&token)
                .unwrap()
                .get_note(*id)
                .await
                .unwrap()
                .unwrap();
            singles
                .notes(&token)
                .unwrap()
                .upsert_note(note.clone())
                .await
                .unwrap();
            singles.reindex_note(&token, &note).await.unwrap();
        }
        assert_eq!(single_calls.load(Ordering::SeqCst), 260);
        let batched_entities = runtime
            .vectors_for_model(&token, BACKFILL_BATCH_MODEL)
            .unwrap()
            .get_vectors(&entity_ids, "local", "entity.body")
            .await
            .unwrap();
        let single_entities = singles
            .vectors_for_model(&token, BACKFILL_BATCH_MODEL)
            .unwrap()
            .get_vectors(&entity_ids, "local", "entity.body")
            .await
            .unwrap();
        assert_eq!(batched_entities.len(), 257);
        assert_eq!(batched_entities, single_entities);
        let batched_notes = runtime
            .vectors_for_model(&token, BACKFILL_BATCH_MODEL)
            .unwrap()
            .get_vectors(&note_ids, "local", "note.content")
            .await
            .unwrap();
        let single_notes = singles
            .vectors_for_model(&token, BACKFILL_BATCH_MODEL)
            .unwrap()
            .get_vectors(&note_ids, "local", "note.content")
            .await
            .unwrap();
        assert_eq!(batched_notes.len(), 3);
        assert_eq!(batched_notes, single_notes);
    }

    #[tokio::test]
    async fn backfill_failed_pages_retry_singly_without_duplicate_index_rows() {
        use khive_storage::types::{SqlStatement, SqlValue};

        let token = NamespaceToken::local();
        let runtime = KhiveRuntime::memory().unwrap();
        for name in ["good one", "poison", "good two"] {
            runtime
                .create_entity(&token, "concept", None, name, Some("body"), None, vec![])
                .await
                .unwrap();
            runtime
                .create_note(&token, "observation", None, name, None, None, vec![])
                .await
                .unwrap();
        }
        let calls = Arc::new(AtomicUsize::new(0));
        runtime.register_embedder(CountingBackfillProvider {
            calls: Arc::clone(&calls),
            reject_poison: true,
        });
        let backfilled = runtime.backfill_missing_embeddings(&token).await.unwrap();
        assert_eq!(backfilled, 4);
        assert_eq!(
            calls.load(Ordering::SeqCst),
            8,
            "two failed pages plus six singleton retries"
        );
        let mut reader = runtime.sql().reader().await.unwrap();
        for (table, expected) in [("ann_write_log", 4), ("vector_provenance", 0)] {
            let count = reader
                .query_scalar(SqlStatement {
                    sql: format!("SELECT COUNT(*) FROM {table} WHERE namespace = ?1"),
                    params: vec![SqlValue::Text("local".into())],
                    label: Some("backfill-batch-no-duplicate-rows".into()),
                })
                .await
                .unwrap();
            assert!(
                matches!(count, Some(SqlValue::Integer(value)) if value == expected),
                "{table} must match the guarded per-record writer"
            );
        }
        assert_eq!(
            runtime.backfill_missing_embeddings(&token).await.unwrap(),
            0
        );
    }
}