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