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