Skip to content

symfonic.platform.audit

audit

AUD-1..4 — one narrow seam, and what happens when it breaks.

AUD-4 is the clause with teeth and it is the platform-side answer to T3.4.5's finding F-1: a broken objector that never reaches degraded_stages, i.e. a monitoring surface reporting health while a guard is down. So a failing sink does two things it does not do today — it raises a degradation signal, and for a destructive operation it fails the operation closed. An erasure that proceeds after its intent record was dropped is an untraceable destruction, and "we tried to log it" is not a trail.

AuditSeam 

AuditSeam(sink: AuditSink | None = None, *, on_degraded: Callable[[AuditRecord], None] | None = None)

The one path platform services write audit facts through.

Handlers do not construct sinks and do not reach for a module-level emitter; they are handed a seam. That is what makes "every administrative operation is audited before it mutates" (ADM-6) checkable rather than conventional.

Source code in src/symfonic/platform/audit.py 
81
82
83
84
85
86
87
88
89
def __init__(
    self,
    sink: AuditSink | None = None,
    *,
    on_degraded: Callable[[AuditRecord], None] | None = None,
) -> None:
    self._sink = sink if sink is not None else NullAuditSink()
    self._on_degraded = on_degraded
    self._degraded = False

degraded property 

degraded: bool

True once a record failed to land. Sticky: it describes the host.

record async 

record(record: AuditRecord, *, destructive: bool = False) -> None

Emit. On failure, degrade loudly — and for destruction, fail closed.

Source code in src/symfonic/platform/audit.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
async def record(self, record: AuditRecord, *, destructive: bool = False) -> None:
    """Emit. On failure, degrade loudly — and for destruction, fail closed."""
    try:
        await self._sink.emit(record)
    except Exception as exc:  # noqa: BLE001 - the failure is the subject
        self._degraded = True
        logger.error(
            "audit sink failed for action=%s scope=%s: %s",
            record.action,
            record.scope_key,
            exc,
        )
        if self._on_degraded is not None:
            self._on_degraded(record)
        if destructive:
            raise AuditSinkError(
                f"the audit seam could not record {record.action!r} and the "
                "operation is destructive; proceeding would destroy data with "
                "no trail (ADM-6, AUD-4)"
            ) from exc

AuditSinkError

Bases: SymfonicError

The seam could not record, and the operation may not proceed.

NullAuditSink

Drops records. The default only because a host that wants none says so.

RecordingAuditSink 

RecordingAuditSink()

Keeps records in memory — the test double, and a usable dev sink.

Source code in src/symfonic/platform/audit.py 
63
64
def __init__(self) -> None:
    self.records: list[AuditRecord] = []