FSMA 204 TraceabilityCompliance automation with Python

Automating Recurring Mock Recall Drills in Python with APScheduler

A mock recall drill you have to remember to run is a drill that quietly stops happening. The first few weeks after a compliance push, someone triggers the drill by hand; a quarter later the manual run has slipped, the traceability graph has drifted, and the next time anyone scores a reconstruction is during an actual FDA request. The fix is to make the drill a scheduled job that fires on its own, picks a seed lot without human bias, runs the reconstruction, writes a scorecard to durable history, and pages someone the moment a metric regresses. This guide takes the scored Mock Recall Drills runner and wraps it in an APScheduler job that turns a one-off exercise into continuous, self-auditing assurance that the 24-hour reconstruction still works.

The task has four moving parts that a naive implementation gets wrong: the schedule (it must fire reliably without blocking), the seed selection (it must rotate so coverage is real), the persistence (each scorecard must survive a restart to enable regression comparison), and the alerting (a failed or degraded drill must reach a human). Get any one wrong and the automation gives false confidence — the worst possible state for a compliance control.

Root Cause Analysis

Naive drill automation fails in ways that are individually plausible and collectively fatal. The most common defect is a while True loop with a time.sleep(604800) that blocks the process for a week: the drill runs once at startup, then the container restarts on a deploy and the timer resets, so in practice the drill fires far less often than intended and never at a predictable time. The second defect is a fixed seed lot — the loop always drills the same known-good lot, so after the first pass it re-verifies a path that already works and learns nothing about the thousands of other lots that might not.

The third defect is the absence of persistence. A drill that prints its scorecard and moves on has no memory, so it cannot answer the question that makes drills valuable over time: is our trace completeness getting worse? Regression is the failure mode that matters most, because a traceability system rarely breaks all at once — it erodes when a supplier changes an export format, a co-packer is onboarded without proper lot linkage, or a schema migration silently drops a reference document field. Without a stored baseline, each drill is an isolated snapshot and the slow slide from 100% to 96% completeness goes unnoticed until it is a live emergency.

The regulatory stakes make this more than an engineering tidiness issue. 21 CFR Part 1, Subpart S requires a covered facility to furnish sortable electronic traceability records within 24 hours of an FDA request (§ 1.1455). A drill regime that silently degrades is worse than none, because it manufactures documented confidence that the 24-hour capability exists while the underlying data quietly rots. Reliable scheduling, rotating seed selection, durable scorecards, and regression alerting are the four controls that keep the automation honest — and the metrics they watch are the same ones the Data Quality Monitoring layer tracks against per-supplier SLAs.

Minimal Reproducible Example

The snippet below is the well-intentioned version that fails. It blocks on a week-long sleep, always drills the same lot, and discards every scorecard, so it can neither rotate coverage nor detect a regression.

import time

def run_drill(seed_tlc: str) -> None:
    # imagine this calls the scored DrillRunner from the mock-drills guide
    print(f"ran drill for {seed_tlc}")

def naive_drill_loop() -> None:
    while True:
        # BUG 1: always the SAME lot -> coverage never rotates, learns nothing new.
        run_drill("LOT-ROMA-0419-RC")
        # BUG 2: scorecard is printed inside run_drill and discarded -> no history,
        #         so a slide from 100% to 96% completeness is invisible.
        # BUG 3: blocks the process for 7 days; a restart resets the timer and the
        #         drill fires unpredictably, not on a defensible cadence.
        time.sleep(604800)

naive_drill_loop()

Three defects compound. The fixed seed means the drill only ever proves one path; the discarded scorecard means there is no baseline to regress against; and the blocking sleep means the schedule is a fiction that a single redeploy erases. In production this looks like a drill that “runs” but whose history is empty and whose coverage is a single lot — a control that exists on paper and nowhere else.

Fix Implementation

The corrected design uses APScheduler for a real cron-style trigger, random.choice to rotate the seed lot across a scenario catalogue, SQLite for durable append-only scorecards, and a regression check against a rolling baseline that alerts on any failed or degraded run. The drill itself is the scored DrillRunner from the parent guide; this code is the automation shell around it.

Figure — The automated drill pipeline:

The automated mock recall drill pipeline from cron trigger to regression alert A five-stage automated pipeline. APScheduler fires on a cron trigger, a seed lot is selected at random from the scenario catalogue, the mock drill runs a traceback and scores it, the scorecard is persisted to an append-only store, and a regression check compares the result to a baseline and alerts on any drop. A dashed return path shows the pipeline repeating on the next scheduled run. next scheduled run · weekly cron APScheduler cron trigger Select seed lot randomized Run mock drill traceback + score Persist scorecard append-only store Regression check alert on drop

The scheduler uses a non-blocking BackgroundScheduler with a cron trigger, so the drill fires at a defensible fixed time and a redeploy re-registers the job rather than resetting a sleep timer. Seed selection draws from a scenario catalogue with random.choice, so coverage rotates across the risk surface over successive runs. Persistence and regression detection share one SQLite table.

from __future__ import annotations

import json
import logging
import random
import sqlite3
from datetime import datetime, timezone
from decimal import Decimal
from pathlib import Path

from apscheduler.schedulers.background import BackgroundScheduler

# The scored runner and models come from the mock-drills guide.
from mock_drill import DrillRunner, DrillScenario, DrillScorecard, DrillThresholds

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

SCORECARD_DB = Path("drill_scorecards.db")
REGRESSION_TOLERANCE = Decimal("1.0")  # allowed drop from baseline, in points


def _connect() -> sqlite3.Connection:
    conn = sqlite3.connect(SCORECARD_DB)
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS scorecards (
            run_at TEXT NOT NULL,
            scenario_id TEXT NOT NULL,
            completeness_pct TEXT NOT NULL,
            reconstruct_seconds TEXT NOT NULL,
            mass_balance_pct TEXT NOT NULL,
            bol_reconciliation_pct TEXT NOT NULL,
            orphaned_events INTEGER NOT NULL,
            passed INTEGER NOT NULL
        )
        """
    )
    return conn


def persist_scorecard(card: DrillScorecard, run_at: str) -> None:
    """Append the scorecard to durable history (never update in place)."""
    conn = _connect()
    with conn:
        conn.execute(
            "INSERT INTO scorecards VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
            (
                run_at,
                card.scenario_id,
                str(card.completeness_pct),
                str(card.reconstruct_seconds),
                str(card.mass_balance_pct),
                str(card.bol_reconciliation_pct),
                card.orphaned_events,
                int(card.passed),
            ),
        )
    conn.close()


def baseline_completeness(scenario_id: str, window: int = 5) -> Decimal:
    """Rolling mean completeness over the last `window` runs of this scenario."""
    conn = _connect()
    rows = conn.execute(
        """
        SELECT completeness_pct FROM scorecards
        WHERE scenario_id = ?
        ORDER BY run_at DESC
        LIMIT ?
        """,
        (scenario_id, window),
    ).fetchall()
    conn.close()
    if not rows:
        return Decimal("0")
    values = [Decimal(r[0]) for r in rows]
    return sum(values, Decimal(0)) / Decimal(len(values))


def alert(subject: str, card: DrillScorecard) -> None:
    """Route to the same on-call path a live recall would use."""
    logger.error(
        "DRILL ALERT | %s | %s", subject, json.dumps(card.model_dump(), default=str)
    )
    # In production: page on-call and open a remediation ticket.


def run_scheduled_drill(
    runner: DrillRunner,
    scenarios: list[DrillScenario],
    thresholds: DrillThresholds,
) -> DrillScorecard:
    """One automated drill: pick a lot, run, persist, and check for regression."""
    scenario = random.choice(scenarios)
    run_at = datetime.now(timezone.utc).isoformat()

    # Capture the baseline BEFORE this run is written, so we compare to history.
    baseline = baseline_completeness(scenario.scenario_id)

    card = runner.run(scenario)
    persist_scorecard(card, run_at)

    if not card.passed:
        alert(f"drill {scenario.scenario_id} FAILED thresholds", card)
    elif baseline > 0 and card.completeness_pct < baseline - REGRESSION_TOLERANCE:
        alert(
            f"drill {scenario.scenario_id} REGRESSED "
            f"{card.completeness_pct}% vs baseline {baseline}%",
            card,
        )
    else:
        logger.info(
            "drill %s ok | completeness=%s%% | passed",
            scenario.scenario_id,
            card.completeness_pct,
        )
    return card


def start_scheduler(
    runner: DrillRunner,
    scenarios: list[DrillScenario],
    thresholds: DrillThresholds,
) -> BackgroundScheduler:
    scheduler = BackgroundScheduler(timezone="UTC")
    scheduler.add_job(
        run_scheduled_drill,
        trigger="cron",
        day_of_week="mon",
        hour=3,
        minute=0,
        args=[runner, scenarios, thresholds],
        id="weekly_mock_recall",
        replace_existing=True,  # a redeploy re-registers rather than duplicating
        misfire_grace_time=3600,
    )
    scheduler.start()
    logger.info("mock recall drill scheduler started | job=weekly_mock_recall")
    return scheduler

The replace_existing=True and misfire_grace_time settings are what make the schedule survive real operations: a redeploy re-registers the job under a stable id instead of stacking duplicates, and a brief outage at the trigger time still fires the drill within the grace window rather than skipping it silently. Capturing baseline before persist_scorecard is deliberate — comparing the fresh run against history that already includes itself would dilute the signal. Passing the scenario list means each run rotates the seed lot, so the drift the naive version never saw becomes visible within a few cycles.

Verification Steps

Confirm the automation along three axes: scorecards are actually persisted, a regression trips an alert, and a threshold failure trips an alert. Each is a tight unit test against the functions above.

1. Assert the scorecard is persisted and readable. Run one drill and confirm a row lands in history:

def test_scorecard_is_persisted(tmp_path, monkeypatch) -> None:
    import mock_drill_scheduler as sched
    monkeypatch.setattr(sched, "SCORECARD_DB", tmp_path / "test.db")
    runner, scenarios, thresholds = build_test_fixture()  # passing scenario
    card = sched.run_scheduled_drill(runner, scenarios, thresholds)

    conn = sched._connect()
    rows = conn.execute("SELECT scenario_id, passed FROM scorecards").fetchall()
    conn.close()
    assert len(rows) == 1
    assert rows[0][0] == card.scenario_id
    assert rows[0][1] == 1  # passed persisted as 1

2. Confirm a completeness regression raises an alert. Seed history with a high baseline, then force a degraded run and assert alert fires:

def test_regression_triggers_alert(tmp_path, monkeypatch) -> None:
    import mock_drill_scheduler as sched
    monkeypatch.setattr(sched, "SCORECARD_DB", tmp_path / "test.db")
    calls: list[str] = []
    monkeypatch.setattr(sched, "alert", lambda subject, card: calls.append(subject))

    # Prior good history at 100% completeness, then a run that drops to 97%.
    seed_history(sched, scenario_id="DRILL-2026-Q2-07", completeness="100.00", n=5)
    runner, scenarios, thresholds = build_degraded_fixture(completeness="97.00")
    sched.run_scheduled_drill(runner, scenarios, thresholds)

    assert any("REGRESSED" in s for s in calls)

3. Confirm a threshold failure raises an alert. A drill that misses a hard threshold — say an orphaned event — must alert regardless of baseline:

def test_threshold_failure_triggers_alert(tmp_path, monkeypatch) -> None:
    import mock_drill_scheduler as sched
    monkeypatch.setattr(sched, "SCORECARD_DB", tmp_path / "test.db")
    calls: list[str] = []
    monkeypatch.setattr(sched, "alert", lambda subject, card: calls.append(subject))

    runner, scenarios, thresholds = build_failing_fixture()  # one orphaned event
    card = sched.run_scheduled_drill(runner, scenarios, thresholds)

    assert card.passed is False
    assert any("FAILED thresholds" in s for s in calls)

Running the three together proves the automation does what the naive loop could not: it records every run, notices when a scenario slips below its own history, and escalates a hard failure immediately. Watch the scheduler log line at the cron time to confirm the job actually fires; an empty scorecards table after a scheduled window is the tell-tale of a misregistered job.

2026-07-13 03:00:00 | INFO | fsma204.drill_scheduler | drill DRILL-2026-Q2-07 ok | completeness=100.00% | passed
2026-07-20 03:00:01 | ERROR | fsma204.drill_scheduler | DRILL ALERT | drill DRILL-2026-Q2-07 REGRESSED 97.00% vs baseline 100.00% | {...}

Once the scheduled drill runs cleanly, three adjacent failure modes deserve a check:

  • Overlapping runs on a slow drill. If a drill against a large lot takes longer than the interval between triggers, APScheduler can start a second run before the first finishes and corrupt shared state. Set max_instances=1 on the job so a slow run is skipped rather than overlapped, and if drills routinely run long, hand the reconstruction to a background worker as described in Async Batch Processing.
  • Baseline poisoned by a bad run. The rolling mean includes failed runs, so a single anomalous drill drags the baseline down and can mask a subsequent real regression. Exclude failed runs from the baseline window, or compare against the median of passing runs only, so the baseline reflects the system’s healthy state rather than its worst day.
  • Seed lot no longer in the graph. A scenario can name a lot that has aged out under retention, so the traceback returns nothing and the drill reports 0% completeness for a reason that is not a real defect. Validate each scenario’s seed against the live graph before drilling, and refresh the catalogue against the same retention boundaries covered in Data Retention Policies.

Automating the drill is a compliance control, not a convenience: a scheduled, self-auditing drill is documented, repeatable evidence that the 24-hour reconstruction still works, and the regression alert is what converts slow data erosion into an actionable ticket weeks before it would surface as a failed FDA request. Track drill pass rate, per-scenario completeness trend, and alert volume as first-class telemetry alongside the Data Quality Monitoring signals.

Frequently Asked Questions

Why use APScheduler instead of a plain sleep loop or an OS cron entry?

A sleep loop blocks the process and resets on every redeploy, so the drill fires unpredictably. APScheduler gives a real cron-style trigger with a stable job id, a misfire grace window, and instance limits, all inside the application where it can share the scenario catalogue and the scorecard store. An OS cron entry also works, but then persistence, alerting, and regression logic still have to live in the invoked script — APScheduler keeps the whole control in one place.

Why randomize the seed lot instead of drilling a fixed one?

A fixed seed re-verifies a single path that already works and learns nothing about the rest of the traceability graph. Drawing the seed from a scenario catalogue with random selection rotates coverage across suppliers, co-packers, and transformation types over successive runs, so real defects on rarely exercised paths surface. Bias the catalogue toward high-risk commodities so the rotation still concentrates on the product that matters most.

How does the automation detect a regression rather than just a hard failure?

It captures a rolling baseline of recent completeness for the scenario before the new run is written, then compares the fresh score against that baseline minus a small tolerance. A hard failure trips the threshold gate and alerts immediately; a softer slide — say 100% down to 97% while still technically passing — trips the regression check. Both route to the same on-call alert, so gradual erosion is caught as early as an outright break.

Where should scorecards be stored for audit and regression history?

In durable, append-only storage — SQLite for a single node, Postgres for shared history — so the record survives restarts and is available for both regression comparison and FDA-facing evidence. Never update a scorecard in place; each run is a new row. Align retention of the scorecard history with the facility’s broader record-retention policy so the drill trail itself is auditable.

Up: Mock Recall Drills & Traceability Exercises for FSMA 204 — this guide automates the drill that page defines and scores.

Related content

Up Mock Recall Drills & Traceability Exercises for FSMA 204