Skip to content

symfonic.capabilities.memory.phase_context

phase_context

What a phase is handed, and what a phase is.

Split from :mod:symfonic.capabilities.memory.consolidation for the same reason :mod:.cycle_state was: the runtime and the thing it hands to a phase are read by different people. A phase author needs this module and nothing else in the runtime; the runtime's own file is then about running a roster.

Everything here is re-exported from consolidation, which is where every phase in the tree imports it from and where the public surface stays.

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 
80
81
82
async def run(self, context: PhaseContext) -> int:
    """Do the work; return how many things changed."""
    ...

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 
62
63
64
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 
58
59
60
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 
66
67
68
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)

applies 

applies(phase: Any, context: PhaseContext) -> bool

Whether phase has anything to work on this cycle.

A phase without an applies always does. Optional because the answer is "yes" for most phases and a protocol that demanded it would be asking every implementation to write return True.

Source code in src/symfonic/capabilities/memory/phase_context.py 
85
86
87
88
89
90
91
92
93
def applies(phase: Any, context: PhaseContext) -> bool:
    """Whether ``phase`` has anything to work on this cycle.

    A phase without an ``applies`` always does. Optional because the answer is
    "yes" for most phases and a protocol that demanded it would be asking every
    implementation to write ``return True``.
    """
    decide = getattr(phase, "applies", None)
    return True if decide is None else bool(decide(context))