symfonic.capabilities.memory.consolidation

consolidation

Deliverable 3: one runtime behind quick nap, nightly nap, and Deep Sleep.

The shipped consolidator is a 600-line method per cadence with the roster inlined: run names its phases in sequence, quick_nap re-implements a four-phase subset, and nightly_nap calls run while temporarily mutating instance state to apply per-call overrides. Adding a phase means editing every cadence that should have it, and forgetting one is invisible until an adopter reports that a configuration flag does nothing.

Here the cadence is a roster (:data:PHASE_ROSTER) and the runtime is the thing that executes one. The properties that make a half-finished run safe are the runtime's, so every phase gets them for free:

• A roster is complete or the runtime refuses it. A cycle whose roster named phases nothing implemented used to run the ones it had, skip the rest and report clean — so a runtime composed with no phases at all answered "consolidation succeeded" having done nothing, which is the worst available answer: it is indistinguishable from a scope with nothing left to do. Completeness is checked at construction for every declared cadence, and again at :meth:ConsolidationRuntime.run for a runtime that declared none.
• A phase's name is recorded before it runs. Presence with a zero counter means "ran and found nothing"; absence means "never ran". Both produce a zero otherwise, and they call for opposite responses.
• A phase may decline. applies is how a phase says it has nothing to work on — no episodic layer, no profile schema, no judge model — which is the condition legacy expressed by wrapping the call in an if. It lands in skipped, which is a different sentence from "no implementation" and now cannot be confused with it.
• A failing phase does not end the cycle. Consolidation is maintenance: the pruning failing is no reason to skip the decay.
• Cancellation is not a failure. CancelledError propagates unswallowed (CXL-2), so a cancelled run is never reported as a completed one — and because cancellation skips the commit, a cycle cut short publishes nothing.
• A cycle that recorded an error publishes nothing. Writes stay pending, so a partial consolidation is invisible until a clean cycle commits it. This is the rollback the write/flush split exists to provide.
• A cycle that lost its scope stops rather than finishing. still_authorised is the lease, asked before every durable mutation (through :class:~.fencing.FencedGraph) and once more at the end. The status is then lease_lost, which is neither clean nor degraded: the work is being redone properly by whoever holds the scope now, and this worker standing down is the mechanism working rather than anything to alert on.

What runs inside a phase is not the runtime's business — a phase reaches whatever store it owns and returns how many things it changed. That is what lets the same runtime drive the in-process reference HMS and a graph backend.

ConsolidationPhase

Bases: Protocol

One maintenance step. Named, so a roster can ask for it.

name property

name: str

The roster name this phase answers to.

run async

run(context: PhaseContext) -> int

Do the work; return how many things changed.

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

async def run(self, context: PhaseContext) -> int:
    """Do the work; return how many things changed."""
    ...

ConsolidationRuntime

ConsolidationRuntime(*, phases: Sequence[ConsolidationPhase] = (), writes: MemoryWriteCoordinator | None = None, cycles: Sequence[ConsolidationCycle] = (), participants: Sequence[Any] = (), max_new_edges_per_sweep: int | None = None)

Runs one cadence's roster over one scope.

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

def __init__(
    self,
    *,
    phases: Sequence[ConsolidationPhase] = (),
    writes: MemoryWriteCoordinator | None = None,
    cycles: Sequence[ConsolidationCycle] = (),
    #: The stores this runtime's phases write through, for a composition
    #: root to check against its transaction domain *before* a cycle runs.
    #: Taken from the roster when the phases came from one of the shipped
    #: factories, which is where the graph was resolved; pass it explicitly
    #: for a hand-built roster, or the mismatch is only caught at the
    #: commit -- after seventeen phases and their model calls.
    participants: Sequence[Any] = (),
    #: Gross new edges per durable scope/cadence sweep; None is unlimited.
    #: Exhaustion defers groups, not cursor progress. Mid-sweep changes refuse.
    max_new_edges_per_sweep: int | None = None,
) -> None:
    from symfonic.capabilities.memory.growth import validate_limit
    validate_limit(max_new_edges_per_sweep)
    self._edge_limit = max_new_edges_per_sweep
    self._phases = {phase.name: phase for phase in phases}
    self._writes = writes
    from symfonic.capabilities.memory.cycle_writes import validate_writes
    if self._edge_limit is not None:
        validate_writes(writes)
    declared = getattr(phases, "graph", None)
    self._participants: tuple[Any, ...] = tuple(participants) or (
        (declared,) if declared is not None else ()
    )
    for cycle in cycles:
        self._require_complete(cycle, at="composition")

participants property

participants: tuple[Any, ...]

The stores this runtime's phases write through.

Published so a composition root can prove, before any cycle runs, that the graph the phases mutate is in the same transaction domain as the lease that authorises them. Empty for a hand-built roster that declared none, which is the one case where the check has to wait for the commit.

writes property

writes: MemoryWriteCoordinator | None

The coordinator this runtime publishes through, if it has one.

Published so a composition root can check, before any cycle runs, that the staged records and the graph mutations share one transaction domain.

run async

run(scope: MemoryScope, cycle: ConsolidationCycle, *, run_id: str = '', root_run_id: str = '', still_authorised: Any = None, commit_authority: Any = None, domain: Any = None) -> ConsolidationState

Execute cycle's roster over scope and report what happened.

run_id/root_run_id name the turn this cycle is attributable to. A quick nap fires from a turn and one of its phases can spend a model call, so the cost has an owner rather than landing on whichever run was in flight when the background task was scheduled. Empty is the honest answer for a cycle a scheduler ran outside any turn.

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

async def run(
    self,
    scope: MemoryScope,
    cycle: ConsolidationCycle,
    *,
    run_id: str = "",
    root_run_id: str = "",
    still_authorised: Any = None,
    commit_authority: Any = None,
    domain: Any = None,
) -> ConsolidationState:
    """Execute ``cycle``'s roster over ``scope`` and report what happened.

    ``run_id``/``root_run_id`` name the turn this cycle is attributable to.
    A quick nap fires from a turn and one of its phases can spend a model
    call, so the cost has an owner rather than landing on whichever run was
    in flight when the background task was scheduled. Empty is the honest
    answer for a cycle a scheduler ran outside any turn.
    """
    # Second gate, for a runtime composed without declaring its cadences.
    # A caller that asks for a cycle this runtime cannot serve gets an
    # error rather than a state, because a state is a report of work.
    self._require_complete(cycle, at="the call")

    state = ConsolidationState(
        scope_path=scope.path,
        cycle=cycle,
        registered=self.registered,
        run_id=run_id,
        root_run_id=root_run_id,
    )
    from symfonic.capabilities.memory.cycle_writes import prepare_cycle
    buffered, domain = prepare_cycle(self._writes, self._edge_limit, domain)
    context = PhaseContext(
        scope=scope,
        cycle=cycle,
        started_at=state.started_at,
        writes=buffered,
        run_id=run_id,
        root_run_id=root_run_id,
    )
    fence = (
        Fence(still_authorised, commit_authority)
        if still_authorised is not None
        else None
    )
    journal = CycleJournal(cycle=cycle.value, max_new_edges_per_sweep=self._edge_limit)
    # The phases run against the journal: their mutations are held, and
    # read back, but nothing is durable until this cycle earns it below.
    async with held(fence), journalled(journal):
        await self._roster(cycle, context, state, journal)
        await still_ours(state, fence)
        state.ledger = dict(context.ledger)
        state.counters.update(context.legacy)
        state.ledger["candidates"] = len(context.seen)
        state.committed = await publish_cycle(
            scope, state, journal, fence, domain, buffered
        )
    state.ledger["published"] = len(state.committed)
    state.ledger.update(journal.growth_counts)
    state.finished_at = datetime.now(UTC)
    return state

serves

serves(cycle: ConsolidationCycle) -> bool

Whether every phase on cycle's roster has an implementation.

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

def serves(self, cycle: ConsolidationCycle) -> bool:
    """Whether every phase on ``cycle``'s roster has an implementation."""
    return not self._missing(cycle)

ConsolidationState dataclass

ConsolidationState(scope_path: str, cycle: ConsolidationCycle, tenant_id: str = '', started_at: datetime = (lambda: datetime.now(UTC))(), finished_at: datetime | None = None, run_id: str = '', root_run_id: str = '', registered: tuple[str, ...] = (), phases_run: tuple[str, ...] = (), skipped: tuple[str, ...] = (), failed: tuple[str, ...] = (), deferred: tuple[str, ...] = (), mutations: dict[str, int] = dict(), ledger: dict[str, int] = dict(), counters: dict[str, int] = dict(), errors: tuple[str, ...] = (), committed: tuple[str, ...] = (), already_running: bool = False, lease_lost: bool = False)

What one cycle did — in this capability's terms and in legacy's.

clean property

clean: bool

Whether every phase that ran, ran without error.

duration_ms property

duration_ms: int

How long the cycle took, or 0 while it is still running.

status property

status: str

The cycle's final word.

running, clean, degraded -- and two more that are none of those and must not be reported as any of them.

already_running: another holder had the scope, so this cycle never started. Saying clean would make "somebody else is consolidating this" indistinguishable from "there was nothing to consolidate", and a scheduler reading a dashboard would conclude the cycle had run.

lease_lost: this cycle started, then lost the scope partway. Not degraded either, though it is a kind of failure, because the two want opposite responses: degraded is a phase that broke and wants looking at, while lease_lost is a worker that correctly stood down so another one could do the work properly. Alerting on the second is alerting on the mechanism working. Ranked above degraded because a cycle that loses its lease also collects the fence's error, and the specific fact is the useful one.

telemetry

telemetry() -> dict[str, Any]

The safe record of this cycle: integers, roster names, and a status.

Everything here is either a framework constant or a count. The scope is not: tenant_id identifies whose consolidation this was, which is what makes the model cost a phase spends attributable, and it is already the key every other metric in the system carries.

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

def telemetry(self) -> dict[str, Any]:
    """The safe record of this cycle: integers, roster names, and a status.

    Everything here is either a framework constant or a count. The scope
    is *not*: ``tenant_id`` identifies whose consolidation this was, which
    is what makes the model cost a phase spends attributable, and it is
    already the key every other metric in the system carries.
    """
    return {
        "cycle_kind": self.cycle.value,
        "tenant_id": self.tenant_id or scope_from_path(self.scope_path).tenant,
        "scope_path": self.scope_path,
        # What a phase's model call is charged against. A nap fires from a
        # turn, so its cost belongs to that turn's root run rather than to
        # whichever run was in flight when the task was scheduled.
        "run_id": self.run_id,
        "root_run_id": self.root_run_id,
        "phases_registered": list(self.registered),
        "phases_run": list(self.phases_run),
        "phases_skipped": list(self.skipped),
        "phases_failed": list(self.failed),
        "phases_deferred": list(self.deferred),
        "registered": len(self.registered),
        "ran": len(self.phases_run),
        "skipped": len(self.skipped),
        "failed": len(self.failed),
        **{name: self.ledger.get(name, 0) for name in CYCLE_LEDGER},
        "duration_ms": self.duration_ms,
        "status": self.status,
    }

to_legacy_dict

to_legacy_dict() -> dict[str, Any]

The shipped ConsolidationReport.to_dict() shape, plus this cycle.

Every legacy key is present with its legacy type, so a reader written against the old report needs no change. The additions (cycle, scope_path, skipped, committed) are new keys, which a dict consumer ignores.

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

def to_legacy_dict(self) -> dict[str, Any]:
    """The shipped ``ConsolidationReport.to_dict()`` shape, plus this cycle.

    Every legacy key is present with its legacy type, so a reader written
    against the old report needs no change. The additions
    (``cycle``, ``scope_path``, ``skipped``, ``committed``) are new keys,
    which a dict consumer ignores.
    """
    payload: dict[str, Any] = {key: 0 for key in LEGACY_REPORT_KEYS}
    for phase, count in self.mutations.items():
        for counter in PHASE_COUNTERS.get(phase, ()):
            payload[counter] = payload.get(counter, 0) + count
    # Accumulated, not assigned: legacy's ``run`` does ``+=`` into these
    # fields too, and a counter fed from both sides must not lose one.
    for counter, count in self.counters.items():
        payload[counter] = payload.get(counter, 0) + count
    payload.update(
        {
            "tenant_id": self.tenant_id or scope_from_path(self.scope_path).tenant,
            "started_at": self.started_at.isoformat(),
            "finished_at": (
                self.finished_at.isoformat() if self.finished_at else None
            ),
            "errors": list(self.errors),
            "phases_run": list(self.phases_run),
            "cycle": self.cycle.value,
            "scope_path": self.scope_path,
            "skipped": list(self.skipped),
            "committed": list(self.committed),
        }
    )
    return payload

PhaseContext dataclass

PhaseContext(scope: MemoryScope, cycle: ConsolidationCycle, started_at: datetime, writes: MemoryWriteCoordinator | None = None, run_id: str = '', root_run_id: str = '', ledger: dict[str, int] = dict(), legacy: dict[str, int] = dict(), seen: set[str] = set())

What a phase is told about the cycle it is running inside.

Frozen, and the two mutable fields are the cycle's own accumulators rather than state a phase can rewrite: a phase adds to the ledger and names what it read, and cannot reach anything another phase decided.

contributed

contributed(counter: str, delta: int) -> None

Add to a legacy report counter this phase owns but cannot headline.

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

def contributed(self, counter: str, delta: int) -> None:
    """Add to a legacy report counter this phase owns but cannot headline."""
    self.legacy[counter] = self.legacy.get(counter, 0) + int(delta)

counted

counted(name: str, delta: int = 1) -> None

Add to a cycle counter. Integers and closed names only.

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

def counted(self, name: str, delta: int = 1) -> None:
    """Add to a cycle counter. Integers and closed names only."""
    merge_ledger(self.ledger, {name: delta})

examined

examined(record_ids: Iterable[Any]) -> None

Name what this phase read, for the cycle's candidates count.

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

def examined(self, record_ids: Iterable[Any]) -> None:
    """Name what this phase read, for the cycle's ``candidates`` count."""
    self.seen.update(str(record_id) for record_id in record_ids)