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)

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

Source code in src/symfonic/capabilities/memory/phases/window.py

def __init__(self, graph: Any, *, lookback_hours: float = 24.0) -> None:
    self._graph = graph
    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

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
    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