From c34b3266f0b23d45c7f8c3cbe065dbdbe5fab794 Mon Sep 17 00:00:00 2001 From: ebigunso Date: Thu, 3 Sep 2026 20:01:59 +0900 Subject: [PATCH 1/3] feat: report vector recall completeness --- src/adapters/oxigraph/tests.rs | 25 ++- src/adapters/qdrant.rs | 1 + src/adapters/qdrant/store.rs | 284 +++++++++++++------------- src/adapters/qdrant/tie_closure.rs | 217 ++++++++++++++++++++ src/api/types/retrieval.rs | 33 +++ src/memory.rs | 60 +++++- src/models/vector/candidate_record.rs | 20 +- src/ports/vector_candidate.rs | 15 +- src/test_support.rs | 111 ++++++++-- src/usecases/correct_forget.rs | 41 ++-- src/usecases/remember.rs | 16 +- src/usecases/retrieve.rs | 91 ++++++++- 12 files changed, 707 insertions(+), 207 deletions(-) create mode 100644 src/adapters/qdrant/tie_closure.rs diff --git a/src/adapters/oxigraph/tests.rs b/src/adapters/oxigraph/tests.rs index 08a594e8..46d8f3a6 100644 --- a/src/adapters/oxigraph/tests.rs +++ b/src/adapters/oxigraph/tests.rs @@ -26,7 +26,7 @@ mod tests { GraphExpansionFanoutOverride, GraphExpansionFilteredReason, GraphExpansionLifecyclePolicy, GraphExpansionQuery, GraphObjectQuery, }; - use crate::ports::vector_candidate::VectorCandidateStore; + use crate::ports::vector_candidate::{VectorCandidateRecall, VectorCandidateStore}; use crate::test_support::{ high_fanout_graph_fixture, representative_fixtures, FakeGraphAuthorityStore, }; @@ -1894,8 +1894,27 @@ mod tests { async fn search_candidates( &self, query: &VectorCandidateSearch, - ) -> Result { - Ok(CanonicalCandidates::new(self.candidates.clone()).truncated(query.limit)) + ) -> Result { + if query.limit == 0 || query.object_types.is_empty() { + return Ok(VectorCandidateRecall { + candidates: CanonicalCandidates::new([]), + completeness: + crate::api::types::retrieval::VectorRecallCompleteness::NotRequested, + }); + } + let candidates = self + .candidates + .iter() + .filter(|candidate| query.object_types.contains(&candidate.object_type)) + .cloned() + .collect::>(); + let scanned = candidates.len(); + Ok(VectorCandidateRecall { + candidates: CanonicalCandidates::new(candidates).truncated(query.limit), + completeness: crate::api::types::retrieval::VectorRecallCompleteness::Exhaustive { + scanned, + }, + }) } async fn delete_candidates(&self, _object_ids: &[MemoryId]) -> Result<(), CustomError> { diff --git a/src/adapters/qdrant.rs b/src/adapters/qdrant.rs index bdd7faa1..3d582220 100644 --- a/src/adapters/qdrant.rs +++ b/src/adapters/qdrant.rs @@ -1,4 +1,5 @@ mod payload; mod store; +pub(crate) mod tie_closure; pub(crate) use store::QdrantVectorCandidateStore; diff --git a/src/adapters/qdrant/store.rs b/src/adapters/qdrant/store.rs index 64ec51cd..f6dab53a 100644 --- a/src/adapters/qdrant/store.rs +++ b/src/adapters/qdrant/store.rs @@ -7,8 +7,8 @@ use async_trait::async_trait; use qdrant_client::qdrant::{ points_selector::PointsSelectorOneOf, value::Kind, vectors_config, Condition, CreateCollectionBuilder, CreateFieldIndexCollectionBuilder, DeletePointsBuilder, Distance, - Filter, PointStruct, ScoredPoint, SearchPointsBuilder, UpsertPointsBuilder, VectorParams, - VectorsConfig, + Filter, PointStruct, ScoredPoint, ScrollPointsBuilder, SearchPointsBuilder, + UpsertPointsBuilder, VectorParams, VectorsConfig, }; use qdrant_client::{config::QdrantConfig, Qdrant, QdrantError}; @@ -18,18 +18,16 @@ use crate::errors::{ VectorDatabaseError, VectorDatabaseErrorKind, }; use crate::models::vector::{ - CanonicalCandidates, VectorCandidateMatch, VectorCandidateSearch, VectorRecordEmbedding, - VectorSurface, + VectorCandidateMatch, VectorCandidateSearch, VectorRecordEmbedding, VectorSurface, }; -use crate::ports::vector_candidate::VectorCandidateStore; +use crate::ports::vector_candidate::{VectorCandidateRecall, VectorCandidateStore}; use super::payload::{ qdrant_payload_map, QdrantPayloadSchema, OBJECT_ID_FIELD, OBJECT_TYPE_FIELD, SURFACE_FIELD, }; +use super::tie_closure::close_tie_cohort; const QDRANT_CANDIDATE_TIMEOUT_SECS: u64 = 30; -const QDRANT_TIE_COHORT_MIN_EXTRA_CANDIDATES: usize = 4_096; -const QDRANT_TIE_COHORT_LIMIT_MULTIPLIER: usize = 16; const QDRANT_CONNECT_FAILURE_PREFIX: &str = "Failed to connect to "; pub(crate) struct QdrantVectorCandidateStore { @@ -155,9 +153,7 @@ impl QdrantVectorCandidateStore { .with_payload(true) .with_vectors(false); - if let Some(filter) = qdrant_candidate_filter(query) { - builder = builder.filter(filter); - } + builder = builder.filter(qdrant_candidate_filter(query)); let response = self .client @@ -170,48 +166,27 @@ impl QdrantVectorCandidateStore { .map(scored_point_to_match) .collect() } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum TieCohortFetchDecision { - Return, - ReturnAtBound, - Grow(usize), -} - -fn tie_cohort_fetch_bound(limit: usize) -> usize { - limit - .saturating_mul(QDRANT_TIE_COHORT_LIMIT_MULTIPLIER) - .max(limit.saturating_add(QDRANT_TIE_COHORT_MIN_EXTRA_CANDIDATES)) -} - -fn tie_cohort_fetch_decision( - admitted_limit: usize, - fetch_limit: usize, - fetch_bound: usize, - fetched_count: usize, - candidates: &[VectorCandidateMatch], -) -> TieCohortFetchDecision { - if fetched_count < fetch_limit || tie_cohort_is_closed(candidates, admitted_limit) { - return TieCohortFetchDecision::Return; - } - if fetch_limit >= fetch_bound { - return TieCohortFetchDecision::ReturnAtBound; - } - - TieCohortFetchDecision::Grow(fetch_limit.saturating_mul(2).min(fetch_bound)) -} -fn tie_cohort_is_closed(candidates: &[VectorCandidateMatch], admitted_limit: usize) -> bool { - if admitted_limit == 0 || candidates.len() <= admitted_limit { - return false; + async fn scroll_zero_norm_candidate_batch( + &self, + query: &VectorCandidateSearch, + fetch_limit: usize, + ) -> Result, CustomError> { + let request = ScrollPointsBuilder::new(&self.collection_name) + .filter(qdrant_candidate_filter(query)) + .limit(fetch_limit as u32) + .with_payload(true) + .with_vectors(false) + .build(); + self.client + .scroll(request) + .await + .map_err(qdrant_error)? + .result + .into_iter() + .map(|point| qdrant_payload_to_match(&point.payload, 0.0)) + .collect() } - - candidates.last().is_some_and(|tail| { - tail.score - .total_cmp(&candidates[admitted_limit - 1].score) - .is_lt() - }) } fn validate_collection_vector_config( @@ -285,37 +260,29 @@ impl VectorCandidateStore for QdrantVectorCandidateStore { async fn search_candidates( &self, query: &VectorCandidateSearch, - ) -> Result { - if query.limit == 0 { - return Ok(CanonicalCandidates::new([])); + ) -> Result { + if query.limit == 0 || query.object_types.is_empty() { + return Ok(VectorCandidateRecall { + candidates: crate::models::vector::CanonicalCandidates::new([]), + completeness: crate::api::types::retrieval::VectorRecallCompleteness::NotRequested, + }); } - // Fetch past K until the boundary tie is closed. Growth is bounded by - // max(K * 16, K + 4096), which avoids unbounded reads when an entire - // collection ties. At the bound, results are canonical and deterministic - // for the fetched set, but membership can still vary if the equal-score - // cohort itself exceeds the bound. A future adapter-observability channel - // in the RetrievalTrace family should surface that degradation. - let fetch_bound = tie_cohort_fetch_bound(query.limit); - let mut fetch_limit = query.limit.saturating_add(1).min(fetch_bound); - loop { - let fetched = self.search_candidate_batch(query, fetch_limit).await?; - let fetched_count = fetched.len(); - let candidates = CanonicalCandidates::new(fetched); - - match tie_cohort_fetch_decision( - query.limit, - fetch_limit, - fetch_bound, - fetched_count, - &candidates, - ) { - TieCohortFetchDecision::Grow(next_limit) => fetch_limit = next_limit, - TieCohortFetchDecision::Return | TieCohortFetchDecision::ReturnAtBound => { - return Ok(candidates.truncated(query.limit)); - } + let zero_norm = query.is_zero_norm(); + let closed = close_tie_cohort(query.limit, |fetch_limit| async move { + if zero_norm { + self.scroll_zero_norm_candidate_batch(query, fetch_limit) + .await + } else { + self.search_candidate_batch(query, fetch_limit).await } - } + }) + .await?; + let completeness = closed.completeness(zero_norm.then_some(closed.fetched)); + Ok(VectorCandidateRecall { + candidates: closed.candidates, + completeness, + }) } async fn delete_candidates(&self, object_ids: &[MemoryId]) -> Result<(), CustomError> { @@ -498,13 +465,11 @@ fn qdrant_candidate_config(url: &str) -> QdrantConfig { .keep_alive_while_idle() } -fn qdrant_candidate_filter(query: &VectorCandidateSearch) -> Option { - (!query.object_types.is_empty()).then(|| { - Filter::must([any_field_matches( - OBJECT_TYPE_FIELD, - query.object_types.iter().copied().map(object_type_name), - )]) - }) +fn qdrant_candidate_filter(query: &VectorCandidateSearch) -> Filter { + Filter::must([any_field_matches( + OBJECT_TYPE_FIELD, + query.object_types.iter().copied().map(object_type_name), + )]) } fn any_field_matches( @@ -540,19 +505,26 @@ fn qdrant_point_structs( } fn scored_point_to_match(point: ScoredPoint) -> Result { - let object_id = payload_string(&point.payload, OBJECT_ID_FIELD)?; + qdrant_payload_to_match(&point.payload, point.score) +} + +fn qdrant_payload_to_match( + payload: &HashMap, + score: f32, +) -> Result { + let object_id = payload_string(payload, OBJECT_ID_FIELD)?; let object_id = uuid::Uuid::parse_str(&object_id).map_err(|error| { CustomError::DatabaseError(format!("Invalid Qdrant object_id payload UUID: {error}")) })?; - let object_type = parse_object_type(payload_string(&point.payload, OBJECT_TYPE_FIELD)?)?; - let surface = parse_vector_surface(payload_string(&point.payload, SURFACE_FIELD)?)?; + let object_type = parse_object_type(payload_string(payload, OBJECT_TYPE_FIELD)?)?; + let surface = parse_vector_surface(payload_string(payload, SURFACE_FIELD)?)?; Ok(VectorCandidateMatch::new( object_id, object_type, surface, - point.score, + score, )) } @@ -647,9 +619,11 @@ mod tests { CONTENT_TEXT_FIELD, GRAPH_URI_FIELD, IS_CURRENT_FIELD, RETENTION_STATE_FIELD, }; use super::*; + use crate::api::types::retrieval::VectorRecallCompleteness; use crate::domain::{graph_uri, RetentionState, DEFAULT_SCHEMA_VERSION}; use crate::models::vector::{ - VectorRecord, VectorRecordEmbedding, VectorRelationshipHints, VectorSurface, + CanonicalCandidates, VectorRecord, VectorRecordEmbedding, VectorRelationshipHints, + VectorSurface, }; use qdrant_client::qdrant::condition::ConditionOneOf; use qdrant_client::qdrant::{ @@ -675,11 +649,8 @@ mod tests { #[test] fn candidate_filter_maps_live_object_type_scope() { - assert!(qdrant_candidate_filter(&VectorCandidateSearch::new(vec![1.0, 0.0], 10)).is_none()); - - let query = VectorCandidateSearch::new(vec![1.0, 0.0], 10) - .with_object_types(vec![ObjectType::Episode]); - let filter = qdrant_candidate_filter(&query).expect("object-type scope should build"); + let query = VectorCandidateSearch::new(vec![1.0, 0.0], 10, vec![ObjectType::Episode]); + let filter = qdrant_candidate_filter(&query); let Some(ConditionOneOf::Field(field)) = &filter.must[0].condition_one_of else { panic!("single object type should map to a field condition"); }; @@ -687,6 +658,22 @@ mod tests { assert_eq!(field.key, OBJECT_TYPE_FIELD); } + #[tokio::test] + async fn empty_scope_and_zero_limit_return_without_contacting_qdrant() { + let store = + QdrantVectorCandidateStore::new("http://127.0.0.1:1", "not_contacted", 2).unwrap(); + let queries = [ + VectorCandidateSearch::new(vec![1.0, 0.0], 10, Vec::new()), + VectorCandidateSearch::new(vec![1.0, 0.0], 0, vec![ObjectType::Episode]), + ]; + + for query in queries { + let recall = store.search_candidates(&query).await.unwrap(); + assert!(recall.candidates.is_empty()); + assert_eq!(recall.completeness, VectorRecallCompleteness::NotRequested); + } + } + #[test] fn qdrant_response_error_preserves_typed_transport_status() { let error = qdrant_error(QdrantError::ResponseError { @@ -1126,44 +1113,6 @@ mod tests { assert_eq!(matches[2].object_id, second_tied_id); } - #[test] - fn all_tied_cohort_at_fetch_bound_degrades_to_canonical_fetched_membership() { - let admitted_limit = 2; - let fetched = (1..=6) - .rev() - .map(|value| { - VectorCandidateMatch::new( - Uuid::from_u128(value), - ObjectType::Episode, - VectorSurface::Summary, - 1.0, - ) - }) - .collect::>(); - let candidates = CanonicalCandidates::new(fetched); - let fetch_bound = candidates.len(); - - assert_eq!( - tie_cohort_fetch_decision( - admitted_limit, - fetch_bound, - fetch_bound, - fetch_bound, - &candidates, - ), - TieCohortFetchDecision::ReturnAtBound - ); - - let candidates = candidates.truncated(admitted_limit); - assert_eq!( - candidates - .iter() - .map(|candidate| candidate.object_id) - .collect::>(), - vec![Uuid::from_u128(1), Uuid::from_u128(2)] - ); - } - #[test] fn candidate_mapping_does_not_return_lifecycle_hints_as_authority() { let object_id = Uuid::new_v4(); @@ -1213,15 +1162,20 @@ mod tests { .expect("upsert succeeds"); let matches = store - .search_candidates( - &VectorCandidateSearch::new(vec![1.0, 0.0], 1) - .with_object_types(vec![ObjectType::DerivedMemory]), - ) + .search_candidates(&VectorCandidateSearch::new( + vec![1.0, 0.0], + 1, + vec![ObjectType::DerivedMemory], + )) .await .expect("search succeeds"); - assert_eq!(matches.len(), 1); - assert_eq!(matches[0].object_id, object_id); + assert_eq!(matches.candidates.len(), 1); + assert_eq!(matches.candidates[0].object_id, object_id); + assert_eq!( + matches.completeness, + VectorRecallCompleteness::BoundaryTieClosed { fetched: 1 } + ); store .delete_candidates(&[object_id]) @@ -1230,6 +1184,50 @@ mod tests { let _ = store.client.delete_collection(&collection_name).await; } + #[tokio::test] + #[ignore = "requires local Qdrant: docker compose -f docker-compose.qdrant.yml up -d and QDRANT_CONNECTION_STRING"] + async fn qdrant_candidate_store_live_scores_zero_norm_query_candidates_zero() { + let url = env::var("QDRANT_CONNECTION_STRING") + .expect("QDRANT_CONNECTION_STRING is required for live Qdrant regression"); + let collection_name = format!("cm_zero_norm_{}", Uuid::new_v4().simple()); + let store = QdrantVectorCandidateStore::new(&url, &collection_name, 2).unwrap(); + let records = [ + idle_gap_vector_record(ObjectType::Episode), + idle_gap_vector_record(ObjectType::Episode), + ]; + let embeddings = [vec![1.0, 0.0], vec![0.0, 1.0]]; + let record_embeddings = records + .iter() + .zip(&embeddings) + .map(|(record, embedding)| VectorRecordEmbedding::new(record, embedding)) + .collect::>(); + + store.init_collection().await.expect("collection init"); + store + .upsert_vector_records(&record_embeddings) + .await + .expect("upsert succeeds"); + let recall = store + .search_candidates(&VectorCandidateSearch::new( + vec![0.0, 0.0], + 10, + vec![ObjectType::Episode], + )) + .await + .expect("zero-norm search succeeds"); + + assert_eq!(recall.candidates.len(), 2); + assert!(recall + .candidates + .iter() + .all(|candidate| candidate.score == 0.0)); + assert_eq!( + recall.completeness, + VectorRecallCompleteness::Exhaustive { scanned: 2 } + ); + let _ = store.client.delete_collection(&collection_name).await; + } + #[tokio::test] #[ignore = "requires local Qdrant: docker compose -f docker-compose.qdrant.yml up -d and QDRANT_CONNECTION_STRING"] async fn qdrant_candidate_store_live_closes_equal_score_boundary_deterministically() { @@ -1271,8 +1269,7 @@ mod tests { .await .expect("upsert succeeds"); - let query = VectorCandidateSearch::new(vec![1.0, 0.0], 5) - .with_object_types(vec![ObjectType::Episode]); + let query = VectorCandidateSearch::new(vec![1.0, 0.0], 5, vec![ObjectType::Episode]); let expected = object_ids[..5].to_vec(); for _ in 0..8 { let matches = store @@ -1281,11 +1278,16 @@ mod tests { .expect("equal-score search succeeds"); assert_eq!( matches + .candidates .iter() .map(|candidate| candidate.object_id) .collect::>(), expected ); + assert_eq!( + matches.completeness, + VectorRecallCompleteness::BoundaryTieClosed { fetched: 12 } + ); } let _ = store.client.delete_collection(&collection_name).await; diff --git a/src/adapters/qdrant/tie_closure.rs b/src/adapters/qdrant/tie_closure.rs new file mode 100644 index 00000000..751c2ffd --- /dev/null +++ b/src/adapters/qdrant/tie_closure.rs @@ -0,0 +1,217 @@ +use std::future::Future; + +use crate::api::types::retrieval::VectorRecallCompleteness; +use crate::models::vector::{CanonicalCandidates, VectorCandidateMatch}; + +const TIE_COHORT_MIN_EXTRA_CANDIDATES: usize = 4_096; +const TIE_COHORT_LIMIT_MULTIPLIER: usize = 16; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum FetchDecision { + Return, + ReturnAtBound, + Grow(usize), +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum TieClosure { + Closed, + OpenAtBound, +} + +pub(crate) struct TieClosureResult { + pub(crate) candidates: CanonicalCandidates, + pub(crate) fetched: usize, + fetch_bound: usize, + closure: TieClosure, +} + +impl TieClosureResult { + pub(crate) fn completeness( + &self, + exhaustive_scanned: Option, + ) -> VectorRecallCompleteness { + match (self.closure, exhaustive_scanned) { + (TieClosure::Closed, Some(scanned)) => VectorRecallCompleteness::Exhaustive { scanned }, + (TieClosure::Closed, None) => VectorRecallCompleteness::BoundaryTieClosed { + fetched: self.fetched, + }, + (TieClosure::OpenAtBound, _) => VectorRecallCompleteness::BoundaryTieOpen { + fetched: self.fetched, + fetch_bound: self.fetch_bound, + }, + } + } +} + +pub(crate) async fn close_tie_cohort( + admitted_limit: usize, + mut fetch: F, +) -> Result +where + F: FnMut(usize) -> Fut, + Fut: Future, E>>, +{ + let fetch_bound = tie_cohort_fetch_bound(admitted_limit); + let mut fetch_limit = admitted_limit.saturating_add(1).min(fetch_bound); + + loop { + let fetched = fetch(fetch_limit).await?; + let fetched_count = fetched.len(); + let candidates = CanonicalCandidates::new(fetched); + + match fetch_decision( + admitted_limit, + fetch_limit, + fetch_bound, + fetched_count, + &candidates, + ) { + FetchDecision::Grow(next_limit) => fetch_limit = next_limit, + FetchDecision::Return => { + return Ok(TieClosureResult { + candidates: candidates.truncated(admitted_limit), + fetched: fetched_count, + fetch_bound, + closure: TieClosure::Closed, + }); + } + FetchDecision::ReturnAtBound => { + return Ok(TieClosureResult { + candidates: candidates.truncated(admitted_limit), + fetched: fetched_count, + fetch_bound, + closure: TieClosure::OpenAtBound, + }); + } + } + } +} + +fn tie_cohort_fetch_bound(limit: usize) -> usize { + limit + .saturating_mul(TIE_COHORT_LIMIT_MULTIPLIER) + .max(limit.saturating_add(TIE_COHORT_MIN_EXTRA_CANDIDATES)) +} + +fn fetch_decision( + admitted_limit: usize, + fetch_limit: usize, + fetch_bound: usize, + fetched_count: usize, + candidates: &[VectorCandidateMatch], +) -> FetchDecision { + if fetched_count < fetch_limit || tie_cohort_is_closed(candidates, admitted_limit) { + return FetchDecision::Return; + } + if fetch_limit >= fetch_bound { + return FetchDecision::ReturnAtBound; + } + + FetchDecision::Grow(fetch_limit.saturating_mul(2).min(fetch_bound)) +} + +fn tie_cohort_is_closed(candidates: &[VectorCandidateMatch], admitted_limit: usize) -> bool { + if admitted_limit == 0 || candidates.len() <= admitted_limit { + return false; + } + + candidates.last().is_some_and(|tail| { + tail.score + .total_cmp(&candidates[admitted_limit - 1].score) + .is_lt() + }) +} + +#[cfg(test)] +mod tests { + use std::cell::RefCell; + use std::future::ready; + + use uuid::Uuid; + + use super::*; + use crate::domain::ObjectType; + use crate::models::vector::VectorSurface; + + fn candidate(id: u128, score: f32) -> VectorCandidateMatch { + VectorCandidateMatch::new( + Uuid::from_u128(id), + ObjectType::Episode, + VectorSurface::Summary, + score, + ) + } + + #[test] + fn fetch_decision_returns_when_the_cutoff_cohort_is_closed() { + let candidates = + CanonicalCandidates::new([candidate(1, 1.0), candidate(2, 1.0), candidate(3, 0.5)]); + + assert_eq!( + fetch_decision(2, 3, 10, 3, &candidates), + FetchDecision::Return + ); + let result = TieClosureResult { + candidates: candidates.truncated(2), + fetched: 3, + fetch_bound: 10, + closure: TieClosure::Closed, + }; + assert_eq!( + result.completeness(None), + VectorRecallCompleteness::BoundaryTieClosed { fetched: 3 } + ); + } + + #[test] + fn all_tied_cohort_at_fetch_bound_is_open() { + let candidates = CanonicalCandidates::new((1..=6).rev().map(|id| candidate(id, 1.0))); + + assert_eq!( + fetch_decision(2, 6, 6, 6, &candidates), + FetchDecision::ReturnAtBound + ); + let result = TieClosureResult { + candidates: candidates.truncated(2), + fetched: 6, + fetch_bound: 6, + closure: TieClosure::OpenAtBound, + }; + assert_eq!( + result.completeness(None), + VectorRecallCompleteness::BoundaryTieOpen { + fetched: 6, + fetch_bound: 6, + } + ); + assert_eq!(result.candidates[0].object_id, Uuid::from_u128(1)); + assert_eq!(result.candidates[1].object_id, Uuid::from_u128(2)); + } + + #[tokio::test] + async fn closure_loop_grows_and_canonicalizes_before_returning() { + let candidates = [candidate(2, 1.0), candidate(1, 1.0), candidate(3, 0.5)]; + let fetch_limits = RefCell::new(Vec::new()); + + let result = close_tie_cohort(1, |fetch_limit| { + fetch_limits.borrow_mut().push(fetch_limit); + ready(Ok::<_, ()>( + candidates.iter().take(fetch_limit).cloned().collect(), + )) + }) + .await + .unwrap(); + + assert_eq!(*fetch_limits.borrow(), vec![2, 4]); + assert_eq!(result.candidates[0].object_id, Uuid::from_u128(1)); + assert_eq!( + result.completeness(None), + VectorRecallCompleteness::BoundaryTieClosed { fetched: 3 } + ); + assert_eq!( + result.completeness(Some(3)), + VectorRecallCompleteness::Exhaustive { scanned: 3 } + ); + } +} diff --git a/src/api/types/retrieval.rs b/src/api/types/retrieval.rs index 01a02fa1..a987d1d0 100644 --- a/src/api/types/retrieval.rs +++ b/src/api/types/retrieval.rs @@ -5,6 +5,7 @@ use crate::domain::{ GraphFailureMode, MemoryId, MemoryObjectRef, MemoryThread, ObjectType, Observation, RelationType, RetentionState, ThreadStatus, }; +use crate::errors::{ConfigValidationError, ConfigValidationReason}; #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct RetrievalContext { @@ -35,6 +36,20 @@ impl RetrievalContext { self.include_trace = true; self } + + pub(crate) fn validate(&self) -> Result<(), ConfigValidationError> { + if self.object_type_defaults.is_empty() { + return Err(ConfigValidationError { + keys: vec!["object_type_defaults"], + reason: ConfigValidationReason::OutOfDomain { + expected: "at least one retrieval object type", + actual: "[]".to_owned(), + }, + }); + } + + Ok(()) + } } impl Default for RetrievalContext { @@ -254,6 +269,7 @@ pub struct RetrievalTelemetry { pub configured_lifecycle_policy: RetrievalLifecyclePolicy, pub query_embedding_dimension: usize, pub returned_vector_candidate_count: usize, + pub vector_recall_completeness: VectorRecallCompleteness, pub unique_graph_root_candidate_count: usize, pub selected_graph_root_count: usize, pub graph_root_omission_count: usize, @@ -262,6 +278,23 @@ pub struct RetrievalTelemetry { pub section_pressure: Vec, } +#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum VectorRecallCompleteness { + #[default] + NotRequested, + Exhaustive { + scanned: usize, + }, + BoundaryTieClosed { + fetched: usize, + }, + BoundaryTieOpen { + fetched: usize, + fetch_bound: usize, + }, +} + #[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)] pub struct SelectivityTelemetry { pub decision_count: usize, diff --git a/src/memory.rs b/src/memory.rs index 5d109ab6..d08111bc 100644 --- a/src/memory.rs +++ b/src/memory.rs @@ -160,7 +160,7 @@ mod tests { use crate::config::Settings; use crate::ports::embedder::MemoryEmbedder; use crate::ports::graph_authority::{GraphAuthorityStore, GraphObjectQuery}; - use crate::ports::vector_candidate::VectorCandidateStore; + use crate::ports::vector_candidate::{VectorCandidateRecall, VectorCandidateStore}; use crate::*; use async_trait::async_trait; use secrecy::SecretString; @@ -458,6 +458,26 @@ mod tests { assert_eq!(outcome.trace.as_ref().unwrap().vector_candidates.len(), 1); } + #[tokio::test] + async fn retrieve_rejects_an_empty_configured_object_type_scope_at_the_boundary() { + let memory = injected_memory(); + let mut context = RetrievalContext::new("invalid empty scope"); + context.object_type_defaults.clear(); + + let error = memory.retrieve(context).await.unwrap_err(); + + assert!(matches!( + error, + CustomError::ConfigValidation(ConfigValidationError { + keys, + reason: ConfigValidationReason::OutOfDomain { + expected: "at least one retrieval object type", + actual, + }, + }) if keys == vec!["object_type_defaults"] && actual == "[]" + )); + } + #[tokio::test] async fn injected_facade_corrects_derived_memory_and_retrieval_excludes_superseded_memory() { let (memory, fixtures, replacement_id) = lifecycle_memory().await; @@ -1070,8 +1090,27 @@ mod tests { async fn search_candidates( &self, query: &VectorCandidateSearch, - ) -> Result { - Ok(CanonicalCandidates::new(self.candidates.clone()).truncated(query.limit)) + ) -> Result { + if query.limit == 0 || query.object_types.is_empty() { + return Ok(VectorCandidateRecall { + candidates: CanonicalCandidates::new([]), + completeness: + crate::api::types::retrieval::VectorRecallCompleteness::NotRequested, + }); + } + let candidates = self + .candidates + .iter() + .filter(|candidate| query.object_types.contains(&candidate.object_type)) + .cloned() + .collect::>(); + let scanned = candidates.len(); + Ok(VectorCandidateRecall { + candidates: CanonicalCandidates::new(candidates).truncated(query.limit), + completeness: crate::api::types::retrieval::VectorRecallCompleteness::Exhaustive { + scanned, + }, + }) } async fn delete_candidates(&self, _object_ids: &[MemoryId]) -> Result<(), CustomError> { @@ -1098,9 +1137,18 @@ mod tests { async fn search_candidates( &self, - _query: &VectorCandidateSearch, - ) -> Result { - Ok(CanonicalCandidates::new([])) + query: &VectorCandidateSearch, + ) -> Result { + Ok(VectorCandidateRecall { + candidates: CanonicalCandidates::new([]), + completeness: if query.limit == 0 || query.object_types.is_empty() { + crate::api::types::retrieval::VectorRecallCompleteness::NotRequested + } else { + crate::api::types::retrieval::VectorRecallCompleteness::Exhaustive { + scanned: 0, + } + }, + }) } async fn delete_candidates(&self, _object_ids: &[MemoryId]) -> Result<(), CustomError> { diff --git a/src/models/vector/candidate_record.rs b/src/models/vector/candidate_record.rs index ead8774e..2ab0e423 100644 --- a/src/models/vector/candidate_record.rs +++ b/src/models/vector/candidate_record.rs @@ -71,17 +71,20 @@ pub(crate) struct VectorCandidateSearch { } impl VectorCandidateSearch { - pub(crate) fn new(query_embedding: Vec, limit: usize) -> Self { + pub(crate) fn new( + query_embedding: Vec, + limit: usize, + object_types: Vec, + ) -> Self { Self { query_embedding, limit, - object_types: Vec::new(), + object_types, } } - pub(crate) fn with_object_types(mut self, object_types: Vec) -> Self { - self.object_types = object_types; - self + pub(crate) fn is_zero_norm(&self) -> bool { + self.query_embedding.iter().all(|value| *value == 0.0) } } @@ -205,8 +208,11 @@ mod tests { #[test] fn vector_candidate_search_can_scope_by_canonical_object_types() { - let search = VectorCandidateSearch::new(vec![1.0, 0.0], 10) - .with_object_types(vec![ObjectType::Episode, ObjectType::DerivedMemory]); + let search = VectorCandidateSearch::new( + vec![1.0, 0.0], + 10, + vec![ObjectType::Episode, ObjectType::DerivedMemory], + ); assert_eq!(search.query_embedding, vec![1.0, 0.0]); assert_eq!(search.limit, 10); diff --git a/src/ports/vector_candidate.rs b/src/ports/vector_candidate.rs index 2cfa1223..2abd68e4 100644 --- a/src/ports/vector_candidate.rs +++ b/src/ports/vector_candidate.rs @@ -2,10 +2,17 @@ // tests use deterministic fake stores. use async_trait::async_trait; +use crate::api::types::retrieval::VectorRecallCompleteness; use crate::domain::MemoryId; use crate::errors::CustomError; use crate::models::vector::{CanonicalCandidates, VectorCandidateSearch, VectorRecordEmbedding}; +#[derive(Debug, Clone, PartialEq)] +pub(crate) struct VectorCandidateRecall { + pub(crate) candidates: CanonicalCandidates, + pub(crate) completeness: VectorRecallCompleteness, +} + #[async_trait] pub(crate) trait VectorCandidateStore: Send + Sync { async fn upsert_vector_records( @@ -14,13 +21,11 @@ pub(crate) trait VectorCandidateStore: Send + Sync { ) -> Result<(), CustomError>; /// Returns at most `query.limit` unique object/surface matches in canonical - /// score-descending, object-type, object-id, surface order. Adapters close - /// equal-score cutoff cohorts before canonical truncation, subject to their - /// documented bounded-overfetch degradation policy. + /// score-descending, object-type, object-id, surface order. async fn search_candidates( &self, query: &VectorCandidateSearch, - ) -> Result; + ) -> Result; async fn delete_candidates(&self, object_ids: &[MemoryId]) -> Result<(), CustomError>; } @@ -37,7 +42,7 @@ impl VectorCandidateStore for Box { async fn search_candidates( &self, query: &VectorCandidateSearch, - ) -> Result { + ) -> Result { (**self).search_candidates(query).await } diff --git a/src/test_support.rs b/src/test_support.rs index 72104661..0f1b0fc0 100644 --- a/src/test_support.rs +++ b/src/test_support.rs @@ -7,6 +7,7 @@ use async_trait::async_trait; use chrono::{DateTime, Utc}; use uuid::Uuid; +use crate::api::types::retrieval::VectorRecallCompleteness; use crate::domain::{ DerivedMemory, DerivedType, Entity, EntityType, Episode, MemoryId, MemoryLink, MemoryObject, MemoryObjectRef, MemoryThread, Modality, ObjectType, Observation, RelationType, RetentionState, @@ -25,7 +26,7 @@ use crate::ports::graph_authority::{ GraphAuthorityStore, GraphDerivedMemoryProvenanceQuery, GraphDerivedMemoryThreadQuery, GraphExpansion, GraphExpansionQuery, GraphObjectQuery, }; -use crate::ports::vector_candidate::VectorCandidateStore; +use crate::ports::vector_candidate::{VectorCandidateRecall, VectorCandidateStore}; #[derive(Debug, Default)] pub(crate) struct FakeVectorCandidateStore { @@ -74,13 +75,18 @@ impl VectorCandidateStore for FakeVectorCandidateStore { async fn search_candidates( &self, query: &VectorCandidateSearch, - ) -> Result { + ) -> Result { + if query.limit == 0 || query.object_types.is_empty() { + return Ok(VectorCandidateRecall { + candidates: CanonicalCandidates::new([]), + completeness: VectorRecallCompleteness::NotRequested, + }); + } + let records = lock(&self.records)?; let matches: Vec<_> = records .iter() - .filter(|record| { - query.object_types.is_empty() || query.object_types.contains(&record.object_type) - }) + .filter(|record| query.object_types.contains(&record.object_type)) .map(|record| { VectorCandidateMatch::new( record.object_id, @@ -91,7 +97,11 @@ impl VectorCandidateStore for FakeVectorCandidateStore { }) .collect(); - Ok(CanonicalCandidates::new(matches).truncated(query.limit)) + let scanned = matches.len(); + Ok(VectorCandidateRecall { + candidates: CanonicalCandidates::new(matches).truncated(query.limit), + completeness: VectorRecallCompleteness::Exhaustive { scanned }, + }) } async fn delete_candidates(&self, object_ids: &[MemoryId]) -> Result<(), CustomError> { @@ -101,6 +111,32 @@ impl VectorCandidateStore for FakeVectorCandidateStore { } } +pub(crate) fn zero_norm_vector_fixture() -> (Vec, VectorCandidateSearch) { + ( + vec![ + VectorCandidateRecord::new( + Uuid::from_u128(1), + ObjectType::Episode, + VectorSurface::Summary, + vec![1.0, 0.0], + ), + VectorCandidateRecord::new( + Uuid::from_u128(2), + ObjectType::Episode, + VectorSurface::Summary, + vec![0.0, 1.0], + ), + VectorCandidateRecord::new( + Uuid::from_u128(3), + ObjectType::Observation, + VectorSurface::Text, + vec![1.0, 0.0], + ), + ], + VectorCandidateSearch::new(vec![0.0, 0.0], 10, vec![ObjectType::Episode]), + ) +} + #[derive(Debug, Default)] pub(crate) struct FakeGraphAuthorityStore { objects: Mutex>, @@ -1004,14 +1040,22 @@ mod tests { .await .unwrap(); - let query = VectorCandidateSearch::new(vec![1.0, 0.0], 10); + let query = VectorCandidateSearch::new( + vec![1.0, 0.0], + 10, + vec![ObjectType::Episode, ObjectType::Observation], + ); let first_result = store.search_candidates(&query).await.unwrap(); let second_result = store.search_candidates(&query).await.unwrap(); assert_eq!(first_result, second_result); - assert_eq!(first_result[0].object_id, fixtures.episode.id); - assert_eq!(first_result[0].object_type, ObjectType::Episode); - assert_eq!(first_result[0].surface, VectorSurface::Summary); + assert_eq!( + first_result.completeness, + VectorRecallCompleteness::Exhaustive { scanned: 2 } + ); + assert_eq!(first_result.candidates[0].object_id, fixtures.episode.id); + assert_eq!(first_result.candidates[0].object_type, ObjectType::Episode); + assert_eq!(first_result.candidates[0].surface, VectorSurface::Summary); store .delete_candidates(&[fixtures.episode.id]) @@ -1019,8 +1063,11 @@ mod tests { .unwrap(); let after_delete = store.search_candidates(&query).await.unwrap(); - assert_eq!(after_delete.len(), 1); - assert_eq!(after_delete[0].object_id, fixtures.salient_observation.id); + assert_eq!(after_delete.candidates.len(), 1); + assert_eq!( + after_delete.candidates[0].object_id, + fixtures.salient_observation.id + ); } #[tokio::test] @@ -1056,13 +1103,18 @@ mod tests { first_store.upsert_candidates(&candidates).await.unwrap(); second_store.upsert_candidates(&reversed).await.unwrap(); - let query = VectorCandidateSearch::new(vec![1.0, 0.0], 2); + let query = VectorCandidateSearch::new( + vec![1.0, 0.0], + 2, + vec![ObjectType::Episode, ObjectType::Observation], + ); let first = first_store.search_candidates(&query).await.unwrap(); let second = second_store.search_candidates(&query).await.unwrap(); assert_eq!(first, second); assert_eq!( first + .candidates .iter() .map(|candidate| candidate.object_id) .collect::>(), @@ -1080,14 +1132,37 @@ mod tests { store.upsert_vector_records(&records).await.unwrap(); let matches = store - .search_candidates(&VectorCandidateSearch::new(vec![1.0, 0.0], 10)) + .search_candidates(&VectorCandidateSearch::new( + vec![1.0, 0.0], + 10, + vec![ObjectType::Episode], + )) .await .unwrap(); - assert_eq!(matches.len(), 1); - assert_eq!(matches[0].object_id, fixtures.episode.id); - assert_eq!(matches[0].object_type, ObjectType::Episode); - assert_eq!(matches[0].surface, VectorSurface::Summary); + assert_eq!(matches.candidates.len(), 1); + assert_eq!(matches.candidates[0].object_id, fixtures.episode.id); + assert_eq!(matches.candidates[0].object_type, ObjectType::Episode); + assert_eq!(matches.candidates[0].surface, VectorSurface::Summary); + } + + #[tokio::test] + async fn vector_fake_zero_norm_fixture_scores_every_scoped_candidate_zero() { + let store = FakeVectorCandidateStore::new(); + let (records, query) = zero_norm_vector_fixture(); + store.upsert_candidates(&records).await.unwrap(); + + let recall = store.search_candidates(&query).await.unwrap(); + + assert_eq!(recall.candidates.len(), 2); + assert!(recall + .candidates + .iter() + .all(|candidate| candidate.score == 0.0)); + assert_eq!( + recall.completeness, + VectorRecallCompleteness::Exhaustive { scanned: 2 } + ); } #[tokio::test] diff --git a/src/usecases/correct_forget.rs b/src/usecases/correct_forget.rs index 4deafa94..6f6d82a8 100644 --- a/src/usecases/correct_forget.rs +++ b/src/usecases/correct_forget.rs @@ -1300,6 +1300,7 @@ mod tests { RetrievalStatsCounter, RetrievalStatsCounterKey, RetrievalStatsEdge, RetrievalStatsHealth, RetrievalStatsObjectState, }; + use crate::ports::vector_candidate::VectorCandidateRecall; use crate::test_support::{ representative_fixtures, DeterministicMemoryEmbedder, FakeGraphAuthorityStore, FakeVectorCandidateStore, @@ -1586,13 +1587,15 @@ mod tests { assert!(first.vector_maintenance_failure.is_some()); let graph_writes_after_first = graph_write_count(&graph.calls()); let candidates_after_first = vector - .search_candidates( - &VectorCandidateSearch::new(vec![1.0, 0.0, 0.0, 0.0], 10) - .with_object_types(vec![ObjectType::DerivedMemory]), - ) + .search_candidates(&VectorCandidateSearch::new( + vec![1.0, 0.0, 0.0, 0.0], + 10, + vec![ObjectType::DerivedMemory], + )) .await .unwrap(); assert!(candidates_after_first + .candidates .iter() .any(|candidate| candidate.object_id == ids.old)); @@ -1613,16 +1616,19 @@ mod tests { assert!(maintained_ids.contains(&ids.old)); assert!(maintained_ids.contains(&replacement_id)); let candidates_after_retry = vector - .search_candidates( - &VectorCandidateSearch::new(vec![1.0, 0.0, 0.0, 0.0], 10) - .with_object_types(vec![ObjectType::DerivedMemory]), - ) + .search_candidates(&VectorCandidateSearch::new( + vec![1.0, 0.0, 0.0, 0.0], + 10, + vec![ObjectType::DerivedMemory], + )) .await .unwrap(); assert!(!candidates_after_retry + .candidates .iter() .any(|candidate| candidate.object_id == ids.old)); assert!(candidates_after_retry + .candidates .iter() .any(|candidate| candidate.object_id == replacement_id)); } @@ -3516,9 +3522,18 @@ mod tests { async fn search_candidates( &self, - _query: &VectorCandidateSearch, - ) -> Result { - Ok(CanonicalCandidates::new([])) + query: &VectorCandidateSearch, + ) -> Result { + Ok(VectorCandidateRecall { + candidates: CanonicalCandidates::new([]), + completeness: if query.limit == 0 || query.object_types.is_empty() { + crate::api::types::retrieval::VectorRecallCompleteness::NotRequested + } else { + crate::api::types::retrieval::VectorRecallCompleteness::Exhaustive { + scanned: 0, + } + }, + }) } async fn delete_candidates(&self, object_ids: &[MemoryId]) -> Result<(), CustomError> { @@ -3653,7 +3668,7 @@ mod tests { async fn search_candidates( &self, query: &VectorCandidateSearch, - ) -> Result { + ) -> Result { self.inner.search_candidates(query).await } @@ -3758,7 +3773,7 @@ mod tests { async fn search_candidates( &self, query: &VectorCandidateSearch, - ) -> Result { + ) -> Result { self.inner.search_candidates(query).await } diff --git a/src/usecases/remember.rs b/src/usecases/remember.rs index 86a48648..c4f2604e 100644 --- a/src/usecases/remember.rs +++ b/src/usecases/remember.rs @@ -338,6 +338,7 @@ mod tests { RetrievalStatsCounter, RetrievalStatsCounterKey, RetrievalStatsEdge, RetrievalStatsHealth, RetrievalStatsObjectState, RetrievalStatsStore, }; + use crate::ports::vector_candidate::VectorCandidateRecall; use crate::test_support::{representative_fixtures, FakeGraphAuthorityStore}; use crate::usecases::write_planning::RememberPlanDefaults; @@ -1230,9 +1231,18 @@ mod tests { async fn search_candidates( &self, - _query: &VectorCandidateSearch, - ) -> Result { - Ok(CanonicalCandidates::new([])) + query: &VectorCandidateSearch, + ) -> Result { + Ok(VectorCandidateRecall { + candidates: CanonicalCandidates::new([]), + completeness: if query.limit == 0 || query.object_types.is_empty() { + crate::api::types::retrieval::VectorRecallCompleteness::NotRequested + } else { + crate::api::types::retrieval::VectorRecallCompleteness::Exhaustive { + scanned: 0, + } + }, + }) } async fn delete_candidates(&self, _object_ids: &[MemoryId]) -> Result<(), CustomError> { diff --git a/src/usecases/retrieve.rs b/src/usecases/retrieve.rs index 56aafcac..8732f8a0 100644 --- a/src/usecases/retrieve.rs +++ b/src/usecases/retrieve.rs @@ -33,7 +33,7 @@ use crate::ports::graph_authority::{ GraphExpansionLifecyclePolicy, GraphExpansionQuery, TraceMode, }; use crate::ports::retrieval_stats::RetrievalStatsStore; -use crate::ports::vector_candidate::VectorCandidateStore; +use crate::ports::vector_candidate::{VectorCandidateRecall, VectorCandidateStore}; pub(crate) struct RetrievePipeline<'a, G, V, E> where @@ -85,14 +85,18 @@ where &self, context: RetrievalContext, ) -> Result { + context.validate()?; let query_embedding = self.embed_query(&context).await?; let query_embedding_dimension = query_embedding.len(); let vector_search = VectorCandidateSearch::new( query_embedding, context.candidate_limits.max_vector_candidates, - ) - .with_object_types(context.object_type_defaults.clone()); - let vector_candidates = self.vector_store.search_candidates(&vector_search).await?; + context.object_type_defaults.clone(), + ); + let VectorCandidateRecall { + candidates: vector_candidates, + completeness: vector_recall_completeness, + } = self.vector_store.search_candidates(&vector_search).await?; let trace_mode = TraceMode::from_enabled(context.include_trace); let root_selection = @@ -218,6 +222,7 @@ where configured_lifecycle_policy: context.lifecycle_policy, query_embedding_dimension, returned_vector_candidate_count: vector_candidates.len(), + vector_recall_completeness, unique_graph_root_candidate_count: root_selection.unique_count, selected_graph_root_count: candidate_roots.len(), graph_root_omission_count: root_selection.omitted_count, @@ -1356,6 +1361,7 @@ mod tests { use chrono::{DateTime, Utc}; use crate::adapters::stats::InMemoryRetrievalStatsStore; + use crate::api::types::retrieval::VectorRecallCompleteness; use crate::api::types::{ ContinuitySectionLimits, RetrievalCandidateLimits, RetrievalLifecyclePolicy, }; @@ -1647,6 +1653,38 @@ mod tests { })); } + #[tokio::test] + async fn retrieval_telemetry_preserves_every_vector_recall_completeness_verdict() { + let cases = [ + VectorRecallCompleteness::NotRequested, + VectorRecallCompleteness::Exhaustive { scanned: 7 }, + VectorRecallCompleteness::BoundaryTieClosed { fetched: 9 }, + VectorRecallCompleteness::BoundaryTieOpen { + fetched: 16, + fetch_bound: 16, + }, + ]; + + for completeness in cases { + let graph = FakeGraphAuthorityStore::new(); + let vector = RecordingVectorStore::with_completeness(Vec::new(), completeness); + let embedder = RecordingEmbedder::new(vec![1.0, 0.0]); + let outcome = RetrievePipeline::new(&graph, &vector, &embedder) + .retrieve(RetrievalContext::new("completeness telemetry")) + .await + .unwrap(); + + assert_eq!( + outcome.rationale.telemetry.vector_recall_completeness, + completeness + ); + assert_eq!( + outcome.rationale.telemetry.returned_vector_candidate_count, + 0 + ); + } + } + #[tokio::test] async fn selectivity_allows_high_selectivity_entity_about_expansion() { let fixture = high_fanout_graph_fixture(); @@ -2527,11 +2565,10 @@ mod tests { RecordingVectorStore::new(vec![candidate(object_id, ObjectType::MemoryLink, 0.99)]); let embedder = RecordingEmbedder::new(vec![1.0, 0.0]); let pipeline = RetrievePipeline::new(&graph, &vector, &embedder); + let mut context = RetrievalContext::new("propagate graph errors"); + context.object_type_defaults.push(ObjectType::MemoryLink); - let error = pipeline - .retrieve(RetrievalContext::new("propagate graph errors")) - .await - .unwrap_err(); + let error = pipeline.retrieve(context).await.unwrap_err(); assert!( matches!(error, CustomError::MemoryValidation(message) if message.contains("unsupported root")) @@ -2929,11 +2966,25 @@ mod tests { #[derive(Debug)] struct RecordingVectorStore { candidates: Vec, + completeness: Option, } impl RecordingVectorStore { fn new(candidates: Vec) -> Self { - Self { candidates } + Self { + candidates, + completeness: None, + } + } + + fn with_completeness( + candidates: Vec, + completeness: VectorRecallCompleteness, + ) -> Self { + Self { + candidates, + completeness: Some(completeness), + } } } @@ -2949,8 +3000,26 @@ mod tests { async fn search_candidates( &self, query: &VectorCandidateSearch, - ) -> Result { - Ok(CanonicalCandidates::new(self.candidates.clone()).truncated(query.limit)) + ) -> Result { + if query.limit == 0 || query.object_types.is_empty() { + return Ok(VectorCandidateRecall { + candidates: CanonicalCandidates::new([]), + completeness: VectorRecallCompleteness::NotRequested, + }); + } + let candidates = self + .candidates + .iter() + .filter(|candidate| query.object_types.contains(&candidate.object_type)) + .cloned() + .collect::>(); + let scanned = candidates.len(); + Ok(VectorCandidateRecall { + candidates: CanonicalCandidates::new(candidates).truncated(query.limit), + completeness: self + .completeness + .unwrap_or(VectorRecallCompleteness::Exhaustive { scanned }), + }) } async fn delete_candidates(&self, _object_ids: &[MemoryId]) -> Result<(), CustomError> { From b6a3edc1a1ebc0d0308b994b2a87cd1d69a40a1c Mon Sep 17 00:00:00 2001 From: ebigunso Date: Thu, 3 Sep 2026 20:40:32 +0900 Subject: [PATCH 2/3] fix: preserve vector recall completeness at backend limits --- src/adapters/qdrant/store.rs | 29 +++++++++++++++-- src/adapters/qdrant/tie_closure.rs | 50 ++++++++++++++++++++++++------ src/api/types.rs | 2 +- src/api/types/retrieval.rs | 8 +++++ src/lib.rs | 3 +- 5 files changed, 78 insertions(+), 14 deletions(-) diff --git a/src/adapters/qdrant/store.rs b/src/adapters/qdrant/store.rs index f6dab53a..52dad7b4 100644 --- a/src/adapters/qdrant/store.rs +++ b/src/adapters/qdrant/store.rs @@ -172,9 +172,10 @@ impl QdrantVectorCandidateStore { query: &VectorCandidateSearch, fetch_limit: usize, ) -> Result, CustomError> { + let backend_limit = qdrant_scroll_fetch_limit(fetch_limit)?; let request = ScrollPointsBuilder::new(&self.collection_name) .filter(qdrant_candidate_filter(query)) - .limit(fetch_limit as u32) + .limit(backend_limit) .with_payload(true) .with_vectors(false) .build(); @@ -269,7 +270,12 @@ impl VectorCandidateStore for QdrantVectorCandidateStore { } let zero_norm = query.is_zero_norm(); - let closed = close_tie_cohort(query.limit, |fetch_limit| async move { + let fetch_limit_cap = if zero_norm { + usize::try_from(u32::MAX).unwrap_or(usize::MAX) + } else { + usize::MAX + }; + let closed = close_tie_cohort(query.limit, fetch_limit_cap, |fetch_limit| async move { if zero_norm { self.scroll_zero_norm_candidate_batch(query, fetch_limit) .await @@ -465,6 +471,15 @@ fn qdrant_candidate_config(url: &str) -> QdrantConfig { .keep_alive_while_idle() } +fn qdrant_scroll_fetch_limit(fetch_limit: usize) -> Result { + u32::try_from(fetch_limit).map_err(|_| { + CustomError::DatabaseError(format!( + "Qdrant scroll limit {fetch_limit} exceeds the backend maximum {}", + u32::MAX + )) + }) +} + fn qdrant_candidate_filter(query: &VectorCandidateSearch) -> Filter { Filter::must([any_field_matches( OBJECT_TYPE_FIELD, @@ -674,6 +689,16 @@ mod tests { } } + #[test] + fn qdrant_scroll_limit_checks_backend_width_without_narrowing() { + let backend_max = usize::try_from(u32::MAX).unwrap(); + assert_eq!(qdrant_scroll_fetch_limit(backend_max).unwrap(), u32::MAX); + + if let Some(too_large) = backend_max.checked_add(1) { + assert!(qdrant_scroll_fetch_limit(too_large).is_err()); + } + } + #[test] fn qdrant_response_error_preserves_typed_transport_status() { let error = qdrant_error(QdrantError::ResponseError { diff --git a/src/adapters/qdrant/tie_closure.rs b/src/adapters/qdrant/tie_closure.rs index 751c2ffd..d3f2464b 100644 --- a/src/adapters/qdrant/tie_closure.rs +++ b/src/adapters/qdrant/tie_closure.rs @@ -46,13 +46,14 @@ impl TieClosureResult { pub(crate) async fn close_tie_cohort( admitted_limit: usize, + fetch_limit_cap: usize, mut fetch: F, ) -> Result where F: FnMut(usize) -> Fut, Fut: Future, E>>, { - let fetch_bound = tie_cohort_fetch_bound(admitted_limit); + let fetch_bound = tie_cohort_fetch_bound(admitted_limit, fetch_limit_cap); let mut fetch_limit = admitted_limit.saturating_add(1).min(fetch_bound); loop { @@ -61,11 +62,10 @@ where let candidates = CanonicalCandidates::new(fetched); match fetch_decision( - admitted_limit, fetch_limit, fetch_bound, fetched_count, - &candidates, + tie_cohort_is_closed(&candidates, admitted_limit), ) { FetchDecision::Grow(next_limit) => fetch_limit = next_limit, FetchDecision::Return => { @@ -88,20 +88,20 @@ where } } -fn tie_cohort_fetch_bound(limit: usize) -> usize { +fn tie_cohort_fetch_bound(limit: usize, fetch_limit_cap: usize) -> usize { limit .saturating_mul(TIE_COHORT_LIMIT_MULTIPLIER) .max(limit.saturating_add(TIE_COHORT_MIN_EXTRA_CANDIDATES)) + .min(fetch_limit_cap) } fn fetch_decision( - admitted_limit: usize, fetch_limit: usize, fetch_bound: usize, fetched_count: usize, - candidates: &[VectorCandidateMatch], + tie_cohort_closed: bool, ) -> FetchDecision { - if fetched_count < fetch_limit || tie_cohort_is_closed(candidates, admitted_limit) { + if fetched_count < fetch_limit || tie_cohort_closed { return FetchDecision::Return; } if fetch_limit >= fetch_bound { @@ -149,7 +149,7 @@ mod tests { CanonicalCandidates::new([candidate(1, 1.0), candidate(2, 1.0), candidate(3, 0.5)]); assert_eq!( - fetch_decision(2, 3, 10, 3, &candidates), + fetch_decision(3, 10, 3, tie_cohort_is_closed(&candidates, 2)), FetchDecision::Return ); let result = TieClosureResult { @@ -169,7 +169,7 @@ mod tests { let candidates = CanonicalCandidates::new((1..=6).rev().map(|id| candidate(id, 1.0))); assert_eq!( - fetch_decision(2, 6, 6, 6, &candidates), + fetch_decision(6, 6, 6, tie_cohort_is_closed(&candidates, 2)), FetchDecision::ReturnAtBound ); let result = TieClosureResult { @@ -189,12 +189,42 @@ mod tests { assert_eq!(result.candidates[1].object_id, Uuid::from_u128(2)); } + #[test] + fn backend_fetch_cap_reports_an_open_boundary_without_allocating_rows() { + let Ok(fetch_limit_cap) = usize::try_from(u32::MAX) else { + return; + }; + let Some(admitted_limit) = fetch_limit_cap.checked_add(1) else { + return; + }; + let fetch_bound = tie_cohort_fetch_bound(admitted_limit, fetch_limit_cap); + + assert_eq!(fetch_bound, fetch_limit_cap); + assert_eq!( + fetch_decision(fetch_limit_cap, fetch_bound, fetch_limit_cap, false), + FetchDecision::ReturnAtBound + ); + let result = TieClosureResult { + candidates: CanonicalCandidates::new([]), + fetched: fetch_limit_cap, + fetch_bound, + closure: TieClosure::OpenAtBound, + }; + assert_eq!( + result.completeness(Some(fetch_limit_cap)), + VectorRecallCompleteness::BoundaryTieOpen { + fetched: fetch_limit_cap, + fetch_bound: fetch_limit_cap, + } + ); + } + #[tokio::test] async fn closure_loop_grows_and_canonicalizes_before_returning() { let candidates = [candidate(2, 1.0), candidate(1, 1.0), candidate(3, 0.5)]; let fetch_limits = RefCell::new(Vec::new()); - let result = close_tie_cohort(1, |fetch_limit| { + let result = close_tie_cohort(1, usize::MAX, |fetch_limit| { fetch_limits.borrow_mut().push(fetch_limit); ready(Ok::<_, ()>( candidates.iter().take(fetch_limit).cloned().collect(), diff --git a/src/api/types.rs b/src/api/types.rs index 47602f1e..dac11f01 100644 --- a/src/api/types.rs +++ b/src/api/types.rs @@ -29,7 +29,7 @@ pub use retrieval::{ SectionPressureSummary, SectionScoreComponents, SectionVectorScoreSource, SelectivityCountScope, SelectivityDecision, SelectivityTelemetry, SelectivityTrace, StaleCandidateOmission, StaleCandidateOmissionSummary, StaleCandidateReason, - VectorCandidateTrace, VectorSurface, + VectorCandidateTrace, VectorRecallCompleteness, VectorSurface, }; pub use write_plan::{ CandidateCount, CandidateProducerKind, CandidateProvenance, CandidateRationale, CommitOptions, diff --git a/src/api/types/retrieval.rs b/src/api/types/retrieval.rs index a987d1d0..8549d367 100644 --- a/src/api/types/retrieval.rs +++ b/src/api/types/retrieval.rs @@ -278,6 +278,14 @@ pub struct RetrievalTelemetry { pub section_pressure: Vec, } +/// Completeness of the vector candidate set reported for a retrieval. +/// +/// ``` +/// use character_memory::api::types::VectorRecallCompleteness as ApiCompleteness; +/// use character_memory::VectorRecallCompleteness; +/// +/// let _: VectorRecallCompleteness = ApiCompleteness::NotRequested; +/// ``` #[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum VectorRecallCompleteness { diff --git a/src/lib.rs b/src/lib.rs index d52bd4cc..5222a1a9 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -41,7 +41,8 @@ pub use crate::api::types::{ StaleCandidateOmissionSummary, StaleCandidateReason, StatsUpdateCandidate, StatsUpdateFailure, StatsUpdateStatus, SupersededByEvidence, SuppressionPolicy, VectorCandidateTrace, VectorIndexCandidate, VectorIndexingFailure, VectorMaintenanceFailure, - VectorMaintenanceFailureItem, VectorMaintenanceOperation, VectorSurface, + VectorMaintenanceFailureItem, VectorMaintenanceOperation, VectorRecallCompleteness, + VectorSurface, }; pub use crate::config::{ GraphStoreMode, RetrievalStatsHealthFailMode, RetrievalStatsStoreMode, Settings, From acfdc08a697c8706a3d79a60bfd78702171025d6 Mon Sep 17 00:00:00 2001 From: ebigunso Date: Fri, 4 Sep 2026 01:23:34 +0900 Subject: [PATCH 3/3] fix: validate zero-norm query dimensions --- src/adapters/qdrant/store.rs | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/src/adapters/qdrant/store.rs b/src/adapters/qdrant/store.rs index 52dad7b4..9554b349 100644 --- a/src/adapters/qdrant/store.rs +++ b/src/adapters/qdrant/store.rs @@ -269,6 +269,18 @@ impl VectorCandidateStore for QdrantVectorCandidateStore { }); } + let actual_vector_size = u64::try_from(query.query_embedding.len()).unwrap_or(u64::MAX); + if actual_vector_size != self.vector_size { + return Err(CollectionCompatibilityError { + collection: self.collection_name.clone(), + mismatch: CollectionMismatch::VectorSize { + expected: self.vector_size, + actual: actual_vector_size, + }, + } + .into()); + } + let zero_norm = query.is_zero_norm(); let fetch_limit_cap = if zero_norm { usize::try_from(u32::MAX).unwrap_or(usize::MAX) @@ -689,6 +701,26 @@ mod tests { } } + #[tokio::test] + async fn wrong_dimension_zero_norm_query_fails_before_contacting_qdrant() { + let store = + QdrantVectorCandidateStore::new("http://127.0.0.1:1", "not_contacted", 2).unwrap(); + let query = VectorCandidateSearch::new(vec![0.0], 10, vec![ObjectType::Episode]); + + let error = store.search_candidates(&query).await.unwrap_err(); + + assert!(matches!( + error, + CustomError::CollectionIncompatible(CollectionCompatibilityError { + collection, + mismatch: CollectionMismatch::VectorSize { + expected: 2, + actual: 1, + }, + }) if collection == "not_contacted" + )); + } + #[test] fn qdrant_scroll_limit_checks_backend_width_without_narrowing() { let backend_max = usize::try_from(u32::MAX).unwrap();