Audit

Art. 30 audit — an append-only trail as the source of truth.

AuditEvent

class AuditEvent(BaseModel):
    event_id: UUID
    event_type: AuditEventType
    occurred_at: datetime
    payload: dict[str, str | int | bool] = Field(default_factory=dict)
    subject_ref: str = Field(min_length=1, max_length=255)

One immutable entry in the audit trail.

Events carry references and metadata, never rich PII — an audit trail that itself hoards personal data would defeat its purpose. payload values are intentionally restricted to short, loggable scalars.

Fields:

• event_id (UUID): Unique id, assigned at creation, never reused.
• event_type (AuditEventType): What happened.
• occurred_at (datetime): When it happened (UTC).
• payload (dict[str, str | int | bool]): Small structured details (counts, table names, versions).
• subject_ref (str): Opaque subject reference (NOT an email or name).

AuditEventType

class AuditEventType(StrEnum):
    ...

Every kind of event the audit trail records.

Adding members is a MINOR change; removing or renaming is MAJOR (old trails must stay readable forever).

Member	Value
CONSENT_GRANTED	consent_granted
CONSENT_WITHDRAWN	consent_withdrawn
EXPORT_REQUESTED	export_requested
EXPORT_COMPLETED	export_completed
ERASURE_REQUESTED	erasure_requested
ERASURE_LOCAL_COMPLETED	erasure_local_completed
ERASURE_EXPIRY_SCHEDULED	erasure_expiry_scheduled
ERASURE_STEP_SUCCEEDED	erasure_step_succeeded
ERASURE_STEP_FAILED	erasure_step_failed
ERASURE_VERIFIED	erasure_verified
ERASURE_VERIFICATION_FAILED	erasure_verification_failed
ERASURE_EXTERNAL_VERIFIED	erasure_external_verified
ERASURE_EXTERNAL_VERIFICATION_FAILED	erasure_external_verification_failed
ERASURE_COMPLETED	erasure_completed
ERASURE_REQUEUED	erasure_requeued
ERASURE_REPLAYED	erasure_replayed
MANIFEST_SNAPSHOT	manifest_snapshot
RECTIFICATION_REQUESTED	rectification_requested
RECTIFICATION_LOCAL_COMPLETED	rectification_local_completed
RECTIFICATION_STEP_SUCCEEDED	rectification_step_succeeded
RECTIFICATION_STEP_FAILED	rectification_step_failed
RECTIFICATION_COMPLETED	rectification_completed
RESTRICTION_PLACED	restriction_placed
RESTRICTION_LIFTED	restriction_lifted
RETENTION_EXPIRED	retention_expired

AuditSink

Protocol — implement these members in your own class; do not subclass.

class AuditSink(Protocol):
    ...

Anything that can durably append and read back audit events.

This protocol is public API. It is extended additively only (new optional methods with default implementations) — existing custom sinks must never break on upgrade.

AuditSink.append

def append(event: AuditEvent) -> None

Durably append one event.

Must be atomic per event and must never overwrite anything. Sync by design — appends run inside the erasure/consent transaction path (ADR 0006); an async external sink would be an additive separate adapter, never a change to this protocol.

Args:

• event (AuditEvent): The event to persist.

AuditSink.read

def read(subject_ref: str) -> Sequence[AuditEvent]

Read all events for one subject, oldest first.

Args:

• subject_ref (str): The opaque subject reference to filter by.

Returns:

• Sequence[AuditEvent] — The subject’s full trail — what a regulator asks for first.

DatabaseAuditSink

class DatabaseAuditSink:
    def __init__(session_factory: sessionmaker, audit_events: Table) -> None

Append-only audit storage in the application’s own database.

The default sink: zero data leaves the user’s system in OSS mode. Rows are insert-only; the table carries no update path and the sink exposes none. Each append commits in its own short transaction (ADR 0006), so an event survives even when the caller’s surrounding transaction later rolls back — audit evidence is never lost to an unrelated failure.

DatabaseAuditSink.append

def append(event: AuditEvent) -> None

Durably append one event (insert-only).

Commits immediately in a transaction of its own. A duplicate event_id raises the database’s integrity error — an existing row is never overwritten.

Args:

• event (AuditEvent): The event to persist.

DatabaseAuditSink.read

def read(subject_ref: str) -> Sequence[AuditEvent]

Read one subject’s trail, oldest first.

Ordering is by occurred_at, ties broken by event_id so repeated reads always agree.

Args:

• subject_ref (str): The opaque subject reference to filter by.

Returns:

• Sequence[AuditEvent] — All events recorded for the subject.

Raises:

• AuditIntegrityError — If the trail contains an event_type this version of effaced cannot interpret (recorded by a newer release). This is deliberately all-or-nothing: one unreadable entry fails the whole read rather than serving a silently incomplete trail — partial evidence presented as complete would be worse than no answer. Upgrading effaced restores readability; nothing is lost.

DatabaseAuditSink.read_since

def read_since(since: datetime) -> Sequence[AuditEvent]

Read every subject’s events from since onward, oldest first.

The ReplaySource capability (ADR 0023): the window a backup-replay derivation consumes. The boundary is inclusive (occurred_at >= since) — matching the replay rule that an erasure at exactly the backup instant is replayed — and ordering ties in occurred_at resolve by event_id, exactly as read does.

Args:

• since (datetime): The instant to read from, inclusive. Must be timezone-aware — the trail’s timestamps are UTC, and a naive comparison could silently shift the window boundary by the session offset, dropping events from the read.

Returns:

• Sequence[AuditEvent] — All events at or after since, across all subjects.

Raises:

• ConfigurationError — If since is timezone-naive — the same guard ReplayPlan.derive> applies to its cutoff, for the same reason.
• AuditIntegrityError — If the window contains an event_type this version of effaced cannot interpret — all-or-nothing, as in read.