`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,
)

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

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)

distinct_labels `async` ¶

distinct_labels(
    scope: TenantScope, *, include_retracted: bool = False
) -> list[dict[str, Any]]

Get distinct labels with counts and last_updated.

Aggregates all nodes in the tenant scope by their label field.

Soft-retracted nodes are excluded by default so a label whose only node has been retracted is NOT reported as present -- otherwise a required-memory discovery check (retrieval.discovery) would mark a block satisfied when no live node backs it. Pass include_retracted=True for audit aggregation.

Returns:

Type	Description
`list[dict[str, Any]]`	List of dicts with keys `label`, `count`, `last_updated`.

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

async def distinct_labels(
    self, scope: TenantScope, *, include_retracted: bool = False,
) -> list[dict[str, Any]]:
    """Get distinct labels with counts and last_updated.

    Aggregates all nodes in the tenant scope by their ``label`` field.

    Soft-retracted nodes are excluded by default so a label whose only
    node has been retracted is NOT reported as present -- otherwise a
    required-memory discovery check (``retrieval.discovery``) would mark a
    block satisfied when no live node backs it. Pass
    ``include_retracted=True`` for audit aggregation.

    Returns:
        List of dicts with keys ``label``, ``count``, ``last_updated``.
    """
    # Full aggregation, no cap: route through the store's own
    # ``query_nodes`` (the single retraction choke point). A flat
    # backend cap was unsafe here -- with retractions excluded AFTER a
    # truncated fetch, N>=cap retracted rows ahead of a live node made
    # its label vanish (retrieval.discovery would then mark a satisfied
    # block UNsatisfied). ``limit=None`` scans every node and refills
    # past retracted rows via the same doubling over-fetch as siblings.
    nodes = await self.query_nodes(
        scope, limit=None, include_retracted=include_retracted,
    )
    groups: dict[str, dict[str, Any]] = defaultdict(
        lambda: {"count": 0, "last_updated": None},
    )
    for node in nodes:
        g = groups[node.label]
        g["count"] += 1
        if node.updated_at and (
            g["last_updated"] is None or node.updated_at > g["last_updated"]
        ):
            g["last_updated"] = node.updated_at
    return [
        {"label": label, "count": g["count"], "last_updated": g["last_updated"]}
        for label, g in groups.items()
    ]

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)

query_nodes `async` ¶

query_nodes(
    scope: TenantScope,
    layer: MemoryLayer | None = None,
    label: str | None = None,
    label_prefix: str | None = None,
    filters: dict[str, object] | None = None,
    limit: int | None = None,
    include_retracted: bool = False,
) -> list[MemoryNode]

Query nodes by layer, label, label_prefix, and/or arbitrary filters.

Parameters:

Name	Type	Description	Default
`scope`	`TenantScope`	Tenant isolation scope.	required
`layer`	`MemoryLayer \| None`	Exact match on the memory layer.	`None`
`label`	`str \| None`	Exact match on the node label.	`None`
`label_prefix`	`str \| None`	Prefix match on the node label (e.g. `"SOUL"` matches `"SOUL: Joy"` and `"SOUL: name: Amiel"`). `label` and `label_prefix` can coexist (AND logic).	`None`
`filters`	`dict[str, object] \| None`	Arbitrary key/value filters passed through to the backend.	`None`
`limit`	`int \| None`	Maximum rows to return. `None` (default) returns all matching nodes. Prior to v8.7.2 this wrapper silently dropped the caller's request and the backend truncated every read to its ~50-row default — starving retrieval scans, dedupe, and (worst) inverting `working_graph_retention >= 50` pruning.	`None`
`include_retracted`	`bool`	When `False` (default) nodes soft-retracted via a `retract_node` extraction op (see `symfonic.memory.retraction` for the namespaced marker) are filtered out. This is the single read-side choke point that keeps a corrected/false-positive memory out of hydration, routing, and dedup. Consolidation's prune phase passes `True` so it can find and delete them. See `docs/guides/09-consolidation-and-deep-sleep.md`.	`False`

A positive limit combined with include_retracted=False no longer under-returns when a retracted row happens to fall inside the backend's page: the wrapper over-fetches (doubling the requested window) and refills with live nodes until limit is satisfied or the backend is exhausted, then trims to exactly limit. This keeps limit semantics honest for every backend (InMemory/Postgres/ Mongo) without any backend-side change.

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

async def query_nodes(
    self,
    scope: TenantScope,
    layer: MemoryLayer | None = None,
    label: str | None = None,
    label_prefix: str | None = None,
    filters: dict[str, object] | None = None,
    limit: int | None = None,
    include_retracted: bool = False,
) -> list[MemoryNode]:
    """Query nodes by layer, label, label_prefix, and/or arbitrary filters.

    Args:
        scope: Tenant isolation scope.
        layer: Exact match on the memory layer.
        label: Exact match on the node label.
        label_prefix: Prefix match on the node label (e.g. ``"SOUL"``
            matches ``"SOUL: Joy"`` and ``"SOUL: name: Amiel"``).
            ``label`` and ``label_prefix`` can coexist (AND logic).
        filters: Arbitrary key/value filters passed through to the backend.
        limit: Maximum rows to return.  ``None`` (default) returns *all*
            matching nodes.  Prior to v8.7.2 this wrapper silently dropped
            the caller's request and the backend truncated every read to
            its ~50-row default — starving retrieval scans, dedupe, and
            (worst) inverting ``working_graph_retention >= 50`` pruning.
        include_retracted: When ``False`` (default) nodes soft-retracted
            via a ``retract_node`` extraction op (see
            ``symfonic.memory.retraction`` for the namespaced marker) are
            filtered out. This is the single read-side choke point that
            keeps a corrected/false-positive memory out of hydration,
            routing, and dedup. Consolidation's prune phase passes
            ``True`` so it can find and delete them. See
            ``docs/guides/09-consolidation-and-deep-sleep.md``.

    A positive ``limit`` combined with ``include_retracted=False`` no
    longer under-returns when a retracted row happens to fall inside the
    backend's page: the wrapper over-fetches (doubling the requested
    window) and refills with live nodes until ``limit`` is satisfied or
    the backend is exhausted, then trims to exactly ``limit``. This keeps
    ``limit`` semantics honest for every backend (InMemory/Postgres/
    Mongo) without any backend-side change.
    """
    filter_dict: dict[str, object] = {}
    if layer is not None:
        filter_dict["layer"] = layer.value
    if label is not None:
        filter_dict["label"] = label
    if label_prefix is not None:
        filter_dict["label_prefix"] = label_prefix
    if filters:
        filter_dict.update(filters)
    # v8.7.1: unify limit==0 semantics across backends. ``None`` = all,
    # a positive int caps the result, 0 means "no rows".  Mongo's native
    # ``.limit(0)`` otherwise means *unlimited* (the opposite) and InMemory
    # returned 1 row — so short-circuit here before dispatch.
    if limit == 0:
        return []
    if include_retracted:
        return await self._backend.query_nodes(scope, filter_dict, limit=limit)
    if limit is None:
        nodes = await self._backend.query_nodes(scope, filter_dict, limit=None)
        return [n for n in nodes if not is_retracted(n.properties)]
    # Retraction exclusion: a soft-retracted node stays in the store (for
    # audit + reversibility) but must never surface to any reader. Applied
    # here so every consumer of query_nodes -- semantic/procedural
    # retrieve, dedup, spreading activation -- inherits the exclusion for
    # free. Over-fetch with a doubling window so a retracted row inside
    # the backend's page doesn't silently starve the caller's ``limit``.
    fetch = limit
    while True:
        batch = await self._backend.query_nodes(scope, filter_dict, limit=fetch)
        live = [n for n in batch if not is_retracted(n.properties)]
        if len(live) >= limit or len(batch) < fetch:
            break
        fetch *= 2
    return live[:limit]

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

symfonic.memory.graph.store¶

store ¶

GraphMemoryStore ¶

add_edge async ¶

add_node async ¶

bump_spreading async ¶

delete_edge async ¶

delete_node async ¶

distinct_labels async ¶

get_neighbors async ¶

get_node async ¶

list_edges async ¶

query_nodes async ¶

update_node async ¶

upsert_edge async ¶

upsert_edge_props async ¶

`symfonic.memory.graph.store`¶

add_edge `async` ¶

add_node `async` ¶

bump_spreading `async` ¶

delete_edge `async` ¶

delete_node `async` ¶

distinct_labels `async` ¶

get_neighbors `async` ¶

get_node `async` ¶

list_edges `async` ¶

query_nodes `async` ¶

update_node `async` ¶

upsert_edge `async` ¶

upsert_edge_props `async` ¶