Audit Log Schema Design for Claims Systems

Designing the audit-event schema is the single decision that determines whether a claims platform can reconstruct its own past under examination, and this guide is one component of the broader Core Architecture & Compliance Mapping domain that defines the schema contracts every downstream system writes against. An audit log is not application logging with a longer retention period — it is an append-only, immutable ledger of every consequential action the platform takes, structured so that a regulator, a fraud investigator, or a future engineer can replay a claim’s entire life without touching live production state.

Teams that treat the audit trail as free-text logging discover its inadequacy at the worst possible moment: a market-conduct examination, a coverage dispute in litigation, or a fraud referral that hinges on who changed a reserve figure and when. Three failure modes recur.

The first is shape drift. When every service invents its own event format, the log becomes a heterogeneous pile that no query can traverse. One service writes user, another writes actor_id, a third writes changed_by; timestamps arrive in local time from one worker and UTC from another; some events carry the claim id, some bury it in a message string. Reconstructing a single claim’s history then means writing bespoke parsers per producer, and every parser is a place the reconstruction can silently drop an event.

The second is mutability. If audit records live in a table that supports UPDATE and DELETE, the log proves nothing — an examiner has no cryptographic reason to believe the sequence they are reading is the sequence that actually occurred. The whole evidentiary value of an audit trail rests on the guarantee that no entry, once written, can be altered or removed without detection.

The third is PII sprawl. Naively logging full event payloads means policyholder names, Social Security numbers, and medical details land in a store that is, by design, never deleted and widely readable for analysis. That collides directly with data-retention statutes and with the perimeter controls that Data Boundary Enforcement establishes. An audit schema must record what happened with enough fidelity to be evidentiary, while carrying the minimum personal data necessary to do so.

The remedy to all three is a single, versioned event envelope that every stage writes to, immutable storage semantics enforced at the schema layer, and a discipline of referencing sensitive subjects rather than embedding them.

The schema itself needs only Pydantic for validation and structlog for structured audit lines; the hash chain uses the standard library. Pin the versions so serialization and coercion semantics cannot drift between the producer that writes an event and the verifier that reads it years later.

Copy
python -m venv .venv && source .venv/bin/activate
pip install "pydantic==2.*" "structlog==24.*"
# stdlib only for hashing/serialization: hashlib, json, uuid, datetime, decimal

Two upstream invariants must already hold before this stage runs. First, currency and quantitative fields entering a payload must already be normalized to Decimal at the ingestion boundary — never let a float reach the canonical serializer, because its repr is platform-dependent and will break replay. Second, the actor identity must be resolved to a stable principal (a service account name or an authenticated user id), not a display string that can change.

The envelope is the fixed set of fields that identify any event regardless of what happened. Its discipline is that these fields never vary: an examiner learns the shape once and can then read the entire log. The payload — the part that differs by action — is a separate, typed object nested inside.

The envelope carries seven load-bearing fields. The event id is a UUID assigned at construction, the primary key of the record. The timestamp is always UTC in ISO 8601, generated with datetime.now(timezone.utc) and never from a naive local clock. The actor is the principal responsible for the action. The action is a controlled verb from an enumeration (record.validated, deductible.applied, referral.raised) so the log is queryable by event type rather than by grepping prose. The subject is a reference to the entity acted upon — a claim id or a tokenized policyholder handle, never the raw personal data itself. The correlation id threads every event for one claim together, which is what makes single-claim reconstruction a filter rather than a forensic exercise. And the previous hash binds this record to the one before it, the mechanism explored in depth in the guide on building append-only hash-chained audit logs.

Copy
from __future__ import annotations

import uuid
from datetime import datetime, timezone
from decimal import Decimal
from enum import Enum
from typing import Literal, Union

import structlog
from pydantic import BaseModel, ConfigDict, Field

log = structlog.get_logger("audit.schema")


def _utc_now() -> datetime:
    return datetime.now(timezone.utc)


class Action(str, Enum):
    """Controlled vocabulary of auditable actions. Extend additively only."""

    RECORD_VALIDATED = "record.validated"
    DEDUCTIBLE_APPLIED = "deductible.applied"
    REFERRAL_RAISED = "referral.raised"
    FIELD_MAPPED = "field.mapped"


class RecordValidatedPayload(BaseModel):
    """Typed body for a record.validated event."""

    kind: Literal["record.validated"] = "record.validated"
    schema_ref: str
    passed: bool
    violation_count: int = 0


class DeductibleAppliedPayload(BaseModel):
    kind: Literal["deductible.applied"] = "deductible.applied"
    threshold: Decimal
    loss_amount: Decimal
    verdict: Literal["CLEARS", "BELOW_DEDUCTIBLE"]
    boundary_mode: Literal["strict", "inclusive"]


class ReferralRaisedPayload(BaseModel):
    kind: Literal["referral.raised"] = "referral.raised"
    score: Decimal
    ruleset_version: str
    reason_codes: tuple[str, ...]


class FieldMappedPayload(BaseModel):
    kind: Literal["field.mapped"] = "field.mapped"
    source_field: str
    target_field: str
    confidence: Decimal


Payload = Union[
    RecordValidatedPayload,
    DeductibleAppliedPayload,
    ReferralRaisedPayload,
    FieldMappedPayload,
]


class AuditEvent(BaseModel):
    """Immutable, append-only audit envelope written by every pipeline stage.

    Frozen so that once constructed an event's identity and content cannot be
    mutated in memory before it reaches the writer — a precondition for the
    deterministic replay auditors expect.
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    event_id: uuid.UUID = Field(default_factory=uuid.uuid4)
    occurred_at: datetime = Field(default_factory=_utc_now)
    schema_version: str = "2026.1"
    actor: str
    action: Action
    subject_ref: str          # claim id / tokenized handle — never raw PII
    correlation_id: str
    prev_hash: str            # hex sha256 of the preceding record; genesis = 64 zeros
    payload: Payload = Field(discriminator="kind")

The discriminator="kind" on the payload field is what makes the typed union work: Pydantic reads the kind literal and validates against exactly one payload model, so a malformed deductible.applied body can never masquerade as a valid event. extra="forbid" means an unexpected field is a hard validation error rather than a silently retained blob — critical when the envelope is your evidentiary record.

The writer is the only component permitted to add to the ledger, and it does exactly three things: it reads the current tail hash, it seals the event by computing this record’s hash over the previous hash plus the canonical body, and it appends. It never updates or deletes. Every failure path raises an explicit, named exception rather than returning a sentinel, because a silently dropped audit write is indistinguishable from an event that never happened.

Copy
from __future__ import annotations

import hashlib
import json
from pathlib import Path
from typing import Final

GENESIS_HASH: Final[str] = "0" * 64


class AuditWriteError(RuntimeError):
    """Raised when an audit event cannot be durably appended."""


class ChainContinuityError(AuditWriteError):
    """Raised when the event's prev_hash does not match the ledger tail."""


def _canonical_body(event: AuditEvent) -> bytes:
    """Deterministic JSON of everything except the record's own hash.

    Sorted keys, no insignificant whitespace, and Decimals/UUIDs/datetimes
    rendered as stable strings so the same event always hashes identically.
    """
    body = event.model_dump(mode="json", exclude={"prev_hash"})
    return json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8")


def seal(event: AuditEvent) -> str:
    """Return the hex sha256 binding this event to its predecessor."""
    material = event.prev_hash.encode("ascii") + _canonical_body(event)
    return hashlib.sha256(material).hexdigest()


class AuditLedger:
    """Append-only writer over a newline-delimited JSON ledger."""

    def __init__(self, path: Path) -> None:
        self._path = path
        self._path.touch(exist_ok=True)

    def tail_hash(self) -> str:
        last = GENESIS_HASH
        with self._path.open("r", encoding="utf-8") as fh:
            for line in fh:
                if line.strip():
                    last = json.loads(line)["record_hash"]
        return last

    def append(self, event: AuditEvent) -> str:
        """Seal and durably append one event; return its record hash."""
        expected = self.tail_hash()
        if event.prev_hash != expected:
            log.error("audit.chain_break", expected=expected, got=event.prev_hash)
            raise ChainContinuityError(
                f"prev_hash {event.prev_hash[:12]}… != tail {expected[:12]}…"
            )
        record_hash = seal(event)
        row = {**event.model_dump(mode="json"), "record_hash": record_hash}
        line = json.dumps(row, sort_keys=True, separators=(",", ":"))
        try:
            with self._path.open("a", encoding="utf-8") as fh:
                fh.write(line + "\n")
                fh.flush()
        except OSError as exc:  # disk full, permission, etc.
            log.error("audit.write_failed", error=str(exc))
            raise AuditWriteError("could not append audit event") from exc
        log.info(
            "audit.appended",
            action=event.action.value,
            correlation_id=event.correlation_id,
            record_hash=record_hash[:12],
        )
        return record_hash

The exclude={"prev_hash"} in the canonical body is deliberate: the previous hash is fed into the digest as separate material, not as part of the serialized body, so the record’s hash covers both its content and its position in the chain without double-counting the link field.

An audit schema is a long-lived contract, so its evolution rules matter as much as its initial shape. Version additively. Every event carries schema_version, and new fields are added as optional with defaults; existing fields are never renamed or repurposed, because a verifier reading a five-year-old event must still interpret it under the version it was written with. When a genuinely incompatible change is unavoidable, introduce a new Action value and a new payload type rather than mutating an existing one — the union grows, it never rewrites.

Partition by time, keyed for retention. Regulatory retention windows differ by record class and jurisdiction — many state insurance regulations require claims records to be retained for several years past close, and some lines demand longer. Partitioning the ledger by month (or by a retention_class derived from line of business) lets you expire whole partitions when their window lapses without ever performing a row-level delete that would fracture the hash chain. Retention is enforced at the partition boundary; within a partition, the log stays strictly append-only.

Copy
from datetime import date


def partition_key(event: AuditEvent, retention_class: str) -> str:
    """Route an event to a monthly, retention-tagged partition.

    e.g. 'claims-standard/2026-07' — whole partitions expire together so
    retention never requires an in-chain deletion.
    """
    month = event.occurred_at.date().replace(day=1)
    return f"{retention_class}/{month.isoformat()[:7]}"


assert partition_key.__name__ == "partition_key"
assert isinstance(date(2026, 7, 1), date)

Tune the flush and durability posture to the risk: for compliance-critical actions (a settlement, a referral, a reserve change) flush synchronously so the event is on disk before the action’s effect is acknowledged upstream; for high-volume, low-consequence telemetry, batch appends behind a bounded buffer. Never let audit-write latency become a reason to skip the write — the correct backpressure response is to slow the action, not to drop its record.

The tension at the center of audit design is that the log must be complete enough to prove what happened yet must not become a permanent honeypot of personal data. The resolution is reference, don’t embed. The envelope’s subject_ref holds a claim id or a tokenized handle that resolves to the policyholder only through the controlled path that Data Boundary Enforcement governs; the log records that a field was mapped, or that a referral was raised, without carrying the underlying name, SSN, or diagnosis. When a payload genuinely needs a sensitive value to be evidentiary — say, the last four digits of an account — capture the minimum discriminating fragment, never the whole.

This aligns with how the platform models field semantics against Policy Schema Design: the schema layer already knows which fields are classified as PII/PHI, and that classification is exactly what the audit serializer consults to decide whether a value is referenced or redacted before it is written. Events emitted when documents are normalized through field mapping strategies therefore record the mapping decision and its confidence rather than the raw extracted values, and fraud-side events emitted during SIU referral orchestration record the score and reason codes rather than the investigative narrative. The audit log proves the decision was made, at what version, by whom — which is precisely what a NAIC market-conduct examiner needs — without duplicating regulated data into an immutable store.

Because every event is a pure statement of fact keyed by correlation id, a claim’s entire history is a filter-and-fold: select all events for the correlation id, order by the chain, and apply each to a fresh state accumulator. This is how you answer “why was this claim settled the way it was” without querying — or trusting — the live policy service, whose state has since moved on.

Copy
from typing import Iterable


def replay_claim(events: Iterable[dict[str, object]]) -> dict[str, object]:
    """Fold ordered audit rows into the derived state they produced.

    Pure over its inputs: the same event sequence always yields the same
    reconstructed state, which is what makes it defensible in an audit.
    """
    state: dict[str, object] = {"validated": False, "verdict": None, "referred": False}
    for row in events:
        action = row["action"]
        payload = row["payload"]
        if action == Action.RECORD_VALIDATED.value:
            state["validated"] = bool(payload["passed"])  # type: ignore[index]
        elif action == Action.DEDUCTIBLE_APPLIED.value:
            state["verdict"] = payload["verdict"]          # type: ignore[index]
        elif action == Action.REFERRAL_RAISED.value:
            state["referred"] = True
    return state

• Non-deterministic hashes across producers. Symptom: the verifier reports a break on records that were never tampered with. Cause: two producers serialized the same logical event differently — one included a field the other omitted, or a float rendered differently across platforms. Fix: route every write through the single _canonical_body function with sorted keys and string-rendered Decimal, and forbid float in payloads at the schema layer.
• Naive timestamps. Symptom: events appear out of order or cross a day boundary into the wrong retention partition. Cause: datetime.now() without timezone.utc. Fix: generate all timestamps with the _utc_now factory; reject naive datetimes in validation.
• Silent audit-write drop. Symptom: a claim’s history is missing a step everyone remembers happening. Cause: an audit append that failed was swallowed by a bare except. Fix: let AuditWriteError propagate and apply backpressure to the action; an action whose audit record cannot be written must not be treated as complete.
• PII leaking into the ledger. Symptom: a data-subject deletion request cannot be honored because personal data sits in an immutable store. Cause: a payload embedded raw personal fields. Fix: reference by token, redact at serialization using the schema’s PII classification, and keep resolvable identifiers behind the boundary layer.
• Retention delete fractures the chain. Symptom: verification fails after a records-management job runs. Cause: row-level deletion of expired events broke the prev-hash linkage. Fix: never delete within a chain; partition by retention class and expire whole partitions, sealing each partition’s terminal hash beforehand.

This component has a focused companion guide that implements the tamper-evidence mechanism end to end: building append-only hash-chained audit logs walks through the sha256 linkage, the canonical-serialization pitfalls that silently break a chain, a verifier that detects tampering, and the pytest suite that proves it — the concrete implementation of the envelope and writer sketched above.

• Core Architecture & Compliance Mapping — the parent domain defining the schema contracts this log serves
• Building Append-Only Hash-Chained Audit Logs — the tamper-evident hash chain and its verifier in full
• Data Boundary Enforcement — where subject references resolve and PII classification lives
• Policy Schema Design — the versioned contracts whose validation events feed the log
• SIU Referral Orchestration — a producer whose referral decisions are written to this ledger