symfonic.services.privacy

privacy

Subject privacy: the erasure fence, the coverage registry, and the guard.

Runtime-service layer (LAY-ADR §1.4), and deliberately so. Port-registry rows 14 (ErasureFence) and 20 (SubjectDataStore) are owned here rather than by platform because every storage adapter has to implement them, and SEC-TEN-5's rule — backends enforce isolation without importing platform — applies with even more force to erasure: a background worker must be able to ask "is this subject deleted?" without dragging the HTTP layer into the process.

What lives here:

• :class:InMemoryErasureFence — the atomic dual-condition commit (EFX-ER-3) and the tombstone-plus-generation transition (PRIV-4);
• :class:SubjectDataStoreRegistry — coverage as a registry (PRIV-1);
• :class:InMemorySubjectDataStore — the reference adapter (CON-S-4);
• :class:InMemoryErasureSagaStore — durable per-participant completion (PRIV-3);
• :class:SubjectGuard — the inward admission / read / write checks (PRIV-5);
• :func:run_subject_data_store_suite — CS-20, runnable against any adapter;
• :class:LegacyParticipantAdapter — the migration-window bridge for the stores already registered with the shipped erasure sweep (SEC-PRIV-5).

What does not live here: the saga that drives the registry to completion. That is a platform service (symfonic.platform.erasure), because deciding to erase a subject is an administrative act and this package is the machinery it acts through.

ErasureFence

Bases: Protocol

Registry row 14 — the erasure generation and the subject tombstone.

The write side is one method on purpose. Exposing "check" and "write" separately would make the time-of-check/time-of-use gap a supported usage of the port, and EFX-F-3 requires the gap to be closed by the atomic operation rather than by callers agreeing to be careful.

An implementation resolves a scope key to the nearest tombstone over its lineage, not to its exact row. Erasure is subtree-scoped, so a tenant tombstone governs every scope beneath it, and an exact-key implementation fails open for sub-scopes — the sub-scope write commits, and the subtree-scoped export reads the resurrected data straight back out. :func:symfonic.services.privacy.lineage.scope_key_lineage is the walk.

complete_erasure async

complete_erasure(scope_key: str) -> ErasureGeneration

Advance the generation again on completion; the tombstone stays.

Source code in src/symfonic/services/privacy/ports.py

async def complete_erasure(self, scope_key: str) -> ErasureGeneration:
    """Advance the generation again on completion; the tombstone stays."""
    ...

conditional_write async

conditional_write(scope_key: str, *, expected_generation: int, apply: Callable[[], Any]) -> WriteOutcome

Commit apply only if the generation matches AND no tombstone exists.

Both conditions and the write happen in one linearizable operation — a transaction with a predicate, a conditional update, or a single-lock section, per backend. Never as separate steps.

"No tombstone exists" means none over the scope's lineage: a write at tenant/team1 is refused by the tombstone published for tenant.

Source code in src/symfonic/services/privacy/ports.py

async def conditional_write(
    self,
    scope_key: str,
    *,
    expected_generation: int,
    apply: Callable[[], Any],
) -> WriteOutcome:
    """Commit ``apply`` only if the generation matches AND no tombstone exists.

    Both conditions and the write happen in one linearizable operation — a
    transaction with a predicate, a conditional update, or a single-lock
    section, per backend. Never as separate steps.

    "No tombstone exists" means none over the scope's lineage: a write at
    ``tenant/team1`` is refused by the tombstone published for ``tenant``.
    """
    ...

publish_tombstone async

publish_tombstone(scope_key: str, *, reason: str) -> ErasureGeneration

Plant the tombstone and advance the generation as ONE transition.

Source code in src/symfonic/services/privacy/ports.py

async def publish_tombstone(self, scope_key: str, *, reason: str) -> ErasureGeneration:
    """Plant the tombstone and advance the generation as ONE transition."""
    ...

read_generation async

read_generation(scope_key: str) -> ErasureGeneration

What a writer observes at admission (EFX-ER-2).

The returned scope_key names the scope the tombstone was published for, which may be an ancestor of the one asked about.

Source code in src/symfonic/services/privacy/ports.py

async def read_generation(self, scope_key: str) -> ErasureGeneration:
    """What a writer observes at admission (EFX-ER-2).

    The returned ``scope_key`` names the scope the tombstone was published
    for, which may be an ancestor of the one asked about.
    """
    ...

ErasureFenceDenied

ErasureFenceDenied(message: str, *, reason: str, scope_key: str, expected_generation: int | None = None, observed_generation: int | None = None)

Bases: PrivacyError

A conditional write was refused by the fence (EFX-ER-4).

Carries reason — generation_mismatch or subject_tombstoned — because the two are operationally different: the first says "somebody else moved", the second says "this subject is gone and retrying is resurrection".

Source code in src/symfonic/services/privacy/errors.py

def __init__(
    self,
    message: str,
    *,
    reason: str,
    scope_key: str,
    expected_generation: int | None = None,
    observed_generation: int | None = None,
) -> None:
    super().__init__(message)
    self.reason = reason
    self.scope_key = scope_key
    self.expected_generation = expected_generation
    self.observed_generation = observed_generation

ErasureGeneration dataclass

ErasureGeneration(scope_key: str, generation: int = 0, tombstoned: bool = False, tombstoned_at: float | None = None, reason: str = '')

EFX-ER-2 — what a writer observes before it commits.

The pair is read together and, at publication, advanced together: PRIV-4 makes the tombstone and the generation bump ONE linearizable transition, so there is no interval in which an observer can see a new generation without the tombstone that came with it.

ErasureReceipt dataclass

ErasureReceipt(participant_id: str, erased: int, generation: int, confirmed_absent: bool, detail: Mapping[str, Any] = dict())

What one store did, and whether absence was then verified.

erased is reporting (PRIV-6). confirmed_absent is the proof, and it is a separate field because a store that returns a confident count and still holds rows is precisely the failure SEC-PRIV-3 exists to catch.

ErasureSagaError

Bases: PrivacyError

The saga was asked about a subject it has no state for.

ErasureSagaStore

Bases: Protocol

The saga's durable per-participant completion state (PRIV-3).

Runtime-service-owned because the guard reads it — an in-flight invocation asks "is any store still unconfirmed?" before serving a read, and the guard may not import platform to find out.

claim_generation_advance async

claim_generation_advance(scope_key: str) -> bool

Compare-and-set: True exactly once per saga, for the caller that won.

There is no plain "mark it advanced" setter on this port, on purpose. Exactly-once completion cannot be assembled from a read plus a write — the await between them is where an operator retry and a scheduled resume both decide to advance, and the generation ends two ahead of the erasure that caused it. The store owns the transition because the store is the only thing both runs share.

Source code in src/symfonic/services/privacy/ports.py

async def claim_generation_advance(self, scope_key: str) -> bool:
    """Compare-and-set: True exactly once per saga, for the caller that won.

    There is no plain "mark it advanced" setter on this port, on purpose.
    Exactly-once completion cannot be assembled from a read plus a write —
    the await between them is where an operator retry and a scheduled resume
    both decide to advance, and the generation ends two ahead of the erasure
    that caused it. The store owns the transition because the store is the
    only thing both runs share.
    """
    ...

incomplete_under async

incomplete_under(scope_key: str) -> SagaState | None

Any saga at or below scope_key that still has a store unconfirmed.

The descendant half of read suppression, and it is not symmetry for its own sake. Reads match subtree-wide with narrows, so a reader at tenant sees the rows an in-flight erasure of tenant/team1 has not destroyed yet. Walking ancestors alone suppresses the readers at or below the erased scope and serves the one above it — the same rows, through a wider query.

Matching is segment-wise (SEC-TEN-1): acme-evil is not under acme, whatever a string comparison would say.

Source code in src/symfonic/services/privacy/ports.py

async def incomplete_under(self, scope_key: str) -> SagaState | None:
    """Any saga at or *below* ``scope_key`` that still has a store unconfirmed.

    The descendant half of read suppression, and it is not symmetry for its
    own sake. Reads match subtree-wide with ``narrows``, so a reader at
    ``tenant`` sees the rows an in-flight erasure of ``tenant/team1`` has
    not destroyed yet. Walking ancestors alone suppresses the readers at or
    below the erased scope and serves the one above it — the same rows,
    through a wider query.

    Matching is segment-wise (SEC-TEN-1): ``acme-evil`` is not under
    ``acme``, whatever a string comparison would say.
    """
    ...

start async

start(scope_key: str, *, participants: Sequence[str]) -> SagaState

Create or resume; never restart a saga that is already under way.

Source code in src/symfonic/services/privacy/ports.py

async def start(self, scope_key: str, *, participants: Sequence[str]) -> SagaState:
    """Create or resume; never restart a saga that is already under way."""
    ...

ExportFragment dataclass

ExportFragment(participant_id: str, records: tuple[Mapping[str, Any], ...] = (), exportable: bool = True, note: str = '')

One store's contribution to a subject export (SEC-PRIV-2).

FenceAuditBindingError

Bases: PrivacyError

Something tried to take over a fence's denial recorder (HOST-3).

Two hosts in one process — a test suite, a sidecar, an adopter mounting two agents — can share a fence. A silent rebind sends the first host's EFX-ER-4 denials to the second host's sink, which is worse than either host having no audit: the events exist, in the wrong place, and nobody is looking for them there. Rebinding is therefore an explicit act, not a constructor side effect.

FenceDenial dataclass

FenceDenial(scope_key: str, reason: DenialReason, expected_generation: int, observed_generation: int)

One refused write, kept so the audit seam has something to emit.

InMemoryErasureFence

InMemoryErasureFence(state: MutableMapping[str, ErasureGeneration] | None = None, *, on_denial: DenialRecorder | None = None, denial_buffer: int = _DENIAL_BUFFER)

Registry row 14, reference implementation.

state is injectable so a durable backend can own the bytes while this class owns the rule. A restart over the same mapping keeps refusing, which is what makes the tombstone persistent rather than a process fact.

Source code in src/symfonic/services/privacy/fence.py

def __init__(
    self,
    state: MutableMapping[str, ErasureGeneration] | None = None,
    *,
    on_denial: DenialRecorder | None = None,
    denial_buffer: int = _DENIAL_BUFFER,
) -> None:
    self._state: MutableMapping[str, ErasureGeneration] = (
        state if state is not None else {}
    )
    self._denials: deque[FenceDenial] = deque(maxlen=denial_buffer)
    self._on_denial = on_denial
    self._lock = threading.Lock()

has_denial_recorder property

has_denial_recorder: bool

Whether a host has already claimed this fence's denial stream.

complete_erasure async

complete_erasure(scope_key: str) -> ErasureGeneration

Advance once more, and keep the tombstone forever (PRIV-7).

Source code in src/symfonic/services/privacy/fence.py

async def complete_erasure(self, scope_key: str) -> ErasureGeneration:
    """Advance once more, and keep the tombstone forever (PRIV-7)."""
    with self._lock:
        governing = self._effective(scope_key)
        if not governing.tombstoned:
            raise ErasureFenceDenied(
                "cannot complete an erasure that was never published; the "
                "tombstone is what bars the writers this completion claims "
                "to have outlived",
                reason="generation_mismatch",
                scope_key=scope_key,
                observed_generation=governing.generation,
            )
        base = self._exact(scope_key)
        if base is None or not base.tombstoned:
            # An inherited tombstone: give this scope its own row so the
            # completion advance belongs to the scope that completed rather
            # than bumping an ancestor's generation on its behalf.
            base = replace(governing, scope_key=scope_key)
        advanced = replace(base, generation=base.generation + 1)
        self._state[scope_key] = advanced
        return advanced

conditional_write async

conditional_write(scope_key: str, *, expected_generation: int, apply: Callable[[], Any]) -> WriteOutcome

The atomic dual-condition commit (EFX-ER-3).

apply must be synchronous. An awaitable would suspend inside the very section whose indivisibility is the contract, and "atomic except while it awaits" is not atomic.

Source code in src/symfonic/services/privacy/fence.py

async def conditional_write(
    self,
    scope_key: str,
    *,
    expected_generation: int,
    apply: Callable[[], Any],
) -> WriteOutcome:
    """The atomic dual-condition commit (EFX-ER-3).

    ``apply`` must be synchronous. An awaitable would suspend inside the
    very section whose indivisibility is the contract, and "atomic except
    while it awaits" is not atomic.
    """
    if inspect.iscoroutinefunction(apply):
        raise TypeError(
            "conditional_write applies its effect inside the linearizable "
            "section, so `apply` must be synchronous; an awaited effect "
            "would reopen the time-of-check/time-of-use window this port "
            "exists to close"
        )
    with self._lock:
        current = self._effective(scope_key)
        if current.tombstoned:
            denial = self._record(
                scope_key, "subject_tombstoned", expected_generation, current.generation
            )
        elif current.generation != expected_generation:
            denial = self._record(
                scope_key, "generation_mismatch", expected_generation, current.generation
            )
        else:
            result = apply()
            if inspect.isawaitable(result):
                raise TypeError(
                    "`apply` returned an awaitable; see the synchronous-effect "
                    "rule above — the effect has already run and cannot be undone, "
                    "so fix the caller rather than awaiting here"
                )
            return WriteOutcome(
                committed=True, generation=current.generation, result=result
            )

    # Outside the section: the seam is somebody else's I/O and the decision
    # is already made. Emitting under the lock would let a slow audit sink
    # serialize every writer in the process.
    await self._emit(denial)
    return WriteOutcome(
        committed=False,
        generation=denial.observed_generation,
        denial_reason=denial.reason,
    )

denials

denials() -> tuple[FenceDenial, ...]

The retained denial window (bounded; see :data:_DENIAL_BUFFER).

Source code in src/symfonic/services/privacy/fence.py

def denials(self) -> tuple[FenceDenial, ...]:
    """The retained denial window (bounded; see :data:`_DENIAL_BUFFER`)."""
    return tuple(self._denials)

drain_denials

drain_denials() -> tuple[FenceDenial, ...]

Take the window and clear it, for a host that batches its own emit.

Source code in src/symfonic/services/privacy/fence.py

def drain_denials(self) -> tuple[FenceDenial, ...]:
    """Take the window and clear it, for a host that batches its own emit."""
    with self._lock:
        drained = tuple(self._denials)
        self._denials.clear()
    return drained

publish_tombstone async

publish_tombstone(scope_key: str, *, reason: str) -> ErasureGeneration

One transition: tombstone planted, generation advanced (PRIV-4).

Idempotent, because a resumed saga re-publishes and a second advance would invalidate every generation an in-flight reader holds for no reason at all.

Source code in src/symfonic/services/privacy/fence.py

async def publish_tombstone(self, scope_key: str, *, reason: str) -> ErasureGeneration:
    """One transition: tombstone planted, generation advanced (PRIV-4).

    Idempotent, because a resumed saga re-publishes and a second advance
    would invalidate every generation an in-flight reader holds for no
    reason at all.
    """
    if not reason:
        raise ErasureFenceDenied(
            "an erasure needs a reason; an unexplained tombstone is not "
            "reviewable evidence",
            reason="generation_mismatch",
            scope_key=scope_key,
        )
    with self._lock:
        current = self._exact(scope_key) or ErasureGeneration(scope_key=scope_key)
        if current.tombstoned:
            return current
        published = ErasureGeneration(
            scope_key=scope_key,
            generation=current.generation + 1,
            tombstoned=True,
            tombstoned_at=time.time(),
            reason=reason,
        )
        self._state[scope_key] = published
        return published

set_denial_recorder

set_denial_recorder(recorder: DenialRecorder | None, *, replace: bool = False) -> None

Late-bind the audit seam (EFX-ER-4). Once, unless told otherwise.

A fence is usually built before the service that audits it — the host wires storage first — so the recorder is attachable rather than constructor-only. What it is not is silently replaceable: two hosts in one process can share a fence, and a second attachment would redirect the first host's denials into the second host's sink without either of them saying so. replace=True is how a host that means it says so.

Source code in src/symfonic/services/privacy/fence.py

def set_denial_recorder(
    self, recorder: DenialRecorder | None, *, replace: bool = False
) -> None:
    """Late-bind the audit seam (EFX-ER-4). Once, unless told otherwise.

    A fence is usually built before the service that audits it — the host
    wires storage first — so the recorder is attachable rather than
    constructor-only. What it is *not* is silently replaceable: two hosts in
    one process can share a fence, and a second attachment would redirect
    the first host's denials into the second host's sink without either of
    them saying so. ``replace=True`` is how a host that means it says so.
    """
    if recorder is not None and self._on_denial is not None and not replace:
        raise FenceAuditBindingError(
            "this fence already has a denial recorder; attaching a second "
            "one would send the first host's EFX-ER-4 events to the second "
            "host's audit sink. Pass replace=True if that is the intent, or "
            "leave the binding to the host that made it"
        )
    self._on_denial = recorder

InMemoryErasureSagaStore

InMemoryErasureSagaStore(state: MutableMapping[str, SagaState] | None = None)

Reference implementation (CON-S-4); state is injectable for durability.

The lock is a thread lock, not a per-loop asyncio.Lock: an operator retry and a scheduled resume can land in different loops (or threads), and :meth:claim_generation_advance is only a compare-and-set if the compare and the set are indivisible for every caller, not just for co-scheduled ones. Nothing awaits inside a section, so a plain lock cannot deadlock.

Source code in src/symfonic/services/privacy/saga_store.py

def __init__(self, state: MutableMapping[str, SagaState] | None = None) -> None:
    self._state: MutableMapping[str, SagaState] = state if state is not None else {}
    self._lock = threading.Lock()

claim_generation_advance async

claim_generation_advance(scope_key: str) -> bool

Compare-and-set the completion transition; True for the winner only.

This is the only way to make the transition — there is deliberately no plain "mark it advanced" setter to reach for. The saga cannot do it as "read the flag, then advance": an operator retry racing a scheduled resume — a race PrivacyService.resume explicitly invites — would have both runs observe False across the await and both call complete_erasure, advancing the generation twice. The decision has to be one indivisible step in the store, which is the only place both runs meet.

Source code in src/symfonic/services/privacy/saga_store.py

async def claim_generation_advance(self, scope_key: str) -> bool:
    """Compare-and-set the completion transition; True for the winner only.

    This is the *only* way to make the transition — there is deliberately no
    plain "mark it advanced" setter to reach for. The saga cannot do it as
    "read the flag, then advance": an operator retry racing a scheduled
    ``resume`` — a race ``PrivacyService.resume`` explicitly invites — would
    have both runs observe ``False`` across the await and both call
    ``complete_erasure``, advancing the generation twice. The decision has to
    be one indivisible step *in the store*, which is the only place both runs
    meet.
    """
    with self._lock:
        state = self._require(scope_key)
        if state.generation_advanced:
            return False
        self._state[scope_key] = replace(state, generation_advanced=True)
        return True

incomplete_under async

incomplete_under(scope_key: str) -> SagaState | None

The descendant half of read suppression (see the port's docstring).

A scan, because the reference store is a mapping: a durable backend indexes the tenant segment and answers the same question with a query. Keys that are not scope keys are compared exactly rather than skipped — an unparseable key still names a saga somebody started, and dropping it would fail open for whoever recorded it.

Source code in src/symfonic/services/privacy/saga_store.py

async def incomplete_under(self, scope_key: str) -> SagaState | None:
    """The descendant half of read suppression (see the port's docstring).

    A scan, because the reference store is a mapping: a durable backend
    indexes the tenant segment and answers the same question with a query.
    Keys that are not scope keys are compared exactly rather than skipped —
    an unparseable key still names a saga somebody started, and dropping it
    would fail *open* for whoever recorded it.
    """
    try:
        reader = SubjectScope.from_key(scope_key)
    except SubjectScopeError:
        state = await self.read(scope_key)
        return state if state is not None and not state.complete else None
    with self._lock:
        for key in sorted(self._state):
            state = self._state[key]
            if state.complete:
                continue
            try:
                scope = SubjectScope.from_key(key)
            except SubjectScopeError:
                continue
            if scope.narrows(reader):
                return state
    return None

mark_exhausted async

mark_exhausted(scope_key: str, participant_id: str, *, error: str) -> SagaState

Bounded retry ran out. Recorded, never raised.

Raising here would abort the sweep and skip every participant sorted after the broken one — the exact failure mode the shipped legacy sweep already learned to avoid.

Source code in src/symfonic/services/privacy/saga_store.py

async def mark_exhausted(
    self, scope_key: str, participant_id: str, *, error: str
) -> SagaState:
    """Bounded retry ran out. Recorded, never raised.

    Raising here would abort the sweep and skip every participant sorted
    after the broken one — the exact failure mode the shipped legacy sweep
    already learned to avoid.
    """
    return await self._update(
        scope_key,
        participant_id,
        lambda p: replace(p, exhausted=True, last_error=error or p.last_error),
    )

start async

start(scope_key: str, *, participants: Sequence[str]) -> SagaState

Create, or resume and widen.

Widen, because PRIV-1's registry is live: a store registered while the saga was running still holds the subject's data, and a saga that captured its participant list at t=0 would leave it holding it.

Source code in src/symfonic/services/privacy/saga_store.py

async def start(self, scope_key: str, *, participants: Sequence[str]) -> SagaState:
    """Create, or resume and widen.

    Widen, because PRIV-1's registry is live: a store registered while the
    saga was running still holds the subject's data, and a saga that
    captured its participant list at t=0 would leave it holding it.
    """
    with self._lock:
        state = self._state.get(scope_key)
        if state is None:
            state = SagaState(
                scope_key=scope_key,
                started_at=time.time(),
                progress={
                    pid: ParticipantProgress(participant_id=pid) for pid in participants
                },
            )
        else:
            progress = dict(state.progress)
            for pid in participants:
                if pid not in progress:
                    progress[pid] = ParticipantProgress(participant_id=pid)
            state = replace(state, progress=progress)
        self._state[scope_key] = state
        return state

InMemorySubjectDataStore

InMemorySubjectDataStore(participant_id: str, *, holds: str = 'records', replica_of: str | None = None, backup_retention_days: int | None = None)

A dictionary of rows keyed by scope key, behind the row-20 port.

Source code in src/symfonic/services/privacy/reference.py

def __init__(
    self,
    participant_id: str,
    *,
    holds: str = "records",
    replica_of: str | None = None,
    backup_retention_days: int | None = None,
) -> None:
    self.participant_id = participant_id
    self._holds = holds
    self._replica_of = replica_of
    self._backup_retention_days = backup_retention_days
    self._rows: dict[str, list[Mapping[str, Any]]] = {}

put

put(scope: SubjectScope, record: Mapping[str, Any]) -> None

The raw effect. Only ever called inside a conditional write.

Source code in src/symfonic/services/privacy/reference.py

def put(self, scope: SubjectScope, record: Mapping[str, Any]) -> None:
    """The raw effect. Only ever called *inside* a conditional write."""
    self._rows.setdefault(scope.scope_key, []).append(dict(record))

write async

write(scope: SubjectScope, record: Mapping[str, Any], *, fence: ErasureFence, expected_generation: int) -> WriteOutcome

EFX-ER-3: the store never commits except through the fence.

Source code in src/symfonic/services/privacy/reference.py

async def write(
    self,
    scope: SubjectScope,
    record: Mapping[str, Any],
    *,
    fence: ErasureFence,
    expected_generation: int,
) -> WriteOutcome:
    """EFX-ER-3: the store never commits except through the fence."""
    return await fence.conditional_write(
        scope.scope_key,
        expected_generation=expected_generation,
        apply=lambda: self.put(scope, record),
    )

LegacyParticipantAdapter

LegacyParticipantAdapter(participant: _LegacyParticipant)

A TenantErasureParticipant seen through the row-20 port.

The legacy contract is keyed by tenant_id alone, so the adapter passes the subject's tenant segment and documents the consequence: erasing a sub-tenant through a legacy participant erases the whole tenant's rows in that store. That is the conservative direction — over-erasure of the caller's own subtree, never under-erasure — and it is recorded here rather than discovered later.

Source code in src/symfonic/services/privacy/legacy.py

def __init__(self, participant: _LegacyParticipant) -> None:
    self._participant = participant

ParticipantDescriptor dataclass

ParticipantDescriptor(participant_id: str, holds: str, durable: bool = True, replica_of: str | None = None, backup_retention_days: int | None = None, exportable: bool = True)

PRIV-1/PRIV-7 — what a store declares about the subject data it holds.

replica_of and backup_retention_days exist so PRIV-7 is checkable rather than aspirational: a read replica names its primary, and a backup set names the horizon the tombstone must outlive for a restore to re-apply the erasure before the data is served again.

ParticipantProgress dataclass

ParticipantProgress(participant_id: str, confirmed: bool = False, exhausted: bool = False, attempts: int = 0, erased: int = 0, last_error: str = '')

One store's place in the saga.

confirmed means absence was verified (SEC-PRIV-3), not that an erase call returned. exhausted means the bounded retry ran out — recorded rather than raised, because one unreachable backend must not abort the other eight.

ParticipantRegistrationError

Bases: PrivacyError

A store is missing from, or duplicated in, the coverage registry.

PRIV-1 makes coverage checkable by making it declared. A silent overwrite would drop a store out of the erasure path with no diagnostic at all.

PrivacyError

Bases: SymfonicError

Root of the privacy taxonomy.

ReadSuppressedError

ReadSuppressedError(message: str, *, scope_key: str, unconfirmed: tuple[str, ...])

Bases: PrivacyError

An erasure is in flight and at least one store is unconfirmed.

Distinct from :class:SubjectErasedError on purpose: this one is a transient fail-closed state an operator can act on by driving the saga to completion, and it maps to a different status than "this subject is gone".

Source code in src/symfonic/services/privacy/errors.py

def __init__(self, message: str, *, scope_key: str, unconfirmed: tuple[str, ...]) -> None:
    super().__init__(message)
    self.scope_key = scope_key
    self.unconfirmed = unconfirmed

SagaState dataclass

SagaState(scope_key: str, started_at: float, progress: dict[str, ParticipantProgress] = dict(), generation_advanced: bool = False)

Everything a resumed saga needs, and nothing a transport would add.

exhausted

exhausted() -> tuple[str, ...]

Participants whose bounded retry ran out.

Source code in src/symfonic/services/privacy/saga_values.py

def exhausted(self) -> tuple[str, ...]:
    """Participants whose bounded retry ran out."""
    return tuple(
        pid for pid in sorted(self.progress) if self.progress[pid].exhausted
    )

pending

pending() -> tuple[str, ...]

Participants that still owe a confirmed absence.

Source code in src/symfonic/services/privacy/saga_values.py

def pending(self) -> tuple[str, ...]:
    """Participants that still owe a confirmed absence."""
    return tuple(
        pid
        for pid in sorted(self.progress)
        if not self.progress[pid].confirmed and not self.progress[pid].exhausted
    )

unconfirmed

unconfirmed() -> tuple[str, ...]

Everything not yet proven absent — the read-suppression set.

Source code in src/symfonic/services/privacy/saga_values.py

def unconfirmed(self) -> tuple[str, ...]:
    """Everything not yet proven absent — the read-suppression set."""
    return tuple(
        pid for pid in sorted(self.progress) if not self.progress[pid].confirmed
    )

SubjectDataStore

Bases: Protocol

Registry row 20 — one store's side of export and erasure.

confirm_absent is what turns SEC-PRIV-3's verification-by-absence from a claim into a call. A store that only reports how many rows it deleted cannot answer the question an auditor actually asks.

confirm_absent async

confirm_absent(scope: SubjectScope) -> bool

Prove it: is the subject's subtree gone from this store?

Source code in src/symfonic/services/privacy/ports.py

async def confirm_absent(self, scope: SubjectScope) -> bool:
    """Prove it: is the subject's subtree gone from this store?"""
    ...

describe

describe() -> ParticipantDescriptor

Declare what subject data this store holds (PRIV-1/PRIV-7).

Source code in src/symfonic/services/privacy/ports.py

def describe(self) -> ParticipantDescriptor:
    """Declare what subject data this store holds (PRIV-1/PRIV-7)."""
    ...

erase_subject async

erase_subject(scope: SubjectScope, fence: ErasureFence) -> ErasureReceipt

Erase the subject's subtree under the fence.

Source code in src/symfonic/services/privacy/ports.py

async def erase_subject(self, scope: SubjectScope, fence: ErasureFence) -> ErasureReceipt:
    """Erase the subject's subtree under the fence."""
    ...

export_subject async

export_subject(scope: SubjectScope) -> ExportFragment

Everything this store holds for the subject's subtree (SEC-PRIV-2).

Source code in src/symfonic/services/privacy/ports.py

async def export_subject(self, scope: SubjectScope) -> ExportFragment:
    """Everything this store holds for the subject's subtree (SEC-PRIV-2)."""
    ...

SubjectDataStoreRegistry

SubjectDataStoreRegistry(registry_id: str = 'default')

The set of stores the erasure saga enumerates.

An instance, not a module global: HOST-3 forbids process-global mutable state for per-deployment facts, and two hosts in one process (a test suite, a sidecar, an adopter mounting two agents) must not share an erasure surface neither of them declared.

Source code in src/symfonic/services/privacy/registry.py

def __init__(self, registry_id: str = "default") -> None:
    self._registry_id = registry_id
    self._stores: dict[str, SubjectDataStore] = {}

participants

participants() -> tuple[SubjectDataStore, ...]

Stable order, so a resumed saga sweeps in the same sequence.

Source code in src/symfonic/services/privacy/registry.py

def participants(self) -> tuple[SubjectDataStore, ...]:
    """Stable order, so a resumed saga sweeps in the same sequence."""
    return tuple(self._stores[key] for key in sorted(self._stores))

register

register(store: SubjectDataStore) -> None

Add a participant. A duplicate id is an error, not an overwrite.

Silently replacing would drop a live store out of the erasure path and leave a registry that looks complete — the worst of both.

Source code in src/symfonic/services/privacy/registry.py

def register(self, store: SubjectDataStore) -> None:
    """Add a participant. A duplicate id is an error, not an overwrite.

    Silently replacing would drop a live store out of the erasure path and
    leave a registry that *looks* complete — the worst of both.
    """
    participant_id = store.describe().participant_id
    if not participant_id:
        raise ParticipantRegistrationError(
            "a subject-data store must declare a participant id; an unnamed "
            "participant cannot be reported as covered or missing"
        )
    if participant_id in self._stores:
        raise ParticipantRegistrationError(
            f"{participant_id!r} is already registered with the "
            f"{self._registry_id!r} erasure registry; overwriting it would "
            "silently remove a store from the erasure path"
        )
    self._stores[participant_id] = store

SubjectDataStoreWriter

Bases: Protocol

Row 20's write obligation, for the stores that accept writes.

Separate from :class:SubjectDataStore rather than folded into it, because not every participant has a write path — a legacy erase-and-count adapter implements the four methods above and nothing else, and demanding a writer of it would misreport a complete participant as a broken one.

Where it does exist, EFX-ER-3 says what it must be: the commit goes through :meth:ErasureFence.conditional_write, never straight to the backend. CS-20 asks for this entry point by name, so a store that owns subject rows and does not declare it fails the suite's first write case instead of crashing it.

write async

write(scope: SubjectScope, record: Any, *, fence: ErasureFence, expected_generation: int) -> WriteOutcome

Commit record for scope through the fence, or not at all.

Source code in src/symfonic/services/privacy/ports.py

async def write(
    self,
    scope: SubjectScope,
    record: Any,
    *,
    fence: ErasureFence,
    expected_generation: int,
) -> WriteOutcome:
    """Commit ``record`` for ``scope`` through the fence, or not at all."""
    ...

SubjectErasedError

SubjectErasedError(message: str, *, scope_key: str)

Bases: PrivacyError

The subject carries a tombstone: no admission, no read, no write.

Terminal by design. There is no "the erasure finished so it is safe again" branch, because that branch is exactly how a restored backup serves data the subject asked to have destroyed (PRIV-7).

Source code in src/symfonic/services/privacy/errors.py

def __init__(self, message: str, *, scope_key: str) -> None:
    super().__init__(message)
    self.scope_key = scope_key

SubjectGuard

SubjectGuard(*, fence: ErasureFence, saga_store: ErasureSagaStore | None = None)

Constructor-injected, narrow, and holding no state of its own.

Source code in src/symfonic/services/privacy/guard.py

def __init__(self, *, fence: ErasureFence, saga_store: ErasureSagaStore | None = None) -> None:
    self._fence = fence
    self._saga_store = saga_store

guarded_write async

guarded_write(scope: SubjectScope, *, apply: Callable[[], Any]) -> WriteOutcome

Observe the generation, then commit under both conditions.

Reading the generation first is not a check-then-write: the read is EFX-ER-2's admission observation, and the decision is made inside the conditional write against whatever is true at commit time. A writer that loses the race is denied there, not here.

A denied write is never retried. EFX-ER-4 is unambiguous: retrying against the new generation is resurrection, and a caller that wants to proceed must re-derive from post-erasure sources instead.

Source code in src/symfonic/services/privacy/guard.py

async def guarded_write(
    self, scope: SubjectScope, *, apply: Callable[[], Any]
) -> WriteOutcome:
    """Observe the generation, then commit under both conditions.

    Reading the generation first is not a check-then-write: the read is
    EFX-ER-2's admission observation, and the *decision* is made inside the
    conditional write against whatever is true at commit time. A writer that
    loses the race is denied there, not here.

    A denied write is never retried. EFX-ER-4 is unambiguous: retrying
    against the new generation is resurrection, and a caller that wants to
    proceed must re-derive from post-erasure sources instead.
    """
    generation = await self._fence.read_generation(scope.scope_key)
    return await self._fence.conditional_write(
        scope.scope_key,
        expected_generation=generation.generation,
        apply=apply,
    )

require_admission async

require_admission(scope: SubjectScope) -> None

SCOPE-14 step 2. Fail closed: a tombstoned subject does not run.

Source code in src/symfonic/services/privacy/guard.py

async def require_admission(self, scope: SubjectScope) -> None:
    """SCOPE-14 step 2. Fail closed: a tombstoned subject does not run."""
    tombstone = await self._tombstone(scope)
    if tombstone is not None:
        raise SubjectErasedError(
            f"subject {scope.scope_key!r} was erased"
            f"{_under(tombstone, scope)}; admitting an invocation for it "
            "is how deleted data comes back",
            scope_key=scope.scope_key,
        )

require_readable async

require_readable(scope: SubjectScope) -> None

Reads: suppressed while any store is unconfirmed, refused once erased.

Both outcomes deny; the distinction is what an operator can do about it. An incomplete saga is a transient state somebody can clear by driving the remaining stores to confirmed absence, so it reports which stores are outstanding. A completed erasure is terminal — finishing the saga does not reopen the subject, it closes it — so the tombstone check is the fallthrough rather than the first gate.

Source code in src/symfonic/services/privacy/guard.py

async def require_readable(self, scope: SubjectScope) -> None:
    """Reads: suppressed while any store is unconfirmed, refused once erased.

    Both outcomes deny; the distinction is what an operator can do about it.
    An *incomplete* saga is a transient state somebody can clear by driving
    the remaining stores to confirmed absence, so it reports which stores
    are outstanding. A completed erasure is terminal — finishing the saga
    does not reopen the subject, it closes it — so the tombstone check is
    the fallthrough rather than the first gate.
    """
    state = await self._incomplete_saga(scope)
    if state is not None:
        raise ReadSuppressedError(
            f"an erasure for {state.scope_key!r} has stores that have not "
            "confirmed absence; serving a read now could return data the "
            "subject asked to have destroyed",
            scope_key=scope.scope_key,
            unconfirmed=state.unconfirmed(),
        )
    tombstone = await self._tombstone(scope)
    if tombstone is not None:
        raise SubjectErasedError(
            f"subject {scope.scope_key!r} was erased"
            f"{_under(tombstone, scope)} and is not readable",
            scope_key=scope.scope_key,
        )

unconfirmed_stores async

unconfirmed_stores(scope: SubjectScope) -> tuple[str, ...]

Diagnostics: which participants still owe a confirmed absence.

Source code in src/symfonic/services/privacy/guard.py

async def unconfirmed_stores(self, scope: SubjectScope) -> tuple[str, ...]:
    """Diagnostics: which participants still owe a confirmed absence."""
    state = await self._incomplete_saga(scope)
    return () if state is None else state.unconfirmed()

WriteOutcome dataclass

WriteOutcome(committed: bool, generation: int, denial_reason: DenialReason | None = None, result: Any = None)

The result of the atomic dual-condition commit (EFX-ER-3).

committed is the whole contract. A caller that wants to know why it failed reads denial_reason; a caller that retries on subject_tombstoned is performing resurrection and EFX-ER-4 forbids it.

adapt_legacy_registry

adapt_legacy_registry(legacy: object, *, into: SubjectDataStoreRegistry) -> tuple[str, ...]

Register every legacy participant with into; return the ids added.

Takes the registry structurally (participants()) so this module does not import the shadow package — the erasure path must not acquire a dependency on the cutover machinery in order to sweep it.

Source code in src/symfonic/services/privacy/legacy.py

def adapt_legacy_registry(legacy: object, *, into: SubjectDataStoreRegistry) -> tuple[str, ...]:
    """Register every legacy participant with ``into``; return the ids added.

    Takes the registry structurally (``participants()``) so this module does not
    import the shadow package — the erasure path must not acquire a dependency
    on the cutover machinery in order to sweep it.
    """
    added: list[str] = []
    participants = getattr(legacy, "participants", None)
    if not callable(participants):
        return ()
    for participant in participants():
        adapter = LegacyParticipantAdapter(participant)
        participant_id = adapter.describe().participant_id
        if into.holds(participant_id):
            continue
        into.register(adapter)
        added.append(participant_id)
    return tuple(added)

run_subject_data_store_suite async

run_subject_data_store_suite(store_factory: Callable[[], SubjectDataStore], *, fence_factory: Callable[[], ErasureFence]) -> SuiteReport

Run every CS-20 case against a freshly built store, and report.

A fresh store per case: sharing one would let an earlier case's leftovers decide a later one, and a suite whose verdict depends on execution order is not evidence.

Source code in src/symfonic/services/privacy/suite.py

async def run_subject_data_store_suite(
    store_factory: Callable[[], SubjectDataStore],
    *,
    fence_factory: Callable[[], ErasureFence],
) -> SuiteReport:
    """Run every CS-20 case against a freshly built store, and report.

    A fresh store per case: sharing one would let an earlier case's leftovers
    decide a later one, and a suite whose verdict depends on execution order is
    not evidence.
    """
    cases: list[SuiteCase] = []
    participant_id = store_factory().describe().participant_id

    cases.append(
        SuiteCase("describe-declares-a-participant-id", bool(participant_id))
    )

    # Every case after this one commits through the store's conditional-write
    # entry point (EFX-ER-3). A store that does not declare one is *reported*
    # here rather than crashing the suite three cases later: an integrator who
    # implemented the published port and got an AttributeError learns nothing
    # about which obligation they missed.
    if not isinstance(store_factory(), SubjectDataStoreWriter):
        cases.append(
            SuiteCase(
                "declares-a-conditional-write-entry-point",
                False,
                "no `write(scope, record, *, fence, expected_generation)`; the "
                "remaining cases commit through it and were not run",
            )
        )
        return SuiteReport(
            participant_id=participant_id, subject=_SUBJECT, cases=tuple(cases)
        )
    cases.append(SuiteCase("declares-a-conditional-write-entry-point", True))

    store, fence = store_factory(), fence_factory()
    outcome = await _write(store, fence, _SUBJECT, "one")
    cases.append(
        SuiteCase("write-commits-at-the-current-generation", bool(outcome.committed))
    )

    store, fence = store_factory(), fence_factory()
    await _write(store, fence, _SUBJECT, "mine")
    await _write(store, fence, _CHILD, "child")
    await _write(store, fence, _NEIGHBOUR, "not-mine")
    fragment = await store.export_subject(_SUBJECT)
    bodies = {record.get("body") for record in fragment.records}
    cases.append(
        SuiteCase(
            "export-is-subtree-scoped",
            not fragment.exportable or bodies == {"mine", "child"},
            f"exported {sorted(str(b) for b in bodies)}",
        )
    )

    store, fence = store_factory(), fence_factory()
    await _write(store, fence, _SUBJECT, "one")
    await fence.publish_tombstone(_SUBJECT.scope_key, reason="cs-20")
    receipt = await store.erase_subject(_SUBJECT, fence)
    absent = await store.confirm_absent(_SUBJECT)
    cases.append(
        SuiteCase(
            "erase-then-confirm-absent",
            bool(receipt.confirmed_absent and absent),
            f"erased={receipt.erased}",
        )
    )

    store, fence = store_factory(), fence_factory()
    await fence.publish_tombstone(_SUBJECT.scope_key, reason="cs-20")
    denied = await _write(store, fence, _SUBJECT, "after")
    cases.append(
        SuiteCase(
            "write-denied-after-tombstone",
            not denied.committed and await store.confirm_absent(_SUBJECT),
        )
    )

    # The sub-scope version of the same case, and the one an adapter is most
    # likely to fail: erasure is subtree-scoped, so the tenant tombstone has to
    # refuse a write at the child too. A store (or a fence) that matches the
    # exact key lets the child write commit, and the subtree-scoped export then
    # returns the subject that was just erased.
    store, fence = store_factory(), fence_factory()
    await fence.publish_tombstone(_SUBJECT.scope_key, reason="cs-20")
    denied_child = await _write(store, fence, _CHILD, "after-child")
    exported = await store.export_subject(_SUBJECT)
    cases.append(
        SuiteCase(
            "write-denied-after-ancestor-tombstone",
            not denied_child.committed
            and not exported.records
            and await store.confirm_absent(_SUBJECT),
            f"child={_CHILD.scope_key} tombstone={_SUBJECT.scope_key}",
        )
    )

    cases.append(await _concurrency_case(store_factory, fence_factory))
    return SuiteReport(
        participant_id=participant_id, subject=_SUBJECT, cases=tuple(cases)
    )

scope_key_lineage

scope_key_lineage(scope_key: str) -> tuple[str, ...]

:func:scope_lineage for a raw key.

A key that is not a scope key gets a one-element lineage rather than an exception: the fence is a safety control, and refusing to answer "is this tombstoned?" for an odd key would fail open at every call site that only wanted a boolean.

Source code in src/symfonic/services/privacy/lineage.py

def scope_key_lineage(scope_key: str) -> tuple[str, ...]:
    """:func:`scope_lineage` for a raw key.

    A key that is not a scope key gets a one-element lineage rather than an
    exception: the fence is a safety control, and refusing to answer "is this
    tombstoned?" for an odd key would fail *open* at every call site that only
    wanted a boolean.
    """
    try:
        scope = SubjectScope.from_key(scope_key)
    except SubjectScopeError:
        return (scope_key,)
    return scope_lineage(scope)

scope_lineage

scope_lineage(scope: SubjectScope) -> tuple[str, ...]

scope's own key, then every ancestor key, most specific first.

Source code in src/symfonic/services/privacy/lineage.py

def scope_lineage(scope: SubjectScope) -> tuple[str, ...]:
    """``scope``'s own key, then every ancestor key, most specific first."""
    keys: list[str] = [scope.scope_key]
    segments = scope.segments
    for depth in range(len(segments), 0, -1):
        for key in _spellings(segments[:depth]):
            if key not in keys:
                keys.append(key)
    return tuple(keys)