FSMA 204 TraceabilityCompliance automation with Python

Parsing EDI 856 for FSMA 204 Compliance: Resolving HL Loop Misalignment During KDE Extraction

The exact production failure this page resolves: an EDI 856 Advance Ship Notice parses without raising a single error, yet the Key Data Elements it writes to your traceability ledger are attached to the wrong physical aggregation level — a lot code meant for an item ends up on the pallet, or a purchase-order number is silently recorded as the traceability lot. The symptom surfaces weeks later, during recall scoping, when the lot chain fails to reconstruct. The root causes are two closely related X12 defects: HL (Hierarchical Level) scope drift, where REF and MAN segments fall outside their parent loop, and qualifier substitution, where a supplier sends REF*ON (Purchase Order Number) in place of the FSMA-mandated REF*BT (Batch/Lot) qualifier. This guide isolates both, then implements a deterministic state-machine parser that anchors every segment to its owning HL context, enforces qualifier precedence, and routes malformed records to quarantine before they reach the compliance ledger.

This is the EDI half of the CSV/EDI Parser Setup contract: the flat-file path and this X12 path both converge on one canonical KDE record, but the HL-loop traversal is where EDI 856 ingestion breaks most often.

Root Cause: HL Scope Drift and Qualifier Substitution

X12 856 encodes physical aggregation with a parent-child pointer system. A compliant pallet-to-case-to-item transmission follows a strict sequence — HL*1**P (Pallet), HL*2*1*C (Case), HL*3*2*I (Item) — where the third element of each HL segment points at its parent’s ID. Every Key Data Element that follows an HL belongs to that loop until the next HL opens. KDE extraction depends entirely on this nesting: the MAN*GM serial, the REF*BT lot code, and the LIN GTIN under an item-level HL describe that item, not the pallet three loops above it.

Two supplier behaviors break the assumption:

  1. Scope drift. Suppliers flatten the hierarchy, omit parent pointers, or emit REF/MAN segments at the transaction (ST/SE) or order (O) level. A parser that simply attaches each segment to the last-seen HL — the naive default — misattributes identifiers the moment a segment appears outside its expected loop.
  2. Qualifier substitution. Many legacy ERP systems default to REF*ON for internal purchase-order reconciliation and drop the REF*BT qualifier entirely. A parser that treats any REF value as “the lot” records a PO number where FSMA 204 requires a traceability lot code — a corrupt KDE that still looks valid.

Under 21 CFR Part 1, Subpart S, these are not cosmetic. The shipping Critical Tracking Event captured in an ASN is only defensible if its traceability lot code (§ 1.1320) and shipping KDEs (§ 1.1340) are bound to the correct item. Scope drift plus qualifier substitution produces a broken CTE lineage graph that fails audits and obscures recall scoping — the precise outcome the rule exists to prevent.

Figure — EDI 856 HL hierarchy and qualifier precedence:

EDI 856 HL loop nesting and REF/MAN qualifier precedence Three nested boxes show how each Key Data Element binds to the HL loop that owns it. The outer Pallet loop (HL*1**P, no parent) carries REF*ON*PO-998877, a purchase order at qualifier ON. Nested inside it, the Case loop (HL*2*1*C, parent pointer to HL 1) carries a LIN GTIN under qualifier BP and REF*BT*LOT-ALPHA-01, the batch/lot at qualifier BT. Nested inside the case, the Item loop (HL*3*2*I, parent pointer to HL 2) carries MAN*GM, a GTIN plus serial at qualifier GM. A precedence strip below shows the ranking that resolves conflicts within a single loop: MAN*GM at rank 3 wins, then REF*BT at rank 2, then REF*ON at rank 1, which is flagged as a substitution rather than trusted as a lot. HL*1**P Pallet · root loop, no parent REF*ON*PO-998877 purchase order — qualifier ON HL*2*1*C Case · parent → HL 1 LIN*1*BP*00012345678901 GTIN — qualifier BP REF*BT*LOT-ALPHA-01 batch/lot — qualifier BT HL*3*2*I Item · parent → HL 2 MAN*GM*000123…234 GTIN + serial — qualifier GM Qualifier precedence — highest rank wins within one HL loop MAN*GM · rank 3 GTIN + serial wins REF*BT · rank 2 batch / lot code FSMA-mandated KDE REF*ON · rank 1 purchase order flagged, never trusted > >

Minimal Reproducible Example

The failure is easiest to see against a naive parser that keeps a single mutable “current lot” and never tracks which HL owns it. Feed it a realistic ASN that mixes a pallet-level REF*ON, a case-level REF*BT, and an item-level MAN*GM:

raw_856 = (
    "ST*856*0001~"
    "HL*1**P~"                 # pallet
    "REF*ON*PO-998877~"        # purchase order at the PALLET level
    "HL*2*1*C~"                # case, parent = pallet 1
    "REF*BT*LOT-ALPHA-01~"     # true batch/lot for the case
    "HL*3*2*I~"                # item, parent = case 2
    "MAN*GM*000123456789012345678901234~"  # GTIN+serial for the item
    "SE*10*0001~"
)

current_lot: str | None = None
for segment in (s for s in raw_856.split("~") if s):
    parts = segment.split("*")
    if parts[0] == "REF":          # BUG: no qualifier check, no HL scope
        current_lot = parts[2]
    elif parts[0] == "MAN" and parts[1] == "GM":
        current_lot = parts[2]

print(current_lot)
# -> '000123456789012345678901234'
# The pallet PO overwrote nothing useful, LOT-ALPHA-01 was clobbered by the
# item serial, and there is NO record of which HL / GTIN this value belongs to.

Every line runs; nothing raises. Yet LOT-ALPHA-01 — the actual batch/lot KDE for the case — has been silently overwritten, the REF*ON PO number was accepted as a lot, and the surviving value carries no HL provenance at all. That is the corrupt state that reaches the ledger.

Fix: A Deterministic HL State-Machine Extractor

The fix maintains active HL context, validates segment placement, and enforces a strict qualifier precedence hierarchy: MAN*GM (GTIN + serial) supersedes REF*BT (Batch/Lot), which supersedes REF*ON (Purchase Order). Segments that arrive with no open HL are quarantined rather than attached to a stale context. KDEs are only committed once GTIN, lot/batch, and the shipping timestamp are all populated, and a configurable error-rate circuit breaker halts ingestion before a malformed supplier batch overwhelms downstream systems. This decoupling of raw ingestion from KDE commitment is the same quarantine-first contract used throughout Supplier Data Ingestion & Sync Automation.

from __future__ import annotations

import logging
from datetime import datetime, timezone
from dataclasses import dataclass, field
from enum import Enum

logger = logging.getLogger("fsma204_edi856_parser")
logger.setLevel(logging.INFO)


class KDEStatus(Enum):
    VALID = "VALID"
    QUARANTINE = "QUARANTINE"
    CIRCUIT_BREAK = "CIRCUIT_BREAK"


@dataclass
class FSMAKDE:
    hl_id: str
    parent_hl: str | None
    hl_level: str
    gtin: str | None = None
    lot_code: str | None = None
    ship_date: datetime | None = None
    source_qualifier: str | None = None
    validation_errors: list[str] = field(default_factory=list)
    status: KDEStatus = KDEStatus.VALID


class QuarantineRecord:
    def __init__(self, segment: str, reason: str, hl_context: str | None = None) -> None:
        self.segment = segment
        self.reason = reason
        self.hl_context = hl_context
        self.timestamp = datetime.now(timezone.utc)


class EDI856FSMAExtractor:
    # Qualifier precedence: a higher rank overwrites a lower one, never the reverse.
    _QUALIFIER_RANK = {"ON": 1, "BT": 2, "GM": 3}

    def __init__(self, max_error_rate: float = 0.05) -> None:
        self.active_hl: str | None = None
        self.hl_context: dict[str, dict] = {}
        self.committed_kdes: list[FSMAKDE] = []
        self.quarantine: list[QuarantineRecord] = []
        self.total_segments = 0
        self.error_segments = 0
        self.max_error_rate = max_error_rate
        self.circuit_open = False

    def _check_circuit_breaker(self) -> bool:
        if self.total_segments == 0:
            return False
        error_rate = self.error_segments / self.total_segments
        if error_rate > self.max_error_rate:
            self.circuit_open = True
            logger.critical(
                "Circuit breaker triggered. Error rate %.2f%% exceeds threshold %.2f%%",
                error_rate * 100, self.max_error_rate * 100,
            )
            return True
        return False

    def ingest_segment(self, raw_segment: str) -> None:
        if self.circuit_open:
            return

        self.total_segments += 1
        clean = raw_segment.strip().rstrip("~")
        if not clean:
            return

        parts = clean.split("*")
        seg_id = parts[0]

        try:
            if seg_id == "HL":
                self._handle_hl(parts)
            elif seg_id in ("REF", "MAN", "LIN", "DTM"):
                self._handle_kde_segments(parts, seg_id)
        except Exception as e:  # defensive: one bad segment must not abort the file
            self.error_segments += 1
            self.quarantine.append(
                QuarantineRecord(clean, f"Parse exception: {e}", self.active_hl)
            )
            logger.warning("Segment quarantined due to exception: %s", clean)

        self._check_circuit_breaker()

    def _handle_hl(self, parts: list[str]) -> None:
        # HL*<id>*<parent>*<level> — a malformed HL desynchronizes the whole loop tree.
        if len(parts) < 4:
            self.error_segments += 1
            self.quarantine.append(
                QuarantineRecord("*".join(parts), "Malformed HL segment")
            )
            return

        hl_id = parts[1]
        parent_id = parts[2] if parts[2] else None
        level = parts[3]
        self.active_hl = hl_id
        self.hl_context[hl_id] = {
            "parent": parent_id,
            "level": level,
            "gtin": None,
            "lot": None,
            "date": None,
            "qualifier": None,
        }
        logger.debug("HL context set: ID=%s Parent=%s Level=%s", hl_id, parent_id, level)

    def _apply_lot(self, ctx: dict, value: str, qualifier: str) -> None:
        # Only overwrite if the incoming qualifier ranks at least as high as the current.
        current = ctx["qualifier"]
        if current is None or self._QUALIFIER_RANK[qualifier] >= self._QUALIFIER_RANK[current]:
            ctx["lot"] = value
            ctx["qualifier"] = qualifier

    def _handle_kde_segments(self, parts: list[str], seg_id: str) -> None:
        # The defect fix: a KDE segment with no open HL is an orphan — quarantine it,
        # never attach it to a stale context.
        if not self.active_hl:
            self.error_segments += 1
            self.quarantine.append(
                QuarantineRecord(
                    "*".join(parts), f"{seg_id} segment outside HL scope", self.active_hl
                )
            )
            return

        ctx = self.hl_context[self.active_hl]

        if seg_id == "LIN":
            # LIN*1*BP*00012345678901*UP*... — GTIN under qualifier BP, UP, or IN.
            for i in range(2, len(parts) - 1, 2):
                if parts[i] in ("BP", "UP", "IN"):
                    ctx["gtin"] = parts[i + 1]
                    break

        elif seg_id == "REF":
            # REF*BT*LOT... (mandated) or REF*ON*PO... (substitution to flag).
            if len(parts) >= 3:
                qualifier, value = parts[1], parts[2]
                if qualifier == "BT":
                    self._apply_lot(ctx, value, "BT")
                elif qualifier == "ON":
                    logger.warning(
                        "REF*ON used instead of FSMA-mandated REF*BT at HL %s", self.active_hl
                    )
                    self._apply_lot(ctx, value, "ON")

        elif seg_id == "MAN":
            # MAN*GM*<GTIN+serial> — highest precedence; overrides REF*BT/REF*ON.
            if len(parts) >= 3 and parts[1] == "GM":
                self._apply_lot(ctx, parts[2], "GM")
                logger.debug("MAN*GM assigned at HL %s", self.active_hl)

        elif seg_id == "DTM":
            # DTM*011*YYYYMMDD — shipping date, the CTE timestamp KDE.
            if len(parts) >= 3 and parts[1] == "011":
                try:
                    ctx["date"] = datetime.strptime(parts[2], "%Y%m%d").replace(
                        tzinfo=timezone.utc
                    )
                except ValueError:
                    logger.warning(
                        "Invalid DTM*011 format at HL %s: %s", self.active_hl, parts[2]
                    )

    def commit_kdes(self) -> list[FSMAKDE]:
        if self.circuit_open:
            logger.error("Commit aborted: circuit breaker open")
            return []

        for hl_id, ctx in self.hl_context.items():
            kde = FSMAKDE(
                hl_id=hl_id,
                parent_hl=ctx["parent"],
                hl_level=ctx["level"],
                gtin=ctx.get("gtin"),
                lot_code=ctx.get("lot"),
                ship_date=ctx.get("date"),
                source_qualifier=ctx.get("qualifier"),
            )

            # FSMA 204 KDE completeness: GTIN + lot/batch + shipping timestamp.
            if not kde.gtin:
                kde.validation_errors.append("Missing GTIN")
            if not kde.lot_code:
                kde.validation_errors.append("Missing Lot/Batch Code")
            if not kde.ship_date:
                kde.validation_errors.append("Missing Shipping Date (CTE)")

            if kde.validation_errors:
                kde.status = KDEStatus.QUARANTINE
                self.quarantine.append(
                    QuarantineRecord(
                        f"HL*{hl_id}*",
                        f"KDE incomplete: {', '.join(kde.validation_errors)}",
                        hl_id,
                    )
                )
                logger.warning("KDE quarantined for HL %s: %s", hl_id, kde.validation_errors)
            else:
                self.committed_kdes.append(kde)

        return self.committed_kdes

    def get_quarantine_report(self) -> list[dict]:
        return [
            {
                "segment": q.segment,
                "reason": q.reason,
                "hl_context": q.hl_context,
                "timestamp": q.timestamp.isoformat(),
            }
            for q in self.quarantine
        ]

The critical change over the naive parser is _apply_lot: qualifier precedence is a ranked comparison, so REF*ON can never clobber a REF*BT already captured for the same loop, and MAN*GM always wins. Combined with the orphan check in _handle_kde_segments, every KDE is now bound to a specific HL with an auditable qualifier provenance.

Verification Steps

Run the corrected extractor against the same drifted transmission from the reproducible example and inspect what it commits versus what it quarantines:

raw_edi = """
ST*856*0001~
HL*1**P~
REF*ON*PO-998877~
DTM*011*20240520~
HL*2*1*C~
LIN*1*BP*00012345678901~
REF*BT*LOT-ALPHA-01~
HL*3*2*I~
MAN*GM*000123456789012345678901234~
SE*10*0001~
""".strip().split("\n")

extractor = EDI856FSMAExtractor(max_error_rate=0.10)
for seg in raw_edi:
    extractor.ingest_segment(seg)

valid = extractor.commit_kdes()
print(f"Committed KDEs: {len(valid)}")
print(f"Quarantined:    {len(extractor.quarantine)}")

# Assert the qualifier precedence held per loop, independent of segment order.
case = extractor.hl_context["2"]
item = extractor.hl_context["3"]
assert case["lot"] == "LOT-ALPHA-01" and case["qualifier"] == "BT"
assert item["qualifier"] == "GM"          # MAN*GM won at the item level
assert extractor.hl_context["1"]["qualifier"] == "ON"  # PO isolated to the pallet

Confirm three things in the output. First, the assertions pass: LOT-ALPHA-01 stays bound to case HL 2 and is not overwritten by the item-level serial — the exact corruption the naive parser produced is gone. Second, the log stream contains a REF*ON used instead of FSMA-mandated REF*BT at HL 1 warning, giving compliance officers an auditable record of the substitution. Third, every loop is quarantined here because none supplies a complete GTIN + lot + shipping-date triple — the pallet has a date but no GTIN, the item has a GTIN and lot but no DTM*011. get_quarantine_report() returns one row per incomplete loop with its hl_context, so reconciliation targets the exact HL at fault rather than the whole file.

  • Multi-shipment interchanges (ST/SE loops). A single ISA/GS envelope can carry several 856 transactions. Reset active_hl and the hl_context map on each SE, or loop IDs from transaction n will bleed into transaction n+1. When these arrive over a pollable endpoint rather than a file drop, pair this parser with API Polling Strategies for idempotent, de-duplicated fetch cycles.
  • Composite GTIN qualifiers in LIN. Some suppliers send the GTIN under a composite SLN/LIN sub-element or use a non-standard qualifier. Validate the extracted GTIN against a GS1 mod-10 check digit before commit, mirroring the Schema Validation Rules applied at every ingestion boundary.
  • PII and location data in N1/REF loops. ASN party loops can carry ship-to identifiers and contact details that must not land in the same store as raw KDEs. Route them per the Security Boundaries for Trace Data so audit-relevant KDEs and sensitive party data stay segregated.

Frequently Asked Questions

Why does REF*ON never overwrite a REF*BT lot code?

Because the two qualifiers mean different things and only one is a KDE. REF*BT carries the batch/lot code FSMA 204 requires under 21 CFR 1.1320; REF*ON carries a purchase-order number used for internal reconciliation. The _apply_lot precedence ranks GM > BT > ON, so once a BT value is captured for a loop, a later ON in the same loop is logged but cannot replace it. The ON value is still recorded when no BT exists, so nothing is lost — it is just flagged as a substitution rather than trusted as a lot.

What happens to a REF or MAN segment that arrives before any HL opens?

It is quarantined as an orphan, never attached to a context. The naive failure mode is to bind such a segment to the last-seen HL, which silently misattributes the identifier. The state machine checks self.active_hl first: if no loop is open, the segment goes to the quarantine store with the reason <seg> segment outside HL scope and its (null) hl_context, so a reviewer can see exactly where the transmission drifted.

Why does MAN*GM take precedence over REF*BT for the lot value?

MAN*GM encodes an application-identifier-structured GS1 element that embeds the GTIN and often the batch/serial in one field, so when a supplier sends it at the item level it is the most specific and machine-verifiable identity for that unit. Ranking it above REF*BT means a serialized item transmission resolves to the serial rather than a looser lot reference, while non-serialized cases still resolve correctly to their REF*BT lot. The precedence is per-loop, so a case and its child item can each carry the qualifier appropriate to their level.

Why quarantine an incomplete KDE instead of committing the partial record?

FSMA 204 shipping records are only defensible when the traceability lot code, the product identity (GTIN), and the shipping CTE timestamp are all present. A partial commit produces a record that looks valid in the ledger but cannot reconstruct a lot chain during recall scoping. The parser commits only complete GTIN + lot + DTM*011 triples and routes anything short of that to quarantine with the specific missing fields, so reconciliation is a targeted fix rather than a full re-parse.

How do I stop loop IDs from one transaction bleeding into the next?

Reset the extractor state on every SE segment. HL IDs restart at 1 in each ST/SE transaction, so if you reuse one extractor across an interchange, transaction two’s HL*1 collides with transaction one’s. Either instantiate a fresh EDI856FSMAExtractor per transaction, or clear active_hl and hl_context when you see SE. The circuit-breaker counters can stay shared if you want a per-file error budget, but the loop map must not.

What should the circuit-breaker error rate be for a new supplier?

Tighter than for an established one. New suppliers whose 856 output has not been validated warrant a max_error_rate around 0.02, so a systemic problem — wholesale HL flattening, a missing qualifier across the file — halts ingestion fast and pages a data steward. Established partners with a clean history can run at 0.05. Always pair the circuit breaker with alerting: a silent halt is itself a compliance gap, because it means KDEs stopped flowing to the ledger without anyone noticing.

Up: CSV/EDI Parser Setup — this EDI 856 state machine is the X12 ingestion vector of the parent parser.

Related content

Up CSV/EDI Parser Setup for FSMA 204 KDE Ingestion