From 57f3e23c939b8e5d308c9789d5ccc7eeaa0e0eea Mon Sep 17 00:00:00 2001 From: ebigunso Date: Mon, 21 Sep 2026 08:20:11 +0900 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20Preserve=20typed=20memory=20iden?= =?UTF-8?q?tity=20in=20vector=20writes=20and=20deletes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .github/workflows/pr_validation.yaml | 3 + src/adapters/qdrant/payload.rs | 5 +- src/adapters/qdrant/store.rs | 75 ++++++++++++++++------ src/adapters/qdrant_edge/mod.rs | 63 +++++++++++------- src/memory.rs | 96 +++++++++++++++++++++++++++- src/ports/vector_candidate.rs | 9 +-- src/test_support.rs | 10 +-- src/usecases/correct_forget.rs | 12 ++-- src/usecases/remember.rs | 2 +- src/usecases/retrieve.rs | 4 +- src/usecases/vector_indexing.rs | 6 +- 11 files changed, 217 insertions(+), 68 deletions(-) diff --git a/.github/workflows/pr_validation.yaml b/.github/workflows/pr_validation.yaml index bdc98884..080d936e 100644 --- a/.github/workflows/pr_validation.yaml +++ b/.github/workflows/pr_validation.yaml @@ -133,3 +133,6 @@ jobs: - name: Run service parity tests run: cargo test --verbose --test vector_port_contract_tests service_and_embedded_ + + - name: Verify service deletes preserve other typed objects + run: cargo test --verbose --lib adapters::qdrant::store::tests::port_contract::service_delete_removes_every_surface_of_only_the_requested_object -- --exact diff --git a/src/adapters/qdrant/payload.rs b/src/adapters/qdrant/payload.rs index 29a7d7ba..7c9b3f61 100644 --- a/src/adapters/qdrant/payload.rs +++ b/src/adapters/qdrant/payload.rs @@ -109,7 +109,10 @@ pub(crate) const OBJECT_ID_FIELD: &str = QdrantPayloadField::ObjectId.name(); pub(crate) const OBJECT_TYPE_FIELD: &str = QdrantPayloadField::ObjectType.name(); pub(crate) fn qdrant_point_id(record: &VectorRecord) -> MemoryId { - MemoryId::new_v5(&record.object_id, record.surface.to_string().as_bytes()) + MemoryId::new_v5( + &record.object_id, + format!("{}:{}", record.object_type, record.surface).as_bytes(), + ) } pub(crate) fn read_candidate_match<'a>( diff --git a/src/adapters/qdrant/store.rs b/src/adapters/qdrant/store.rs index ffa3f5fc..252333d4 100644 --- a/src/adapters/qdrant/store.rs +++ b/src/adapters/qdrant/store.rs @@ -12,7 +12,7 @@ use qdrant_client::qdrant::{ }; use qdrant_client::{config::QdrantConfig, Qdrant, QdrantError}; -use crate::domain::MemoryId; +use crate::domain::MemoryObjectRef; use crate::errors::{ CollectionCompatibilityError, CollectionMismatch, CustomError, IoErrorKind, TransportStatus, VectorDatabaseError, VectorDatabaseErrorKind, @@ -316,14 +316,19 @@ impl VectorCandidateStore for QdrantVectorCandidateStore { }) } - async fn delete_candidates(&self, object_ids: &[MemoryId]) -> Result<(), CustomError> { - if object_ids.is_empty() { + async fn delete_candidates(&self, objects: &[MemoryObjectRef]) -> Result<(), CustomError> { + if objects.is_empty() { return Ok(()); } - let conditions: Vec<_> = object_ids + let conditions: Vec<_> = objects .iter() - .map(|id| Condition::matches(OBJECT_ID_FIELD, id.to_string())) + .map(|object| { + Condition::from(Filter::must([ + Condition::matches(OBJECT_ID_FIELD, object.id.to_string()), + Condition::matches(OBJECT_TYPE_FIELD, object.object_type.to_string()), + ])) + }) .collect(); let selector = PointsSelectorOneOf::Filter(Filter::should(conditions)); let request = DeletePointsBuilder::new(&self.collection_name) @@ -574,7 +579,7 @@ mod tests { use super::*; use crate::adapters::qdrant::payload::QdrantPayloadField; use crate::api::types::retrieval::VectorRecallCompleteness; - use crate::domain::{ObjectType, VectorSurface, DEFAULT_SCHEMA_VERSION}; + use crate::domain::{MemoryId, ObjectType, VectorSurface, DEFAULT_SCHEMA_VERSION}; use crate::models::vector::{VectorRecord, VectorRecordEmbedding}; use qdrant_client::qdrant::{ point_id::PointIdOptions, value::Kind, vector, vectors, DeleteCollectionBuilder, PointId, @@ -705,14 +710,16 @@ mod tests { store: &dyn VectorCandidateStore, ) -> Result<(VectorCandidateRecall, VectorCandidateRecall), CustomError> { let records = [ - (MemoryId::from_u128(1), VectorSurface::Summary), - (MemoryId::from_u128(1), VectorSurface::Text), - (MemoryId::from_u128(2), VectorSurface::Summary), + (1, ObjectType::Episode, VectorSurface::Summary), + (1, ObjectType::Episode, VectorSurface::Text), + (2, ObjectType::Episode, VectorSurface::Summary), + (1, ObjectType::MemoryThread, VectorSurface::Summary), + (2, ObjectType::MemoryThread, VectorSurface::Summary), ] - .map(|(id, surface)| { + .map(|(id, object_type, surface)| { VectorRecord::new( - id, - ObjectType::Episode, + MemoryId::from_u128(id), + object_type, surface, DEFAULT_SCHEMA_VERSION, "surface", @@ -723,9 +730,18 @@ mod tests { .map(|record| VectorRecordEmbedding::new(record, &[1.0, 0.0])) .collect::>(); store.upsert_vector_records(&embeddings).await?; - let query = VectorCandidateSearch::new(vec![1.0, 0.0], 10, vec![ObjectType::Episode]); + let query = VectorCandidateSearch::new( + vec![1.0, 0.0], + 10, + vec![ObjectType::Episode, ObjectType::MemoryThread], + ); let before = store.search_candidates(&query).await?; - store.delete_candidates(&[MemoryId::from_u128(1)]).await?; + store + .delete_candidates(&[ + MemoryObjectRef::new(ObjectType::Episode, MemoryId::from_u128(1)), + MemoryObjectRef::new(ObjectType::MemoryThread, MemoryId::from_u128(2)), + ]) + .await?; let after = store.search_candidates(&query).await?; Ok((before, after)) } @@ -737,20 +753,37 @@ mod tests { recall .candidates .iter() - .map(|candidate| (candidate.object_id, candidate.surface)) + .map(|candidate| { + ( + candidate.object_id, + candidate.object_type, + candidate.surface, + ) + }) .collect::>() }; assert_eq!( surfaces(&before), - HashSet::from([ - (MemoryId::from_u128(1), VectorSurface::Summary), - (MemoryId::from_u128(1), VectorSurface::Text), - (MemoryId::from_u128(2), VectorSurface::Summary), - ]) + [ + (1, ObjectType::Episode, VectorSurface::Summary), + (1, ObjectType::Episode, VectorSurface::Text), + (2, ObjectType::Episode, VectorSurface::Summary), + (1, ObjectType::MemoryThread, VectorSurface::Summary), + (2, ObjectType::MemoryThread, VectorSurface::Summary), + ] + .map(|(id, object_type, surface)| (MemoryId::from_u128(id), object_type, surface)) + .into_iter() + .collect() ); assert_eq!( surfaces(&after), - HashSet::from([(MemoryId::from_u128(2), VectorSurface::Summary)]) + [ + (2, ObjectType::Episode, VectorSurface::Summary), + (1, ObjectType::MemoryThread, VectorSurface::Summary), + ] + .map(|(id, object_type, surface)| (MemoryId::from_u128(id), object_type, surface)) + .into_iter() + .collect() ); } } diff --git a/src/adapters/qdrant_edge/mod.rs b/src/adapters/qdrant_edge/mod.rs index 70390dff..f6aab4af 100644 --- a/src/adapters/qdrant_edge/mod.rs +++ b/src/adapters/qdrant_edge/mod.rs @@ -8,10 +8,10 @@ use std::time::Duration; use async_trait::async_trait; use qdrant_edge::{ Condition, CountRequest, CreateIndex, Distance, EdgeConfig, EdgeOptimizersConfig, EdgeShard, - EdgeVectorParams, FieldCondition, FieldIndexOperations, Filter, Match, MatchValue, NamedQuery, - PayloadFieldSchema, PayloadSchemaType, PointInsertOperations, PointOperations, PointStruct, - QueryEnum, ScoredPoint, ScrollRequest, SearchParams, SearchRequestBuilder, UpdateOperation, - ValueVariants, WithPayloadInterface, DEFAULT_VECTOR_NAME, + EdgeVectorParams, FieldCondition, FieldIndexOperations, Filter, NamedQuery, PayloadFieldSchema, + PayloadSchemaType, PointInsertOperations, PointOperations, PointStruct, QueryEnum, ScoredPoint, + ScrollRequest, SearchParams, SearchRequestBuilder, UpdateOperation, WithPayloadInterface, + DEFAULT_VECTOR_NAME, }; use serde::{Deserialize, Serialize}; use tokio::sync::{oneshot, Mutex}; @@ -21,7 +21,7 @@ use crate::adapters::qdrant::payload::{ QdrantPayloadSchema, OBJECT_ID_FIELD, OBJECT_TYPE_FIELD, }; use crate::adapters::qdrant::tie_closure::close_tie_cohort; -use crate::domain::{MemoryId, DEFAULT_SCHEMA_VERSION}; +use crate::domain::{MemoryObjectRef, DEFAULT_SCHEMA_VERSION}; use crate::errors::{ CollectionCompatibilityError, CollectionMismatch, ConfigValidationError, ConfigValidationReason, CustomError, IoErrorKind, VectorDatabaseError, VectorDatabaseErrorKind, @@ -59,7 +59,7 @@ enum Command { reply: Reply<()>, }, Delete { - object_ids: Vec, + filter: Filter, reply: Reply<()>, }, Search { @@ -263,14 +263,29 @@ impl VectorCandidateStore for QdrantEdgeVectorCandidateStore { }) } - async fn delete_candidates(&self, object_ids: &[MemoryId]) -> Result<(), CustomError> { - if object_ids.is_empty() { + async fn delete_candidates(&self, objects: &[MemoryObjectRef]) -> Result<(), CustomError> { + if objects.is_empty() { return Ok(()); } let _operation = self.operation.lock().await; let (reply, receiver) = oneshot::channel(); + let conditions = objects + .iter() + .map(|object| { + Condition::Filter(Filter { + must: Some(vec![ + string_condition(OBJECT_ID_FIELD, object.id.to_string()), + string_condition(OBJECT_TYPE_FIELD, object.object_type.to_string()), + ]), + ..Filter::new() + }) + }) + .collect(); self.send(Command::Delete { - object_ids: object_ids.iter().map(ToString::to_string).collect(), + filter: Filter { + should: Some(conditions), + ..Filter::new() + }, reply, })?; receive(receiver).await @@ -319,8 +334,7 @@ fn owner_loop(shard: EdgeShard, commands: mpsc::Receiver) { .and_then(|_| shard.flush().map_err(edge_error)); let _ = reply.send(result); } - Command::Delete { object_ids, reply } => { - let filter = string_filter(OBJECT_ID_FIELD, object_ids); + Command::Delete { filter, reply } => { let result = shard .update(UpdateOperation::PointOperation( PointOperations::DeletePointsByFilter(filter), @@ -586,16 +600,7 @@ fn payload_to_match( fn string_filter(field: &str, values: Vec) -> Filter { let conditions = values .into_iter() - .map(|value| { - Condition::Field(FieldCondition::new_match( - field - .try_into() - .expect("adapter fields are valid JSON paths"), - Match::Value(MatchValue { - value: ValueVariants::String(value), - }), - )) - }) + .map(|value| string_condition(field, value)) .collect(); Filter { should: Some(conditions), @@ -603,6 +608,15 @@ fn string_filter(field: &str, values: Vec) -> Filter { } } +fn string_condition(field: &str, value: String) -> Condition { + Condition::Field(FieldCondition::new_match( + field + .try_into() + .expect("adapter fields are valid JSON paths"), + value.into(), + )) +} + fn object_type_tokens(query: &VectorCandidateSearch) -> Vec { query.object_types.iter().map(ToString::to_string).collect() } @@ -738,7 +752,7 @@ mod tests { use super::*; use crate::api::types::retrieval::VectorRecallCompleteness; - use crate::domain::{ObjectType, VectorSurface}; + use crate::domain::{MemoryId, ObjectType, VectorSurface}; use crate::models::vector::VectorRecord; const CHILD_MODE: &str = "CM_QDRANT_EDGE_CHILD_MODE"; @@ -889,7 +903,10 @@ mod tests { assert_eq!(store.search_candidates(&query(3)).await.unwrap(), first); store - .delete_candidates(&[records[0].object_id]) + .delete_candidates(&[MemoryObjectRef::new( + records[0].object_type, + records[0].object_id, + )]) .await .unwrap(); store.close().await.unwrap(); diff --git a/src/memory.rs b/src/memory.rs index c8fac056..acc66d47 100644 --- a/src/memory.rs +++ b/src/memory.rs @@ -732,6 +732,100 @@ mod tests { memory.close().await.unwrap(); } + #[tokio::test] + async fn episode_and_thread_with_the_same_id_keep_distinct_vectors_on_write_and_forget() { + let memory = injected_memory().await; + let shared_id = MemoryId::from_u128(1901); + let episode_text = "A heron crossed the lake at dawn."; + let thread_text = "Planning the next observatory visit."; + let mut episode = EpisodeDraft::new(episode_text); + episode.id = Some(shared_id); + let mut thread = MemoryThreadDraft::new("Observatory visit", thread_text); + thread.id = Some(shared_id); + let plan = memory + .prepare( + RememberInput::new(episode_text) + .with_episode(episode) + .with_memory_thread(thread), + PrepareOptions::default(), + ) + .await + .unwrap(); + let committed = memory.commit(plan, CommitOptions::default()).await.unwrap(); + assert!(committed.vector_indexing_failure.is_none()); + let mut recalled = Vec::new(); + for (object_type, text) in [ + (ObjectType::Episode, episode_text), + (ObjectType::MemoryThread, thread_text), + ] { + let mut context = RetrievalContext::new(text).with_trace(); + context.object_type_defaults = vec![object_type]; + context.graph_limits.max_depth = 0; + let outcome = memory.retrieve(context).await.unwrap(); + let candidates = outcome + .trace + .unwrap() + .vector_candidates + .into_iter() + .map(|candidate| candidate.object) + .collect::>(); + let summaries = if object_type == ObjectType::Episode { + outcome + .pack + .relevant_episodes + .into_iter() + .map(|episode| (episode.id, episode.summary)) + .collect::>() + } else { + outcome + .pack + .active_threads + .into_iter() + .map(|thread| (thread.id, thread.summary)) + .collect::>() + }; + recalled.push((candidates, summaries)); + } + assert_eq!( + recalled, + vec![ + ( + vec![MemoryObjectRef::new(ObjectType::Episode, shared_id)], + vec![(shared_id, episode_text.to_owned())] + ), + ( + vec![MemoryObjectRef::new(ObjectType::MemoryThread, shared_id)], + vec![(shared_id, thread_text.to_owned())] + ), + ] + ); + let forgotten = memory + .forget(ForgetMemoryDraft::suppress( + LifecycleTargetRef::episode(shared_id), + "Forget the episode, preserving the separate thread.", + )) + .await + .unwrap(); + assert!(forgotten.vector_maintenance_failure.is_none()); + let mut context = RetrievalContext::new(thread_text).with_trace(); + context.object_type_defaults = vec![ObjectType::MemoryThread]; + context.graph_limits.max_depth = 0; + let outcome = memory.retrieve(context).await.unwrap(); + memory.close().await.unwrap(); + assert_eq!( + outcome + .trace + .unwrap() + .vector_candidates + .into_iter() + .map(|candidate| candidate.object) + .collect::>(), + vec![MemoryObjectRef::new(ObjectType::MemoryThread, shared_id)] + ); + assert_eq!(outcome.pack.active_threads.len(), 1); + assert_eq!(outcome.pack.active_threads[0].summary, thread_text); + } + #[tokio::test] async fn retrieve_rejects_an_empty_configured_object_type_scope_at_the_boundary() { let memory = injected_memory().await; @@ -1258,7 +1352,7 @@ mod tests { }) } - async fn delete_candidates(&self, _object_ids: &[MemoryId]) -> Result<(), CustomError> { + async fn delete_candidates(&self, _objects: &[MemoryObjectRef]) -> Result<(), CustomError> { Ok(()) } } diff --git a/src/ports/vector_candidate.rs b/src/ports/vector_candidate.rs index 2fb77ef8..beb93347 100644 --- a/src/ports/vector_candidate.rs +++ b/src/ports/vector_candidate.rs @@ -2,7 +2,7 @@ use async_trait::async_trait; use crate::api::types::retrieval::VectorRecallCompleteness; -use crate::domain::MemoryId; +use crate::domain::MemoryObjectRef; use crate::errors::CustomError; use crate::models::vector::{CanonicalCandidates, VectorCandidateSearch, VectorRecordEmbedding}; @@ -38,7 +38,8 @@ pub(crate) trait VectorCandidateStore: Send + Sync { query: &VectorCandidateSearch, ) -> Result; - async fn delete_candidates(&self, object_ids: &[MemoryId]) -> Result<(), CustomError>; + /// Deletes every surface of each typed object, preserving other object types with the same ID. + async fn delete_candidates(&self, objects: &[MemoryObjectRef]) -> Result<(), CustomError>; } #[async_trait] @@ -61,7 +62,7 @@ impl VectorCandidateStore for Box { (**self).search_candidates(query).await } - async fn delete_candidates(&self, object_ids: &[MemoryId]) -> Result<(), CustomError> { - (**self).delete_candidates(object_ids).await + async fn delete_candidates(&self, objects: &[MemoryObjectRef]) -> Result<(), CustomError> { + (**self).delete_candidates(objects).await } } diff --git a/src/test_support.rs b/src/test_support.rs index 35b080d2..3cc73f90 100644 --- a/src/test_support.rs +++ b/src/test_support.rs @@ -7,9 +7,9 @@ use uuid::Uuid; use crate::adapters::oxigraph::OxigraphGraphAuthorityStore; use crate::adapters::qdrant_edge::QdrantEdgeVectorCandidateStore; use crate::domain::{ - DerivedMemory, DerivedType, Entity, Episode, MemoryId, MemoryLink, MemoryObject, MemoryThread, - Modality, ObjectType, Observation, RelationType, RetentionState, ThreadStatus, - DEFAULT_SCHEMA_VERSION, + DerivedMemory, DerivedType, Entity, Episode, MemoryId, MemoryLink, MemoryObject, + MemoryObjectRef, MemoryThread, Modality, ObjectType, Observation, RelationType, RetentionState, + ThreadStatus, DEFAULT_SCHEMA_VERSION, }; use crate::errors::CustomError; use crate::models::vector::{ @@ -98,8 +98,8 @@ impl VectorCandidateStore for TemporaryVectorCandidateStore { self.store().search_candidates(query).await } - async fn delete_candidates(&self, object_ids: &[MemoryId]) -> Result<(), CustomError> { - self.store().delete_candidates(object_ids).await + async fn delete_candidates(&self, objects: &[MemoryObjectRef]) -> Result<(), CustomError> { + self.store().delete_candidates(objects).await } } diff --git a/src/usecases/correct_forget.rs b/src/usecases/correct_forget.rs index 3124b44c..d10488e1 100644 --- a/src/usecases/correct_forget.rs +++ b/src/usecases/correct_forget.rs @@ -3301,7 +3301,7 @@ mod tests { GraphLinks(Vec<(MemoryId, MemoryId)>), EmbedBatch(Vec), VectorUpsert(Vec), - VectorDelete(Vec), + VectorDelete(Vec), StatsEdges(usize), StatsObjectStates(usize), StatsUnhealthy, @@ -3498,8 +3498,8 @@ mod tests { }) } - async fn delete_candidates(&self, object_ids: &[MemoryId]) -> Result<(), CustomError> { - lock(&self.calls).push(StoreCall::VectorDelete(object_ids.to_vec())); + async fn delete_candidates(&self, objects: &[MemoryObjectRef]) -> Result<(), CustomError> { + lock(&self.calls).push(StoreCall::VectorDelete(objects.to_vec())); if self.fail_delete { return Err(CustomError::VectorDatabaseError(VectorDatabaseError::new( "test", @@ -3628,7 +3628,7 @@ mod tests { self.inner.search_candidates(query).await } - async fn delete_candidates(&self, object_ids: &[MemoryId]) -> Result<(), CustomError> { + async fn delete_candidates(&self, objects: &[MemoryObjectRef]) -> Result<(), CustomError> { if std::mem::take(&mut *lock(&self.fail_next_delete)) { return Err(CustomError::VectorDatabaseError(VectorDatabaseError::new( "test", @@ -3637,7 +3637,7 @@ mod tests { "one-shot vector delete failed", ))); } - self.inner.delete_candidates(object_ids).await + self.inner.delete_candidates(objects).await } } @@ -3733,7 +3733,7 @@ mod tests { self.inner.search_candidates(query).await } - async fn delete_candidates(&self, _object_ids: &[MemoryId]) -> Result<(), CustomError> { + async fn delete_candidates(&self, _objects: &[MemoryObjectRef]) -> Result<(), CustomError> { Err(CustomError::VectorDatabaseError(VectorDatabaseError::new( "test", VectorDatabaseErrorKind::Response, diff --git a/src/usecases/remember.rs b/src/usecases/remember.rs index 7c51d126..f5529b0a 100644 --- a/src/usecases/remember.rs +++ b/src/usecases/remember.rs @@ -1284,7 +1284,7 @@ mod tests { }) } - async fn delete_candidates(&self, _object_ids: &[MemoryId]) -> Result<(), CustomError> { + async fn delete_candidates(&self, _objects: &[MemoryObjectRef]) -> Result<(), CustomError> { Ok(()) } } diff --git a/src/usecases/retrieve.rs b/src/usecases/retrieve.rs index fe4047e9..1267e753 100644 --- a/src/usecases/retrieve.rs +++ b/src/usecases/retrieve.rs @@ -3300,8 +3300,8 @@ mod tests { Ok(recall) } - async fn delete_candidates(&self, object_ids: &[MemoryId]) -> Result<(), CustomError> { - self.inner.delete_candidates(object_ids).await + async fn delete_candidates(&self, objects: &[MemoryObjectRef]) -> Result<(), CustomError> { + self.inner.delete_candidates(objects).await } } diff --git a/src/usecases/vector_indexing.rs b/src/usecases/vector_indexing.rs index 8af7329e..aafc3aee 100644 --- a/src/usecases/vector_indexing.rs +++ b/src/usecases/vector_indexing.rs @@ -138,8 +138,7 @@ pub(crate) async fn delete_vectors( if objects.is_empty() { return Ok(None); } - let ids = objects.iter().map(|object| object.id).collect::>(); - match vector_store.delete_candidates(&ids).await { + match vector_store.delete_candidates(objects).await { Ok(()) => Ok(None), Err(CustomError::VectorDatabaseError(error)) => { Ok(Some(crate::api::types::VectorMaintenanceFailureItem { @@ -167,7 +166,6 @@ mod tests { use super::*; use async_trait::async_trait; - use crate::domain::MemoryId; use crate::models::vector::{zero_norm_record_fixture, EmbeddingInput, VectorCandidateSearch}; use crate::ports::vector_candidate::VectorCandidateRecall; @@ -205,7 +203,7 @@ mod tests { unreachable!("search is not part of this test") } - async fn delete_candidates(&self, _object_ids: &[MemoryId]) -> Result<(), CustomError> { + async fn delete_candidates(&self, _objects: &[MemoryObjectRef]) -> Result<(), CustomError> { unreachable!("deletion is not part of this test") } }