Structured Audit Log Export for FDA Submission

When an FDA investigator asks who touched a traceability record and when, an approximate answer is a finding. The FSMA 204 Architecture & KDE Compliance Mapping program stores validated Critical Tracking Events on an immutable ledger, but the ledger alone does not prove custody — it proves content. What proves custody is a second, orthogonal stream: an append-only audit log that records every read, mutation, and export of Key Data Elements (KDEs), and that can be sliced by retention window and rendered into the exact columnar layout the FDA expects to receive. This guide defines that export component — the audit-event schema, a deterministic CSV/JSON exporter aligned to FDA submission columns, and the retention-window queries that pull the right records without scanning the whole history.

The requirement is narrow but unforgiving. During an inspection you must be able to demonstrate, for any lot under scope, a complete and tamper-evident account of who accessed the underlying KDEs, what changed, and which exports left the system — for the full two-year window that the Data Retention Policies guide governs. A structured export is only defensible if it is reproducible: the same query over the same data must always emit byte-identical rows in a stable order, so an investigator can re-run it and see the same result. This page builds that exporter on top of the append-only audit stream, pairs it with the tamper-evidence guarantees from the Append-Only Ledger & Hash Chaining guide, and feeds the recall workflow’s FDA 24-Hour Response Automation so the same audit evidence can be attached to a traceability record request.

Export Architecture: From Audit Store to Submission Package

The export pipeline is a linear, side-effect-light transformation with a strict boundary at each stage. Audit events are written once, at the moment a subject acts on a KDE, into an append-only store partitioned by month. When an export is requested, the retention-window filter selects only the events whose event_timestamp falls inside the two-year floor (or a narrower recall scope), touching only the partitions that can contain matching rows. Selected events are then mapped field-by-field onto the FDA submission template — a fixed column order with fixed header names — and serialized to CSV or JSON with deterministic ordering. The final artifact is a self-contained submission package: the rows, a manifest of the query that produced them, and the ledger hash that anchors them.

The design keeps three concerns separate so each can be verified in isolation: selection (which events belong in this export), mapping (which internal field becomes which FDA column), and serialization (how the rows are rendered to bytes). A defect in any one of them is diagnosable without re-reading the whole pipeline, and the mapping stage in particular is where most template-mismatch findings originate — a reordered column or a renamed header is enough to fail schema validation on the FDA side.

Data Contract: The Audit Event Schema

An audit event answers five questions — who, what, when, which record, and what changed — plus the integrity fields that make the answer provable. Every field below is mandatory on the export; a null in any load-bearing column is a quarantine trigger, not a blank cell in the FDA file. The Regulatory Source column cites the Subpart S provision that makes each field necessary for a defensible submission.

Two rules keep the export defensible. First, event_timestamp is validated as timezone-aware and normalized to UTC at the boundary, so the retention-window filter never compares a naive datetime against the two-year cutoff. Second, prev_hash links each event to its predecessor, so an investigator can confirm that no audit row was silently removed between two exports — the same tamper-evidence property the Append-Only Ledger & Hash Chaining guide applies to the KDE ledger itself.

Production Implementation: Audit Record Model and Exporter

The implementation below models an audit event with pydantic v2, then serializes a filtered batch to CSV or JSON with a single, shared column order so the two formats never disagree. The exporter is deterministic: events are sorted by (event_timestamp, audit_id) before rendering, the CSV writer quotes every field to survive commas and newlines in payloads, and every export decision is written to a structured log that itself becomes an audit event. A record missing a load-bearing field is routed to quarantine rather than emitted as a partial row.

import csv
import hashlib
import io
import json
import logging
from datetime import datetime, timezone
from enum import Enum
from typing import Optional

from pydantic import BaseModel, ConfigDict, ValidationError, field_validator

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    handlers=[logging.StreamHandler()],
)
logger = logging.getLogger("fsma204.audit_export")

# The FDA submission template: internal field -> exact output column header.
# Order is fixed; both CSV and JSON honour it so the two formats never diverge.
FDA_COLUMN_MAP: list[tuple[str, str]] = [
    ("event_timestamp", "Event Date/Time (UTC)"),
    ("traceability_lot_code", "Traceability Lot Code"),
    ("target_record_id", "Reference Record"),
    ("action", "Action Type"),
    ("actor_id", "Performed By"),
    ("field_changed", "Field Changed"),
    ("checksum", "Record Checksum"),
]


class AuditAction(str, Enum):
    READ = "READ"
    CREATE = "CREATE"
    UPDATE = "UPDATE"
    DELETE = "DELETE"
    EXPORT = "EXPORT"


class MissingFieldError(Exception):
    """A load-bearing audit field is null — the row is quarantined, not emitted."""


class AuditEvent(BaseModel):
    model_config = ConfigDict(extra="forbid")

    audit_id: str
    event_timestamp: datetime
    actor_id: str
    action: AuditAction
    target_record_id: str
    traceability_lot_code: str
    field_changed: Optional[str] = None
    prev_hash: str
    checksum: Optional[str] = None

    @field_validator("event_timestamp", mode="before")
    @classmethod
    def enforce_utc(cls, v: datetime) -> datetime:
        # A naive timestamp makes the retention-window filter ambiguous;
        # reject it at the boundary rather than assuming an offset.
        if isinstance(v, datetime):
            if v.tzinfo is None or v.tzinfo.utcoffset(v) is None:
                raise ValueError("event_timestamp must be timezone-aware (UTC required)")
            return v.astimezone(timezone.utc)
        return v

    def compute_checksum(self) -> str:
        identity = (
            f"{self.audit_id}|{self.actor_id}|{self.action.value}"
            f"|{self.target_record_id}|{self.event_timestamp.isoformat()}"
        )
        return hashlib.sha256(identity.encode("utf-8")).hexdigest()

    def to_template_row(self) -> dict[str, str]:
        """Project the event onto the FDA columns, in template order."""
        row: dict[str, str] = {}
        for field, header in FDA_COLUMN_MAP:
            value = getattr(self, field)
            if value is None and field not in ("field_changed",):
                raise MissingFieldError(f"{self.audit_id}: required field {field} is null")
            if isinstance(value, datetime):
                value = value.isoformat()
            elif isinstance(value, Enum):
                value = value.value
            row[header] = "" if value is None else str(value)
        return row


class AuditExporter:
    HEADERS: list[str] = [header for _, header in FDA_COLUMN_MAP]

    def __init__(self, quarantine_client):
        self.quarantine = quarantine_client

    def _prepare(self, events: list[AuditEvent]) -> list[dict[str, str]]:
        """Validate, project, and deterministically order the batch."""
        # Stable ordering: newest-relevant sort key that a re-run reproduces.
        ordered = sorted(events, key=lambda e: (e.event_timestamp, e.audit_id))
        rows: list[dict[str, str]] = []
        for event in ordered:
            try:
                AuditEvent.model_validate(event.model_dump())
                rows.append(event.to_template_row())
            except (ValidationError, MissingFieldError) as exc:
                logger.error("Excluded from export | audit_id=%s | error=%s",
                             event.audit_id, exc)
                self.quarantine.route(event.audit_id, reason="incomplete_audit_row")
        logger.info("Prepared export batch | included=%d | excluded=%d",
                    len(rows), len(events) - len(rows))
        return rows

    def to_csv(self, events: list[AuditEvent]) -> str:
        rows = self._prepare(events)
        buffer = io.StringIO()
        # QUOTE_ALL survives commas, quotes, and newlines inside KDE payloads.
        writer = csv.DictWriter(buffer, fieldnames=self.HEADERS,
                                quoting=csv.QUOTE_ALL, lineterminator="\n")
        writer.writeheader()
        writer.writerows(rows)
        return buffer.getvalue()

    def to_json(self, events: list[AuditEvent]) -> str:
        rows = self._prepare(events)
        # sort_keys=False keeps the FDA column order; ensure_ascii keeps it portable.
        return json.dumps({"columns": self.HEADERS, "rows": rows},
                          indent=2, sort_keys=False, ensure_ascii=True)


if __name__ == "__main__":
    exporter = AuditExporter(quarantine_client=None)
    events = [
        AuditEvent(
            audit_id="aud-0001",
            event_timestamp=datetime(2024, 5, 3, 14, 22, tzinfo=timezone.utc),
            actor_id="user:kmorales",
            action=AuditAction.EXPORT,
            target_record_id="cte-7781",
            traceability_lot_code="LOT-2024-05-03-QK2",
            field_changed=None,
            prev_hash="0" * 64,
        )
    ]
    for e in events:
        e.checksum = e.compute_checksum()
    print(exporter.to_csv(events))

The exporter never emits a partial row. _prepare re-validates each event against the schema, projects it through FDA_COLUMN_MAP, and sorts the survivors by a stable key so two runs over the same batch produce identical bytes. Because CSV and JSON both read FDA_COLUMN_MAP, a change to the template propagates to both formats from a single edit — the class of defect the FDA template export guide walks through fixing.

Error Handling and Quarantine Strategy

An audit export fails closed. A row that cannot be rendered completely is excluded and logged, never guessed at, because a submission with a fabricated or blank required cell is worse than a submission that is transparently short one record. Three failure classes route to quarantine, each with a distinct reason code so a compliance operator can reconcile it:

• Missing required field — a null actor_id, target_record_id, or traceability_lot_code raises MissingFieldError. The event is quarantined with reason incomplete_audit_row; a partial row is never written to the FDA file. The gap itself is evidence that the upstream write path dropped a field, which is a defect to fix, not to paper over.
• Schema-invalid event — a naive event_timestamp, an unknown action, or an extra field (blocked by extra="forbid") fails pydantic revalidation. These are quarantined the same way and surfaced with the full raw event so the ingestion path can be corrected.
• Broken hash chain — if an event’s prev_hash does not match the checksum of its predecessor when the batch is verified, the export is halted rather than shipped. A break means an audit row was removed or reordered, which invalidates the custody claim until it is reconciled through the Fallback Routing Logic reconciliation path.

Because every quarantine decision is itself logged as a structured event, the exclusion list is auditable: an investigator can see not only what was submitted but exactly which rows were withheld and why. That transparency is what separates a defensible short export from a suspicious one.

Integration With the Parent Architecture

This export component sits alongside the immutable KDE ledger, not inside it. The FSMA 204 Architecture & KDE Compliance Mapping reference defines where validated KDEs land; the audit stream records every subsequent interaction with those KDEs, and this page defines how that stream is queried and rendered for the FDA. The dependency runs one way: the exporter consumes audit events that the access layer has already written, and it trusts the field contract that layer guarantees. The retention window it filters on is the same two-year floor the Data Retention Policies engine enforces, so an export can never be asked to produce an event that retention has already purged — archived events remain reachable precisely so this exporter can still reach them.

Downstream, the same exporter feeds the recall workflow. When a lot is scoped for an FDA request, the FDA 24-Hour Response Automation attaches the audit export for the affected lots to the traceability record package, so the custody trail and the KDE content ship together inside the 24-hour window. The tamper-evidence anchor — the prev_hash chain and the ledger hash in the manifest — comes from the Append-Only Ledger & Hash Chaining guide, which is what lets an investigator verify that the exported rows were not edited after the fact.

Operational Notes

• Runtime and dependencies. Python 3.10+, pydantic>=2.6. The csv, json, hashlib, io, and logging modules are standard library. Pin versions in a lockfile so the exact bytes an export produces are reproducible across environments during an inspection.
• Configuration variables. Keep FDA_COLUMN_MAP in version control as the single source of truth for template layout; never hardcode headers at more than one site. Expose the audit-store connection, the quarantine table name, and the export bucket as environment variables.
• Determinism. Always sort by the same key before serializing, and always quote every CSV field. An export that changes row order or quoting between runs cannot be re-verified by an investigator and undermines the custody claim.
• UTC everywhere. Store and export timestamps in UTC with an explicit offset. The retention-window filter and the Event Date/Time (UTC) column both assume it; a local-time export silently shifts the window boundary.
• Manifest. Ship every export with a manifest recording the query parameters, the row count, the excluded count, and the ledger hash, so the export is self-describing when it is opened months later.

Frequently Asked Questions

Conclusion

A structured audit-log export turns a pile of access records into inspection-ready evidence. By modeling each audit event with pydantic v2, projecting it through a single FDA column map, serializing CSV and JSON deterministically, and filtering by the two-year retention window, teams can produce a reproducible custody trail for any lot on demand. The discipline that makes it defensible is the same throughout: normalize to UTC, order deterministically, fail closed on missing fields, and anchor the sequence with a hash chain so no row can vanish unseen. Built this way, the export is not a scramble during an inspection — it is a query.

Related

• Exporting audit logs to FDA submission templates — mapping internal rows to the exact FDA columns and emitting a validated CSV
• Retention-period query patterns for audit logs — selecting the two-year window across partitioned tables without full scans
• Data Retention Policies — the two-year floor the export window filters against
• Append-Only Ledger & Hash Chaining — the tamper-evidence anchor the audit chain reuses
• FDA 24-Hour Response Automation — the recall workflow that attaches this export to a traceability record request

Up: FSMA 204 Architecture & KDE Compliance Mapping — this export component renders the audit trail that sits alongside the parent architecture’s immutable KDE ledger.