Lot-Level Recall Scoping for FSMA 204 Traceback

When a positive pathogen result lands, the difference between a surgical recall and a brand-wide one is a scoping decision made in the first hour. A contamination signal names a symptom — a finished case, a clinical isolate, a supplier notification — not the set of traceability lot codes that actually carry the hazard. The job of scoping is to convert that single signal into the minimal set of affected lots: every lot that could have received contaminated material and every lot that contaminated material could have flowed into, and nothing else. Over-scope and you destroy salable product, exhaust customer goodwill, and bury investigators in noise; under-scope and contaminated lots stay on shelves and the recall fails its only purpose. This work sits at the front of the Recall Simulation & FDA 24-Hour Response reference, because every downstream artifact — the reconstructed chain, the sortable spreadsheet, the drill report — is only as correct as the lot set it starts from.

The triggering requirement is the traceback itself. FSMA 204 (21 CFR Part 1, Subpart S) obligates a covered facility to hand the FDA electronic, sortable traceability records within 24 hours of a request during an outbreak investigation, and the agency’s traceback works by walking Critical Tracking Events one link at a time — one step back to the immediate source, one step forward to the immediate recipient. Scoping mirrors that walk in software: it seeds from the signal, expands backward and forward across the append-only ledger, and stops at a defensible boundary. This guide defines the scoping algorithm, the Key Data Elements it reads, a runnable pydantic v2 engine that walks the ledger graph, the quantity reconciliation that keeps the boundary honest, and the collision control that stops a reused lot code from silently doubling the recall. It feeds two focused walkthroughs — simulating an FDA 24-hour record request and scoping a recall by traceability lot code — and connects sideways to the one-up, one-back reconstruction that expands each lot into its full evidentiary chain.

The Scoping Algorithm: Seed, Expand, Bound

Scoping is a bounded breadth-first walk over a directed graph whose nodes are lots and whose edges are the shipping, receiving, and transformation events that connect them. The critical modeling decision comes first: a node is not a traceability lot code, it is the pair (assigning_location_gln, traceability_lot_code). A lot code alone is not globally unique — two unrelated facilities can and do stamp the same string — so keying the graph on the code alone lets the walk jump across facilities that merely share a label. Keying on the location that assigned the code, per § 1.1320, makes each node a real physical lot, which is the single most important defense against a false-positive explosion.

The walk has three phases. Seeding turns the contamination signal into one or more starting nodes: a finished-goods lot from a customer complaint, a raw ingredient lot from a supplier advisory, or several lots from a shared cooling event. Expansion grows the affected set outward. Backward expansion answers “where could this have come from” — for a lot’s receiving events it follows the reference document and lot code back to the shipping facility’s node (one step back), and for a transformed lot it follows the transformation record from the output lot to each input lot within the same facility (§ 1.1350). Forward expansion answers “where could this have gone” — for a lot’s shipping events it follows to each recipient’s receiving node (one step forward), and for an input lot it follows the transformation to the output lot it became. Bounding is what makes the result minimal rather than merely reachable: unbounded reachability in a dense supply graph eventually touches most of the network, so the engine prunes edges that quantity reconciliation shows could not have carried the hazard, caps the walk at a configurable generation depth, and stops at nodes that quantity accounting proves are fully explained by uncontaminated inputs.

The expansion diagram below shows one generation of the walk from a single seed. In production the same step repeats breadth-first until the frontier empties or the depth cap is hit, and every edge traversed is recorded so the affected set carries its own provenance.

Figure — one generation of scoping expansion from a seed lot:

KDE Contract the Scoping Engine Reads

Scoping never touches raw supplier files; it reads normalized, validated events already committed to the ledger by Supplier Data Ingestion & Sync Automation. The engine depends on a specific subset of Key Data Elements — the identifiers that define graph nodes, the location codes that define edges, the quantities that bound the walk, and the reference documents that pair a shipment with its matching receipt. The table maps each field to the role it plays in the walk and the Subpart S section that mandates it.

The reference_document deserves emphasis. One-up, one-back matching is only reliable when a shipping event and the receiving event on the far side of the same transaction carry a shared document identifier; a lot code match alone is ambiguous when the same lot ships to several customers. The engine treats the reference document as the primary edge key and the lot code as a confirmation, which is how the detailed reconstructing trace chains with NetworkX walkthrough resolves multi-recipient fan-out without inventing phantom links.

Production Scoping Engine

The engine below indexes the ledger once into lookup structures keyed by lot node, then runs a bounded BFS that expands each node through its shipping, receiving, and transformation edges. Ledger reads are I/O-bound and can fail transiently, so the fetch is wrapped in a tenacity retry; the graph walk itself is pure and deterministic and is never retried. Every node admitted to the affected set carries the reason it was admitted — the seed, or the edge and generation depth that reached it — so the output is self-documenting for an FDA submission. Quantity reconciliation runs as the walk expands and routes any lot with an unexplained shortfall to a trace-gap quarantine rather than silently trusting or silently dropping it.

from __future__ import annotations

import logging
from collections import defaultdict, deque
from decimal import Decimal
from datetime import datetime, timezone
from enum import Enum

from pydantic import BaseModel, ConfigDict, Field, field_validator
from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential

scope_logger = logging.getLogger("fsma204.recall.scoping")
scope_logger.setLevel(logging.INFO)

UOM_ENUM = {"kg", "lb", "case", "pallet", "ea", "liter", "gallon"}


class CTEType(str, Enum):
    SHIPPING = "Shipping"
    RECEIVING = "Receiving"
    TRANSFORMATION = "Transformation"


class LedgerReadError(Exception):
    """Raised when the append-only ledger store is transiently unavailable."""


class LotNode(BaseModel):
    """A physical lot: the pair that a traceability lot code alone cannot express."""

    model_config = ConfigDict(frozen=True)

    assigning_location_gln: str = Field(..., pattern=r"^\d{13}$")
    traceability_lot_code: str = Field(..., min_length=1, max_length=50)

    def key(self) -> tuple[str, str]:
        return (self.assigning_location_gln, self.traceability_lot_code)


class LedgerEvent(BaseModel):
    # Lax parsing: the scoping engine reads already-validated rows back from the
    # ledger, so JSON-typed values (enum-by-value, Decimal-from-number) are expected.
    model_config = ConfigDict(extra="allow")

    event_id: str
    cte_type: CTEType
    assigning_location_gln: str = Field(..., pattern=r"^\d{13}$")
    traceability_lot_code: str = Field(..., min_length=1, max_length=50)
    counterparty_gln: str | None = None
    input_lot_codes: list[str] = Field(default_factory=list)
    output_lot_code: str | None = None
    quantity: Decimal = Field(..., gt=0)
    unit_of_measure: str
    reference_document: str | None = None
    event_timestamp: str

    @field_validator("unit_of_measure")
    @classmethod
    def check_uom(cls, v: str) -> str:
        if v not in UOM_ENUM:
            raise ValueError(f"unit_of_measure must be one of {sorted(UOM_ENUM)}")
        return v

    @field_validator("event_timestamp")
    @classmethod
    def check_timestamp(cls, v: str) -> str:
        dt = datetime.fromisoformat(v.replace("Z", "+00:00"))
        if dt.tzinfo is None:
            raise ValueError("event_timestamp must carry a timezone offset")
        if dt.astimezone(timezone.utc) > datetime.now(timezone.utc):
            raise ValueError("event_timestamp cannot be in the future")
        return v

    def node(self) -> LotNode:
        return LotNode(
            assigning_location_gln=self.assigning_location_gln,
            traceability_lot_code=self.traceability_lot_code,
        )


class ScopeHit(BaseModel):
    node: LotNode
    depth: int
    reason: str


class LedgerIndex:
    """In-memory adjacency built once per traceback from validated ledger events."""

    def __init__(self, events: list[LedgerEvent]) -> None:
        self._events = events
        self._by_node: dict[tuple[str, str], list[LedgerEvent]] = defaultdict(list)
        # reference_document -> events sharing it, used to pair shipments to receipts
        self._by_ref: dict[str, list[LedgerEvent]] = defaultdict(list)
        # output_lot_code -> the transformation event that produced it
        self._by_output: dict[str, list[LedgerEvent]] = defaultdict(list)
        for ev in events:
            self._by_node[ev.node().key()].append(ev)
            if ev.reference_document:
                self._by_ref[ev.reference_document].append(ev)
            if ev.output_lot_code:
                self._by_output[ev.output_lot_code].append(ev)

    def events_for(self, node: LotNode) -> list[LedgerEvent]:
        return self._by_node.get(node.key(), [])

    def one_back(self, node: LotNode) -> list[LotNode]:
        """Immediate sources: shipping counterparties of receipts, and transform inputs."""
        neighbors: list[LotNode] = []
        for ev in self.events_for(node):
            if ev.cte_type is CTEType.RECEIVING and ev.reference_document:
                for peer in self._by_ref[ev.reference_document]:
                    if peer.cte_type is CTEType.SHIPPING and peer.node() != node:
                        neighbors.append(peer.node())
            if ev.cte_type is CTEType.TRANSFORMATION:
                for input_tlc in ev.input_lot_codes:
                    neighbors.append(
                        LotNode(
                            assigning_location_gln=ev.assigning_location_gln,
                            traceability_lot_code=input_tlc,
                        )
                    )
        return neighbors

    def one_forward(self, node: LotNode) -> list[LotNode]:
        """Immediate recipients: receiving counterparties of shipments, and transform outputs."""
        neighbors: list[LotNode] = []
        for ev in self.events_for(node):
            if ev.cte_type is CTEType.SHIPPING and ev.reference_document:
                for peer in self._by_ref[ev.reference_document]:
                    if peer.cte_type is CTEType.RECEIVING and peer.node() != node:
                        neighbors.append(peer.node())
        # A transformation that consumed this lot points forward to its output lot.
        for ev in self._events:
            if (
                ev.cte_type is CTEType.TRANSFORMATION
                and node.traceability_lot_code in ev.input_lot_codes
                and ev.assigning_location_gln == node.assigning_location_gln
                and ev.output_lot_code
            ):
                neighbors.append(
                    LotNode(
                        assigning_location_gln=ev.assigning_location_gln,
                        traceability_lot_code=ev.output_lot_code,
                    )
                )
        return neighbors


@retry(
    retry=retry_if_exception_type(LedgerReadError),
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=0.5, max=8),
    reraise=True,
)
def load_ledger_events(raw_rows: list[dict[str, object]]) -> list[LedgerEvent]:
    """Fetch and validate the append-only ledger slice for the traceback window.

    The retry guards only the transient store read; validation is deterministic
    and any structurally broken row raises immediately for quarantine upstream.
    """
    return [LedgerEvent(**row) for row in raw_rows]


def reconcile_quantity(node: LotNode, index: LedgerIndex) -> Decimal:
    """Received/produced quantity minus shipped/consumed quantity for one lot.

    A positive residual is normal on-hand stock. A negative residual means more
    left the lot than ever entered it — an unrecorded intake, i.e. a trace gap.
    """
    inflow = Decimal("0")
    outflow = Decimal("0")
    for ev in index.events_for(node):
        if ev.cte_type is CTEType.RECEIVING:
            inflow += ev.quantity
        elif ev.cte_type is CTEType.TRANSFORMATION and ev.output_lot_code == node.traceability_lot_code:
            inflow += ev.quantity
        elif ev.cte_type is CTEType.SHIPPING:
            outflow += ev.quantity
        elif ev.cte_type is CTEType.TRANSFORMATION and node.traceability_lot_code in ev.input_lot_codes:
            outflow += ev.quantity
    return inflow - outflow


def scope_recall(
    seeds: list[LotNode],
    index: LedgerIndex,
    max_depth: int = 6,
) -> tuple[list[ScopeHit], list[LotNode]]:
    """Bounded BFS over the ledger graph from the seed lots.

    Returns the minimal affected-lot set with provenance, plus the trace-gap
    nodes whose quantity could not be reconciled and must be quarantined.
    """
    seen: set[tuple[str, str]] = set()
    frontier: deque[tuple[LotNode, int]] = deque()
    affected: list[ScopeHit] = []
    gaps: list[LotNode] = []

    for seed in seeds:
        if seed.key() not in seen:
            seen.add(seed.key())
            frontier.append((seed, 0))
            affected.append(ScopeHit(node=seed, depth=0, reason="seed"))

    while frontier:
        node, depth = frontier.popleft()
        if reconcile_quantity(node, index) < 0:
            gaps.append(node)
            scope_logger.warning(
                "TRACE_GAP | gln=%s | tlc=%s | negative quantity residual",
                node.assigning_location_gln,
                node.traceability_lot_code,
            )
        if depth >= max_depth:
            continue
        expansions = (
            [("one_back", n) for n in index.one_back(node)]
            + [("one_forward", n) for n in index.one_forward(node)]
        )
        for direction, neighbor in expansions:
            if neighbor.key() in seen:
                continue
            seen.add(neighbor.key())
            frontier.append((neighbor, depth + 1))
            affected.append(
                ScopeHit(node=neighbor, depth=depth + 1, reason=f"{direction} of {node.traceability_lot_code}")
            )

    scope_logger.info(
        "SCOPE_COMPLETE | seeds=%d | affected=%d | gaps=%d",
        len(seeds), len(affected), len(gaps),
    )
    return affected, gaps

The engine keeps the graph honest in two ways that a naive SELECT ... WHERE traceability_lot_code = ? cannot. First, because every node is (assigning_location_gln, traceability_lot_code), a lot code reused at an unrelated facility is a different node and is never reached unless a real event connects the two — the collision defense is structural, not a post-hoc filter. Second, expansion follows the reference_document to pair a shipment with its true receipt instead of matching lot codes broadly, so a lot that shipped to five customers fans out to exactly those five receipts rather than to every event that mentions the code.

Error Handling and Trace-Gap Quarantine

Two failure classes get two different responses. A transient ledger read failure — a timed-out object store, a replica lag error, a throttled query — is non-deterministic, so load_ledger_events retries it with exponential backoff before surfacing the error; losing the ledger slice mid-traceback would silently under-scope the recall, the worst outcome. A structural or continuity failure — a lot whose outbound quantity exceeds everything that ever entered it, a receiving event with no matching shipment, a transformation whose input lot never appears in the ledger — is not retryable and means the trace itself is broken. The engine routes these to a trace-gap quarantine through the same discipline as the fallback routing logic: the gap is recorded with the node, the residual, and the events that produced it, and a human investigator decides whether to widen scope conservatively or chase the missing record with the supplier.

Quantity reconciliation is the mechanism that surfaces those gaps. For each lot the engine sums inflow (receipts plus transformation output) and outflow (shipments plus transformation consumption); a negative residual means product left a lot that the ledger says never held it, which can only happen if an intake event was never recorded. Rather than trust the graph or drop the lot, the engine flags it. This matters for scope integrity because an unrecorded intake is exactly the edge a traceback must not miss — the missing link could be the path the contamination actually took. When a gap appears, conservative practice is to widen the affected set to include the counterparties implied by the shortfall until the missing event is recovered, a policy the resolving orphaned receiving events in recall queries walkthrough works through in detail.

Integration with the Recall Response Reference

Scoping is the first stage of the Recall Simulation & FDA 24-Hour Response reference and it hands off cleanly to the rest. The affected-lot set it produces becomes the input to one-up, one-back reconstruction, which expands each lot into the full chain of CTEs and KDEs an investigator reads, and then to FDA 24-hour response automation, which renders that chain into the sortable spreadsheet the agency requires. Because the engine records why each lot entered scope, the provenance travels with the data all the way to the export — an auditor can see that lot RC-88213 is in scope because it was one step forward from the seed under reference document BOL-4471, not because it happened to match a string.

The engine reads only what the append-only ledger has already committed, so its correctness rests on the ledger being tamper-evident and complete. That is why scoping is deliberately a read-only consumer of the immutable record and never mutates it: the traceback must be reproducible, and re-running the same seeds against the same ledger slice must always yield the same affected set. Rehearsing that determinism on synthetic outbreaks is the job of the mock recall drills, which replay known-answer scenarios through this engine to prove the boundary holds before a real signal arrives.

Operational Notes

Run scoping as an on-demand job triggered by a recall coordinator, not a standing service — a traceback is an event, and a fresh process guarantees the engine reads the current ledger state rather than a stale cache. Recommended runtime and dependencies:

• Python 3.10+ (the code uses from __future__ import annotations, X | Y unions, and list[...] / dict[...] generics).
• pydantic ≥ 2.5 — the v2 field_validator, ConfigDict, and model_config API. Never mix in the v1 validator decorator.
• tenacity ≥ 8.2 for the retry primitives guarding the ledger read.

Configuration belongs in the environment. Provide a LEDGER_ENDPOINT for the append-only store, a SCOPE_MAX_DEPTH bound (six generations covers most multi-tier produce chains without over-reaching), and a TRACE_GAP_QUEUE target for quarantined nodes. Set max_depth conservatively and raise it only when a drill shows a legitimately longer chain, because each extra generation widens the blast radius. Keep the UOM enum and the seed-selection rules under version control so a scoping change is a reviewable diff, and log every traceback with its seed set, resulting lot count, gap count, and the ledger slice window so a later audit can reproduce the exact decision. For the FDA rule text these obligations flow from, see the FDA Food Traceability Final Rule.

Frequently Asked Questions

Related

• Simulate an FDA 24-hour traceability record request — drives this scoping engine end to end against a wall-clock budget.
• Scoping a recall by traceability lot code — the over-scoping and under-scoping failures a reused lot code causes, and the GLN-namespacing fix.
• One-Up, One-Back Reconstruction — expands each scoped lot into its full evidentiary CTE chain.
• FDA 24-Hour Response Automation — renders the scoped chain into the sortable spreadsheet the FDA requires.
• FSMA 204 Architecture & KDE Compliance Mapping — the append-only ledger and CTE model this engine reads.

Up: Recall Simulation & FDA 24-Hour Response — scoping is the first stage of the recall response reference, producing the affected-lot set every later stage consumes.