Audit trail

An audit trail answers "who did what, to whose data, and was it allowed". The way one goes wrong is rarely by recording too little: an audit log built from request bodies is a second copy of the data it protects — unencrypted, long-retained, readable by whoever reads the logs. So the point of the audit plane is the allowlist: an action records only the metadata keys it declared, and a call that tries to record anything else raises instead of being filtered. A filter is invisible and the over-collecting call site survives to be copied; a refusal fails the test that introduced it.

Nothing is audited unless declared.

Declaring an action¶

from forze.application.contracts.audit import AuditObjectRef, AuditSpec
from forze.application.hooks.audit import Audited

INTERVAL_CORRECT = AuditSpec(
    action="interval.correct",
    allowed_metadata=frozenset({"root_id", "reason_code"}),
)

audited = Audited(
    spec=INTERVAL_CORRECT,
    metadata=lambda args, result: {"root_id": args.root_id, "reason_code": args.reason_code},
    object_ref=lambda args, result: AuditObjectRef(type="interval", id=str(args.root_id)),
)

Then bind it to the operations it covers, alongside their guards and transaction:

registry = audited.bind(
    registry.bind("interval.correct").bind_tx().set_route("pg").finish()
).finish()

• Metadata is scalars — strings, integers, finite floats, booleans, None, and UUIDs (stored as strings). A key outside allowed_metadata, or a value that is not a scalar, raises configuration (code audit_metadata_refused), naming the key and never the value.
• A name the log scrubber treats as a secret (password, token, api_key, session…) is refused when the spec is declared: an allowlist naming one is a declaration nobody meant.
• The actor and subject come from the authenticated identity, never from the arguments. subject_id is the principal the call runs for; actor_id is who performed it — the same principal, or the nearest delegate when the call is delegated. actor_ids holds the whole delegation chain, nearest first, so an agent acting through another is on record too, and the trail can be read by any of them ({"$values": {"actor_ids": {"$superset": [agent]}}}). The nearest actor is not always the one you configured: on the MCP path the server's own agent sits deeper in the chain, and actor_ids has it.

What gets recorded, and where¶

An audited write ends in one row, with an outcome of allowed, denied or failed; reads follow rules of their own, below. Where the row is written depends on the outcome, because that decides whether it can be trusted:

A denial is the event a trail most needs, and it happens before the handler and before the transaction opens — which is why the audit hooks are bound to the operation's outer scope as well as its transaction. An operation without a transaction binds with transactional=False, and its allowed row is written after it returns.

A failed or denied row is written after the operation's own transaction, not after every transaction: an audited operation called inside another one's transaction writes its row into that transaction, and the row goes if the outer one rolls back.

Reads (QUERY operations) follow two rules of their own, set by audit_reads:

• "after_authz" (the default) records a read once it is admitted and completes. A refused or failed read is not recorded: a log of refused reads is a record of who wanted to see whose data, a second disclosure surface. The row is written after the read, since the read's own transaction is read-only.
• A principal reading their own data, undelegated, is not recorded. Pass owner= — a callable naming whose data the read returned — for the audit to tell; without it, every admitted read is recorded, since the skip is never a guess.
• "never" records no reads for that action.

Auditing a kit's operations¶

An AggregateKit binds the hooks to the operations it generates, keyed by kernel op like its handlers:

from forze_kits.aggregates import AggregateKit
from forze_kits.aggregates.document import DocumentKernelOp

ORDERS = AggregateKit(
    spec=ORDER_SPEC,
    audit={
        DocumentKernelOp.CREATE: Audited(spec=AuditSpec(action="order.create")),
        DocumentKernelOp.GET: Audited(spec=AuditSpec(action="order.read")),
    },
)

An audited write runs in a transaction on the registry's tx_route, so its allowed row commits with the write or rolls back with it. An audited read gets no transaction it did not have; its row is written after it completes. A key naming an operation the kit does not compose is refused when the registry is built. An update's result is the row wrapped with its diff, so a callable that reads it unwraps it with written_read_model(result).

When the audit write fails¶

on_failure="fail" (the default): an operation whose audit row could not be written fails, and its write rolls back — an audited action nobody can account for did not happen. on_failure="ignore": the operation succeeds and a warning is logged. Neither direction is free — failing closed turns an audit-store outage into an outage, failing open turns it into a silent gap — so the choice is per action.

A metadata refusal raises under either policy: it is the over-collecting call site, not an outage. And a failed or denied operation keeps its own error — a row that cannot be written for it is logged, never raised in its place.

The collection¶

The trail is an ordinary document collection, so it is tenant-scoped, queryable and exportable like any other; only its metadata can be sealed, since the other fields are what it is queried by. forze_kits.integrations.audit provides the spec and the port:

from forze_kits.integrations.audit import AuditDepsModule, audit_record_spec

AUDIT = audit_record_spec()           # route "audit_events"; wire it like any document spec
audit_deps = AuditDepsModule(tx_route="pg")

Route AUDIT through your document deps module on the database the audited operations' own transactions use — an allowed row is written inside them — and register it in your spec inventory. The collection is yours to create:

CREATE TABLE audit_events (
    id uuid PRIMARY KEY,
    rev integer NOT NULL,
    created_at timestamptz NOT NULL,
    last_update_at timestamptz NOT NULL,
    tenant_id uuid NOT NULL,          -- when wired tenant-aware
    action text NOT NULL,
    outcome text NOT NULL,
    actor_id uuid,
    subject_id uuid,
    actor_ids uuid[] NOT NULL DEFAULT '{}',
    object_type text,
    object_id text,
    metadata jsonb NOT NULL DEFAULT '{}',
    at timestamptz NOT NULL
);

A table created before actor_ids existed needs the column. Until it is added, startup schema validation refuses the table (missing columns ['actor_ids']), and without that validation every audited operation fails closed on the write:

ALTER TABLE audit_events ADD COLUMN actor_ids uuid[] NOT NULL DEFAULT '{}';

On MongoDB, rows written before the field existed have no actor_ids at all. They read back as [], but {"$empty": true} does not match a missing field, so a "direct calls only" query there must also accept its absence:

{"$or": [{"$values": {"actor_ids": {"$empty": True}}}, {"$values": {"actor_ids": {"$null": True}}}]}

To keep the trail somewhere else, implement AuditPort — one record(entry) that joins the open transaction when there is one — and register its factory under AuditDepKey, as AuditDepsModule does.

audit_record_spec(encryption=...) may seal metadata; the other fields are what the trail is queried by, and sealing one is refused.

A deterministic-simulation invariant, audit_row_per_effect, checks the one-row claim under operation-level retries and injected faults: see Invariants.