Skip to content

symfonic.services.switching.evidence

evidence

SCP-FRZ-2 / SCP-REV-4 — retirement evidence, keyed to its freeze epoch.

Evidence is never "the drain was clean"; it is "the drain was clean under freeze epoch F". Keying it that way is what lets revocation invalidate a whole generation of results in one operation, and what makes reads that filter on the current epoch a structural guarantee instead of a review checklist item.

EvidenceInvalidated dataclass 

EvidenceInvalidated(freeze_epoch_id: str, bundle_id: str, reason: str, evidence_ids: tuple[str, ...], kinds: tuple[str, ...], invalidated_at: float = 0.0)

The event T4.4.7 consumes when a freeze revocation lands.

RetirementEvidence dataclass 

RetirementEvidence(evidence_id: str, freeze_epoch_id: str, bundle_id: str, kind: str, detail: str, recorded_at: float = 0.0, invalidated_by: str | None = None)

One gathered result, permanently attributed to one freeze epoch.

RetirementEvidenceStore 

RetirementEvidenceStore(*, clock: Callable[[], float] = time.time)

Append-only evidence with epoch-scoped invalidation and a subscriber seam.

Source code in src/symfonic/services/switching/evidence.py 
58
59
60
61
62
def __init__(self, *, clock: Callable[[], float] = time.time) -> None:
    self._rows: list[RetirementEvidence] = []
    self._subscribers: list[Callable[[EvidenceInvalidated], None]] = []
    self._seq = itertools.count(1)
    self._clock = clock

covers 

covers(freeze_epoch_id: str, kinds: Sequence[str] = ()) -> bool

Whether every required evidence kind is present and still valid.

Source code in src/symfonic/services/switching/evidence.py 
92
93
94
95
96
97
98
def covers(
    self, freeze_epoch_id: str, kinds: Sequence[str] = ()
) -> bool:
    """Whether every required evidence kind is present and still valid."""
    required = frozenset(kinds) if kinds else REQUIRED_KINDS
    present = {row.kind for row in self.valid_for(freeze_epoch_id)}
    return required <= present

invalidate_epoch 

invalidate_epoch(freeze_epoch_id: str, *, reason: str) -> EvidenceInvalidated

Atomically mark every row of this epoch invalid and emit the event.

Source code in src/symfonic/services/switching/evidence.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def invalidate_epoch(self, freeze_epoch_id: str, *, reason: str) -> EvidenceInvalidated:
    """Atomically mark every row of this epoch invalid and emit the event."""
    affected: list[RetirementEvidence] = []
    for index, row in enumerate(self._rows):
        if row.freeze_epoch_id == freeze_epoch_id and row.valid:
            marked = replace(row, invalidated_by=freeze_epoch_id)
            self._rows[index] = marked
            affected.append(marked)
    event = EvidenceInvalidated(
        freeze_epoch_id=freeze_epoch_id,
        bundle_id=affected[0].bundle_id if affected else "",
        reason=reason,
        evidence_ids=tuple(row.evidence_id for row in affected),
        kinds=tuple(sorted({row.kind for row in affected})),
        invalidated_at=self._clock(),
    )
    for subscriber in self._subscribers:
        subscriber(event)
    return event

valid_for 

valid_for(freeze_epoch_id: str) -> tuple[RetirementEvidence, ...]

SCP-REV-4: reads filter on non-invalidated AND matching epoch.

Source code in src/symfonic/services/switching/evidence.py 
81
82
83
84
85
86
87
def valid_for(self, freeze_epoch_id: str) -> tuple[RetirementEvidence, ...]:
    """SCP-REV-4: reads filter on non-invalidated AND matching epoch."""
    return tuple(
        row
        for row in self._rows
        if row.freeze_epoch_id == freeze_epoch_id and row.valid
    )