The three memory ports, persisted through the shipped GraphBackend.
Persists MemoryNode rows through GraphBackend using compatible formats.
forget erases a scope subtree, retracted rows stay hidden until pruning,
and pending rows remain invisible until flush. Those rules are enforced
at the shared read/lifecycle choke points below rather than by callers.
Each node is updated and deleted at its own scope, reconstructed from the
stored scope_path. The backends gate mutation by ancestor visibility, so a
descendant's row cannot be written through its parent's scope -- which is the
correct rule for a read and would be a silent no-op here.
GraphBackedHms
GraphBackedHms(graph: Any, *, vectors: Any = None, embedder: Any = None)
All three memory ports, over a graph backend and optionally a vector one.
Graph and vector candidates merge by record ID before final ranking.
Source code in src/symfonic/capabilities/memory/graph_store.py
| def __init__(
self, graph: Any, *, vectors: Any = None, embedder: Any = None
) -> None:
if (vectors is None) != (embedder is None):
raise ConfigurationError(
"semantic recall needs a vector backend and an embedder "
f"together; got vectors={type(vectors).__name__} and "
f"embedder={type(embedder).__name__}. One without the other "
"recalls nothing and looks composed."
)
self._graph = graph
self._vectors = vectors
self._direct_claims: dict[tuple[str, str], str] = {}
self._transaction_participants = (graph,) + ((vectors,) if vectors is not None else ())
self._recall = (
VectorRecall(vectors, embedder) if vectors is not None else None
)
from symfonic.capabilities.memory.mutations import mutation_fence_for
self._mutations = mutation_fence_for(graph, vectors)
|
graph
property
The graph store used by consolidation's domain check.
transaction_participants
property
transaction_participants: tuple[Any, ...]
Backends participating in mutation and vector publication.
discard
async
discard(scope: MemoryScope, *, operation: Any = None) -> LifecycleReceipt
Delete the pending rows under scope; committed history survives.
Row-by-row rather than through delete_subtree, and deliberately:
the subtree sweep is the privacy verb and takes everything. A turn
taking back its own writes may not take the previous turns' with them,
so the pending predicate has to be part of the selection.
Source code in src/symfonic/capabilities/memory/graph_store.py
| async def discard(self, scope: MemoryScope, *, operation: Any = None) -> LifecycleReceipt:
"""Delete the pending rows under ``scope``; committed history survives.
Row-by-row rather than through ``delete_subtree``, and deliberately:
the subtree sweep is the *privacy* verb and takes everything. A turn
taking back its own writes may not take the previous turns' with them,
so the pending predicate has to be part of the selection.
"""
from symfonic.capabilities.memory.graph_claimed_lifecycle import discard_pending
from symfonic.capabilities.memory.graph_direct_claims import external_transaction, retire
from symfonic.capabilities.memory.mutations import lifecycle_fence
graph_receipt.validate(
self, operation, graph_receipt.MemoryOperationVerb.DISCARD, scope)
self._check_mutation(scope, graph_receipt.MemoryOperationVerb.DISCARD)
outer = external_transaction(self, scope)
async with lifecycle_fence(
self._graph, self._vectors, scope, self._mutations,
operation=operation, verb=graph_receipt.MemoryOperationVerb.DISCARD,
):
claims = self._direct_claims.copy()
# Sealed terminal procedures survive; the rest go in one transaction.
discarded = await discard_pending(self, await self._pending_under(scope))
retire(self, scope, discarded, outer, claims)
return LifecycleReceipt(scope_path=scope.path, discarded=tuple(sorted(discarded)))
|
forget
async
forget(scope: MemoryScope, *, operation: Any = None) -> LifecycleReceipt | ForgetInProgress
Erase the subtree; native PG reports deleted IDs atomically.
Other backends still inventory before deletion and can omit late rows.
A composed CORE-199 eraser answers ForgetInProgress until its job ends.
Source code in src/symfonic/capabilities/memory/graph_store.py
| async def forget(
self, scope: MemoryScope, *, operation: Any = None,
) -> LifecycleReceipt | ForgetInProgress:
"""Erase the subtree; native PG reports deleted IDs atomically.
Other backends still inventory before deletion and can omit late rows.
A composed CORE-199 eraser answers ForgetInProgress until its job ends.
"""
from symfonic.capabilities.memory.graph_forget import forget_subtree
graph_receipt.validate(
self, operation, graph_receipt.MemoryOperationVerb.FORGET, scope)
self._check_mutation(scope, graph_receipt.MemoryOperationVerb.FORGET)
return await forget_subtree(self, scope, operation=operation)
|
retrieve
async
retrieve(query: MemoryQuery) -> RetrievalResult
Both routes, merged by record_id and ranked once.
Source code in src/symfonic/capabilities/memory/graph_store.py
| async def retrieve(self, query: MemoryQuery) -> RetrievalResult:
"""Both routes, merged by ``record_id`` and ranked once."""
candidates, sources, unavailable = await graph_receipt.gather(self, query)
return select(candidates, query, sources=sources, unavailable=unavailable)
|
scan_candidates
async
scan_candidates(query: MemoryQuery) -> RetrievalResult
Every visible candidate, uncapped by query.limit (CandidateScan).
Retrieval's select would apply limits this scan leaves to callers.
Vector candidates retain their own similarity-search top-k budget.
Source code in src/symfonic/capabilities/memory/graph_store.py
| async def scan_candidates(self, query: MemoryQuery) -> RetrievalResult:
"""Every visible candidate, uncapped by ``query.limit`` (CandidateScan).
Retrieval's ``select`` would apply limits this scan leaves to callers.
Vector candidates retain their own similarity-search top-k budget.
"""
candidates, sources, unavailable = await graph_receipt.gather(self, query)
return RetrievalResult(
memories=rank(candidates), sources=sources, unavailable=unavailable
)
|
node_payload
node_payload(node: MemoryNode) -> dict[str, Any]
The mapping compat.record_from_legacy_node reads, off a real node.
Source code in src/symfonic/capabilities/memory/graph_rows.py
| def node_payload(node: MemoryNode) -> dict[str, Any]:
"""The mapping ``compat.record_from_legacy_node`` reads, off a real node."""
return {
"id": str(node.id),
"layer": node.layer.value,
"tenant_id": node.tenant_id,
"label": node.label,
"properties": dict(node.properties),
"importance": node.importance,
"updated_at": node.updated_at,
}
|
tenant_scope
tenant_scope(scope: MemoryScope) -> StorageScope
GraphBackedHms-only conversion, after its graph/vector admission check.
Source code in src/symfonic/capabilities/memory/graph_rows.py
| def _adapter_scope(scope: MemoryScope) -> StorageScope:
"""GraphBackedHms-only conversion, after its graph/vector admission check."""
if scope.identity_version:
return NativeStorageScope.from_scope(scope)
return tenant_scope(scope)
|