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    #[allow(unused_mut)]
2123    let mut vector_outcome = vector_search_with_backend(
2124        conn,
2125        query_embedding,
2126        config.candidate_pool_size,
2127        config.min_similarity,
2128        config,
2129        context,
2130        namespaces,
2131        source_types,
2132        session_ids,
2133    )?;
2134
2135    // Task 3: Matryoshka 2-stage search — truncate query embedding to candidate_dims
2136    // for coarse retrieval, then rerank with full embedding. Falls back to direct
2137    // search if the 64d index doesn't exist or matryoshka feature is off.
2138    #[cfg(feature = "matryoshka")]
2139    {
2140        if let Some(candidate_dim) = config.candidate_dims {
2141            if candidate_dim > 0 && candidate_dim < query_embedding.len()
2142                && context.exactness_profile != crate::types::ExactnessProfile::PreferExact
2143            {
2144                use crate::matryoshka::truncate_embedding;
2145                let truncated_query = truncate_embedding(query_embedding, candidate_dim);
2146                match vector_search_with_backend(
2147                    conn,
2148                    &truncated_query,
2149                    config.candidate_pool_size.saturating_mul(2),
2150                    config.min_similarity * 0.5,
2151                    config,
2152                    context,
2153                    namespaces,
2154                    source_types,
2155                    session_ids,
2156                ) {
2157                    Ok(coarse_outcome) => {
2158                        // Rerank coarse candidates with full-dimension embedding.
2159                        let reranked_hits: Vec<VectorHit> = coarse_outcome
2160                            .hits
2161                            .into_iter()
2162                            .map(|mut hit| {
2163                                if let Ok(full_sim) =
2164                                    rerank_hit_with_full_embedding(conn, query_embedding, &hit)
2165                                {
2166                                    hit.similarity = full_sim;
2167                                    hit.reranked_from_f32 = true;
2168                                }
2169                                hit
2170                            })
2171                            .filter(|hit| hit.similarity >= config.min_similarity)
2172                            .collect();
2173                        let mut reranked = reranked_hits;
2174                        reranked.sort_by(|a, b| {
2175                            b.similarity
2176                                .partial_cmp(&a.similarity)
2177                                .unwrap_or(std::cmp::Ordering::Equal)
2178                        });
2179                        reranked.truncate(config.candidate_pool_size);
2180                        let new_receipt_metadata = coarse_outcome.receipt_metadata.clone();
2181                        vector_outcome = VectorSearchOutcome {
2182                            hits: reranked,
2183                            candidate_backend: format!(
2184                                "matryoshka_2stage_{}d_to_{}d",
2185                                candidate_dim,
2186                                query_embedding.len()
2187                            ),
2188                            receipt_metadata: new_receipt_metadata,
2189                            ..coarse_outcome
2190                        };
2191                    }
2192                    Err(_) => { /* keep original vector_outcome */ }
2193                }
2194            }
2195        }
2196    }
2197
2198    let results = if config.late_interaction_weight > 0.0 {
2199        // Late interaction 3rd RRF signal: compute proxy MaxSim scores by
2200        // splitting the query embedding into segments and comparing against
2201        // document embeddings. This is an approximation of ColBERT late
2202        // interaction using existing dense embeddings.
2203        let li_scores = compute_proxy_late_interaction_scores(
2204            query_embedding,
2205            &vector_outcome.hits,
2206        );
2207        #[cfg(feature = "late-interaction")]
2208        {
2209            rrf_fuse_with_late_interaction(
2210                &bm25_hits,
2211                &vector_outcome.hits,
2212                &li_scores,
2213                config,
2214                context,
2215                top_k,
2216            )
2217        }
2218        #[cfg(not(feature = "late-interaction"))]
2219        {
2220            let _ = li_scores;
2221            rrf_fuse_detailed_with_context(&bm25_hits, &vector_outcome.hits, config, context, top_k)
2222        }
2223    } else {
2224        rrf_fuse_detailed_with_context(&bm25_hits, &vector_outcome.hits, config, context, top_k)
2225    };
2226    let receipt = build_receipt_with_metadata(
2227        context,
2228        query_embedding,
2229        "hybrid",
2230        &vector_outcome.candidate_backend,
2231        vector_outcome.requested_candidates,
2232        vector_outcome.returned_candidates,
2233        vector_outcome.post_filter_candidates,
2234        vector_outcome.fallback,
2235        vector_outcome.exact_rerank,
2236        &results,
2237        vector_outcome.degradations,
2238        vector_outcome.receipt_metadata,
2239    );
2240    Ok(SearchExecution { results, receipt })
2241}
2242
2243#[allow(clippy::too_many_arguments)]
2244pub(crate) fn hybrid_search_detailed(
2245    conn: &Connection,
2246    query: &str,
2247    query_embedding: &[f32],
2248    config: &SearchConfig,
2249    top_k: usize,
2250    namespaces: Option<&[&str]>,
2251    source_types: Option<&[SearchSourceType]>,
2252    session_ids: Option<&[&str]>,
2253) -> Result<Vec<ExplainedResult>, MemoryError> {
2254    let context = SearchContext::default_now();
2255    Ok(hybrid_search_detailed_with_context(
2256        conn,
2257        query,
2258        query_embedding,
2259        config,
2260        &context,
2261        top_k,
2262        namespaces,
2263        source_types,
2264        session_ids,
2265    )?
2266    .results)
2267}
2268
2269/// Perform a hybrid search and return the exact score decomposition.
2270#[allow(clippy::too_many_arguments)]
2271pub fn hybrid_search_explained(
2272    conn: &Connection,
2273    query: &str,
2274    query_embedding: &[f32],
2275    config: &SearchConfig,
2276    top_k: usize,
2277    namespaces: Option<&[&str]>,
2278    source_types: Option<&[SearchSourceType]>,
2279    session_ids: Option<&[&str]>,
2280) -> Result<Vec<ExplainedResult>, MemoryError> {
2281    hybrid_search_detailed(
2282        conn,
2283        query,
2284        query_embedding,
2285        config,
2286        top_k,
2287        namespaces,
2288        source_types,
2289        session_ids,
2290    )
2291}
2292
2293/// Perform a hybrid search (BM25 + vector + RRF).
2294#[allow(clippy::too_many_arguments)]
2295pub fn hybrid_search(
2296    conn: &Connection,
2297    query: &str,
2298    query_embedding: &[f32],
2299    config: &SearchConfig,
2300    top_k: usize,
2301    namespaces: Option<&[&str]>,
2302    source_types: Option<&[SearchSourceType]>,
2303    session_ids: Option<&[&str]>,
2304) -> Result<Vec<SearchResult>, MemoryError> {
2305    let results: Vec<SearchResult> = hybrid_search_detailed(
2306        conn,
2307        query,
2308        query_embedding,
2309        config,
2310        top_k,
2311        namespaces,
2312        source_types,
2313        session_ids,
2314    )?
2315    .into_iter()
2316    .map(|result| result.result)
2317    .collect();
2318
2319    // Content dedup: remove results with identical or near-identical content,
2320    // keeping the highest-scoring one. This prevents duplicate chunks from
2321    // different document copies appearing in search results.
2322    let mut seen_content: std::collections::HashSet<String> = std::collections::HashSet::new();
2323    let deduped: Vec<SearchResult> = results
2324        .into_iter()
2325        .filter(|r| {
2326            // Normalize whitespace and use first 200 chars as fingerprint.
2327            // This catches near-duplicates with minor whitespace differences.
2328            let fingerprint: String = r
2329                .content
2330                .split_whitespace()
2331                .take(30)
2332                .collect::<Vec<_>>()
2333                .join(" ")
2334                .to_lowercase();
2335            seen_content.insert(fingerprint)
2336        })
2337        .collect();
2338
2339    Ok(deduped)
2340}
2341
2342#[cfg(feature = "hnsw")]
2343#[derive(Clone)]
2344struct HnswCandidateSeed {
2345    source_rank: usize,
2346    source_similarity: f64,
2347}
2348
2349#[cfg(feature = "hnsw")]
2350#[allow(clippy::type_complexity)]
2351fn resolve_hnsw_hits_batched(
2352    conn: &Connection,
2353    query_embedding: &[f32],
2354    config: &SearchConfig,
2355    namespaces: Option<&[&str]>,
2356    source_types: Option<&[SearchSourceType]>,
2357    session_ids: Option<&[&str]>,
2358    hnsw_hits: &[crate::hnsw::HnswHit],
2359) -> Result<Vec<VectorHit>, MemoryError> {
2360    let search_facts = source_types
2361        .map(|st| st.contains(&SearchSourceType::Facts))
2362        .unwrap_or(true);
2363    let search_chunks = source_types
2364        .map(|st| st.contains(&SearchSourceType::Chunks))
2365        .unwrap_or(true);
2366    let search_messages = source_types
2367        .map(|st| st.contains(&SearchSourceType::Messages))
2368        .unwrap_or(false);
2369    let search_episodes = source_types
2370        .map(|st| st.contains(&SearchSourceType::Episodes))
2371        .unwrap_or(true);
2372
2373    // CONVENTION EXCEPTION: O(1) lookup required for performance-critical search path
2374    let mut fact_entries: HashMap<String, HnswCandidateSeed> = HashMap::new();
2375    // CONVENTION EXCEPTION: O(1) lookup required for performance-critical search path
2376    let mut chunk_entries: HashMap<String, HnswCandidateSeed> = HashMap::new();
2377    // CONVENTION EXCEPTION: O(1) lookup required for performance-critical search path
2378    let mut message_entries: HashMap<i64, HnswCandidateSeed> = HashMap::new();
2379    // CONVENTION EXCEPTION: O(1) lookup required for performance-critical search path
2380    let mut episode_entries: HashMap<String, HnswCandidateSeed> = HashMap::new();
2381
2382    for (rank_0, hit) in hnsw_hits.iter().enumerate() {
2383        let similarity = hit.similarity() as f64;
2384        if similarity < config.min_similarity {
2385            continue;
2386        }
2387
2388        let (domain, raw_id) = hit.parse_key()?;
2389        let seed = HnswCandidateSeed {
2390            source_rank: rank_0 + 1,
2391            source_similarity: similarity,
2392        };
2393
2394        match domain {
2395            "fact" if search_facts => {
2396                fact_entries.entry(raw_id.to_string()).or_insert(seed);
2397            }
2398            "chunk" if search_chunks => {
2399                chunk_entries.entry(raw_id.to_string()).or_insert(seed);
2400            }
2401            "msg" if search_messages => {
2402                if let Ok(message_id) = raw_id.parse::<i64>() {
2403                    message_entries.entry(message_id).or_insert(seed);
2404                }
2405            }
2406            "episode" if search_episodes => {
2407                episode_entries.entry(raw_id.to_string()).or_insert(seed);
2408            }
2409            _ => {}
2410        }
2411    }
2412
2413    let mut hits = Vec::new();
2414    batch_load_fact_hits(
2415        conn,
2416        query_embedding,
2417        config,
2418        namespaces,
2419        &fact_entries,
2420        &mut hits,
2421    )?;
2422    batch_load_chunk_hits(
2423        conn,
2424        query_embedding,
2425        config,
2426        namespaces,
2427        &chunk_entries,
2428        &mut hits,
2429    )?;
2430    batch_load_message_hits(
2431        conn,
2432        query_embedding,
2433        config,
2434        session_ids,
2435        &message_entries,
2436        &mut hits,
2437    )?;
2438    batch_load_episode_hits(
2439        conn,
2440        query_embedding,
2441        config,
2442        namespaces,
2443        &episode_entries,
2444        &mut hits,
2445    )?;
2446
2447    hits.sort_by(|a, b| {
2448        b.similarity
2449            .partial_cmp(&a.similarity)
2450            .unwrap_or(std::cmp::Ordering::Equal)
2451            .then_with(|| {
2452                a.source_rank
2453                    .unwrap_or(usize::MAX)
2454                    .cmp(&b.source_rank.unwrap_or(usize::MAX))
2455            })
2456    });
2457    hits.truncate(config.candidate_pool_size);
2458    Ok(hits)
2459}
2460
2461#[cfg(feature = "hnsw")]
2462fn exact_similarity_from_blob(
2463    query_embedding: &[f32],
2464    blob: &[u8],
2465) -> Result<Option<f64>, MemoryError> {
2466    if blob.is_empty() {
2467        return Ok(None);
2468    }
2469    let stored = crate::db::bytes_to_embedding(blob)?;
2470    if stored.len() != query_embedding.len() {
2471        return Ok(None);
2472    }
2473    Ok(Some(cosine_similarity(query_embedding, &stored)? as f64))
2474}
2475
2476#[cfg(feature = "hnsw")]
2477#[allow(clippy::too_many_arguments)]
2478fn build_ranked_vector_hit(
2479    id: String,
2480    content: String,
2481    source: SearchSource,
2482    updated_at: Option<String>,
2483    embedding_blob: Option<Vec<u8>>,
2484    seed: &HnswCandidateSeed,
2485    query_embedding: &[f32],
2486    config: &SearchConfig,
2487) -> Result<Option<VectorHit>, MemoryError> {
2488    let similarity = if config.rerank_from_f32 {
2489        match embedding_blob {
2490            Some(blob) => exact_similarity_from_blob(query_embedding, &blob)?,
2491            None => None,
2492        }
2493        .unwrap_or(seed.source_similarity)
2494    } else {
2495        seed.source_similarity
2496    };
2497
2498    if similarity < config.min_similarity {
2499        return Ok(None);
2500    }
2501
2502    Ok(Some(VectorHit {
2503        id,
2504        content,
2505        source,
2506        similarity,
2507        updated_at,
2508        source_rank: Some(seed.source_rank),
2509        source_similarity: Some(seed.source_similarity),
2510        reranked_from_f32: config.rerank_from_f32,
2511                temporal_weight: None,
2512    provenance_confidence: None,
2513    }))
2514}
2515
2516#[cfg(feature = "hnsw")]
2517fn batch_load_fact_hits(
2518    conn: &Connection,
2519    query_embedding: &[f32],
2520    config: &SearchConfig,
2521    namespaces: Option<&[&str]>,
2522    // CONVENTION EXCEPTION: O(1) lookup required for performance-critical search path
2523    entries: &HashMap<String, HnswCandidateSeed>,
2524    output: &mut Vec<VectorHit>,
2525) -> Result<(), MemoryError> {
2526    if entries.is_empty() {
2527        return Ok(());
2528    }
2529
2530    let placeholders = (1..=entries.len())
2531        .map(|idx| format!("?{idx}"))
2532        .collect::<Vec<_>>()
2533        .join(", ");
2534    let sql = format!(
2535        "SELECT id, content, namespace, updated_at, embedding
2536         FROM facts
2537         WHERE id IN ({placeholders})"
2538    );
2539    let params: Vec<SqlValue> = entries
2540        .keys()
2541        .map(|id| SqlValue::Text(id.clone()))
2542        .collect();
2543    let mut stmt = conn.prepare(&sql)?;
2544    let rows = stmt.query_map(rusqlite::params_from_iter(&params), |row| {
2545        Ok((
2546            row.get::<_, String>(0)?,
2547            row.get::<_, String>(1)?,
2548            row.get::<_, String>(2)?,
2549            row.get::<_, Option<String>>(3)?,
2550            row.get::<_, Option<Vec<u8>>>(4)?,
2551        ))
2552    })?;
2553
2554    for row in rows {
2555        let (fact_id, content, namespace, updated_at, embedding_blob) = row?;
2556        if let Some(filter) = namespaces {
2557            if !filter.contains(&namespace.as_str()) {
2558                continue;
2559            }
2560        }
2561        if let Some(seed) = entries.get(&fact_id) {
2562            if let Some(hit) = build_ranked_vector_hit(
2563                format!("fact:{fact_id}"),
2564                content,
2565                SearchSource::Fact { fact_id, namespace },
2566                updated_at,
2567                embedding_blob,
2568                seed,
2569                query_embedding,
2570                config,
2571            )? {
2572                output.push(hit);
2573            }
2574        }
2575    }
2576
2577    Ok(())
2578}
2579
2580#[cfg(feature = "hnsw")]
2581fn batch_load_chunk_hits(
2582    conn: &Connection,
2583    query_embedding: &[f32],
2584    config: &SearchConfig,
2585    namespaces: Option<&[&str]>,
2586    // CONVENTION EXCEPTION: O(1) lookup required for performance-critical search path
2587    entries: &HashMap<String, HnswCandidateSeed>,
2588    output: &mut Vec<VectorHit>,
2589) -> Result<(), MemoryError> {
2590    if entries.is_empty() {
2591        return Ok(());
2592    }
2593
2594    let placeholders = (1..=entries.len())
2595        .map(|idx| format!("?{idx}"))
2596        .collect::<Vec<_>>()
2597        .join(", ");
2598    let sql = format!(
2599        "SELECT c.id, c.content, c.document_id, d.title, c.chunk_index, c.created_at, d.namespace, c.embedding
2600         FROM chunks c
2601         JOIN documents d ON d.id = c.document_id
2602         WHERE c.id IN ({placeholders})"
2603    );
2604    let params: Vec<SqlValue> = entries
2605        .keys()
2606        .map(|id| SqlValue::Text(id.clone()))
2607        .collect();
2608    let mut stmt = conn.prepare(&sql)?;
2609    let rows = stmt.query_map(rusqlite::params_from_iter(&params), |row| {
2610        Ok((
2611            row.get::<_, String>(0)?,
2612            row.get::<_, String>(1)?,
2613            row.get::<_, String>(2)?,
2614            row.get::<_, String>(3)?,
2615            row.get::<_, i64>(4)?,
2616            row.get::<_, Option<String>>(5)?,
2617            row.get::<_, String>(6)?,
2618            row.get::<_, Option<Vec<u8>>>(7)?,
2619        ))
2620    })?;
2621
2622    for row in rows {
2623        let (
2624            chunk_id,
2625            content,
2626            document_id,
2627            document_title,
2628            chunk_index,
2629            updated_at,
2630            namespace,
2631            embedding_blob,
2632        ) = row?;
2633        if let Some(filter) = namespaces {
2634            if !filter.contains(&namespace.as_str()) {
2635                continue;
2636            }
2637        }
2638        if let Some(seed) = entries.get(&chunk_id) {
2639            if let Some(hit) = build_ranked_vector_hit(
2640                format!("chunk:{chunk_id}"),
2641                content,
2642                SearchSource::Chunk {
2643                    chunk_id,
2644                    document_id,
2645                    document_title,
2646                    chunk_index: chunk_index as usize,
2647                },
2648                updated_at,
2649                embedding_blob,
2650                seed,
2651                query_embedding,
2652                config,
2653            )? {
2654                output.push(hit);
2655            }
2656        }
2657    }
2658
2659    Ok(())
2660}
2661
2662#[cfg(feature = "hnsw")]
2663fn batch_load_message_hits(
2664    conn: &Connection,
2665    query_embedding: &[f32],
2666    config: &SearchConfig,
2667    session_ids: Option<&[&str]>,
2668    // CONVENTION EXCEPTION: O(1) lookup required for performance-critical search path
2669    entries: &HashMap<i64, HnswCandidateSeed>,
2670    output: &mut Vec<VectorHit>,
2671) -> Result<(), MemoryError> {
2672    if entries.is_empty() {
2673        return Ok(());
2674    }
2675
2676    let placeholders = (1..=entries.len())
2677        .map(|idx| format!("?{idx}"))
2678        .collect::<Vec<_>>()
2679        .join(", ");
2680    let sql = format!(
2681        "SELECT id, content, session_id, role, created_at, embedding
2682         FROM messages
2683         WHERE id IN ({placeholders})"
2684    );
2685    let params: Vec<SqlValue> = entries.keys().map(|id| SqlValue::Integer(*id)).collect();
2686    let mut stmt = conn.prepare(&sql)?;
2687    let rows = stmt.query_map(rusqlite::params_from_iter(&params), |row| {
2688        Ok((
2689            row.get::<_, i64>(0)?,
2690            row.get::<_, String>(1)?,
2691            row.get::<_, String>(2)?,
2692            row.get::<_, String>(3)?,
2693            row.get::<_, Option<String>>(4)?,
2694            row.get::<_, Option<Vec<u8>>>(5)?,
2695        ))
2696    })?;
2697
2698    for row in rows {
2699        let (message_id, content, session_id, role, updated_at, embedding_blob) = row?;
2700        if let Some(filter) = session_ids {
2701            if !filter.contains(&session_id.as_str()) {
2702                continue;
2703            }
2704        }
2705        if let Some(seed) = entries.get(&message_id) {
2706            if let Some(hit) = build_ranked_vector_hit(
2707                format!("msg:{message_id}"),
2708                content,
2709                SearchSource::Message {
2710                    message_id,
2711                    session_id,
2712                    role,
2713                },
2714                updated_at,
2715                embedding_blob,
2716                seed,
2717                query_embedding,
2718                config,
2719            )? {
2720                output.push(hit);
2721            }
2722        }
2723    }
2724
2725    Ok(())
2726}
2727
2728#[cfg(feature = "hnsw")]
2729fn batch_load_episode_hits(
2730    conn: &Connection,
2731    query_embedding: &[f32],
2732    config: &SearchConfig,
2733    namespaces: Option<&[&str]>,
2734    // CONVENTION EXCEPTION: O(1) lookup required for performance-critical search path
2735    entries: &HashMap<String, HnswCandidateSeed>,
2736    output: &mut Vec<VectorHit>,
2737) -> Result<(), MemoryError> {
2738    if entries.is_empty() {
2739        return Ok(());
2740    }
2741
2742    let placeholders = (1..=entries.len())
2743        .map(|idx| format!("?{idx}"))
2744        .collect::<Vec<_>>()
2745        .join(", ");
2746    let sql = format!(
2747        "SELECT e.episode_id, e.document_id, e.search_text, e.effect_type, e.outcome, e.updated_at, d.namespace, e.embedding
2748         FROM episodes e
2749         JOIN documents d ON d.id = e.document_id
2750         WHERE e.episode_id IN ({placeholders})"
2751    );
2752    let params: Vec<SqlValue> = entries
2753        .keys()
2754        .map(|id| SqlValue::Text(id.clone()))
2755        .collect();
2756    let mut stmt = conn.prepare(&sql)?;
2757    let rows = stmt.query_map(rusqlite::params_from_iter(&params), |row| {
2758        Ok((
2759            row.get::<_, String>(0)?,
2760            row.get::<_, String>(1)?,
2761            row.get::<_, String>(2)?,
2762            row.get::<_, String>(3)?,
2763            row.get::<_, String>(4)?,
2764            row.get::<_, Option<String>>(5)?,
2765            row.get::<_, String>(6)?,
2766            row.get::<_, Option<Vec<u8>>>(7)?,
2767        ))
2768    })?;
2769
2770    for row in rows {
2771        let (
2772            episode_id,
2773            document_id,
2774            content,
2775            effect_type,
2776            outcome,
2777            updated_at,
2778            namespace,
2779            embedding_blob,
2780        ) = row?;
2781        if let Some(filter) = namespaces {
2782            if !filter.contains(&namespace.as_str()) {
2783                continue;
2784            }
2785        }
2786        if let Some(seed) = entries.get(&episode_id) {
2787            if let Some(hit) = build_ranked_vector_hit(
2788                episodes::episode_item_key(&episode_id),
2789                content,
2790                SearchSource::Episode {
2791                    episode_id,
2792                    document_id,
2793                    effect_type,
2794                    outcome,
2795                },
2796                updated_at,
2797                embedding_blob,
2798                seed,
2799                query_embedding,
2800                config,
2801            )? {
2802                output.push(hit);
2803            }
2804        }
2805    }
2806
2807    Ok(())
2808}
2809
2810/// Perform a hybrid search using pre-computed HNSW hits for the vector component.
2811#[cfg(feature = "hnsw")]
2812#[allow(clippy::too_many_arguments)]
2813pub fn hybrid_search_with_hnsw(
2814    conn: &Connection,
2815    query: &str,
2816    query_embedding: &[f32],
2817    config: &SearchConfig,
2818    top_k: usize,
2819    namespaces: Option<&[&str]>,
2820    source_types: Option<&[SearchSourceType]>,
2821    session_ids: Option<&[&str]>,
2822    hnsw_hits: &[crate::hnsw::HnswHit],
2823) -> Result<Vec<SearchResult>, MemoryError> {
2824    Ok(hybrid_search_with_hnsw_detailed(
2825        conn,
2826        query,
2827        query_embedding,
2828        config,
2829        top_k,
2830        namespaces,
2831        source_types,
2832        session_ids,
2833        hnsw_hits,
2834    )?
2835    .into_iter()
2836    .map(|result| result.result)
2837    .collect())
2838}
2839
2840#[cfg(feature = "hnsw")]
2841#[allow(clippy::too_many_arguments)]
2842pub(crate) fn hybrid_search_with_hnsw_detailed_with_context(
2843    conn: &Connection,
2844    query: &str,
2845    query_embedding: &[f32],
2846    config: &SearchConfig,
2847    context: &SearchContext,
2848    top_k: usize,
2849    namespaces: Option<&[&str]>,
2850    source_types: Option<&[SearchSourceType]>,
2851    session_ids: Option<&[&str]>,
2852    hnsw_hits: &[crate::hnsw::HnswHit],
2853) -> Result<SearchExecution, MemoryError> {
2854    let bm25_hits = match sanitize_fts_query(query) {
2855        Some(sanitized) => bm25_search(
2856            conn,
2857            &sanitized,
2858            config.candidate_pool_size,
2859            namespaces,
2860            source_types,
2861            session_ids,
2862        )?,
2863        None => Vec::new(),
2864    };
2865
2866    let mut vector_hits = resolve_hnsw_hits_batched(
2867        conn,
2868        query_embedding,
2869        config,
2870        namespaces,
2871        source_types,
2872        session_ids,
2873        hnsw_hits,
2874    )?;
2875    let mut fallback = None;
2876    let mut degradations = Vec::new();
2877    let mut backend = "hnsw";
2878    let mut exact_rerank = config.rerank_from_f32;
2879
2880    if !hnsw_hits.is_empty()
2881        && vector_hits.len() < top_k
2882        && filters_are_active(namespaces, source_types, session_ids)
2883    {
2884        fallback = Some("hnsw_filtered_underreturn_fallback".to_string());
2885        degradations.push(format!(
2886            "HNSW returned {} post-filter vector candidates for requested top_k {}; exact filtered fallback was used",
2887            vector_hits.len(),
2888            top_k
2889        ));
2890        vector_hits = vector_search(
2891            conn,
2892            query_embedding,
2893            config.candidate_pool_size,
2894            config.min_similarity,
2895            namespaces,
2896            source_types,
2897            session_ids,
2898        )?;
2899        backend = "hnsw_then_brute_force_f32";
2900        exact_rerank = true;
2901    }
2902
2903    let results = rrf_fuse_detailed_with_context(&bm25_hits, &vector_hits, config, context, top_k);
2904    let receipt = build_receipt(
2905        context,
2906        query_embedding,
2907        "hybrid",
2908        backend,
2909        config.candidate_pool_size,
2910        hnsw_hits.len(),
2911        vector_hits.len(),
2912        fallback,
2913        exact_rerank,
2914        &results,
2915        degradations,
2916    );
2917
2918    Ok(SearchExecution { results, receipt })
2919}
2920
2921#[cfg(feature = "hnsw")]
2922#[allow(clippy::too_many_arguments)]
2923pub(crate) fn hybrid_search_with_hnsw_detailed(
2924    conn: &Connection,
2925    query: &str,
2926    query_embedding: &[f32],
2927    config: &SearchConfig,
2928    top_k: usize,
2929    namespaces: Option<&[&str]>,
2930    source_types: Option<&[SearchSourceType]>,
2931    session_ids: Option<&[&str]>,
2932    hnsw_hits: &[crate::hnsw::HnswHit],
2933) -> Result<Vec<ExplainedResult>, MemoryError> {
2934    let context = SearchContext::default_now();
2935    Ok(hybrid_search_with_hnsw_detailed_with_context(
2936        conn,
2937        query,
2938        query_embedding,
2939        config,
2940        &context,
2941        top_k,
2942        namespaces,
2943        source_types,
2944        session_ids,
2945        hnsw_hits,
2946    )?
2947    .results)
2948}
2949
2950/// Perform a hybrid HNSW-backed search and return the exact score decomposition.
2951#[cfg(feature = "hnsw")]
2952#[allow(clippy::too_many_arguments)]
2953pub fn hybrid_search_explained_with_hnsw(
2954    conn: &Connection,
2955    query: &str,
2956    query_embedding: &[f32],
2957    config: &SearchConfig,
2958    top_k: usize,
2959    namespaces: Option<&[&str]>,
2960    source_types: Option<&[SearchSourceType]>,
2961    session_ids: Option<&[&str]>,
2962    hnsw_hits: &[crate::hnsw::HnswHit],
2963) -> Result<Vec<ExplainedResult>, MemoryError> {
2964    hybrid_search_with_hnsw_detailed(
2965        conn,
2966        query,
2967        query_embedding,
2968        config,
2969        top_k,
2970        namespaces,
2971        source_types,
2972        session_ids,
2973        hnsw_hits,
2974    )
2975}
2976
2977pub(crate) fn fts_only_search_detailed(
2978    conn: &Connection,
2979    query: &str,
2980    config: &SearchConfig,
2981    top_k: usize,
2982    namespaces: Option<&[&str]>,
2983    source_types: Option<&[SearchSourceType]>,
2984    session_ids: Option<&[&str]>,
2985) -> Result<Vec<ExplainedResult>, MemoryError> {
2986    let sanitized = match sanitize_fts_query(query) {
2987        Some(value) => value,
2988        None => return Ok(Vec::new()),
2989    };
2990    let bm25_hits = bm25_search(
2991        conn,
2992        &sanitized,
2993        top_k,
2994        namespaces,
2995        source_types,
2996        session_ids,
2997    )?;
2998    Ok(rrf_fuse_detailed(&bm25_hits, &[], config, top_k))
2999}
3000
3001/// Full-text search only (no embeddings needed). Synchronous.
3002pub fn fts_only_search(
3003    conn: &Connection,
3004    query: &str,
3005    config: &SearchConfig,
3006    top_k: usize,
3007    namespaces: Option<&[&str]>,
3008    source_types: Option<&[SearchSourceType]>,
3009    session_ids: Option<&[&str]>,
3010) -> Result<Vec<SearchResult>, MemoryError> {
3011    Ok(fts_only_search_detailed(
3012        conn,
3013        query,
3014        config,
3015        top_k,
3016        namespaces,
3017        source_types,
3018        session_ids,
3019    )?
3020    .into_iter()
3021    .map(|result| result.result)
3022    .collect())
3023}
3024
3025#[allow(clippy::too_many_arguments)]
3026pub(crate) fn vector_only_search_detailed_with_context(
3027    conn: &Connection,
3028    query_embedding: &[f32],
3029    config: &SearchConfig,
3030    context: &SearchContext,
3031    top_k: usize,
3032    namespaces: Option<&[&str]>,
3033    source_types: Option<&[SearchSourceType]>,
3034    session_ids: Option<&[&str]>,
3035) -> Result<SearchExecution, MemoryError> {
3036    let vector_outcome = vector_search_with_backend(
3037        conn,
3038        query_embedding,
3039        top_k,
3040        config.min_similarity,
3041        config,
3042        context,
3043        namespaces,
3044        source_types,
3045        session_ids,
3046    )?;
3047    let results = rrf_fuse_detailed_with_context(&[], &vector_outcome.hits, config, context, top_k);
3048    let receipt = build_receipt_with_metadata(
3049        context,
3050        query_embedding,
3051        "vector_only",
3052        &vector_outcome.candidate_backend,
3053        vector_outcome.requested_candidates,
3054        vector_outcome.returned_candidates,
3055        vector_outcome.post_filter_candidates,
3056        vector_outcome.fallback,
3057        vector_outcome.exact_rerank,
3058        &results,
3059        vector_outcome.degradations,
3060        vector_outcome.receipt_metadata,
3061    );
3062    Ok(SearchExecution { results, receipt })
3063}
3064
3065pub(crate) fn vector_only_search_detailed(
3066    conn: &Connection,
3067    query_embedding: &[f32],
3068    config: &SearchConfig,
3069    top_k: usize,
3070    namespaces: Option<&[&str]>,
3071    source_types: Option<&[SearchSourceType]>,
3072    session_ids: Option<&[&str]>,
3073) -> Result<Vec<ExplainedResult>, MemoryError> {
3074    let context = SearchContext::default_now();
3075    Ok(vector_only_search_detailed_with_context(
3076        conn,
3077        query_embedding,
3078        config,
3079        &context,
3080        top_k,
3081        namespaces,
3082        source_types,
3083        session_ids,
3084    )?
3085    .results)
3086}
3087
3088/// Vector-only search. Called after embedding the query.
3089pub fn vector_only_search(
3090    conn: &Connection,
3091    query_embedding: &[f32],
3092    config: &SearchConfig,
3093    top_k: usize,
3094    namespaces: Option<&[&str]>,
3095    source_types: Option<&[SearchSourceType]>,
3096    session_ids: Option<&[&str]>,
3097) -> Result<Vec<SearchResult>, MemoryError> {
3098    Ok(vector_only_search_detailed(
3099        conn,
3100        query_embedding,
3101        config,
3102        top_k,
3103        namespaces,
3104        source_types,
3105        session_ids,
3106    )?
3107    .into_iter()
3108    .map(|result| result.result)
3109    .collect())
3110}
3111
3112#[cfg(test)]
3113mod digest_tests {
3114    use super::query_embedding_digest;
3115
3116    #[test]
3117    fn query_embedding_digest_includes_dimension_and_bytes() {
3118        let two_dims = query_embedding_digest(&[1.0, 2.0]);
3119        let three_dims = query_embedding_digest(&[1.0, 2.0, 0.0]);
3120        let changed_byte = query_embedding_digest(&[1.0, 2.000_001]);
3121
3122        assert!(two_dims.starts_with("blake3:"));
3123        assert_eq!(two_dims.len(), 71);
3124        assert_ne!(two_dims, three_dims);
3125        assert_ne!(two_dims, changed_byte);
3126        assert_eq!(two_dims, query_embedding_digest(&[1.0, 2.0]));
3127    }
3128}
3129
3130/// Vector-only search using pre-computed HNSW hits.
3131#[cfg(feature = "hnsw")]
3132#[allow(clippy::too_many_arguments)]
3133pub fn vector_only_search_with_hnsw(
3134    conn: &Connection,
3135    query_embedding: &[f32],
3136    config: &SearchConfig,
3137    top_k: usize,
3138    namespaces: Option<&[&str]>,
3139    source_types: Option<&[SearchSourceType]>,
3140    session_ids: Option<&[&str]>,
3141    hnsw_hits: &[crate::hnsw::HnswHit],
3142) -> Result<Vec<SearchResult>, MemoryError> {
3143    Ok(vector_only_search_with_hnsw_detailed(
3144        conn,
3145        query_embedding,
3146        config,
3147        top_k,
3148        namespaces,
3149        source_types,
3150        session_ids,
3151        hnsw_hits,
3152    )?
3153    .into_iter()
3154    .map(|result| result.result)
3155    .collect())
3156}
3157
3158#[cfg(feature = "hnsw")]
3159#[allow(clippy::too_many_arguments)]
3160pub(crate) fn vector_only_search_with_hnsw_detailed_with_context(
3161    conn: &Connection,
3162    query_embedding: &[f32],
3163    config: &SearchConfig,
3164    context: &SearchContext,
3165    top_k: usize,
3166    namespaces: Option<&[&str]>,
3167    source_types: Option<&[SearchSourceType]>,
3168    session_ids: Option<&[&str]>,
3169    hnsw_hits: &[crate::hnsw::HnswHit],
3170) -> Result<SearchExecution, MemoryError> {
3171    let mut vector_hits = resolve_hnsw_hits_batched(
3172        conn,
3173        query_embedding,
3174        config,
3175        namespaces,
3176        source_types,
3177        session_ids,
3178        hnsw_hits,
3179    )?;
3180    let mut fallback = None;
3181    let mut degradations = Vec::new();
3182    let mut backend = "hnsw";
3183    let mut exact_rerank = config.rerank_from_f32;
3184
3185    if !hnsw_hits.is_empty()
3186        && vector_hits.len() < top_k
3187        && filters_are_active(namespaces, source_types, session_ids)
3188    {
3189        fallback = Some("hnsw_filtered_underreturn_fallback".to_string());
3190        degradations.push(format!(
3191            "HNSW returned {} post-filter vector candidates for requested top_k {}; exact filtered fallback was used",
3192            vector_hits.len(),
3193            top_k
3194        ));
3195        vector_hits = vector_search(
3196            conn,
3197            query_embedding,
3198            top_k,
3199            config.min_similarity,
3200            namespaces,
3201            source_types,
3202            session_ids,
3203        )?;
3204        backend = "hnsw_then_brute_force_f32";
3205        exact_rerank = true;
3206    }
3207
3208    let results = rrf_fuse_detailed_with_context(&[], &vector_hits, config, context, top_k);
3209    let receipt = build_receipt(
3210        context,
3211        query_embedding,
3212        "vector_only",
3213        backend,
3214        top_k,
3215        hnsw_hits.len(),
3216        vector_hits.len(),
3217        fallback,
3218        exact_rerank,
3219        &results,
3220        degradations,
3221    );
3222    Ok(SearchExecution { results, receipt })
3223}
3224
3225#[cfg(feature = "hnsw")]
3226#[allow(clippy::too_many_arguments)]
3227pub(crate) fn vector_only_search_with_hnsw_detailed(
3228    conn: &Connection,
3229    query_embedding: &[f32],
3230    config: &SearchConfig,
3231    top_k: usize,
3232    namespaces: Option<&[&str]>,
3233    source_types: Option<&[SearchSourceType]>,
3234    session_ids: Option<&[&str]>,
3235    hnsw_hits: &[crate::hnsw::HnswHit],
3236) -> Result<Vec<ExplainedResult>, MemoryError> {
3237    let context = SearchContext::default_now();
3238    Ok(vector_only_search_with_hnsw_detailed_with_context(
3239        conn,
3240        query_embedding,
3241        config,
3242        &context,
3243        top_k,
3244        namespaces,
3245        source_types,
3246        session_ids,
3247        hnsw_hits,
3248    )?
3249    .results)
3250}
3251
3252fn build_filter_clause(
3253    column: &str,
3254    values: Option<&[&str]>,
3255    param_offset: usize,
3256) -> (String, Vec<SqlValue>) {
3257    match values {
3258        Some(values) if !values.is_empty() => {
3259            let placeholders = (0..values.len())
3260                .map(|idx| format!("?{}", param_offset + idx))
3261                .collect::<Vec<_>>();
3262            let clause = format!(" AND {} IN ({})", column, placeholders.join(", "));
3263            let params = values
3264                .iter()
3265                .map(|value| SqlValue::Text((*value).to_string()))
3266                .collect();
3267            (clause, params)
3268        }
3269        _ => (String::new(), Vec::new()),
3270    }
3271}
3272
3273/// Deduplicate results by (source_type, source_id), keeping the first occurrence.
3274pub fn deduplicate_results(results: Vec<SearchResult>) -> Vec<SearchResult> {
3275    let mut seen = HashSet::new();
3276    results
3277        .into_iter()
3278        .filter(|result| seen.insert(source_dedup_key(&result.source)))
3279        .collect()
3280}
3281
3282#[cfg(test)]
3283mod tests {
3284    use super::*;
3285
3286    fn vector_row(id: &str) -> VectorRow {
3287        VectorRow {
3288            id: id.to_string(),
3289            content: format!("content {id}"),
3290            blob: bytemuck::cast_slice(&[1.0_f32, 0.0]).to_vec(),
3291            updated_at: None,
3292            source_type: SearchSourceType::Facts,
3293            filter_namespace: Some("default".to_string()),
3294            filter_session_id: None,
3295            source: SearchSource::Fact {
3296                fact_id: id.to_string(),
3297                namespace: "default".to_string(),
3298            },
3299        }
3300    }
3301
3302    #[test]
3303    fn timestamp_parser_accepts_sql_fractional_and_rfc3339_and_warns_by_returning_none() {
3304        assert!(parse_search_timestamp("2026-05-07 12:34:56").is_some());
3305        assert!(parse_search_timestamp("2026-05-07 12:34:56.123").is_some());
3306        assert!(parse_search_timestamp("2026-05-07T12:34:56Z").is_some());
3307        assert!(parse_search_timestamp("not-a-timestamp").is_none());
3308    }
3309
3310    #[test]
3311    fn vector_scan_hard_limit_blocks_before_unbounded_scan() {
3312        let old_warn = VECTOR_SCAN_WARN_LIMIT.swap(1, Ordering::SeqCst);
3313        let old_hard = VECTOR_SCAN_BLOCK_LIMIT.swap(2, Ordering::SeqCst);
3314        let rows = ["a", "b", "c"].into_iter().map(|id| Ok(vector_row(id)));
3315        let result = scan_vector_rows(rows, &[1.0, 0.0], -1.0, "fact");
3316        VECTOR_SCAN_WARN_LIMIT.store(old_warn, Ordering::SeqCst);
3317        VECTOR_SCAN_BLOCK_LIMIT.store(old_hard, Ordering::SeqCst);
3318
3319        match result {
3320            Err(MemoryError::VectorScanLimitExceeded {
3321                table,
3322                scanned,
3323                limit,
3324            }) => {
3325                assert_eq!(table, "fact");
3326                assert_eq!(scanned, 3);
3327                assert_eq!(limit, 2);
3328            }
3329            other => panic!("expected vector scan limit error, got {other:?}"),
3330        }
3331    }
3332}