FSMA 204 TraceabilityCompliance automation with Python

Setting Up FSMA 204 Data Retention That Survives an FDA Audit

The task looks simple: schedule a job that keeps traceability records for two years and archives the rest. The failure it produces in production is not simple. The most damaging outcome of a naive retention setup is the silent archival of incomplete records — a payload with a timezone-naive creation_date or a missing traceability_lot_number slips past a permissive filter, lands in cold storage looking healthy, and only surfaces months later as a ValidationError the first time the FDA runs a Subpart S traceback against that partition. By then the record is unrecoverable and the audit is already failing.

This guide sets up a retention ingestion boundary that refuses to archive a record it cannot prove is complete and correctly timestamped. It intercepts malformed Key Data Elements (KDEs) before they enter the retention queue, normalizes every timestamp to UTC, and routes anything ambiguous to quarantine instead of deleting or blindly archiving it. It is the concrete, single-scheduler version of the lifecycle engine described in the parent Data Retention Policies guide.

Root Cause: Why Retention Setups Fail an Audit

The retention window under FSMA 204 (21 CFR Part 1, Subpart S) is computed per record, from the timestamp of the Critical Tracking Event it documents. That single fact is where most setups break, because the expiration math is only ever as trustworthy as the fields feeding it. Three compounding defects account for nearly every retention finding an auditor raises:

  1. Timezone-naive timestamps. A supplier feed delivers creation_date as "2024-03-15T08:30:00" with no offset. A naive scheduler computes the 730-day cutoff against server-local time, so the same record is “expired” in one region and “active” in another. Archival and purge become non-deterministic, and chronological reconstruction of a lot’s lineage silently reorders events.
  2. Absent mandatory KDEs. A record missing traceability_lot_number or transformation_input_traceability_lot_number passes a filter that only checks for a non-null primary key. It archives cleanly, but the moment an FDA query joins on the lot code, the row cannot participate in the traceback and the query raises rather than returns.
  3. Unhandled null states in the retention engine itself. A misconfigured TTL, an epoch-vs-ISO format mix, or a cutoff comparison against a None timestamp throws mid-batch. Whatever records had not yet been processed are left in an indeterminate state — some archived, some not — with no clean way to prove which.

Because these defects originate upstream, the correct place to stop them is the ingestion boundary, before a record is ever queued for retention. If the Supplier Data Ingestion feed hands the scheduler a naive datetime, the fix is to reject it at the door — not to guess an offset and archive it anyway.

Retention lifecycle of a single KDE record, from ingestion to purge A left-to-right timeline with three phase bands — active hot storage up to Day 730, checksum-verified WORM cold storage just past the 730-day FDA floor, and eligible-for-purge beyond. A dashed callout shows that a retention_hold freezes the record at any point, pausing both archival and purge. retention_hold freezes the record at any point litigation hold or open recall — archival and purge pause 730-day FDA floor Active in hot storage WORM cold storage Eligible for live · queryable · indexed checksum-verified archival purge Day 0 Day 730 >730 days ingestion + validation cutoff computed per record post-retention window

Minimal Reproducible Example: The Silent-Archival Failure

The snippet below reproduces the failure in isolation. A permissive retention filter keeps anything whose creation_date looks recent and whose primary key is present. Two realistic supplier payloads defeat it: one carries a naive timestamp, the other is missing a mandatory KDE. Both are waved through to the archive, and the failure only appears when the audit query runs.

from datetime import datetime, timedelta

# Two payloads as they actually arrive from heterogeneous supplier feeds.
incoming = [
    {  # naive timestamp: no UTC offset — the cutoff math is ambiguous
        "record_id": "cte-9001",
        "traceability_lot_number": "LOT-2024-03-15-AA1",
        "creation_date": "2024-03-15T08:30:00",
        "product_description": "romaine, chopped",
    },
    {  # missing traceability_lot_number — unusable in a Subpart S traceback
        "record_id": "cte-9002",
        "creation_date": "2024-03-15T08:31:00+00:00",
        "product_description": "romaine, chopped",
    },
]

def naive_retention_filter(rec: dict) -> bool:
    # BUG 1: parses a naive string against server-local time.
    # BUG 2: only checks record_id, never the mandatory KDEs.
    created = datetime.fromisoformat(rec["creation_date"].replace("Z", ""))
    return bool(rec.get("record_id")) and created > datetime.now() - timedelta(days=730)

archived = [r for r in incoming if naive_retention_filter(r)]
# Both records archive "successfully" here.

# ...months later, the FDA 24-hour traceback query runs:
for r in archived:
    lot = r["traceability_lot_number"]   # KeyError on cte-9002
    print(f"tracing {lot} from {datetime.fromisoformat(r['creation_date'])}")

Running this archives both records, then raises KeyError: 'traceability_lot_number' on cte-9002 at query time — exactly the class of error that presents as a ValidationError during real Subpart S execution. The naive record (cte-9001) is worse: it does not raise, it just sorts into the wrong position in the event chain.

Fix Implementation: A Validation Boundary With a Circuit Breaker

The fix moves all trust to the ingestion boundary. A pydantic v2 model rejects any payload missing a mandatory KDE and coerces every creation_date to timezone-aware UTC — a naive value is refused, never silently localized. A CircuitBreaker halts the pipeline after a run of consecutive failures so a degraded upstream feed cannot flood cold storage with garbage. Nothing is ever dropped: non-compliant payloads are serialized to a quarantine directory with explicit failure metadata, preserving an immutable trail of ingestion anomalies.

import json
import logging
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Optional

from pydantic import BaseModel, ValidationError, field_validator

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S%z",
)
logger = logging.getLogger("fsma204.retention_setup")

RETENTION_DAYS = 730  # FDA two-year floor; never lower this below the mandate.


class RetentionRecord(BaseModel):
    """The narrow contract the retention scheduler needs to trust a record."""

    record_id: str
    traceability_lot_number: str
    product_description: str
    creation_date: datetime
    transformation_input_traceability_lot_number: Optional[str] = None
    location_id: Optional[str] = None

    @field_validator("creation_date", mode="before")
    @classmethod
    def coerce_utc(cls, v: Any) -> datetime:
        # Accept epoch or ISO string, but REJECT a naive datetime rather than
        # guess an offset — a wrong offset silently corrupts the retention clock.
        if isinstance(v, (int, float)):
            return datetime.fromtimestamp(v, tz=timezone.utc)
        if isinstance(v, str):
            dt = datetime.fromisoformat(v.replace("Z", "+00:00"))
        elif isinstance(v, datetime):
            dt = v
        else:
            raise ValueError(f"unsupported creation_date type: {type(v)!r}")
        if dt.tzinfo is None or dt.tzinfo.utcoffset(dt) is None:
            raise ValueError("creation_date must be timezone-aware (UTC offset required)")
        return dt.astimezone(timezone.utc)

    @field_validator("traceability_lot_number")
    @classmethod
    def lot_not_blank(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("traceability_lot_number cannot be blank")
        return v.strip()


@dataclass
class CircuitBreaker:
    failure_threshold: int = 5
    consecutive_failures: int = 0
    is_open: bool = False

    def record_failure(self) -> None:
        self.consecutive_failures += 1
        if self.consecutive_failures >= self.failure_threshold:
            self.is_open = True
            logger.critical("Circuit breaker OPEN: consecutive validation failures exceeded threshold.")

    def record_success(self) -> None:
        if self.is_open:
            logger.info("Circuit breaker CLOSED: pipeline stability restored.")
        self.consecutive_failures = 0
        self.is_open = False


class RetentionIngestBoundary:
    def __init__(self, quarantine_dir: Path = Path("/var/log/fsma204/quarantine")) -> None:
        self.quarantine_dir = quarantine_dir
        self.quarantine_dir.mkdir(parents=True, exist_ok=True)
        self.breaker = CircuitBreaker()

    def _quarantine(self, payload: dict[str, Any], reason: str) -> None:
        ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%f")
        path = self.quarantine_dir / f"quarantined_{ts}.json"
        path.write_text(json.dumps(
            {"original_payload": payload, "failure_reason": reason, "quarantined_at": ts},
            indent=2,
        ))
        logger.info("Record quarantined | reason=%s | file=%s", reason, path)

    def accept(self, payload: dict[str, Any]) -> Optional[RetentionRecord]:
        """Validate one payload at the boundary. Returns a queue-ready record or None."""
        if self.breaker.is_open:
            logger.error("Pipeline halted: circuit breaker active. Manual intervention required.")
            return None

        try:
            record = RetentionRecord(**payload)
        except ValidationError as e:
            self.breaker.record_failure()
            reason = "; ".join(f"{err['loc'][0]}:{err['type']}" for err in e.errors())
            self._quarantine(payload, f"schema_invalid ({reason})")
            return None

        # A record already older than the window should have been archived upstream;
        # its arrival here signals a replay or a misconfigured scheduler, not a purge trigger.
        cutoff = datetime.now(timezone.utc) - timedelta(days=RETENTION_DAYS)
        if record.creation_date < cutoff:
            self._quarantine(payload, "outside_active_retention_window")
            self.breaker.record_success()
            return None

        self.breaker.record_success()
        logger.info(
            "Record accepted to retention queue | lot=%s | created_utc=%s",
            record.traceability_lot_number, record.creation_date.isoformat(),
        )
        return record

Figure — KDE payload validation and fallback flow:

KDE payload validation and fallback flow at the ingestion boundary A top-down decision spine of four gates — circuit breaker, mandatory-KDE presence, timestamp-to-UTC normalization, and the 730-day window. The circuit-breaker branch halts the pipeline; every other failure branch routes right into one shared Quarantine queue. Only a fully valid record reaches the retention queue. Incoming KDE payload Halt — manual intervention Route to retention queue Quarantine queue serialized with failure_reason metadata Circuit breakeropen? Mandatory KDEspresent? Timestamp → UTCvalid? Within 730-daywindow? no complete valid active yes · breaker open missing fields drift / malformed expired

Three decisions in this code are load-bearing for the audit. The mode="before" validator refuses a naive creation_date instead of localizing it, which is what keeps the retention clock deterministic across regions. The mandatory-field contract is enforced by pydantic’s required-field mechanics, so a missing traceability_lot_number fails validation before archival rather than at query time. And the quarantine path fails closed — an ambiguous record is preserved with its reason, never dropped and never archived on a guess.

Verification Steps

Confirm the boundary works before you point it at production storage. Feed it the two payloads from the reproduction and one clean record, then check three independent signals.

1. Log output. A correctly configured boundary emits one acceptance and two quarantines:

2026-07-02T09:14:02+0000 | INFO     | fsma204.retention_setup | Record quarantined | reason=schema_invalid (creation_date:value_error) | file=/var/log/fsma204/quarantine/quarantined_20260702T091402...json
2026-07-02T09:14:02+0000 | INFO     | fsma204.retention_setup | Record quarantined | reason=schema_invalid (traceability_lot_number:missing) | file=...
2026-07-02T09:14:02+0000 | INFO     | fsma204.retention_setup | Record accepted to retention queue | lot=LOT-2024-06-01-BB2 | created_utc=2024-06-01T12:00:00+00:00

2. Unit assertions. Encode the invariants so a regression cannot ship silently:

def test_naive_timestamp_is_rejected() -> None:
    boundary = RetentionIngestBoundary(quarantine_dir=tmp_path)
    result = boundary.accept({
        "record_id": "cte-9001",
        "traceability_lot_number": "LOT-2024-03-15-AA1",
        "product_description": "romaine, chopped",
        "creation_date": "2024-03-15T08:30:00",  # no offset
    })
    assert result is None                       # never queued
    assert list(tmp_path.glob("quarantined_*.json"))  # trail preserved

def test_missing_kde_never_queues() -> None:
    boundary = RetentionIngestBoundary(quarantine_dir=tmp_path)
    assert boundary.accept({
        "record_id": "cte-9002",
        "creation_date": "2024-03-15T08:31:00+00:00",
    }) is None

3. SQL state check. After a real run, no archived record may sit inside the window without a timezone-aware timestamp. This query must return zero rows:

SELECT record_id, creation_date
FROM fsma204_retention_archive
WHERE creation_date > (now() AT TIME ZONE 'UTC') - INTERVAL '730 days'
  AND (creation_date IS NULL OR traceability_lot_number IS NULL);

Any row returned is a record that reached the archive without a valid clock or lot code — the exact defect this setup exists to prevent.

  • Daylight-saving drift from legacy pytz. If any upstream stage still localizes with pytz rather than the standard-library zoneinfo, historical timestamps near a DST boundary can shift by an hour and land a record in the wrong retention week. Normalize with zoneinfo end to end.
  • Lot splits that drop the retention anchor. When a lot is split or transformed, the child records must inherit the parent’s creation_date as their retention anchor. A split that re-stamps the child with the transformation time resets the two-year clock and can hold a record past its lawful window — verify against the KDE Field Mapping Guide propagation rules.
  • Replayed payloads outside the window. A retry storm from a flaky feed can re-deliver already-archived records. The boundary quarantines these as outside_active_retention_window, but you should also confirm the delivery layer is backing off correctly — the discipline in API Polling Strategies keeps a transient fault from being mistaken for a stream of retention failures.

Up: Data Retention Policies — this how-to configures the ingestion boundary in front of that cluster’s retention lifecycle engine.

Related content

Up FSMA 204 Data Retention Policies