From 74bc7e51e3bb9bdbd96fcbb95a962050e5d6e5c1 Mon Sep 17 00:00:00 2001 From: ebigunso Date: Mon, 21 Sep 2026 05:35:44 +0900 Subject: [PATCH 01/13] =?UTF-8?q?=F0=9F=90=9B=20Report=20stats=20projectio?= =?UTF-8?q?n=20graph=20reads=20accurately?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 2 +- docs/design/database/graph_schema_design.md | 2 ++ src/errors.rs | 16 +++++++--------- src/usecases/remember.rs | 21 +++++++++++++++++++-- src/usecases/stats_projection.rs | 10 +++++----- tests/write_planning_tests.rs | 2 +- 6 files changed, 35 insertions(+), 18 deletions(-) diff --git a/README.md b/README.md index bc3797e4..423b4476 100644 --- a/README.md +++ b/README.md @@ -149,7 +149,7 @@ if validation.iter().all(|candidate| candidate.status == CandidateValidationStat } ``` -`commit` revalidates the plan before writing. Graph-authoritative objects, links, provenance, lifecycle, and currentness are critical writes; vector indexing and retrieval-stat updates are repairable and are reported in `RememberOutcome`. +`commit` revalidates the plan before writing. Graph-authoritative objects, links, provenance, lifecycle, and currentness are critical writes; vector indexing and retrieval-stat updates are repairable and are reported in `RememberOutcome`. Failed graph reads feeding the stats projection, including endpoint hydration and currency lookup, report `StatsUpdateCause::GraphRead`; the nested `GraphQueryError` retains the underlying failure. For callers that want the standard write lifecycle in one call, `remember(RememberInput, RememberOptions)` composes `prepare`, `validate_plan`, and `commit` over the same graph-authoritative machinery. diff --git a/docs/design/database/graph_schema_design.md b/docs/design/database/graph_schema_design.md index aea1223c..650a42cb 100644 --- a/docs/design/database/graph_schema_design.md +++ b/docs/design/database/graph_schema_design.md @@ -100,6 +100,8 @@ The policy combination that includes suppressed memories while excluding superse Graph writes are critical. Vector and stats updates are repairable parts of the write outcome, with typed failures and affected IDs. A graph-only memory loses semantic recall until vector indexing is repaired; stale candidates are checked against graph existence and lifecycle before inclusion. Replaying an old plan does not re-index an interpreted memory that graph authority identifies as superseded. +A failed graph read feeding the stats projection is reported as `StatsUpdateCause::GraphRead` and retained in unhealthy stats as `RetrievalStatsHealthCause::GraphRead`. Both carry the underlying `GraphQueryError`, whether the read was endpoint hydration or currency lookup. + The library has no cross-store reconciliation or stats-rebuild operation. Stats health remains unhealthy after an internal failure; recovery requires a fresh store and caller-managed replay. Raw-source resolution and cross-store census operations belong to callers or operators. See the [schema cheat sheet](schema_cheat_sheet.md) for predicate tables and the [vector payload contract](vector_payload_design.md) for recall record fields. diff --git a/src/errors.rs b/src/errors.rs index d36c7dbd..f1eac546 100644 --- a/src/errors.rs +++ b/src/errors.rs @@ -224,8 +224,8 @@ pub enum VectorIndexingCause { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Error)] #[serde(tag = "cause", rename_all = "snake_case")] pub enum StatsUpdateCause { - #[error("stats endpoint hydration failed: {error}")] - EndpointHydration { error: GraphQueryError }, + #[error("graph read for stats projection failed: {error}")] + GraphRead { error: GraphQueryError }, #[error("stats edge write failed: {error}")] EdgeWrite { error: RetrievalStatsStoreError }, #[error("stats object-state write failed: {error}")] @@ -274,8 +274,8 @@ pub enum RetrievalStatsStoreError { pub enum RetrievalStatsHealthCause { #[error("stats store initialization failed: {error}")] StoreInitialization { error: RetrievalStatsStoreError }, - #[error("stats endpoint hydration failed: {error}")] - EndpointHydration { error: GraphQueryError }, + #[error("graph read for stats projection failed: {error}")] + GraphRead { error: GraphQueryError }, #[error("stats edge write failed: {error}")] EdgeWrite { error: RetrievalStatsStoreError }, #[error("stats object-state write failed: {error}")] @@ -291,11 +291,9 @@ pub enum RetrievalStatsHealthCause { impl StatsUpdateCause { pub(crate) fn health_cause(&self) -> Option { match self { - Self::EndpointHydration { error } => { - Some(RetrievalStatsHealthCause::EndpointHydration { - error: error.clone(), - }) - } + Self::GraphRead { error } => Some(RetrievalStatsHealthCause::GraphRead { + error: error.clone(), + }), Self::EdgeWrite { error } => Some(RetrievalStatsHealthCause::EdgeWrite { error: error.clone(), }), diff --git a/src/usecases/remember.rs b/src/usecases/remember.rs index 7c51d126..2941646a 100644 --- a/src/usecases/remember.rs +++ b/src/usecases/remember.rs @@ -419,13 +419,30 @@ mod tests { outcome .stats_update_status .failure + .as_ref() .unwrap() .causes .as_slice(), - [StatsUpdateCause::EndpointHydration { + [StatsUpdateCause::GraphRead { error: crate::errors::GraphQueryError::Selection { .. } }] )); + assert_eq!( + serde_json::to_value(&outcome.stats_update_status.failure.as_ref().unwrap().causes[0]) + .unwrap()["cause"], + "graph_read" + ); + let health_cause = stats.health().await.unwrap().last_error_cause.unwrap(); + assert!(matches!( + health_cause, + RetrievalStatsHealthCause::GraphRead { + error: crate::errors::GraphQueryError::Selection { .. } + } + )); + assert_eq!( + serde_json::to_value(&health_cause).unwrap()["operation"], + "graph_read" + ); assert!(outcome .repair_needed .iter() @@ -862,7 +879,7 @@ mod tests { assert!(matches!( failure.causes.as_slice(), [ - StatsUpdateCause::EndpointHydration { .. }, + StatsUpdateCause::GraphRead { .. }, StatsUpdateCause::EdgeWrite { .. } ] )); diff --git a/src/usecases/stats_projection.rs b/src/usecases/stats_projection.rs index 2717c2ca..617947b2 100644 --- a/src/usecases/stats_projection.rs +++ b/src/usecases/stats_projection.rs @@ -71,7 +71,7 @@ where let causes = if missing_endpoint_refs.is_empty() { Vec::new() } else { - vec![StatsUpdateCause::EndpointHydration { + vec![StatsUpdateCause::GraphRead { error: GraphQueryError::Hydration { detail: format_missing_endpoint_refs(&missing_endpoint_refs), }, @@ -84,7 +84,7 @@ where } Err(error) => ( objects.to_vec(), - vec![StatsUpdateCause::EndpointHydration { error }], + vec![StatsUpdateCause::GraphRead { error }], ), } }; @@ -138,7 +138,7 @@ where .await { Ok(ids) => ids.into_iter().collect(), - Err(error) => return vec![StatsUpdateCause::EndpointHydration { error }], + Err(error) => return vec![StatsUpdateCause::GraphRead { error }], } }; let states = retrieval_stats_object_states(objects, &superseded); @@ -509,13 +509,13 @@ mod tests { ); assert!(matches!( failure.causes.as_slice(), - [StatsUpdateCause::EndpointHydration { + [StatsUpdateCause::GraphRead { error: GraphQueryError::Hydration { .. }, }] )); assert!(matches!( stats_store.marked_causes.lock().unwrap().as_slice(), - [RetrievalStatsHealthCause::EndpointHydration { + [RetrievalStatsHealthCause::GraphRead { error: GraphQueryError::Hydration { .. }, }] )); diff --git a/tests/write_planning_tests.rs b/tests/write_planning_tests.rs index 3b370b00..fce75563 100644 --- a/tests/write_planning_tests.rs +++ b/tests/write_planning_tests.rs @@ -317,7 +317,7 @@ async fn authority_split_outcome_fields_are_coherent_on_healthy_commit() { assert!(matches!( failure.causes.as_slice(), [ - StatsUpdateCause::EndpointHydration { .. } + StatsUpdateCause::GraphRead { .. } | StatsUpdateCause::EdgeWrite { .. } | StatsUpdateCause::ObjectStateWrite { .. } | StatsUpdateCause::HealthCheck { .. } From 57f3e23c939b8e5d308c9789d5ccc7eeaa0e0eea Mon Sep 17 00:00:00 2001 From: ebigunso Date: Mon, 21 Sep 2026 08:20:11 +0900 Subject: [PATCH 02/13] =?UTF-8?q?=F0=9F=90=9B=20Preserve=20typed=20memory?= =?UTF-8?q?=20identity=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") } } From 9bb6ce64368ff53c6f6ea68cfa2a5ee6472898be Mon Sep 17 00:00:00 2001 From: ebigunso Date: Wed, 23 Sep 2026 05:34:46 +0900 Subject: [PATCH 03/13] =?UTF-8?q?=E2=9C=85=20Cover=20typed=20identity=20an?= =?UTF-8?q?d=20deletion=20across=20scene=20surfaces?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/adapters/qdrant/payload.rs | 24 ++++++++++- src/adapters/qdrant/store.rs | 16 ++++++- src/memory/scene_tests.rs | 76 ++++++++++++++++++++++++++++++++++ 3 files changed, 114 insertions(+), 2 deletions(-) diff --git a/src/adapters/qdrant/payload.rs b/src/adapters/qdrant/payload.rs index 913413f7..fdd990dd 100644 --- a/src/adapters/qdrant/payload.rs +++ b/src/adapters/qdrant/payload.rs @@ -305,7 +305,7 @@ mod tests { } #[test] - fn point_identity_is_stable_per_object_and_distinct_per_surface() { + fn point_identity_is_stable_and_distinct_per_type_id_and_surface() { let summary = VectorRecord::new( MemoryId::from_u128(7), ObjectType::Episode, @@ -321,6 +321,28 @@ mod tests { assert_eq!(qdrant_point_id(&summary), qdrant_point_id(&summary)); assert_eq!(qdrant_point_id(&summary), qdrant_point_id(&re_embedded)); assert_ne!(qdrant_point_id(&summary), qdrant_point_id(&text)); + let mut identities = std::collections::HashSet::new(); + for id in [7, 8] { + for object_type in [ObjectType::Episode, ObjectType::MemoryThread] { + for surface in [ + VectorSurface::Summary, + VectorSurface::Text, + VectorSurface::DerivedText, + VectorSurface::SceneSetting, + VectorSurface::SceneParticipants, + VectorSurface::Query, + ] { + let record = VectorRecord::new( + MemoryId::from_u128(id), + object_type, + surface, + DEFAULT_SCHEMA_VERSION, + "same text", + ); + assert!(identities.insert(qdrant_point_id(&record)), "{record:?}"); + } + } + } } #[test] diff --git a/src/adapters/qdrant/store.rs b/src/adapters/qdrant/store.rs index 975a95c7..5940352f 100644 --- a/src/adapters/qdrant/store.rs +++ b/src/adapters/qdrant/store.rs @@ -720,7 +720,11 @@ mod tests { let records = [ (1, ObjectType::Episode, VectorSurface::Summary), (1, ObjectType::Episode, VectorSurface::Text), + (1, ObjectType::Episode, VectorSurface::SceneSetting), + (1, ObjectType::Episode, VectorSurface::SceneParticipants), (2, ObjectType::Episode, VectorSurface::Summary), + (2, ObjectType::Episode, VectorSurface::SceneSetting), + (2, ObjectType::Episode, VectorSurface::SceneParticipants), (1, ObjectType::MemoryThread, VectorSurface::Summary), (2, ObjectType::MemoryThread, VectorSurface::Summary), ] @@ -738,11 +742,15 @@ mod tests { .map(|record| VectorRecordEmbedding::new(record, &[1.0, 0.0])) .collect::>(); store.upsert_vector_records(&embeddings).await?; - let query = VectorCandidateSearch::new( + let mut query = VectorCandidateSearch::new( vec![1.0, 0.0], 10, vec![ObjectType::Episode, ObjectType::MemoryThread], ); + query.surfaces.extend([ + VectorSurface::SceneSetting, + VectorSurface::SceneParticipants, + ]); let before = store.search_candidates(&query).await?; store .delete_candidates(&[ @@ -775,7 +783,11 @@ mod tests { [ (1, ObjectType::Episode, VectorSurface::Summary), (1, ObjectType::Episode, VectorSurface::Text), + (1, ObjectType::Episode, VectorSurface::SceneSetting), + (1, ObjectType::Episode, VectorSurface::SceneParticipants), (2, ObjectType::Episode, VectorSurface::Summary), + (2, ObjectType::Episode, VectorSurface::SceneSetting), + (2, ObjectType::Episode, VectorSurface::SceneParticipants), (1, ObjectType::MemoryThread, VectorSurface::Summary), (2, ObjectType::MemoryThread, VectorSurface::Summary), ] @@ -787,6 +799,8 @@ mod tests { surfaces(&after), [ (2, ObjectType::Episode, VectorSurface::Summary), + (2, ObjectType::Episode, VectorSurface::SceneSetting), + (2, ObjectType::Episode, VectorSurface::SceneParticipants), (1, ObjectType::MemoryThread, VectorSurface::Summary), ] .map(|(id, object_type, surface)| (MemoryId::from_u128(id), object_type, surface)) diff --git a/src/memory/scene_tests.rs b/src/memory/scene_tests.rs index f063ed22..a48cc38e 100644 --- a/src/memory/scene_tests.rs +++ b/src/memory/scene_tests.rs @@ -313,6 +313,82 @@ async fn remember_keeps_distinct_vectors_for_episode_and_observation_with_the_sa memory.close().await.unwrap(); } +#[tokio::test] +async fn typed_vector_identity_preserves_content_and_scene_surfaces_through_forget() { + let (memory, _) = memory().await; + let id = MemoryId::from_u128(8802); + let mut thread = MemoryThreadDraft::new("A thread", "An ongoing topic"); + thread.id = Some(id); + let mut observation = ObservationDraft::new(id, "An observed detail"); + observation.id = Some(id); + memory + .remember( + RememberInput::new("An experience") + .with_episode(episode_draft(id.as_u128(), Some(words_scene()))) + .with_observation(observation) + .with_memory_thread(thread), + RememberOptions::default(), + ) + .await + .unwrap(); + + let mut observed = Vec::new(); + for stage in 0..3 { + if stage > 0 { + let target = if stage == 1 { + LifecycleTargetRef::observation(id) + } else { + LifecycleTargetRef::episode(id) + }; + memory + .forget(ForgetMemoryDraft::suppress( + target, + "Only this typed object", + )) + .await + .unwrap(); + } + for (object_type, surface) in [ + (ObjectType::Episode, VectorSurface::Summary), + (ObjectType::Episode, VectorSurface::SceneSetting), + (ObjectType::Episode, VectorSurface::SceneParticipants), + (ObjectType::MemoryThread, VectorSurface::Summary), + ] { + let mut query = RetrievalContext::default().with_trace(); + query.object_type_defaults = vec![object_type]; + query.graph_limits.max_depth = 0; + match surface { + VectorSurface::SceneSetting => { + query.scene.setting.words = words_scene().setting.words + } + VectorSurface::SceneParticipants => { + query.scene.participants = words_scene().participants + } + _ => query.topic = Some("An experience".to_owned()), + } + let result = memory.retrieve(query).await.unwrap(); + let found = result + .trace + .unwrap() + .vector_candidates + .iter() + .any(|candidate| { + candidate.object == MemoryObjectRef::new(object_type, id) + && candidate.surface == surface + }); + observed.push((stage, object_type, surface, found)); + } + } + memory.close().await.unwrap(); + for (stage, object_type, surface, found) in &observed { + assert_eq!( + *found, + *stage < 2 || *object_type == ObjectType::MemoryThread, + "{stage} {object_type:?} {surface:?}: {observed:?}" + ); + } +} + #[tokio::test] async fn setting_words_recall_the_episode_when_the_summary_does_not_name_the_place() { // A lexical provider makes the recall test depend on words, not hash collisions. From 8b2c24fc07f671bb878fdd5956bd71ca36fed97b Mon Sep 17 00:00:00 2001 From: ebigunso Date: Wed, 23 Sep 2026 05:40:56 +0900 Subject: [PATCH 04/13] =?UTF-8?q?=E2=9A=A1=20Hydrate=20only=20requested=20?= =?UTF-8?q?object=20graphs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/adapters/oxigraph/shared.rs | 30 +++++++++++++- src/adapters/oxigraph/tests.rs | 69 +++++++++++++++++++++++++++++++++ 2 files changed, 97 insertions(+), 2 deletions(-) diff --git a/src/adapters/oxigraph/shared.rs b/src/adapters/oxigraph/shared.rs index 947587d3..a425ba03 100644 --- a/src/adapters/oxigraph/shared.rs +++ b/src/adapters/oxigraph/shared.rs @@ -142,7 +142,20 @@ pub(super) fn hydrate_objects_by_refs_from_store( store: &Store, refs: &[MemoryObjectRef], ) -> Result, CustomError> { - let subjects = rdf_subject_values(store)?; + let graphs = refs + .iter() + .filter(|object| object.object_type != ObjectType::MemoryLink) + .map(|object| NamedNode::new(graph_uri(object.object_type, object.id))) + .collect::, _>>()?; + // Include assertion subjects owned by each graph, not just its root subject. + let subjects = rdf_subject_values_from_quads(graphs.iter().flat_map(|graph| { + store.quads_for_pattern( + None, + None, + None, + Some(GraphNameRef::NamedNode(graph.as_ref())), + ) + }))?; let mut objects = Vec::new(); for object_ref in refs { if object_ref.object_type == ObjectType::MemoryLink { @@ -220,9 +233,17 @@ pub(super) fn hydrate_links_by_id_sets_from_store( pub(super) fn rdf_subject_values( store: &Store, +) -> Result, CustomError> { + rdf_subject_values_from_quads(store.iter()) +} + +fn rdf_subject_values_from_quads( + quads: impl IntoIterator>, ) -> Result, CustomError> { let mut subjects = HashMap::::new(); - for quad in store.iter() { + for quad in quads { + #[cfg(test)] + RDF_QUADS_READ.with(|count| count.set(count.get() + 1)); let quad = quad.map_err(oxigraph_error)?; if !matches!(quad.graph_name, GraphName::NamedNode(_)) { continue; @@ -245,6 +266,11 @@ pub(super) fn rdf_subject_values( Ok(subjects) } +#[cfg(test)] +thread_local! { + pub(super) static RDF_QUADS_READ: std::cell::Cell = const { std::cell::Cell::new(0) }; +} + fn rdf_subject_values_for_named_graph( store: &Store, graph_uri: &str, diff --git a/src/adapters/oxigraph/tests.rs b/src/adapters/oxigraph/tests.rs index 05067bd4..46b36067 100644 --- a/src/adapters/oxigraph/tests.rs +++ b/src/adapters/oxigraph/tests.rs @@ -37,6 +37,75 @@ mod tests { } } + #[tokio::test] + async fn by_refs_hydration_reads_only_requested_graphs_in_a_large_store() { + use super::super::shared::RDF_QUADS_READ; + use crate::domain::graph_uri; + use oxigraph::model::{GraphNameRef, NamedNode}; + + let store = OxigraphGraphAuthorityStore::new_in_memory().unwrap(); + let mut fixtures = representative_fixtures(); + fixtures.user_preference.entity_ids = vec![fixtures.user_entity.id]; + fixtures.user_preference.assertions = vec![crate::domain::BeliefAssertion { + subject: fixtures.user_entity.id, + predicate: crate::domain::BeliefPredicate::KnownAs { + name: "A reader".to_owned(), + }, + }]; + let expected = fixtures.objects(); + store.upsert_objects(&expected).await.unwrap(); + let refs = expected + .iter() + .map(MemoryObject::object_ref) + .collect::>(); + let budget = refs + .iter() + .map(|object| { + let graph = NamedNode::new(graph_uri(object.object_type, object.id)).unwrap(); + store + .store + .quads_for_pattern( + None, + None, + None, + Some(GraphNameRef::NamedNode(graph.as_ref())), + ) + .count() + }) + .sum::(); + for unrelated in [0, 2000] { + let extras = (0..unrelated) + .map(|index| { + let mut entity = fixtures.user_entity.clone(); + entity.id = MemoryId::from_u128(100_000 + index); + MemoryObject::Entity(entity) + }) + .collect::>(); + store.upsert_objects(&extras).await.unwrap(); + RDF_QUADS_READ.with(|count| count.set(0)); + let objects = store + .query_objects(&GraphObjectQuery::by_refs(refs.clone())) + .await + .unwrap(); + let read = RDF_QUADS_READ.with(|count| count.get()); + assert_eq!(objects.len(), expected.len()); + for object in &expected { + assert!(objects.contains(object)); + } + assert_eq!( + read, budget, + "unrelated={unrelated}, read={read}, budget={budget}" + ); + } + RDF_QUADS_READ.with(|count| count.set(0)); + assert!(store + .query_objects(&GraphObjectQuery::by_refs(Vec::new())) + .await + .unwrap() + .is_empty()); + assert_eq!(RDF_QUADS_READ.with(|count| count.get()), 0); + } + #[tokio::test] async fn oxigraph_store_upserts_and_queries_canonical_objects() { let store = OxigraphGraphAuthorityStore::new_in_memory().unwrap(); From d54b3f31f94de13e4903f72a484e4df1ac3a6dc5 Mon Sep 17 00:00:00 2001 From: ebigunso Date: Wed, 23 Sep 2026 06:53:16 +0900 Subject: [PATCH 05/13] =?UTF-8?q?=E2=9A=A1=20Bound=20retrieval=20state=20a?= =?UTF-8?q?nd=20participant=20reads?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 2 + docs/design/database/graph_schema_design.md | 4 +- src/adapters/oxigraph/embedded.rs | 24 +- src/adapters/oxigraph/shared.rs | 51 +-- src/adapters/oxigraph/sparql_selectors.rs | 358 +++++++++++++++----- src/adapters/oxigraph/tests.rs | 291 ++++++++++++++++ src/memory/retrieval_scene_tests.rs | 16 +- src/memory/write_turn_tests.rs | 3 +- src/policy/graph_expansion.rs | 9 +- src/ports/graph_authority.rs | 12 +- src/usecases/correct_forget.rs | 3 +- src/usecases/link.rs | 3 +- src/usecases/remember.rs | 3 +- src/usecases/retrieve.rs | 19 +- src/usecases/retrieve/activity.rs | 2 +- 15 files changed, 652 insertions(+), 148 deletions(-) diff --git a/README.md b/README.md index 11f7e4b7..1a570afe 100644 --- a/README.md +++ b/README.md @@ -93,6 +93,8 @@ Each cue kind has a floor at the caps it encounters. The measured defaults reser With tracing enabled, `floor_admissions` identifies the object, stage and cue kind when a reserved or spare turn admitted an object outside that stage's original capped prefix. Earlier-stage evidence remains even when a later stage omits the object. +State and participant-occasion reads bound eligible results and exclusion evidence separately. Exclusion evidence is a latest-first prefix within the caller's budget, not a complete audit. Fanout omission telemetry counts only omissions seen within fetched prefixes; it does not count the entire matching store. + The trace's `scene_cue_searches` reports the best similarity found for the setting and for the joined participant words, with the references sharing each search; a search with no matches has no score. Each entry identifies its cue kind and counts recallable matches left out by the description's occasion limit in `omitted_count`. There is no minimum similarity: whether a description is too weak to remind the character of anything remains a question for measurement, not a decision made by this score. A participant present in most experiences brings fewer past encounters to mind, while retaining the latest eligible one within the requested limits. A participant recognized by key or name brings only encounters at or before the scene time. Several remarks or participants in one encounter do not make it count more than once. Familiarity limits recalled encounters, not beliefs about that participant. This applies to every notion, including whichever one the application regards as the character. diff --git a/docs/design/database/graph_schema_design.md b/docs/design/database/graph_schema_design.md index feebaf32..b4a55855 100644 --- a/docs/design/database/graph_schema_design.md +++ b/docs/design/database/graph_schema_design.md @@ -120,7 +120,9 @@ Incoming `Resolves` and `FulfillsCommitment` links exclude targets from named-su Each resolved scene reference reports a `last_interactions` map keyed by notion ID, including every candidate of an ambiguous name. Values contain the last episode ID, recorded scene time and whole seconds elapsed relative to the retrieval scene; `null` means no eligible encounter at or before that time. Descriptions and unknown references have an empty map. [Scene resolution (`scene.rs:98`)](../../../src/usecases/retrieve/scene.rs#L98) collects these facts independently of pack caps and tracing. The [last-interaction selector (`sparql_selectors.rs:414`)](../../../src/adapters/oxigraph/sparql_selectors.rs#L414) uses one SPARQL query over `Involves` episode links and `Mentions` observation-to-parent-episode links, in either orientation. Both neighbor and parent retention checks, controlled by `include_suppressed`, and the scene-time cutoff precede `ORDER BY` descending timestamp, ascending episode ID and `LIMIT 1`. Scene-time strings are cast to timestamps for comparison, preserving fractional-time ordering. One result row crosses into Rust; the query engine may still examine all matching candidates. -Pack expansion needs multiple occasions and exclusion evidence, so it retains a separate occasion read. The [participant limiter (`graph_expansion.rs:1145`)](../../../src/policy/graph_expansion.rs#L1145) shares one episode budget across direct episode and observation routes, with at most one observation per selected episode. Frequency counts distinct episodes, and selection prefers recorded scene time, latest first. Only participants recognized by key or name apply the retrieval scene-time cutoff here; beliefs and other retrieval paths have no new as-of rule. +Pack expansion reads bounded occasion prefixes and exclusion evidence separately. The [participant limiter](../../../src/policy/graph_expansion.rs) shares one episode budget across direct episode and observation routes, with at most one observation per selected episode. Eligibility and distinct-episode selection precede the query limit. Frequency counts distinct episodes, and selection prefers recorded scene time, latest first. Only participants recognized by key or name apply the retrieval scene-time cutoff here; beliefs and other retrieval paths have no new as-of rule. + +State reads also apply lifecycle eligibility before their caller's query limit; subject state additionally requires a traversable About link. Exclusion evidence has a separate latest-first prefix of the same budget. Fanout omission telemetry counts omissions within fetched prefixes, without a whole-store count query. Thread expansion excludes resolved members independently of that bounded evidence. ## Cross-Store Failures diff --git a/src/adapters/oxigraph/embedded.rs b/src/adapters/oxigraph/embedded.rs index 7742e8e6..b8d8971e 100644 --- a/src/adapters/oxigraph/embedded.rs +++ b/src/adapters/oxigraph/embedded.rs @@ -245,7 +245,7 @@ impl GraphAuthorityStore for OxigraphGraphAuthorityStore { .map(|id| MemoryObjectRef::new(ObjectType::DerivedMemory, *id)) .collect::>(); let mut ids = SparqlGraphSelectors::new(&self.store) - .select_links_touching(&refs) + .select_links_touching(&refs, false) .map_err(|error| GraphQueryError::Selection { detail: error.to_string(), })? @@ -301,6 +301,25 @@ impl GraphAuthorityStore for OxigraphGraphAuthorityStore { &self, query: &GraphDerivedMemoryThreadQuery, ) -> Result<(Vec, Vec), CustomError> { + if let Some(limit) = query.current_state_limit { + let (ids, filtered) = + SparqlGraphSelectors::new(&self.store).select_thread_state(query, limit)?; + let objects = hydrate_objects_by_refs_from_store( + &self.store, + &ids.into_iter() + .map(|id| MemoryObjectRef::new(ObjectType::DerivedMemory, id)) + .collect::>(), + )?; + let mut memories = objects + .into_iter() + .filter_map(|object| match object { + MemoryObject::DerivedMemory(memory) => Some(memory), + _ => None, + }) + .collect::>(); + memories.sort_by_key(|memory| (std::cmp::Reverse(memory.created_at), memory.id)); + return Ok((memories, filtered)); + } let selected_ids = SparqlGraphSelectors::new(&self.store) .select_derived_memories_by_thread(query)? .into_iter() @@ -328,8 +347,9 @@ impl GraphAuthorityStore for OxigraphGraphAuthorityStore { &self, key: &ScopeKey, policy: GraphExpansionLifecyclePolicy, + limit: usize, ) -> Result<(Vec, Vec), CustomError> { - SparqlGraphSelectors::new(&self.store).select_scope_state(key, policy) + SparqlGraphSelectors::new(&self.store).select_scope_state(key, policy, limit) } async fn expand_bounded( diff --git a/src/adapters/oxigraph/shared.rs b/src/adapters/oxigraph/shared.rs index a425ba03..2e61ce46 100644 --- a/src/adapters/oxigraph/shared.rs +++ b/src/adapters/oxigraph/shared.rs @@ -627,7 +627,16 @@ pub(super) fn bounded_graph_visible_refs( let mut bounded_failure = None; let mut frontier = vec![root_ref]; let (state_ids, state_filtered) = if query.current_subject_state { - selectors.select_subject_state(query.root_id, query.lifecycle_policy)? + selectors.select_subject_state( + query.root_id, + query.lifecycle_policy, + crate::policy::graph_expansion::fanout_limit_for_pair( + query, + RelationType::About, + ObjectType::DerivedMemory, + ), + query.trace_mode.is_enabled(), + )? } else { (Vec::new(), Vec::new()) }; @@ -643,23 +652,39 @@ pub(super) fn bounded_graph_visible_refs( if frontier.is_empty() { break; } - let link_refs = selectors.select_links_touching(&frontier)?; + let mut link_refs = + selectors.select_links_touching(&frontier, depth == 0 && query.current_thread_state)?; if depth == 0 && query .fanout_overrides .iter() .any(|entry| is_participant_pair(entry.relation, entry.object_type)) { + let mut seen = HashSet::new(); let neighbors = link_refs .iter() .filter_map(|link| { let neighbor = link.other_endpoint(root_ref); - is_participant_pair(link.relation, neighbor.object_type).then_some(neighbor) + (is_participant_pair(link.relation, neighbor.object_type) + && query.allows_object(neighbor) + && (query.allowed_relation_types.is_empty() + || query.allowed_relation_types.contains(&link.relation)) + && crate::policy::graph_expansion::fanout_limit_for_pair( + query, + link.relation, + neighbor.object_type, + ) > 0 + && seen.insert(neighbor)) + .then_some(neighbor) }) - .collect::>() - .into_iter() .collect::>(); - participant_occasions = selectors.select_participant_occasions(&neighbors)?; + participant_occasions = + selectors.select_bounded_participant_occasions(&neighbors, query)?; + link_refs.retain(|link| { + let neighbor = link.other_endpoint(root_ref); + !is_participant_pair(link.relation, neighbor.object_type) + || participant_occasions.contains_key(&neighbor) + }); } let link_refs_by_endpoint = link_refs_by_endpoint(&link_refs); let mut next_frontier = Vec::new(); @@ -700,18 +725,6 @@ pub(super) fn bounded_graph_visible_refs( }, ); &ordered - } else if depth == 0 && !query.resolved_thread_members.is_empty() { - ordered = incident_link_refs - .iter() - .filter(|link| { - link.relation != RelationType::PartOfThread - || !query - .resolved_thread_members - .contains(&link.other_endpoint(*object_ref)) - }) - .copied() - .collect(); - &ordered } else { incident_link_refs }; @@ -749,7 +762,7 @@ pub(super) fn bounded_graph_visible_refs( let candidate_refs = graph_refs.iter().copied().collect::>(); let lifecycle_link_ids = selectors - .select_links_touching(&candidate_refs)? + .select_links_touching(&candidate_refs, false)? .into_iter() .filter(|link_ref| { matches!( diff --git a/src/adapters/oxigraph/sparql_selectors.rs b/src/adapters/oxigraph/sparql_selectors.rs index b6938f3a..da3dca46 100644 --- a/src/adapters/oxigraph/sparql_selectors.rs +++ b/src/adapters/oxigraph/sparql_selectors.rs @@ -219,55 +219,126 @@ impl<'a> SparqlGraphSelectors<'a> { ) } - // Do not LIMIT here: own-subject metadata can exist without a traversable About - // link. Limiting before that intersection can leave the state bucket underfilled. pub(crate) fn select_subject_state( &self, subject_id: MemoryId, policy: GraphExpansionLifecyclePolicy, + limit: usize, + read_more: bool, ) -> Result<(Vec, Vec), CustomError> { let subject = graph_uri(ObjectType::Entity, subject_id); - self.select_state(&format!("<{}> <{subject}>", vocab::ABOUT_ENTITY), policy) + let scope = format!( + r#" + GRAPH ?memory {{ ?memory <{about}> <{subject}> . }} + GRAPH ?aboutLink {{ + ?aboutLink a <{link_class}> ; <{relation}> "about" . + {{ ?aboutLink <{from}> ?memory ; <{to}> <{subject}> ; <{from_type}> "derived_memory" ; <{to_type}> "entity" . }} + UNION {{ ?aboutLink <{to}> ?memory ; <{from}> <{subject}> ; <{to_type}> "derived_memory" ; <{from_type}> "entity" . }} + }} + "#, + about = vocab::ABOUT_ENTITY, + link_class = vocab::CLASS_MEMORY_LINK, + relation = vocab::RELATION, + from = vocab::FROM, + to = vocab::TO, + from_type = vocab::FROM_TYPE, + to_type = vocab::TO_TYPE + ); + self.select_state( + &scope, + policy, + limit, + read_more, + "?superseded DESC(xsd:double(?salience)) DESC(xsd:dateTime(?created)) ?id", + ) } pub(crate) fn select_scope_state( &self, key: &crate::domain::ScopeKey, policy: GraphExpansionLifecyclePolicy, + limit: usize, ) -> Result<(Vec, Vec), CustomError> { let value = serde_json::to_string(key).expect("scope keys contain strings only"); - let predicate = format!( - "<{}> {}", + let scope = format!( + "GRAPH ?memory {{ ?memory <{}> {} . }}", vocab::SCOPE_KEY, oxigraph::model::Literal::new_simple_literal(value) ); - self.select_state(&predicate, policy) + self.select_state( + &scope, + policy, + limit, + false, + "?superseded DESC(xsd:double(?salience)) DESC(xsd:dateTime(?created)) ?id", + ) + } + + pub(crate) fn select_thread_state( + &self, + query: &GraphDerivedMemoryThreadQuery, + limit: usize, + ) -> Result<(Vec, Vec), CustomError> { + if query.thread_ids.is_empty() { + return Ok((Vec::new(), Vec::new())); + } + let threads = query + .thread_ids + .iter() + .map(|id| graph_uri(ObjectType::MemoryThread, *id)) + .collect::>(); + let scope = format!( + "{} GRAPH ?memory {{ ?memory <{}> ?thread . }}", + sparql_iri_values("thread", threads.iter().map(String::as_str)), + vocab::PART_OF_THREAD + ); + self.select_state( + &scope, + query.lifecycle_policy, + limit, + false, + "DESC(xsd:dateTime(?created)) ?id", + ) } fn select_state( &self, - scope_predicate: &str, + scope: &str, policy: GraphExpansionLifecyclePolicy, + limit: usize, + read_more: bool, + order: &str, ) -> Result<(Vec, Vec), CustomError> { - let query = format!( + let eligible_limit = limit.saturating_add(usize::from(read_more)); + if eligible_limit == 0 { + return Ok((Vec::new(), Vec::new())); + } + // Join lifecycle links once; per-memory EXISTS repeats the same link scan. + let body = format!( r#" - PREFIX xsd: - SELECT ?id ?retention ?relation ?source WHERE {{ - GRAPH ?g {{ + {scope} + GRAPH ?memory {{ ?memory a <{derived_class}> ; <{object_id}> ?id ; - {scope_predicate} ; <{salience}> ?salience ; + <{salience}> ?salience ; <{created}> ?created ; <{retention}> ?retention . }} OPTIONAL {{ - GRAPH ?linkGraph {{ - ?link a <{link_class}> ; <{from_type}> "derived_memory" ; - <{to_type}> "derived_memory" ; <{relation}> ?relation ; - <{from}> ?source ; <{to}> ?memory . - VALUES ?relation {{ "supersedes" "resolves" "fulfills_commitment" }} + GRAPH ?supersessionLink {{ + ?supersessionLink a <{link_class}> ; <{from_type}> "derived_memory" ; + <{to_type}> "derived_memory" ; <{relation}> "supersedes" ; <{to}> ?memory . }} }} - }} - ORDER BY DESC(xsd:double(?salience)) DESC(xsd:dateTime(?created)) ?id ?relation ?source + BIND(BOUND(?supersessionLink) AS ?superseded) + OPTIONAL {{ + GRAPH ?resolution {{ + ?resolution a <{link_class}> ; <{from_type}> "derived_memory" ; + <{to_type}> "derived_memory" ; <{relation}> ?resolutionKind ; <{to}> ?memory . + VALUES ?resolutionKind {{ "resolves" "fulfills_commitment" }} + }} + }} + BIND(BOUND(?resolution) AS ?resolved) + BIND(({include_suppressed} || ?retention != "suppressed") && + ({include_superseded} || !?superseded) && !?resolved AS ?eligible) "#, derived_class = vocab::CLASS_DERIVED_MEMORY, object_id = vocab::OBJECT_ID, @@ -278,73 +349,98 @@ impl<'a> SparqlGraphSelectors<'a> { from_type = vocab::FROM_TYPE, to_type = vocab::TO_TYPE, relation = vocab::RELATION, - from = vocab::FROM, to = vocab::TO, + include_suppressed = policy.include_suppressed, + include_superseded = policy.include_superseded, ); - // Aggregate all lifecycle rows before filtering: a memory can be both - // superseded and resolved, with several incoming links of either kind. - let mut states = Vec::<(MemoryId, RetentionState, Vec, bool)>::new(); - for solution in self.query_solutions(&query)? { - let id = memory_id_binding(&solution, "id")?; - if states.last().is_none_or(|entry| entry.0 != id) { - states.push((id, enum_binding(&solution, "retention")?, Vec::new(), false)); - } - let state = states.last_mut().unwrap(); - if let Some(source) = solution.get("source") { - let Term::NamedNode(source) = source else { - return Err(oxigraph_sparql_error(format!( - "expected lifecycle source IRI, got {source}" - ))); - }; - match enum_binding::(&solution, "relation")? { - RelationType::Supersedes => state - .2 - .push(super::shared::memory_id_from_resource(source.as_str())?), - RelationType::Resolves | RelationType::FulfillsCommitment => state.3 = true, - _ => unreachable!("query restricts lifecycle relations"), - } - } + let prefix = "PREFIX xsd: "; + let ranked_ids = self + .query_solutions(&format!( + r#"{prefix} + SELECT DISTINCT ?id ?superseded ?salience ?created WHERE {{ {body} FILTER(?eligible) }} + ORDER BY {order} LIMIT {eligible_limit} + "# + ))? + .iter() + .map(|row| memory_id_binding(row, "id")) + .collect::, _>>()?; + if limit == 0 { + return Ok((ranked_ids, Vec::new())); } - let mut ranked_ids = Vec::new(); - let mut historical_ids = Vec::new(); + // Exclusion evidence has its own latest-first budget; it cannot crowd out eligible state. + let excluded = format!( + r#"{prefix} + SELECT ?id ?retention ?created (GROUP_CONCAT(DISTINCT COALESCE(STR(?source), ""); SEPARATOR=" ") AS ?sources) WHERE {{ + {{ SELECT DISTINCT ?memory ?id ?retention ?created WHERE {{ {body} FILTER(!?eligible) }} + ORDER BY DESC(xsd:dateTime(?created)) ?id LIMIT {limit} }} + OPTIONAL {{ GRAPH ?supersession {{ + ?supersession a <{link_class}> ; <{from_type}> "derived_memory" ; + <{to_type}> "derived_memory" ; <{relation}> "supersedes" ; <{from}> ?source ; <{to}> ?memory . + }} }} + }} GROUP BY ?id ?retention ?created + ORDER BY DESC(xsd:dateTime(?created)) ?id LIMIT {limit} + "#, + link_class = vocab::CLASS_MEMORY_LINK, + from_type = vocab::FROM_TYPE, + to_type = vocab::TO_TYPE, + relation = vocab::RELATION, + from = vocab::FROM, + to = vocab::TO + ); let mut filtered = Vec::new(); - for (id, retention, mut superseded_by, resolved) in states { + for row in self.query_solutions(&excluded)? { + let id = memory_id_binding(&row, "id")?; + let retention = enum_binding::(&row, "retention")?; + let mut superseded_by = literal_binding(&row, "sources")? + .split_whitespace() + .map(super::shared::memory_id_from_resource) + .collect::, _>>()?; superseded_by.sort_unstable(); - superseded_by.dedup(); let reason = if retention == RetentionState::Suppressed && !policy.include_suppressed { - Some(GraphExpansionFilteredReason::Suppressed) + GraphExpansionFilteredReason::Suppressed } else if !superseded_by.is_empty() && !policy.include_superseded { - Some(GraphExpansionFilteredReason::Superseded) - } else if resolved { - Some(GraphExpansionFilteredReason::Resolved) + GraphExpansionFilteredReason::Superseded } else { - None + GraphExpansionFilteredReason::Resolved }; - if let Some(reason) = reason { - filtered.push(GraphExpansionFilteredNode { - object_ref: MemoryObjectRef::new(ObjectType::DerivedMemory, id), - reason, - superseded_by, - }); - } else if superseded_by.is_empty() { - ranked_ids.push(id); - } else { - historical_ids.push(id); - } + filtered.push(GraphExpansionFilteredNode { + object_ref: MemoryObjectRef::new(ObjectType::DerivedMemory, id), + reason, + superseded_by, + }); } - ranked_ids.extend(historical_ids); Ok((ranked_ids, filtered)) } pub(crate) fn select_links_touching( &self, object_refs: &[MemoryObjectRef], + current_thread_state: bool, ) -> Result, CustomError> { if object_refs.is_empty() { return Ok(Vec::new()); } let node_values = sparql_node_iri_values("node", object_refs); + let thread_filter = if current_thread_state { + format!( + r#" + BIND(IF(?fromType = "derived_memory", ?from, ?to) AS ?member) + FILTER(?relation != "part_of_thread" || NOT EXISTS {{ GRAPH ?resolution {{ + ?resolution a <{link_class}> ; <{from_type}> "derived_memory" ; + <{to_type}> "derived_memory" ; <{relation}> ?resolutionKind ; <{to}> ?member . + VALUES ?resolutionKind {{ "resolves" "fulfills_commitment" }} + }} }}) + "#, + link_class = vocab::CLASS_MEMORY_LINK, + from_type = vocab::FROM_TYPE, + to_type = vocab::TO_TYPE, + relation = vocab::RELATION, + to = vocab::TO + ) + } else { + String::new() + }; let query_text = format!( r#" SELECT DISTINCT ?linkId ?fromId ?fromType ?toId ?toType ?relation WHERE {{ @@ -369,6 +465,7 @@ impl<'a> SparqlGraphSelectors<'a> { ?to <{object_id}> ?toId ; <{object_type}> ?toType . }} + {thread_filter} }} "#, link_class = vocab::CLASS_MEMORY_LINK, @@ -478,26 +575,7 @@ impl<'a> SparqlGraphSelectors<'a> { if neighbors.is_empty() { return Ok(ParticipantOccasions::new()); } - let node_values = sparql_node_iri_values("node", neighbors); - let query_text = format!( - r#" - SELECT DISTINCT ?id ?objectType ?episodeId ?sceneTime ?retention ?episodeRetention WHERE {{ - {node_values} - GRAPH ?node {{ ?node <{object_id}> ?id ; <{object_type}> ?objectType ; <{retention}> ?retention . }} - {{ - GRAPH ?node {{ ?node <{object_id}> ?episodeId ; <{scene_time}> ?sceneTime ; <{retention}> ?episodeRetention . }} - }} UNION {{ - GRAPH ?node {{ ?node <{episode}> ?episode . }} - GRAPH ?episode {{ ?episode <{object_id}> ?episodeId ; <{scene_time}> ?sceneTime ; <{retention}> ?episodeRetention . }} - }} - }} - "#, - object_id = vocab::OBJECT_ID, - object_type = vocab::OBJECT_TYPE, - scene_time = vocab::SCENE_TIME, - episode = vocab::EPISODE, - retention = vocab::RETENTION_STATE, - ); + let query_text = format!("SELECT DISTINCT ?id ?objectType ?episodeId ?sceneTime ?retention ?episodeRetention WHERE {{ {} }}", participant_occasion_pattern(neighbors)); let mut occasions = ParticipantOccasions::new(); for solution in self.query_solutions(&query_text)? { let neighbor = MemoryObjectRef::from_id_type( @@ -524,6 +602,80 @@ impl<'a> SparqlGraphSelectors<'a> { Ok(occasions) } + pub(crate) fn select_bounded_participant_occasions( + &self, + neighbors: &[MemoryObjectRef], + query: &crate::ports::graph_authority::GraphExpansionQuery, + ) -> Result { + let budget = query + .fanout_overrides + .iter() + .filter(|entry| { + crate::policy::graph_expansion::is_participant_pair( + entry.relation, + entry.object_type, + ) + }) + .map(|entry| entry.max_fanout) + .max() + .unwrap_or(0) + .min(query.max_fanout_per_node); + if neighbors.is_empty() || budget == 0 { + return Ok(ParticipantOccasions::new()); + } + let prefix = "PREFIX xsd: "; + let cutoff = if query.current_subject_state { + format!( + "FILTER(?time <= {}^^xsd:dateTime)", + sparql_string_literal(&query.participant_reference_time.to_rfc3339()) + ) + } else { + String::new() + }; + let body = format!( + r#"{} + BIND(xsd:dateTime(?sceneTime) AS ?time) {cutoff} + BIND(({} || (?retention = "active" && ?episodeRetention = "active")) AS ?eligible) + "#, + participant_occasion_pattern(neighbors), + query.lifecycle_policy.include_suppressed + ); + let mut selected = Vec::new(); + for eligible in [true, false] { + let limit = + budget.saturating_add(usize::from(eligible && query.trace_mode.is_enabled())); + // Reserve distinct occasions before choosing one stable route of each type. + let episodes = self + .query_solutions(&format!( + r#"{prefix} + SELECT DISTINCT ?episodeId ?time WHERE {{ {body} FILTER(?eligible = {eligible}) }} + ORDER BY DESC(?time) ?episodeId LIMIT {limit} + "# + ))? + .iter() + .map(|row| memory_id_binding(row, "episodeId")) + .collect::, _>>()?; + if episodes.is_empty() { + continue; + } + let values = + sparql_literal_values("episodeId", episodes.iter().map(ToString::to_string)); + for row in self.query_solutions(&format!( + r#"{prefix} + SELECT ?episodeId ?objectType (MIN(?rank) AS ?selectedRank) WHERE {{ + {body} {values} FILTER(?eligible = {eligible}) + }} GROUP BY ?episodeId ?objectType + "# + ))? { + let rank = literal_binding(&row, "selectedRank")? + .parse::() + .map_err(oxigraph_sparql_error)?; + selected.push(neighbors[rank]); + } + } + self.select_participant_occasions(&selected) + } + pub(crate) fn select_anniversaries( &self, date: chrono::NaiveDate, @@ -728,12 +880,46 @@ impl<'a> SparqlGraphSelectors<'a> { )); }; - solutions + let rows = solutions .collect::, _>>() - .map_err(oxigraph_sparql_error) + .map_err(oxigraph_sparql_error)?; + #[cfg(test)] + MAX_SELECT_ROWS.with(|count| count.set(count.get().max(rows.len()))); + Ok(rows) } } +#[cfg(test)] +thread_local! { + pub(super) static MAX_SELECT_ROWS: std::cell::Cell = const { std::cell::Cell::new(0) }; +} + +fn participant_occasion_pattern(neighbors: &[MemoryObjectRef]) -> String { + let values = neighbors + .iter() + .enumerate() + .map(|(rank, object)| format!("(<{}> {rank})", graph_uri(object.object_type, object.id))) + .collect::>() + .join(" "); + format!( + r#" + VALUES (?node ?rank) {{ {values} }} + GRAPH ?node {{ ?node <{object_id}> ?id ; <{object_type}> ?objectType ; <{retention}> ?retention . }} + {{ + GRAPH ?node {{ ?node <{object_id}> ?episodeId ; <{scene_time}> ?sceneTime ; <{retention}> ?episodeRetention . }} + }} UNION {{ + GRAPH ?node {{ ?node <{episode}> ?episode . }} + GRAPH ?episode {{ ?episode <{object_id}> ?episodeId ; <{scene_time}> ?sceneTime ; <{retention}> ?episodeRetention . }} + }} + "#, + object_id = vocab::OBJECT_ID, + object_type = vocab::OBJECT_TYPE, + retention = vocab::RETENTION_STATE, + scene_time = vocab::SCENE_TIME, + episode = vocab::EPISODE + ) +} + fn occasion_retention_filter(policy: GraphExpansionLifecyclePolicy) -> &'static str { if policy.include_suppressed { "" diff --git a/src/adapters/oxigraph/tests.rs b/src/adapters/oxigraph/tests.rs index 46b36067..80f7081f 100644 --- a/src/adapters/oxigraph/tests.rs +++ b/src/adapters/oxigraph/tests.rs @@ -1215,6 +1215,297 @@ mod tests { )); } + #[tokio::test] + async fn retrieval_selectors_bound_state_and_occasion_prefixes() { + use super::super::sparql_selectors::{SparqlGraphSelectors, MAX_SELECT_ROWS}; + use crate::domain::ScopeKey; + let store = OxigraphGraphAuthorityStore::new_in_memory().unwrap(); + let fixtures = representative_fixtures(); + let key = ScopeKey::Setting("home".to_owned()); + let mut objects = vec![ + MemoryObject::Entity(fixtures.hub_entity.clone()), + MemoryObject::MemoryThread(fixtures.soft_thread.clone()), + ]; + let mut links = Vec::new(); + let mut expected_state = Vec::new(); + let mut expected_subject = Vec::new(); + let mut neighbors = Vec::new(); + for index in 0..20_u128 { + let mut memory = fixtures.open_loop.clone(); + memory.id = MemoryId::from_u128(10_000 + index); + memory.created_at += chrono::Duration::minutes(index as i64); + memory.salience_score = (index % 7) as f32 / 10.0; + memory.entity_ids = vec![fixtures.hub_entity.id]; + memory.thread_ids = vec![fixtures.soft_thread.id]; + memory.scope_keys = vec![key.clone()]; + if index % 5 == 0 { + memory.retention_state = RetentionState::Suppressed; + } + if index % 5 >= 3 { + expected_state.push(memory.clone()); + if index % 13 != 0 { + expected_subject.push(memory.clone()); + } + } + let mut link = fixtures.soft_thread_link.clone(); + link.id = MemoryId::from_u128(100_000 + index); + link.from_id = memory.id; + link.from_type = ObjectType::DerivedMemory; + link.to_id = fixtures.hub_entity.id; + link.to_type = ObjectType::Entity; + link.relation = RelationType::About; + if index % 13 != 0 { + links.push(link.clone()); + } + if matches!(index % 5, 1 | 2) { + link.id = MemoryId::from_u128(110_000 + index); + link.to_id = memory.id; + link.to_type = ObjectType::DerivedMemory; + link.from_id = fixtures.correction.id; + link.relation = if index % 5 == 1 { + RelationType::Resolves + } else { + RelationType::Supersedes + }; + links.push(link); + } + objects.push(MemoryObject::DerivedMemory(memory)); + } + + for index in 0..10_u128 { + let mut episode = fixtures.episode.clone(); + episode.id = MemoryId::from_u128(20_000 + index); + episode.scene.time += chrono::Duration::minutes(index as i64); + if index % 2 == 0 { + episode.retention_state = RetentionState::Suppressed; + } + neighbors.push(MemoryObjectRef::new(ObjectType::Episode, episode.id)); + objects.push(MemoryObject::Episode(episode.clone())); + for offset in 0..2 { + let mut observation = fixtures.salient_observation.clone(); + observation.id = MemoryId::from_u128(30_000 + 2 * index + offset); + observation.episode_id = episode.id; + neighbors.push(MemoryObjectRef::new( + ObjectType::Observation, + observation.id, + )); + objects.push(MemoryObject::Observation(observation)); + } + } + store.upsert_objects(&objects).await.unwrap(); + store.upsert_links(&links).await.unwrap(); + let policy = GraphExpansionLifecyclePolicy::default(); + let selectors = SparqlGraphSelectors::new(&store.store); + MAX_SELECT_ROWS.with(|count| count.set(0)); + let (scope, scope_filtered) = selectors.select_scope_state(&key, policy, 3).unwrap(); + assert!(MAX_SELECT_ROWS.with(|count| count.get()) <= 3); + let (subject, subject_filtered) = selectors + .select_subject_state(fixtures.hub_entity.id, policy, 3, false) + .unwrap(); + assert!(MAX_SELECT_ROWS.with(|count| count.get()) <= 3); + let mut query = GraphDerivedMemoryThreadQuery::by_threads(vec![fixtures.soft_thread.id]); + query.current_state_limit = Some(3); + let (mut thread, thread_filtered) = store + .query_derived_memories_by_thread(&query) + .await + .unwrap(); + thread.sort_by_key(|memory| (std::cmp::Reverse(memory.created_at), memory.id)); + assert!(MAX_SELECT_ROWS.with(|count| count.get()) <= 3); + let mut occasion_query = + GraphExpansionQuery::new(fixtures.hub_entity.id, ObjectType::Entity, 1, 10) + .with_fanout_overrides(vec![ + GraphExpansionFanoutOverride { + relation: RelationType::Involves, + object_type: ObjectType::Episode, + max_fanout: 3, + }, + GraphExpansionFanoutOverride { + relation: RelationType::Mentions, + object_type: ObjectType::Observation, + max_fanout: 3, + }, + ]); + occasion_query.current_subject_state = true; + occasion_query.participant_reference_time = + fixtures.episode.scene.time.to_utc() + chrono::Duration::minutes(8); + let occasions = selectors + .select_bounded_participant_occasions(&neighbors, &occasion_query) + .unwrap(); + assert!(MAX_SELECT_ROWS.with(|count| count.get()) <= 12); + for entries in [&scope_filtered, &subject_filtered, &thread_filtered] { + assert_eq!( + entries + .iter() + .map(|entry| entry.object_ref.id) + .collect::>(), + [10017, 10016, 10015].map(MemoryId::from_u128) + ); + } + for (eligible, indices) in [(true, [7, 5, 3]), (false, [8, 6, 4])] { + let ids = occasions + .iter() + .filter(|(neighbor, occasion)| { + occasion.filtered_reason(**neighbor, policy).is_none() == eligible + }) + .map(|(_, occasion)| occasion.episode_id) + .collect::>(); + assert_eq!( + ids, + indices + .map(|index| MemoryId::from_u128(20_000 + index)) + .into_iter() + .collect() + ); + } + for memories in [&mut expected_state, &mut expected_subject] { + memories.sort_by(|a, b| { + b.salience_score + .total_cmp(&a.salience_score) + .then_with(|| b.created_at.cmp(&a.created_at)) + .then_with(|| a.id.cmp(&b.id)) + }); + } + assert_eq!( + &scope[..3], + &expected_state + .iter() + .take(3) + .map(|memory| memory.id) + .collect::>() + ); + let traversable = subject + .iter() + .copied() + .filter(|id| expected_subject.iter().any(|memory| memory.id == *id)) + .take(3) + .collect::>(); + assert_eq!( + traversable, + expected_subject + .iter() + .take(3) + .map(|memory| memory.id) + .collect::>() + ); + expected_state.sort_by_key(|memory| (std::cmp::Reverse(memory.created_at), memory.id)); + assert_eq!( + thread + .iter() + .take(3) + .map(|memory| memory.id) + .collect::>(), + expected_state + .iter() + .take(3) + .map(|memory| memory.id) + .collect::>() + ); + let counts = [ + scope.len(), + scope_filtered.len(), + subject.len(), + subject_filtered.len(), + thread.len(), + thread_filtered.len(), + occasions.len(), + ]; + assert_eq!(counts, [3, 3, 3, 3, 3, 3, 12]); + let observation_ids = occasions + .keys() + .filter(|neighbor| neighbor.object_type == ObjectType::Observation) + .map(|neighbor| neighbor.id) + .collect::>(); + assert_eq!( + observation_ids, + [7, 5, 3, 8, 6, 4] + .map(|index| MemoryId::from_u128(30_000 + 2 * index)) + .into_iter() + .collect() + ); + MAX_SELECT_ROWS.with(|count| count.set(0)); + assert!(selectors + .select_scope_state(&key, policy, 0) + .unwrap() + .0 + .is_empty()); + assert!(selectors + .select_subject_state(fixtures.hub_entity.id, policy, 0, false) + .unwrap() + .0 + .is_empty()); + query.current_state_limit = Some(0); + assert!(store + .query_derived_memories_by_thread(&query) + .await + .unwrap() + .0 + .is_empty()); + occasion_query.max_fanout_per_node = 0; + assert!(selectors + .select_bounded_participant_occasions(&neighbors, &occasion_query) + .unwrap() + .is_empty()); + assert_eq!(MAX_SELECT_ROWS.with(|count| count.get()), 0); + } + + #[tokio::test] + async fn thread_expansion_excludes_resolved_members_outside_its_trace_prefix() { + let store = OxigraphGraphAuthorityStore::new_in_memory().unwrap(); + let fixtures = representative_fixtures(); + let mut objects = vec![MemoryObject::MemoryThread(fixtures.soft_thread.clone())]; + let mut links = Vec::new(); + for index in 0..8 { + let mut memory = fixtures.open_loop.clone(); + memory.id = MemoryId::from_u128(5000 + index); + memory.thread_ids = vec![fixtures.soft_thread.id]; + memory.created_at += chrono::Duration::minutes(index as i64); + let mut link = fixtures.soft_thread_link.clone(); + link.id = MemoryId::from_u128(6000 + index); + link.from_id = memory.id; + link.from_type = ObjectType::DerivedMemory; + link.to_id = fixtures.soft_thread.id; + link.to_type = ObjectType::MemoryThread; + link.relation = RelationType::PartOfThread; + links.push(link.clone()); + if [0, 1, 6].contains(&index) { + link.id = MemoryId::from_u128(7000 + index); + link.from_id = fixtures.correction.id; + link.to_id = memory.id; + link.to_type = ObjectType::DerivedMemory; + link.relation = RelationType::Resolves; + links.push(link); + } + objects.push(MemoryObject::DerivedMemory(memory)); + } + store.upsert_objects(&objects).await.unwrap(); + store.upsert_links(&links).await.unwrap(); + let mut state = GraphDerivedMemoryThreadQuery::by_threads(vec![fixtures.soft_thread.id]); + state.current_state_limit = Some(1); + let (_, excluded) = store + .query_derived_memories_by_thread(&state) + .await + .unwrap(); + assert_eq!( + excluded + .iter() + .map(|row| row.object_ref.id) + .collect::>(), + [MemoryId::from_u128(5006)] + ); + let mut query = + GraphExpansionQuery::new(fixtures.soft_thread.id, ObjectType::MemoryThread, 1, 3) + .with_max_fanout_per_node(1); + query.current_thread_state = true; + let expansion = store.expand_bounded(&query).await.unwrap(); + let members = expansion + .objects + .iter() + .filter(|object| object.object_type() == ObjectType::DerivedMemory) + .map(MemoryObject::id) + .collect::>(); + assert_eq!(members, [MemoryId::from_u128(5002)]); + } + #[tokio::test] async fn named_subject_state_intersects_links_before_applying_the_cap() { let store = OxigraphGraphAuthorityStore::new_in_memory().unwrap(); diff --git a/src/memory/retrieval_scene_tests.rs b/src/memory/retrieval_scene_tests.rs index 9ab6eaa3..921b69cf 100644 --- a/src/memory/retrieval_scene_tests.rs +++ b/src/memory/retrieval_scene_tests.rs @@ -637,10 +637,8 @@ async fn assert_participant_recall_after_suppression(active_sibling: bool) { }) .unwrap(); assert_eq!(utilization.retained_count, 1); - assert_eq!( - utilization.omitted_by_fanout_count, - 8 + usize::from(active_sibling) - ); + // One extra eligible occasion is fetched to retain the omission indication. + assert_eq!(utilization.omitted_by_fanout_count, 1); }; assert_lifecycle_evidence(&first); // Suppression after the occasion budget fills must not inflate lifecycle evidence. @@ -785,14 +783,8 @@ async fn assert_participant_recall_after_suppression(active_sibling: bool) { continue; } assert_eq!(row.selected_cap, 1); - assert_eq!( - row.omitted_by_fanout_count, - if row.relation == RelationType::Involves { - 8 + usize::from(forget_last) - } else { - 9 - } - ); + // The fetched prefix contains lifecycle exclusions and no omitted eligible route. + assert_eq!(row.omitted_by_fanout_count, 0); } } memory.close().await.unwrap(); diff --git a/src/memory/write_turn_tests.rs b/src/memory/write_turn_tests.rs index f4c373cb..713c09e5 100644 --- a/src/memory/write_turn_tests.rs +++ b/src/memory/write_turn_tests.rs @@ -636,8 +636,9 @@ impl GraphAuthorityStore for GatedGraph { &self, key: &ScopeKey, policy: GraphExpansionLifecyclePolicy, + limit: usize, ) -> Result<(Vec, Vec), CustomError> { - self.store.query_scope_state(key, policy).await + self.store.query_scope_state(key, policy, limit).await } async fn expand_bounded( diff --git a/src/policy/graph_expansion.rs b/src/policy/graph_expansion.rs index 5da4ada9..6a8bb54a 100644 --- a/src/policy/graph_expansion.rs +++ b/src/policy/graph_expansion.rs @@ -227,7 +227,7 @@ pub(crate) fn derived_memories_by_thread( .is_none() }) .filter(|memory| { - if query.current_state && resolved.contains_key(&memory.id) { + if query.current_state_limit.is_some() && resolved.contains_key(&memory.id) { push_filtered_node( &mut filtered, MemoryObjectRef::new(ObjectType::DerivedMemory, memory.id), @@ -546,10 +546,9 @@ fn bounded_expansion_plan<'a>( }) .collect::>(); incident_links.sort_by_key(|(link, _)| stable_link_key(link)); - if depth == 0 { + if depth == 0 && query.current_thread_state { incident_links.retain(|(link, neighbor)| { - link.relation != RelationType::PartOfThread - || !query.resolved_thread_members.contains(neighbor) + link.relation != RelationType::PartOfThread || !resolved.contains_key(&neighbor.id) }); } if depth == 0 && query.current_subject_state { @@ -836,7 +835,7 @@ pub(crate) fn bounded_hub_retention_limit( } } -fn fanout_limit_for_pair( +pub(crate) fn fanout_limit_for_pair( query: &GraphExpansionQuery, relation: RelationType, object_type: ObjectType, diff --git a/src/ports/graph_authority.rs b/src/ports/graph_authority.rs index f366fa50..8102caaa 100644 --- a/src/ports/graph_authority.rs +++ b/src/ports/graph_authority.rs @@ -37,7 +37,7 @@ pub(crate) struct GraphDerivedMemoryThreadQuery { pub(crate) thread_ids: Vec, pub(crate) lifecycle_policy: GraphExpansionLifecyclePolicy, // Activity state excludes resolved members; lifecycle operations still see them. - pub(crate) current_state: bool, + pub(crate) current_state_limit: Option, } impl GraphDerivedMemoryProvenanceQuery { @@ -63,7 +63,7 @@ impl GraphDerivedMemoryThreadQuery { Self { thread_ids, lifecycle_policy: GraphExpansionLifecyclePolicy::default(), - current_state: false, + current_state_limit: None, } } @@ -157,7 +157,7 @@ pub(crate) struct GraphExpansionQuery { pub(crate) current_subject_state: bool, pub(crate) reminder_only: bool, pub(crate) participant_reference_time: DateTime, - pub(crate) resolved_thread_members: std::collections::HashSet, + pub(crate) current_thread_state: bool, // Hydrated lifecycle evidence may lie outside the adapter's selected traversal. pub(crate) traversal_link_ids: Option>, pub(crate) trace_mode: TraceMode, @@ -192,7 +192,7 @@ impl GraphExpansionQuery { current_subject_state: false, reminder_only: false, participant_reference_time: DateTime::::MAX_UTC, - resolved_thread_members: std::collections::HashSet::new(), + current_thread_state: false, traversal_link_ids: None, trace_mode: TraceMode::Disabled, lifecycle_policy: GraphExpansionLifecyclePolicy::default(), @@ -421,6 +421,7 @@ pub(crate) trait GraphAuthorityStore: Send + Sync { &self, key: &ScopeKey, policy: GraphExpansionLifecyclePolicy, + limit: usize, ) -> Result<(Vec, Vec), CustomError>; async fn query_last_interaction( @@ -541,8 +542,9 @@ impl GraphAuthorityStore for Box { &self, key: &ScopeKey, policy: GraphExpansionLifecyclePolicy, + limit: usize, ) -> Result<(Vec, Vec), CustomError> { - (**self).query_scope_state(key, policy).await + (**self).query_scope_state(key, policy, limit).await } async fn expand_bounded( diff --git a/src/usecases/correct_forget.rs b/src/usecases/correct_forget.rs index b5044a3f..1899b38b 100644 --- a/src/usecases/correct_forget.rs +++ b/src/usecases/correct_forget.rs @@ -3632,8 +3632,9 @@ mod tests { &self, key: &ScopeKey, policy: GraphExpansionLifecyclePolicy, + limit: usize, ) -> Result<(Vec, Vec), CustomError> { - self.store.query_scope_state(key, policy).await + self.store.query_scope_state(key, policy, limit).await } async fn expand_bounded( diff --git a/src/usecases/link.rs b/src/usecases/link.rs index 24109a08..554d095e 100644 --- a/src/usecases/link.rs +++ b/src/usecases/link.rs @@ -670,8 +670,9 @@ mod tests { &self, key: &ScopeKey, policy: GraphExpansionLifecyclePolicy, + limit: usize, ) -> Result<(Vec, Vec), CustomError> { - let _ = (key, policy); + let _ = (key, policy, limit); unreachable!("scope selector is not used by this failure fixture") } diff --git a/src/usecases/remember.rs b/src/usecases/remember.rs index b2dbe9f2..7b927dfb 100644 --- a/src/usecases/remember.rs +++ b/src/usecases/remember.rs @@ -1332,8 +1332,9 @@ mod tests { &self, key: &ScopeKey, policy: GraphExpansionLifecyclePolicy, + limit: usize, ) -> Result<(Vec, Vec), CustomError> { - self.store.query_scope_state(key, policy).await + self.store.query_scope_state(key, policy, limit).await } async fn expand_bounded( diff --git a/src/usecases/retrieve.rs b/src/usecases/retrieve.rs index 5604ab60..7b276189 100644 --- a/src/usecases/retrieve.rs +++ b/src/usecases/retrieve.rs @@ -133,6 +133,7 @@ where include_suppressed: context.lifecycle_policy.include_suppressed, include_superseded: context.lifecycle_policy.include_superseded, }, + context.candidate_limits.max_graph_roots, ) .await?; assembly @@ -144,12 +145,7 @@ where &entry.superseded_by, ) })); - // Lifecycle filtering and successor deduplication precede this cap. - for (rank, id) in ids - .into_iter() - .take(context.candidate_limits.max_graph_roots) - .enumerate() - { + for (rank, id) in ids.into_iter().enumerate() { root_order.insert( ( cues.participants.len() + offset, @@ -176,10 +172,6 @@ where } } let (activity, activity_roots, filtered) = self.activity_roots(&context).await?; - let resolved_thread_members = filtered - .iter() - .map(|entry| entry.object_ref) - .collect::>(); assembly .lifecycle_decisions .extend(filtered.into_iter().map(|entry| { @@ -369,8 +361,8 @@ where if context.activity == Some(crate::api::types::ActivityRef::Thread(candidate.object_id)) && candidate.object_type == ObjectType::MemoryThread { - // Reuse the activity selector's omitted members; no second state read. - query.resolved_thread_members = resolved_thread_members.clone(); + // Filter traversal itself: the bounded audit cannot list every resolved member. + query.current_thread_state = true; } graph_expansion_telemetry.attempted_root_count += 1; match self.graph_store.expand_bounded(&query).await { @@ -3598,8 +3590,9 @@ mod tests { &self, key: &ScopeKey, policy: GraphExpansionLifecyclePolicy, + limit: usize, ) -> Result<(Vec, Vec), CustomError> { - let _ = (key, policy); + let _ = (key, policy, limit); unreachable!("scope selector is not used by this failure fixture") } diff --git a/src/usecases/retrieve/activity.rs b/src/usecases/retrieve/activity.rs index 6c8f1c5b..b0263663 100644 --- a/src/usecases/retrieve/activity.rs +++ b/src/usecases/retrieve/activity.rs @@ -84,7 +84,7 @@ where include_suppressed: context.lifecycle_policy.include_suppressed, include_superseded: context.lifecycle_policy.include_superseded, }); - query.current_state = true; + query.current_state_limit = Some(context.candidate_limits.max_graph_roots); let (mut memories, omitted) = self .graph_store .query_derived_memories_by_thread(&query) From 445b69a16c6b419fd18c2d4ed1a2975c21592471 Mon Sep 17 00:00:00 2001 From: ebigunso Date: Wed, 23 Sep 2026 07:21:05 +0900 Subject: [PATCH 06/13] =?UTF-8?q?=F0=9F=90=9B=20Keep=20resolution=20eviden?= =?UTF-8?q?ce=20scoped=20to=20derived=20memories?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/policy/graph_expansion.rs | 4 ++- src/usecases/retrieve.rs | 53 ++++++++++++++++++++++++++--- tests/retrieval_resolution_tests.rs | 49 ++++++++++++++++++++++++++ 3 files changed, 101 insertions(+), 5 deletions(-) diff --git a/src/policy/graph_expansion.rs b/src/policy/graph_expansion.rs index 6a8bb54a..6ce39be1 100644 --- a/src/policy/graph_expansion.rs +++ b/src/policy/graph_expansion.rs @@ -548,7 +548,9 @@ fn bounded_expansion_plan<'a>( incident_links.sort_by_key(|(link, _)| stable_link_key(link)); if depth == 0 && query.current_thread_state { incident_links.retain(|(link, neighbor)| { - link.relation != RelationType::PartOfThread || !resolved.contains_key(&neighbor.id) + link.relation != RelationType::PartOfThread + || neighbor.object_type != ObjectType::DerivedMemory + || !resolved.contains_key(&neighbor.id) }); } if depth == 0 && query.current_subject_state { diff --git a/src/usecases/retrieve.rs b/src/usecases/retrieve.rs index 7b276189..90b2fbe8 100644 --- a/src/usecases/retrieve.rs +++ b/src/usecases/retrieve.rs @@ -702,10 +702,12 @@ impl RetrieveAssembly { ranked.is_root = object_ref == candidate_ref; ranked }); - if let Some(resolvers) = expansion.resolved_by.get(&object_ref.id) { - ranked.resolved_by.extend(resolvers); - ranked.resolved_by.sort_unstable(); - ranked.resolved_by.dedup(); + if object_ref.object_type == ObjectType::DerivedMemory { + if let Some(resolvers) = expansion.resolved_by.get(&object_ref.id) { + ranked.resolved_by.extend(resolvers); + ranked.resolved_by.sort_unstable(); + ranked.resolved_by.dedup(); + } } } @@ -1853,6 +1855,49 @@ mod tests { TemporaryVectorCandidateStore, }; + #[test] + fn resolution_metadata_stays_on_derived_memory_when_ids_collide() { + let fixtures = representative_fixtures(); + let entity = fixtures.hub_entity; + let mut memory = fixtures.open_loop; + memory.id = entity.id; + let resolver = fixtures.correction.id; + let candidate = CandidateRoot { + object_id: entity.id, + object_type: ObjectType::Entity, + score: 1.0, + source: GraphRootSource::Participant, + vector_score: None, + cue_kinds: BTreeSet::from([CueKind::Participant]), + full_standing_score: Some(1.0), + full_standing_kinds: BTreeSet::from([CueKind::Participant]), + date_match_floor_eligible: false, + }; + let query = GraphExpansionQuery::new(entity.id, ObjectType::Entity, 1, 2); + let mut expansion = GraphExpansion::new( + vec![ + MemoryObject::Entity(entity.clone()), + MemoryObject::DerivedMemory(memory), + ], + Vec::new(), + ); + expansion.resolved_by.insert(entity.id, vec![resolver]); + let mut assembly = RetrieveAssembly::new(TraceMode::Disabled); + assembly + .absorb_expansion(&candidate, &query, expansion) + .unwrap(); + assert!( + assembly.objects[&MemoryObjectRef::new(ObjectType::Entity, entity.id)] + .resolved_by + .is_empty() + ); + assert_eq!( + assembly.objects[&MemoryObjectRef::new(ObjectType::DerivedMemory, entity.id)] + .resolved_by, + vec![resolver] + ); + } + #[test] fn single_kind_reserves_own_head_then_fills_final_ranked_room() { let objects = diff --git a/tests/retrieval_resolution_tests.rs b/tests/retrieval_resolution_tests.rs index d6c1002d..ca62ce6a 100644 --- a/tests/retrieval_resolution_tests.rs +++ b/tests/retrieval_resolution_tests.rs @@ -543,3 +543,52 @@ async fn recalled_resolution_has_no_omission_but_section_capped_resolution_does( memory.close().await.unwrap(); root.close().unwrap(); } + +#[tokio::test] +async fn thread_resolution_keeps_same_id_observation_on_an_independent_route() { + let (memory, root) = fixture(false).await; + let mut observation = + character_memory::ObservationDraft::new(id(101), "independent observed fact"); + observation.id = Some(id(301)); + observation.created_at = Some(at(11)); + commit( + &memory, + RememberInput::new("observation thread membership") + .with_thread_id(id(601)) + .with_observation(observation), + ) + .await; + let mut association = MemoryLinkDraft::new( + ObjectType::MemoryThread, + id(601), + RelationType::AssociatedWith, + ObjectType::DerivedMemory, + id(301), + ); + association.id = Some(id(9001)); + association.created_at = Some(at(31)); + memory.link(association).await.unwrap(); + let mut missing = Vec::new(); + for trace in [false, true] { + let mut context = request("thread_expansion"); + context.include_trace = trace; + context.candidate_limits.max_graph_roots = 1; + context.graph_limits.allowed_relation_types = + vec![RelationType::PartOfThread, RelationType::AssociatedWith]; + let outcome = memory.retrieve(context).await.unwrap(); + if !outcome + .pack + .salient_observations + .iter() + .any(|object| object.id == id(301)) + { + missing.push(trace); + } + } + memory.close().await.unwrap(); + root.close().unwrap(); + assert!( + missing.is_empty(), + "Observation 301 lost for trace modes {missing:?}; resolution belongs to DerivedMemory 301" + ); +} From 800b96e13a4c9319134cd3e9e26b268ac8e03c60 Mon Sep 17 00:00:00 2001 From: ebigunso Date: Wed, 23 Sep 2026 07:26:12 +0900 Subject: [PATCH 07/13] =?UTF-8?q?=F0=9F=90=9B=20Exclude=20trace=20evidence?= =?UTF-8?q?=20from=20retrieval=20hub=20accounting?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/design/database/graph_schema_design.md | 2 + src/adapters/oxigraph/tests.rs | 56 +++++++++++++++++++++ src/policy/graph_expansion.rs | 26 +++++++--- tests/retrieval_resolution_tests.rs | 42 ++++++++++++++++ 4 files changed, 119 insertions(+), 7 deletions(-) diff --git a/docs/design/database/graph_schema_design.md b/docs/design/database/graph_schema_design.md index b4a55855..294cd495 100644 --- a/docs/design/database/graph_schema_design.md +++ b/docs/design/database/graph_schema_design.md @@ -124,6 +124,8 @@ Pack expansion reads bounded occasion prefixes and exclusion evidence separately State reads also apply lifecycle eligibility before their caller's query limit; subject state additionally requires a traversable About link. Exclusion evidence has a separate latest-first prefix of the same budget. Fanout omission telemetry counts omissions within fetched prefixes, without a whole-store count query. Thread expansion excludes resolved members independently of that bounded evidence. +The hub limit counts the fetched, eligibility-pruned traversal prefix, excluding trace-only omission evidence, so a larger stored history alone does not cause a hub failure. + ## Cross-Store Failures Graph writes are critical. Vector and stats updates are repairable parts of the write outcome, with typed failures and affected IDs. A graph-only memory loses semantic recall until vector indexing is repaired; stale candidates are checked against graph existence and lifecycle before inclusion. Replaying an old plan does not re-index an interpreted memory that graph authority identifies as superseded. diff --git a/src/adapters/oxigraph/tests.rs b/src/adapters/oxigraph/tests.rs index 80f7081f..6f7a3034 100644 --- a/src/adapters/oxigraph/tests.rs +++ b/src/adapters/oxigraph/tests.rs @@ -1448,6 +1448,62 @@ mod tests { assert_eq!(MAX_SELECT_ROWS.with(|count| count.get()), 0); } + #[tokio::test] + async fn participant_hub_limit_counts_only_the_admitted_occasion_prefix() { + use crate::ports::graph_authority::TraceMode; + let store = OxigraphGraphAuthorityStore::new_in_memory().unwrap(); + let fixtures = representative_fixtures(); + let mut objects = vec![MemoryObject::Entity(fixtures.hub_entity.clone())]; + let mut links = Vec::new(); + for index in 0..4_u128 { + let mut episode = fixtures.episode.clone(); + episode.id = MemoryId::from_u128(20_000 + index); + episode.scene.time += chrono::Duration::minutes(index as i64); + links.push( + crate::MemoryLinkDraft::new( + ObjectType::Episode, + episode.id, + RelationType::Involves, + ObjectType::Entity, + fixtures.hub_entity.id, + ) + .into_domain() + .unwrap(), + ); + objects.push(MemoryObject::Episode(episode)); + } + store.upsert_objects(&objects).await.unwrap(); + store.upsert_links(&links).await.unwrap(); + let mut outputs = Vec::new(); + for trace in [TraceMode::Disabled, TraceMode::Enabled] { + let query = GraphExpansionQuery::new(fixtures.hub_entity.id, ObjectType::Entity, 1, 10) + .with_max_hub_edges(1) + .with_max_fanout_per_node(1) + .with_failure_policy(GraphExpansionFailurePolicy { + timeout_ms: None, + mode: crate::domain::GraphFailureMode::FailClosed, + }) + .with_fanout_overrides(vec![GraphExpansionFanoutOverride { + relation: RelationType::Involves, + object_type: ObjectType::Episode, + max_fanout: 1, + }]) + .with_fanout_utilization_recording(trace); + let expansion = store.expand_bounded(&query).await.unwrap(); + assert!(expansion.bounded_failure.is_none()); + assert!(expansion + .objects + .iter() + .any(|object| object.id() == MemoryId::from_u128(20_003))); + outputs.push(( + expansion.objects, + expansion.links, + expansion.selection_order, + )); + } + assert_eq!(outputs[0], outputs[1]); + } + #[tokio::test] async fn thread_expansion_excludes_resolved_members_outside_its_trace_prefix() { let store = OxigraphGraphAuthorityStore::new_in_memory().unwrap(); diff --git a/src/policy/graph_expansion.rs b/src/policy/graph_expansion.rs index 6ce39be1..953c8556 100644 --- a/src/policy/graph_expansion.rs +++ b/src/policy/graph_expansion.rs @@ -1052,6 +1052,25 @@ pub(crate) fn bounded_incident_link_refs( }) }); + let exclusions = limit_participant_occasions( + query, + &mut incident_links, + root_fanout_mode, + occasions, + |link| (link.relation(), link.other_endpoint(object_ref)), + ); + if depth == 0 && query.current_subject_state && query.trace_mode.is_enabled() { + // The selector's extra state row is omission evidence, not a traversal edge. + let cap = fanout_limit_for_pair(query, RelationType::About, ObjectType::DerivedMemory); + let mut selected = HashSet::new(); + incident_links.retain(|link| { + let neighbor = link.other_endpoint(object_ref); + link.relation() != RelationType::About + || neighbor.object_type != ObjectType::DerivedMemory + || selected.contains(&neighbor.id) + || (selected.len() < cap && selected.insert(neighbor.id)) + }); + } let exceeds_hub_limit = incident_links.len() > query.max_hub_edges; if exceeds_hub_limit { let failure = GraphExpansionBoundedFailure { @@ -1063,13 +1082,6 @@ pub(crate) fn bounded_incident_link_refs( } bounded_failure.get_or_insert(failure); } - let exclusions = limit_participant_occasions( - query, - &mut incident_links, - root_fanout_mode, - occasions, - |link| (link.relation(), link.other_endpoint(object_ref)), - ); if exceeds_hub_limit { incident_links.truncate(bounded_hub_retention_limit(query, root_fanout_mode)); } diff --git a/tests/retrieval_resolution_tests.rs b/tests/retrieval_resolution_tests.rs index ca62ce6a..e80106ce 100644 --- a/tests/retrieval_resolution_tests.rs +++ b/tests/retrieval_resolution_tests.rs @@ -592,3 +592,45 @@ async fn thread_resolution_keeps_same_id_observation_on_an_independent_route() { "Observation 301 lost for trace modes {missing:?}; resolution belongs to DerivedMemory 301" ); } + +#[tokio::test] +async fn named_state_trace_does_not_change_recall_at_the_hub_limit() { + let (memory, root) = fixture(false).await; + let mut input = RememberInput::new("Two active subject memories"); + for n in [303, 304] { + let mut belief = derived(n, DerivedType::Claim, "active subject state", 101, n as i64); + belief.entity_ids.push(id(501)); + input = input.with_derived_memory(belief); + } + commit(&memory, input).await; + let mut packs = Vec::new(); + for trace in [false, true] { + let mut context = request("named"); + context.include_trace = trace; + context.candidate_limits.max_graph_roots = 1; + context.graph_limits.max_fanout_per_node = 1; + context.graph_limits.max_hub_edges = 1; + context.graph_limits.failure_mode = character_memory::GraphFailureMode::FailClosed; + context.graph_limits.allowed_relation_types = vec![RelationType::About]; + packs.push(memory.retrieve(context).await.map(|outcome| { + if trace { + assert!(outcome + .trace + .as_ref() + .unwrap() + .fanout_utilization + .iter() + .any(|entry| entry.relation == RelationType::About + && entry.omitted_by_fanout_count > 0)); + } + serde_json::to_value(outcome.pack).unwrap() + })); + } + memory.close().await.unwrap(); + root.close().unwrap(); + let packs = packs + .into_iter() + .collect::, _>>() + .expect("both trace modes must succeed at the hub boundary"); + assert_eq!(packs[0], packs[1], "trace mode must not change recall"); +} From 0f78ccbb8b096d954f32c89ff2b2481426b3701b Mon Sep 17 00:00:00 2001 From: ebigunso Date: Wed, 23 Sep 2026 07:27:18 +0900 Subject: [PATCH 08/13] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20Remove=20unreachable?= =?UTF-8?q?=20thread=20state=20filtering?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/policy/graph_expansion.rs | 21 +-------------------- 1 file changed, 1 insertion(+), 20 deletions(-) diff --git a/src/policy/graph_expansion.rs b/src/policy/graph_expansion.rs index 953c8556..3435cc54 100644 --- a/src/policy/graph_expansion.rs +++ b/src/policy/graph_expansion.rs @@ -204,11 +204,6 @@ pub(crate) fn derived_memories_by_thread( let links = links.into_iter().collect::>(); let link_refs = links.iter().collect::>(); let superseded = incoming_derived_memory_ids(&link_refs, &[RelationType::Supersedes]); - let resolved = incoming_derived_memory_ids( - &link_refs, - &[RelationType::Resolves, RelationType::FulfillsCommitment], - ); - let mut filtered = Vec::new(); let mut memories = objects .into_iter() .filter_map(|object| match object { @@ -226,24 +221,10 @@ pub(crate) fn derived_memories_by_thread( derived_memory_lifecycle_filter_reason(memory, &superseded, query.lifecycle_policy) .is_none() }) - .filter(|memory| { - if query.current_state_limit.is_some() && resolved.contains_key(&memory.id) { - push_filtered_node( - &mut filtered, - MemoryObjectRef::new(ObjectType::DerivedMemory, memory.id), - GraphExpansionFilteredReason::Resolved, - &superseded, - ); - false - } else { - true - } - }) .collect::>(); memories.sort_by_key(|memory| memory.id); - filtered.sort_by_key(|entry| entry.object_ref.id); - (memories, filtered) + (memories, Vec::new()) } fn derived_memory_matches_provenance( From b21c57da799882d905aba11ed555a029c8573422 Mon Sep 17 00:00:00 2001 From: ebigunso Date: Wed, 23 Sep 2026 07:29:45 +0900 Subject: [PATCH 09/13] =?UTF-8?q?=F0=9F=90=9B=20Share=20the=20clamped=20pa?= =?UTF-8?q?rticipant=20occasion=20budget?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/adapters/oxigraph/sparql_selectors.rs | 15 +----- src/policy/graph_expansion.rs | 64 ++++++++++++++++++++--- 2 files changed, 59 insertions(+), 20 deletions(-) diff --git a/src/adapters/oxigraph/sparql_selectors.rs b/src/adapters/oxigraph/sparql_selectors.rs index da3dca46..30bc416e 100644 --- a/src/adapters/oxigraph/sparql_selectors.rs +++ b/src/adapters/oxigraph/sparql_selectors.rs @@ -607,19 +607,8 @@ impl<'a> SparqlGraphSelectors<'a> { neighbors: &[MemoryObjectRef], query: &crate::ports::graph_authority::GraphExpansionQuery, ) -> Result { - let budget = query - .fanout_overrides - .iter() - .filter(|entry| { - crate::policy::graph_expansion::is_participant_pair( - entry.relation, - entry.object_type, - ) - }) - .map(|entry| entry.max_fanout) - .max() - .unwrap_or(0) - .min(query.max_fanout_per_node); + let budget = + crate::policy::graph_expansion::participant_occasion_budget(query).unwrap_or(0); if neighbors.is_empty() || budget == 0 { return Ok(ParticipantOccasions::new()); } diff --git a/src/policy/graph_expansion.rs b/src/policy/graph_expansion.rs index 3435cc54..894d373f 100644 --- a/src/policy/graph_expansion.rs +++ b/src/policy/graph_expansion.rs @@ -1101,6 +1101,15 @@ pub(crate) fn is_participant_pair(relation: RelationType, object_type: ObjectTyp ) } +pub(crate) fn participant_occasion_budget(query: &GraphExpansionQuery) -> Option { + query + .fanout_overrides + .iter() + .filter(|entry| is_participant_pair(entry.relation, entry.object_type)) + .map(|entry| entry.max_fanout.min(query.max_fanout_per_node)) + .max() +} + #[derive(Default)] struct ParticipantExclusions { // The excluded route's type can differ from its suppressed parent object's type. @@ -1136,13 +1145,7 @@ fn limit_participant_occasions( if !root_fanout_mode.applies_selectivity() || occasions.is_empty() { return exclusions; } - let Some(budget) = query - .fanout_overrides - .iter() - .filter(|entry| is_participant_pair(entry.relation, entry.object_type)) - .map(|entry| entry.max_fanout) - .max() - else { + let Some(budget) = participant_occasion_budget(query) else { return exclusions; }; let is_participant = |item: &T| { @@ -1253,6 +1256,53 @@ mod tests { use crate::test_support::{high_fanout_graph_fixture, representative_fixtures}; + #[test] + fn participant_occasion_budget_clamps_across_route_types() { + let episode = MemoryObjectRef::new(ObjectType::Episode, MemoryId::from_u128(1)); + let observation = MemoryObjectRef::new(ObjectType::Observation, MemoryId::from_u128(2)); + let query = GraphExpansionQuery::new(MemoryId::from_u128(3), ObjectType::Entity, 1, 10) + .with_max_fanout_per_node(1) + .with_fanout_overrides(vec![ + GraphExpansionFanoutOverride { + relation: RelationType::Involves, + object_type: ObjectType::Episode, + max_fanout: 2, + }, + GraphExpansionFanoutOverride { + relation: RelationType::Mentions, + object_type: ObjectType::Observation, + max_fanout: 2, + }, + ]); + let time = representative_fixtures().episode.scene.time.to_utc(); + let occasions = [episode, observation] + .into_iter() + .map(|object| { + ( + object, + ParticipantOccasion { + episode_id: object.id, + time: time + chrono::Duration::minutes(object.id.as_u128() as i64), + retention_state: RetentionState::Active, + episode_retention_state: RetentionState::Active, + }, + ) + }) + .collect(); + let mut items = vec![ + (RelationType::Involves, episode), + (RelationType::Mentions, observation), + ]; + limit_participant_occasions( + &query, + &mut items, + RootFanoutMode::SelectivityOverrides, + &occasions, + |item| *item, + ); + assert_eq!(items, vec![(RelationType::Mentions, observation)]); + } + #[test] fn hub_retention_limit_keeps_large_window_only_for_root_selectivity_overrides() { let root_id = MemoryId::new_v4(); From d685baa08ab6ee78d4f85359591bb93f1ea3e3f4 Mon Sep 17 00:00:00 2001 From: ebigunso Date: Wed, 23 Sep 2026 07:31:26 +0900 Subject: [PATCH 10/13] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20Leave=20activity=20m?= =?UTF-8?q?ember=20ordering=20to=20the=20activity=20caller?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/adapters/oxigraph/embedded.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/adapters/oxigraph/embedded.rs b/src/adapters/oxigraph/embedded.rs index b8d8971e..a5ce360b 100644 --- a/src/adapters/oxigraph/embedded.rs +++ b/src/adapters/oxigraph/embedded.rs @@ -310,14 +310,13 @@ impl GraphAuthorityStore for OxigraphGraphAuthorityStore { .map(|id| MemoryObjectRef::new(ObjectType::DerivedMemory, id)) .collect::>(), )?; - let mut memories = objects + let memories = objects .into_iter() .filter_map(|object| match object { MemoryObject::DerivedMemory(memory) => Some(memory), _ => None, }) .collect::>(); - memories.sort_by_key(|memory| (std::cmp::Reverse(memory.created_at), memory.id)); return Ok((memories, filtered)); } let selected_ids = SparqlGraphSelectors::new(&self.store) From 60ede0d218ab903de6dd6ff9ff0aaa950169e75a Mon Sep 17 00:00:00 2001 From: ebigunso Date: Wed, 23 Sep 2026 07:36:31 +0900 Subject: [PATCH 11/13] =?UTF-8?q?=E2=9A=A1=20Bound=20forget=20cascade=20li?= =?UTF-8?q?nk=20hydration=20to=20affected=20memories?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/adapters/oxigraph/embedded.rs | 16 +++- src/adapters/oxigraph/shared.rs | 2 + src/adapters/oxigraph/sparql_selectors.rs | 27 +++++++ src/adapters/oxigraph/tests.rs | 99 +++++++++++++++++++++++ 4 files changed, 142 insertions(+), 2 deletions(-) diff --git a/src/adapters/oxigraph/embedded.rs b/src/adapters/oxigraph/embedded.rs index a5ce360b..4afc6198 100644 --- a/src/adapters/oxigraph/embedded.rs +++ b/src/adapters/oxigraph/embedded.rs @@ -287,7 +287,13 @@ impl GraphAuthorityStore for OxigraphGraphAuthorityStore { .map(|id| MemoryObjectRef::from_id_type(id, ObjectType::DerivedMemory)) .collect::>(), )?; - let links = hydrate_all_links_from_store(&self.store)?; + let link_ids = SparqlGraphSelectors::new(&self.store).select_link_ids_touching( + &objects + .iter() + .map(MemoryObject::object_ref) + .collect::>(), + )?; + let links = hydrate_links_by_ids_from_store(&self.store, &link_ids)?; Ok(derived_memories_by_provenance( query, objects.into_iter().filter( @@ -332,7 +338,13 @@ impl GraphAuthorityStore for OxigraphGraphAuthorityStore { .map(|id| MemoryObjectRef::from_id_type(id, ObjectType::DerivedMemory)) .collect::>(), )?; - let links = hydrate_all_links_from_store(&self.store)?; + let link_ids = SparqlGraphSelectors::new(&self.store).select_link_ids_touching( + &objects + .iter() + .map(MemoryObject::object_ref) + .collect::>(), + )?; + let links = hydrate_links_by_ids_from_store(&self.store, &link_ids)?; Ok(derived_memories_by_thread( query, objects.into_iter().filter( diff --git a/src/adapters/oxigraph/shared.rs b/src/adapters/oxigraph/shared.rs index 2e61ce46..191d9477 100644 --- a/src/adapters/oxigraph/shared.rs +++ b/src/adapters/oxigraph/shared.rs @@ -284,6 +284,8 @@ fn rdf_subject_values_for_named_graph( None, Some(GraphNameRef::NamedNode(graph_name.as_ref())), ) { + #[cfg(test)] + RDF_QUADS_READ.with(|count| count.set(count.get() + 1)); let quad = quad.map_err(oxigraph_error)?; let NamedOrBlankNode::NamedNode(subject) = quad.subject else { continue; diff --git a/src/adapters/oxigraph/sparql_selectors.rs b/src/adapters/oxigraph/sparql_selectors.rs index 30bc416e..2bfe3b31 100644 --- a/src/adapters/oxigraph/sparql_selectors.rs +++ b/src/adapters/oxigraph/sparql_selectors.rs @@ -412,6 +412,33 @@ impl<'a> SparqlGraphSelectors<'a> { Ok((ranked_ids, filtered)) } + pub(crate) fn select_link_ids_touching( + &self, + object_refs: &[MemoryObjectRef], + ) -> Result, CustomError> { + if object_refs.is_empty() { + return Ok(Vec::new()); + } + let query = format!( + r#"SELECT DISTINCT ?id WHERE {{ + {nodes} + GRAPH ?link {{ + ?link a <{link_class}> ; <{object_id}> ?id . + {{ ?link <{from}> ?node . }} UNION {{ ?link <{to}> ?node . }} + }} + }} ORDER BY ?id"#, + nodes = sparql_node_iri_values("node", object_refs), + link_class = vocab::CLASS_MEMORY_LINK, + object_id = vocab::OBJECT_ID, + from = vocab::FROM, + to = vocab::TO, + ); + self.query_solutions(&query)? + .iter() + .map(|row| memory_id_binding(row, "id")) + .collect() + } + pub(crate) fn select_links_touching( &self, object_refs: &[MemoryObjectRef], diff --git a/src/adapters/oxigraph/tests.rs b/src/adapters/oxigraph/tests.rs index 6f7a3034..a51e5947 100644 --- a/src/adapters/oxigraph/tests.rs +++ b/src/adapters/oxigraph/tests.rs @@ -1504,6 +1504,105 @@ mod tests { assert_eq!(outputs[0], outputs[1]); } + #[tokio::test] + async fn forget_cascade_reads_only_affected_object_and_link_graphs() { + use super::super::shared::RDF_QUADS_READ; + let store = OxigraphGraphAuthorityStore::new_in_memory().unwrap(); + let fixtures = representative_fixtures(); + let mut objects = Vec::new(); + for index in 0..3_u128 { + let mut memory = fixtures.open_loop.clone(); + memory.id = MemoryId::from_u128(50 + index); + memory.thread_ids = vec![fixtures.soft_thread.id]; + memory.derived_from_episode_ids = vec![fixtures.episode.id]; + memory.derived_from_observation_ids.clear(); + if index == 2 { + memory.retention_state = RetentionState::Suppressed; + } + objects.push(MemoryObject::DerivedMemory(memory)); + } + store.upsert_objects(&objects).await.unwrap(); + // Lifecycle links still count when their source object is absent. + let links = + [(RelationType::Resolves, 50), (RelationType::Supersedes, 51)].map(|(relation, id)| { + let mut link = fixtures.soft_thread_link.clone(); + link.id = MemoryId::from_u128(900 + id); + link.from_type = ObjectType::DerivedMemory; + link.from_id = MemoryId::from_u128(999); + link.to_type = ObjectType::DerivedMemory; + link.to_id = MemoryId::from_u128(id); + link.relation = relation; + link + }); + store.upsert_links(&links).await.unwrap(); + let thread_query = GraphDerivedMemoryThreadQuery::by_threads(vec![fixtures.soft_thread.id]); + let provenance_query = + GraphDerivedMemoryProvenanceQuery::by_sources(vec![fixtures.episode.id], Vec::new()); + let mut reads = Vec::new(); + for unrelated in [false, true] { + if unrelated { + let mut extra_objects = Vec::new(); + let mut extra_links = Vec::new(); + for n in 0..12_u128 { + let mut entity = fixtures.hub_entity.clone(); + entity.id = MemoryId::from_u128(1000 + n); + extra_links.push( + crate::MemoryLinkDraft::new( + ObjectType::Entity, + entity.id, + RelationType::AssociatedWith, + ObjectType::Entity, + fixtures.hub_entity.id, + ) + .into_domain() + .unwrap(), + ); + extra_objects.push(MemoryObject::Entity(entity)); + } + store.upsert_objects(&extra_objects).await.unwrap(); + store.upsert_links(&extra_links).await.unwrap(); + } + RDF_QUADS_READ.with(|count| count.set(0)); + let (thread, filtered) = store + .query_derived_memories_by_thread(&thread_query) + .await + .unwrap(); + let thread_reads = RDF_QUADS_READ.with(|count| count.get()); + RDF_QUADS_READ.with(|count| count.set(0)); + let provenance = store + .query_derived_memories_by_provenance(&provenance_query) + .await + .unwrap(); + reads.push((thread_reads, RDF_QUADS_READ.with(|count| count.get()))); + for memories in [thread, provenance] { + assert_eq!( + memories.iter().map(|memory| memory.id).collect::>(), + vec![MemoryId::from_u128(50)] + ); + } + assert!(filtered.is_empty()); + } + assert_eq!( + reads[0], reads[1], + "unrelated graphs must not increase either cascade read: {reads:?}" + ); + RDF_QUADS_READ.with(|count| count.set(0)); + store + .query_derived_memories_by_thread( + &GraphDerivedMemoryThreadQuery::by_threads(Vec::new()), + ) + .await + .unwrap(); + store + .query_derived_memories_by_provenance(&GraphDerivedMemoryProvenanceQuery::by_sources( + Vec::new(), + Vec::new(), + )) + .await + .unwrap(); + assert_eq!(RDF_QUADS_READ.with(|count| count.get()), 0); + } + #[tokio::test] async fn thread_expansion_excludes_resolved_members_outside_its_trace_prefix() { let store = OxigraphGraphAuthorityStore::new_in_memory().unwrap(); From 806208548c2673bb81cef26b387fb5cc8b93b59f Mon Sep 17 00:00:00 2001 From: ebigunso Date: Wed, 23 Sep 2026 08:43:08 +0900 Subject: [PATCH 12/13] =?UTF-8?q?=E2=9A=A1=20Bound=20hydration=20after=20c?= =?UTF-8?q?ompact=20selector=20scans?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/design/database/graph_schema_design.md | 4 +- src/adapters/oxigraph/sparql_selectors.rs | 420 +++++++++++--------- src/adapters/oxigraph/tests.rs | 86 +++- 3 files changed, 313 insertions(+), 197 deletions(-) diff --git a/docs/design/database/graph_schema_design.md b/docs/design/database/graph_schema_design.md index 294cd495..dc6df0e1 100644 --- a/docs/design/database/graph_schema_design.md +++ b/docs/design/database/graph_schema_design.md @@ -120,9 +120,9 @@ Incoming `Resolves` and `FulfillsCommitment` links exclude targets from named-su Each resolved scene reference reports a `last_interactions` map keyed by notion ID, including every candidate of an ambiguous name. Values contain the last episode ID, recorded scene time and whole seconds elapsed relative to the retrieval scene; `null` means no eligible encounter at or before that time. Descriptions and unknown references have an empty map. [Scene resolution (`scene.rs:98`)](../../../src/usecases/retrieve/scene.rs#L98) collects these facts independently of pack caps and tracing. The [last-interaction selector (`sparql_selectors.rs:414`)](../../../src/adapters/oxigraph/sparql_selectors.rs#L414) uses one SPARQL query over `Involves` episode links and `Mentions` observation-to-parent-episode links, in either orientation. Both neighbor and parent retention checks, controlled by `include_suppressed`, and the scene-time cutoff precede `ORDER BY` descending timestamp, ascending episode ID and `LIMIT 1`. Scene-time strings are cast to timestamps for comparison, preserving fractional-time ordering. One result row crosses into Rust; the query engine may still examine all matching candidates. -Pack expansion reads bounded occasion prefixes and exclusion evidence separately. The [participant limiter](../../../src/policy/graph_expansion.rs) shares one episode budget across direct episode and observation routes, with at most one observation per selected episode. Eligibility and distinct-episode selection precede the query limit. Frequency counts distinct episodes, and selection prefers recorded scene time, latest first. Only participants recognized by key or name apply the retrieval scene-time cutoff here; beliefs and other retrieval paths have no new as-of rule. +Pack expansion reads bounded occasion prefixes and exclusion evidence separately. The [participant limiter](../../../src/policy/graph_expansion.rs) shares one episode budget across direct episode and observation routes, with at most one observation per selected episode. Compact occasion metadata is read first; Rust applies eligibility and distinct-episode limits before full RDF hydration. Frequency counts distinct episodes, and selection prefers recorded scene time, latest first. Only participants recognized by key or name apply the retrieval scene-time cutoff here; beliefs and other retrieval paths have no new as-of rule. -State reads also apply lifecycle eligibility before their caller's query limit; subject state additionally requires a traversable About link. Exclusion evidence has a separate latest-first prefix of the same budget. Fanout omission telemetry counts omissions within fetched prefixes, without a whole-store count query. Thread expansion excludes resolved members independently of that bounded evidence. +State reads scan keyed IDs and rank/lifecycle columns, then apply eligibility, deterministic ordering and the caller's budget in Rust before full RDF hydration; subject state additionally requires a traversable About link. Exclusion evidence has a separate latest-first prefix of the same budget. Fanout omission telemetry counts omissions within fetched prefixes, without a whole-store count query. Thread expansion excludes resolved members independently of that bounded evidence. The hub limit counts the fetched, eligibility-pruned traversal prefix, excluding trace-only omission evidence, so a larger stored history alone does not cause a hub failure. diff --git a/src/adapters/oxigraph/sparql_selectors.rs b/src/adapters/oxigraph/sparql_selectors.rs index 2bfe3b31..c30d1aa7 100644 --- a/src/adapters/oxigraph/sparql_selectors.rs +++ b/src/adapters/oxigraph/sparql_selectors.rs @@ -1,6 +1,6 @@ // Internal SPARQL selectors for the embedded Oxigraph authority. These helpers // return backend-neutral IDs/refs; canonical object hydration reads RDF state. -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use chrono::{DateTime, Utc}; use oxigraph::model::Term; @@ -226,30 +226,47 @@ impl<'a> SparqlGraphSelectors<'a> { limit: usize, read_more: bool, ) -> Result<(Vec, Vec), CustomError> { + if limit == 0 && !read_more { + return Ok((Vec::new(), Vec::new())); + } let subject = graph_uri(ObjectType::Entity, subject_id); - let scope = format!( - r#" - GRAPH ?memory {{ ?memory <{about}> <{subject}> . }} - GRAPH ?aboutLink {{ - ?aboutLink a <{link_class}> ; <{relation}> "about" . - {{ ?aboutLink <{from}> ?memory ; <{to}> <{subject}> ; <{from_type}> "derived_memory" ; <{to_type}> "entity" . }} - UNION {{ ?aboutLink <{to}> ?memory ; <{from}> <{subject}> ; <{to_type}> "derived_memory" ; <{from_type}> "entity" . }} - }} - "#, - about = vocab::ABOUT_ENTITY, + // Keep About membership separate from the keyed rank scan: joining every + // candidate to its links makes the planner repeat work before the budget. + let about_query = format!( + r#"SELECT DISTINCT ?memory WHERE {{ + {{ GRAPH ?g {{ ?link a <{link_class}> ; <{relation}> "about" ; + <{from}> ?memory ; <{from_type}> "derived_memory" ; + <{to}> <{subject}> ; <{to_type}> "entity" . }} }} + UNION {{ GRAPH ?g {{ ?link a <{link_class}> ; <{relation}> "about" ; + <{to}> ?memory ; <{to_type}> "derived_memory" ; + <{from}> <{subject}> ; <{from_type}> "entity" . }} }} + }}"#, link_class = vocab::CLASS_MEMORY_LINK, relation = vocab::RELATION, from = vocab::FROM, to = vocab::TO, from_type = vocab::FROM_TYPE, - to_type = vocab::TO_TYPE + to_type = vocab::TO_TYPE, ); + let traversable = self + .query_solutions(&about_query)? + .iter() + .map(|row| match row.get("memory") { + Some(Term::NamedNode(memory)) => { + super::shared::memory_id_from_resource(memory.as_str()) + } + value => Err(oxigraph_sparql_error(format!( + "expected About memory IRI, got {value:?}" + ))), + }) + .collect::, _>>()?; self.select_state( - &scope, + &format!("?memory <{}> <{subject}> .", vocab::ABOUT_ENTITY), policy, limit, read_more, - "?superseded DESC(xsd:double(?salience)) DESC(xsd:dateTime(?created)) ?id", + true, + Some(&traversable), ) } @@ -260,18 +277,12 @@ impl<'a> SparqlGraphSelectors<'a> { limit: usize, ) -> Result<(Vec, Vec), CustomError> { let value = serde_json::to_string(key).expect("scope keys contain strings only"); - let scope = format!( - "GRAPH ?memory {{ ?memory <{}> {} . }}", + let predicate = format!( + "?memory <{}> {} .", vocab::SCOPE_KEY, oxigraph::model::Literal::new_simple_literal(value) ); - self.select_state( - &scope, - policy, - limit, - false, - "?superseded DESC(xsd:double(?salience)) DESC(xsd:dateTime(?created)) ?id", - ) + self.select_state(&predicate, policy, limit, false, true, None) } pub(crate) fn select_thread_state( @@ -287,59 +298,51 @@ impl<'a> SparqlGraphSelectors<'a> { .iter() .map(|id| graph_uri(ObjectType::MemoryThread, *id)) .collect::>(); - let scope = format!( - "{} GRAPH ?memory {{ ?memory <{}> ?thread . }}", + let predicate = format!( + "{} ?memory <{}> ?thread .", sparql_iri_values("thread", threads.iter().map(String::as_str)), - vocab::PART_OF_THREAD + vocab::PART_OF_THREAD, ); self.select_state( - &scope, + &predicate, query.lifecycle_policy, limit, false, - "DESC(xsd:dateTime(?created)) ?id", + false, + None, ) } fn select_state( &self, - scope: &str, + scope_predicate: &str, policy: GraphExpansionLifecyclePolicy, limit: usize, read_more: bool, - order: &str, + salience_first: bool, + traversable: Option<&HashSet>, ) -> Result<(Vec, Vec), CustomError> { let eligible_limit = limit.saturating_add(usize::from(read_more)); if eligible_limit == 0 { return Ok((Vec::new(), Vec::new())); } - // Join lifecycle links once; per-memory EXISTS repeats the same link scan. - let body = format!( - r#" - {scope} - GRAPH ?memory {{ - ?memory a <{derived_class}> ; <{object_id}> ?id ; - <{salience}> ?salience ; - <{created}> ?created ; <{retention}> ?retention . - }} - OPTIONAL {{ - GRAPH ?supersessionLink {{ - ?supersessionLink a <{link_class}> ; <{from_type}> "derived_memory" ; - <{to_type}> "derived_memory" ; <{relation}> "supersedes" ; <{to}> ?memory . - }} - }} - BIND(BOUND(?supersessionLink) AS ?superseded) - OPTIONAL {{ - GRAPH ?resolution {{ - ?resolution a <{link_class}> ; <{from_type}> "derived_memory" ; - <{to_type}> "derived_memory" ; <{relation}> ?resolutionKind ; <{to}> ?memory . - VALUES ?resolutionKind {{ "resolves" "fulfills_commitment" }} + // ponytail: scan keyed rank rows; reshape selectors if this compact scan + // grows costly. Full RDF hydration stays behind the eligible/evidence caps. + let query = format!( + r#"SELECT ?id ?retention ?salience ?created ?relation ?source WHERE {{ + GRAPH ?g {{ + ?memory a <{derived_class}> ; <{object_id}> ?id ; + <{salience}> ?salience ; <{created}> ?created ; + <{retention}> ?retention . + {scope_predicate} }} - }} - BIND(BOUND(?resolution) AS ?resolved) - BIND(({include_suppressed} || ?retention != "suppressed") && - ({include_superseded} || !?superseded) && !?resolved AS ?eligible) - "#, + OPTIONAL {{ GRAPH ?linkGraph {{ + ?link a <{link_class}> ; <{from_type}> "derived_memory" ; + <{to_type}> "derived_memory" ; <{relation}> ?relation ; + <{from}> ?source ; <{to}> ?memory . + VALUES ?relation {{ "supersedes" "resolves" "fulfills_commitment" }} + }} }} + }}"#, derived_class = vocab::CLASS_DERIVED_MEMORY, object_id = vocab::OBJECT_ID, salience = vocab::SALIENCE_SCORE, @@ -349,67 +352,109 @@ impl<'a> SparqlGraphSelectors<'a> { from_type = vocab::FROM_TYPE, to_type = vocab::TO_TYPE, relation = vocab::RELATION, + from = vocab::FROM, to = vocab::TO, - include_suppressed = policy.include_suppressed, - include_superseded = policy.include_superseded, ); - let prefix = "PREFIX xsd: "; - let ranked_ids = self - .query_solutions(&format!( - r#"{prefix} - SELECT DISTINCT ?id ?superseded ?salience ?created WHERE {{ {body} FILTER(?eligible) }} - ORDER BY {order} LIMIT {eligible_limit} - "# - ))? - .iter() - .map(|row| memory_id_binding(row, "id")) - .collect::, _>>()?; - if limit == 0 { - return Ok((ranked_ids, Vec::new())); + struct State { + id: MemoryId, + retention: RetentionState, + salience: f32, + created: DateTime, + superseded_by: Vec, + resolved: bool, } - // Exclusion evidence has its own latest-first budget; it cannot crowd out eligible state. - let excluded = format!( - r#"{prefix} - SELECT ?id ?retention ?created (GROUP_CONCAT(DISTINCT COALESCE(STR(?source), ""); SEPARATOR=" ") AS ?sources) WHERE {{ - {{ SELECT DISTINCT ?memory ?id ?retention ?created WHERE {{ {body} FILTER(!?eligible) }} - ORDER BY DESC(xsd:dateTime(?created)) ?id LIMIT {limit} }} - OPTIONAL {{ GRAPH ?supersession {{ - ?supersession a <{link_class}> ; <{from_type}> "derived_memory" ; - <{to_type}> "derived_memory" ; <{relation}> "supersedes" ; <{from}> ?source ; <{to}> ?memory . - }} }} - }} GROUP BY ?id ?retention ?created - ORDER BY DESC(xsd:dateTime(?created)) ?id LIMIT {limit} - "#, - link_class = vocab::CLASS_MEMORY_LINK, - from_type = vocab::FROM_TYPE, - to_type = vocab::TO_TYPE, - relation = vocab::RELATION, - from = vocab::FROM, - to = vocab::TO - ); - let mut filtered = Vec::new(); - for row in self.query_solutions(&excluded)? { + // Collect every lifecycle row before applying policy or the budget. + let mut states = HashMap::::new(); + for row in self.query_solutions(&query)? { let id = memory_id_binding(&row, "id")?; - let retention = enum_binding::(&row, "retention")?; - let mut superseded_by = literal_binding(&row, "sources")? - .split_whitespace() - .map(super::shared::memory_id_from_resource) - .collect::, _>>()?; - superseded_by.sort_unstable(); - let reason = if retention == RetentionState::Suppressed && !policy.include_suppressed { - GraphExpansionFilteredReason::Suppressed - } else if !superseded_by.is_empty() && !policy.include_superseded { - GraphExpansionFilteredReason::Superseded + if traversable.is_some_and(|ids| !ids.contains(&id)) { + continue; + } + if let std::collections::hash_map::Entry::Vacant(entry) = states.entry(id) { + entry.insert(State { + id, + retention: enum_binding(&row, "retention")?, + salience: literal_binding(&row, "salience")? + .parse() + .map_err(oxigraph_sparql_error)?, + created: literal_binding(&row, "created")? + .parse() + .map_err(oxigraph_sparql_error)?, + superseded_by: Vec::new(), + resolved: false, + }); + } + let state = states.get_mut(&id).unwrap(); + if let Some(source) = row.get("source") { + let Term::NamedNode(source) = source else { + return Err(oxigraph_sparql_error(format!( + "expected lifecycle source IRI, got {source}" + ))); + }; + match enum_binding::(&row, "relation")? { + RelationType::Supersedes => state + .superseded_by + .push(super::shared::memory_id_from_resource(source.as_str())?), + RelationType::Resolves | RelationType::FulfillsCommitment => { + state.resolved = true + } + _ => unreachable!("query restricts lifecycle relations"), + } + } + } + let mut eligible = Vec::new(); + let mut excluded = Vec::new(); + for mut state in states.into_values() { + state.superseded_by.sort_unstable(); + state.superseded_by.dedup(); + let reason = + if state.retention == RetentionState::Suppressed && !policy.include_suppressed { + Some(GraphExpansionFilteredReason::Suppressed) + } else if !state.superseded_by.is_empty() && !policy.include_superseded { + Some(GraphExpansionFilteredReason::Superseded) + } else if state.resolved { + Some(GraphExpansionFilteredReason::Resolved) + } else { + None + }; + if let Some(reason) = reason { + excluded.push(( + state.created, + GraphExpansionFilteredNode { + object_ref: MemoryObjectRef::new(ObjectType::DerivedMemory, state.id), + reason, + superseded_by: state.superseded_by, + }, + )); } else { - GraphExpansionFilteredReason::Resolved - }; - filtered.push(GraphExpansionFilteredNode { - object_ref: MemoryObjectRef::new(ObjectType::DerivedMemory, id), - reason, - superseded_by, - }); + eligible.push(state); + } } - Ok((ranked_ids, filtered)) + eligible.sort_by(|a, b| { + (if salience_first { + b.superseded_by + .is_empty() + .cmp(&a.superseded_by.is_empty()) + .then_with(|| b.salience.total_cmp(&a.salience)) + } else { + std::cmp::Ordering::Equal + }) + .then_with(|| b.created.cmp(&a.created)) + .then_with(|| a.id.cmp(&b.id)) + }); + excluded.sort_by_key(|(created, entry)| (std::cmp::Reverse(*created), entry.object_ref.id)); + Ok(( + eligible + .into_iter() + .take(eligible_limit) + .map(|state| state.id) + .collect(), + excluded + .into_iter() + .take(limit) + .map(|(_, entry)| entry) + .collect(), + )) } pub(crate) fn select_link_ids_touching( @@ -603,30 +648,10 @@ impl<'a> SparqlGraphSelectors<'a> { return Ok(ParticipantOccasions::new()); } let query_text = format!("SELECT DISTINCT ?id ?objectType ?episodeId ?sceneTime ?retention ?episodeRetention WHERE {{ {} }}", participant_occasion_pattern(neighbors)); - let mut occasions = ParticipantOccasions::new(); - for solution in self.query_solutions(&query_text)? { - let neighbor = MemoryObjectRef::from_id_type( - memory_id_binding(&solution, "id")?, - enum_binding(&solution, "objectType")?, - ); - let time = literal_binding(&solution, "sceneTime")? - .parse() - .map_err(|error| { - CustomError::DatabaseError(format!( - "Oxigraph SPARQL invalid Scene.time: {error}" - )) - })?; - occasions.insert( - neighbor, - ParticipantOccasion { - episode_id: memory_id_binding(&solution, "episodeId")?, - time, - retention_state: enum_binding(&solution, "retention")?, - episode_retention_state: enum_binding(&solution, "episodeRetention")?, - }, - ); - } - Ok(occasions) + self.query_solutions(&query_text)? + .iter() + .map(participant_occasion_binding) + .collect() } pub(crate) fn select_bounded_participant_occasions( @@ -639,57 +664,53 @@ impl<'a> SparqlGraphSelectors<'a> { if neighbors.is_empty() || budget == 0 { return Ok(ParticipantOccasions::new()); } - let prefix = "PREFIX xsd: "; - let cutoff = if query.current_subject_state { - format!( - "FILTER(?time <= {}^^xsd:dateTime)", - sparql_string_literal(&query.participant_reference_time.to_rfc3339()) - ) - } else { - String::new() - }; - let body = format!( - r#"{} - BIND(xsd:dateTime(?sceneTime) AS ?time) {cutoff} - BIND(({} || (?retention = "active" && ?episodeRetention = "active")) AS ?eligible) - "#, - participant_occasion_pattern(neighbors), - query.lifecycle_policy.include_suppressed - ); - let mut selected = Vec::new(); + let mut occasions = self.select_participant_occasions(neighbors)?; + occasions.retain(|_, occasion| { + !query.current_subject_state || occasion.time <= query.participant_reference_time + }); + let mut episodes = occasions + .iter() + .map(|(neighbor, occasion)| { + ( + occasion.time, + occasion.episode_id, + occasion + .filtered_reason(*neighbor, query.lifecycle_policy) + .is_none(), + ) + }) + .collect::>(); + episodes.sort_unstable_by_key(|(time, id, eligible)| { + (std::cmp::Reverse(*time), *id, *eligible) + }); + episodes.dedup(); + let mut selected_episodes = HashSet::new(); for eligible in [true, false] { let limit = budget.saturating_add(usize::from(eligible && query.trace_mode.is_enabled())); - // Reserve distinct occasions before choosing one stable route of each type. - let episodes = self - .query_solutions(&format!( - r#"{prefix} - SELECT DISTINCT ?episodeId ?time WHERE {{ {body} FILTER(?eligible = {eligible}) }} - ORDER BY DESC(?time) ?episodeId LIMIT {limit} - "# - ))? - .iter() - .map(|row| memory_id_binding(row, "episodeId")) - .collect::, _>>()?; - if episodes.is_empty() { - continue; - } - let values = - sparql_literal_values("episodeId", episodes.iter().map(ToString::to_string)); - for row in self.query_solutions(&format!( - r#"{prefix} - SELECT ?episodeId ?objectType (MIN(?rank) AS ?selectedRank) WHERE {{ - {body} {values} FILTER(?eligible = {eligible}) - }} GROUP BY ?episodeId ?objectType - "# - ))? { - let rank = literal_binding(&row, "selectedRank")? - .parse::() - .map_err(oxigraph_sparql_error)?; - selected.push(neighbors[rank]); - } + selected_episodes.extend( + episodes + .iter() + .filter(|(_, _, allowed)| *allowed == eligible) + .take(limit) + .map(|(_, id, _)| (*id, eligible)), + ); } - self.select_participant_occasions(&selected) + // Keep the first input route per episode/type/eligibility, as the graph + // limiter does; all following object hydration sees only these prefixes. + let mut seen_routes = HashSet::new(); + Ok(neighbors + .iter() + .filter_map(|neighbor| { + let occasion = occasions.remove(neighbor)?; + let eligible = occasion + .filtered_reason(*neighbor, query.lifecycle_policy) + .is_none(); + (selected_episodes.contains(&(occasion.episode_id, eligible)) + && seen_routes.insert((occasion.episode_id, neighbor.object_type, eligible))) + .then_some((*neighbor, occasion)) + }) + .collect()) } pub(crate) fn select_anniversaries( @@ -901,6 +922,8 @@ impl<'a> SparqlGraphSelectors<'a> { .map_err(oxigraph_sparql_error)?; #[cfg(test)] MAX_SELECT_ROWS.with(|count| count.set(count.get().max(rows.len()))); + #[cfg(test)] + SELECT_CALLS.with(|count| count.set(count.get() + 1)); Ok(rows) } } @@ -908,18 +931,37 @@ impl<'a> SparqlGraphSelectors<'a> { #[cfg(test)] thread_local! { pub(super) static MAX_SELECT_ROWS: std::cell::Cell = const { std::cell::Cell::new(0) }; + pub(super) static SELECT_CALLS: std::cell::Cell = const { std::cell::Cell::new(0) }; +} + +fn participant_occasion_binding( + solution: &QuerySolution, +) -> Result<(MemoryObjectRef, ParticipantOccasion), CustomError> { + let neighbor = MemoryObjectRef::from_id_type( + memory_id_binding(solution, "id")?, + enum_binding(solution, "objectType")?, + ); + let time = literal_binding(solution, "sceneTime")? + .parse() + .map_err(|error| { + CustomError::DatabaseError(format!("Oxigraph SPARQL invalid Scene.time: {error}")) + })?; + Ok(( + neighbor, + ParticipantOccasion { + episode_id: memory_id_binding(solution, "episodeId")?, + time, + retention_state: enum_binding(solution, "retention")?, + episode_retention_state: enum_binding(solution, "episodeRetention")?, + }, + )) } fn participant_occasion_pattern(neighbors: &[MemoryObjectRef]) -> String { - let values = neighbors - .iter() - .enumerate() - .map(|(rank, object)| format!("(<{}> {rank})", graph_uri(object.object_type, object.id))) - .collect::>() - .join(" "); + let values = sparql_node_iri_values("node", neighbors); format!( r#" - VALUES (?node ?rank) {{ {values} }} + {values} GRAPH ?node {{ ?node <{object_id}> ?id ; <{object_type}> ?objectType ; <{retention}> ?retention . }} {{ GRAPH ?node {{ ?node <{object_id}> ?episodeId ; <{scene_time}> ?sceneTime ; <{retention}> ?episodeRetention . }} diff --git a/src/adapters/oxigraph/tests.rs b/src/adapters/oxigraph/tests.rs index a51e5947..8cc1369a 100644 --- a/src/adapters/oxigraph/tests.rs +++ b/src/adapters/oxigraph/tests.rs @@ -1217,8 +1217,10 @@ mod tests { #[tokio::test] async fn retrieval_selectors_bound_state_and_occasion_prefixes() { - use super::super::sparql_selectors::{SparqlGraphSelectors, MAX_SELECT_ROWS}; - use crate::domain::ScopeKey; + use super::super::shared::RDF_QUADS_READ; + use super::super::sparql_selectors::{SparqlGraphSelectors, MAX_SELECT_ROWS, SELECT_CALLS}; + use crate::domain::{graph_uri, ScopeKey}; + use oxigraph::model::{GraphNameRef, NamedNode}; let store = OxigraphGraphAuthorityStore::new_in_memory().unwrap(); let fixtures = representative_fixtures(); let key = ScopeKey::Setting("home".to_owned()); @@ -1298,11 +1300,51 @@ mod tests { let selectors = SparqlGraphSelectors::new(&store.store); MAX_SELECT_ROWS.with(|count| count.set(0)); let (scope, scope_filtered) = selectors.select_scope_state(&key, policy, 3).unwrap(); - assert!(MAX_SELECT_ROWS.with(|count| count.get()) <= 3); let (subject, subject_filtered) = selectors .select_subject_state(fixtures.hub_entity.id, policy, 3, false) .unwrap(); - assert!(MAX_SELECT_ROWS.with(|count| count.get()) <= 3); + let quad_budget = |ids: &[MemoryId]| { + ids.iter() + .map(|id| { + let graph = NamedNode::new(graph_uri(ObjectType::DerivedMemory, *id)).unwrap(); + store + .store + .quads_for_pattern( + None, + None, + None, + Some(GraphNameRef::NamedNode(graph.as_ref())), + ) + .count() + }) + .sum::() + }; + // Rank scans may read all keyed IDs. Full objects and exclusion evidence + // remain bounded before canonical RDF hydration. + for (ids, filtered) in [(&scope, &scope_filtered), (&subject, &subject_filtered)] { + let bounded = ids + .iter() + .copied() + .chain(filtered.iter().map(|entry| entry.object_ref.id)) + .collect::>(); + assert_eq!(bounded.len(), 6); + RDF_QUADS_READ.with(|count| count.set(0)); + let hydrated = store + .query_objects(&GraphObjectQuery::by_refs( + bounded + .iter() + .map(|id| MemoryObjectRef::new(ObjectType::DerivedMemory, *id)) + .collect(), + )) + .await + .unwrap(); + assert_eq!(hydrated.len(), 6); + assert_eq!( + RDF_QUADS_READ.with(|count| count.get()), + quad_budget(&bounded) + ); + } + RDF_QUADS_READ.with(|count| count.set(0)); let mut query = GraphDerivedMemoryThreadQuery::by_threads(vec![fixtures.soft_thread.id]); query.current_state_limit = Some(3); let (mut thread, thread_filtered) = store @@ -1310,7 +1352,10 @@ mod tests { .await .unwrap(); thread.sort_by_key(|memory| (std::cmp::Reverse(memory.created_at), memory.id)); - assert!(MAX_SELECT_ROWS.with(|count| count.get()) <= 3); + assert_eq!( + RDF_QUADS_READ.with(|count| count.get()), + quad_budget(&thread.iter().map(|memory| memory.id).collect::>()) + ); let mut occasion_query = GraphExpansionQuery::new(fixtures.hub_entity.id, ObjectType::Entity, 1, 10) .with_fanout_overrides(vec![ @@ -1328,10 +1373,39 @@ mod tests { occasion_query.current_subject_state = true; occasion_query.participant_reference_time = fixtures.episode.scene.time.to_utc() + chrono::Duration::minutes(8); + MAX_SELECT_ROWS.with(|count| count.set(0)); + SELECT_CALLS.with(|count| count.set(0)); let occasions = selectors .select_bounded_participant_occasions(&neighbors, &occasion_query) .unwrap(); - assert!(MAX_SELECT_ROWS.with(|count| count.get()) <= 12); + assert_eq!(SELECT_CALLS.with(|count| count.get()), 1); + assert_eq!(MAX_SELECT_ROWS.with(|count| count.get()), neighbors.len()); + let occasion_refs = occasions.keys().copied().collect::>(); + let occasion_quad_budget = occasion_refs + .iter() + .map(|object| { + let graph = NamedNode::new(graph_uri(object.object_type, object.id)).unwrap(); + store + .store + .quads_for_pattern( + None, + None, + None, + Some(GraphNameRef::NamedNode(graph.as_ref())), + ) + .count() + }) + .sum::(); + RDF_QUADS_READ.with(|count| count.set(0)); + let hydrated_occasions = store + .query_objects(&GraphObjectQuery::by_refs(occasion_refs)) + .await + .unwrap(); + assert_eq!(hydrated_occasions.len(), 12); + assert_eq!( + RDF_QUADS_READ.with(|count| count.get()), + occasion_quad_budget + ); for entries in [&scope_filtered, &subject_filtered, &thread_filtered] { assert_eq!( entries From 9aaa242cd68352be288ba5138bff8c2b887990c1 Mon Sep 17 00:00:00 2001 From: ebigunso Date: Wed, 23 Sep 2026 09:03:36 +0900 Subject: [PATCH 13/13] =?UTF-8?q?=E2=9A=A1=20Filter=20thread=20resolution?= =?UTF-8?q?=20from=20compact=20member=20IDs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/adapters/oxigraph/sparql_selectors.rs | 105 +++++++++++++--------- src/adapters/oxigraph/tests.rs | 1 - 2 files changed, 63 insertions(+), 43 deletions(-) diff --git a/src/adapters/oxigraph/sparql_selectors.rs b/src/adapters/oxigraph/sparql_selectors.rs index c30d1aa7..dee6b107 100644 --- a/src/adapters/oxigraph/sparql_selectors.rs +++ b/src/adapters/oxigraph/sparql_selectors.rs @@ -494,25 +494,6 @@ impl<'a> SparqlGraphSelectors<'a> { } let node_values = sparql_node_iri_values("node", object_refs); - let thread_filter = if current_thread_state { - format!( - r#" - BIND(IF(?fromType = "derived_memory", ?from, ?to) AS ?member) - FILTER(?relation != "part_of_thread" || NOT EXISTS {{ GRAPH ?resolution {{ - ?resolution a <{link_class}> ; <{from_type}> "derived_memory" ; - <{to_type}> "derived_memory" ; <{relation}> ?resolutionKind ; <{to}> ?member . - VALUES ?resolutionKind {{ "resolves" "fulfills_commitment" }} - }} }}) - "#, - link_class = vocab::CLASS_MEMORY_LINK, - from_type = vocab::FROM_TYPE, - to_type = vocab::TO_TYPE, - relation = vocab::RELATION, - to = vocab::TO - ) - } else { - String::new() - }; let query_text = format!( r#" SELECT DISTINCT ?linkId ?fromId ?fromType ?toId ?toType ?relation WHERE {{ @@ -537,7 +518,6 @@ impl<'a> SparqlGraphSelectors<'a> { ?to <{object_id}> ?toId ; <{object_type}> ?toType . }} - {thread_filter} }} "#, link_class = vocab::CLASS_MEMORY_LINK, @@ -567,6 +547,51 @@ impl<'a> SparqlGraphSelectors<'a> { refs.push(link_ref); } } + if current_thread_state { + let mut members = refs + .iter() + .filter(|link| link.relation == RelationType::PartOfThread) + .flat_map(|link| [link.from, link.to]) + .filter(|object| object.object_type == ObjectType::DerivedMemory) + .collect::>(); + members.sort_unstable_by_key(|object| object.id); + members.dedup(); + if !members.is_empty() { + let query = format!( + r#"SELECT DISTINCT ?member WHERE {{ + {members} + GRAPH ?g {{ + ?link a <{link_class}> ; <{from_type}> "derived_memory" ; + <{to_type}> "derived_memory" ; <{relation}> ?kind ; <{to}> ?member . + VALUES ?kind {{ "resolves" "fulfills_commitment" }} + }} + }}"#, + members = sparql_node_iri_values("member", &members), + link_class = vocab::CLASS_MEMORY_LINK, + from_type = vocab::FROM_TYPE, + to_type = vocab::TO_TYPE, + relation = vocab::RELATION, + to = vocab::TO, + ); + let resolved = self + .query_solutions(&query)? + .iter() + .map(|row| match row.get("member") { + Some(Term::NamedNode(member)) => Ok(MemoryObjectRef::new( + ObjectType::DerivedMemory, + super::shared::memory_id_from_resource(member.as_str())?, + )), + value => Err(oxigraph_sparql_error(format!( + "expected resolved member IRI, got {value:?}" + ))), + }) + .collect::, _>>()?; + refs.retain(|link| { + link.relation != RelationType::PartOfThread + || (!resolved.contains(&link.from) && !resolved.contains(&link.to)) + }); + } + } refs.sort_by_key(|link_ref| { ( link_ref.to.id, @@ -647,7 +672,24 @@ impl<'a> SparqlGraphSelectors<'a> { if neighbors.is_empty() { return Ok(ParticipantOccasions::new()); } - let query_text = format!("SELECT DISTINCT ?id ?objectType ?episodeId ?sceneTime ?retention ?episodeRetention WHERE {{ {} }}", participant_occasion_pattern(neighbors)); + let values = sparql_node_iri_values("node", neighbors); + let query_text = format!( + r#"SELECT DISTINCT ?id ?objectType ?episodeId ?sceneTime ?retention ?episodeRetention WHERE {{ + {values} + GRAPH ?node {{ ?node <{object_id}> ?id ; <{object_type}> ?objectType ; <{retention}> ?retention . }} + {{ + GRAPH ?node {{ ?node <{object_id}> ?episodeId ; <{scene_time}> ?sceneTime ; <{retention}> ?episodeRetention . }} + }} UNION {{ + GRAPH ?node {{ ?node <{episode}> ?episode . }} + GRAPH ?episode {{ ?episode <{object_id}> ?episodeId ; <{scene_time}> ?sceneTime ; <{retention}> ?episodeRetention . }} + }} + }}"#, + object_id = vocab::OBJECT_ID, + object_type = vocab::OBJECT_TYPE, + retention = vocab::RETENTION_STATE, + scene_time = vocab::SCENE_TIME, + episode = vocab::EPISODE, + ); self.query_solutions(&query_text)? .iter() .map(participant_occasion_binding) @@ -957,27 +999,6 @@ fn participant_occasion_binding( )) } -fn participant_occasion_pattern(neighbors: &[MemoryObjectRef]) -> String { - let values = sparql_node_iri_values("node", neighbors); - format!( - r#" - {values} - GRAPH ?node {{ ?node <{object_id}> ?id ; <{object_type}> ?objectType ; <{retention}> ?retention . }} - {{ - GRAPH ?node {{ ?node <{object_id}> ?episodeId ; <{scene_time}> ?sceneTime ; <{retention}> ?episodeRetention . }} - }} UNION {{ - GRAPH ?node {{ ?node <{episode}> ?episode . }} - GRAPH ?episode {{ ?episode <{object_id}> ?episodeId ; <{scene_time}> ?sceneTime ; <{retention}> ?episodeRetention . }} - }} - "#, - object_id = vocab::OBJECT_ID, - object_type = vocab::OBJECT_TYPE, - retention = vocab::RETENTION_STATE, - scene_time = vocab::SCENE_TIME, - episode = vocab::EPISODE - ) -} - fn occasion_retention_filter(policy: GraphExpansionLifecyclePolicy) -> &'static str { if policy.include_suppressed { "" diff --git a/src/adapters/oxigraph/tests.rs b/src/adapters/oxigraph/tests.rs index 8cc1369a..f91c428c 100644 --- a/src/adapters/oxigraph/tests.rs +++ b/src/adapters/oxigraph/tests.rs @@ -1298,7 +1298,6 @@ mod tests { store.upsert_links(&links).await.unwrap(); let policy = GraphExpansionLifecyclePolicy::default(); let selectors = SparqlGraphSelectors::new(&store.store); - MAX_SELECT_ROWS.with(|count| count.set(0)); let (scope, scope_filtered) = selectors.select_scope_state(&key, policy, 3).unwrap(); let (subject, subject_filtered) = selectors .select_subject_state(fixtures.hub_entity.id, policy, 3, false)