`symfonic.capabilities.memory.graph_admin`¶

graph_admin ¶

Reading the memory graph: edges, neighbourhoods, and a portable export.

MemoryAdminService answers what a scope remembers -- the published records. This answers how those records relate, which is a different question with a different backend call, and the one a graph view and a GDPR export both need.

Split rather than added because the two have different blast radii. Records are what a turn recalls; edges are what a maintenance job rewrites and a privacy request must include. An adopter mounting a read-only graph view should not have to hold the object that can publish and forget.

Why it takes a backend and not a store. Relationships live on the GraphBackend protocol -- query_edges, get_neighbors, traverse -- and the capability-native store deliberately exposes only the three memory ports. Reaching through the store for an edge would make every store owe a graph, when most deployments want one.

GraphAdminService ¶

GraphAdminService(graph: Any, records: Any = None, *, conversation: Any = None)

Relationship reads for one deployment's graph backend.

Parameters:

Name	Type	Description	Default
`graph`	`Any`	the `GraphBackend` the memory store persists through.	required
`records`	`Any`	a `MemoryAdminService`, needed only by :meth:`export`. Absent, an export would contain relationships and no memories, which is a worse answer to a portability request than an error.	`None`

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

def __init__(
    self, graph: Any, records: Any = None, *, conversation: Any = None
) -> None:
    """
    Args:
        graph: the ``GraphBackend`` the memory store persists through.
        records: a ``MemoryAdminService``, needed only by :meth:`export`.
            Absent, an export would contain relationships and no memories,
            which is a worse answer to a portability request than an error.
    """
    self._graph = graph
    self._records = records
    self._conversation = conversation if conversation is not None else getattr(
        records, "_conversation", None
    )

clusters `async` ¶

clusters(scope: MemoryScope) -> list[dict[str, Any]]

Published semantic label groups, including visible ancestors.

Unlike legacy clusters, this hides staged and quarantined writes. It is not the descendant inventory served by :meth:nodes. Reads never increment counters.

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

async def clusters(self, scope: MemoryScope) -> list[dict[str, Any]]:
    """Published semantic label groups, including visible ancestors.

    Unlike legacy clusters, this hides staged and quarantined writes. It
    is not the descendant inventory served by :meth:`nodes`. Reads never
    increment counters.
    """
    from symfonic.memory.pending import is_pending
    from symfonic.memory.retraction import is_retracted
    from symfonic.memory.types import MemoryLayer

    nodes = await self._graph.query_nodes(
        tenant_scope(scope), {"layer": MemoryLayer.SEMANTIC}, limit=None,
    )
    groups: dict[str, list[Any]] = {}
    for node in nodes:
        if is_pending(node.properties) or is_retracted(node.properties):
            continue
        label = str(node.label)
        prefix = label.split(":", 1)[0].strip() if ":" in label else (
            label.split()[0] if label else "unknown"
        )
        groups.setdefault(prefix, []).append(node)
    return [
        {
            "prefix": prefix,
            "count": len(members),
            "members": [{"id": str(node.id), "label": str(node.label)}
                        for node in members],
            "representative": {
                "id": str(best.id), "label": str(best.label),
            },
        }
        for prefix, members in sorted(
            groups.items(), key=lambda item: len(item[1]), reverse=True,
        )
        for best in [max(members, key=lambda node: node.importance)]
    ]

edges `async` ¶

edges(scope: MemoryScope, *, limit: int = 500, offset: int = 0, admit: Any = None) -> list[dict[str, Any]]

One bounded page of relationships, in newest backend order.

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

async def edges(
    self, scope: MemoryScope, *, limit: int = 500, offset: int = 0, admit: Any = None,
) -> list[dict[str, Any]]:
    """One bounded page of relationships, in newest backend order."""
    found = await self._graph.query_edges(
        tenant_scope(scope), {}, limit=limit, offset=offset
    )
    from symfonic.capabilities.memory.graph_owner_reads import admitted_edges
    return [edge_body(edge) for edge in await admitted_edges(self._graph, scope, found, admit)]

export `async` ¶

export(scope: MemoryScope, *, admit: Any = None) -> dict[str, Any]

Everything this scope owns, as a portable document (GDPR Art. 20).

complete is part of the payload rather than an exception, because a partial export is still owed to the subject -- and a request that silently returned nine layers of ten would be a portability failure nobody could see. What could not be read is named.

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

async def export(self, scope: MemoryScope, *, admit: Any = None) -> dict[str, Any]:
    """Everything this scope owns, as a portable document (GDPR Art. 20).

    ``complete`` is part of the payload rather than an exception, because a
    partial export is still owed to the subject -- and a request that
    silently returned nine layers of ten would be a portability failure
    nobody could see. What could not be read is named.
    """
    from symfonic.capabilities.memory.graph_owner_reads import export
    return await export(
        self, scope, admit, body=_record_body, schema_version=EXPORT_SCHEMA_VERSION,
    )

neighborhood `async` ¶

neighborhood(scope: MemoryScope, node_id: str, *, depth: int = 2, min_weight: float = 0, admit: Any = None) -> dict[str, Any]

A visible center, filtered BFS neighbors, and induced edges.

Neighbor reachability is independent of min_weight: that parameter filters only returned edges, as on the established graph route.

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

async def neighborhood(
    self, scope: MemoryScope, node_id: str, *, depth: int = 2,
    min_weight: float = 0, admit: Any = None,
) -> dict[str, Any]:
    """A visible center, filtered BFS neighbors, and induced edges.

    Neighbor reachability is independent of ``min_weight``: that parameter
    filters only returned edges, as on the established graph route.
    """
    from symfonic.memory.retraction import is_retracted
    from symfonic.memory.types import NodeId

    # An overriding subclass must declare support for its own query_edges;
    # inheriting the parent's marker does not certify a replacement.
    if type(self._graph).__dict__.get("supports_visible_edge_owners") is not True:
        raise RuntimeError("graph backend lacks visible edge owner reads")
    trusted = tenant_scope(scope)
    center = await self._graph.get_node(trusted, NodeId(node_id))
    if center is None or is_retracted(center.properties) or (
        admit is not None and not await admit(center)
    ):
        raise GraphNodeNotFound(f"Node {node_id} not found")

    visited = {node_id}
    neighbors: list[dict[str, Any]] = []
    queue = deque([(node_id, 0)])
    while queue:
        current, distance = queue.popleft()
        if distance >= depth:
            continue
        incident = await self._graph.query_edges(
            trusted, {"incident_node_id": current, "visible_edge_owners": True},
            limit=None,
        )
        for edge in incident:
            source, target = str(edge.source), str(edge.target)
            if current not in (source, target):
                continue
            neighbor_id = target if source == current else source
            if neighbor_id in visited:
                continue
            node = await self._graph.get_node(trusted, NodeId(neighbor_id))
            if node is None or is_retracted(node.properties) or (
                admit is not None and not await admit(node)
            ):
                continue
            visited.add(neighbor_id)
            neighbors.append({**_node_body(node), "depth": distance + 1})
            queue.append((neighbor_id, distance + 1))

    # The backend filters both endpoints before sorting or limiting. A
    # single call avoids repeatedly scanning unrelated tenant edges and
    # keeps the read proportional to the output's induced edge set.
    induced = await self._graph.query_edges(
        trusted, {"induced_node_ids": tuple(sorted(visited)),
                  "visible_edge_owners": True},
        limit=None,
    )
    edges = []
    for edge in induced:
        if (str(edge.source) not in visited or str(edge.target) not in visited
            or edge.weight < min_weight):
            continue
        serialized = edge_body(edge)
        edges.append({key: serialized[key] for key in (
            "id", "source", "target", "relationship", "weight",
        )})
    return {"center": _node_body(center), "neighbors": neighbors, "edges": edges}

nodes `async` ¶

nodes(scope: MemoryScope, *, limit: int = 500, offset: int = 0, admit: Any = None) -> list[dict[str, Any]]

Drawable nodes owned by scope, including its descendants.

A graph browser and a recall answer are different views. Recall is ranked and ancestor-facing; a graph needs the endpoints of the edges it was given, including conversation descendants, or it silently drops almost every relationship as dangling.

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

async def nodes(
    self, scope: MemoryScope, *, limit: int = 500, offset: int = 0, admit: Any = None,
) -> list[dict[str, Any]]:
    """Drawable nodes owned by ``scope``, including its descendants.

    A graph browser and a recall answer are different views.  Recall is
    ranked and ancestor-facing; a graph needs the endpoints of the edges
    it was given, including conversation descendants, or it silently drops
    almost every relationship as dangling.
    """
    legacy = tenant_scope(scope)
    from symfonic.capabilities.memory.graph_owner_reads import admitted_nodes
    query = getattr(self._graph, "query_subtree", None)
    if limit <= 0:
        return []
    native_page = getattr(self._graph, "query_subtree_page", None)
    if callable(native_page):
        found = await native_page(legacy, {}, limit=limit, offset=offset)
        return [_node_body(node) for node in await admitted_nodes(found, admit)]
    # GraphBackend's historical node contract has no offset. Fetch only
    # through the requested page and slice here; this bounds process memory
    # while keeping custom backends compatible. Native cursor pagination
    # can replace this without changing the public admin/API contract.
    through = offset + limit
    found = await (
        query(legacy, {}, limit=through)
        if callable(query)
        else self._graph.query_nodes(legacy, {}, limit=through)
    )
    return [_node_body(node) for node in await admitted_nodes(found[offset:through], admit)]

shortest_path `async` ¶

shortest_path(scope: MemoryScope, start: str, end: str, *, admit: Any = None) -> dict[str, Any]

Fewest-hop published path through visible nodes, without counters.

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

async def shortest_path(
    self, scope: MemoryScope, start: str, end: str, *, admit: Any = None,
) -> dict[str, Any]:
    """Fewest-hop published path through visible nodes, without counters."""
    from symfonic.capabilities.memory.graph_paths import shortest_path

    return await shortest_path(self._graph, scope, start, end, admit=admit)

GraphNodeNotFound ¶

Bases: LookupError

The requested neighborhood center is absent from the caller's scope.

edge_body ¶

edge_body(edge: Any) -> dict[str, Any]

One relationship as JSON. Named fields, so a model gaining an internal one does not silently widen a privacy export.

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

def edge_body(edge: Any) -> dict[str, Any]:
    """One relationship as JSON. Named fields, so a model gaining an internal
    one does not silently widen a privacy export."""
    return {
        "id": str(getattr(edge, "id", "")),
        "source": str(getattr(edge, "source", "")),
        "target": str(getattr(edge, "target", "")),
        "relationship": str(getattr(edge, "relationship", "")),
        "weight": getattr(edge, "weight", None),
        "uses": getattr(edge, "uses", None),
    }

symfonic.capabilities.memory.graph_admin¶

graph_admin ¶

GraphAdminService ¶

clusters async ¶

edges async ¶

export async ¶

neighborhood async ¶

nodes async ¶

shortest_path async ¶

GraphNodeNotFound ¶

edge_body ¶

`symfonic.capabilities.memory.graph_admin`¶

clusters `async` ¶

edges `async` ¶

export `async` ¶

neighborhood `async` ¶

nodes `async` ¶

shortest_path `async` ¶