Audit log with per-user attribution

An audit log is a signed, append-only record of what the agent did and on whose behalf: runs started and finished, tools called, results returned. Turn it on when you need to answer “who did what, and when”, for compliance, incident review, or customer disputes.

Every primitive in Loomflow writes audit entries when an AuditLog is configured. AuditEntry carries a top-level user_id field; the HMAC signature covers it, so tampering that swaps a user id breaks verification. Each entry also carries prev_hash, the previous entry’s signature, inside its signed content, so the entries form a hash chain: deleting, inserting or reordering entries breaks verification too, not only editing one.

Two backends behind the same protocol:

Quick wiring

from loomflow import Agent
from loomflow.security import FileAuditLog
 
audit = FileAuditLog("./audit.jsonl", secret="prod-secret")
agent = Agent("...", model="claude-opus-4-7", audit_log=audit)
 
await agent.run("export my data", user_id="alice", session_id="s1")

Pass the same audit_log to a Workflow and step entries interleave with agent entries under the same session_id:

from loomflow import Workflow
 
wf = Workflow.chain([step_a, step_b], audit_log=audit)
await wf.run("...", user_id="alice")

What the default captures (and doesn’t)

The default audit log is compliance-friendly. It captures enough to prove what happened, but it doesn’t store the customer’s content verbatim:

• Prompts get truncated at 500 characters.
• The model’s final output isn’t recorded at all.
• Tool results carry only ok, denied, error, and reason. The actual return value of the tool stays out of the log.

That’s the right shape for any regime that prohibits logging customer content. If you’re debugging an internal flow on synthetic data and want everything captured verbatim, you opt in.

Verbatim capture: audit_log={"scope_full": True}

Instead of constructing an AuditLog instance, you can hand Agent or Workflow a config dict. The resolver builds the right backend for you. Set scope_full: True and prompts, outputs, and full tool-result bodies all land in the log:

from loomflow import Agent
 
agent = Agent(
    "You are a helpful assistant.",
    model="claude-opus-4-7",
    audit_log={
        "name": "./audit.jsonl",    # path; omit this to get an in-memory log
        "scope_full": True,         # capture everything verbatim
        "secret": "my-org-hmac-key", # optional, for HMAC signing
    },
)

The same dict works on Workflow. One parameter, two primitives:

wf = Workflow.chain(
    [step_a, step_b],
    audit_log={"scope_full": True, "secret": "my-org-hmac-key"},
)

Internally the resolver wraps the backend in FullTranscriptAuditLog. Signatures still verify, because the wrapper just forwards entries through to the underlying log.

When to flip scope_full on:

• Incident review where you need the exact prompts that hit the model.
• Pre-prod debugging on synthetic data.
• A short-window post-mortem, with PII redaction handled downstream.

Leave it off in production unless compliance has signed off. The default truncation isn’t a bug, it’s the safer default.

from loomflow.security import FullTranscriptAuditLog
# (you don't usually import this; the resolver picks it for you
# when you pass `scope_full: True` in the dict)

Querying by user, session or action

# All entries for alice
alice_entries = await audit.query(user_id="alice")
 
# Entries for one session
session_entries = await audit.query(session_id="s1")
 
# Combine filters: alice's completed runs only
done = await audit.query(user_id="alice", action="run_completed")

query() filters on the entry’s top-level user_id, session_id and action fields, so you never dig through payloads. FileAuditLog reads the whole file for each query; for large logs or heavy analysis, ship the JSONL into a database.

HMAC tamper detection

from loomflow.security.audit import verify_signature
 
entries = await audit.query(user_id="alice")
sample = entries[0]
 
verify_signature(sample, secret="prod-secret")  # → True
 
# Mutate the payload, and verification fails:
tampered = sample.model_copy(
    update={"payload": {**sample.payload, "tampered": True}}
)
verify_signature(tampered, secret="prod-secret")  # → False
 
# A wrong secret also fails, which catches secret-rotation mistakes
verify_signature(sample, secret="wrong-key")  # → False

The signature is over the canonical serialized body. Additions, changes, or removals all break it. The signature covers user_id, so swapping a different user’s id into an entry breaks verification too. Payloads are signed in the same JSON form they’re stored in, so an entry with a datetime, set or tuple in its payload still verifies after it’s read back.

Verifying the whole log

verify_signature checks one entry. To check a whole log, call verify() on either backend, or verify_chain(entries, secret) on entries you loaded yourself. Both check every signature, that seq starts at 1 and goes up by one, and that each prev_hash links to the entry before it, so a deleted, inserted or reordered entry shows up:

from loomflow.security import FileAuditLog, verify_chain
 
audit = FileAuditLog("./audit.jsonl", secret="prod-secret")
 
report = await audit.verify()
report.ok         # True when nothing is wrong
report.entries    # how many entries were checked
report.head       # the last entry's signature
for problem in report.problems:
    print(problem.line, problem.seq, problem.reason)
    # e.g. 2 3 "seq 3 where 2 was expected"
    #      2 3 "prev_hash does not match the entry before it"
 
# The same check on entries from anywhere (a database export, a copy):
report = verify_chain(await audit.all_entries(), "prod-secret")

verify() returns an AuditVerification(ok, entries, head, problems); each AuditProblem has a reason, the entry’s seq and, for FileAuditLog, the 1-based line in the file. FileAuditLog.verify() also reports lines that aren’t valid entries (a corrupted or hand-edited line) by line number.

Cutting entries off the end leaves a valid chain. To catch it, store report.head somewhere the log’s writer can’t change (another system, a ticket, a signed release note) and pass it back later:

later = await audit.verify(expected_head=saved_head)
# problems include "the expected head is not in the log (entries were removed)"

Entries written before the hash chain existed (no prev_hash) still verify; they’re checked for their signature and seq only.

Restart recovery and several writers

FileAuditLog takes an exclusive file lock for each append and continues from the file’s last entry, so a restarted process, several processes, or several log objects can append to one file without repeating a seq or forking the chain. If the last line was torn by a crash mid-write, the next append starts a new line instead of corrupting itself. query() skips lines it can’t parse and logs how many it skipped; verify() reports them by line number.

# Crash → restart. New FileAuditLog reads the existing file:
restarted = FileAuditLog("./audit.jsonl", secret="prod-secret")
 
before = (await restarted.query(user_id="alice"))[-1].seq
 
# Run more work:
await wf.run("more", user_id="alice")
 
after = (await restarted.query(user_id="alice"))[-1].seq
# `after > before`, no seq collisions.

Full worked example

The framework ships examples/12_audit_log.py exercising the behaviors above with EchoModel (no API key needed):

python examples/12_audit_log.py

Output (Part 5, verbatim capture, trimmed):

============================================================
Part 1 — InMemoryAuditLog (fast, ephemeral)
============================================================
  Alice's audit entries: 2
  Bob's audit entries:   2
  Sample action chain:   ['run_started', 'run_completed']

============================================================
Part 2 — FileAuditLog (durable, JSONL)
============================================================
  Workflow result: 'HELLO!'
  Wrote 1568 bytes to _audit_log_demo.jsonl
  Entries written: 4
  Actions:         ['step_started', 'step_completed', 'step_started', 'step_completed']

============================================================
Part 3 — Tamper detection via HMAC
============================================================
  Original entry verifies:    True
  Tampered entry verifies:    False ← False detects the change
  Wrong-secret check passes:  False ← catches secret-rotation mistakes

============================================================
Part 4 — Restart recovery (seq counter persists)
============================================================
  Highest seq before new run:  4
  Highest seq after new run:   8
  Δ = 4 new entries appended, no seq collisions.

Read examples/12_audit_log.py.