symfonic.services.shadow.privacy

privacy

SEC-PRIV-5 tenant erasure contract and legacy deletion wiring.

The process-global participant registry remains for the live legacy fallback. Native hosts explicitly register their recording store with the injected SubjectDataStoreRegistry via shadow.native_privacy. Neither path implicitly imports participants from the other. Native registration is synthetic-only until its synchronous writer can commit through the native erasure fence. Legacy production attestation still requires a seeded erase round trip and an absence check.

DeletionWiringAttestation dataclass

DeletionWiringAttestation(participant_id: str, registry_id: str, verified_at: datetime, probe_tenant: str, seeded: int, residual: int)

Evidence that erasure actually reaches this participant.

ErasureParticipantRegistry

ErasureParticipantRegistry(registry_id: str = 'default')

The set of stores the tenant-erasure path must sweep.

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

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

erase_all async

erase_all(tenant_id: str) -> dict[str, int]

Sweep every participant. Returns participant_id -> rows removed.

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

async def erase_all(self, tenant_id: str) -> dict[str, int]:
    """Sweep every participant. Returns ``participant_id -> rows removed``."""
    return (await self.sweep(tenant_id)).counts

sweep async

sweep(tenant_id: str) -> ErasureSweep

Erase from every participant, surviving one that fails.

A participant is somebody else's store — a remote backend having a bad minute must not abort the tenant's erasure, skip every participant sorted after it, and lose the completion trail. Each failure is logged and reported by id instead.

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

async def sweep(self, tenant_id: str) -> ErasureSweep:
    """Erase from every participant, surviving one that fails.

    A participant is somebody else's store — a remote backend having a bad
    minute must not abort the tenant's erasure, skip every participant
    sorted after it, and lose the completion trail. Each failure is logged
    and reported by id instead.
    """
    result = ErasureSweep()
    for participant in self.participants():
        try:
            result.counts[participant.participant_id] = await participant.erase_tenant(
                tenant_id
            )
        except Exception as exc:  # noqa: BLE001 - erasure must not abort mid-sweep
            logger.warning(
                "Erase: participant %s failed for tenant %s",
                participant.participant_id,
                tenant_id,
                exc_info=True,
            )
            result.failed[participant.participant_id] = f"{type(exc).__name__}: {exc}"
    return result

ErasureSweep dataclass

ErasureSweep(counts: dict[str, int] = dict(), failed: dict[str, str] = dict())

The outcome of one sweep: what was erased, and what refused to be.

A sweep is deliberately not all-or-nothing. Erasure is the one operation where a partial success must still be reported rather than rolled back — the rows that went are gone — so a participant that raises is recorded by id and the sweep continues to the rest.

TenantErasureParticipant

Bases: Protocol

A store that holds tenant-scoped data and can be erased on request.

count_for_tenant async

count_for_tenant(tenant_id: str) -> int

How many rows this participant still holds for tenant_id.

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

async def count_for_tenant(self, tenant_id: str) -> int:
    """How many rows this participant still holds for ``tenant_id``."""
    ...

erase_tenant async

erase_tenant(tenant_id: str) -> int

Remove every row owned by tenant_id; return how many went.

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

async def erase_tenant(self, tenant_id: str) -> int:
    """Remove every row owned by ``tenant_id``; return how many went."""
    ...

verify_deletion_wiring async

verify_deletion_wiring(participant: TenantErasureParticipant, *, registry: ErasureParticipantRegistry = PRIVACY_DELETION_PARTICIPANTS, probe_tenant: str = 'deletion-wiring-probe', seed: int = 1) -> DeletionWiringAttestation

Run a real erase round-trip and attest to the result.

Raises when the participant is not registered — an attestation for a store the deletion path cannot reach would be worse than none at all.

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

async def verify_deletion_wiring(
    participant: TenantErasureParticipant,
    *,
    registry: ErasureParticipantRegistry = PRIVACY_DELETION_PARTICIPANTS,
    probe_tenant: str = "deletion-wiring-probe",
    seed: int = 1,
) -> DeletionWiringAttestation:
    """Run a real erase round-trip and attest to the result.

    Raises when the participant is not registered — an attestation for a store
    the deletion path cannot reach would be worse than none at all.
    """
    if not registry.holds(participant.participant_id):
        raise RecordingStoreError(
            f"{participant.participant_id!r} is not registered with the "
            f"{registry.registry_id!r} deletion path; wiring cannot be attested"
        )
    seeded = await participant.count_for_tenant(probe_tenant)
    if seeded < seed:
        raise RecordingStoreError(
            f"the deletion-wiring probe needs at least {seed} seeded row(s) for "
            f"{probe_tenant!r}; found {seeded}. Attesting against an empty store "
            "would prove nothing."
        )
    await registry.erase_all(probe_tenant)
    residual = await participant.count_for_tenant(probe_tenant)
    return DeletionWiringAttestation(
        participant_id=participant.participant_id,
        registry_id=registry.registry_id,
        verified_at=datetime.now(UTC),
        probe_tenant=probe_tenant,
        seeded=seeded,
        residual=residual,
    )