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_STEP_SUCCEEDED	erasure_step_succeeded
ERASURE_STEP_FAILED	erasure_step_failed
ERASURE_COMPLETED	erasure_completed
MANIFEST_SNAPSHOT	manifest_snapshot

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.