Audit trail export

AuditStore is the append-only log behind every governed decision. It records the full input/output of each gate evaluation — tool name, args, session context, the PolicyDecision that came back, the policy version that produced it, and the resulting outcome — in a SQLite table protected by SQL triggers that block updates and deletes. The store is the authoritative answer to "who decided what, on which rules, when?".

What gets recorded

Every record is an AuditRecord:

Field	What it carries
id	UUID — the record's primary key
session_id	The originating session
tool_name	The action attempted (approve_refund, read_file, etc.)
args	Full argument dict the agent passed
policy_decision	{allow, deny, requires_hitl, risk_tier, reason, ...}
policy_version	SHA-256 (truncated to 16 hex chars) over every .rego file
hitl_action_id	Set when the call went to the HITL queue
hitl_decided_by	Operator who approved/denied (when applicable)
outcome	One of seven canonical outcomes — see below
timestamp	UTC ISO-8601
context	Full SessionContext snapshot (role, scopes, depth, parent token, …)
parent_session_id / delegation_depth	Lineage for delegated agents

The seven outcomes

Outcome	When it's written
allowed	Policy returned allow=True and the tool ran
blocked	Policy returned deny=True
soft_denied	Policy returned allow=False without an explicit deny (no rule matched)
hitl_queued	Policy returned requires_hitl=True and the call was parked
hitl_approved	Operator approved the queued call
hitl_denied	Operator denied the queued call
hitl_timeout	The queued call expired without an operator decision

Every governed call resolves to exactly one of those — there are no holes in the log.

Tamper-resistance

The schema installs two triggers at table creation:

CREATE TRIGGER prevent_audit_update
BEFORE UPDATE ON audit_records
BEGIN SELECT RAISE(ABORT, 'audit_records is append-only'); END;

CREATE TRIGGER prevent_audit_delete
BEFORE DELETE ON audit_records
BEGIN SELECT RAISE(ABORT, 'audit_records is append-only'); END;

Any UPDATE or DELETE against the table aborts with a clear error. This is enforcement at the engine layer, not application code — even direct sqlite3 shell access will raise. The store also runs in WAL journal mode for concurrent readers + a single writer, with indexes on session_id, outcome, and timestamp DESC.

The SQLite file itself isn't cryptographically sealed against an attacker with shell access — that's an operational concern (file permissions, write-once mounts, off-host shipping). Combined with the content-addressed policy_version, the trigger-protected schema gives you the audit guarantee application code cannot retroactively edit the log.

Setup

from kitelogik.audit.store import AuditStore

store = AuditStore(db_path="audit.db")   # ":memory:" is rejected
await store.setup()                       # one-time table + trigger creation

The store is async and safe to share across an entire process.

Query

query() is the read path:

# Recent activity for one session
records = await store.query(session_id="sess_001", limit=50)

# All denials in the last batch of records
denials = await store.query(outcome="blocked", limit=200)

# All calls to one tool
calls = await store.query(tool_name="approve_refund", limit=100)

Filters are AND-combined; limit defaults to 50 and orders results newest-first.

Export a session

export_session() returns a JSON-serialisable bundle suitable for shipping to a SOC team, a regulator, or a downstream forensics pipeline:

bundle = await store.export_session("sess_001")

# bundle = {
#   "session_id":     "sess_001",
#   "record_count":   17,
#   "policy_version": "a3f9e1c2b4d7e6f5",
#   "exported_at":    "2026-04-28T14:00:00Z",
#   "records":        [ ... 17 AuditRecord dicts in chronological order ... ],
#   "integrity_hash": "sha256:..."  # over the records JSON
# }

import json
with open("session_001_export.json", "w") as f:
    json.dump(bundle, f, indent=2, default=str)

The integrity_hash is a SHA-256 over the records' canonical JSON. A downstream auditor can recompute the hash from the exported records; any tampering between export and review changes the digest.

Replay against a new policy

PolicyReplayer re-evaluates historical records against the live gate — useful for "what would this policy change have done to last month's traffic?":

from kitelogik.audit.replay import PolicyReplayer

replayer = PolicyReplayer(gate)
results  = await replayer.replay_session(store, session_id="sess_001")

changed = [r for r in results if r.outcome_changed]
print(f"{len(changed)} of {len(results)} decisions would differ")

for r in changed:
    print(f"  {r.tool_name:24} {r.original_outcome:>12} -> {r.replayed_outcome}")

ReplayResult carries both the original and replayed outcome, the risk tiers from each, and the replayed reason — so a compliance team can preview the blast radius of a Rego change before it ships, and an incident response team can identify exactly which historical calls a new rule would have caught.

Vocabulary mismatch

AuditStore.outcome uses seven values (allowed, blocked, soft_denied, hitl_queued, …). ReplayResult.replayed_outcome uses three (allowed, denied, pending_review) — it's a synthesised view of the current policy's decision, not a record of execution. Compare on intent, not on the literal string.

Operational pattern: ship to long-term storage

SQLite is the right home for recent audit traffic. For multi-year retention you want a tiered approach: keep the SQLite file as the hot, queryable surface, then ship completed sessions off-host as exported bundles. A simple cron job is enough:

from datetime import datetime, timedelta, timezone
import boto3, json

s3 = boto3.client("s3")
cutoff = (datetime.now(timezone.utc) - timedelta(days=7)).isoformat()

# Find sessions whose most recent record is older than the cutoff
old_sessions = {
    r.session_id
    for r in await store.query(limit=10_000)
    if r.timestamp < cutoff
}

for sid in old_sessions:
    bundle = await store.export_session(sid)
    s3.put_object(
        Bucket="kitelogik-audit-archive",
        Key=f"{sid}/{bundle['exported_at']}.json",
        Body=json.dumps(bundle, default=str),
    )

The exported bundle includes the integrity_hash, so any later read from S3 can verify the JSON wasn't altered in transit or at rest.

What you'd build yourself

• Cross-tenant indexing. The bundled store is single-tenant. A multi-tenant audit with per-tenant query isolation is a backend swap on top of the same row schema.
• Postgres backend. The default is SQLite — fine to billions of rows with WAL + the bundled indexes. HA / connection pooling means bringing a Postgres-backed AuditStore implementation.
• Built-in shipping. No bundled exporter for Splunk / Datadog / S3. Wire the loop yourself — the export bundle is plain JSON.

Related

• Risk tiers & HITL — the queue and the outcomes it produces (hitl_queued, hitl_approved, hitl_denied, hitl_timeout)
• OpenTelemetry — for live request-path observability, complementary to the durable audit log