symfonic.capabilities.memory.graph_store

graph_store

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.

Composed with vectors and embedder, recall runs both routes and merges them: the graph answers what matches the cue's words, the vector index what matches its meaning, and a query sharing no vocabulary with the memory it needs is answered by the second. Composed without them, nothing changes -- which is why they are one optional pair rather than a second store class.

A pair because neither is useful alone: an embedder with nowhere to put a vector writes nothing, and a vector backend with no embedder cannot be queried. One without the other is refused at construction rather than discovered later as recall that was silently lexical.

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

graph: Any

The store or backend this HMS writes through.

Published for the consolidation commit, which must establish that the records it publishes land in the same transaction domain as the graph mutations it applies beside them.

transaction_participants property

transaction_participants: tuple[Any, ...]

Every backend a write/flush can mutate, including vector publication.

discard async

discard(scope: MemoryScope) -> 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) -> 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.
    """
    discarded: list[str] = []
    async with self._mutations.hold(tenant_scope(scope)):
        for node in await self._pending_under(scope):
            if self._recall is not None:
                await self._recall.forget(
                    tenant_scope(node_scope(node)), (str(node.id),)
                )
            await self._graph.delete_node(
                tenant_scope(node_scope(node)), NodeId(str(node.id)), cascade=True
            )
            discarded.append(str(node.id))
    return LifecycleReceipt(scope_path=scope.path, discarded=tuple(sorted(discarded)))

forget async

forget(scope: MemoryScope) -> LifecycleReceipt

Erase scope and everything below it, pending or not.

The ids are read before the sweep because the receipt names them, not because the sweep needs them: delete_subtree is one backend-native statement over the whole subtree, so a row written into a descendant scope between the read and the delete is still erased.

Source code in src/symfonic/capabilities/memory/graph_store.py

async def forget(self, scope: MemoryScope) -> LifecycleReceipt:
    """Erase ``scope`` and everything below it, pending or not.

    The ids are read before the sweep because the receipt names them, not
    because the sweep needs them: ``delete_subtree`` is one backend-native
    statement over the whole subtree, so a row written into a descendant
    scope between the read and the delete is still erased.
    """
    legacy = tenant_scope(scope)
    async with self._mutations.hold(legacy):
        doomed_nodes = await self._graph.query_subtree(legacy, {}, limit=None)
        doomed = sorted(str(node.id) for node in doomed_nodes)
        await self._graph.delete_subtree(legacy)
        if self._recall is not None:
            await self._recall.forget_subtree(legacy)
    return LifecycleReceipt(scope_path=scope.path, discarded=tuple(doomed))

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 self._gather(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).

Not a call to :meth:retrieve: that one ends in select, which applies the limit and the character ceilings, and those are exactly the decisions a scan must leave to its caller.

Both routes, because this is the method a turn's hydration actually calls -- a scan that asked only the lexical route would leave the vector index composed, written to, and never consulted, which is the shape this pair exists to remove.

The vector half is bounded by its own top-k, which is the "store's own budget" this contract allows: a similarity search has no unbounded form, and asking for every vector in the scope would be a scan of the index rather than a search of it.

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

    Not a call to :meth:`retrieve`: that one ends in ``select``, which
    applies the limit and the character ceilings, and those are exactly
    the decisions a scan must leave to its caller.

    Both routes, because this is the method a turn's hydration actually
    calls -- a scan that asked only the lexical route would leave the
    vector index composed, written to, and never consulted, which is the
    shape this pair exists to remove.

    The vector half is bounded by its own top-k, which is the "store's own
    budget" this contract allows: a similarity search has no unbounded
    form, and asking for every vector in the scope would be a scan of the
    index rather than a search of it.
    """
    candidates, sources, unavailable = await self._gather(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,
    }

tenant_scope

tenant_scope(scope: MemoryScope) -> TenantScope

The legacy scope value for a capability scope. Kinds from compat.

Source code in src/symfonic/capabilities/memory/graph_rows.py

def tenant_scope(scope: MemoryScope) -> TenantScope:
    """The legacy scope value for a capability scope. Kinds from ``compat``."""
    levels = [
        ScopeLevel(kind=kind, id=segment)
        for kind, segment in zip(LEGACY_KINDS, scope.segments, strict=False)
    ]
    return TenantScope.from_path(levels)