An immutable audit trail for AI agent actions (FastAPI + async SQLAlchemy)

Rule zero: the audit log is append-only

The single most important design decision is a discipline, not a library: the audit table is never updated and never deleted from. Not by a feature, not by a migration, not by an admin in a hurry. An event happened, so a row exists, forever.

That sounds obvious until you realise how many ORMs make UPDATE and DELETE one method call away. So the rule has to be enforced by convention and by review: in this codebase, the audit module exposes exactly one operation — append.

from datetime import datetime, timezone
from sqlalchemy import String, DateTime, JSON
from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase
import uuid


class Base(DeclarativeBase):
    pass


class AuditLog(Base):
    """Append-only. Never UPDATE, never DELETE."""
    __tablename__ = "audit_logs"

    id: Mapped[str] = mapped_column(String, primary_key=True,
                                    default=lambda: str(uuid.uuid4()))
    event_type: Mapped[str] = mapped_column(String, index=True)   # "action.blocked", ...
    actor: Mapped[str | None] = mapped_column(String, nullable=True)
    resource_type: Mapped[str | None] = mapped_column(String, nullable=True)
    resource_id: Mapped[str | None] = mapped_column(String, nullable=True)
    details: Mapped[dict | None] = mapped_column(JSON, nullable=True)
    timestamp: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), index=True
    )

Note the indexes on event_type and timestamp — you will query this by "what happened to this agent, in this window," and you don't want a sequential scan over millions of rows when an auditor is waiting.

Rule one: writing the log must never break the agent's path

If your audit write is in the hot path and the database hiccups, you've just taken down the agent over a log line. That's backwards. Logging is critical for compliance but it must not be a single point of failure for execution.

FastAPI's BackgroundTasks is the simple, dependency-free way to defer the write until after the response is sent:

from fastapi import APIRouter, BackgroundTasks
from sqlalchemy.ext.asyncio import AsyncSession

router = APIRouter()


async def append_audit(
    session_factory,
    *,
    event_type: str,
    actor: str | None,
    resource_type: str | None,
    resource_id: str | None,
    details: dict | None,
) -> None:
    """One operation, append-only. Runs after the response is returned."""
    async with session_factory() as session:  # type: AsyncSession
        session.add(AuditLog(
            event_type=event_type,
            actor=actor,
            resource_type=resource_type,
            resource_id=resource_id,
            details=details,
        ))
        await session.commit()


@router.post("/v1/actions")
async def submit_action(payload: dict, background_tasks: BackgroundTasks) -> dict:
    decision = evaluate(payload)  # from the policy engine

    background_tasks.add_task(
        append_audit,
        session_factory,
        event_type=f"action.{decision.outcome}",
        actor="agent:" + payload["agent_id"],
        resource_type="action",
        resource_id=payload["action_id"],
        details={"violations": decision.violations, "reason": decision.reason},
    )
    return {"status": decision.outcome}

A couple of async gotchas worth saying out loud, because they cost me time:

• Use the asyncpg driver (postgresql+asyncpg://...), never a sync driver, in an async app. Mixing sync DB calls into an async event loop is a classic way to block everything.
• If your DATABASE_URL comes from an environment variable, .strip() it. A trailing newline pasted into a dashboard env var produces connection errors that look like everything and nothing. ## Rule two: make it tamper-evident, not just append-only

Append-only protects you from your own code. It does not, on its own, prove to a third party that no one with database access quietly edited a row. For that you want each entry to be cryptographically chained to the one before it — the same idea behind a hash chain.

Here's the technique. When you append an event, you hash its contents together with the hash of the previous event:

import hashlib
import json


def hash_entry(prev_hash: str, event: dict) -> str:
    # Canonical serialization matters: same input must always produce same bytes.
    payload = json.dumps(event, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256((prev_hash + payload).encode("utf-8")).hexdigest()

Store that hash on each row. Now, to verify the trail hasn't been altered, you recompute the chain from the start: if any single row was changed, deleted, or reordered, every hash after it breaks, and the final hash won't match. You don't have to trust the storage layer — you can prove integrity by recomputation.

(I'm being honest about state here: the append-only, non-blocking design above is what's running. The hash-chaining is the direction I'm adding next — if you build the same thing, do them in this order, because immutability-by-discipline is what makes the chaining meaningful.)

Why this is the unglamorous core

Nobody demos an audit table. But it's the thing that turns "we think the agent behaved" into "here is the record, in order, provably unaltered, exportable as evidence." In a bank, an insurer, a hospital — that distinction is the difference between an agent that stays a prototype and one that's allowed into production.

https://horkos.eu — a governance layer that sits in front of AI agents. If you've had to defend an automated system to an auditor, I'd love to know what they actually asked you for. That's the spec I'm trying to hit.