Audit Trails for LLM Actions

What an audit trail records

For each agent run: who triggered it (user id, system trigger), when, what prompt, what tools called with what arguments, what tool results returned, what final answer the agent produced, and what permission decisions a human made. Together, these answer 'what happened' months later.

Hooks are your audit hook (literally)

The simplest implementation: a PostToolUse hook that appends a structured line per tool call, plus a UserPromptSubmit hook for the trigger and a final summary on Stop. Each line carries a session id so you can stitch them into a timeline.

Retention vs cost

Audit logs grow. Decide retention policy up front — 30/90/365 days based on regulatory or business need. Cold-storage tiers (S3 Glacier, GCS Coldline) make multi-year retention cheap; hot-storage queries should target the last N days only.

Principle: An action without an audit trail is a fingerprint-free incident. Build the trail first; add the cool features later.

Code

Structured audit logger·python

import hashlib, json, time, uuid
from claude_agent_sdk import HookContext, HookMatcher

def fingerprint(value) -> str:
    # stable across processes, unlike Python's hash()
    return hashlib.sha256(json.dumps(value, sort_keys=True, default=str).encode()).hexdigest()[:16]

class Audit:
    def __init__(self, path: str, run_id: str | None = None):
        self.path = path
        self.run_id = run_id or str(uuid.uuid4())

    def emit(self, event_type: str, **fields):
        line = {"ts": time.time(), "run_id": self.run_id, "event": event_type, **fields}
        with open(self.path, "a") as f:
            f.write(json.dumps(line) + "\n")

def make_hooks(audit: Audit) -> dict:
    async def on_user_prompt(input_data: dict, tool_use_id: str | None, context: HookContext) -> dict:
        audit.emit("user_prompt", session=input_data["session_id"], prompt_sha=fingerprint(input_data["prompt"]))
        return {}

    async def on_tool_post(input_data: dict, tool_use_id: str | None, context: HookContext) -> dict:
        audit.emit("tool_use", session=input_data["session_id"], tool=input_data["tool_name"],
                   tool_use_id=tool_use_id, input_keys=sorted(input_data["tool_input"]))
        return {}

    async def on_tool_failure(input_data: dict, tool_use_id: str | None, context: HookContext) -> dict:
        audit.emit("tool_failure", session=input_data["session_id"], tool=input_data["tool_name"],
                   tool_use_id=tool_use_id, error=input_data["error"])
        return {}

    async def on_stop(input_data: dict, tool_use_id: str | None, context: HookContext) -> dict:
        audit.emit("stop", session=input_data["session_id"])
        return {}

    return {
        "UserPromptSubmit": [HookMatcher(hooks=[on_user_prompt])],
        "PostToolUse": [HookMatcher(hooks=[on_tool_post])],
        "PostToolUseFailure": [HookMatcher(hooks=[on_tool_failure])],
        "Stop": [HookMatcher(hooks=[on_stop])],
    }

# options = ClaudeAgentOptions(hooks=make_hooks(Audit("/var/log/agent-audit.jsonl")))

Stitching events into a session timeline·python

import json
from collections import defaultdict

def session_timeline(audit_path: str):
    runs = defaultdict(list)
    for line in open(audit_path):
        ev = json.loads(line)
        runs[ev["run_id"]].append(ev)
    for run_id, events in runs.items():
        events.sort(key=lambda e: e["ts"])
        print(f"\n== run {run_id} ({len(events)} events) ==")
        for e in events:
            print(f"  {e['ts']:.3f}  {e['event']}")

Exercise

Implement the audit logger pattern in one agent. Run a session, then write a query that answers 'how many tool calls in the last hour, by tool name'. Verify the log answers it cheaply.

Hint

If your log is text-only, switch to JSONL — every observability question gets harder when the format is not structured.

What an audit trail records

Hooks are your audit hook (literally)

Retention vs cost

Code

External links

Exercise

Progress

Comments 0