CAPA Routing Automation for Temperature Excursions

When a confirmed temperature excursion arrives at the quality management system, someone has to own it — and the speed, formality, and seniority of that ownership must scale with the risk to the product, not with who happens to be watching the queue. Manual triage cannot do this reliably: a night-shift deviation on a high-value biologic can wait hours for an email to be read, while a trivial ambient blip generates the same heavyweight investigation as a genuine cold-chain breach. CAPA routing automation replaces that queue with a deterministic engine that reads the excursion score, resolves the correct corrective and preventive action (CAPA) priority, assigns an accountable owner with a target closure interval, and writes the assignment to a tamper-evident audit trail. The regulatory anchor is twofold: ICH Q10 establishes CAPA as a core element of the pharmaceutical quality system, and 21 CFR Part 11 §11.10(e) requires that the routing action be a secure, time-stamped, attributable record, while §11.10(k) places the routing logic itself under change control.

Problem Statement: Triage Cannot Be Discretionary

Three concrete failures push teams from manual triage to automated routing.

• Proportionality is a regulatory expectation, not a preference. ICH Q9 requires the effort and formality of a quality action to be commensurate with risk. A human triaging by intuition applies inconsistent thresholds across shifts and reviewers, which is precisely the variability the routing engine exists to remove.
• Latency on a critical excursion is a patient-safety exposure. A P1 deviation on a cold-sensitive product that sits unassigned overnight erodes the window in which corrective action can still preserve the batch. Deterministic assignment with escalation timers bounds that latency.
• Undocumented routing is an audit finding. If an inspector cannot see why a given excursion became a Major rather than a Critical CAPA, and which version of the mapping decided it, the disposition is indefensible. The routing decision and the logic version must both be recorded.

CAPA routing sits at the head of the QMS integration, audit trails and automated batch disposition workstream. It consumes the normalized risk produced by duration-based scoring for temperature excursions, and it hands a quarantined batch to the automated batch invalidation and quarantine workflows once an owner is engaged. Its two implementation deep-dives cover mapping excursion severity to CAPA priority in Python and auto-generating CAPA records from excursion events.

Concept & Specification: The Routing Decision Record

The engine is a pure function surrounded by an audited side-effect boundary. The pure part maps an event to a priority; the audited part assigns an owner, sets a closure deadline, and appends a record. The persisted routing record carries the following fields; the Regulatory anchor column states why each must exist for the record to be inspection-ready.

The routing_version field is what makes the mapping defensible rather than a black box. Because the entire priority table is serialized and hashed into every record, a reviewer confronted with a disposition years later can prove which set of thresholds and escalation rules produced it — the change-control evidence §11.10(k) demands over any system that makes regulated decisions.

Architecture Diagram: Score to Owned CAPA

The routing pipeline is stateful only in its owner-resolution and escalation state; the classification itself is deterministic and replayable. A scored event is classified into a priority band, an owner is resolved from a change-controlled responsibility matrix, a closure deadline is derived, and the assignment is appended to the audit chain before any notification fires.

Production Python Implementation

The module is a complete routing engine. Classification is a pure function of the event and a fingerprinted band table; assignment resolves an owner from a responsibility matrix, derives a closure deadline, and appends an audit record before notifying. Persistence, notification, and the clock are injected so the engine is deterministic under test. Each block cites the clause it satisfies.

from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from enum import Enum
from typing import Callable, Mapping, Optional


class CapaPriority(str, Enum):
    P1_CRITICAL = "P1_CRITICAL"
    P2_MAJOR = "P2_MAJOR"
    P3_MINOR = "P3_MINOR"


@dataclass(frozen=True)
class ExcursionEvent:
    excursion_id: str
    batch_id: str
    score: float
    stability_class: str
    site: str


# (min_score_inclusive, priority, closure_hours). Ordered high to low so the
# first match wins. This table is decision logic under 21 CFR 11.10(k) change
# control and is fingerprinted into every routing record below.
_BANDS: tuple[tuple[float, CapaPriority, int], ...] = (
    (76.0, CapaPriority.P1_CRITICAL, 24),
    (26.0, CapaPriority.P2_MAJOR, 72),
    (0.0, CapaPriority.P3_MINOR, 240),
)

# Change-controlled responsibility matrix: (site, priority) -> accountable role.
_OWNERS: Mapping[tuple[str, CapaPriority], str] = {
    ("WAREHOUSE-A", CapaPriority.P1_CRITICAL): "qa.head@site",
    ("WAREHOUSE-A", CapaPriority.P2_MAJOR): "qa.reviewer@site",
    ("WAREHOUSE-A", CapaPriority.P3_MINOR): "coldchain.ops@site",
}
# Escalation path when a deadline lapses (§11.10(d) authority, ICH Q10 closure).
_ESCALATION: Mapping[str, str] = {
    "coldchain.ops@site": "qa.reviewer@site",
    "qa.reviewer@site": "qa.head@site",
    "qa.head@site": "site.qp@site",  # Qualified Person, terminal escalation
}


def _routing_fingerprint() -> str:
    # Bind bands + owner matrix + escalation to each record for §11.10(k).
    payload = json.dumps(
        {
            "bands": [[b[0], b[1].value, b[2]] for b in _BANDS],
            "owners": {f"{k[0]}|{k[1].value}": v for k, v in _OWNERS.items()},
            "escalation": dict(_ESCALATION),
        },
        sort_keys=True, separators=(",", ":"),
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def classify(event: ExcursionEvent) -> tuple[CapaPriority, int]:
    """Pure, replayable score -> (priority, closure_hours) mapping (ICH Q9)."""
    biologic = event.stability_class.startswith("biologic")
    for min_score, priority, hours in _BANDS:
        if event.score >= min_score:
            # Cold-sensitive product never routes below Major on real exposure.
            if biologic and priority is CapaPriority.P3_MINOR and event.score > 0:
                return CapaPriority.P2_MAJOR, 72
            return priority, hours
    return CapaPriority.P3_MINOR, 240


def resolve_owner(site: str, priority: CapaPriority) -> str:
    # §11.10(d): authority to act is assigned, never left to whoever is watching.
    owner = _OWNERS.get((site, priority))
    if owner is None:
        raise KeyError(f"no owner configured for {site}/{priority.value}")
    return owner


class CapaRouter:
    def __init__(
        self,
        persist: Callable[[dict], None],
        notify: Callable[[str, CapaPriority, str], None],
        clock: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
        actor: str = "system:capa-router",
    ) -> None:
        self._persist = persist
        self._notify = notify
        self._clock = clock
        self._actor = actor
        self._prev_hash = "0" * 64

    def _append(self, record: dict) -> dict:
        # §11.10(e): attributable, time-stamped, hash-linked action record.
        record = {**record, "actor": self._actor,
                  "assigned_at": self._clock().isoformat(),
                  "prev_hash": self._prev_hash}
        canonical = json.dumps(record, sort_keys=True, separators=(",", ":"))
        record["record_hash"] = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
        self._prev_hash = record["record_hash"]
        self._persist(record)
        return record

    def route(self, event: ExcursionEvent) -> dict:
        priority, closure_hours = classify(event)
        owner = resolve_owner(event.site, priority)
        deadline = (self._clock() + timedelta(hours=closure_hours)).isoformat()
        capa_id = "CAPA-" + hashlib.sha256(
            f"{event.excursion_id}:{event.batch_id}".encode("utf-8")
        ).hexdigest()[:12].upper()

        record = self._append({
            "action": "CAPA_ROUTED",
            "capa_id": capa_id,
            "excursion_id": event.excursion_id,
            "batch_id": event.batch_id,
            "priority": priority.value,
            "owner": owner,
            "target_closure_at": deadline,
            "routing_version": _routing_fingerprint(),  # §11.10(k)
        })
        # Notify AFTER the record is durable, so an alert can never exist
        # without its audit trail (ALCOA+ complete).
        self._notify(owner, priority, capa_id)
        return record

    def escalate_if_overdue(self, record: dict, now: Optional[datetime] = None) -> Optional[dict]:
        now = now or self._clock()
        due = datetime.fromisoformat(record["target_closure_at"])
        if now < due:
            return None  # still within ICH Q10 closure window
        next_owner = _ESCALATION.get(record["owner"])
        if next_owner is None:
            return None  # already at terminal authority
        esc = self._append({
            "action": "CAPA_ESCALATED",
            "capa_id": record["capa_id"],
            "from_owner": record["owner"],
            "owner": next_owner,
            "reason": "target_closure_lapsed",
            "routing_version": _routing_fingerprint(),
        })
        self._notify(next_owner, CapaPriority(record["priority"]), record["capa_id"])
        return esc

The deep mechanics of the score-to-priority table — including per-stability-class overrides and the MKT and duration inputs that feed it — are developed in mapping excursion severity to CAPA priority in Python, and the structured CAPA payload emitted into the QMS is built in auto-generating CAPA records from excursion events.

Configuration & Deployment

Routing behaviour is entirely configuration-driven so the decision logic stays version-controlled and re-validatable. Treat the band table, responsibility matrix, and escalation path as a single controlled document under your ICH Q10 pharmaceutical quality system: any change is a change-management event that may require re-validation.

Deploy the router behind the same message broker that carries scored events, and rotate the broker’s mTLS client certificates on a fixed schedule, failing closed on an expired certificate so an unauthenticated source can never inject a routing decision. Pin the active routing_version in your deployment manifest and fail the release if the running fingerprint does not match the approved value, so staging logic can never silently reach production.

Verification & Testing

The router informs regulated disposition, so its tests are validation evidence, not developer convenience. Build the suite around the cases an inspector will probe.

• Determinism (replay) tests. Route the same event twice with a fixed injected clock and assert identical capa_id, priority, and routing_version. Idempotent classification is the demonstration of §11.10(a) validation.
• Band boundary tests. Assert the exact P3 → P2 (26) and P2 → P1 (76) transitions, including equality, so proportional routing under ICH Q9 is deterministic.
• Owner-resolution tests. Assert every (site, priority) pair resolves, and that a missing pair raises rather than silently assigning no one — an unowned CAPA is an ICH Q10 gap.
• Escalation tests. Advance the injected clock past target_closure_at and assert a CAPA_ESCALATED record moves ownership one step up _ESCALATION, terminating at the Qualified Person.
• Audit-chain tests. Mutate one stored field and assert the recomputed record_hash no longer matches the successor’s prev_hash, demonstrating tamper-evidence for §11.10(e).
• CSV protocol hook. Expose a fixture loader that reads an OQ test-vector CSV (score, stability class, site, expected priority, expected owner) so Operational Qualification runs against documented expected outputs.

from datetime import datetime, timezone


def _fixed_clock():
    return datetime(2026, 7, 14, 6, 0, tzinfo=timezone.utc)


def test_biologic_never_routes_minor():
    sink = []
    r = CapaRouter(persist=sink.append, notify=lambda *a: None, clock=_fixed_clock)
    ev = ExcursionEvent("EXC-1", "BATCH-1", score=10.0,
                        stability_class="biologic_2_8", site="WAREHOUSE-A")
    rec = r.route(ev)
    # ICH Q9: a real excursion on cold-sensitive product cannot be Minor.
    assert rec["priority"] == "P2_MAJOR"


def test_routing_is_deterministic():
    a, b = [], []
    ra = CapaRouter(persist=a.append, notify=lambda *x: None, clock=_fixed_clock)
    rb = CapaRouter(persist=b.append, notify=lambda *x: None, clock=_fixed_clock)
    ev = ExcursionEvent("EXC-2", "BATCH-2", 82.0, "biologic_2_8", "WAREHOUSE-A")
    # §11.10(a): same input, same decision and same logic fingerprint.
    assert ra.route(ev)["capa_id"] == rb.route(ev)["capa_id"]
    assert a[0]["routing_version"] == b[0]["routing_version"]

Known Failure Modes & Mitigations

When the owner matrix has no entry for a rare (site, priority) combination, fail loudly to a documented default owner with elevated priority rather than dropping the CAPA — a routed-but-misassigned action is recoverable, an unrouted excursion is a monitoring gap.

Compliance Q&A

Related

• QMS integration, audit trails & automated batch disposition — the wider disposition boundary this routing engine heads.
• Mapping excursion severity to CAPA priority in Python — the deterministic severity-to-priority function in depth.
• Auto-generating CAPA records from excursion events — emitting the structured CAPA payload into the QMS.
• Automated batch invalidation & quarantine workflows — where the affected batch is held pending disposition.
• Duration-based scoring for temperature excursions — the source of the score this engine routes on.

For architectural context, this page sits under QMS integration, audit trails & automated batch disposition.