FDA 24-Hour Response Automation for FSMA 204 Recalls

When an FDA investigator issues a records request during an outbreak, the clock that matters is not measured in days. Under FSMA 204, a facility that holds records for a food on the Traceability List must provide an electronic, sortable spreadsheet of the requested traceability information within 24 hours of the request, or within a reasonable time the agency agrees to. That obligation is the sharpest edge of the whole Recall Simulation & FDA 24-Hour Response program: every upstream investment in clean Key Data Elements exists so that this one deliverable can be assembled, formatted, and handed over before the deadline expires. This guide owns the orchestration that turns a request into a compliant file on time, every time.

The problem is deceptively narrow but operationally brutal. A request names a product, a lot, or a window, and the response must reconstruct the relevant portion of the trace graph, project it into the exact column layout the FDA expects, and deliver it as a sortable electronic spreadsheet — all while a stopwatch runs. Doing this by hand invites missed deadlines and inconsistent files; doing it as one monolithic script invites a single slow query that silently eats the entire budget. The answer is a staged orchestrator with an explicit timing budget per stage, hard deadlines enforced in code, and observability that tells an operator, mid-incident, exactly how much of the 24 hours remains. This page builds that orchestrator on top of the sibling lot-level recall scoping and one-up, one-back reconstruction guides, and wires its evidence trail into the audit-log export.

Response Workflow and SLA Budget

The response runs as a fixed sequence of five stages, each with its own time budget carved out of the 24-hour window. The pipeline is: request intake (parse and authenticate the FDA request, resolve it to concrete identifiers), scope (expand the request to the full set of implicated Traceability Lot Codes), reconstruct (walk the one-up/one-back chain to gather every relevant Critical Tracking Event), format (project the reconstructed records into the FDA sortable-spreadsheet column layout), and deliver (write the file, compute its hash, and register the delivery artifact). Treating these as discrete stages is what makes the deadline enforceable: each stage is timed independently, and an overrun is caught the moment a stage exceeds its slice rather than at the end when nothing can be done.

The budgets are deliberately conservative and sum to well under 24 hours, because the buffer is not slack — it is the margin an operator needs to rerun a stage after an FDA clarification, absorb a slow database, or handle a partial trace gap surfaced by the Fallback Routing Logic. Reconstruction is given the largest slice because full-chain graph traversal under production data volumes is the stage most sensitive to load; intake and delivery are cheap. The orchestrator records the wall-clock cost of every stage so that repeated drills, run through the mock recall drills program, produce a benchmark distribution rather than a single anecdote. When the p95 of any stage creeps toward its budget, that is the early warning that the query layer needs indexing work long before a real request arrives.

Output Columns and the FDA Sortable Spreadsheet

The deliverable is a sortable electronic spreadsheet whose columns FDA expects to see in a recognizable, machine-readable layout. The orchestrator’s format stage projects each reconstructed Critical Tracking Event into exactly these columns, drawing the canonical field definitions from the KDE Field Mapping Guide. The mapping below binds each internal KDE to its output column, its type, and the Subpart S provision that requires it. Optional columns are always emitted, filled with an explicit empty value rather than omitted, so the column count is stable across every response and the file remains sortable on any field.

The column order is contractual, not cosmetic: an investigator sorts and filters this file directly, so a stable header row and a deterministic row order are part of the compliance guarantee, not a nicety. The mechanics of writing that file — deterministic sort keys, UTF-8 encoding, and CSV or XLSX output that opens cleanly — are the subject of the dedicated sortable spreadsheet export guide, which this orchestrator calls in its format stage. Here the concern is only that every reconstructed event carries the full column set before it reaches that writer.

Production Orchestrator with Per-Stage Deadlines

The orchestrator is the piece that turns five capabilities into one deadline-bound response. It models the request and each stage result with pydantic v2, times every stage against its budget with asyncio.timeout, and aborts a stage the instant it overruns so the failure is loud and early rather than a silent deadline miss discovered at hour 23. The stage bodies below stand in for the real scoping and reconstruction calls; in production each delegates to the sibling guides’ implementations, but the timing, deadline enforcement, and structured observability are exactly as shown.

from __future__ import annotations

import asyncio
import logging
import time
from dataclasses import dataclass, field
from datetime import date, datetime, timezone
from decimal import Decimal

from pydantic import BaseModel, ConfigDict, Field, field_validator

logger = logging.getLogger("fsma204.response")
logger.setLevel(logging.INFO)

# Per-stage budgets in seconds, summing to 14.5h — well under the 24h SLA.
STAGE_BUDGETS: dict[str, float] = {
    "intake": 0.5 * 3600,
    "scope": 3 * 3600,
    "reconstruct": 8 * 3600,
    "format": 2 * 3600,
    "deliver": 1 * 3600,
}
SLA_DEADLINE_SECONDS: float = 24 * 3600


class StageOverrunError(RuntimeError):
    """Raised when a response stage exceeds its allotted time budget."""

    def __init__(self, stage: str, budget_s: float) -> None:
        super().__init__(f"stage '{stage}' overran its {budget_s / 3600:.2f}h budget")
        self.stage = stage
        self.budget_s = budget_s


class DeadlineExceededError(RuntimeError):
    """Raised when cumulative elapsed time would breach the 24-hour SLA."""


class FDARequest(BaseModel):
    """A parsed, authenticated FDA records request."""

    model_config = ConfigDict(strict=True)

    request_id: str = Field(..., min_length=3, max_length=64)
    product_description: str = Field(..., min_length=2)
    lot_codes: list[str] = Field(default_factory=list)
    window_start: date
    window_end: date
    received_at: datetime

    @field_validator("received_at")
    @classmethod
    def _tz_aware(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("received_at must be timezone-aware")
        return v.astimezone(timezone.utc)


class TraceRow(BaseModel):
    """One reconstructed CTE, already shaped to the FDA output columns."""

    model_config = ConfigDict(strict=True)

    traceability_lot_code: str
    product_description: str
    quantity: Decimal = Field(..., gt=0)
    unit_of_measure: str
    location_description: str
    event_type: str
    event_date: date


@dataclass
class StageTiming:
    stage: str
    elapsed_s: float
    budget_s: float

    @property
    def within_budget(self) -> bool:
        return self.elapsed_s <= self.budget_s


@dataclass
class ResponseResult:
    request_id: str
    rows: list[TraceRow] = field(default_factory=list)
    timings: list[StageTiming] = field(default_factory=list)
    export_path: str | None = None

    @property
    def total_elapsed_s(self) -> float:
        return sum(t.elapsed_s for t in self.timings)


async def _run_stage(name: str, coro) -> tuple[object, StageTiming]:
    """Run one stage under its budget, timing it and enforcing the deadline."""
    budget = STAGE_BUDGETS[name]
    start = time.monotonic()
    try:
        async with asyncio.timeout(budget):
            value = await coro
    except TimeoutError as exc:
        elapsed = time.monotonic() - start
        logger.error(
            "STAGE_OVERRUN | stage=%s | elapsed_s=%.1f | budget_s=%.1f",
            name, elapsed, budget,
        )
        raise StageOverrunError(name, budget) from exc
    elapsed = time.monotonic() - start
    timing = StageTiming(stage=name, elapsed_s=elapsed, budget_s=budget)
    logger.info(
        "STAGE_OK | stage=%s | elapsed_s=%.1f | budget_s=%.1f | within=%s",
        name, elapsed, budget, timing.within_budget,
    )
    return value, timing


class ResponseOrchestrator:
    """Drives the five response stages within the 24-hour SLA."""

    def __init__(self, scoper, reconstructor, exporter) -> None:
        self._scope = scoper
        self._reconstruct = reconstructor
        self._export = exporter

    async def run(self, request: FDARequest) -> ResponseResult:
        result = ResponseResult(request_id=request.request_id)

        lots, t = await _run_stage("intake", self._intake(request))
        result.timings.append(t)
        self._check_deadline(result)

        scoped, t = await _run_stage("scope", self._scope(request, lots))
        result.timings.append(t)
        self._check_deadline(result)

        rows, t = await _run_stage("reconstruct", self._reconstruct(scoped))
        result.rows = rows
        result.timings.append(t)
        self._check_deadline(result)

        formatted, t = await _run_stage("format", self._format(rows))
        result.rows = formatted
        result.timings.append(t)
        self._check_deadline(result)

        path, t = await _run_stage("deliver", self._export(request, formatted))
        result.export_path = path
        result.timings.append(t)
        self._check_deadline(result)

        logger.info(
            "RESPONSE_COMPLETE | request_id=%s | rows=%d | total_h=%.2f",
            request.request_id, len(result.rows), result.total_elapsed_s / 3600,
        )
        return result

    async def _intake(self, request: FDARequest) -> list[str]:
        # Resolve the request to concrete seed lot codes.
        return list(request.lot_codes)

    async def _format(self, rows: list[TraceRow]) -> list[TraceRow]:
        # Rows already conform to the output-column contract; validate the set.
        return [TraceRow.model_validate(r.model_dump()) for r in rows]

    def _check_deadline(self, result: ResponseResult) -> None:
        if result.total_elapsed_s > SLA_DEADLINE_SECONDS:
            logger.critical(
                "SLA_BREACH | request_id=%s | total_h=%.2f",
                result.request_id, result.total_elapsed_s / 3600,
            )
            raise DeadlineExceededError(
                f"cumulative {result.total_elapsed_s / 3600:.2f}h exceeds 24h SLA"
            )

Two enforcement mechanisms work together. asyncio.timeout(budget) bounds each stage so a runaway reconstruction query cannot silently consume hours that belong to formatting and delivery; when it fires, _run_stage translates the raw TimeoutError into a StageOverrunError naming the offending stage. Independently, _check_deadline sums the real elapsed time after each stage and refuses to continue if the cumulative cost has breached the 24-hour SLA, so a series of within-budget-but-slow stages still trips a loud SLA_BREACH log rather than a missed handoff. Every stage emits a structured STAGE_OK line with its elapsed and budgeted seconds, which is exactly the telemetry a benchmark harness needs to build a per-stage latency distribution under load.

Error Handling When a Stage Overruns

An overrun is not merely a logged warning; it is an operational decision point, and the orchestrator is built to make that decision cheap. Because each stage is timed and bounded independently, a StageOverrunError tells the operator precisely which capability blew its budget and how much of the 24-hour window remains — the reconstruct stage overran, eleven hours are gone, and thirteen remain. That is enough context to choose between three responses: rerun the stage with a narrower scope, degrade to a partial-chain result annotated through the Fallback Routing Logic, or escalate to FDA for a reasonable extension. The code never has to guess, because the budget breach surfaces the instant it happens rather than at delivery time.

The distinction between a stage overrun and a hard SLA breach matters. StageOverrunError is recoverable: a single slow stage that trips its slice still leaves the retained buffer intact, so the orchestrator can be re-driven for that stage without abandoning the response. DeadlineExceededError is terminal for the automated path — it means the cumulative budget is genuinely exhausted, and the only correct action is to deliver whatever complete portion of the chain exists and document the gap for the investigator. In both cases the timing record on ResponseResult becomes part of the audit trail. Persisting that record, together with the request and the delivered file’s hash, into the audit log export is what lets a facility prove, after the fact, that it met its obligation and exactly how long each stage took.

Integration with the Recall Response Program

This orchestrator is the coordinating layer of the parent Recall Simulation & FDA 24-Hour Response reference, and it is intentionally thin: it owns timing, deadlines, and observability, and delegates every substantive query to the guides that specialize in it. The scope stage calls the lot-level recall scoping logic to expand a seed request into the full set of implicated Traceability Lot Codes, honoring the transformation rules that create new lot codes downstream. The reconstruct stage calls the one-up, one-back reconstruction logic to walk the trace graph and gather every relevant Critical Tracking Event. The format stage invokes the sortable spreadsheet export writer to project those events into the FDA column layout.

The evidence side of the response closes the loop with the compliance model. The delivered file, its SHA-256 hash, the parsed request, and the full per-stage timing record are written to the audit log export, which projects them into the FDA submission template and the facility’s own recordkeeping. Because the whole pipeline is built to be run on demand, it is also the exact code path exercised by the mock recall drills — a drill is simply a response run against a synthetic request, and the timing distribution those drills produce is the benchmark that proves the 24-hour guarantee is real rather than aspirational. Upstream, the quality of every row depends on the Critical Tracking Event records captured at ingestion, so a facility that neglects KDE completeness cannot reconstruct a defensible chain no matter how fast this orchestrator runs.

Operational Notes

Run the orchestrator as an on-demand job — a triggered worker or a container invoked by the incident tooling — not a long-lived service, so a crashed response is restarted cleanly and the request record remains the single source of truth. Recommended runtime and dependencies:

• Python 3.11+ — asyncio.timeout is a 3.11 addition, and the code uses X | Y unions and modern generics.
• pydantic ≥ 2.5 for the v2 field_validator / model_validate / model_dump API; never mix in the v1 validator decorator.
• openpyxl ≥ 3.1 (or the standard-library csv module) in the export stage, covered in the dedicated export guide.

Configuration belongs in the environment. Provide the STAGE_BUDGETS as tunable values so a facility with a larger graph can reallocate the 24 hours without editing code, an AUDIT_LOG_SINK for the timing and delivery record, and the connection details for the trace store the scope and reconstruct stages query. Every response must log the request identifier, the per-stage elapsed and budgeted seconds, the total wall-clock cost, and the delivered file hash, because those are the fields an investigator or an internal auditor will ask for when confirming the response met its deadline. Rehearse the pipeline on a schedule; a benchmark that is only ever run during a real outbreak is not a benchmark. Retention of the response artifacts must satisfy the FSMA 204 two-year recordkeeping mandate described in the Data Retention Policies and detailed in the FDA FSMA 204 final rule.

Frequently Asked Questions

Related

• Sortable spreadsheet export — the deterministic CSV/XLSX writer the format stage invokes.
• Lot-level recall scoping — expands a seed request into the full set of implicated Traceability Lot Codes for the scope stage.
• One-up, one-back reconstruction — walks the trace graph to gather Critical Tracking Events for the reconstruct stage.
• Audit log export — records the delivered file, its hash, and the per-stage timing as compliance evidence.
• Mock recall drills — runs this orchestrator against synthetic requests to benchmark stage timings under load.

Up: Recall Simulation & FDA 24-Hour Response — this orchestrator is the deadline-bound coordinating layer of the parent recall-response reference.