Implementing a Quarantine State Machine in Python

A batch quarantine finite state machine (FSM) is the control that makes an illegal disposition unrepresentable rather than merely discouraged. This guide builds that FSM in Python end to end: a declarative transition table, guards that reject illegal moves and unsigned dispositions, idempotent replay so a re-delivered excursion cannot double-quarantine a batch, and a hash-linked audit record on every accepted transition. It is the implementation companion to automated batch invalidation and quarantine workflows, which frames the states and the regulatory rationale; here the focus is correct, testable code.

Regulatory hook

EU GDP §5.5 requires that product which has been outside its authorized storage conditions be segregated and assessed before any return to saleable stock. Encoding batch status as an FSM turns that requirement into a structural guarantee: only the transitions the standard permits are representable, so a batch cannot skip assessment. 21 CFR Part 11 §11.10(e) then requires each status change to be a secure, time-stamped, attributable record that does not obscure prior information, which the hash-linked transition log provides. Transitions that constitute a disposition — release or rejection — additionally require a §11.50 signature reference.

The engineering value of the FSM formulation is that it converts a class of compliance failures from runtime bugs a reviewer must catch into states that cannot be constructed. A validation team can only sample the behaviors it thinks to test; an inspector only sees the records that happen to exist. But an illegal transition that the type system refuses to represent needs no test to catch and generates no bad record to find, because it never executes. That is a stronger guarantee than a validated procedure, and it is the reason regulators increasingly expect batch status to be modeled as a controlled state graph rather than a mutable attribute.

Prerequisites

• Python 3.11 or newer (uses frozenset, enum.Enum, and dataclasses).
• Dependencies: the FSM core uses only the standard library. Install nothing beyond a clean interpreter for the core; pytest is used for the test examples:

python3.11 -m venv .venv && source .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install pytest

• Persistence: an append-only or WORM-backed store for transition records, addressed by DSN from your secrets manager.
• Access control: the actor on each transition is an authenticated identity resolved upstream; the FSM records it but does not authenticate. Only the disposition service may drive terminal transitions.

Step-by-Step Implementation

Step 1 — Declare states and the legal transition table

Model status as an enum and the permitted moves as data. Anything absent from the table is unrepresentable, which is the whole point.

from __future__ import annotations

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


class BatchState(str, Enum):
    IN_STOCK = "IN_STOCK"
    QUARANTINE = "QUARANTINE"
    UNDER_REVIEW = "UNDER_REVIEW"
    RELEASED = "RELEASED"
    REJECTED = "REJECTED"


# EU GDP §5.5: legal edges only. Absence of an edge is a hard segregation control.
_LEGAL: dict[BatchState, frozenset[BatchState]] = {
    BatchState.IN_STOCK:     frozenset({BatchState.QUARANTINE}),
    BatchState.QUARANTINE:   frozenset({BatchState.UNDER_REVIEW, BatchState.IN_STOCK}),
    BatchState.UNDER_REVIEW: frozenset({BatchState.RELEASED, BatchState.REJECTED}),
    BatchState.RELEASED:     frozenset(),
    BatchState.REJECTED:     frozenset(),
}
_REQUIRES_SIGNATURE = frozenset({BatchState.RELEASED, BatchState.REJECTED})

Step 2 — Implement guarded transitions

The transition method enforces three guards before it mutates anything: the move must be legal, it must carry an actor and reason, and a disposition must carry a signature.

class IllegalTransition(ValueError):
    """Move not permitted from the current state (EU GDP §5.5)."""


def _guard(state: BatchState, to_state: BatchState,
           actor: str, reason: str, signature_ref: Optional[str]) -> None:
    if to_state not in _LEGAL[state]:
        raise IllegalTransition(f"{state.value} -> {to_state.value} not permitted")
    if not actor or not reason:
        # §11.10(e): every action is attributable and justified.
        raise ValueError("actor and reason are mandatory")
    if to_state in _REQUIRES_SIGNATURE and not signature_ref:
        # §11.50: a disposition must be signed.
        raise ValueError(f"{to_state.value} requires a signature reference")

Prove an illegal move is refused:

try:
    _guard(BatchState.QUARANTINE, BatchState.RELEASED, "qp@site", "hurry", None)
    raise AssertionError("should have been illegal")
except IllegalTransition:
    pass  # QUARANTINE -> RELEASED is not a legal edge

Step 3 — Add idempotent replay protection

A re-delivered excursion event must not create a second identical transition. Key each transition on an idempotency token derived from the triggering event, and skip a token already applied. Re-delivery is not an edge case in a regulated telemetry pipeline; it is the normal consequence of at-least-once messaging, gateway reconnects, and consumer restarts, so a workflow that is not idempotent will accumulate duplicate quarantine records in ordinary operation. Two duplicate holds for one excursion are not merely untidy — they are two conflicting versions of the batch’s history, and reconciling them after the fact is itself a data-integrity investigation. The token approach shown here keeps the applied set in memory for clarity; a production deployment persists it alongside the audit records so the guard survives a restart, which is covered in the persistence step below.

@dataclass
class QuarantineFSM:
    batch_id: str
    state: BatchState = BatchState.IN_STOCK
    _prev_hash: str = "0" * 64
    _applied: set[str] = field(default_factory=set)  # seen idempotency tokens
    persist: Callable[[dict], None] = lambda r: None
    clock: Callable[[], datetime] = lambda: datetime.now(timezone.utc)

    def transition(
        self,
        to_state: BatchState,
        actor: str,
        reason: str,
        idempotency_key: str,
        signature_ref: Optional[str] = None,
    ) -> Optional[dict]:
        if idempotency_key in self._applied:
            # ALCOA+ consistent: a replayed event is a no-op, not a duplicate.
            return None
        _guard(self.state, to_state, actor, reason, signature_ref)

        record = {
            "batch_id": self.batch_id,
            "from_state": self.state.value,
            "to_state": to_state.value,
            "actor": actor,                       # §11.10(e) attributable
            "reason": reason,                     # §11.10(e) reason for change
            "signature_ref": signature_ref,       # §11.50 disposition signature
            "idempotency_key": idempotency_key,
            "occurred_at": self.clock().isoformat(),  # §11.10(e) contemporaneous
            "prev_hash": self._prev_hash,
        }
        canonical = json.dumps(record, sort_keys=True, separators=(",", ":"))
        # §11.10(e): tamper-evident chain link.
        record["record_hash"] = hashlib.sha256(canonical.encode("utf-8")).hexdigest()

        self._prev_hash = record["record_hash"]
        self.state = to_state
        self._applied.add(idempotency_key)
        self.persist(record)
        return record

Step 4 — Verify the chain and idempotency

Confirm that a repeated key is a no-op and that the chain links correctly.

sink: list[dict] = []
fsm = QuarantineFSM("BATCH-42", persist=sink.append)

r1 = fsm.transition(BatchState.QUARANTINE, "system:disposition",
                    "confirmed excursion", idempotency_key="EXC-9")
r2 = fsm.transition(BatchState.QUARANTINE, "system:disposition",
                    "confirmed excursion", idempotency_key="EXC-9")  # replay
assert r1 is not None and r2 is None            # replay produced no second record
assert len(sink) == 1                            # exactly one transition persisted
assert sink[0]["prev_hash"] == "0" * 64          # first link anchors the chain

Step 5 — Attach a persistence adapter

The FSM writes through an injected sink so the core stays deterministic. Back it with an append-only store and load the DSN from the environment. Injecting the sink rather than calling a database directly is what keeps the FSM unit-testable without infrastructure — the validation suite drives thousands of transition sequences against an in-memory list, and the same code path in production writes to WORM media. When the service restarts, it must rehydrate two pieces of state from the persisted records before accepting new transitions: the current state, taken from the most recent transition’s to_state, and _prev_hash, taken from that record’s record_hash, so the chain continues unbroken. Recovering to the logged state rather than a default is what makes the workflow’s history the single source of truth an inspector can rely on.

import os


class AppendOnlySink:
    def __init__(self, writer):
        # `writer.append(record)` must target WORM or append-only media so a
        # transition record can never be overwritten (§11.10(e)).
        self._writer = writer

    def __call__(self, record: dict) -> None:
        self._writer.append(record)


if __name__ == "__main__":
    dsn = os.environ.get("BATCH_AUDIT_DSN")
    if not dsn:
        raise SystemExit("BATCH_AUDIT_DSN environment variable is required")
    # writer = build_append_only_writer(dsn)
    # fsm = QuarantineFSM("BATCH-42", persist=AppendOnlySink(writer))

Confirm a full lawful path reaches a signed disposition:

fsm2 = QuarantineFSM("BATCH-7")
fsm2.transition(BatchState.QUARANTINE, "system", "excursion", "EXC-1")
fsm2.transition(BatchState.UNDER_REVIEW, "qp@site", "open assessment", "REV-1")
final = fsm2.transition(BatchState.REJECTED, "qp@site",
                        "failed stability", "DISP-1", signature_ref="SIG-88")
assert final["to_state"] == "REJECTED" and final["signature_ref"] == "SIG-88"

Compliance Validation Checklist

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

• Only legal edges representable — every transition absent from _LEGAL raises IllegalTransition, enforcing EU GDP §5.5 segregation structurally.
• Dispositions are signed — RELEASED and REJECTED raise without a signature_ref and succeed with one (§11.50).
• Terminal states are terminal — no transition is permitted out of REJECTED or RELEASED, proving irreversibility.
• Replay is a no-op — a repeated idempotency_key produces no second record and no state change (ALCOA+ consistent).
• Every move attributable — an empty actor or reason raises before any mutation (§11.10(e)).
• Chain is tamper-evident — mutating a stored transition breaks the recomputed record_hash against the successor’s prev_hash (§11.10(e)).
• Contemporaneous timestamps — occurred_at is timezone-aware UTC from a disciplined clock (§11.10(e)).
• Append-only persistence — the sink writes to WORM or append-only media; no transition can be overwritten.

Troubleshooting

Related

• Automated batch invalidation & quarantine workflows — the states and regulatory rationale this FSM implements.
• Reversible vs irreversible batch holds under EU GDP — which destinations may be lifted and which are terminal.
• CAPA routing automation for temperature excursions — the trigger that quarantines the batch.
• Implementing 21 CFR Part 11 electronic signatures — the signature the disposition guard requires.
• QMS integration, audit trails & automated batch disposition — the wider disposition boundary.

For architectural context, see automated batch invalidation & quarantine workflows, part of the broader QMS integration, audit trails & automated batch disposition section.