Skip to content

symfonic.capabilities.memory.status

status

Scoped semantic block status over the public graph backend protocol.

MemoryStatusService 

MemoryStatusService(graph: Any = None, *, required_labels: Iterable[str] = (), enabled_layers: Iterable[str] = ('semantic',), records: Any = None)

Report discovered labels and required pending blocks for one scope.

The graph backend is the same scoped backend used by native memory writes. enabled_layers is deployment policy, not inferred from stored rows.

Source code in src/symfonic/capabilities/memory/status.py 
29
30
31
32
33
34
35
36
37
38
39
40
def __init__(
    self,
    graph: Any = None,
    *,
    required_labels: Iterable[str] = (),
    enabled_layers: Iterable[str] = ("semantic",),
    records: Any = None,
) -> None:
    self._graph = graph
    self._records = records
    self._required_labels = tuple(required_labels)
    self._enabled_layers = frozenset(enabled_layers)

blocks async 

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

Return the legacy status fields without reaching an orchestrator.

Source code in src/symfonic/capabilities/memory/status.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
async def blocks(self, scope: MemoryScope) -> list[dict[str, Any]]:
    """Return the legacy status fields without reaching an orchestrator."""
    # A kind-aware scope has no positional graph address; a deployment wired
    # with records alone has no graph to query. Both read the inventory.
    if scope.identity_version or (self._graph is None and self._records is not None):
        return self._results(await self._native_groups(scope))
    nodes = (
        await self._graph.query_nodes(
            tenant_scope(scope),
            {"layer": MemoryLayer.SEMANTIC},
            limit=None,
        )
        if self._graph is not None
        else ()
    )
    groups: dict[str, dict[str, Any]] = {}
    for node in nodes:
        # Legacy distinct_labels excludes retracted rows only. Native status
        # also hides unpublished pending writes: a draft cannot satisfy a
        # required block before publish.
        if is_pending(node.properties) or is_retracted(node.properties):
            continue
        entry = groups.setdefault(node.label, {"count": 0, "last_updated": None})
        entry["count"] += 1
        stamp = store_time(node.updated_at)
        if stamp is not None and (
            entry["last_updated"] is None or stamp > entry["last_updated"]
        ):
            entry["last_updated"] = stamp
    return self._results(groups)