Skip to main content

semantic_memory/
search.rs

1//! Hybrid search engine: BM25 + vector similarity + Reciprocal Rank Fusion.
2
3use crate::config::{DerivedVectorBackendPolicy, SearchConfig};
4use crate::episodes;
5use crate::error::MemoryError;
6use crate::types::{
7    ExplainedResult, ScoreBreakdown, SearchContext, SearchResult, SearchSource, SearchSourceType,
8    VectorSearchReceiptV1,
9};
10use rusqlite::types::Value as SqlValue;
11use rusqlite::Connection;
12// `OptionalExtension` provides `Result::optional()` for `rusqlite::query_row`.
13// Four unconditional call sites in this file use it; keep the import
14// always available. The trait is light (zero runtime cost) so the
15// `#[allow(unused_imports)]` is the only cost when no callsite is in
16// scope on a given feature set.
17#[allow(unused_imports)]
18use rusqlite::OptionalExtension;
19use stack_ids::DigestBuilder;
20#[cfg(feature = "turbo-quant-codec")]
21use std::collections::BinaryHeap;
22use std::collections::{HashMap, HashSet};
23use std::sync::atomic::{AtomicUsize, Ordering};
24
25/// Per-table row count above which vector search emits a warning.
26const VECTOR_SCAN_WARN_THRESHOLD: usize = 50_000;
27/// Per-table row count above which brute-force vector search is refused.
28const VECTOR_SCAN_HARD_LIMIT: usize = 250_000;
29
30static VECTOR_SCAN_WARN_LIMIT: AtomicUsize = AtomicUsize::new(VECTOR_SCAN_WARN_THRESHOLD);
31static VECTOR_SCAN_BLOCK_LIMIT: AtomicUsize = AtomicUsize::new(VECTOR_SCAN_HARD_LIMIT);
32
33
34/// Expand query terms to match hyphenated variants.
35/// "turbo-quant" -> "turbo-quant OR turboquant"
36/// This improves BM25 recall for technical terms with hyphens.
37#[allow(dead_code)]
38fn expand_query_for_fts(query: &str) -> String {
39    let terms: Vec<&str> = query.split_whitespace().collect();
40    let expanded: Vec<String> = terms.iter().map(|term| {
41        if term.contains('-') {
42            let no_hyphen = term.replace('-', "");
43            if no_hyphen != *term {
44                format!("{term} OR {no_hyphen}")
45            } else {
46                term.to_string()
47            }
48        } else {
49            term.to_string()
50        }
51    }).collect();
52    expanded.join(" ")
53}
54
55
56/// Classify whether a query needs retrieval at all.
57/// Simple greetings, confirmations, and single-word responses don't need search.
58pub fn should_retrieve(query: &str) -> bool {
59    let trimmed = query.trim();
60    if trimmed.len() < 12 {
61        return false;
62    }
63    let lower = trimmed.to_lowercase();
64    let skip_phrases = [
65        "ok", "yes", "no", "thanks", "done", "sure", "yeah", "right",
66        "correct", "agreed", "ok thanks", "got it", "sounds good",
67        "that works", "makes sense", "i see", "understood", "gotcha",
68    ];
69    for phrase in &skip_phrases {
70        if lower == *phrase {
71            return false;
72        }
73    }
74    if lower.starts_with("can you") || lower.starts_with("could you")
75        || lower.starts_with("would you") || lower.starts_with("will you")
76    {
77        if lower.len() <= 20 {
78            return false;
79        }
80    }
81    if trimmed.starts_with('/') {
82        return false;
83    }
84    true
85}
86
87/// Sanitize a raw query string for safe use in an FTS5 MATCH expression.
88///
89/// Replaces any character that is not alphanumeric, whitespace, or a Unicode
90/// letter/digit with a space, then strips FTS5 boolean keywords (`AND`, `OR`,
91/// `NOT`, `NEAR`).  Returns `None` when no searchable tokens remain.
92///
93/// This uses an allowlist strategy so that *any* FTS5 operator or punctuation
94/// — including `?`, `.`, `/`, `!`, etc. — is neutralised without needing an
95/// exhaustive denylist.
96pub fn sanitize_fts_query(raw: &str) -> Option<String> {
97    let cleaned: String = raw
98        .chars()
99        .map(|c| {
100            if c.is_alphanumeric() || c.is_whitespace() || c == '_' {
101                c
102            } else {
103                ' '
104            }
105        })
106        .collect();
107
108    let tokens: Vec<&str> = cleaned
109        .split_whitespace()
110        .filter(|t| !matches!(t.to_uppercase().as_str(), "AND" | "OR" | "NOT" | "NEAR"))
111        .collect();
112
113    if tokens.is_empty() {
114        None
115    } else {
116        Some(
117            tokens
118                .into_iter()
119                .map(|token| format!("\"{}\"", token.replace('"', "\"\"")))
120                .collect::<Vec<_>>()
121                .join(" OR "),
122        )
123    }
124}
125
126/// Compute cosine similarity between two vectors.
127pub fn cosine_similarity(a: &[f32], b: &[f32]) -> Result<f32, MemoryError> {
128    if a.len() != b.len() {
129        return Err(MemoryError::EmbeddingDimensionMismatch {
130            expected: a.len(),
131            actual: b.len(),
132        });
133    }
134    if let Some((index, _)) = a.iter().enumerate().find(|(_, value)| !value.is_finite()) {
135        return Err(MemoryError::NonFiniteEmbeddingValue { index });
136    }
137    if let Some((index, _)) = b.iter().enumerate().find(|(_, value)| !value.is_finite()) {
138        return Err(MemoryError::NonFiniteEmbeddingValue { index });
139    }
140    let dot: f32 = a.iter().zip(b.iter()).map(|(x, y)| x * y).sum();
141    let norm_a: f32 = a.iter().map(|x| x * x).sum::<f32>().sqrt();
142    let norm_b: f32 = b.iter().map(|x| x * x).sum::<f32>().sqrt();
143    if norm_a == 0.0 || norm_b == 0.0 {
144        return Ok(0.0);
145    }
146    let similarity = dot / (norm_a * norm_b);
147    if !similarity.is_finite() {
148        return Err(MemoryError::Other(
149            "cosine similarity produced a non-finite score".to_string(),
150        ));
151    }
152    Ok(similarity)
153}
154
155fn days_since(timestamp: &str, evaluation_time: chrono::DateTime<chrono::Utc>) -> Option<f64> {
156    let dt = parse_search_timestamp(timestamp)?;
157    let duration = evaluation_time.naive_utc() - dt;
158    Some(duration.num_seconds() as f64 / 86_400.0)
159}
160
161fn parse_search_timestamp(timestamp: &str) -> Option<chrono::NaiveDateTime> {
162    if let Ok(dt) = chrono::NaiveDateTime::parse_from_str(timestamp, "%Y-%m-%d %H:%M:%S") {
163        return Some(dt);
164    }
165    if let Ok(dt) = chrono::NaiveDateTime::parse_from_str(timestamp, "%Y-%m-%d %H:%M:%S%.f") {
166        return Some(dt);
167    }
168    if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(timestamp) {
169        return Some(dt.naive_utc());
170    }
171    tracing::warn!(
172        timestamp,
173        "failed to parse search timestamp for recency scoring; recency contribution dropped"
174    );
175    None
176}
177
178fn recency_contribution(
179    config: &SearchConfig,
180    context: &SearchContext,
181    updated_at: Option<&str>,
182    best_rank: Option<usize>,
183) -> Option<f64> {
184    match (config.recency_half_life_days, updated_at) {
185        (Some(half_life), Some(ts)) if half_life > 0.0 => {
186            let age_days = days_since(ts, context.evaluation_time).map(|days| days.max(0.0))?;
187            let decay = 2.0_f64.powf(-age_days / half_life);
188            let rank = best_rank.unwrap_or(1).max(1) as f64;
189            Some(config.recency_weight * decay / (config.rrf_k + rank))
190        }
191        _ => None,
192    }
193}
194
195pub(crate) fn search_result_id(source: &SearchSource) -> String {
196    match source {
197        SearchSource::Fact { fact_id, .. } => format!("fact:{fact_id}"),
198        SearchSource::Chunk { chunk_id, .. } => format!("chunk:{chunk_id}"),
199        SearchSource::Message { message_id, .. } => format!("msg:{message_id}"),
200        SearchSource::Episode { episode_id, .. } => format!("episode:{episode_id}"),
201        SearchSource::Projection { projection_id, .. } => format!("projection:{projection_id}"),
202    }
203}
204
205pub fn source_dedup_key(source: &SearchSource) -> (u8, String) {
206    match source {
207        SearchSource::Fact { fact_id, .. } => (0, fact_id.clone()),
208        SearchSource::Chunk { chunk_id, .. } => (1, chunk_id.clone()),
209        SearchSource::Message {
210            message_id,
211            session_id,
212            ..
213        } => (2, format!("{session_id}:{message_id}")),
214        SearchSource::Episode { episode_id, .. } => (3, episode_id.clone()),
215        SearchSource::Projection { projection_id, .. } => (4, projection_id.clone()),
216    }
217}
218
219/// A BM25 search hit from FTS5.
220#[derive(Debug, Clone)]
221pub struct Bm25Hit {
222    /// Search item key such as `fact:{uuid}` or `episode:{episode_id}`.
223    pub id: String,
224    /// Text content returned to callers.
225    pub content: String,
226    /// Source info.
227    pub source: SearchSource,
228    /// Raw BM25 score reported by SQLite FTS5.
229    pub raw_score: f64,
230    /// Timestamp used for recency scoring.
231    pub updated_at: Option<String>,
232    /// Temporal weight for stale-fact downranking (0.0-1.0, default 1.0).
233    pub temporal_weight: Option<f64>,
234    /// Provenance confidence (0.0-1.0, default 0.5).
235    pub provenance_confidence: Option<f64>,
236}
237
238/// A vector search hit.
239#[derive(Debug, Clone)]
240pub struct VectorHit {
241    /// Search item key such as `fact:{uuid}` or `episode:{episode_id}`.
242    pub id: String,
243    /// Text content returned to callers.
244    pub content: String,
245    /// Source info.
246    pub source: SearchSource,
247    /// Final similarity used for vector ranking.
248    pub similarity: f64,
249    /// Timestamp used for recency scoring.
250    pub updated_at: Option<String>,
251    /// Rank from the underlying retrieval stage before exact reranking.
252    pub source_rank: Option<usize>,
253    /// Similarity from the underlying retrieval stage before exact reranking.
254    pub source_similarity: Option<f64>,
255    /// Whether exact f32 reranking changed or confirmed this candidate ordering.
256    pub reranked_from_f32: bool,
257    /// Temporal weight for stale-fact downranking (0.0-1.0, default 1.0).
258    pub temporal_weight: Option<f64>,
259    /// Provenance confidence (0.0-1.0, default 0.5).
260    pub provenance_confidence: Option<f64>,
261}
262
263#[allow(dead_code)]
264struct VectorRow {
265    id: String,
266    content: String,
267    blob: Vec<u8>,
268    updated_at: Option<String>,
269    source_type: SearchSourceType,
270    filter_namespace: Option<String>,
271    filter_session_id: Option<String>,
272    source: SearchSource,
273}
274
275struct RrfCandidate {
276    content: String,
277    source: SearchSource,
278    updated_at: Option<String>,
279    bm25_score: Option<f64>,
280    bm25_rank: Option<usize>,
281    vector_score: Option<f64>,
282    vector_rank: Option<usize>,
283    vector_source_rank: Option<usize>,
284    vector_source_score: Option<f64>,
285    vector_reranked_from_f32: bool,
286    /// Late interaction (ColBERT MaxSim) rank — 3rd RRF signal.
287    late_interaction_rank: Option<usize>,
288    /// Late interaction raw score. Populated only with the `late-interaction` feature.
289    #[allow(dead_code)]
290    late_interaction_score: Option<f64>,
291    /// Temporal weight for stale-fact downranking (0.0-1.0, default 1.0).
292    temporal_weight: Option<f64>,
293    /// Provenance confidence (0.0-1.0, default 0.5). Higher = more trustworthy.
294    provenance_confidence: Option<f64>,
295}
296
297impl RrfCandidate {
298    fn explained(self, config: &SearchConfig, context: &SearchContext) -> ExplainedResult {
299        let bm25_contribution = self
300            .bm25_rank
301            .map(|rank| config.bm25_weight / (config.rrf_k + rank as f64));
302        let vector_contribution = self
303            .vector_rank
304            .map(|rank| config.vector_weight / (config.rrf_k + rank as f64));
305        // Late interaction contribution: uses same RRF formula with late_interaction_weight.
306        // Defaults to 0.0 weight when not configured (backward compatible).
307        let late_interaction_weight = config.late_interaction_weight;
308        let late_interaction_contribution = self
309            .late_interaction_rank
310            .map(|rank| late_interaction_weight / (config.rrf_k + rank as f64));
311        let best_rank = match (self.bm25_rank, self.vector_rank) {
312            (Some(a), Some(b)) => Some(a.min(b)),
313            (Some(a), None) | (None, Some(a)) => Some(a),
314            (None, None) => None,
315        };
316        let recency_score =
317            recency_contribution(config, context, self.updated_at.as_deref(), best_rank);
318        let base_score = bm25_contribution.unwrap_or(0.0)
319            + vector_contribution.unwrap_or(0.0)
320            + late_interaction_contribution.unwrap_or(0.0)
321            + recency_score.unwrap_or(0.0);
322        // Apply temporal weight: stale facts (weight < 1.0) get downranked.
323        // Default weight is 1.0 (no effect) when temporal feature is not active.
324        let temporal_factor = self.temporal_weight.unwrap_or(1.0);
325        let provenance_factor = 1.0 + (self.provenance_confidence.unwrap_or(0.5) - 0.5) * 0.2;
326        let rrf_score = base_score * temporal_factor * provenance_factor;
327        // Apply namespace weight if configured
328        let ns_weight = match &self.source {
329            SearchSource::Fact { namespace, .. } => {
330                config.namespace_weights.get(namespace).copied().unwrap_or(1.0)
331            }
332            _ => 1.0,
333        };
334        let rrf_score = rrf_score * ns_weight;
335
336        let breakdown = ScoreBreakdown {
337            rrf_score,
338            bm25_score: self.bm25_score,
339            vector_score: self.vector_score,
340            recency_score,
341            bm25_rank: self.bm25_rank,
342            vector_rank: self.vector_rank,
343            vector_source_rank: self.vector_source_rank,
344            vector_source_score: self.vector_source_score,
345            bm25_contribution,
346            vector_contribution,
347            vector_reranked_from_f32: self.vector_reranked_from_f32,
348            bm25_weight: config.bm25_weight,
349            vector_weight: config.vector_weight,
350            recency_weight: config.recency_half_life_days.map(|_| config.recency_weight),
351            rrf_k: config.rrf_k,
352        };
353
354        ExplainedResult {
355            result: SearchResult {
356                content: self.content,
357                source: self.source,
358                score: rrf_score,
359                bm25_rank: breakdown.bm25_rank,
360                vector_rank: breakdown.vector_rank,
361                cosine_similarity: breakdown.vector_score,
362            },
363            breakdown,
364        }
365    }
366}
367
368fn scan_vector_rows(
369    rows: impl Iterator<Item = Result<VectorRow, rusqlite::Error>>,
370    query_embedding: &[f32],
371    min_similarity: f64,
372    table_label: &str,
373) -> Result<(Vec<VectorHit>, usize), MemoryError> {
374    let expected_dims = query_embedding.len();
375    let mut hits = Vec::new();
376    let mut row_count = 0usize;
377    let warn_limit = VECTOR_SCAN_WARN_LIMIT.load(Ordering::Relaxed);
378    let hard_limit = VECTOR_SCAN_BLOCK_LIMIT.load(Ordering::Relaxed);
379
380    for row in rows {
381        let row = row?;
382        row_count += 1;
383        if warn_limit > 0 && row_count == warn_limit.saturating_add(1) {
384            tracing::warn!(
385                table = table_label,
386                count = row_count,
387                threshold = warn_limit,
388                "vector scan warning threshold exceeded"
389            );
390        }
391        if hard_limit > 0 && row_count > hard_limit {
392            return Err(MemoryError::VectorScanLimitExceeded {
393                table: table_label.to_string(),
394                scanned: row_count,
395                limit: hard_limit,
396            });
397        }
398
399        let stored_embedding = match crate::db::decode_f32_le(&row.blob, expected_dims) {
400            Ok(embedding) => embedding,
401            Err(error) => {
402                tracing::warn!(
403                    error = %error,
404                    table = table_label,
405                    item = %row.id,
406                    "Skipping row with invalid embedding blob"
407                );
408                continue;
409            }
410        };
411
412        if stored_embedding.len() != expected_dims {
413            tracing::warn!(
414                expected = expected_dims,
415                actual = stored_embedding.len(),
416                "Skipping {} with wrong embedding dimensions",
417                table_label
418            );
419            continue;
420        }
421
422        let similarity = cosine_similarity(query_embedding, &stored_embedding)? as f64;
423        if similarity >= min_similarity {
424            hits.push(VectorHit {
425                id: row.id,
426                content: row.content,
427                source: row.source,
428                similarity,
429                updated_at: row.updated_at,
430                source_rank: None,
431                source_similarity: None,
432                reranked_from_f32: false,
433                temporal_weight: None,
434                provenance_confidence: None,
435            });
436        }
437    }
438
439    Ok((hits, row_count))
440}
441
442fn rank_vector_hits(mut hits: Vec<VectorHit>, pool_size: usize) -> Vec<VectorHit> {
443    hits.sort_by(|a, b| {
444        b.similarity.partial_cmp(&a.similarity).unwrap_or_else(|| {
445            if a.similarity.is_nan() {
446                std::cmp::Ordering::Greater
447            } else {
448                std::cmp::Ordering::Less
449            }
450        })
451    });
452
453    for (idx, hit) in hits.iter_mut().enumerate() {
454        hit.source_rank = Some(idx + 1);
455        hit.source_similarity = Some(hit.similarity);
456    }
457
458    hits.truncate(pool_size);
459    hits
460}
461
462/// Run BM25 search over facts_fts, chunks_fts, episodes_fts, and optionally messages_fts.
463pub(crate) fn bm25_search(
464    conn: &Connection,
465    sanitized_query: &str,
466    pool_size: usize,
467    namespaces: Option<&[&str]>,
468    source_types: Option<&[SearchSourceType]>,
469    session_ids: Option<&[&str]>,
470) -> Result<Vec<Bm25Hit>, MemoryError> {
471    let mut hits = Vec::new();
472
473    let search_facts = source_types
474        .map(|st| st.contains(&SearchSourceType::Facts))
475        .unwrap_or(true);
476    let search_chunks = source_types
477        .map(|st| st.contains(&SearchSourceType::Chunks))
478        .unwrap_or(true);
479    let search_messages = source_types
480        .map(|st| st.contains(&SearchSourceType::Messages))
481        .unwrap_or(false);
482    let search_episodes = source_types
483        .map(|st| st.contains(&SearchSourceType::Episodes))
484        .unwrap_or(true);
485
486    if search_facts {
487        let (ns_clause, ns_params) = build_filter_clause("f.namespace", namespaces, 3);
488        let sql = format!(
489            "SELECT fm.fact_id, f.content, f.namespace, bm25(facts_fts) AS score, f.updated_at, f.temporal_weight
490             FROM facts_fts
491             JOIN facts_rowid_map fm ON facts_fts.rowid = fm.rowid
492             JOIN facts f ON f.id = fm.fact_id
493             WHERE facts_fts MATCH ?1 {}
494             ORDER BY score ASC
495             LIMIT ?2",
496            ns_clause
497        );
498
499        let mut params = vec![
500            SqlValue::Text(sanitized_query.to_string()),
501            SqlValue::Integer(pool_size as i64),
502        ];
503        params.extend(ns_params);
504
505        let mut stmt = conn.prepare(&sql)?;
506        let rows = stmt.query_map(rusqlite::params_from_iter(&params), |row| {
507            let fact_id: String = row.get(0)?;
508            let content: String = row.get(1)?;
509            let namespace: String = row.get(2)?;
510            let raw_score: f64 = row.get(3)?;
511            let updated_at: Option<String> = row.get(4)?;
512            let temporal_weight: Option<f64> = row.get(5)?;
513            Ok(Bm25Hit {
514                id: format!("fact:{fact_id}"),
515                content,
516                source: SearchSource::Fact { fact_id, namespace },
517                raw_score,
518                updated_at,
519                temporal_weight,
520                provenance_confidence: None,
521            })
522        })?;
523
524        for row in rows {
525            hits.push(row?);
526        }
527    }
528
529    if search_chunks {
530        let (ns_clause, ns_params) = build_filter_clause("d.namespace", namespaces, 3);
531        let sql = format!(
532            "SELECT cm.chunk_id, c.content, c.document_id, d.title, c.chunk_index,
533                    bm25(chunks_fts) AS score, c.created_at
534             FROM chunks_fts
535             JOIN chunks_rowid_map cm ON chunks_fts.rowid = cm.rowid
536             JOIN chunks c ON c.id = cm.chunk_id
537             JOIN documents d ON d.id = c.document_id
538             WHERE chunks_fts MATCH ?1 {}
539             ORDER BY score ASC
540             LIMIT ?2",
541            ns_clause
542        );
543
544        let mut params = vec![
545            SqlValue::Text(sanitized_query.to_string()),
546            SqlValue::Integer(pool_size as i64),
547        ];
548        params.extend(ns_params);
549
550        let mut stmt = conn.prepare(&sql)?;
551        let rows = stmt.query_map(rusqlite::params_from_iter(&params), |row| {
552            let chunk_id: String = row.get(0)?;
553            let content: String = row.get(1)?;
554            let document_id: String = row.get(2)?;
555            let document_title: String = row.get(3)?;
556            let chunk_index: i64 = row.get(4)?;
557            let raw_score: f64 = row.get(5)?;
558            let updated_at: Option<String> = row.get(6)?;
559            Ok(Bm25Hit {
560                id: format!("chunk:{chunk_id}"),
561                content,
562                source: SearchSource::Chunk {
563                    chunk_id,
564                    document_id,
565                    document_title,
566                    chunk_index: chunk_index as usize,
567                },
568                raw_score,
569                updated_at,
570                temporal_weight: None,
571            provenance_confidence: None,
572            })
573        })?;
574
575        for row in rows {
576            hits.push(row?);
577        }
578    }
579
580    if search_messages {
581        let (sid_clause, sid_params) = build_filter_clause("m.session_id", session_ids, 3);
582        let sql = format!(
583            "SELECT mm.message_id, m.content, m.session_id, m.role,
584                    bm25(messages_fts) AS score, m.created_at
585             FROM messages_fts
586             JOIN messages_rowid_map mm ON messages_fts.rowid = mm.rowid
587             JOIN messages m ON m.id = mm.message_id
588             WHERE messages_fts MATCH ?1 {}
589             ORDER BY score ASC
590             LIMIT ?2",
591            sid_clause
592        );
593
594        let mut params = vec![
595            SqlValue::Text(sanitized_query.to_string()),
596            SqlValue::Integer(pool_size as i64),
597        ];
598        params.extend(sid_params);
599
600        let mut stmt = conn.prepare(&sql)?;
601        let rows = stmt.query_map(rusqlite::params_from_iter(&params), |row| {
602            let message_id: i64 = row.get(0)?;
603            let content: String = row.get(1)?;
604            let session_id: String = row.get(2)?;
605            let role: String = row.get(3)?;
606            let raw_score: f64 = row.get(4)?;
607            let updated_at: Option<String> = row.get(5)?;
608            Ok(Bm25Hit {
609                id: format!("msg:{message_id}"),
610                content,
611                source: SearchSource::Message {
612                    message_id,
613                    session_id,
614                    role,
615                },
616                raw_score,
617                updated_at,
618                temporal_weight: None,
619            provenance_confidence: None,
620            })
621        })?;
622
623        for row in rows {
624            hits.push(row?);
625        }
626    }
627
628    if search_episodes {
629        let (ns_clause, ns_params) = build_filter_clause("d.namespace", namespaces, 3);
630        let sql = format!(
631            "SELECT e.episode_id, e.document_id, e.search_text, e.effect_type, e.outcome,
632                    bm25(episodes_fts) AS score, e.updated_at
633             FROM episodes_fts
634             JOIN episodes_rowid_map rm ON episodes_fts.rowid = rm.rowid
635             JOIN episodes e ON e.episode_id = rm.episode_id
636             JOIN documents d ON d.id = e.document_id
637             WHERE episodes_fts MATCH ?1 {}
638             ORDER BY score ASC
639             LIMIT ?2",
640            ns_clause
641        );
642
643        let mut params = vec![
644            SqlValue::Text(sanitized_query.to_string()),
645            SqlValue::Integer(pool_size as i64),
646        ];
647        params.extend(ns_params);
648
649        let mut stmt = conn.prepare(&sql)?;
650        let rows = stmt.query_map(rusqlite::params_from_iter(&params), |row| {
651            let episode_id: String = row.get(0)?;
652            let document_id: String = row.get(1)?;
653            let content: String = row.get(2)?;
654            let effect_type: String = row.get(3)?;
655            let outcome: String = row.get(4)?;
656            let raw_score: f64 = row.get(5)?;
657            let updated_at: Option<String> = row.get(6)?;
658            Ok(Bm25Hit {
659                id: episodes::episode_item_key(&episode_id),
660                content,
661                source: SearchSource::Episode {
662                    episode_id,
663                    document_id,
664                    effect_type,
665                    outcome,
666                },
667                raw_score,
668                updated_at,
669                temporal_weight: None,
670            provenance_confidence: None,
671            })
672        })?;
673
674        for row in rows {
675            hits.push(row?);
676        }
677    }
678
679    Ok(hits)
680}
681
682/// Run brute-force vector search over facts, chunks, episodes, and optionally messages.
683pub(crate) fn vector_search(
684    conn: &Connection,
685    query_embedding: &[f32],
686    pool_size: usize,
687    min_similarity: f64,
688    namespaces: Option<&[&str]>,
689    source_types: Option<&[SearchSourceType]>,
690    session_ids: Option<&[&str]>,
691) -> Result<Vec<VectorHit>, MemoryError> {
692    let mut hits = Vec::new();
693
694    let search_facts = source_types
695        .map(|st| st.contains(&SearchSourceType::Facts))
696        .unwrap_or(true);
697    let search_chunks = source_types
698        .map(|st| st.contains(&SearchSourceType::Chunks))
699        .unwrap_or(true);
700    let search_messages = source_types
701        .map(|st| st.contains(&SearchSourceType::Messages))
702        .unwrap_or(false);
703    let search_episodes = source_types
704        .map(|st| st.contains(&SearchSourceType::Episodes))
705        .unwrap_or(true);
706
707    if search_facts {
708        let (ns_clause, ns_params) = build_filter_clause("namespace", namespaces, 1);
709        let sql = format!(
710            "SELECT id, content, namespace, embedding, updated_at
711             FROM facts
712             WHERE embedding IS NOT NULL {}",
713            ns_clause
714        );
715
716        let mut stmt = conn.prepare(&sql)?;
717        let rows = stmt.query_map(rusqlite::params_from_iter(&ns_params), |row| {
718            let id: String = row.get(0)?;
719            let content: String = row.get(1)?;
720            let namespace: String = row.get(2)?;
721            let blob: Vec<u8> = row.get(3)?;
722            let updated_at: Option<String> = row.get(4)?;
723            Ok(VectorRow {
724                id: format!("fact:{id}"),
725                content,
726                blob,
727                updated_at,
728                source_type: SearchSourceType::Facts,
729                filter_namespace: Some(namespace.clone()),
730                filter_session_id: None,
731                source: SearchSource::Fact {
732                    fact_id: id,
733                    namespace,
734                },
735            })
736        })?;
737
738        let (fact_hits, fact_count) =
739            scan_vector_rows(rows, query_embedding, min_similarity, "fact")?;
740        hits.extend(fact_hits);
741
742        if vector_scan_warn_exceeded(fact_count) {
743            tracing::warn!(
744                count = fact_count,
745                "facts table exceeds vector scan threshold ({} rows)",
746                fact_count
747            );
748        }
749    }
750
751    if search_chunks {
752        let (ns_clause, ns_params) = build_filter_clause("d.namespace", namespaces, 1);
753        let sql = format!(
754            "SELECT c.id, c.content, c.document_id, d.title, c.chunk_index, c.embedding, c.created_at, d.namespace
755             FROM chunks c
756             JOIN documents d ON d.id = c.document_id
757             WHERE c.embedding IS NOT NULL {}",
758            ns_clause
759        );
760
761        let mut stmt = conn.prepare(&sql)?;
762        let rows = stmt.query_map(rusqlite::params_from_iter(&ns_params), |row| {
763            let id: String = row.get(0)?;
764            let content: String = row.get(1)?;
765            let document_id: String = row.get(2)?;
766            let document_title: String = row.get(3)?;
767            let chunk_index: i64 = row.get(4)?;
768            let blob: Vec<u8> = row.get(5)?;
769            let updated_at: Option<String> = row.get(6)?;
770            let namespace: String = row.get(7)?;
771            Ok(VectorRow {
772                id: format!("chunk:{id}"),
773                content,
774                blob,
775                updated_at,
776                source_type: SearchSourceType::Chunks,
777                filter_namespace: Some(namespace),
778                filter_session_id: None,
779                source: SearchSource::Chunk {
780                    chunk_id: id,
781                    document_id,
782                    document_title,
783                    chunk_index: chunk_index as usize,
784                },
785            })
786        })?;
787
788        let (chunk_hits, chunk_count) =
789            scan_vector_rows(rows, query_embedding, min_similarity, "chunk")?;
790        hits.extend(chunk_hits);
791
792        if vector_scan_warn_exceeded(chunk_count) {
793            tracing::warn!(
794                count = chunk_count,
795                "chunks table exceeds vector scan threshold ({} rows)",
796                chunk_count
797            );
798        }
799    }
800
801    if search_messages {
802        let (sid_clause, sid_params) = build_filter_clause("m.session_id", session_ids, 1);
803        let sql = format!(
804            "SELECT m.id, m.content, m.session_id, m.role, m.embedding, m.created_at
805             FROM messages m
806             WHERE m.embedding IS NOT NULL {}",
807            sid_clause
808        );
809
810        let mut stmt = conn.prepare(&sql)?;
811        let rows = stmt.query_map(rusqlite::params_from_iter(&sid_params), |row| {
812            let message_id: i64 = row.get(0)?;
813            let content: String = row.get(1)?;
814            let session_id: String = row.get(2)?;
815            let role: String = row.get(3)?;
816            let blob: Vec<u8> = row.get(4)?;
817            let updated_at: Option<String> = row.get(5)?;
818            Ok(VectorRow {
819                id: format!("msg:{message_id}"),
820                content,
821                blob,
822                updated_at,
823                source_type: SearchSourceType::Messages,
824                filter_namespace: None,
825                filter_session_id: Some(session_id.clone()),
826                source: SearchSource::Message {
827                    message_id,
828                    session_id,
829                    role,
830                },
831            })
832        })?;
833
834        let (message_hits, message_count) =
835            scan_vector_rows(rows, query_embedding, min_similarity, "message")?;
836        hits.extend(message_hits);
837
838        if vector_scan_warn_exceeded(message_count) {
839            tracing::warn!(
840                count = message_count,
841                "messages table exceeds vector scan threshold ({} rows)",
842                message_count
843            );
844        }
845    }
846
847    if search_episodes {
848        let (ns_clause, ns_params) = build_filter_clause("d.namespace", namespaces, 1);
849        let sql = format!(
850            "SELECT e.episode_id, e.document_id, e.search_text, e.effect_type, e.outcome, e.embedding, e.updated_at, d.namespace
851             FROM episodes e
852             JOIN documents d ON d.id = e.document_id
853             WHERE e.embedding IS NOT NULL {}",
854            ns_clause
855        );
856
857        let mut stmt = conn.prepare(&sql)?;
858        let rows = stmt.query_map(rusqlite::params_from_iter(&ns_params), |row| {
859            let episode_id: String = row.get(0)?;
860            let document_id: String = row.get(1)?;
861            let content: String = row.get(2)?;
862            let effect_type: String = row.get(3)?;
863            let outcome: String = row.get(4)?;
864            let blob: Vec<u8> = row.get(5)?;
865            let updated_at: Option<String> = row.get(6)?;
866            let namespace: String = row.get(7)?;
867            Ok(VectorRow {
868                id: episodes::episode_item_key(&episode_id),
869                content,
870                blob,
871                updated_at,
872                source_type: SearchSourceType::Episodes,
873                filter_namespace: Some(namespace),
874                filter_session_id: None,
875                source: SearchSource::Episode {
876                    episode_id,
877                    document_id,
878                    effect_type,
879                    outcome,
880                },
881            })
882        })?;
883
884        let (episode_hits, episode_count) =
885            scan_vector_rows(rows, query_embedding, min_similarity, "episode")?;
886        hits.extend(episode_hits);
887
888        if vector_scan_warn_exceeded(episode_count) {
889            tracing::warn!(
890                count = episode_count,
891                "episodes table exceeds vector scan threshold ({} rows)",
892                episode_count
893            );
894        }
895    }
896
897    Ok(rank_vector_hits(hits, pool_size))
898}
899
900fn brute_force_vector_outcome(
901    conn: &Connection,
902    query_embedding: &[f32],
903    pool_size: usize,
904    min_similarity: f64,
905    namespaces: Option<&[&str]>,
906    source_types: Option<&[SearchSourceType]>,
907    session_ids: Option<&[&str]>,
908) -> Result<VectorSearchOutcome, MemoryError> {
909    let hits = vector_search(
910        conn,
911        query_embedding,
912        pool_size,
913        min_similarity,
914        namespaces,
915        source_types,
916        session_ids,
917    )?;
918    Ok(VectorSearchOutcome {
919        requested_candidates: pool_size,
920        returned_candidates: hits.len(),
921        post_filter_candidates: hits.len(),
922        hits,
923        candidate_backend: "brute_force_f32".to_string(),
924        fallback: None,
925        exact_rerank: true,
926        degradations: Vec::new(),
927        receipt_metadata: VectorReceiptMetadata::default(),
928    })
929}
930
931#[allow(clippy::too_many_arguments)]
932fn vector_search_with_backend(
933    conn: &Connection,
934    query_embedding: &[f32],
935    pool_size: usize,
936    min_similarity: f64,
937    config: &SearchConfig,
938    context: &SearchContext,
939    namespaces: Option<&[&str]>,
940    source_types: Option<&[SearchSourceType]>,
941    session_ids: Option<&[&str]>,
942) -> Result<VectorSearchOutcome, MemoryError> {
943    if context.exactness_profile == crate::types::ExactnessProfile::PreferExact {
944        return brute_force_vector_outcome(
945            conn,
946            query_embedding,
947            pool_size,
948            min_similarity,
949            namespaces,
950            source_types,
951            session_ids,
952        );
953    }
954
955    match config.derived_vector_backend {
956        DerivedVectorBackendPolicy::Disabled => brute_force_vector_outcome(
957            conn,
958            query_embedding,
959            pool_size,
960            min_similarity,
961            namespaces,
962            source_types,
963            session_ids,
964        ),
965        DerivedVectorBackendPolicy::TurboQuantCandidateOnly => turbo_quant_vector_outcome(
966            conn,
967            query_embedding,
968            pool_size,
969            min_similarity,
970            config,
971            namespaces,
972            source_types,
973            session_ids,
974        ),
975        DerivedVectorBackendPolicy::ProveKvPoolCandidateOnly => provekv_pool_vector_outcome(
976            conn,
977            query_embedding,
978            pool_size,
979            min_similarity,
980            config,
981            namespaces,
982            source_types,
983            session_ids,
984        ),
985    }
986}
987
988#[allow(clippy::too_many_arguments)]
989fn provekv_pool_vector_outcome(
990    conn: &Connection,
991    query_embedding: &[f32],
992    pool_size: usize,
993    min_similarity: f64,
994    config: &SearchConfig,
995    namespaces: Option<&[&str]>,
996    source_types: Option<&[SearchSourceType]>,
997    session_ids: Option<&[&str]>,
998) -> Result<VectorSearchOutcome, MemoryError> {
999    if !config.turbo_quant_require_exact_rerank {
1000        return Err(MemoryError::InvalidConfig {
1001            field: "search.turbo_quant_require_exact_rerank",
1002            reason: "proveKV pool candidate backend requires exact f32 rerank".to_string(),
1003        });
1004    }
1005
1006    let mut outcome = brute_force_vector_outcome(
1007        conn,
1008        query_embedding,
1009        pool_size,
1010        min_similarity,
1011        namespaces,
1012        source_types,
1013        session_ids,
1014    )?;
1015    outcome.candidate_backend = "provekv_pool_candidate_then_exact_f32".to_string();
1016    outcome.receipt_metadata.codec_family = Some("provekv_pool".to_string());
1017    match crate::db::latest_ready_provekv_pool_generation(conn)? {
1018        Some(row) => {
1019            let item_map =
1020                crate::db::load_provekv_pool_item_map(conn, &row.generation.generation_id)?;
1021            let _payload =
1022                crate::db::load_provekv_pool_payload(conn, &row.generation.generation_id)?;
1023            outcome.receipt_metadata.artifact_generation_id = Some(row.generation.generation_id);
1024            outcome.receipt_metadata.vector_artifact_manifest_digest =
1025                Some(row.generation.pool_manifest_digest);
1026            outcome.receipt_metadata.vector_artifact_count = Some(item_map.len());
1027            outcome.degradations.push(
1028                "proveKV pool generation materialized for candidate provenance; authoritative f32 exact rerank remains final"
1029                    .to_string(),
1030            );
1031        }
1032        None => {
1033            outcome.fallback = Some("provekv_pool_generation_not_materialized".to_string());
1034            outcome.degradations.push(
1035                "proveKV pool backend requested; using authoritative f32 exact path until a pool generation is materialized"
1036                    .to_string(),
1037            );
1038        }
1039    }
1040    Ok(outcome)
1041}
1042
1043#[cfg(not(feature = "turbo-quant-codec"))]
1044#[allow(clippy::too_many_arguments)]
1045fn turbo_quant_vector_outcome(
1046    conn: &Connection,
1047    query_embedding: &[f32],
1048    pool_size: usize,
1049    min_similarity: f64,
1050    _config: &SearchConfig,
1051    namespaces: Option<&[&str]>,
1052    source_types: Option<&[SearchSourceType]>,
1053    session_ids: Option<&[&str]>,
1054) -> Result<VectorSearchOutcome, MemoryError> {
1055    let mut outcome = brute_force_vector_outcome(
1056        conn,
1057        query_embedding,
1058        pool_size,
1059        min_similarity,
1060        namespaces,
1061        source_types,
1062        session_ids,
1063    )?;
1064    outcome.candidate_backend = "turbo_quant_candidate_then_exact_f32".to_string();
1065    outcome.fallback = Some("turbo_quant_feature_disabled".to_string());
1066    outcome
1067        .degradations
1068        .push("TurboQuant backend requested without turbo-quant-codec feature".to_string());
1069    Ok(outcome)
1070}
1071
1072#[cfg(feature = "turbo-quant-codec")]
1073#[allow(clippy::too_many_arguments)]
1074fn turbo_quant_vector_outcome(
1075    conn: &Connection,
1076    query_embedding: &[f32],
1077    pool_size: usize,
1078    min_similarity: f64,
1079    config: &SearchConfig,
1080    namespaces: Option<&[&str]>,
1081    source_types: Option<&[SearchSourceType]>,
1082    session_ids: Option<&[&str]>,
1083) -> Result<VectorSearchOutcome, MemoryError> {
1084    use crate::vector_codec::{TurboQuantCodec, VectorArtifactV1, VectorCodec};
1085
1086    if !config.turbo_quant_require_exact_rerank {
1087        return Err(MemoryError::InvalidConfig {
1088            field: "search.turbo_quant_require_exact_rerank",
1089            reason: "TurboQuant candidate backend requires exact f32 rerank".to_string(),
1090        });
1091    }
1092
1093    let dim = query_embedding.len();
1094    let codec = TurboQuantCodec::new(
1095        dim,
1096        config.turbo_quant_bits,
1097        config.turbo_quant_projections,
1098        config.turbo_quant_seed,
1099    )?;
1100    let profile = codec.profile().clone();
1101    let profile_digest = profile.digest();
1102    let mut metadata = VectorReceiptMetadata {
1103        codec_family: Some("turbo_quant".to_string()),
1104        codec_profile_digest: Some(profile_digest.clone()),
1105        ..VectorReceiptMetadata::default()
1106    };
1107
1108    let filtered = namespaces.is_some_and(|values| !values.is_empty())
1109        || source_types.is_some_and(|values| !values.is_empty())
1110        || session_ids.is_some_and(|values| !values.is_empty());
1111    metadata.filter_strategy = Some(if filtered {
1112        "adaptive_oversampling_after_approximate_scoring".to_string()
1113    } else {
1114        "unfiltered_top_k_heap".to_string()
1115    });
1116
1117    let raw_count = authoritative_vector_row_count(conn)?;
1118    let (current_source_snapshot_digest, current_source_row_count) =
1119        crate::db::current_source_snapshot_digest(conn, dim)?;
1120    let Some(generation) =
1121        crate::db::current_derived_vector_generation(conn, "turbo_quant", &profile_digest)?
1122    else {
1123        metadata.artifact_missing_count = Some(raw_count);
1124        metadata.vector_artifact_missing_count = Some(raw_count);
1125        let mut outcome = brute_force_vector_outcome(
1126            conn,
1127            query_embedding,
1128            pool_size,
1129            min_similarity,
1130            namespaces,
1131            source_types,
1132            session_ids,
1133        )?;
1134        outcome.candidate_backend = "turbo_quant_candidate_then_exact_f32".to_string();
1135        outcome.fallback = Some("turbo_quant_generation_missing_or_invalidated".to_string());
1136        outcome.degradations.push("No active TurboQuant artifact generation is available; authoritative raw f32 search was used".to_string());
1137        outcome.receipt_metadata = metadata;
1138        return Ok(outcome);
1139    };
1140
1141    metadata.artifact_generation_id = Some(generation.generation_id.clone());
1142    metadata.vector_artifact_manifest_digest = Some(generation.artifact_manifest_digest.clone());
1143    metadata.artifact_count = Some(generation.artifact_count);
1144
1145    let artifacts =
1146        crate::db::load_derived_vector_artifacts_by_generation(conn, &generation.generation_id)?;
1147    metadata.vector_artifact_count = Some(artifacts.len());
1148
1149    if generation.dim != dim
1150        || generation.encoding != "turbo_code_wire_v1"
1151        || generation.status != "active"
1152        || generation.source_row_count != raw_count
1153        || generation.source_row_count != current_source_row_count
1154        || generation.source_snapshot_digest != current_source_snapshot_digest
1155        || generation.artifact_count != artifacts.len()
1156    {
1157        let missing = raw_count.saturating_sub(artifacts.len());
1158        metadata.artifact_missing_count = Some(missing);
1159        metadata.vector_artifact_missing_count = Some(missing);
1160        let mut outcome = brute_force_vector_outcome(
1161            conn,
1162            query_embedding,
1163            pool_size,
1164            min_similarity,
1165            namespaces,
1166            source_types,
1167            session_ids,
1168        )?;
1169        outcome.candidate_backend = "turbo_quant_candidate_then_exact_f32".to_string();
1170        outcome.fallback = Some("turbo_quant_generation_incomplete_or_stale".to_string());
1171        outcome.degradations.push(format!(
1172            "TurboQuant generation validation failed: generation={}, status={}, dim={}, source_rows={}, artifacts={}, authoritative_rows={}, snapshot_current={}",
1173            generation.generation_id,
1174            generation.status,
1175            generation.dim,
1176            generation.source_row_count,
1177            artifacts.len(),
1178            raw_count,
1179            generation.source_snapshot_digest == current_source_snapshot_digest
1180        ));
1181        outcome.receipt_metadata = metadata;
1182        return Ok(outcome);
1183    }
1184
1185    let prepared = codec.prepare_query(query_embedding)?;
1186    let candidate_cap = if filtered {
1187        artifacts
1188            .len()
1189            .min(pool_size.saturating_mul(16).max(pool_size))
1190    } else {
1191        pool_size.min(artifacts.len())
1192    };
1193    let mut scored = BinaryHeap::with_capacity(candidate_cap.saturating_add(1));
1194    let mut corrupt_count = 0usize;
1195    let mut scanned_count = 0usize;
1196    for (seq, artifact_row) in artifacts.into_iter().enumerate() {
1197        scanned_count += 1;
1198        if artifact_row.encoding != "turbo_code_wire_v1"
1199            || artifact_row.dim != dim
1200            || artifact_row.status != "active"
1201        {
1202            corrupt_count += 1;
1203            continue;
1204        }
1205        let artifact = VectorArtifactV1::new(profile.clone(), artifact_row.encoded);
1206        if artifact.profile_digest != artifact_row.codec_profile_digest
1207            || artifact.artifact_digest != artifact_row.encoded_digest
1208        {
1209            corrupt_count += 1;
1210            continue;
1211        }
1212        let approx = match codec.score_inner_product_prepared(&artifact, &prepared) {
1213            Ok(score) if score.is_finite() => score as f64,
1214            Ok(_) => {
1215                corrupt_count += 1;
1216                continue;
1217            }
1218            Err(err) => {
1219                tracing::warn!(
1220                    error = %err,
1221                    item = %artifact_row.item_key,
1222                    "corrupt TurboQuant artifact encountered; falling back to raw f32"
1223                );
1224                corrupt_count += 1;
1225                continue;
1226            }
1227        };
1228        if candidate_cap == 0 {
1229            continue;
1230        }
1231        let candidate = ApproxCandidate {
1232            score: approx,
1233            seq,
1234            item_key: artifact_row.item_key,
1235        };
1236        if scored.len() < candidate_cap {
1237            scored.push(candidate);
1238        } else if scored
1239            .peek()
1240            .is_some_and(|worst: &ApproxCandidate| candidate.score > worst.score)
1241        {
1242            scored.pop();
1243            scored.push(candidate);
1244        }
1245    }
1246
1247    metadata.artifact_corruption_count = Some(corrupt_count);
1248    metadata.approximate_scanned_count = Some(scanned_count);
1249    if corrupt_count > 0 {
1250        let mut outcome = brute_force_vector_outcome(
1251            conn,
1252            query_embedding,
1253            pool_size,
1254            min_similarity,
1255            namespaces,
1256            source_types,
1257            session_ids,
1258        )?;
1259        outcome.candidate_backend = "turbo_quant_candidate_then_exact_f32".to_string();
1260        outcome.fallback = Some("turbo_quant_artifact_validation_failed".to_string());
1261        outcome.degradations.push(format!(
1262            "TurboQuant artifact validation failed: {corrupt_count} corrupt artifacts in generation {}",
1263            generation.generation_id
1264        ));
1265        outcome.receipt_metadata = metadata;
1266        return Ok(outcome);
1267    }
1268
1269    let mut scored = scored.into_vec();
1270    scored.sort_by(|a, b| {
1271        b.score
1272            .partial_cmp(&a.score)
1273            .unwrap_or(std::cmp::Ordering::Equal)
1274            .then_with(|| a.seq.cmp(&b.seq))
1275    });
1276    let approximate_returned = scored.len();
1277    metadata.approximate_candidate_count = Some(approximate_returned);
1278    metadata.approximate_returned_count = Some(approximate_returned);
1279    let mut exact_hits = Vec::new();
1280    let mut raw_rows_loaded_count = 0usize;
1281    let mut missing_count = 0usize;
1282    for (approx_rank_0, candidate) in scored.into_iter().enumerate() {
1283        let Some(row) = load_vector_row_by_item_key(conn, &candidate.item_key)? else {
1284            missing_count += 1;
1285            continue;
1286        };
1287        raw_rows_loaded_count += 1;
1288        if !vector_row_matches_filters(&row, namespaces, source_types, session_ids) {
1289            continue;
1290        }
1291        let stored_embedding = crate::db::decode_f32_le(&row.blob, dim)?;
1292        let similarity = cosine_similarity(query_embedding, &stored_embedding)? as f64;
1293        if similarity >= min_similarity {
1294            exact_hits.push(VectorHit {
1295                id: row.id,
1296                content: row.content,
1297                source: row.source,
1298                similarity,
1299                updated_at: row.updated_at,
1300                source_rank: Some(approx_rank_0 + 1),
1301                source_similarity: Some(candidate.score),
1302                reranked_from_f32: true,
1303                temporal_weight: None,
1304                provenance_confidence: None,
1305            });
1306        }
1307    }
1308    let post_filter_candidates = exact_hits.len();
1309    metadata.artifact_missing_count = Some(missing_count);
1310    metadata.vector_artifact_missing_count = Some(missing_count);
1311    metadata.vector_artifact_stale_count = Some(0);
1312    metadata.raw_rows_loaded_count = Some(raw_rows_loaded_count);
1313    metadata.exact_rerank_count = Some(raw_rows_loaded_count);
1314    let mut degradations = Vec::new();
1315    if filtered && post_filter_candidates < pool_size && candidate_cap < scanned_count {
1316        degradations.push(format!(
1317            "TurboQuant filter-aware candidate generation under-returned {post_filter_candidates} candidates for requested pool {pool_size} after scanning {scanned_count} artifacts with candidate budget {candidate_cap}"
1318        ));
1319    }
1320    if missing_count > 0 {
1321        degradations.push(format!(
1322            "TurboQuant exact rerank skipped {missing_count} candidates whose authoritative rows were missing"
1323        ));
1324    }
1325    let hits = rank_vector_hits(exact_hits, pool_size);
1326    Ok(VectorSearchOutcome {
1327        hits,
1328        candidate_backend: "turbo_quant_candidate_then_exact_f32".to_string(),
1329        requested_candidates: pool_size,
1330        returned_candidates: approximate_returned,
1331        post_filter_candidates,
1332        fallback: None,
1333        exact_rerank: true,
1334        degradations,
1335        receipt_metadata: metadata,
1336    })
1337}
1338
1339#[cfg(feature = "turbo-quant-codec")]
1340#[derive(Debug, Clone)]
1341struct ApproxCandidate {
1342    score: f64,
1343    seq: usize,
1344    item_key: String,
1345}
1346
1347#[cfg(feature = "turbo-quant-codec")]
1348impl PartialEq for ApproxCandidate {
1349    fn eq(&self, other: &Self) -> bool {
1350        self.score == other.score && self.seq == other.seq
1351    }
1352}
1353
1354#[cfg(feature = "turbo-quant-codec")]
1355impl Eq for ApproxCandidate {}
1356
1357#[cfg(feature = "turbo-quant-codec")]
1358impl PartialOrd for ApproxCandidate {
1359    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
1360        Some(self.cmp(other))
1361    }
1362}
1363
1364#[cfg(feature = "turbo-quant-codec")]
1365impl Ord for ApproxCandidate {
1366    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
1367        other
1368            .score
1369            .partial_cmp(&self.score)
1370            .unwrap_or(std::cmp::Ordering::Equal)
1371            .then_with(|| other.seq.cmp(&self.seq))
1372    }
1373}
1374
1375#[cfg(feature = "turbo-quant-codec")]
1376fn vector_row_matches_filters(
1377    row: &VectorRow,
1378    namespaces: Option<&[&str]>,
1379    source_types: Option<&[SearchSourceType]>,
1380    session_ids: Option<&[&str]>,
1381) -> bool {
1382    if source_types.is_some_and(|values| !values.contains(&row.source_type)) {
1383        return false;
1384    }
1385    if let Some(namespaces) = namespaces.filter(|values| !values.is_empty()) {
1386        let Some(namespace) = row.filter_namespace.as_deref() else {
1387            return false;
1388        };
1389        if !namespaces.contains(&namespace) {
1390            return false;
1391        }
1392    }
1393    if let Some(session_ids) = session_ids.filter(|values| !values.is_empty()) {
1394        let Some(session_id) = row.filter_session_id.as_deref() else {
1395            return false;
1396        };
1397        if !session_ids.contains(&session_id) {
1398            return false;
1399        }
1400    }
1401    true
1402}
1403
1404#[cfg(feature = "turbo-quant-codec")]
1405fn authoritative_vector_row_count(conn: &Connection) -> Result<usize, MemoryError> {
1406    let count: i64 = conn.query_row(
1407        "SELECT
1408             (SELECT COUNT(*) FROM facts WHERE embedding IS NOT NULL) +
1409             (SELECT COUNT(*) FROM chunks WHERE embedding IS NOT NULL) +
1410             (SELECT COUNT(*) FROM messages WHERE embedding IS NOT NULL) +
1411             (SELECT COUNT(*) FROM episodes WHERE embedding IS NOT NULL)",
1412        [],
1413        |row| row.get(0),
1414    )?;
1415    usize::try_from(count)
1416        .map_err(|err| MemoryError::Other(format!("authoritative vector count overflow: {err}")))
1417}
1418
1419#[cfg(feature = "turbo-quant-codec")]
1420fn load_vector_row_by_item_key(
1421    conn: &Connection,
1422    item_key: &str,
1423) -> Result<Option<VectorRow>, MemoryError> {
1424    let Some((domain, id)) = item_key.split_once(':') else {
1425        return Ok(None);
1426    };
1427    match domain {
1428        "fact" => conn
1429            .query_row(
1430                "SELECT id, content, namespace, embedding, updated_at
1431                 FROM facts WHERE id = ?1 AND embedding IS NOT NULL",
1432                [id],
1433                |row| {
1434                    let fact_id: String = row.get(0)?;
1435                    let content: String = row.get(1)?;
1436                    let namespace: String = row.get(2)?;
1437                    let blob: Vec<u8> = row.get(3)?;
1438                    let updated_at: Option<String> = row.get(4)?;
1439                    Ok(VectorRow {
1440                        id: format!("fact:{fact_id}"),
1441                        content,
1442                        blob,
1443                        updated_at,
1444                        source_type: SearchSourceType::Facts,
1445                        filter_namespace: Some(namespace.clone()),
1446                        filter_session_id: None,
1447                        source: SearchSource::Fact { fact_id, namespace },
1448                    })
1449                },
1450            )
1451            .optional()
1452            .map_err(MemoryError::from),
1453        "chunk" => conn
1454            .query_row(
1455                "SELECT c.id, c.content, c.document_id, d.title, c.chunk_index, c.embedding, c.created_at, d.namespace
1456                 FROM chunks c
1457                 JOIN documents d ON d.id = c.document_id
1458                 WHERE c.id = ?1 AND c.embedding IS NOT NULL",
1459                [id],
1460                |row| {
1461                    let chunk_id: String = row.get(0)?;
1462                    let content: String = row.get(1)?;
1463                    let document_id: String = row.get(2)?;
1464                    let document_title: String = row.get(3)?;
1465                    let chunk_index: i64 = row.get(4)?;
1466                    let blob: Vec<u8> = row.get(5)?;
1467                    let updated_at: Option<String> = row.get(6)?;
1468                    let namespace: String = row.get(7)?;
1469                    Ok(VectorRow {
1470                        id: format!("chunk:{chunk_id}"),
1471                        content,
1472                        blob,
1473                        updated_at,
1474                        source_type: SearchSourceType::Chunks,
1475                        filter_namespace: Some(namespace),
1476                        filter_session_id: None,
1477                        source: SearchSource::Chunk {
1478                            chunk_id,
1479                            document_id,
1480                            document_title,
1481                            chunk_index: chunk_index as usize,
1482                        },
1483                    })
1484                },
1485            )
1486            .optional()
1487            .map_err(MemoryError::from),
1488        "msg" => {
1489            let Ok(message_id) = id.parse::<i64>() else {
1490                return Ok(None);
1491            };
1492            conn.query_row(
1493                "SELECT id, content, session_id, role, embedding, created_at
1494                 FROM messages WHERE id = ?1 AND embedding IS NOT NULL",
1495                [message_id],
1496                |row| {
1497                    let message_id: i64 = row.get(0)?;
1498                    let content: String = row.get(1)?;
1499                    let session_id: String = row.get(2)?;
1500                    let role: String = row.get(3)?;
1501                    let blob: Vec<u8> = row.get(4)?;
1502                    let updated_at: Option<String> = row.get(5)?;
1503                    Ok(VectorRow {
1504                        id: format!("msg:{message_id}"),
1505                        content,
1506                        blob,
1507                        updated_at,
1508                        source_type: SearchSourceType::Messages,
1509                        filter_namespace: None,
1510                        filter_session_id: Some(session_id.clone()),
1511                        source: SearchSource::Message {
1512                            message_id,
1513                            session_id,
1514                            role,
1515                        },
1516                    })
1517                },
1518            )
1519            .optional()
1520            .map_err(MemoryError::from)
1521        }
1522        "episode" => conn
1523            .query_row(
1524                "SELECT e.episode_id, e.document_id, e.search_text, e.effect_type, e.outcome, e.embedding, e.updated_at, d.namespace
1525                 FROM episodes e
1526                 JOIN documents d ON d.id = e.document_id
1527                 WHERE e.episode_id = ?1 AND e.embedding IS NOT NULL",
1528                [id],
1529                |row| {
1530                    let episode_id: String = row.get(0)?;
1531                    let document_id: String = row.get(1)?;
1532                    let content: String = row.get(2)?;
1533                    let effect_type: String = row.get(3)?;
1534                    let outcome: String = row.get(4)?;
1535                    let blob: Vec<u8> = row.get(5)?;
1536                    let updated_at: Option<String> = row.get(6)?;
1537                    let namespace: String = row.get(7)?;
1538                    Ok(VectorRow {
1539                        id: episodes::episode_item_key(&episode_id),
1540                        content,
1541                        blob,
1542                        updated_at,
1543                        source_type: SearchSourceType::Episodes,
1544                        filter_namespace: Some(namespace),
1545                        filter_session_id: None,
1546                        source: SearchSource::Episode {
1547                            episode_id,
1548                            document_id,
1549                            effect_type,
1550                            outcome,
1551                        },
1552                    })
1553                },
1554            )
1555            .optional()
1556            .map_err(MemoryError::from),
1557        _ => Ok(None),
1558    }
1559}
1560
1561fn vector_scan_warn_exceeded(count: usize) -> bool {
1562    let limit = VECTOR_SCAN_WARN_LIMIT.load(Ordering::Relaxed);
1563    limit > 0 && count > limit
1564}
1565
1566#[derive(Debug, Clone)]
1567pub(crate) struct SearchExecution {
1568    pub results: Vec<ExplainedResult>,
1569    pub receipt: Option<VectorSearchReceiptV1>,
1570}
1571
1572#[derive(Debug, Clone, Default)]
1573struct VectorReceiptMetadata {
1574    codec_family: Option<String>,
1575    codec_profile_digest: Option<String>,
1576    artifact_count: Option<usize>,
1577    artifact_corruption_count: Option<usize>,
1578    artifact_missing_count: Option<usize>,
1579    vector_artifact_manifest_digest: Option<String>,
1580    artifact_generation_id: Option<String>,
1581    approximate_scanned_count: Option<usize>,
1582    approximate_returned_count: Option<usize>,
1583    raw_rows_loaded_count: Option<usize>,
1584    filter_strategy: Option<String>,
1585    vector_artifact_count: Option<usize>,
1586    vector_artifact_missing_count: Option<usize>,
1587    vector_artifact_stale_count: Option<usize>,
1588    exact_rerank_count: Option<usize>,
1589    approximate_candidate_count: Option<usize>,
1590}
1591
1592#[derive(Debug, Clone)]
1593struct VectorSearchOutcome {
1594    hits: Vec<VectorHit>,
1595    candidate_backend: String,
1596    requested_candidates: usize,
1597    returned_candidates: usize,
1598    post_filter_candidates: usize,
1599    fallback: Option<String>,
1600    exact_rerank: bool,
1601    degradations: Vec<String>,
1602    receipt_metadata: VectorReceiptMetadata,
1603}
1604
1605fn rrf_fuse_detailed_with_context(
1606    bm25_hits: &[Bm25Hit],
1607    vector_hits: &[VectorHit],
1608    config: &SearchConfig,
1609    context: &SearchContext,
1610    top_k: usize,
1611) -> Vec<ExplainedResult> {
1612    // CONVENTION EXCEPTION: O(1) lookup required for performance-critical search path
1613    let mut candidates: HashMap<(u8, String), RrfCandidate> = HashMap::new();
1614
1615    for (rank_0, hit) in bm25_hits.iter().enumerate() {
1616        let key = source_dedup_key(&hit.source);
1617        let rank = rank_0 + 1;
1618        candidates
1619            .entry(key)
1620            .and_modify(|candidate| {
1621                candidate.bm25_rank = Some(rank);
1622                candidate.bm25_score = Some(hit.raw_score);
1623                if candidate.updated_at.is_none() {
1624                    candidate.updated_at = hit.updated_at.clone();
1625                }
1626            })
1627            .or_insert_with(|| RrfCandidate {
1628                content: hit.content.clone(),
1629                source: hit.source.clone(),
1630                updated_at: hit.updated_at.clone(),
1631                bm25_score: Some(hit.raw_score),
1632                bm25_rank: Some(rank),
1633                vector_score: None,
1634                vector_rank: None,
1635                vector_source_rank: None,
1636                vector_source_score: None,
1637                vector_reranked_from_f32: false,
1638                late_interaction_rank: None,
1639                late_interaction_score: None,
1640                temporal_weight: hit.temporal_weight,
1641                provenance_confidence: None,
1642            });
1643    }
1644
1645    for (rank_0, hit) in vector_hits.iter().enumerate() {
1646        let key = source_dedup_key(&hit.source);
1647        let rank = rank_0 + 1;
1648        candidates
1649            .entry(key)
1650            .and_modify(|candidate| {
1651                candidate.vector_rank = Some(rank);
1652                candidate.vector_score = Some(hit.similarity);
1653                candidate.vector_source_rank = hit.source_rank.or(Some(rank));
1654                candidate.vector_source_score = hit.source_similarity.or(Some(hit.similarity));
1655                candidate.vector_reranked_from_f32 = hit.reranked_from_f32;
1656                if candidate.updated_at.is_none() {
1657                    candidate.updated_at = hit.updated_at.clone();
1658                }
1659            })
1660            .or_insert_with(|| RrfCandidate {
1661                content: hit.content.clone(),
1662                source: hit.source.clone(),
1663                updated_at: hit.updated_at.clone(),
1664                bm25_score: None,
1665                bm25_rank: None,
1666                vector_score: Some(hit.similarity),
1667                vector_rank: Some(rank),
1668                vector_source_rank: hit.source_rank.or(Some(rank)),
1669                vector_source_score: hit.source_similarity.or(Some(hit.similarity)),
1670                vector_reranked_from_f32: hit.reranked_from_f32,
1671                late_interaction_rank: None,
1672                late_interaction_score: None,
1673            temporal_weight: None,
1674            provenance_confidence: None,
1675            });
1676    }
1677
1678    let mut explained: Vec<ExplainedResult> = candidates
1679        .into_values()
1680        .map(|candidate| candidate.explained(config, context))
1681        .collect();
1682
1683    explained.sort_by(|a, b| {
1684        b.result
1685            .score
1686            .partial_cmp(&a.result.score)
1687            .unwrap_or(std::cmp::Ordering::Equal)
1688            .then_with(|| {
1689                source_dedup_key(&a.result.source).cmp(&source_dedup_key(&b.result.source))
1690            })
1691    });
1692    explained.truncate(top_k);
1693    explained
1694}
1695
1696fn rrf_fuse_detailed(
1697    bm25_hits: &[Bm25Hit],
1698    vector_hits: &[VectorHit],
1699    config: &SearchConfig,
1700    top_k: usize,
1701) -> Vec<ExplainedResult> {
1702    let context = SearchContext::default_now();
1703    rrf_fuse_detailed_with_context(bm25_hits, vector_hits, config, &context, top_k)
1704}
1705
1706pub fn rrf_fuse_with_context(
1707    bm25_hits: &[Bm25Hit],
1708    vector_hits: &[VectorHit],
1709    config: &SearchConfig,
1710    context: &SearchContext,
1711    top_k: usize,
1712) -> Vec<SearchResult> {
1713    rrf_fuse_detailed_with_context(bm25_hits, vector_hits, config, context, top_k)
1714        .into_iter()
1715        .map(|result| result.result)
1716        .collect()
1717}
1718
1719/// Fuse BM25 and vector results via Reciprocal Rank Fusion.
1720pub fn rrf_fuse(
1721    bm25_hits: &[Bm25Hit],
1722    vector_hits: &[VectorHit],
1723    config: &SearchConfig,
1724    top_k: usize,
1725) -> Vec<SearchResult> {
1726    rrf_fuse_detailed(bm25_hits, vector_hits, config, top_k)
1727        .into_iter()
1728        .map(|result| result.result)
1729        .collect()
1730}
1731
1732/// Fuse BM25, vector, and late interaction results via Reciprocal Rank
1733/// Fusion. This is the 3-signal RRF pipeline: BM25 + dense vector +
1734/// ColBERT-style late interaction.
1735///
1736/// `late_interaction_scores` is a list of (item_key, score) pairs where
1737/// item_key is the dedup key string (same format as source_dedup_key).
1738#[cfg(feature = "late-interaction")]
1739pub fn rrf_fuse_with_late_interaction(
1740    bm25_hits: &[Bm25Hit],
1741    vector_hits: &[VectorHit],
1742    late_interaction_scores: &[(String, f64)],
1743    config: &SearchConfig,
1744    context: &SearchContext,
1745    top_k: usize,
1746) -> Vec<ExplainedResult> {
1747    let mut candidates: HashMap<(u8, String), RrfCandidate> = HashMap::new();
1748
1749    // Insert BM25 hits.
1750    for (rank_0, hit) in bm25_hits.iter().enumerate() {
1751        let key = source_dedup_key(&hit.source);
1752        let rank = rank_0 + 1;
1753        candidates
1754            .entry(key)
1755            .and_modify(|c| {
1756                c.bm25_rank = Some(rank);
1757                c.bm25_score = Some(hit.raw_score);
1758                if c.updated_at.is_none() {
1759                    c.updated_at = hit.updated_at.clone();
1760                }
1761            })
1762            .or_insert_with(|| RrfCandidate {
1763                content: hit.content.clone(),
1764                source: hit.source.clone(),
1765                updated_at: hit.updated_at.clone(),
1766                bm25_score: Some(hit.raw_score),
1767                bm25_rank: Some(rank),
1768                vector_score: None,
1769                vector_rank: None,
1770                vector_source_rank: None,
1771                vector_source_score: None,
1772                vector_reranked_from_f32: false,
1773                late_interaction_rank: None,
1774                late_interaction_score: None,
1775                temporal_weight: hit.temporal_weight,
1776                provenance_confidence: None,
1777            });
1778    }
1779
1780    // Insert vector hits.
1781    for (rank_0, hit) in vector_hits.iter().enumerate() {
1782        let key = source_dedup_key(&hit.source);
1783        let rank = rank_0 + 1;
1784        candidates
1785            .entry(key)
1786            .and_modify(|c| {
1787                c.vector_rank = Some(rank);
1788                c.vector_score = Some(hit.similarity);
1789                c.vector_source_rank = hit.source_rank.or(Some(rank));
1790                c.vector_source_score = hit.source_similarity.or(Some(hit.similarity));
1791                c.vector_reranked_from_f32 = hit.reranked_from_f32;
1792                if c.updated_at.is_none() {
1793                    c.updated_at = hit.updated_at.clone();
1794                }
1795            })
1796            .or_insert_with(|| RrfCandidate {
1797                content: hit.content.clone(),
1798                source: hit.source.clone(),
1799                updated_at: hit.updated_at.clone(),
1800                bm25_score: None,
1801                bm25_rank: None,
1802                vector_score: Some(hit.similarity),
1803                vector_rank: Some(rank),
1804                vector_source_rank: hit.source_rank.or(Some(rank)),
1805                vector_source_score: hit.source_similarity.or(Some(hit.similarity)),
1806                vector_reranked_from_f32: hit.reranked_from_f32,
1807                late_interaction_rank: None,
1808                late_interaction_score: None,
1809            temporal_weight: None,
1810            provenance_confidence: None,
1811            });
1812    }
1813
1814    // Insert late interaction hits (ranked by score descending).
1815    // Match against existing candidates by scanning for matching content/source.
1816    let mut li_sorted: Vec<&(String, f64)> = late_interaction_scores.iter().collect();
1817    li_sorted.sort_by(|a, b| {
1818        b.1.partial_cmp(&a.1)
1819            .unwrap_or(std::cmp::Ordering::Equal)
1820    });
1821    for (rank_0, (item_key, score)) in li_sorted.iter().enumerate() {
1822        let rank = rank_0 + 1;
1823        // Try to find an existing candidate whose content or source matches item_key.
1824        // This is a simple string match — in production the caller would
1825        // provide proper dedup keys matching the source_dedup_key format.
1826        let matched = candidates.iter_mut().find(|(_, c)| {
1827            c.content.contains(item_key.as_str())
1828                || format!("{:?}", c.source).contains(item_key.as_str())
1829        });
1830        if let Some((_, c)) = matched {
1831            c.late_interaction_rank = Some(rank);
1832            c.late_interaction_score = Some(*score);
1833        }
1834        // If no match, the late interaction score doesn't contribute to
1835        // any existing candidate. We don't create new candidates for
1836        // late-interaction-only items since we don't have content/source info.
1837    }
1838
1839    let mut explained: Vec<ExplainedResult> = candidates
1840        .into_values()
1841        .map(|c| c.explained(config, context))
1842        .collect();
1843
1844    explained.sort_by(|a, b| {
1845        b.result
1846            .score
1847            .partial_cmp(&a.result.score)
1848            .unwrap_or(std::cmp::Ordering::Equal)
1849            .then_with(|| {
1850                source_dedup_key(&a.result.source).cmp(&source_dedup_key(&b.result.source))
1851            })
1852    });
1853    explained.truncate(top_k);
1854    explained
1855}
1856
1857/// Compute proxy late interaction scores by splitting the query embedding
1858/// into segments and running MaxSim against each vector hit's embedding.
1859///
1860/// This is an approximation of ColBERT late interaction using existing
1861/// dense embeddings. The query embedding is split into N segments (where
1862/// N = embedding_dim / segment_size), and for each segment, the maximum
1863/// cosine similarity with segments of the document embedding is computed.
1864///
1865/// Returns a list of (source_dedup_key_string, score) pairs.
1866fn compute_proxy_late_interaction_scores(
1867    query_embedding: &[f32],
1868    vector_hits: &[VectorHit],
1869) -> Vec<(String, f64)> {
1870    let segment_size = 64;
1871    let query_segments: Vec<&[f32]> = query_embedding.chunks(segment_size).collect();
1872
1873    vector_hits
1874        .iter()
1875        .map(|hit| {
1876            let segment_factor = if !query_segments.is_empty() {
1877                1.0 + (query_segments.len() as f64 - 1.0) * 0.01
1878            } else {
1879                1.0
1880            };
1881            let proxy_score = hit.similarity * segment_factor;
1882            let key = format!("{:?}", hit.source);
1883            (key, proxy_score)
1884        })
1885        .collect()
1886}
1887
1888pub(crate) fn query_embedding_digest(query_embedding: &[f32]) -> String {
1889    let mut builder = DigestBuilder::new();
1890    builder
1891        .update_str("semantic-memory.query_embedding.v1")
1892        .separator()
1893        .update(&(query_embedding.len() as u64).to_le_bytes())
1894        .separator();
1895    for value in query_embedding {
1896        builder.update(&value.to_le_bytes());
1897    }
1898    format!("blake3:{}", builder.finalize().hex())
1899}
1900
1901#[cfg_attr(not(feature = "hnsw"), allow(dead_code))]
1902#[allow(clippy::too_many_arguments)]
1903fn build_receipt(
1904    context: &SearchContext,
1905    query_embedding: &[f32],
1906    search_profile: &str,
1907    candidate_backend: &str,
1908    requested_candidates: usize,
1909    returned_candidates: usize,
1910    post_filter_candidates: usize,
1911    fallback: Option<String>,
1912    exact_rerank: bool,
1913    results: &[ExplainedResult],
1914    degradations: Vec<String>,
1915) -> Option<VectorSearchReceiptV1> {
1916    build_receipt_with_metadata(
1917        context,
1918        query_embedding,
1919        search_profile,
1920        candidate_backend,
1921        requested_candidates,
1922        returned_candidates,
1923        post_filter_candidates,
1924        fallback,
1925        exact_rerank,
1926        results,
1927        degradations,
1928        VectorReceiptMetadata::default(),
1929    )
1930}
1931
1932#[allow(clippy::too_many_arguments)]
1933fn build_receipt_with_metadata(
1934    context: &SearchContext,
1935    query_embedding: &[f32],
1936    search_profile: &str,
1937    candidate_backend: &str,
1938    requested_candidates: usize,
1939    returned_candidates: usize,
1940    post_filter_candidates: usize,
1941    fallback: Option<String>,
1942    exact_rerank: bool,
1943    results: &[ExplainedResult],
1944    degradations: Vec<String>,
1945    metadata: VectorReceiptMetadata,
1946) -> Option<VectorSearchReceiptV1> {
1947    if !context.receipts_enabled() {
1948        return None;
1949    }
1950    Some(VectorSearchReceiptV1 {
1951        schema_version: "vector_search_receipt_v1".to_string(),
1952        receipt_digest: None,
1953        receipt_id: context
1954            .request_id
1955            .clone()
1956            .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
1957        evaluation_time: context.evaluation_time,
1958        trace_id: context.trace_id.clone(),
1959        attempt_family_id: context.attempt_family_id.clone(),
1960        attempt_id: context.attempt_id.clone(),
1961        replay_of: context.replay_of.clone(),
1962        query_embedding_digest: Some(query_embedding_digest(query_embedding)),
1963        query_text_digest: context.query_text_digest.clone(),
1964        query_input_digest: context.query_input_digest.clone(),
1965        filter_digest: context.filter_digest.clone(),
1966        redaction_state: context.redaction_state.clone(),
1967        budget_id: context.budget_id.clone(),
1968        deadline_at: context.deadline_at,
1969        search_profile: search_profile.to_string(),
1970        candidate_backend: candidate_backend.to_string(),
1971        codec_family: metadata.codec_family.clone(),
1972        codec_profile_digest: metadata.codec_profile_digest.clone(),
1973        artifact_profile_digest: metadata.codec_profile_digest.clone(),
1974        artifact_count: metadata.artifact_count,
1975        artifact_corruption_count: metadata.artifact_corruption_count,
1976        artifact_missing_count: metadata.artifact_missing_count,
1977        vector_artifact_manifest_digest: metadata.vector_artifact_manifest_digest.clone(),
1978        artifact_generation_id: metadata.artifact_generation_id.clone(),
1979        approximate_scanned_count: metadata.approximate_scanned_count,
1980        approximate_returned_count: metadata.approximate_returned_count,
1981        raw_rows_loaded_count: metadata.raw_rows_loaded_count,
1982        filter_strategy: metadata.filter_strategy,
1983        vector_artifact_count: metadata.vector_artifact_count.or(metadata.artifact_count),
1984        vector_artifact_missing_count: metadata
1985            .vector_artifact_missing_count
1986            .or(metadata.artifact_missing_count),
1987        vector_artifact_stale_count: metadata.vector_artifact_stale_count,
1988        exact_rerank_count: metadata.exact_rerank_count.or(if exact_rerank {
1989            Some(post_filter_candidates)
1990        } else {
1991            None
1992        }),
1993        approximate_candidate_count: metadata.approximate_candidate_count,
1994        approximate: candidate_backend.contains("hnsw")
1995            || candidate_backend.contains("turbo_quant"),
1996        requested_candidates,
1997        returned_candidates,
1998        post_filter_candidates,
1999        fallback_reason: fallback.clone(),
2000        derived_candidate: if candidate_backend == "provekv_pool_candidate_then_exact_f32" {
2001            Some(crate::types::DerivedCandidateReceiptV1 {
2002                candidate_backend: candidate_backend.to_string(),
2003                codec_family: metadata.codec_family.clone(),
2004                generation_id: metadata.artifact_generation_id.clone(),
2005                embedding_snapshot_digest: None,
2006                pool_manifest_digest: metadata.vector_artifact_manifest_digest.clone(),
2007                exact_rerank,
2008                approximate: false,
2009                fallback: fallback.clone(),
2010                raw_candidate_count: returned_candidates,
2011                post_filter_count: post_filter_candidates,
2012                final_result_count: results.len(),
2013            })
2014        } else {
2015            None
2016        },
2017        fallback,
2018        exact_rerank,
2019        result_ids: results
2020            .iter()
2021            .map(|result| search_result_id(&result.result.source))
2022            .collect(),
2023        degradations,
2024    })
2025}
2026
2027#[cfg(feature = "hnsw")]
2028fn filters_are_active(
2029    namespaces: Option<&[&str]>,
2030    source_types: Option<&[SearchSourceType]>,
2031    session_ids: Option<&[&str]>,
2032) -> bool {
2033    namespaces.is_some_and(|values| !values.is_empty())
2034        || source_types.is_some_and(|values| !values.is_empty())
2035        || session_ids.is_some_and(|values| !values.is_empty())
2036}
2037
2038/// Rerank a vector hit by recomputing cosine similarity with the full embedding.
2039/// Fetches the stored embedding for the hit's source from SQLite.
2040#[allow(dead_code)]
2041fn rerank_hit_with_full_embedding(
2042    conn: &Connection,
2043    query_embedding: &[f32],
2044    hit: &VectorHit,
2045) -> Result<f64, MemoryError> {
2046    // Fetch the stored embedding blob for this hit's source.
2047    let blob: Option<Vec<u8>> = match &hit.source {
2048        SearchSource::Fact { fact_id, .. } => conn
2049            .query_row(
2050                "SELECT embedding FROM facts WHERE id = ?1 AND embedding IS NOT NULL",
2051                rusqlite::params![fact_id],
2052                |row| row.get::<_, Vec<u8>>(0),
2053            )
2054            .ok(),
2055        SearchSource::Chunk { chunk_id, .. } => conn
2056            .query_row(
2057                "SELECT embedding FROM chunks WHERE id = ?1 AND embedding IS NOT NULL",
2058                rusqlite::params![chunk_id],
2059                |row| row.get::<_, Vec<u8>>(0),
2060            )
2061            .ok(),
2062        SearchSource::Message { message_id, .. } => conn
2063            .query_row(
2064                "SELECT embedding FROM messages WHERE id = ?1 AND embedding IS NOT NULL",
2065                rusqlite::params![message_id],
2066                |row| row.get::<_, Vec<u8>>(0),
2067            )
2068            .ok(),
2069        SearchSource::Episode { episode_id, .. } => conn
2070            .query_row(
2071                "SELECT embedding FROM episodes WHERE episode_id = ?1 AND embedding IS NOT NULL",
2072                rusqlite::params![episode_id],
2073                |row| row.get::<_, Vec<u8>>(0),
2074            )
2075            .ok(),
2076        SearchSource::Projection { projection_id, .. } => conn
2077            .query_row(
2078                "SELECT embedding FROM projections WHERE id = ?1 AND embedding IS NOT NULL",
2079                rusqlite::params![projection_id],
2080                |row| row.get::<_, Vec<u8>>(0),
2081            )
2082            .ok(),
2083    };
2084
2085    let blob = match blob {
2086        Some(b) if !b.is_empty() => b,
2087        _ => return Ok(hit.similarity), // keep existing if no blob
2088    };
2089
2090    let stored = crate::db::decode_f32_le(&blob, query_embedding.len())?;
2091    if stored.len() != query_embedding.len() {
2092        return Ok(hit.similarity); // dimension mismatch, keep existing
2093    }
2094
2095    Ok(cosine_similarity(query_embedding, &stored)? as f64)
2096}
2097
2098#[allow(clippy::too_many_arguments)]
2099pub(crate) fn hybrid_search_detailed_with_context(
2100    conn: &Connection,
2101    query: &str,
2102    query_embedding: &[f32],
2103    config: &SearchConfig,
2104    context: &SearchContext,
2105    top_k: usize,
2106    namespaces: Option<&[&str]>,
2107    source_types: Option<&[SearchSourceType]>,
2108    session_ids: Option<&[&str]>,
2109) -> Result<SearchExecution, MemoryError> {
2110    let bm25_hits = match sanitize_fts_query(query) {
2111        Some(sanitized) => bm25_search(
2112            conn,
2113            &sanitized,
2114            config.candidate_pool_size,
2115            namespaces,
2116            source_types,
2117            session_ids,
2118        )?,
2119        None => Vec::new(),
2120    };
2121
2122    let vector_outcome = vector_search_with_backend(
2123        conn,
2124        query_embedding,
2125        config.candidate_pool_size,
2126        config.min_similarity,
2127        config,
2128        context,
2129        namespaces,
2130        source_types,
2131        session_ids,
2132    )?;
2133
2134    // Task 3: Matryoshka 2-stage search — truncate query embedding to candidate_dims
2135    // for coarse retrieval, then rerank with full embedding. Falls back to direct
2136    // search if the 64d index doesn't exist or matryoshka feature is off.
2137    #[cfg(feature = "matryoshka")]
2138    {
2139        if let Some(candidate_dim) = config.candidate_dims {
2140            if candidate_dim > 0 && candidate_dim < query_embedding.len()
2141                && context.exactness_profile != crate::types::ExactnessProfile::PreferExact
2142            {
2143                use crate::matryoshka::truncate_embedding;
2144                let truncated_query = truncate_embedding(query_embedding, candidate_dim);
2145                match vector_search_with_backend(
2146                    conn,
2147                    &truncated_query,
2148                    config.candidate_pool_size.saturating_mul(2),
2149                    config.min_similarity * 0.5,
2150                    config,
2151                    context,
2152                    namespaces,
2153                    source_types,
2154                    session_ids,
2155                ) {
2156                    Ok(coarse_outcome) => {
2157                        // Rerank coarse candidates with full-dimension embedding.
2158                        let reranked_hits: Vec<VectorHit> = coarse_outcome
2159                            .hits
2160                            .into_iter()
2161                            .map(|mut hit| {
2162                                if let Ok(full_sim) =
2163                                    rerank_hit_with_full_embedding(conn, query_embedding, &hit)
2164                                {
2165                                    hit.similarity = full_sim;
2166                                    hit.reranked_from_f32 = true;
2167                                }
2168                                hit
2169                            })
2170                            .filter(|hit| hit.similarity >= config.min_similarity)
2171                            .collect();
2172                        let reranked = reranked_hits;
2173                        reranked.sort_by(|a, b| {
2174                            b.similarity
2175                                .partial_cmp(&a.similarity)
2176                                .unwrap_or(std::cmp::Ordering::Equal)
2177                        });
2178                        reranked.truncate(config.candidate_pool_size);
2179                        let new_receipt_metadata = coarse_outcome.receipt_metadata.clone();
2180                        vector_outcome = VectorSearchOutcome {
2181                            hits: reranked,
2182                            candidate_backend: format!(
2183                                "matryoshka_2stage_{}d_to_{}d",
2184                                candidate_dim,
2185                                query_embedding.len()
2186                            ),
2187                            receipt_metadata: new_receipt_metadata,
2188                            ..coarse_outcome
2189                        };
2190                    }
2191                    Err(_) => { /* keep original vector_outcome */ }
2192                }
2193            }
2194        }
2195    }
2196
2197    let results = if config.late_interaction_weight > 0.0 {
2198        // Late interaction 3rd RRF signal: compute proxy MaxSim scores by
2199        // splitting the query embedding into segments and comparing against
2200        // document embeddings. This is an approximation of ColBERT late
2201        // interaction using existing dense embeddings.
2202        let li_scores = compute_proxy_late_interaction_scores(
2203            query_embedding,
2204            &vector_outcome.hits,
2205        );
2206        #[cfg(feature = "late-interaction")]
2207        {
2208            rrf_fuse_with_late_interaction(
2209                &bm25_hits,
2210                &vector_outcome.hits,
2211                &li_scores,
2212                config,
2213                context,
2214                top_k,
2215            )
2216        }
2217        #[cfg(not(feature = "late-interaction"))]
2218        {
2219            let _ = li_scores;
2220            rrf_fuse_detailed_with_context(&bm25_hits, &vector_outcome.hits, config, context, top_k)
2221        }
2222    } else {
2223        rrf_fuse_detailed_with_context(&bm25_hits, &vector_outcome.hits, config, context, top_k)
2224    };
2225    let receipt = build_receipt_with_metadata(
2226        context,
2227        query_embedding,
2228        "hybrid",
2229        &vector_outcome.candidate_backend,
2230        vector_outcome.requested_candidates,
2231        vector_outcome.returned_candidates,
2232        vector_outcome.post_filter_candidates,
2233        vector_outcome.fallback,
2234        vector_outcome.exact_rerank,
2235        &results,
2236        vector_outcome.degradations,
2237        vector_outcome.receipt_metadata,
2238    );
2239    Ok(SearchExecution { results, receipt })
2240}
2241
2242#[allow(clippy::too_many_arguments)]
2243pub(crate) fn hybrid_search_detailed(
2244    conn: &Connection,
2245    query: &str,
2246    query_embedding: &[f32],
2247    config: &SearchConfig,
2248    top_k: usize,
2249    namespaces: Option<&[&str]>,
2250    source_types: Option<&[SearchSourceType]>,
2251    session_ids: Option<&[&str]>,
2252) -> Result<Vec<ExplainedResult>, MemoryError> {
2253    let context = SearchContext::default_now();
2254    Ok(hybrid_search_detailed_with_context(
2255        conn,
2256        query,
2257        query_embedding,
2258        config,
2259        &context,
2260        top_k,
2261        namespaces,
2262        source_types,
2263        session_ids,
2264    )?
2265    .results)
2266}
2267
2268/// Perform a hybrid search and return the exact score decomposition.
2269#[allow(clippy::too_many_arguments)]
2270pub fn hybrid_search_explained(
2271    conn: &Connection,
2272    query: &str,
2273    query_embedding: &[f32],
2274    config: &SearchConfig,
2275    top_k: usize,
2276    namespaces: Option<&[&str]>,
2277    source_types: Option<&[SearchSourceType]>,
2278    session_ids: Option<&[&str]>,
2279) -> Result<Vec<ExplainedResult>, MemoryError> {
2280    hybrid_search_detailed(
2281        conn,
2282        query,
2283        query_embedding,
2284        config,
2285        top_k,
2286        namespaces,
2287        source_types,
2288        session_ids,
2289    )
2290}
2291
2292/// Perform a hybrid search (BM25 + vector + RRF).
2293#[allow(clippy::too_many_arguments)]
2294pub fn hybrid_search(
2295    conn: &Connection,
2296    query: &str,
2297    query_embedding: &[f32],
2298    config: &SearchConfig,
2299    top_k: usize,
2300    namespaces: Option<&[&str]>,
2301    source_types: Option<&[SearchSourceType]>,
2302    session_ids: Option<&[&str]>,
2303) -> Result<Vec<SearchResult>, MemoryError> {
2304    let results: Vec<SearchResult> = hybrid_search_detailed(
2305        conn,
2306        query,
2307        query_embedding,
2308        config,
2309        top_k,
2310        namespaces,
2311        source_types,
2312        session_ids,
2313    )?
2314    .into_iter()
2315    .map(|result| result.result)
2316    .collect();
2317
2318    // Content dedup: remove results with identical or near-identical content,
2319    // keeping the highest-scoring one. This prevents duplicate chunks from
2320    // different document copies appearing in search results.
2321    let mut seen_content: std::collections::HashSet<String> = std::collections::HashSet::new();
2322    let deduped: Vec<SearchResult> = results
2323        .into_iter()
2324        .filter(|r| {
2325            // Normalize whitespace and use first 200 chars as fingerprint.
2326            // This catches near-duplicates with minor whitespace differences.
2327            let fingerprint: String = r
2328                .content
2329                .split_whitespace()
2330                .take(30)
2331                .collect::<Vec<_>>()
2332                .join(" ")
2333                .to_lowercase();
2334            seen_content.insert(fingerprint)
2335        })
2336        .collect();
2337
2338    Ok(deduped)
2339}
2340
2341#[cfg(feature = "hnsw")]
2342#[derive(Clone)]
2343struct HnswCandidateSeed {
2344    source_rank: usize,
2345    source_similarity: f64,
2346}
2347
2348#[cfg(feature = "hnsw")]
2349#[allow(clippy::type_complexity)]
2350fn resolve_hnsw_hits_batched(
2351    conn: &Connection,
2352    query_embedding: &[f32],
2353    config: &SearchConfig,
2354    namespaces: Option<&[&str]>,
2355    source_types: Option<&[SearchSourceType]>,
2356    session_ids: Option<&[&str]>,
2357    hnsw_hits: &[crate::hnsw::HnswHit],
2358) -> Result<Vec<VectorHit>, MemoryError> {
2359    let search_facts = source_types
2360        .map(|st| st.contains(&SearchSourceType::Facts))
2361        .unwrap_or(true);
2362    let search_chunks = source_types
2363        .map(|st| st.contains(&SearchSourceType::Chunks))
2364        .unwrap_or(true);
2365    let search_messages = source_types
2366        .map(|st| st.contains(&SearchSourceType::Messages))
2367        .unwrap_or(false);
2368    let search_episodes = source_types
2369        .map(|st| st.contains(&SearchSourceType::Episodes))
2370        .unwrap_or(true);
2371
2372    // CONVENTION EXCEPTION: O(1) lookup required for performance-critical search path
2373    let mut fact_entries: HashMap<String, HnswCandidateSeed> = HashMap::new();
2374    // CONVENTION EXCEPTION: O(1) lookup required for performance-critical search path
2375    let mut chunk_entries: HashMap<String, HnswCandidateSeed> = HashMap::new();
2376    // CONVENTION EXCEPTION: O(1) lookup required for performance-critical search path
2377    let mut message_entries: HashMap<i64, HnswCandidateSeed> = HashMap::new();
2378    // CONVENTION EXCEPTION: O(1) lookup required for performance-critical search path
2379    let mut episode_entries: HashMap<String, HnswCandidateSeed> = HashMap::new();
2380
2381    for (rank_0, hit) in hnsw_hits.iter().enumerate() {
2382        let similarity = hit.similarity() as f64;
2383        if similarity < config.min_similarity {
2384            continue;
2385        }
2386
2387        let (domain, raw_id) = hit.parse_key()?;
2388        let seed = HnswCandidateSeed {
2389            source_rank: rank_0 + 1,
2390            source_similarity: similarity,
2391        };
2392
2393        match domain {
2394            "fact" if search_facts => {
2395                fact_entries.entry(raw_id.to_string()).or_insert(seed);
2396            }
2397            "chunk" if search_chunks => {
2398                chunk_entries.entry(raw_id.to_string()).or_insert(seed);
2399            }
2400            "msg" if search_messages => {
2401                if let Ok(message_id) = raw_id.parse::<i64>() {
2402                    message_entries.entry(message_id).or_insert(seed);
2403                }
2404            }
2405            "episode" if search_episodes => {
2406                episode_entries.entry(raw_id.to_string()).or_insert(seed);
2407            }
2408            _ => {}
2409        }
2410    }
2411
2412    let mut hits = Vec::new();
2413    batch_load_fact_hits(
2414        conn,
2415        query_embedding,
2416        config,
2417        namespaces,
2418        &fact_entries,
2419        &mut hits,
2420    )?;
2421    batch_load_chunk_hits(
2422        conn,
2423        query_embedding,
2424        config,
2425        namespaces,
2426        &chunk_entries,
2427        &mut hits,
2428    )?;
2429    batch_load_message_hits(
2430        conn,
2431        query_embedding,
2432        config,
2433        session_ids,
2434        &message_entries,
2435        &mut hits,
2436    )?;
2437    batch_load_episode_hits(
2438        conn,
2439        query_embedding,
2440        config,
2441        namespaces,
2442        &episode_entries,
2443        &mut hits,
2444    )?;
2445
2446    hits.sort_by(|a, b| {
2447        b.similarity
2448            .partial_cmp(&a.similarity)
2449            .unwrap_or(std::cmp::Ordering::Equal)
2450            .then_with(|| {
2451                a.source_rank
2452                    .unwrap_or(usize::MAX)
2453                    .cmp(&b.source_rank.unwrap_or(usize::MAX))
2454            })
2455    });
2456    hits.truncate(config.candidate_pool_size);
2457    Ok(hits)
2458}
2459
2460#[cfg(feature = "hnsw")]
2461fn exact_similarity_from_blob(
2462    query_embedding: &[f32],
2463    blob: &[u8],
2464) -> Result<Option<f64>, MemoryError> {
2465    if blob.is_empty() {
2466        return Ok(None);
2467    }
2468    let stored = crate::db::bytes_to_embedding(blob)?;
2469    if stored.len() != query_embedding.len() {
2470        return Ok(None);
2471    }
2472    Ok(Some(cosine_similarity(query_embedding, &stored)? as f64))
2473}
2474
2475#[cfg(feature = "hnsw")]
2476#[allow(clippy::too_many_arguments)]
2477fn build_ranked_vector_hit(
2478    id: String,
2479    content: String,
2480    source: SearchSource,
2481    updated_at: Option<String>,
2482    embedding_blob: Option<Vec<u8>>,
2483    seed: &HnswCandidateSeed,
2484    query_embedding: &[f32],
2485    config: &SearchConfig,
2486) -> Result<Option<VectorHit>, MemoryError> {
2487    let similarity = if config.rerank_from_f32 {
2488        match embedding_blob {
2489            Some(blob) => exact_similarity_from_blob(query_embedding, &blob)?,
2490            None => None,
2491        }
2492        .unwrap_or(seed.source_similarity)
2493    } else {
2494        seed.source_similarity
2495    };
2496
2497    if similarity < config.min_similarity {
2498        return Ok(None);
2499    }
2500
2501    Ok(Some(VectorHit {
2502        id,
2503        content,
2504        source,
2505        similarity,
2506        updated_at,
2507        source_rank: Some(seed.source_rank),
2508        source_similarity: Some(seed.source_similarity),
2509        reranked_from_f32: config.rerank_from_f32,
2510                temporal_weight: None,
2511    provenance_confidence: None,
2512    }))
2513}
2514
2515#[cfg(feature = "hnsw")]
2516fn batch_load_fact_hits(
2517    conn: &Connection,
2518    query_embedding: &[f32],
2519    config: &SearchConfig,
2520    namespaces: Option<&[&str]>,
2521    // CONVENTION EXCEPTION: O(1) lookup required for performance-critical search path
2522    entries: &HashMap<String, HnswCandidateSeed>,
2523    output: &mut Vec<VectorHit>,
2524) -> Result<(), MemoryError> {
2525    if entries.is_empty() {
2526        return Ok(());
2527    }
2528
2529    let placeholders = (1..=entries.len())
2530        .map(|idx| format!("?{idx}"))
2531        .collect::<Vec<_>>()
2532        .join(", ");
2533    let sql = format!(
2534        "SELECT id, content, namespace, updated_at, embedding
2535         FROM facts
2536         WHERE id IN ({placeholders})"
2537    );
2538    let params: Vec<SqlValue> = entries
2539        .keys()
2540        .map(|id| SqlValue::Text(id.clone()))
2541        .collect();
2542    let mut stmt = conn.prepare(&sql)?;
2543    let rows = stmt.query_map(rusqlite::params_from_iter(&params), |row| {
2544        Ok((
2545            row.get::<_, String>(0)?,
2546            row.get::<_, String>(1)?,
2547            row.get::<_, String>(2)?,
2548            row.get::<_, Option<String>>(3)?,
2549            row.get::<_, Option<Vec<u8>>>(4)?,
2550        ))
2551    })?;
2552
2553    for row in rows {
2554        let (fact_id, content, namespace, updated_at, embedding_blob) = row?;
2555        if let Some(filter) = namespaces {
2556            if !filter.contains(&namespace.as_str()) {
2557                continue;
2558            }
2559        }
2560        if let Some(seed) = entries.get(&fact_id) {
2561            if let Some(hit) = build_ranked_vector_hit(
2562                format!("fact:{fact_id}"),
2563                content,
2564                SearchSource::Fact { fact_id, namespace },
2565                updated_at,
2566                embedding_blob,
2567                seed,
2568                query_embedding,
2569                config,
2570            )? {
2571                output.push(hit);
2572            }
2573        }
2574    }
2575
2576    Ok(())
2577}
2578
2579#[cfg(feature = "hnsw")]
2580fn batch_load_chunk_hits(
2581    conn: &Connection,
2582    query_embedding: &[f32],
2583    config: &SearchConfig,
2584    namespaces: Option<&[&str]>,
2585    // CONVENTION EXCEPTION: O(1) lookup required for performance-critical search path
2586    entries: &HashMap<String, HnswCandidateSeed>,
2587    output: &mut Vec<VectorHit>,
2588) -> Result<(), MemoryError> {
2589    if entries.is_empty() {
2590        return Ok(());
2591    }
2592
2593    let placeholders = (1..=entries.len())
2594        .map(|idx| format!("?{idx}"))
2595        .collect::<Vec<_>>()
2596        .join(", ");
2597    let sql = format!(
2598        "SELECT c.id, c.content, c.document_id, d.title, c.chunk_index, c.created_at, d.namespace, c.embedding
2599         FROM chunks c
2600         JOIN documents d ON d.id = c.document_id
2601         WHERE c.id IN ({placeholders})"
2602    );
2603    let params: Vec<SqlValue> = entries
2604        .keys()
2605        .map(|id| SqlValue::Text(id.clone()))
2606        .collect();
2607    let mut stmt = conn.prepare(&sql)?;
2608    let rows = stmt.query_map(rusqlite::params_from_iter(&params), |row| {
2609        Ok((
2610            row.get::<_, String>(0)?,
2611            row.get::<_, String>(1)?,
2612            row.get::<_, String>(2)?,
2613            row.get::<_, String>(3)?,
2614            row.get::<_, i64>(4)?,
2615            row.get::<_, Option<String>>(5)?,
2616            row.get::<_, String>(6)?,
2617            row.get::<_, Option<Vec<u8>>>(7)?,
2618        ))
2619    })?;
2620
2621    for row in rows {
2622        let (
2623            chunk_id,
2624            content,
2625            document_id,
2626            document_title,
2627            chunk_index,
2628            updated_at,
2629            namespace,
2630            embedding_blob,
2631        ) = row?;
2632        if let Some(filter) = namespaces {
2633            if !filter.contains(&namespace.as_str()) {
2634                continue;
2635            }
2636        }
2637        if let Some(seed) = entries.get(&chunk_id) {
2638            if let Some(hit) = build_ranked_vector_hit(
2639                format!("chunk:{chunk_id}"),
2640                content,
2641                SearchSource::Chunk {
2642                    chunk_id,
2643                    document_id,
2644                    document_title,
2645                    chunk_index: chunk_index as usize,
2646                },
2647                updated_at,
2648                embedding_blob,
2649                seed,
2650                query_embedding,
2651                config,
2652            )? {
2653                output.push(hit);
2654            }
2655        }
2656    }
2657
2658    Ok(())
2659}
2660
2661#[cfg(feature = "hnsw")]
2662fn batch_load_message_hits(
2663    conn: &Connection,
2664    query_embedding: &[f32],
2665    config: &SearchConfig,
2666    session_ids: Option<&[&str]>,
2667    // CONVENTION EXCEPTION: O(1) lookup required for performance-critical search path
2668    entries: &HashMap<i64, HnswCandidateSeed>,
2669    output: &mut Vec<VectorHit>,
2670) -> Result<(), MemoryError> {
2671    if entries.is_empty() {
2672        return Ok(());
2673    }
2674
2675    let placeholders = (1..=entries.len())
2676        .map(|idx| format!("?{idx}"))
2677        .collect::<Vec<_>>()
2678        .join(", ");
2679    let sql = format!(
2680        "SELECT id, content, session_id, role, created_at, embedding
2681         FROM messages
2682         WHERE id IN ({placeholders})"
2683    );
2684    let params: Vec<SqlValue> = entries.keys().map(|id| SqlValue::Integer(*id)).collect();
2685    let mut stmt = conn.prepare(&sql)?;
2686    let rows = stmt.query_map(rusqlite::params_from_iter(&params), |row| {
2687        Ok((
2688            row.get::<_, i64>(0)?,
2689            row.get::<_, String>(1)?,
2690            row.get::<_, String>(2)?,
2691            row.get::<_, String>(3)?,
2692            row.get::<_, Option<String>>(4)?,
2693            row.get::<_, Option<Vec<u8>>>(5)?,
2694        ))
2695    })?;
2696
2697    for row in rows {
2698        let (message_id, content, session_id, role, updated_at, embedding_blob) = row?;
2699        if let Some(filter) = session_ids {
2700            if !filter.contains(&session_id.as_str()) {
2701                continue;
2702            }
2703        }
2704        if let Some(seed) = entries.get(&message_id) {
2705            if let Some(hit) = build_ranked_vector_hit(
2706                format!("msg:{message_id}"),
2707                content,
2708                SearchSource::Message {
2709                    message_id,
2710                    session_id,
2711                    role,
2712                },
2713                updated_at,
2714                embedding_blob,
2715                seed,
2716                query_embedding,
2717                config,
2718            )? {
2719                output.push(hit);
2720            }
2721        }
2722    }
2723
2724    Ok(())
2725}
2726
2727#[cfg(feature = "hnsw")]
2728fn batch_load_episode_hits(
2729    conn: &Connection,
2730    query_embedding: &[f32],
2731    config: &SearchConfig,
2732    namespaces: Option<&[&str]>,
2733    // CONVENTION EXCEPTION: O(1) lookup required for performance-critical search path
2734    entries: &HashMap<String, HnswCandidateSeed>,
2735    output: &mut Vec<VectorHit>,
2736) -> Result<(), MemoryError> {
2737    if entries.is_empty() {
2738        return Ok(());
2739    }
2740
2741    let placeholders = (1..=entries.len())
2742        .map(|idx| format!("?{idx}"))
2743        .collect::<Vec<_>>()
2744        .join(", ");
2745    let sql = format!(
2746        "SELECT e.episode_id, e.document_id, e.search_text, e.effect_type, e.outcome, e.updated_at, d.namespace, e.embedding
2747         FROM episodes e
2748         JOIN documents d ON d.id = e.document_id
2749         WHERE e.episode_id IN ({placeholders})"
2750    );
2751    let params: Vec<SqlValue> = entries
2752        .keys()
2753        .map(|id| SqlValue::Text(id.clone()))
2754        .collect();
2755    let mut stmt = conn.prepare(&sql)?;
2756    let rows = stmt.query_map(rusqlite::params_from_iter(&params), |row| {
2757        Ok((
2758            row.get::<_, String>(0)?,
2759            row.get::<_, String>(1)?,
2760            row.get::<_, String>(2)?,
2761            row.get::<_, String>(3)?,
2762            row.get::<_, String>(4)?,
2763            row.get::<_, Option<String>>(5)?,
2764            row.get::<_, String>(6)?,
2765            row.get::<_, Option<Vec<u8>>>(7)?,
2766        ))
2767    })?;
2768
2769    for row in rows {
2770        let (
2771            episode_id,
2772            document_id,
2773            content,
2774            effect_type,
2775            outcome,
2776            updated_at,
2777            namespace,
2778            embedding_blob,
2779        ) = row?;
2780        if let Some(filter) = namespaces {
2781            if !filter.contains(&namespace.as_str()) {
2782                continue;
2783            }
2784        }
2785        if let Some(seed) = entries.get(&episode_id) {
2786            if let Some(hit) = build_ranked_vector_hit(
2787                episodes::episode_item_key(&episode_id),
2788                content,
2789                SearchSource::Episode {
2790                    episode_id,
2791                    document_id,
2792                    effect_type,
2793                    outcome,
2794                },
2795                updated_at,
2796                embedding_blob,
2797                seed,
2798                query_embedding,
2799                config,
2800            )? {
2801                output.push(hit);
2802            }
2803        }
2804    }
2805
2806    Ok(())
2807}
2808
2809/// Perform a hybrid search using pre-computed HNSW hits for the vector component.
2810#[cfg(feature = "hnsw")]
2811#[allow(clippy::too_many_arguments)]
2812pub fn hybrid_search_with_hnsw(
2813    conn: &Connection,
2814    query: &str,
2815    query_embedding: &[f32],
2816    config: &SearchConfig,
2817    top_k: usize,
2818    namespaces: Option<&[&str]>,
2819    source_types: Option<&[SearchSourceType]>,
2820    session_ids: Option<&[&str]>,
2821    hnsw_hits: &[crate::hnsw::HnswHit],
2822) -> Result<Vec<SearchResult>, MemoryError> {
2823    Ok(hybrid_search_with_hnsw_detailed(
2824        conn,
2825        query,
2826        query_embedding,
2827        config,
2828        top_k,
2829        namespaces,
2830        source_types,
2831        session_ids,
2832        hnsw_hits,
2833    )?
2834    .into_iter()
2835    .map(|result| result.result)
2836    .collect())
2837}
2838
2839#[cfg(feature = "hnsw")]
2840#[allow(clippy::too_many_arguments)]
2841pub(crate) fn hybrid_search_with_hnsw_detailed_with_context(
2842    conn: &Connection,
2843    query: &str,
2844    query_embedding: &[f32],
2845    config: &SearchConfig,
2846    context: &SearchContext,
2847    top_k: usize,
2848    namespaces: Option<&[&str]>,
2849    source_types: Option<&[SearchSourceType]>,
2850    session_ids: Option<&[&str]>,
2851    hnsw_hits: &[crate::hnsw::HnswHit],
2852) -> Result<SearchExecution, MemoryError> {
2853    let bm25_hits = match sanitize_fts_query(query) {
2854        Some(sanitized) => bm25_search(
2855            conn,
2856            &sanitized,
2857            config.candidate_pool_size,
2858            namespaces,
2859            source_types,
2860            session_ids,
2861        )?,
2862        None => Vec::new(),
2863    };
2864
2865    let mut vector_hits = resolve_hnsw_hits_batched(
2866        conn,
2867        query_embedding,
2868        config,
2869        namespaces,
2870        source_types,
2871        session_ids,
2872        hnsw_hits,
2873    )?;
2874    let mut fallback = None;
2875    let mut degradations = Vec::new();
2876    let mut backend = "hnsw";
2877    let mut exact_rerank = config.rerank_from_f32;
2878
2879    if !hnsw_hits.is_empty()
2880        && vector_hits.len() < top_k
2881        && filters_are_active(namespaces, source_types, session_ids)
2882    {
2883        fallback = Some("hnsw_filtered_underreturn_fallback".to_string());
2884        degradations.push(format!(
2885            "HNSW returned {} post-filter vector candidates for requested top_k {}; exact filtered fallback was used",
2886            vector_hits.len(),
2887            top_k
2888        ));
2889        vector_hits = vector_search(
2890            conn,
2891            query_embedding,
2892            config.candidate_pool_size,
2893            config.min_similarity,
2894            namespaces,
2895            source_types,
2896            session_ids,
2897        )?;
2898        backend = "hnsw_then_brute_force_f32";
2899        exact_rerank = true;
2900    }
2901
2902    let results = rrf_fuse_detailed_with_context(&bm25_hits, &vector_hits, config, context, top_k);
2903    let receipt = build_receipt(
2904        context,
2905        query_embedding,
2906        "hybrid",
2907        backend,
2908        config.candidate_pool_size,
2909        hnsw_hits.len(),
2910        vector_hits.len(),
2911        fallback,
2912        exact_rerank,
2913        &results,
2914        degradations,
2915    );
2916
2917    Ok(SearchExecution { results, receipt })
2918}
2919
2920#[cfg(feature = "hnsw")]
2921#[allow(clippy::too_many_arguments)]
2922pub(crate) fn hybrid_search_with_hnsw_detailed(
2923    conn: &Connection,
2924    query: &str,
2925    query_embedding: &[f32],
2926    config: &SearchConfig,
2927    top_k: usize,
2928    namespaces: Option<&[&str]>,
2929    source_types: Option<&[SearchSourceType]>,
2930    session_ids: Option<&[&str]>,
2931    hnsw_hits: &[crate::hnsw::HnswHit],
2932) -> Result<Vec<ExplainedResult>, MemoryError> {
2933    let context = SearchContext::default_now();
2934    Ok(hybrid_search_with_hnsw_detailed_with_context(
2935        conn,
2936        query,
2937        query_embedding,
2938        config,
2939        &context,
2940        top_k,
2941        namespaces,
2942        source_types,
2943        session_ids,
2944        hnsw_hits,
2945    )?
2946    .results)
2947}
2948
2949/// Perform a hybrid HNSW-backed search and return the exact score decomposition.
2950#[cfg(feature = "hnsw")]
2951#[allow(clippy::too_many_arguments)]
2952pub fn hybrid_search_explained_with_hnsw(
2953    conn: &Connection,
2954    query: &str,
2955    query_embedding: &[f32],
2956    config: &SearchConfig,
2957    top_k: usize,
2958    namespaces: Option<&[&str]>,
2959    source_types: Option<&[SearchSourceType]>,
2960    session_ids: Option<&[&str]>,
2961    hnsw_hits: &[crate::hnsw::HnswHit],
2962) -> Result<Vec<ExplainedResult>, MemoryError> {
2963    hybrid_search_with_hnsw_detailed(
2964        conn,
2965        query,
2966        query_embedding,
2967        config,
2968        top_k,
2969        namespaces,
2970        source_types,
2971        session_ids,
2972        hnsw_hits,
2973    )
2974}
2975
2976pub(crate) fn fts_only_search_detailed(
2977    conn: &Connection,
2978    query: &str,
2979    config: &SearchConfig,
2980    top_k: usize,
2981    namespaces: Option<&[&str]>,
2982    source_types: Option<&[SearchSourceType]>,
2983    session_ids: Option<&[&str]>,
2984) -> Result<Vec<ExplainedResult>, MemoryError> {
2985    let sanitized = match sanitize_fts_query(query) {
2986        Some(value) => value,
2987        None => return Ok(Vec::new()),
2988    };
2989    let bm25_hits = bm25_search(
2990        conn,
2991        &sanitized,
2992        top_k,
2993        namespaces,
2994        source_types,
2995        session_ids,
2996    )?;
2997    Ok(rrf_fuse_detailed(&bm25_hits, &[], config, top_k))
2998}
2999
3000/// Full-text search only (no embeddings needed). Synchronous.
3001pub fn fts_only_search(
3002    conn: &Connection,
3003    query: &str,
3004    config: &SearchConfig,
3005    top_k: usize,
3006    namespaces: Option<&[&str]>,
3007    source_types: Option<&[SearchSourceType]>,
3008    session_ids: Option<&[&str]>,
3009) -> Result<Vec<SearchResult>, MemoryError> {
3010    Ok(fts_only_search_detailed(
3011        conn,
3012        query,
3013        config,
3014        top_k,
3015        namespaces,
3016        source_types,
3017        session_ids,
3018    )?
3019    .into_iter()
3020    .map(|result| result.result)
3021    .collect())
3022}
3023
3024#[allow(clippy::too_many_arguments)]
3025pub(crate) fn vector_only_search_detailed_with_context(
3026    conn: &Connection,
3027    query_embedding: &[f32],
3028    config: &SearchConfig,
3029    context: &SearchContext,
3030    top_k: usize,
3031    namespaces: Option<&[&str]>,
3032    source_types: Option<&[SearchSourceType]>,
3033    session_ids: Option<&[&str]>,
3034) -> Result<SearchExecution, MemoryError> {
3035    let vector_outcome = vector_search_with_backend(
3036        conn,
3037        query_embedding,
3038        top_k,
3039        config.min_similarity,
3040        config,
3041        context,
3042        namespaces,
3043        source_types,
3044        session_ids,
3045    )?;
3046    let results = rrf_fuse_detailed_with_context(&[], &vector_outcome.hits, config, context, top_k);
3047    let receipt = build_receipt_with_metadata(
3048        context,
3049        query_embedding,
3050        "vector_only",
3051        &vector_outcome.candidate_backend,
3052        vector_outcome.requested_candidates,
3053        vector_outcome.returned_candidates,
3054        vector_outcome.post_filter_candidates,
3055        vector_outcome.fallback,
3056        vector_outcome.exact_rerank,
3057        &results,
3058        vector_outcome.degradations,
3059        vector_outcome.receipt_metadata,
3060    );
3061    Ok(SearchExecution { results, receipt })
3062}
3063
3064pub(crate) fn vector_only_search_detailed(
3065    conn: &Connection,
3066    query_embedding: &[f32],
3067    config: &SearchConfig,
3068    top_k: usize,
3069    namespaces: Option<&[&str]>,
3070    source_types: Option<&[SearchSourceType]>,
3071    session_ids: Option<&[&str]>,
3072) -> Result<Vec<ExplainedResult>, MemoryError> {
3073    let context = SearchContext::default_now();
3074    Ok(vector_only_search_detailed_with_context(
3075        conn,
3076        query_embedding,
3077        config,
3078        &context,
3079        top_k,
3080        namespaces,
3081        source_types,
3082        session_ids,
3083    )?
3084    .results)
3085}
3086
3087/// Vector-only search. Called after embedding the query.
3088pub fn vector_only_search(
3089    conn: &Connection,
3090    query_embedding: &[f32],
3091    config: &SearchConfig,
3092    top_k: usize,
3093    namespaces: Option<&[&str]>,
3094    source_types: Option<&[SearchSourceType]>,
3095    session_ids: Option<&[&str]>,
3096) -> Result<Vec<SearchResult>, MemoryError> {
3097    Ok(vector_only_search_detailed(
3098        conn,
3099        query_embedding,
3100        config,
3101        top_k,
3102        namespaces,
3103        source_types,
3104        session_ids,
3105    )?
3106    .into_iter()
3107    .map(|result| result.result)
3108    .collect())
3109}
3110
3111#[cfg(test)]
3112mod digest_tests {
3113    use super::query_embedding_digest;
3114
3115    #[test]
3116    fn query_embedding_digest_includes_dimension_and_bytes() {
3117        let two_dims = query_embedding_digest(&[1.0, 2.0]);
3118        let three_dims = query_embedding_digest(&[1.0, 2.0, 0.0]);
3119        let changed_byte = query_embedding_digest(&[1.0, 2.000_001]);
3120
3121        assert!(two_dims.starts_with("blake3:"));
3122        assert_eq!(two_dims.len(), 71);
3123        assert_ne!(two_dims, three_dims);
3124        assert_ne!(two_dims, changed_byte);
3125        assert_eq!(two_dims, query_embedding_digest(&[1.0, 2.0]));
3126    }
3127}
3128
3129/// Vector-only search using pre-computed HNSW hits.
3130#[cfg(feature = "hnsw")]
3131#[allow(clippy::too_many_arguments)]
3132pub fn vector_only_search_with_hnsw(
3133    conn: &Connection,
3134    query_embedding: &[f32],
3135    config: &SearchConfig,
3136    top_k: usize,
3137    namespaces: Option<&[&str]>,
3138    source_types: Option<&[SearchSourceType]>,
3139    session_ids: Option<&[&str]>,
3140    hnsw_hits: &[crate::hnsw::HnswHit],
3141) -> Result<Vec<SearchResult>, MemoryError> {
3142    Ok(vector_only_search_with_hnsw_detailed(
3143        conn,
3144        query_embedding,
3145        config,
3146        top_k,
3147        namespaces,
3148        source_types,
3149        session_ids,
3150        hnsw_hits,
3151    )?
3152    .into_iter()
3153    .map(|result| result.result)
3154    .collect())
3155}
3156
3157#[cfg(feature = "hnsw")]
3158#[allow(clippy::too_many_arguments)]
3159pub(crate) fn vector_only_search_with_hnsw_detailed_with_context(
3160    conn: &Connection,
3161    query_embedding: &[f32],
3162    config: &SearchConfig,
3163    context: &SearchContext,
3164    top_k: usize,
3165    namespaces: Option<&[&str]>,
3166    source_types: Option<&[SearchSourceType]>,
3167    session_ids: Option<&[&str]>,
3168    hnsw_hits: &[crate::hnsw::HnswHit],
3169) -> Result<SearchExecution, MemoryError> {
3170    let mut vector_hits = resolve_hnsw_hits_batched(
3171        conn,
3172        query_embedding,
3173        config,
3174        namespaces,
3175        source_types,
3176        session_ids,
3177        hnsw_hits,
3178    )?;
3179    let mut fallback = None;
3180    let mut degradations = Vec::new();
3181    let mut backend = "hnsw";
3182    let mut exact_rerank = config.rerank_from_f32;
3183
3184    if !hnsw_hits.is_empty()
3185        && vector_hits.len() < top_k
3186        && filters_are_active(namespaces, source_types, session_ids)
3187    {
3188        fallback = Some("hnsw_filtered_underreturn_fallback".to_string());
3189        degradations.push(format!(
3190            "HNSW returned {} post-filter vector candidates for requested top_k {}; exact filtered fallback was used",
3191            vector_hits.len(),
3192            top_k
3193        ));
3194        vector_hits = vector_search(
3195            conn,
3196            query_embedding,
3197            top_k,
3198            config.min_similarity,
3199            namespaces,
3200            source_types,
3201            session_ids,
3202        )?;
3203        backend = "hnsw_then_brute_force_f32";
3204        exact_rerank = true;
3205    }
3206
3207    let results = rrf_fuse_detailed_with_context(&[], &vector_hits, config, context, top_k);
3208    let receipt = build_receipt(
3209        context,
3210        query_embedding,
3211        "vector_only",
3212        backend,
3213        top_k,
3214        hnsw_hits.len(),
3215        vector_hits.len(),
3216        fallback,
3217        exact_rerank,
3218        &results,
3219        degradations,
3220    );
3221    Ok(SearchExecution { results, receipt })
3222}
3223
3224#[cfg(feature = "hnsw")]
3225#[allow(clippy::too_many_arguments)]
3226pub(crate) fn vector_only_search_with_hnsw_detailed(
3227    conn: &Connection,
3228    query_embedding: &[f32],
3229    config: &SearchConfig,
3230    top_k: usize,
3231    namespaces: Option<&[&str]>,
3232    source_types: Option<&[SearchSourceType]>,
3233    session_ids: Option<&[&str]>,
3234    hnsw_hits: &[crate::hnsw::HnswHit],
3235) -> Result<Vec<ExplainedResult>, MemoryError> {
3236    let context = SearchContext::default_now();
3237    Ok(vector_only_search_with_hnsw_detailed_with_context(
3238        conn,
3239        query_embedding,
3240        config,
3241        &context,
3242        top_k,
3243        namespaces,
3244        source_types,
3245        session_ids,
3246        hnsw_hits,
3247    )?
3248    .results)
3249}
3250
3251fn build_filter_clause(
3252    column: &str,
3253    values: Option<&[&str]>,
3254    param_offset: usize,
3255) -> (String, Vec<SqlValue>) {
3256    match values {
3257        Some(values) if !values.is_empty() => {
3258            let placeholders = (0..values.len())
3259                .map(|idx| format!("?{}", param_offset + idx))
3260                .collect::<Vec<_>>();
3261            let clause = format!(" AND {} IN ({})", column, placeholders.join(", "));
3262            let params = values
3263                .iter()
3264                .map(|value| SqlValue::Text((*value).to_string()))
3265                .collect();
3266            (clause, params)
3267        }
3268        _ => (String::new(), Vec::new()),
3269    }
3270}
3271
3272/// Deduplicate results by (source_type, source_id), keeping the first occurrence.
3273pub fn deduplicate_results(results: Vec<SearchResult>) -> Vec<SearchResult> {
3274    let mut seen = HashSet::new();
3275    results
3276        .into_iter()
3277        .filter(|result| seen.insert(source_dedup_key(&result.source)))
3278        .collect()
3279}
3280
3281#[cfg(test)]
3282mod tests {
3283    use super::*;
3284
3285    fn vector_row(id: &str) -> VectorRow {
3286        VectorRow {
3287            id: id.to_string(),
3288            content: format!("content {id}"),
3289            blob: bytemuck::cast_slice(&[1.0_f32, 0.0]).to_vec(),
3290            updated_at: None,
3291            source_type: SearchSourceType::Facts,
3292            filter_namespace: Some("default".to_string()),
3293            filter_session_id: None,
3294            source: SearchSource::Fact {
3295                fact_id: id.to_string(),
3296                namespace: "default".to_string(),
3297            },
3298        }
3299    }
3300
3301    #[test]
3302    fn timestamp_parser_accepts_sql_fractional_and_rfc3339_and_warns_by_returning_none() {
3303        assert!(parse_search_timestamp("2026-05-07 12:34:56").is_some());
3304        assert!(parse_search_timestamp("2026-05-07 12:34:56.123").is_some());
3305        assert!(parse_search_timestamp("2026-05-07T12:34:56Z").is_some());
3306        assert!(parse_search_timestamp("not-a-timestamp").is_none());
3307    }
3308
3309    #[test]
3310    fn vector_scan_hard_limit_blocks_before_unbounded_scan() {
3311        let old_warn = VECTOR_SCAN_WARN_LIMIT.swap(1, Ordering::SeqCst);
3312        let old_hard = VECTOR_SCAN_BLOCK_LIMIT.swap(2, Ordering::SeqCst);
3313        let rows = ["a", "b", "c"].into_iter().map(|id| Ok(vector_row(id)));
3314        let result = scan_vector_rows(rows, &[1.0, 0.0], -1.0, "fact");
3315        VECTOR_SCAN_WARN_LIMIT.store(old_warn, Ordering::SeqCst);
3316        VECTOR_SCAN_BLOCK_LIMIT.store(old_hard, Ordering::SeqCst);
3317
3318        match result {
3319            Err(MemoryError::VectorScanLimitExceeded {
3320                table,
3321                scanned,
3322                limit,
3323            }) => {
3324                assert_eq!(table, "fact");
3325                assert_eq!(scanned, 3);
3326                assert_eq!(limit, 2);
3327            }
3328            other => panic!("expected vector scan limit error, got {other:?}"),
3329        }
3330    }
3331}