Resolving FSMA 204 KDE Schema Drift and Silent Type Coercion in Supplier Ingestion Pipelines

The failure this page fixes is specific and destructive: during the initial Supplier Onboarding Automation phase, a supplier’s first CSV or EDI payload passes ingestion “successfully” and still corrupts mandatory Key Data Elements (KDEs) on the way in. The two coercion patterns that dominate production incidents are date_received timezone stripping and traceability_lot_code zero-padding loss. Standard pandas or csv readers default to locale-aware parsing and aggressive type inference: they convert an ISO 8601 string like 2024-03-15T14:30:00-05:00 into a naive UTC datetime, and they numeric-cast an alphanumeric lot code so leading zeros vanish. When those altered records enter the compliance ledger, downstream audit trails report false-negative Critical Tracking Events (CTEs), and FDA Subpart S inspection readiness degrades immediately — with no error line to point at.

The root cause is rarely a missing field. It is schema drift colliding with permissive parser defaults. Resolving it requires strict schema validation enforced at the ingestion boundary, coupled with a quarantine-first architecture that stops one malformed batch from blocking the main processing queue. This page reproduces the failure with realistic payloads, shows the corrected pydantic v2 gate, and gives the log, test, and SQL checks that prove the fix holds.

Root Cause: Why Coercion Slips Past “Successful” Ingestion

The FSMA Section 204 Final Rule mandates exact preservation of traceability lot codes and event timestamps for every CTE. Yet production ingestion layers routinely treat incoming supplier payloads as loosely typed data frames, and two failure modes follow from that single design choice:

1. Timezone stripping. When a supplier submits 2024-11-02T09:15:00-07:00, a naive datetime.strptime or pandas.to_datetime call may reparse it as 2024-11-02 16:15:00+00:00 or drop the offset entirely. For compliance, the exact local receipt time and its offset are legally material — converting to naive UTC destroys the original CTE context that a traceback relies on.
2. Zero-padding loss. Lot identifiers like 00482A or LOT-0091 are frequently ingested as numeric types by default CSV parsers. Leading zeros disappear, turning 00482A into a parse error and 0091 into 91. When downstream systems match these against supplier Certificates of Analysis (COAs), the mismatch surfaces as a compliance gap.

These are not data-entry mistakes; they are architectural oversights at the ingestion boundary. The canonical type and constraint for each field the gate enforces lives in the KDE Field Mapping Guide; the transports that deliver these payloads — flat files, EDI, and the REST feeds described in API Polling Strategies — all funnel through the same gate. The fix requires explicit, fail-fast validation that rejects implicit coercion before any record touches the ledger.

Reproducing the Failure

The minimal reproducible example below reads a realistic two-row supplier batch with a permissive parser and prints the corrupted result. No custom code is required to trigger the defect — the parser defaults are enough:

import io
import pandas as pd

# A realistic supplier CSV: alphanumeric lot with leading zeros,
# and an offset-aware receipt timestamp.
raw = io.StringIO(
    "traceability_lot_code,date_received,quantity_value\n"
    "00482A,2024-11-02T09:15:00-07:00,120\n"
    "0091,2024-03-15T14:30:00-05:00,64\n"
)

df = pd.read_csv(raw, parse_dates=["date_received"])

for row in df.itertuples(index=False):
    print(repr(row.traceability_lot_code), "|", repr(row.date_received))

# Output:
#   '00482A' | Timestamp('2024-11-02 16:15:00')   <- offset stripped, now naive UTC
#   91       | Timestamp('2024-03-15 19:30:00')   <- '0091' coerced to int 91

The second lot code 0091 is silently promoted to the integer 91, and both timestamps lose their offset and become naive. Nothing raises. The batch reports two rows ingested, and the compliance ledger now disagrees with the supplier’s source document — the worst possible outcome under a Subpart S traceback, because it is invisible until an inspector asks for the record.

The Fix: Strict KDE Validation at the Boundary

Replacing implicit type conversion with strict schema enforcement is the first line of defense. Pydantic v2 provides a production-grade mechanism to lock down KDE parsing, reject ambiguous formats, and preserve exact string representations. The model below enforces strict typing, forbids extraneous fields, and applies targeted validators to neutralize both coercion risks:

import logging
import time
from datetime import datetime
from typing import Any
from dataclasses import dataclass
from pydantic import BaseModel, Field, field_validator, ValidationError, ConfigDict

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("fsma204_ingestion")


class FSMA204KDE(BaseModel):
    model_config = ConfigDict(strict=True, extra="forbid")

    traceability_lot_identifier: str = Field(
        ...,
        min_length=4,
        max_length=50,
        description="Alphanumeric lot code; leading zeros preserved exactly as submitted",
    )
    product_description: str = Field(..., min_length=3)
    date_received: datetime = Field(
        ..., description="Timezone-aware CTE timestamp (ISO 8601)"
    )
    business_location_id: str = Field(..., pattern=r"^[A-Z0-9]{6,20}$")
    quantity_value: float = Field(..., gt=0)
    unit_of_measure: str = Field(..., pattern=r"^(EA|KG|LB|CASE|PALLET)$")

    @field_validator("traceability_lot_identifier", mode="before")
    @classmethod
    def enforce_string_coercion(cls, v: Any) -> str:
        if v is None:
            raise ValueError("traceability_lot_identifier cannot be null")
        # Explicitly cast to string to block pandas/numeric inference that
        # strips leading zeros from identifiers like "00482A" or "LOT-0091".
        raw = str(v).strip()
        if not raw:
            raise ValueError("traceability_lot_identifier cannot be empty after stripping")
        return raw

    @field_validator("date_received", mode="before")
    @classmethod
    def enforce_timezone_awareness(cls, v: Any) -> datetime:
        if isinstance(v, datetime):
            if v.tzinfo is None:
                raise ValueError(
                    "date_received must be timezone-aware; naive datetimes rejected"
                )
            return v
        if isinstance(v, str):
            try:
                dt = datetime.fromisoformat(v.replace("Z", "+00:00"))
            except ValueError as e:
                raise ValueError(f"Invalid ISO 8601 format: {v}") from e
            if dt.tzinfo is None:
                raise ValueError(
                    "Parsed datetime lacks timezone offset; naive datetimes rejected"
                )
            return dt
        raise ValueError(f"Unsupported type for date_received: {type(v)}")

Each decision in this model maps to a compliance-relevant constraint:

• strict=True disables Pydantic’s automatic type coercion (for example "12.5" to float), forcing explicit parsing so no field is quietly reinterpreted.
• extra="forbid" prevents suppliers from injecting undocumented fields that could mask schema drift instead of surfacing it.
• The date_received validator explicitly rejects naive datetimes, aligning with ISO 8601 requirements for unambiguous temporal tracking, and it must receive the raw string — feed pydantic the CSV value, not a pre-parsed pandas Timestamp, or the damage is already done upstream.
• Leading zeros are preserved by treating the lot identifier as an immutable string at the boundary.

Note that business_location_id uses an alphanumeric pattern ([A-Z0-9]{6,20}) for internal business system IDs, which may differ from GS1 GLNs. When this field maps to a GS1 GLN, the pattern should be ^\d{13}$ (13 numeric digits only), and the value should additionally pass a mod-10 check-digit test before it is trusted.

Quarantine-First Routing with a Circuit Breaker

Validation alone is insufficient if a malformed batch halts the entire ingestion service. A quarantine-first routing pattern paired with a lightweight circuit breaker isolates failures, maintains throughput for valid records, and prevents cascading degradation.

Figure — Quarantine-first ingestion flow:

@dataclass
class IngestionResult:
    valid_records: list[dict[str, Any]]
    quarantined: list[dict[str, Any]]
    errors: list[str]


class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, recovery_timeout: int = 300) -> None:
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time: float = 0.0
        self.state = "closed"  # closed, open, half-open

    def record_failure(self) -> None:
        self.failure_count += 1
        if self.failure_count >= self.failure_threshold:
            self.state = "open"
            self.last_failure_time = time.time()
            logger.warning(
                "Circuit breaker OPENED: consecutive validation failures exceeded threshold"
            )

    def record_success(self) -> None:
        self.failure_count = 0
        self.state = "closed"

    def allow_request(self) -> bool:
        if self.state == "closed":
            return True
        if self.state == "open":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "half-open"
                return True
            return False
        return True  # half-open permits one test payload


def ingest_supplier_batch(
    raw_batch: list[dict[str, Any]], breaker: CircuitBreaker
) -> IngestionResult:
    valid: list[dict[str, Any]] = []
    quarantined: list[dict[str, Any]] = []
    errors: list[str] = []

    for idx, record in enumerate(raw_batch):
        if not breaker.allow_request():
            logger.error("Pipeline halted: circuit breaker open. Skipping remaining records.")
            break

        try:
            parsed = FSMA204KDE(**record)
            valid.append(parsed.model_dump())
            breaker.record_success()
        except ValidationError as e:
            breaker.record_failure()
            first_error = e.errors()[0]
            logger.warning(
                "Record %d quarantined: %s (field: %s)",
                idx, first_error["msg"], first_error.get("loc"),
            )
            quarantined.append({
                "original_index": idx,
                "payload": record,
                "reason": f"{first_error['loc']}: {first_error['msg']}",
            })
        except Exception as e:
            breaker.record_failure()
            logger.error("Unexpected ingestion error at index %d: %s", idx, e)
            errors.append(f"Index {idx}: {e}")

    return IngestionResult(valid_records=valid, quarantined=quarantined, errors=errors)

This architecture guarantees three properties: valid KDEs flow immediately to the compliance ledger; malformed records are captured with exact diagnostic context and routed to a durable quarantine table or message queue for manual supplier reconciliation; and the circuit breaker halts ingestion when a supplier’s payload consistently violates schema rules, preventing resource exhaustion and alerting compliance teams to systemic drift.

Verifying the Fix

Do not trust a validation gate you have not watched reject a bad record. Three checks confirm the fix is live.

1. Log output. Feed the reproduction batch (raw strings, not pandas objects) through ingest_supplier_batch and confirm the naive-timestamp row is quarantined with an explicit reason rather than silently accepted:

2026-07-02 10:14:03,517 | WARNING | fsma204_ingestion | Record 1 quarantined: Value error, date_received must be timezone-aware; naive datetimes rejected (field: ('date_received',))

2. Unit test assertions. Pin both coercion defects as regression tests so a future dependency bump cannot reintroduce them:

def test_lot_code_preserves_leading_zeros() -> None:
    kde = FSMA204KDE(
        traceability_lot_identifier="00482A",
        product_description="Romaine, chopped",
        date_received="2024-11-02T09:15:00-07:00",
        business_location_id="WH0001",
        quantity_value=120.0,
        unit_of_measure="CASE",
    )
    assert kde.traceability_lot_identifier == "00482A"          # not "482A", not 482
    assert kde.date_received.utcoffset() is not None            # offset survived


def test_naive_timestamp_is_rejected() -> None:
    import pytest
    with pytest.raises(ValidationError, match="timezone-aware"):
        FSMA204KDE(
            traceability_lot_identifier="00482A",
            product_description="Romaine, chopped",
            date_received="2024-11-02T09:15:00",   # no offset
            business_location_id="WH0001",
            quantity_value=120.0,
            unit_of_measure="CASE",
        )

3. Ledger state query. After a production run, prove that no ledger row carries a naive timestamp or a numeric-looking lot code that lost its padding. If either query returns rows, a coercion path is still open upstream of the gate:

-- Any naive timestamp reached the ledger? Must return zero rows.
SELECT traceability_lot_identifier, date_received
FROM compliance_ledger
WHERE date_received::text NOT LIKE '%+%'
  AND date_received::text NOT LIKE '%-__:__';

-- Any lot code that looks numerically coerced (no non-digit char)? Investigate each.
SELECT traceability_lot_identifier
FROM compliance_ledger
WHERE traceability_lot_identifier ~ '^[0-9]+$';

Related Edge Cases to Check Next

The same permissive-defaults root cause produces several nearby failures worth auditing once the two headline defects are closed:

• Float rounding on quantity_value. Parsing 10.1 as a binary float can persist 10.099999999999998, which fails exact reconciliation against a supplier COA. Use Decimal with an explicit quantize step for quantities that must match a source document to the cent or gram.
• Whitespace and Unicode in lot codes. A non-breaking space ( ) or a trailing tab makes 00482A compare unequal to 00482A. Normalize with unicodedata.normalize("NFKC", value) and strip before the length check, and log the raw bytes when a match fails.
• Idempotency on redelivery. An EDI 856 retransmission or a webhook replay can register the same CTE twice. Hash the normalized payload as a deduplication key so retries reconcile instead of inflating lot counts — the routing and dead-letter mechanics for this live in the parent Supplier Data Ingestion pipeline, and the quarantine store’s access controls follow the Security Boundaries for Trace Data model.

Frequently Asked Questions

Why not just fix the timestamps and lot codes after ingestion with a cleanup job?

Because post-hoc cleaning cannot recover information the parser already destroyed. Once 2024-11-02T09:15:00-07:00 is stored as a naive UTC timestamp, the original offset is gone — there is no reliable way to reconstruct whether the local receipt time was Pacific, Mountain, or something else. The same is true of a lot code coerced to 91: you cannot know it was 0091 versus 091 versus 00091. The only defensible place to preserve the value is the ingestion boundary, before coercion happens.

Why does the reproduction feed raw strings to pydantic instead of a pandas DataFrame?

Because pandas.read_csv coerces before your validator ever runs. If you pass pydantic a pre-parsed Timestamp, the offset is already stripped and strict=True cannot help. Read the CSV with dtype=str (or the csv module) so every cell arrives as a string, then let FSMA204KDE parse and validate. The gate is only as strong as the least-coerced value it receives.

Why reject naive datetimes outright instead of assuming UTC?

Assuming UTC is a silent data edit. Under 21 CFR Part 1 Subpart S the receipt time of a CTE is legally material, and guessing an offset fabricates a record the supplier never submitted. Rejecting the payload to quarantine forces the supplier to correct its export so the ledger reflects what actually happened, which is the only state that survives an FDA traceback.

What stops one malformed batch from taking down the whole ingestion service?

The quarantine-first loop plus the circuit breaker. Each invalid record is captured with its error path and routed to a durable quarantine store; valid records in the same batch continue to the ledger uninterrupted. If failures cross the threshold — a sign of systemic schema drift from one supplier — the breaker opens, halts further ingestion, and alerts compliance rather than letting the pipeline churn through a broken feed.

Which 21 CFR Part 1 subpart makes these KDEs mandatory?

Subpart S. Traceability lot code assignment is governed by 1.1320, and the shipping and receiving KDEs — product description, quantity and unit of measure, location, and event date — are enumerated in 1.1340. Because these fields anchor every traceback, coercing any of them at ingestion undermines the exact records 1.1340 requires you to produce on request.

Related

• Supplier Onboarding Automation — the gated onboarding workflow this coercion fix protects.
• API Polling Strategies — the REST ingestion vector whose payloads flow through the same validation gate.
• KDE Field Mapping Guide — the canonical type and constraint for every field the gate enforces.
• Security Boundaries for Trace Data — access controls for the quarantine store and compliance ledger.
• FSMA 204 Architecture & KDE Compliance Mapping — how preserved KDEs map to Critical Tracking Events downstream.

Up: Supplier Onboarding Automation — this page is the schema-drift and type-coercion deep dive for the parent onboarding gate.