Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .github/workflows/pr_validation.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -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
4 changes: 3 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,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.
Expand Down Expand Up @@ -214,7 +216,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.

Expand Down
8 changes: 7 additions & 1 deletion docs/design/database/graph_schema_design.md
Original file line number Diff line number Diff line change
Expand Up @@ -120,12 +120,18 @@ 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. 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 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.

## 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.

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.
39 changes: 35 additions & 4 deletions src/adapters/oxigraph/embedded.rs
Original file line number Diff line number Diff line change
Expand Up @@ -245,7 +245,7 @@ impl GraphAuthorityStore for OxigraphGraphAuthorityStore {
.map(|id| MemoryObjectRef::new(ObjectType::DerivedMemory, *id))
.collect::<Vec<_>>();
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(),
})?
Expand Down Expand Up @@ -287,7 +287,13 @@ impl GraphAuthorityStore for OxigraphGraphAuthorityStore {
.map(|id| MemoryObjectRef::from_id_type(id, ObjectType::DerivedMemory))
.collect::<Vec<_>>(),
)?;
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::<Vec<_>>(),
)?;
let links = hydrate_links_by_ids_from_store(&self.store, &link_ids)?;
Ok(derived_memories_by_provenance(
query,
objects.into_iter().filter(
Expand All @@ -301,6 +307,24 @@ impl GraphAuthorityStore for OxigraphGraphAuthorityStore {
&self,
query: &GraphDerivedMemoryThreadQuery,
) -> Result<(Vec<DerivedMemory>, Vec<GraphExpansionFilteredNode>), 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::<Vec<_>>(),
)?;
let memories = objects
.into_iter()
.filter_map(|object| match object {
MemoryObject::DerivedMemory(memory) => Some(memory),
_ => None,
})
.collect::<Vec<_>>();
return Ok((memories, filtered));
}
let selected_ids = SparqlGraphSelectors::new(&self.store)
.select_derived_memories_by_thread(query)?
.into_iter()
Expand All @@ -314,7 +338,13 @@ impl GraphAuthorityStore for OxigraphGraphAuthorityStore {
.map(|id| MemoryObjectRef::from_id_type(id, ObjectType::DerivedMemory))
.collect::<Vec<_>>(),
)?;
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::<Vec<_>>(),
)?;
let links = hydrate_links_by_ids_from_store(&self.store, &link_ids)?;
Ok(derived_memories_by_thread(
query,
objects.into_iter().filter(
Expand All @@ -328,8 +358,9 @@ impl GraphAuthorityStore for OxigraphGraphAuthorityStore {
&self,
key: &ScopeKey,
policy: GraphExpansionLifecyclePolicy,
limit: usize,
) -> Result<(Vec<MemoryId>, Vec<GraphExpansionFilteredNode>), CustomError> {
SparqlGraphSelectors::new(&self.store).select_scope_state(key, policy)
SparqlGraphSelectors::new(&self.store).select_scope_state(key, policy, limit)
}

async fn expand_bounded(
Expand Down
83 changes: 62 additions & 21 deletions src/adapters/oxigraph/shared.rs
Original file line number Diff line number Diff line change
Expand Up @@ -142,7 +142,20 @@ pub(super) fn hydrate_objects_by_refs_from_store(
store: &Store,
refs: &[MemoryObjectRef],
) -> Result<Vec<MemoryObject>, 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::<Result<Vec<_>, _>>()?;
// 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 {
Expand Down Expand Up @@ -220,9 +233,17 @@ pub(super) fn hydrate_links_by_id_sets_from_store(

pub(super) fn rdf_subject_values(
store: &Store,
) -> Result<HashMap<String, RdfSubjectValues>, CustomError> {
rdf_subject_values_from_quads(store.iter())
}

fn rdf_subject_values_from_quads(
quads: impl IntoIterator<Item = Result<Quad, oxigraph::store::StorageError>>,
) -> Result<HashMap<String, RdfSubjectValues>, CustomError> {
let mut subjects = HashMap::<String, RdfSubjectValues>::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;
Expand All @@ -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<usize> = const { std::cell::Cell::new(0) };
}

fn rdf_subject_values_for_named_graph(
store: &Store,
graph_uri: &str,
Expand All @@ -258,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;
Expand Down Expand Up @@ -601,7 +629,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())
};
Expand All @@ -617,23 +654,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::<HashSet<_>>()
.into_iter()
.collect::<Vec<_>>();
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();
Expand Down Expand Up @@ -674,18 +727,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
};
Expand Down Expand Up @@ -723,7 +764,7 @@ pub(super) fn bounded_graph_visible_refs(

let candidate_refs = graph_refs.iter().copied().collect::<Vec<_>>();
let lifecycle_link_ids = selectors
.select_links_touching(&candidate_refs)?
.select_links_touching(&candidate_refs, false)?
.into_iter()
.filter(|link_ref| {
matches!(
Expand Down
Loading
Loading