symfonic.capabilities.memory.cycle_state

cycle_state

What one consolidation cycle did, in three vocabularies.

Split from :mod:symfonic.capabilities.memory.consolidation so the record is readable apart from the runtime that fills it, and because the combined module went over its 300-line budget once the cycle ledger was added.

Three vocabularies, because three readers ask different questions:

• :attr:ConsolidationState.mutations — what each phase reported. The capability's own terms, keyed by roster name.
• :meth:ConsolidationState.to_legacy_dict — the shipped ConsolidationReport.to_dict() shape, so a dashboard written against the old report keeps working while the runtime under it changes.
• :meth:ConsolidationState.telemetry — the safe ledger. Integers and roster names, nothing else: no memory text, no embedding, no metadata, no phase argument. A consolidation reads a tenant's memories, so its telemetry is the one place a stray f-string would publish them.

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

already_running

already_running(scope: Any, cycle: ConsolidationCycle) -> ConsolidationState

The record a worker that lost the lease returns.

A record rather than None: the caller needs to know why nothing happened, and an absent answer reads exactly like a cycle that ran and found nothing to do. Finished, because this worker is finished; empty everywhere else, because it did nothing.

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

def already_running(scope: Any, cycle: ConsolidationCycle) -> ConsolidationState:
    """The record a worker that lost the lease returns.

    A record rather than ``None``: the caller needs to know *why* nothing
    happened, and an absent answer reads exactly like a cycle that ran and
    found nothing to do. Finished, because this worker is finished; empty
    everywhere else, because it did nothing.
    """
    state = ConsolidationState(
        scope_path=scope.path, cycle=cycle, already_running=True
    )
    state.finished_at = state.started_at
    return state

merge_ledger

merge_ledger(target: dict[str, int], source: Mapping[str, int]) -> None

Add one phase's counters into the cycle's, refusing anything unsafe.

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

def merge_ledger(target: dict[str, int], source: Mapping[str, int]) -> None:
    """Add one phase's counters into the cycle's, refusing anything unsafe."""
    for name, delta in source.items():
        target[name] = target.get(name, 0) + _safe_delta(name, delta)