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