Skip to content

symfonic.capabilities.memory.phases.window

window

The recently-updated semantic rows a cycle works over, read once.

Legacy's quick_nap queried the semantic layer once, filtered it to the lookback window, and handed the same list to phase 1 and phase 5. Splitting those into two phase objects would have split the query too -- two full scans of a tenant's semantic graph per nap, on the one cadence that runs inside a turn.

Memoised on the cycle, not on the phase. The key is the scope path and the cycle's start instant, so two scopes consolidating concurrently never see each other's window, and a second cycle over the same scope re-reads rather than replaying a stale one. One entry is kept: a cycle runs its phases back to back, and a cache that grew would be a per-tenant copy of the semantic graph held for the lifetime of the process.

RecentSemanticNodes 

RecentSemanticNodes(graph: Any, *, lookback_hours: float = 24.0, exact_session_owner: bool = False, session_kind: str | None = None)

One scan of a scope's recent semantic rows, shared across a cycle.

Source code in src/symfonic/capabilities/memory/phases/window.py 
39
40
41
42
43
44
45
46
def __init__(self, graph: Any, *, lookback_hours: float = 24.0,
             exact_session_owner: bool = False, session_kind: str | None = None) -> None:
    self._graph = graph
    self._exact_session_owner = exact_session_owner
    self._session_kind = session_kind
    self._lookback_hours = lookback_hours
    self._key: tuple[str, datetime] | None = None
    self._nodes: list[Any] = []

for_cycle async 

for_cycle(context: Any) -> list[Any]

The window for context's cycle, querying at most once.

Source code in src/symfonic/capabilities/memory/phases/window.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
async def for_cycle(self, context: Any) -> list[Any]:
    """The window for ``context``'s cycle, querying at most once."""
    key = (context.scope.path, context.started_at)
    if self._key == key:
        return self._nodes
    if self._exact_session_owner:
        scope = context.scope
        if not isinstance(scope, MemoryScope):
            raise MemoryContractError("exact QUICK owner requires a terminal session scope")
        if scope.identity_version:
            if (not self._session_kind or scope.levels[-1][0] != self._session_kind):
                raise MemoryContractError(
                    "native QUICK requires the app's terminal session_kind"
                )
            require_native_quick_graph(self._graph)
            storage_scope = NativeStorageScope.from_scope(scope)
        else:
            if not scope.session:
                raise MemoryContractError("exact QUICK owner requires a terminal session scope")
            storage_scope = tenant_scope(scope)
        try:
            nodes = await self._graph.query_subtree(
                storage_scope, {"layer": MemoryLayer.SEMANTIC.value}, limit=None,
            )
        except (AttributeError, NotImplementedError, TypeError) as exc:
            raise MemoryContractError("exact QUICK owner requires query_subtree") from exc
        if scope.identity_version:
            nodes = [node for node in nodes if node_scope(node) == scope]
        if any(node_scope(node) != scope or node.layer is not MemoryLayer.SEMANTIC
               for node in nodes):
            raise MemoryContractError("query_subtree returned a row outside exact QUICK owner")
    else:
        nodes = await self._graph.query_nodes(
            tenant_scope(context.scope), layer=MemoryLayer.SEMANTIC
        )
    cutoff = datetime.now(UTC) - timedelta(hours=self._lookback_hours)
    recent = [n for n in nodes if n.updated_at and n.updated_at >= cutoff]
    # Assigned together and only on success: a query that raised must not
    # leave a key claiming a window it does not have, or the next phase in
    # the same cycle would read an empty list as a real answer.
    self._key, self._nodes = key, recent
    return recent