symfonic.core.observability.audit

audit

Audit logger — pluggable hook for destructive-operation telemetry.

Gate 3 of v5.5.0. Provides a process-wide hook that the FastAPI routers invoke after every destructive operation (delete, update, bulk-delete, export, erase). Matches the patterns already established by :func:~symfonic.agent.fastapi.dependencies.set_tenant_auth_verifier (5.4.47) and :func:~symfonic.core.observability.budget.set_budget_tracker (5.4.49):

• Library-level default is "no logger registered" — zero overhead, zero behavioural change for pre-existing agents.
• Scaffolded apps register a Postgres-backed implementation in the lifespan hook so every destructive op is captured for GDPR / SOC 2 audit trails.
• Fire-and-forget dispatch from hot paths — audit store failure MUST NOT crash the user's request.

Why a separate file (not budget.py)? Single-responsibility + the 300-LOC cap. Audit and budget share the hook pattern but otherwise have nothing in common.

AuditLogEntry

Bases: BaseModel

Immutable audit event emitted for every destructive operation.

Fields intentionally kept small so both in-memory and Postgres implementations can store them without a schema migration dance. metadata is the escape hatch for per-action context (counts, previous value, confirmation token, ...).

action class-attribute instance-attribute

action: str = Field(..., min_length=1, max_length=64)

Free-form action code. Convention: snake_case verb + optional resource suffix, e.g. delete_memory, bulk_delete, export_data, erase_all.

resource_type class-attribute instance-attribute

resource_type: str = Field(..., min_length=1, max_length=64)

E.g. memory_node, edge, procedure, tenant.

to_dict

to_dict() -> dict[str, Any]

Return a JSON-safe dict representation.

Source code in src/symfonic/core/observability/audit.py

def to_dict(self) -> dict[str, Any]:
    """Return a JSON-safe dict representation."""
    return {
        "tenant_id": self.tenant_id,
        "user_id": self.user_id,
        "action": self.action,
        "resource_type": self.resource_type,
        "resource_id": self.resource_id,
        "timestamp": self.timestamp.isoformat(),
        "metadata": self.metadata,
        "ip_address": self.ip_address,
        "user_agent": self.user_agent,
    }

AuditLogger

Bases: Protocol

Destination for audit events.

Implementations MAY persist synchronously or asynchronously. The contract is "record this entry, or raise" — retry / batching is an implementation detail. Callers in the router layer shield failures (fire-and-forget) so a sick audit store never 500s a user request.

InMemoryAuditLogger

InMemoryAuditLogger(max_entries: int = 10000)

Default in-process logger — keeps entries in an in-memory list.

Primarily for unit tests and local development. Production deployments should register a durable implementation (e.g. the scaffolded Postgres logger). The list is bounded by max_entries to prevent unbounded growth when no one is draining it.

Source code in src/symfonic/core/observability/audit.py

def __init__(self, max_entries: int = 10_000) -> None:
    self._entries: list[AuditLogEntry] = []
    self._max = max_entries
    self._lock: asyncio.Lock = asyncio.Lock()

drain async

drain() -> list[AuditLogEntry]

Return all buffered entries, clearing the buffer.

Tests use this to assert what was emitted during a request.

Source code in src/symfonic/core/observability/audit.py

async def drain(self) -> list[AuditLogEntry]:
    """Return all buffered entries, clearing the buffer.

    Tests use this to assert what was emitted during a request.
    """
    async with self._lock:
        snapshot = list(self._entries)
        self._entries.clear()
        return snapshot

entries

entries() -> list[AuditLogEntry]

Return a non-draining snapshot (sync, for inspection).

Not guaranteed to be consistent under concurrent record() calls — prefer drain() in assertions.

Source code in src/symfonic/core/observability/audit.py

def entries(self) -> list[AuditLogEntry]:
    """Return a non-draining snapshot (sync, for inspection).

    Not guaranteed to be consistent under concurrent ``record()``
    calls — prefer ``drain()`` in assertions.
    """
    return list(self._entries)

record async

record(entry: AuditLogEntry) -> None

Append entry to the in-memory buffer (drops oldest when full).

Source code in src/symfonic/core/observability/audit.py

async def record(self, entry: AuditLogEntry) -> None:
    """Append ``entry`` to the in-memory buffer (drops oldest when full)."""
    async with self._lock:
        self._entries.append(entry)
        if len(self._entries) > self._max:
            overflow = len(self._entries) - self._max
            del self._entries[:overflow]

emit_audit_event

emit_audit_event(entry: AuditLogEntry) -> None

Schedule an audit write without blocking the caller.

Never raises — failures are logged. If no logger is registered the call is a complete no-op (zero allocations beyond the entry itself).

The FastAPI routers use this directly in handlers so a slow / sick audit backend can never 500 a user request. The task is scheduled on the currently-running loop; when called outside an async context (defensive guard) we log a warning and drop the entry — no audit logger is expected to live that far up the stack.

Source code in src/symfonic/core/observability/audit.py

def emit_audit_event(entry: AuditLogEntry) -> None:
    """Schedule an audit write without blocking the caller.

    Never raises — failures are logged.  If no logger is registered the
    call is a complete no-op (zero allocations beyond the entry itself).

    The FastAPI routers use this directly in handlers so a slow / sick
    audit backend can never 500 a user request.  The task is scheduled
    on the currently-running loop; when called outside an async context
    (defensive guard) we log a warning and drop the entry — no audit
    logger is expected to live that far up the stack.
    """
    logger_ = _audit_logger
    if logger_ is None:
        return

    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        logger.warning(
            "emit_audit_event called outside a running loop; dropping entry "
            "(action=%s resource=%s)",
            entry.action, entry.resource_type,
        )
        return

    coro: Awaitable[None] = logger_.record(entry)
    task = loop.create_task(coro)
    task.add_done_callback(_audit_task_done)

get_audit_logger

get_audit_logger() -> AuditLogger | None

Return the currently registered audit logger, if any.

Source code in src/symfonic/core/observability/audit.py

def get_audit_logger() -> AuditLogger | None:
    """Return the currently registered audit logger, if any."""
    return _audit_logger

set_audit_logger

set_audit_logger(logger_: AuditLogger | None) -> None

Register (or clear) the process-wide audit logger.

Call once at app startup. Passing None disables audit emission — useful for tests that registered a logger and want to tear down.

Source code in src/symfonic/core/observability/audit.py

def set_audit_logger(logger_: AuditLogger | None) -> None:
    """Register (or clear) the process-wide audit logger.

    Call once at app startup.  Passing ``None`` disables audit emission
    — useful for tests that registered a logger and want to tear down.
    """
    global _audit_logger
    _audit_logger = logger_