PRIV-1..8 — the privacy service: export assembly and the erasure boundary.
The service owns the decision and the order: confirm destructive intent
against the resolved scope (ADM-5), audit the intent before anything mutates
(ADM-6), publish the tombstone atomically with the generation advance (PRIV-4),
then hand the sweep to the saga. It owns none of the per-store mechanics — each
participant erases itself through the row-20 port, which is what lets a backend
in another repository join the erasure path without this module changing.
The ordering is the part worth defending. Tombstone first means a request
admitted a microsecond after the operator pressed delete cannot start, and an
in-flight write that already read the old generation cannot land. Sweeping
first and tombstoning after would leave exactly that window open, and it is the
window through which erased data comes back.
PrivacyService(*, registry: SubjectDataStoreRegistry, fence: ErasureFence, saga_store: ErasureSagaStore, audit: AuditSeam | None = None, max_attempts: int = 3, bind_fence_audit: bool = True)
Export (Article 20) and erasure (Article 17), behind one boundary.
Source code in src/symfonic/platform/privacy.py
| def __init__(
self,
*,
registry: SubjectDataStoreRegistry,
fence: ErasureFence,
saga_store: ErasureSagaStore,
audit: AuditSeam | None = None,
max_attempts: int = 3,
bind_fence_audit: bool = True,
) -> None:
self._registry = registry
self._fence = fence
self._audit = audit if audit is not None else AuditSeam()
# EFX-ER-4: the fence refuses writers racing an erasure; this is what
# makes those refusals visible outside the process. The hook is optional
# rather than part of row 14 because an adapter whose backend already
# audits its own conditional-update failures should not be made to
# accept a second path for the same fact.
#
# The binding is *claimed*, not assumed. HOST-3's two-agents-in-one-
# process case can hand the same fence to two services, and the second
# constructor silently rebinding would send the first host's denials to
# the second host's sink. The fence refuses the overwrite; a host that
# deliberately shares a fence passes ``bind_fence_audit=False`` and
# leaves the binding to whoever made it.
if bind_fence_audit:
attach = getattr(fence, "set_denial_recorder", None)
if callable(attach):
attach(fence_denial_recorder(self._audit))
self._guard = SubjectGuard(fence=fence, saga_store=saga_store)
self._saga = ErasureSaga(
registry=registry,
fence=fence,
saga_store=saga_store,
audit=self._audit,
max_attempts=max_attempts,
)
|
admit(principal: AuthenticatedPrincipal) -> None
SCOPE-14 step 2, exposed for the admission gate.
Source code in src/symfonic/platform/privacy.py
| async def admit(self, principal: AuthenticatedPrincipal) -> None:
"""SCOPE-14 step 2, exposed for the admission gate."""
await self._guard.require_admission(principal.scope)
|
erase(principal: AuthenticatedPrincipal, *, confirmation: str) -> dict[str, Any]
Article 17. Confirm, audit intent, tombstone, then sweep.
Source code in src/symfonic/platform/privacy.py
| async def erase(
self, principal: AuthenticatedPrincipal, *, confirmation: str
) -> dict[str, Any]:
"""Article 17. Confirm, audit intent, tombstone, then sweep."""
expected = f"DELETE-{principal.scope.tenant_id}"
if confirmation != expected:
raise DestructiveConfirmationError(
"Missing or incorrect confirmation. Pass "
"?confirmation=DELETE-<tenant_id> matching the *resolved* tenant "
"to proceed."
)
# ADM-6: intent first, and fail closed if it cannot be recorded.
await self._audit.record(
AuditRecord(
action="erase_all",
outcome="initiated",
principal_id=principal.principal_id,
scope_key=principal.scope.scope_key,
metadata={"stage": "initiated", "confirmation": expected},
),
destructive=True,
)
await self._fence.publish_tombstone(
principal.scope.scope_key, reason=f"erasure requested by {principal.principal_id}"
)
outcome = await self._saga.run(principal.scope)
await self._audit.record(
AuditRecord(
action="erase_all",
outcome="completed" if outcome.complete else "partial",
principal_id=principal.principal_id,
scope_key=principal.scope.scope_key,
metadata={
"stage": "completed",
"counts": dict(outcome.counts),
"unconfirmed": list(outcome.unconfirmed),
},
)
)
return self._render(outcome)
|
export(principal: AuthenticatedPrincipal) -> dict[str, Any]
PRIV-8 — exactly what the tenant's own queries could see.
Suppressed for a tombstoned subject like any other read: an export is
not a privileged back door around the erasure it follows.
Source code in src/symfonic/platform/privacy.py
| async def export(self, principal: AuthenticatedPrincipal) -> dict[str, Any]:
"""PRIV-8 — exactly what the tenant's own queries could see.
Suppressed for a tombstoned subject like any other read: an export is
not a privileged back door around the erasure it follows.
"""
await self._guard.require_readable(principal.scope)
fragments: dict[str, list[Any]] = {}
unavailable: list[str] = []
for store in self._registry.participants():
fragment = await store.export_subject(principal.scope)
if not fragment.exportable:
unavailable.append(fragment.participant_id)
continue
fragments[fragment.participant_id] = [dict(r) for r in fragment.records]
payload: dict[str, Any] = {
"scope_key": principal.scope.scope_key,
"tenant_id": principal.scope.tenant_id,
"exported_at": datetime.now(UTC).isoformat(),
"schema_version": EXPORT_SCHEMA_VERSION,
"fragments": fragments,
}
if unavailable:
# Honest about the gap rather than quietly complete.
payload["not_exportable"] = sorted(unavailable)
await self._audit.record(
AuditRecord(
action="export_data",
outcome="ok",
principal_id=principal.principal_id,
scope_key=principal.scope.scope_key,
metadata={
"record_count": sum(len(v) for v in fragments.values()),
"participants": len(fragments),
},
)
)
return payload
|
read_guard(principal: AuthenticatedPrincipal) -> None
PRIV-5: reads are suppressed while any participant is unconfirmed.
Source code in src/symfonic/platform/privacy.py
| async def read_guard(self, principal: AuthenticatedPrincipal) -> None:
"""PRIV-5: reads are suppressed while any participant is unconfirmed."""
await self._guard.require_readable(principal.scope)
|
resume(scope: SubjectScope) -> dict[str, Any]
Drive an interrupted erasure forward. Safe to call repeatedly.
This is what makes PRIV-3's "resumable" operational rather than
theoretical: a scheduler, an operator, or a restart hook calls it, and
the saga picks up exactly the participants that never confirmed.
Source code in src/symfonic/platform/privacy.py
| async def resume(self, scope: SubjectScope) -> dict[str, Any]:
"""Drive an interrupted erasure forward. Safe to call repeatedly.
This is what makes PRIV-3's "resumable" operational rather than
theoretical: a scheduler, an operator, or a restart hook calls it, and
the saga picks up exactly the participants that never confirmed.
"""
return self._render(await self._saga.run(scope))
|