symfonic.memory.graph.store

store

GraphMemoryStore -- node and edge persistence with tenant isolation.

Delegates all storage operations to a GraphBackend protocol implementation. Provides query convenience methods and access-count tracking for scoring.

GraphMemoryStore

GraphMemoryStore(backend: GraphBackend, embedding_provider: EmbeddingProvider | None = None, embedding_cache: EmbeddingCache | None = None)

Bases: GraphQueryMixin

Graph-aware memory store wrapping a GraphBackend.

All operations enforce tenant isolation via TenantScope. Access counts are automatically incremented on reads to support frequency-based retrieval scoring.

v7.3 Item 13.1: optional auto-embed at the store level.

The graph-backed memory LAYERS (semantic / procedural / prospective) wrap their own writes through :func:~symfonic.memory.embeddings.auto_embed.maybe_embed so layer-API callers benefit even when constructing layers without the orchestrator. Consolidation phases that bypass the layers (Item 12 EntityLinker calls GraphMemoryStore.add_node directly) need the same auto-embed pathway -- that's what the constructor kwargs here are for.

The double wrap (layer -> store) is idempotent: maybe_embed short-circuits when node.embedding is already populated, so a node that went through SemanticLayer.store_fact only triggers one provider call regardless of how many layers of wrapping it traverses.

Both kwargs default to None so legacy callers that build GraphMemoryStore directly are byte-identical to v7.2.

Source code in src/symfonic/memory/graph/store.py

def __init__(
    self,
    backend: GraphBackend,
    embedding_provider: EmbeddingProvider | None = None,
    embedding_cache: EmbeddingCache | None = None,
) -> None:
    """v7.3 Item 13.1: optional auto-embed at the store level.

    The graph-backed memory LAYERS (semantic / procedural /
    prospective) wrap their own writes through
    :func:`~symfonic.memory.embeddings.auto_embed.maybe_embed` so
    layer-API callers benefit even when constructing layers without
    the orchestrator. Consolidation phases that bypass the layers
    (Item 12 EntityLinker calls ``GraphMemoryStore.add_node``
    directly) need the same auto-embed pathway -- that's what the
    constructor kwargs here are for.

    The double wrap (layer -> store) is idempotent: ``maybe_embed``
    short-circuits when ``node.embedding`` is already populated, so
    a node that went through ``SemanticLayer.store_fact`` only
    triggers one provider call regardless of how many layers of
    wrapping it traverses.

    Both kwargs default to ``None`` so legacy callers that build
    ``GraphMemoryStore`` directly are byte-identical to v7.2.
    """
    self._backend = backend
    self._embedding_provider = embedding_provider
    self._embedding_cache = embedding_cache

backend property

backend: GraphBackend

What this store persists through. Named, so callers stop reaching.

add_edge async

add_edge(scope: TenantScope, edge: MemoryEdge) -> MemoryEdge

Add an edge to the graph. Returns the persisted edge.

Source code in src/symfonic/memory/graph/store.py

async def add_edge(self, scope: TenantScope, edge: MemoryEdge) -> MemoryEdge:
    """Add an edge to the graph. Returns the persisted edge."""
    edge.tenant_id = scope.tenant_id
    return await self._backend.add_edge(scope, edge)

add_node async

add_node(scope: TenantScope, node: MemoryNode) -> MemoryNode

Add a node to the graph. Returns the persisted node.

v7.3 Item 13.1: when an embedding_provider is wired into this store, nodes the caller left with embedding=None auto-embed before the backend write. Caller-provided embeddings always win (see :func:maybe_embed).

Source code in src/symfonic/memory/graph/store.py

async def add_node(self, scope: TenantScope, node: MemoryNode) -> MemoryNode:
    """Add a node to the graph. Returns the persisted node.

    v7.3 Item 13.1: when an ``embedding_provider`` is wired into
    this store, nodes the caller left with ``embedding=None``
    auto-embed before the backend write. Caller-provided embeddings
    always win (see :func:`maybe_embed`).
    """
    node.tenant_id = scope.tenant_id
    if self._embedding_provider is not None:
        # Late import keeps the symfonic.memory.embeddings package an
        # optional surface for callers that never construct one.
        from symfonic.memory.embeddings.auto_embed import maybe_embed

        node = await maybe_embed(
            node, self._embedding_provider, self._embedding_cache,
        )
    return await self._backend.add_node(scope, node)

bump_spreading async

bump_spreading(scope: TenantScope, node_id: NodeId, *, include_retracted: bool = False) -> MemoryNode | None

Fetch a node and increment its spreading_access_count.

v6.2 T02: BFS-induced reads funnel through this helper instead of :meth:get_node, so the Phase 1 recurrence signal can distinguish direct fetches from one-shot spreading-activation visits. Both counters carry independent monotonic semantics: - access_count grows on get_node - spreading_access_count grows on bump_spreading

The method is intentionally named bump_spreading (rather than get_node_via_spreading or a private _bump_spreading) so alternative traversal strategies in future sprints can call it directly without reaching into private API.

Returns None if the node does not exist or belongs to another tenant; otherwise returns the updated node.

Soft-retracted nodes return None by default (and are not bumped) so a retracted false memory cannot re-enter context via spreading activation -- notably as a live node's neighbor. Pass include_retracted=True only for audit/maintenance reads.

Source code in src/symfonic/memory/graph/store.py

async def bump_spreading(
    self, scope: TenantScope, node_id: NodeId,
    *, include_retracted: bool = False,
) -> MemoryNode | None:
    """Fetch a node and increment its spreading_access_count.

    v6.2 T02: BFS-induced reads funnel through this helper instead
    of :meth:`get_node`, so the Phase 1 recurrence signal can
    distinguish direct fetches from one-shot spreading-activation
    visits. Both counters carry independent monotonic semantics:
    - access_count grows on ``get_node``
    - spreading_access_count grows on ``bump_spreading``

    The method is intentionally named ``bump_spreading`` (rather than
    ``get_node_via_spreading`` or a private ``_bump_spreading``) so
    alternative traversal strategies in future sprints can call it
    directly without reaching into private API.

    Returns None if the node does not exist or belongs to another
    tenant; otherwise returns the updated node.

    Soft-retracted nodes return ``None`` by default (and are not bumped) so
    a retracted false memory cannot re-enter context via spreading
    activation -- notably as a live node's neighbor. Pass
    ``include_retracted=True`` only for audit/maintenance reads.
    """
    node = await self._backend.get_node(scope, node_id)
    if node is None:
        return None
    if not include_retracted and is_retracted(node.properties):
        return None
    updated = await self._backend.update_node(
        scope,
        node_id,
        {
            "spreading_access_count": node.spreading_access_count + 1,
            "updated_at": datetime.now(UTC).isoformat(),
        },
    )
    return updated

delete_edge async

delete_edge(scope: TenantScope, edge_id: EdgeId) -> None

Delete a single edge by its ID.

Source code in src/symfonic/memory/graph/store.py

async def delete_edge(self, scope: TenantScope, edge_id: EdgeId) -> None:
    """Delete a single edge by its ID."""
    await self._backend.delete_edge(scope, edge_id)

delete_node async

delete_node(scope: TenantScope, node_id: NodeId) -> None

Delete a node and cascade-delete all connected edges.

Source code in src/symfonic/memory/graph/store.py

async def delete_node(self, scope: TenantScope, node_id: NodeId) -> None:
    """Delete a node and cascade-delete all connected edges."""
    await self._backend.delete_node(scope, node_id, cascade=True)

get_neighbors async

get_neighbors(scope: TenantScope, node_id: NodeId, relationship: str | None = None, *, include_retracted: bool = False) -> list[MemoryNode]

Get neighboring nodes, optionally filtered by relationship type.

Soft-retracted neighbours are excluded by default so a retracted node cannot re-enter context as a graph neighbour of a live node. Audit / maintenance callers pass include_retracted=True.

Source code in src/symfonic/memory/graph/store.py

async def get_neighbors(
    self,
    scope: TenantScope,
    node_id: NodeId,
    relationship: str | None = None,
    *,
    include_retracted: bool = False,
) -> list[MemoryNode]:
    """Get neighboring nodes, optionally filtered by relationship type.

    Soft-retracted neighbours are excluded by default so a retracted node
    cannot re-enter context as a graph neighbour of a live node. Audit /
    maintenance callers pass ``include_retracted=True``.
    """
    neighbors = await self._backend.get_neighbors(
        scope, node_id, relationship,
    )
    if include_retracted:
        return neighbors
    return [n for n in neighbors if not is_retracted(n.properties)]

get_node async

get_node(scope: TenantScope, node_id: NodeId, *, include_retracted: bool = False) -> MemoryNode | None

Get a node by ID, incrementing its access count.

Returns None if the node does not exist or belongs to another tenant.

Soft-retracted nodes (see symfonic.memory.retraction for the namespaced marker contract) return None by default so a corrected/false-positive memory never re-enters context through a direct fetch; the access count is NOT bumped for them. Audit / erasure / prune callers pass include_retracted=True to reach them.

Source code in src/symfonic/memory/graph/store.py

async def get_node(
    self, scope: TenantScope, node_id: NodeId,
    *, include_retracted: bool = False,
) -> MemoryNode | None:
    """Get a node by ID, incrementing its access count.

    Returns None if the node does not exist or belongs to another tenant.

    Soft-retracted nodes (see ``symfonic.memory.retraction`` for the
    namespaced marker contract) return ``None`` by default so a
    corrected/false-positive memory never re-enters context through a
    direct fetch; the access count is NOT bumped for them. Audit /
    erasure / prune callers pass ``include_retracted=True`` to reach them.
    """
    node = await self._backend.get_node(scope, node_id)
    if node is None:
        return None
    if not include_retracted and is_retracted(node.properties):
        return None
    # Increment access_count for frequency scoring
    updated = await self._backend.update_node(
        scope,
        node_id,
        {
            "access_count": node.access_count + 1,
            "updated_at": datetime.now(UTC).isoformat(),
        },
    )
    return updated

list_edges async

list_edges(scope: TenantScope, *, limit: int = 50, offset: int = 0, relationship: str | None = None) -> list[MemoryEdge]

List edges for the tenant with optional filtering and pagination.

Delegates directly to GraphBackend.query_edges — no node iteration.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant isolation scope.	required
limit	int	Maximum number of edges to return (default 50).	50
offset	int	Number of edges to skip for pagination (default 0).	0
relationship	str | None	If provided, only edges of this type are returned.	None

Source code in src/symfonic/memory/graph/store.py

async def list_edges(
    self,
    scope: TenantScope,
    *,
    limit: int = 50,
    offset: int = 0,
    relationship: str | None = None,
) -> list[MemoryEdge]:
    """List edges for the tenant with optional filtering and pagination.

    Delegates directly to ``GraphBackend.query_edges`` — no node iteration.

    Args:
        scope: Tenant isolation scope.
        limit: Maximum number of edges to return (default 50).
        offset: Number of edges to skip for pagination (default 0).
        relationship: If provided, only edges of this type are returned.
    """
    filters: dict[str, Any] | None = None
    if relationship is not None:
        filters = {"relationship": relationship}
    return await self._backend.query_edges(scope, filters, limit, offset)

neighbor_probe async

neighbor_probe(scope: TenantScope, node_id: NodeId, minimum: int = 1, relationship: str | None = None, target: str | None = None) -> bool

Answer a threshold/existence question without loading a neighborhood.

Source code in src/symfonic/memory/graph/store.py

async def neighbor_probe(self, scope: TenantScope, node_id: NodeId, minimum: int = 1,
                         relationship: str | None = None, target: str | None = None) -> bool:
    """Answer a threshold/existence question without loading a neighborhood."""
    probe = getattr(self._backend, "neighbor_probe", None)
    if callable(probe):
        return bool(await probe(scope, str(node_id), minimum, relationship, target))
    neighbors = await self.get_neighbors(scope, node_id, relationship)
    return sum(target is None or str(node.id) == target for node in neighbors) >= minimum

related_candidates async

related_candidates(scope: TenantScope, cue: str) -> list[MemoryNode]

Bounded exact-owner semantic candidates for offline comparison.

Source code in src/symfonic/memory/graph/store.py

async def related_candidates(self, scope: TenantScope, cue: str) -> list[MemoryNode]:
    """Bounded exact-owner semantic candidates for offline comparison."""
    select = getattr(self._backend, "select_candidates", None)
    if not callable(select):
        return []
    from symfonic.memory.candidates import CandidateRequest
    page = await select(scope, CandidateRequest(
        cue=cue, layers=("semantic",), limit=64, profile_slots=0, exact_scope=True,
    ))
    return list(page.nodes)

update_node async

update_node(scope: TenantScope, node_id: NodeId, properties: dict[str, object]) -> MemoryNode

Partial update of node properties.

Source code in src/symfonic/memory/graph/store.py

async def update_node(
    self, scope: TenantScope, node_id: NodeId, properties: dict[str, object]
) -> MemoryNode:
    """Partial update of node properties."""
    return await self._backend.update_node(scope, node_id, properties)

upsert_edge async

upsert_edge(scope: TenantScope, edge: MemoryEdge) -> MemoryEdge

Insert edge or increment weight if it already exists.

Uniqueness key: (tenant_id, source, target, relationship). Each repeated call on the same pair increments weight by 1.

Source code in src/symfonic/memory/graph/store.py

async def upsert_edge(self, scope: TenantScope, edge: MemoryEdge) -> MemoryEdge:
    """Insert edge or increment weight if it already exists.

    Uniqueness key: (tenant_id, source, target, relationship).
    Each repeated call on the same pair increments weight by 1.
    """
    edge.tenant_id = scope.tenant_id
    return await self._backend.upsert_edge(scope, edge)

upsert_edge_props async

upsert_edge_props(scope: TenantScope, edge_id: EdgeId, props: dict[str, Any]) -> MemoryEdge | None

Update edge properties in-place.

Attempts backend.update_edge when available; otherwise falls back to delete+re-add. Returns the updated edge, or None on failure.

Source code in src/symfonic/memory/graph/store.py

async def upsert_edge_props(
    self,
    scope: TenantScope,
    edge_id: EdgeId,
    props: dict[str, Any],
) -> MemoryEdge | None:
    """Update edge properties in-place.

    Attempts backend.update_edge when available; otherwise falls back
    to delete+re-add. Returns the updated edge, or None on failure.
    """
    if hasattr(self._backend, "update_edge"):
        return await self._backend.update_edge(scope, edge_id, props)
    # Fallback: find edge, apply props, delete, re-add
    edges = await self._backend.query_edges(scope, None, 10_000, 0)
    target_edge = None
    for e in edges:
        if str(e.id) == str(edge_id):
            target_edge = e
            break
    if target_edge is None:
        return None
    valid_props = {k: v for k, v in props.items() if hasattr(target_edge, k)}
    updated_edge = target_edge.model_copy(update=valid_props)
    # Delete-then-add (v8.7.2): the previous add-then-delete order re-added
    # the edge under the SAME id and then deleted that id, permanently
    # destroying the edge — on InMemory the delete popped the just-written
    # key, and on Postgres ``add_edge`` is ``ON CONFLICT DO NOTHING`` so the
    # write was a no-op before the delete removed the original.  Deleting
    # first, then re-adding the mutated copy, is correct on every backend.
    await self._backend.delete_edge(scope, edge_id)
    await self._backend.add_edge(scope, updated_edge)
    return updated_edge