Skip to content

symfonic.services.effects

effects

EFX — the effect-aware security-revert fence (T2.3.8).

Atomic, generation-scoped effect admission at every port T2.3.7 classified, and a security revert that is honest about its own reach: it closes new admissions for the rejected generation, blocks fenced commits, requests cancellation, quarantines late results, drains what it can within a bounded budget, and records as exposed everything it could not stop — irreversible effects that had already left, and anything an opaque in-process extension did outside the ports entirely.

The load-bearing decision is that the fencing claim is derived rather than asserted (see :mod:exposure and :mod:incident). A revert cannot be configured to report success it did not achieve.

BoundedDrain

BoundedDrain(tracker: InFlightTracker, *, clock: Callable[[], float] = time.time, sleep: Callable[[float], Awaitable[None]] = _default_sleep, interval: float = 0.05, budget: float = 5.0)

Waits, briefly, for in-flight work to reach a terminal state.

Source code in src/symfonic/services/effects/drain.py
def __init__(
    self,
    tracker: InFlightTracker,
    *,
    clock: Callable[[], float] = time.time,
    sleep: Callable[[float], Awaitable[None]] = _default_sleep,
    interval: float = 0.05,
    budget: float = 5.0,
) -> None:
    self._tracker = tracker
    self._clock = clock
    self._sleep = sleep
    self._interval = interval
    self._budget = budget

CommitDecision

Bases: StrEnum

EFX-F-3 — what the commit-time conditional check decided.

SUPPRESSED deliberately does not say "cancelled" or "undone". The write did not land; whether an external effect already left the process is a separate question the exposure ledger answers.

CompensationOutcome dataclass

CompensationOutcome(port_id: str, invocation_id: str, invoked: bool, credentials_rotated: bool, note: str, rule_id: str | None = None)

What containment actually did. invoked is never assumed true.

CompensationRegistry

CompensationRegistry(ports: FencedPortRegistry | None = None)

Which irreversible ports have an approved remedy, and which block cutover.

Source code in src/symfonic/services/effects/compensation.py
def __init__(self, ports: FencedPortRegistry | None = None) -> None:
    self._rules: dict[str, CompensationRule] = {}
    self._ports = ports if ports is not None else FencedPortRegistry()

cutover_blocked_ports

cutover_blocked_ports(port_ids: Iterable[str]) -> tuple[str, ...]

Irreversible ports among these that carry no approved rule.

Source code in src/symfonic/services/effects/compensation.py
def cutover_blocked_ports(self, port_ids: Iterable[str]) -> tuple[str, ...]:
    """Irreversible ports among these that carry no approved rule."""
    blocked = {
        port_id
        for port_id in port_ids
        if self._ports.knows(port_id)
        and self._ports.port(port_id).irreversible
        and port_id not in self._rules
    }
    return tuple(sorted(blocked))

CompensationRule dataclass

CompensationRule(rule_id: str, port_id: str, procedure: str, tested: bool, approved_by: str, rotates_credentials: bool = False)

One documented, tested procedure for undoing what can be undone.

ContainmentCoordinator

ContainmentCoordinator(registry: CompensationRegistry, *, rotator: Callable[[str], None] | None = None)

Runs the approved remedy, or records precisely why it could not.

Source code in src/symfonic/services/effects/compensation.py
def __init__(
    self,
    registry: CompensationRegistry,
    *,
    rotator: Callable[[str], None] | None = None,
) -> None:
    self._registry = registry
    self._rotator = rotator

registry property

registry: CompensationRegistry

Exposed so an incident can ask which ports are still blocking cutover.

contain_all

contain_all(exposures: Iterable[ExposureRecord]) -> tuple[CompensationOutcome, ...]

Only irreversible-effect exposures have a port-keyed remedy.

Source code in src/symfonic/services/effects/compensation.py
def contain_all(
    self, exposures: Iterable[ExposureRecord]
) -> tuple[CompensationOutcome, ...]:
    """Only irreversible-effect exposures have a port-keyed remedy."""
    return tuple(
        self.contain(exposure)
        for exposure in exposures
        if exposure.kind is ExposureKind.IRREVERSIBLE_EFFECT
    )

DrainReport dataclass

DrainReport(drained: tuple[str, ...], contained: tuple[str, ...], budget: float, elapsed: float)

What the drain observed. complete is false if anything was contained.

EffectAdmissionController

EffectAdmissionController(*, store: InMemoryLeaseStore, fences: FenceLedger, ports: FencedPortRegistry, tracker: InFlightTracker, quarantine: ResultQuarantine, clock: Callable[[], float] = time.time)

One admission path for every classified effect port.

Source code in src/symfonic/services/effects/admission.py
def __init__(
    self,
    *,
    store: InMemoryLeaseStore,
    fences: FenceLedger,
    ports: FencedPortRegistry,
    tracker: InFlightTracker,
    quarantine: ResultQuarantine,
    clock: Callable[[], float] = time.time,
) -> None:
    self._store = store
    self._fences = fences
    self._ports = ports
    self._tracker = tracker
    self._quarantine = quarantine
    self._clock = clock

admit async

admit(*, lease_id: str, port_id: str, operation: str = '') -> EffectTicket

EFX-L-1 — no effect attempt exists without a live, unfenced lease.

Source code in src/symfonic/services/effects/admission.py
async def admit(
    self, *, lease_id: str, port_id: str, operation: str = ""
) -> EffectTicket:
    """EFX-L-1 — no effect attempt exists without a live, unfenced lease."""
    port = self._ports.port(port_id)
    lease = await self._verify(lease_id)
    ticket = EffectTicket(
        ticket_id=f"efx-{uuid.uuid4().hex[:12]}",
        lease_id=lease.lease_id,
        invocation_id=lease.invocation_id,
        port_id=port_id,
        operation=operation,
        admitted_at=self._clock(),
        irreversible=port.irreversible,
    )
    return self._tracker.open(ticket)

commit async

commit(ticket: EffectTicket, result: Any = None) -> CommitOutcome

EFX-F-3 — the conditional check that decides whether the write lands.

Source code in src/symfonic/services/effects/admission.py
async def commit(self, ticket: EffectTicket, result: Any = None) -> CommitOutcome:
    """EFX-F-3 — the conditional check that decides whether the write lands."""
    try:
        await self._verify(ticket.lease_id)
    except LeaseStoreUnavailableError as exc:
        return self._suppress(ticket, f"lease store unavailable: {exc}")
    except FencedEffectError as exc:
        return self._suppress(ticket, f"fence: {exc}")
    except LeaseInvalidError as exc:
        return self._suppress(ticket, f"lease invalid: {exc}")
    self._tracker.mark(ticket.ticket_id, TicketState.COMMITTED)
    return CommitOutcome(
        decision=CommitDecision.COMMITTED, ticket_id=ticket.ticket_id
    )

commit_or_raise async

commit_or_raise(ticket: EffectTicket, result: Any = None) -> CommitOutcome

The raising variant, for callers that treat suppression as fatal.

Source code in src/symfonic/services/effects/admission.py
async def commit_or_raise(
    self, ticket: EffectTicket, result: Any = None
) -> CommitOutcome:
    """The raising variant, for callers that treat suppression as fatal."""
    await self._verify(ticket.lease_id)
    return await self.commit(ticket, result)

deliver_result async

deliver_result(ticket: EffectTicket, result: Any) -> Any

Hand back a port's answer, or quarantine it if the fence has landed.

Source code in src/symfonic/services/effects/admission.py
async def deliver_result(self, ticket: EffectTicket, result: Any) -> Any:
    """Hand back a port's answer, or quarantine it if the fence has landed."""
    try:
        await self._verify(ticket.lease_id)
    except LeaseStoreUnavailableError as exc:
        self._quarantine.quarantine(ticket, f"lease store unavailable: {exc}")
        return None
    except FencedEffectError as exc:
        self._quarantine.quarantine(ticket, f"fence covered this result: {exc}")
        return None
    except LeaseInvalidError as exc:
        self._quarantine.quarantine(ticket, f"lease invalid: {exc}")
        return None
    return result

dispatch async

dispatch(ticket: EffectTicket) -> EffectTicket

The last check before the effect leaves. Raising here costs nothing.

Source code in src/symfonic/services/effects/admission.py
async def dispatch(self, ticket: EffectTicket) -> EffectTicket:
    """The last check before the effect leaves. Raising here costs nothing."""
    await self._verify(ticket.lease_id)
    self._tracker.mark_dispatched(ticket.ticket_id, self._clock())
    return ticket

EffectFenceError

Bases: SymfonicError

Root of the effect-admission taxonomy. Never raised directly.

EffectLease dataclass

EffectLease(lease_id: str, invocation_id: str, bundle_id: str, admitted_epoch: int, generation_vector_hash: str, tenant_scope_hash: str, acquired_at: float, expires_at: float, stale_binding: bool = False, revoked_reason: str | None = None, parent_lease_id: str | None = None)

EFX-L-1 — the revocable authorization to perform effects.

generation_vector_hash is what a revert keys on, and it is captured at acquisition rather than read at use time: an invocation must be judged on the generation it was admitted under, not on whatever is current when it happens to call.

EffectReversibility

Bases: StrEnum

EFX-F-2 — can the framework undo this port's effect after it landed?

IRREVERSIBLE is the mandatory fence-check set: external side effects, durable deletes, outbound notifications. For those, cancellation is never a remedy, so a revert must compensate and account rather than claim.

EffectTicket dataclass

EffectTicket(ticket_id: str, lease_id: str, invocation_id: str, port_id: str, operation: str, admitted_at: float, irreversible: bool)

One admitted effect attempt against one port under one lease.

ExposureRecord dataclass

ExposureRecord(invocation_id: str, kind: ExposureKind, subject: str, reason: str, ticket_id: str | None = None)

Something a revert could not suppress, named rather than glossed over.

ExtensionExecution dataclass

ExtensionExecution(invocation_id: str, extension_id: str, attested_clean: bool | None = None)

One extension the invocation actually ran.

attested_clean is tri-state on purpose. True is a sentinel attestation that no direct effect was witnessed, False is a witnessed one, and None — the default — is "nobody watched", which is not the same as "nothing happened".

Fence dataclass

Fence(fence_id: str, kind: FenceKind, reason: str, raised_at: float, raised_by: str = '', bundle_id: str | None = None, epoch_ceiling: int | None = None, rejected_vector_hash: str | None = None, tenant_scope_hash: str | None = None)

One barrier. Keyed narrowly on purpose: a revert is not an outage.

covers

covers(lease: EffectLease) -> bool

Does this fence bar the generation (or subject) the lease captured?

Source code in src/symfonic/services/effects/fence.py
def covers(self, lease: EffectLease) -> bool:
    """Does this fence bar the generation (or subject) the lease captured?"""
    if self.kind is FenceKind.SUBJECT:
        return self.tenant_scope_hash == lease.tenant_scope_hash
    if self.bundle_id is not None and self.bundle_id != lease.bundle_id:
        return False
    if self.rejected_vector_hash is not None:
        return self.rejected_vector_hash == lease.generation_vector_hash
    if self.epoch_ceiling is not None:
        return lease.admitted_epoch < self.epoch_ceiling
    return False

FenceLedger dataclass

FenceLedger(_fences: list[Fence] = list())

Append-only. Nothing here lowers a fence, and that is the point.

covering

covering(lease: EffectLease) -> Fence | None

The first fence that bars this lease, or None.

Source code in src/symfonic/services/effects/fence.py
def covering(self, lease: EffectLease) -> Fence | None:
    """The first fence that bars this lease, or ``None``."""
    for fence in self._fences:
        if fence.covers(lease):
            return fence
    return None

FencedEffectError

Bases: LeaseInvalidError

EFX-F-3 — a fence covered this operation at one of its checkpoints.

A subclass of :class:LeaseInvalidError because a fence is one specific way a lease stops authorizing effects, and because a revert revokes the leases it fences: both conditions are true afterwards, and a caller that only cares "may I proceed?" should not have to catch two types to find out. A caller that does care — an incident writer distinguishing "the fence stopped it" from "the lease aged out" — catches this one first.

FencedPort dataclass

FencedPort(port_id: str, family: EffectFamily, reversibility: EffectReversibility, externally_visible: bool, rationale: str)

One classified effect port, plus the verdict the fence needs from it.

FencedPortRegistry

FencedPortRegistry(classification: EffectPortClassification = DEFAULT_EFFECT_CLASSIFICATION)

Every classified effect port, with its reversibility verdict.

Source code in src/symfonic/services/effects/ports.py
def __init__(
    self, classification: EffectPortClassification = DEFAULT_EFFECT_CLASSIFICATION
) -> None:
    rows: dict[str, FencedPort] = {}
    missing: list[str] = []
    for port_id, port in classification.by_id.items():
        verdict = _verdict(port_id)
        if verdict is None:
            missing.append(port_id)
            continue
        reversibility, rationale = verdict
        rows[port_id] = FencedPort(
            port_id=port_id,
            family=port.family,
            reversibility=reversibility,
            externally_visible=port.externally_visible,
            rationale=rationale,
        )
    if missing:
        raise IncompleteFenceCoverageError(
            "these classified effect ports carry no reversibility verdict, so "
            "the fence cannot say whether a revert may claim to have stopped "
            f"them: {sorted(missing)}. Add a row to ports.py — an unclassified "
            "port is a hole in the evidence, not a permission."
        )
    self._by_id = MappingProxyType(rows)

assert_covers_classification

assert_covers_classification(classification: EffectPortClassification = DEFAULT_EFFECT_CLASSIFICATION) -> None

The fence's table and T2.3.7's table name the same ports, or raise.

Source code in src/symfonic/services/effects/ports.py
def assert_covers_classification(
    self, classification: EffectPortClassification = DEFAULT_EFFECT_CLASSIFICATION
) -> None:
    """The fence's table and T2.3.7's table name the same ports, or raise."""
    gap = set(classification.by_id) - set(self._by_id)
    if gap:
        raise IncompleteFenceCoverageError(
            f"the fence does not cover classified ports {sorted(gap)}"
        )

FencingClaim dataclass

FencingClaim(invocation_id: str, suppressed: bool, withheld_reason: str = '')

What the revert is entitled to say about one invocation's effects.

InFlightTracker

InFlightTracker()

Ticket lifecycle, cancellation requests, and dispatch timestamps.

Source code in src/symfonic/services/effects/tracker.py
def __init__(self) -> None:
    self._tickets: dict[str, EffectTicket] = {}
    self._states: dict[str, TicketState] = {}
    self._dispatched_at: dict[str, float] = {}
    self._cancelled: dict[str, str] = {}

dispatched_before

dispatched_before(invocation_id: str, moment: float) -> bool

Did every dispatched effect for this invocation leave before moment?

Source code in src/symfonic/services/effects/tracker.py
def dispatched_before(self, invocation_id: str, moment: float) -> bool:
    """Did every dispatched effect for this invocation leave before ``moment``?"""
    stamps = [
        at
        for ticket_id, at in self._dispatched_at.items()
        if self._tickets[ticket_id].invocation_id == invocation_id
    ]
    return all(at <= moment for at in stamps)

mark

mark(ticket_id: str, state: TicketState) -> None

Terminal states are sticky: a contained ticket never becomes drained.

Source code in src/symfonic/services/effects/tracker.py
def mark(self, ticket_id: str, state: TicketState) -> None:
    """Terminal states are sticky: a contained ticket never becomes drained."""
    current = self._states.get(ticket_id)
    if current is not None and current.terminal:
        return
    self._states[ticket_id] = state

InMemoryLeaseStore

InMemoryLeaseStore(*, fences: FenceLedger, clock: Callable[[], float] = time.time)

A single-process lease table with conditional writes (DMC-3).

Source code in src/symfonic/services/effects/store.py
def __init__(
    self,
    *,
    fences: FenceLedger,
    clock: Callable[[], float] = time.time,
) -> None:
    self._leases: dict[str, EffectLease] = {}
    self._children: dict[str, list[str]] = {}
    self._epochs: dict[str, int] = {}
    self._fences = fences
    self._clock = clock
    self._lock = asyncio.Lock()
    self.unavailable = False

lock property

lock: Lock

Held across every conditional write, and by a landing revert.

acquire_child async

acquire_child(parent_lease_id: str, *, invocation_id: str, ttl: float) -> EffectLease

EFX-L-2 — a child narrows its parent's lease and never outlives it.

Source code in src/symfonic/services/effects/store.py
async def acquire_child(
    self, parent_lease_id: str, *, invocation_id: str, ttl: float
) -> EffectLease:
    """EFX-L-2 — a child narrows its parent's lease and never outlives it."""
    async with self._lock:
        self._require_reachable()
        parent = self._require_live_locked(parent_lease_id)
        now = self._clock()
        return self._acquire_locked(
            invocation_id=invocation_id,
            bundle_id=parent.bundle_id,
            admitted_epoch=parent.admitted_epoch,
            generation_vector_hash=parent.generation_vector_hash,
            tenant_scope_hash=parent.tenant_scope_hash,
            ttl=ttl,
            stale_binding=parent.stale_binding,
            parent_lease_id=parent.lease_id,
            expires_at=min(now + ttl, parent.expires_at),
        )

check async

check(lease_id: str) -> EffectLease

One atomic verification that this lease may still perform effects.

Source code in src/symfonic/services/effects/store.py
async def check(self, lease_id: str) -> EffectLease:
    """One atomic verification that this lease may still perform effects."""
    async with self._lock:
        return self.check_locked(lease_id)

check_locked

check_locked(lease_id: str) -> EffectLease

Fence first, then lease state. Caller holds lock.

The fence is consulted before liveness because a revert revokes what it fences, so both are true afterwards and the more specific answer is the useful one.

Source code in src/symfonic/services/effects/store.py
def check_locked(self, lease_id: str) -> EffectLease:
    """Fence first, then lease state. Caller holds ``lock``.

    The fence is consulted before liveness because a revert revokes what it
    fences, so both are true afterwards and the more specific answer is the
    useful one.
    """
    self._require_reachable()
    lease = self._leases.get(lease_id)
    if lease is not None:
        fence = self._fences.covering(lease)
        if fence is not None:
            raise FencedEffectError(
                f"a {fence.kind.value} fence ({fence.fence_id}: {fence.reason}) "
                f"covers invocation {lease.invocation_id!r}; no effect is "
                "admitted, dispatched, or committed under a rejected "
                "generation or a fenced subject"
            )
    return self._require_live_locked(lease_id)

expire_due

expire_due(now: float) -> tuple[str, ...]

EFX-L-5 — a crashed worker's lease expires and stops holding the drain.

Source code in src/symfonic/services/effects/store.py
def expire_due(self, now: float) -> tuple[str, ...]:
    """EFX-L-5 — a crashed worker's lease expires and stops holding the drain."""
    expired: list[str] = []
    for lease in tuple(self._leases.values()):
        if lease.revoked_reason is None and now >= lease.expires_at:
            expired.extend(self._revoke_locked(lease.lease_id, "lease expired"))
    return tuple(dict.fromkeys(expired))

revoke_covered

revoke_covered(fence: Fence, reason: str) -> tuple[str, ...]

Close every live lease this fence covers. Caller holds lock.

Source code in src/symfonic/services/effects/store.py
def revoke_covered(self, fence: Fence, reason: str) -> tuple[str, ...]:
    """Close every live lease this fence covers. Caller holds ``lock``."""
    self._require_reachable()
    revoked: list[str] = []
    now = self._clock()
    for lease in tuple(self._leases.values()):
        if lease.live(now) and fence.covers(lease):
            revoked.extend(self._revoke_locked(lease.lease_id, reason))
    return tuple(dict.fromkeys(revoked))

IncompleteFenceCoverageError

Bases: ConfigurationError

A classified effect port has no reversibility verdict.

A ConfigurationError on purpose: the fence's coverage of T2.3.7's table is a build-time property, and a gap is a misconfiguration adopters already catch in that taxonomy.

LeaseAcquisitionError

Bases: EffectFenceError

EFX-L-1 — the conditional write that would grant a lease did not hold.

The bundle's epoch moved, or a fence already covers the requested generation or subject. Fail-closed: no lease, no effect.

LeaseInvalidError

Bases: EffectFenceError

A lease that existed is no longer usable: revoked, expired, or superseded.

EFX-L-3 and EFX-L-5 both land here, because the invocation's next effect attempt is denied identically in either case.

LeaseStoreUnavailableError

Bases: EffectFenceError

The lease backend could not be reached, so no fact could be established.

Not a denial of a known state — an inability to know one. The fence treats it as denial anyway (SEC-FCP-1) and records the degradation.

OpaqueExposureAccountant

OpaqueExposureAccountant(trust: ExtensionTrustRegistry)

Turns executed extensions into exposures and a withheld-or-not claim.

Source code in src/symfonic/services/effects/exposure.py
def __init__(self, trust: ExtensionTrustRegistry) -> None:
    self._trust = trust
    self._executions: dict[str, list[ExtensionExecution]] = {}
    #: Extensions this accountant itself demoted for a witnessed direct
    #: effect. Needed because the demotion is a side effect of the first
    #: assessment: without it, re-reading the same execution would find an
    #: opaque extension and lose the mis-declaration finding.
    self._misdeclared: set[str] = set()

account

account(invocation_ids: Iterable[str]) -> tuple[ExposureRecord, ...]

Every extension exposure for these invocations, demoting as it goes.

Source code in src/symfonic/services/effects/exposure.py
def account(self, invocation_ids: Iterable[str]) -> tuple[ExposureRecord, ...]:
    """Every extension exposure for these invocations, demoting as it goes."""
    records: list[ExposureRecord] = []
    for invocation_id in dict.fromkeys(invocation_ids):
        for execution in self.executions_for(invocation_id):
            finding = self._assess(execution)
            if finding is not None:
                kind, reason = finding
                records.append(
                    ExposureRecord(
                        invocation_id=invocation_id,
                        kind=kind,
                        subject=execution.extension_id,
                        reason=reason,
                    )
                )
    return tuple(records)

claim_for

claim_for(invocation_id: str) -> FencingClaim

Derived from live trust state, so a later demotion still withdraws it.

Source code in src/symfonic/services/effects/exposure.py
def claim_for(self, invocation_id: str) -> FencingClaim:
    """Derived from live trust state, so a later demotion still withdraws it."""
    for execution in self.executions_for(invocation_id):
        finding = self._assess(execution)
        if finding is not None:
            return FencingClaim(
                invocation_id=invocation_id,
                suppressed=False,
                withheld_reason=finding[1],
            )
    return FencingClaim(invocation_id=invocation_id, suppressed=True)

QuarantinedResult dataclass

QuarantinedResult(ticket_id: str, invocation_id: str, port_id: str, reason: str, quarantined_at: float)

One held result. The payload is deliberately not stored.

Keeping the payload would create a second copy of tenant data outside the invocation that earned it, on a path nobody has classified for retention. The record proves a result arrived and was withheld; that is what an incident needs.

ResultQuarantine

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

Holds late results. Nothing here ever releases one.

Source code in src/symfonic/services/effects/quarantine.py
def __init__(self, *, clock: Callable[[], float] = time.time) -> None:
    self._records: dict[str, QuarantinedResult] = {}
    self._clock = clock

deliverable

deliverable(ticket_id: str) -> bool

A quarantined result is never deliverable. There is no release path.

Source code in src/symfonic/services/effects/quarantine.py
def deliverable(self, ticket_id: str) -> bool:
    """A quarantined result is never deliverable. There is no release path."""
    return ticket_id not in self._records

RevertAccounting

RevertAccounting(*, accountant: OpaqueExposureAccountant, containment: ContainmentCoordinator, ports: FencedPortRegistry)

Turns the fenced set into exposures, claims, and blocked cutovers.

Source code in src/symfonic/services/effects/accounting.py
def __init__(
    self,
    *,
    accountant: OpaqueExposureAccountant,
    containment: ContainmentCoordinator,
    ports: FencedPortRegistry,
) -> None:
    self._accountant = accountant
    self._containment = containment
    self._ports = ports

blocked

blocked(exposures: tuple[ExposureRecord, ...]) -> tuple[str, ...]

Ports whose exposure has no approved rule, so cutover cannot proceed.

Source code in src/symfonic/services/effects/accounting.py
def blocked(self, exposures: tuple[ExposureRecord, ...]) -> tuple[str, ...]:
    """Ports whose exposure has no approved rule, so cutover cannot proceed."""
    return self._containment.registry.cutover_blocked_ports(
        e.subject for e in exposures if e.kind is ExposureKind.IRREVERSIBLE_EFFECT
    )

claims

claims(invocations: tuple[str, ...], exposures: tuple[ExposureRecord, ...]) -> tuple[FencingClaim, ...]

One claim per affected invocation, over both axes of exposure.

The accountant only knows extensions. An invocation that ran none but whose irreversible effect had already left the process is still exposed, and printing SUPPRESSED beside its own exposure record would be precisely the overstatement this package exists to prevent — so the port-crossing exposures override a clean extension verdict here.

Source code in src/symfonic/services/effects/accounting.py
def claims(
    self, invocations: tuple[str, ...], exposures: tuple[ExposureRecord, ...]
) -> tuple[FencingClaim, ...]:
    """One claim per affected invocation, over *both* axes of exposure.

    The accountant only knows extensions. An invocation that ran none but
    whose irreversible effect had already left the process is still
    exposed, and printing SUPPRESSED beside its own exposure record would
    be precisely the overstatement this package exists to prevent — so the
    port-crossing exposures override a clean extension verdict here.
    """
    exposed: dict[str, str] = {}
    for exposure in exposures:
        if exposure.kind is ExposureKind.IRREVERSIBLE_EFFECT:
            exposed.setdefault(exposure.invocation_id, exposure.reason)
    claims: list[FencingClaim] = []
    for invocation_id in invocations:
        claim = self._accountant.claim_for(invocation_id)
        reason = exposed.get(invocation_id)
        if claim.suppressed and reason is not None:
            claim = FencingClaim(
                invocation_id=invocation_id,
                suppressed=False,
                withheld_reason=(
                    f"an irreversible effect had already been dispatched: {reason}"
                ),
            )
        claims.append(claim)
    return tuple(claims)

contain_all

contain_all(exposures: tuple[ExposureRecord, ...]) -> tuple[CompensationOutcome, ...]

Run the approved compensation for each exposure, or record its absence.

Source code in src/symfonic/services/effects/accounting.py
def contain_all(
    self, exposures: tuple[ExposureRecord, ...]
) -> tuple[CompensationOutcome, ...]:
    """Run the approved compensation for each exposure, or record its absence."""
    return self._containment.contain_all(exposures)

exposures

exposures(dispatched: list[EffectTicket], invocations: tuple[str, ...]) -> tuple[ExposureRecord, ...]

Irreversible effects that already left, plus every opaque extension.

Source code in src/symfonic/services/effects/accounting.py
def exposures(
    self, dispatched: list[EffectTicket], invocations: tuple[str, ...]
) -> tuple[ExposureRecord, ...]:
    """Irreversible effects that already left, plus every opaque extension."""
    records = [
        ExposureRecord(
            invocation_id=ticket.invocation_id,
            kind=ExposureKind.IRREVERSIBLE_EFFECT,
            subject=ticket.port_id,
            reason=(
                f"{ticket.port_id} was dispatched before the fence closed; "
                f"{self._ports.port(ticket.port_id).rationale}, so cancellation "
                "did not undo it"
            ),
            ticket_id=ticket.ticket_id,
        )
        for ticket in dispatched
        if ticket.irreversible
    ]
    records.extend(self._accountant.account(invocations))
    return tuple(records)

invocations staticmethod

invocations(covered: tuple[EffectLease, ...], affected: tuple[EffectTicket, ...]) -> tuple[str, ...]

Every invocation the fence reached, whether or not it crossed a port.

Effect tickets are the wrong sole key here. An opaque in-process extension performs its effects without crossing a classified port, so an invocation can hold a fenced lease and own no ticket at all — exactly the invocation a revert must never leave out of the record. An invocation is accounted for because it was authorized under the rejected generation, not because it happened to cross a port the framework can see.

Source code in src/symfonic/services/effects/accounting.py
@staticmethod
def invocations(
    covered: tuple[EffectLease, ...], affected: tuple[EffectTicket, ...]
) -> tuple[str, ...]:
    """Every invocation the fence reached, whether or not it crossed a port.

    Effect tickets are the wrong sole key here. An opaque in-process
    extension performs its effects without crossing a classified port, so
    an invocation can hold a fenced lease and own no ticket at all —
    exactly the invocation a revert must never leave out of the record. An
    invocation is accounted for because it was *authorized* under the
    rejected generation, not because it happened to cross a port the
    framework can see.
    """
    return tuple(
        dict.fromkeys(
            [lease.invocation_id for lease in covered]
            + [ticket.invocation_id for ticket in affected]
        )
    )

SecurityRevertCoordinator

SecurityRevertCoordinator(*, store: InMemoryLeaseStore, fences: FenceLedger, tracker: InFlightTracker, drain: BoundedDrain, quarantine: ResultQuarantine, accountant: OpaqueExposureAccountant, containment: ContainmentCoordinator, ports: FencedPortRegistry, clock: Callable[[], float] = time.time)

Closes admissions, blocks commits, contains what it cannot stop.

Source code in src/symfonic/services/effects/revert.py
def __init__(
    self,
    *,
    store: InMemoryLeaseStore,
    fences: FenceLedger,
    tracker: InFlightTracker,
    drain: BoundedDrain,
    quarantine: ResultQuarantine,
    accountant: OpaqueExposureAccountant,
    containment: ContainmentCoordinator,
    ports: FencedPortRegistry,
    clock: Callable[[], float] = time.time,
) -> None:
    self._store = store
    self._fences = fences
    self._tracker = tracker
    self._drain = drain
    self._quarantine = quarantine
    self._books = RevertAccounting(
        accountant=accountant, containment=containment, ports=ports
    )
    self._clock = clock
    self.last_linearized_at = 0.0

refresh

refresh(incident: SecurityRevertIncident) -> SecurityRevertIncident

Re-read the suppression and quarantine ledgers for this incident.

A provider or tool result dispatched before the fence landed arrives after revert returns, so the frozen record cannot contain it — the late-result evidence would otherwise be structurally unreachable from the incident. This re-reads both ledgers, still scoped to the tickets this revert accounted for, and returns a new record with the same incident_id. Nothing else is recomputed: the fence, the drain, and the claims are findings about a moment, not a running total.

Source code in src/symfonic/services/effects/revert.py
def refresh(self, incident: SecurityRevertIncident) -> SecurityRevertIncident:
    """Re-read the suppression and quarantine ledgers for this incident.

    A provider or tool result dispatched before the fence landed arrives
    *after* ``revert`` returns, so the frozen record cannot contain it — the
    late-result evidence would otherwise be structurally unreachable from
    the incident. This re-reads both ledgers, still scoped to the tickets
    this revert accounted for, and returns a new record with the same
    ``incident_id``. Nothing else is recomputed: the fence, the drain, and
    the claims are findings about a moment, not a running total.
    """
    suppressed, quarantined = self._ledgers(incident.affected_tickets)
    return replace(
        incident, suppressed_commits=suppressed, quarantined=quarantined
    )

revert async

revert(*, bundle_id: str, reason: str, actor: str, rejected_vector_hash: str | None = None, epoch_ceiling: int | None = None, tenant_scope_hash: str | None = None) -> SecurityRevertIncident

Raise the fence, then account for everything it could not stop.

Source code in src/symfonic/services/effects/revert.py
async def revert(
    self,
    *,
    bundle_id: str,
    reason: str,
    actor: str,
    rejected_vector_hash: str | None = None,
    epoch_ceiling: int | None = None,
    tenant_scope_hash: str | None = None,
) -> SecurityRevertIncident:
    """Raise the fence, then account for everything it could not stop."""
    fence, revoked, covered, degraded_reason = await self._linearize(
        bundle_id=bundle_id,
        reason=reason,
        actor=actor,
        rejected_vector_hash=rejected_vector_hash,
        epoch_ceiling=epoch_ceiling,
        tenant_scope_hash=tenant_scope_hash,
    )
    linearized_at = self.last_linearized_at

    affected = self._tracker.tickets_for_leases(
        lease.lease_id for lease in covered
    )
    invocations = self._books.invocations(covered, affected)
    open_tickets = [t for t in affected if not self._state(t).terminal]
    self._tracker.request_cancellation(
        (t.ticket_id for t in open_tickets), f"security revert: {reason}"
    )

    cancelled = self._cancel_undispatched(open_tickets)
    dispatched = [t for t in affected if self._tracker.dispatched_at(t.ticket_id)]
    report = await self._drain.drain([t.ticket_id for t in dispatched])

    exposures = self._books.exposures(dispatched, invocations)
    compensations = self._books.contain_all(exposures)
    claims = self._books.claims(invocations, exposures)
    scope = tuple(t.ticket_id for t in affected)
    suppressed, quarantined = self._ledgers(scope)
    return SecurityRevertIncident(
        incident_id=f"inc-{uuid.uuid4().hex[:12]}",
        bundle_id=bundle_id,
        reason=reason,
        actor=actor,
        linearized_at=linearized_at,
        fence=fence,
        revoked_leases=revoked,
        stale_admissions=self._stale(revoked),
        cancelled=cancelled,
        affected_tickets=scope,
        suppressed_commits=suppressed,
        quarantined=quarantined,
        exposures=exposures,
        compensations=compensations,
        claims=claims,
        cutover_blocked=self._books.blocked(exposures),
        degraded=bool(degraded_reason),
        degraded_reason=degraded_reason,
        drain=report,
    )

SecurityRevertIncident dataclass

SecurityRevertIncident(incident_id: str, bundle_id: str, reason: str, actor: str, linearized_at: float, fence: Fence, revoked_leases: tuple[str, ...] = (), stale_admissions: tuple[str, ...] = (), cancelled: tuple[str, ...] = (), affected_tickets: tuple[str, ...] = (), suppressed_commits: tuple[str, ...] = (), quarantined: tuple[str, ...] = (), exposures: tuple[ExposureRecord, ...] = (), compensations: tuple[CompensationOutcome, ...] = (), claims: tuple[FencingClaim, ...] = (), cutover_blocked: tuple[str, ...] = (), degraded: bool = False, degraded_reason: str = '', drain: DrainReport = (lambda: DrainReport(drained=(), contained=(), budget=0.0, elapsed=0.0))())

Every leg of one security revert, in one reviewable record.

fencing_claim_withheld property

fencing_claim_withheld: bool

Derived. Exposure, a withheld claim, or degradation all take it true.

TicketState

Bases: StrEnum

Where one effect attempt got to. The drain reads these and only these.

UnapprovedCompensationError

Bases: ConfigurationError

An irreversible port has no tested, approved compensation rule.

Raised at registration for an untested rule and at lookup for a missing one. Either way the capability does not cut over.

UnclassifiedEffectPortError

Bases: EffectFenceError

SEC-FCP-5 precedent — an effect port carries no fence row.

An unclassified port is not "probably reversible". It is a hole in the evidence, and the operation that found it is denied.