Auto-Generating CAPA Records from Excursion Events

Once an excursion has been classified into a priority, the quality system needs an actual corrective and preventive action (CAPA) record — a structured, validated, hash-linked object that a quality reviewer can open, investigate, and eventually sign a disposition against. This guide builds the generator that consumes a scored excursion event and emits exactly that record into the QMS, idempotently and with full provenance. It complements the classification core in mapping excursion severity to CAPA priority in Python and slots into the stateful router described in CAPA routing automation for temperature excursions.

Regulatory hook

ICH Q10 establishes CAPA as a core element of the pharmaceutical quality system, and it expects every CAPA to be traceable to its trigger, owned, and driven to documented closure. 21 CFR Part 11 §11.10(e) makes the creation of that record an audit-trail event: it must be attributable, time-stamped, and preserved so previously recorded information is never obscured. A CAPA that cannot be traced back to the exact excursion that spawned it, or that can be silently recreated with different content, fails both mandates. The generator therefore binds the source event’s identity and hash into the CAPA and makes creation idempotent so one excursion can never yield two divergent records.

Traceability here is bidirectional and that matters for how the record is built. Forward traceability lets a reviewer start from an excursion and find the CAPA it opened; backward traceability lets an inspector start from a CAPA and prove exactly which telemetry event, at which reading and hash, justified it. Binding source_event_hash — the hash of the originating record, not merely its identifier — supplies the backward direction with cryptographic strength: if anyone later alters the source excursion, the hash stored in the CAPA no longer matches, and the tampering is evident. A bare foreign-key reference cannot do this, because a mutable row can be edited without breaking the link. This is the difference between a CAPA that merely points at its trigger and one that commits to it.

Prerequisites

Python 3.11 or newer.
Dependencies: pydantic v2 for schema validation, standard library for hashing and time:

bash

python3.11 -m venv .venv && source .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install "pydantic>=2.6,<3"

Inputs assumed upstream: a scored, classified excursion event carrying an excursion_id, batch_id, priority, and the source event’s record_hash from the detection layer.
QMS sink: an append-only or WORM-backed store for CAPA records, addressed by DSN from your secrets manager. The environment-variable pattern below is for local development only.
Access control: only the routing service identity may write CAPA records; the sink enforces that at the transport layer.

Step-by-Step Implementation

Step 1 — Define a validated CAPA record schema

A CAPA that enters the QMS must satisfy a strict schema, so a malformed record is rejected at the boundary rather than persisted and discovered at inspection. Model it with pydantic v2.

python

from __future__ import annotations

import hashlib
import json
from datetime import datetime, timezone
from enum import Enum
from typing import Optional

from pydantic import BaseModel, Field, field_validator


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


class CapaRecord(BaseModel):
    # §11.10(a): strict typing rejects malformed records at the QMS boundary.
    capa_id: str = Field(..., min_length=8, max_length=32)
    excursion_id: str = Field(..., min_length=8)
    batch_id: str = Field(..., min_length=4)
    priority: CapaPriority
    title: str = Field(..., min_length=8, max_length=160)
    source_event_hash: str = Field(..., min_length=64, max_length=64)  # ICH Q10 traceability
    status: str = "OPEN"
    opened_at: str
    prev_hash: str = Field(..., min_length=64, max_length=64)
    record_hash: Optional[str] = None

    @field_validator("opened_at")
    @classmethod
    def _utc(cls, v: str) -> str:
        # §11.10(e): contemporaneous, timezone-aware UTC timestamp.
        datetime.fromisoformat(v)  # raises on a naive or malformed value
        return v

Step 2 — Derive a deterministic CAPA identity

Idempotency is the whole game: a re-delivered excursion event must map to the same CAPA id so the QMS updates one record rather than opening a duplicate. Derive the id from the immutable source identifiers.

python

def capa_id_for(excursion_id: str, batch_id: str) -> str:
    # Deterministic id => a replayed event resolves to the SAME CAPA, never a
    # duplicate quality record (ALCOA+ consistent; ICH Q10 one-trigger-one-CAPA).
    seed = f"{excursion_id}:{batch_id}".encode("utf-8")
    return "CAPA-" + hashlib.sha256(seed).hexdigest()[:12].upper()

Verify two calls agree:

python

assert capa_id_for("EXC-77", "BATCH-9") == capa_id_for("EXC-77", "BATCH-9")

Step 3 — Generate the record from an excursion event

The generator assembles a validated CAPA, binds the source event’s hash for traceability, and seals the record with its own hash. A human-readable title is composed so the reviewer sees context immediately.

python

def generate_capa(event: dict, prev_hash: str = "0" * 64) -> CapaRecord:
    """Consume a scored, classified excursion event dict; emit a sealed CAPA."""
    priority = CapaPriority(event["priority"])
    title = (
        f"Cold chain excursion on {event['batch_id']} "
        f"(peak {event.get('peak_deviation_c', '?')} C, "
        f"{event.get('cumulative_minutes', '?')} min)"
    )
    record = CapaRecord(
        capa_id=capa_id_for(event["excursion_id"], event["batch_id"]),
        excursion_id=event["excursion_id"],
        batch_id=event["batch_id"],
        priority=priority,
        title=title,
        # ICH Q10 traceability: bind the exact source event that triggered this.
        source_event_hash=event["record_hash"],
        opened_at=datetime.now(timezone.utc).isoformat(),
        prev_hash=prev_hash,
    )
    # §11.10(e): seal with a hash over the canonical record for tamper evidence.
    canonical = json.dumps(
        record.model_dump(exclude={"record_hash"}),
        sort_keys=True, separators=(",", ":"),
    )
    record.record_hash = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
    return record

Prove the record validates and seals:

python

event = {
    "excursion_id": "EXC-2026-07-0093", "batch_id": "BATCH-VAX-5521",
    "priority": "P1_CRITICAL", "peak_deviation_c": "6.4",
    "cumulative_minutes": "214", "record_hash": "a" * 64,
}
capa = generate_capa(event)
assert capa.status == "OPEN" and len(capa.record_hash) == 64

Step 4 — Persist idempotently into the QMS

Creation must be a conditional insert keyed on capa_id: if a record already exists for this excursion, the store keeps the original and does not overwrite it, preserving the first-captured entry required by ALCOA+. The distinction between an insert-if-absent and an ordinary upsert is the whole compliance argument: an upsert silently replaces the earlier CAPA, destroying the original record and its timestamp, which is precisely the obscuring of previously recorded information that §11.10(e) forbids. Any subsequent change to a CAPA — a status update, an added investigation note — must therefore be modeled as a new appended event that references the original, never as a mutation of the first record. The generator’s responsibility ends at creating the immutable opening record; the lifecycle after that is a sequence of further audit-chained events, each attributable and timestamped in its own right.

python

class CapaSink:
    def __init__(self, writer):
        # `writer.insert_if_absent(key, record) -> bool` returns True on a NEW
        # insert, False if the key already existed. Back it with an append-only
        # or WORM store so an existing CAPA is never silently replaced.
        self._writer = writer

    def commit(self, record: CapaRecord) -> bool:
        created = self._writer.insert_if_absent(
            record.capa_id,
            record.model_dump(),
        )
        if not created:
            # §11.10(e): the original record stands; re-delivery is a no-op, not
            # an overwrite. Callers may reconcile status separately if needed.
            return False
        return True

Confirm idempotency against an in-memory writer:

python

class _MemWriter:
    def __init__(self):
        self._store = {}
    def insert_if_absent(self, key, record):
        if key in self._store:
            return False
        self._store[key] = record
        return True


sink = CapaSink(_MemWriter())
assert sink.commit(capa) is True     # first delivery creates
assert sink.commit(capa) is False    # re-delivery is a no-op (no duplicate)

Step 5 — Wire the generator to the event stream

Consume events from the broker, generate, and commit. Load the sink DSN from the environment; never inline credentials.

python

import os


def handle_event(event: dict, sink: CapaSink, prev_hash: str) -> Optional[CapaRecord]:
    record = generate_capa(event, prev_hash=prev_hash)
    created = sink.commit(record)
    # Only a newly created CAPA advances the chain; a duplicate returns None.
    return record if created else None


if __name__ == "__main__":
    dsn = os.environ.get("QMS_CAPA_DSN")
    if not dsn:
        raise SystemExit("QMS_CAPA_DSN environment variable is required")
    # Bind `writer` to an append-only/WORM client addressed by `dsn`.
    # writer = build_worm_writer(dsn)
    # sink = CapaSink(writer)
    # for event in consume(broker): handle_event(event, sink, prev_hash)

Compliance Validation Checklist

Run this as part of computerized-system validation; each item is independently confirmable by an auditor.

Schema enforced at the boundary — a malformed event is rejected by the pydantic model before any persistence, satisfying §11.10(a).
Source event bound — every CAPA carries source_event_hash linking it to the exact excursion that triggered it (ICH Q10 traceability).
Deterministic identity — capa_id is derived from immutable source ids so a replay resolves to the same record (ALCOA+ consistent).
Idempotent creation proven — a test asserts a re-delivered event does not create a second CAPA and does not overwrite the first (§11.10(e)).
Record sealed — each CAPA carries a record_hash over its canonical form, making later edits detectable (§11.10(e)).
Contemporaneous timestamp — opened_at is a timezone-aware UTC value from a disciplined clock (§11.10(e)).
Append-only sink — the store enforces insert_if_absent semantics on WORM or append-only media, never a blind upsert.
Credentials externalized — the sink DSN comes from a secrets manager and never appears in source or logs.

Troubleshooting

Symptom	Root cause	Fix
Duplicate CAPA for one excursion	Non-deterministic id or blind insert	Derive `capa_id` from source ids and use `insert_if_absent`
`record_hash` differs on replay	Serialization order or float formatting	Use `sort_keys=True`, compact separators, and stringified numerics
CAPA lacks a link back to its excursion	`source_event_hash` omitted or blank	Require the field in the schema; reject events without a source hash
Validation rejects a well-formed event	Naive datetime or short id	Ensure UTC-aware `opened_at` and ids meeting the `min_length` bounds
Re-delivery overwrites the original record	Sink performs upsert not conditional insert	Switch the writer to append-only `insert_if_absent` semantics

CAPA routing automation for temperature excursions — the stateful engine that routes and owns the generated CAPA.
Mapping excursion severity to CAPA priority in Python — how the priority on the event is decided.
Automated batch invalidation & quarantine workflows — the batch hold that runs alongside the CAPA.
QMS integration, audit trails & automated batch disposition — the wider disposition boundary.
Schema validation pipelines for temperature telemetry — the upstream validation idiom reused here.

For architectural context, see CAPA routing automation for temperature excursions, part of the broader QMS integration, audit trails & automated batch disposition section.