symfonic.memory.graph.store_queries

store_queries

The graph store's read surface: filtered node queries and label aggregation.

Split out of :mod:symfonic.memory.graph.store (410 lines against the 300-line budget). What stayed behind is the write surface and the single-node reads that also bump access counts. What moved here is the querying half -- query_nodes with its filter matrix, the subtree read and erase, and the label aggregation the admin surfaces use -- which shares only _backend with the rest.

GraphMemoryStore mixes this in, so every method is called as before.

GraphQueryMixin

Filtered reads over a :class:GraphBackend.

_backend is supplied by the host store; it is declared here so the mixin's reads are typed rather than implicit.

delete_subtree async

delete_subtree(scope: TenantScope) -> int

Erase scope and every descendant scope. Returns the node count.

Forwarded unchanged, and deliberately without the retraction filter every read on this wrapper applies. Erasure is not retrieval: a retracted memory is still a stored one, and a privacy sweep that skipped it would leave behind exactly the rows a subject asked to have removed while reporting success.

Forwarded at all because this wrapper is what the factories construct and what MemoryOrchestrator holds, so it is where production wiring reaches for a sweep. Without this method the only way to get one is to reach past the wrapper to the raw backend, which silently skips the access-count and retraction behaviour its sibling reads depend on.

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

async def delete_subtree(self, scope: TenantScope) -> int:
    """Erase ``scope`` and every descendant scope. Returns the node count.

    Forwarded unchanged, and deliberately *without* the retraction filter
    every read on this wrapper applies. Erasure is not retrieval: a
    retracted memory is still a stored one, and a privacy sweep that
    skipped it would leave behind exactly the rows a subject asked to have
    removed while reporting success.

    Forwarded at all because this wrapper is what the factories construct
    and what ``MemoryOrchestrator`` holds, so it is where production wiring
    reaches for a sweep. Without this method the only way to get one is to
    reach past the wrapper to the raw backend, which silently skips the
    access-count and retraction behaviour its sibling reads depend on.
    """
    return await self._backend.delete_subtree(scope)

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_queries.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()
    ]

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_queries.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]

query_subtree async

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

Read scope and every descendant scope — the mirror of the sweep.

Unlike :meth:delete_subtree this does filter retracted rows by default, because it is a read and every other read here does. The two differ for the reason the pair exists: the erase verb must reach everything stored, the read verb must not surface what was corrected.

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

async def query_subtree(
    self,
    scope: TenantScope,
    filters: dict[str, object] | None = None,
    limit: int | None = None,
    include_retracted: bool = False,
) -> list[MemoryNode]:
    """Read ``scope`` and every descendant scope — the mirror of the sweep.

    Unlike :meth:`delete_subtree` this *does* filter retracted rows by
    default, because it is a read and every other read here does. The two
    differ for the reason the pair exists: the erase verb must reach
    everything stored, the read verb must not surface what was corrected.
    """
    nodes = await self._backend.query_subtree(scope, filters or {}, limit)
    if include_retracted:
        return list(nodes)
    return [node for node in nodes if not is_retracted(node)]