Automating Form 106 from Ledger Snapshots

Python recipe to compute the reported quantity for a DEA Form 106 from a frozen ledger snapshot and bind the payload to a SHA-256 non-repudiation hash

The hard part of automating a DEA Form 106 is not laying out the fields — it is guaranteeing that the reported quantity is exactly the number the perpetual inventory held at the instant a loss was discovered, and proving that number was never touched afterward. This recipe implements that guarantee. Given a frozen point-in-time ledger snapshot and a detected discrepancy, it computes the reported dosage-unit quantity, assembles the field payload, and binds the whole thing to a SHA-256 hash of the canonical snapshot so the filing is non-repudiable. It is the implementation companion to DEA Form 106 Theft & Loss Reporting, which frames the discovery-to-filing timeline and the trigger rules; here the focus is narrowly on the arithmetic and the cryptographic binding.

The failure this prevents is subtle. A naive automation reads a live on-hand balance when the report is generated, minutes or hours after discovery, by which time dispensing has moved the number. The figure filed then cannot be reproduced from the ledger, so an inspector who reconstructs the balance at the discovery timestamp finds a mismatch — a discrepancy inside the discrepancy report. Sourcing the quantity from a frozen snapshot and hashing that snapshot into the payload closes the gap.

Binding a Form 106 payload to a frozen snapshot hash A frozen ledger snapshot and a physical count feed a quantity computation that subtracts the count from the snapshot on-hand to produce the reported quantity. The snapshot is also canonicalized and SHA-256 hashed. The reported quantity and the snapshot hash combine into the Form 106 payload, which carries the hash as its non-repudiation binding. Frozen snapshot on-hand at discovery Physical count dosage units on shelf Compute quantity on_hand − count Canonical SHA-256 hash of snapshot Form 106 payload quantity + snapshot_hash

Prerequisites & environment

  • Python 3.11+ — the recipe uses dataclasses(frozen=True) and the built-in union syntax.
  • Standard library only for the core: hashlib, json, logging, datetime. Add pytest for the verification block.
  • You need a frozen snapshot already captured at discovery. Producing that snapshot by replaying the event stream to an arbitrary timestamp is covered in point-in-time inventory reconstruction; this recipe assumes the snapshot exists and is immutable.
  • Regulatory grounding you should already hold: 21 CFR § 1301.76(b) (theft/loss reporting and Form 106), 21 CFR § 1304.11 (the perpetual inventory the quantity must reconcile against), and 21 CFR § 1304.04 (retain the report and its supporting snapshot).

Implementation: compute, assemble, bind

The design keeps three concerns separate: canonicalization (turn the snapshot into one deterministic byte string), derivation (compute the quantity, never accept it), and binding (attach the snapshot hash to the payload). Canonicalization must be stable — the same snapshot must always hash to the same digest — so keys are sorted and separators are fixed.

python
from __future__ import annotations

import hashlib
import json
import logging
from dataclasses import dataclass, asdict
from datetime import datetime, timezone

logger = logging.getLogger("pharmacy.reporting.form106.snapshot")


class SnapshotBindingError(Exception):
    """Raised when a snapshot cannot back a defensible Form 106 quantity."""


@dataclass(frozen=True)
class FrozenSnapshot:
    """Immutable point-in-time ledger view captured at discovery."""
    ndc_11: str
    schedule: str
    on_hand_at_discovery: int
    last_sequence: int
    last_event_hash: str
    discovery_ts: str            # ISO-8601 UTC


def canonical_bytes(snapshot: FrozenSnapshot) -> bytes:
    """Deterministic serialization: sorted keys, no incidental whitespace.

    Two processes on two machines must derive identical bytes for the same
    snapshot, or the non-repudiation hash is worthless.
    """
    return json.dumps(
        asdict(snapshot), sort_keys=True, separators=(",", ":")
    ).encode("utf-8")


def snapshot_hash(snapshot: FrozenSnapshot) -> str:
    """SHA-256 over the canonical snapshot — the non-repudiation binding."""
    return hashlib.sha256(canonical_bytes(snapshot)).hexdigest()


def reported_quantity(snapshot: FrozenSnapshot, physical_count: int) -> int:
    """Derive the dosage units lost. The quantity is COMPUTED, not supplied,
    so it cannot drift from the ledger state under 21 CFR § 1304.11.
    """
    if physical_count < 0:
        raise SnapshotBindingError("Physical count cannot be negative.")
    if physical_count > snapshot.on_hand_at_discovery:
        raise SnapshotBindingError("Count exceeds ledger on-hand: overage, not loss.")
    qty = snapshot.on_hand_at_discovery - physical_count
    if qty == 0:
        raise SnapshotBindingError("No shortage: nothing to report.")
    return qty


def assemble_payload(
    snapshot: FrozenSnapshot,
    physical_count: int,
    registrant_dea: str,
    loss_type: str,
) -> dict:
    """Build the Form 106 field payload and bind it to the snapshot hash."""
    qty = reported_quantity(snapshot, physical_count)
    digest = snapshot_hash(snapshot)
    payload = {
        "registrant_dea": registrant_dea,
        "ndc_11": snapshot.ndc_11,
        "schedule": snapshot.schedule,
        "loss_type": loss_type,
        "reported_quantity": qty,
        "discovery_ts": snapshot.discovery_ts,
        "source_sequence": snapshot.last_sequence,
        "snapshot_hash": digest,               # non-repudiation binding
        "generated_ts": datetime.now(timezone.utc).isoformat(),
    }
    # PHI-free audit line: identifiers, quantity, and hashes only.
    logger.info(
        "form106_bound ndc=%s schedule=%s qty=%d seq=%d snapshot_hash=%s",
        snapshot.ndc_11, snapshot.schedule, qty, snapshot.last_sequence, digest,
    )
    return payload


def verify_binding(payload: dict, snapshot: FrozenSnapshot) -> bool:
    """Recompute the hash from the retained snapshot; the inspection check."""
    return payload.get("snapshot_hash") == snapshot_hash(snapshot)


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    snap = FrozenSnapshot(
        ndc_11="00093310501",
        schedule="II",
        on_hand_at_discovery=420,
        last_sequence=91827,
        last_event_hash="a91f...c30d",
        discovery_ts="2026-07-16T14:05:00+00:00",
    )
    payload = assemble_payload(snap, physical_count=380,
                               registrant_dea="BX1234567",
                               loss_type="SIGNIFICANT_LOSS")
    print(payload["reported_quantity"], verify_binding(payload, snap))

The verify_binding function is the whole point of the exercise: an inspector, or an automated retention check, recomputes the hash from the stored snapshot and compares it to the snapshot_hash inside the payload. A match proves the payload’s quantity still corresponds to the exact ledger state captured at discovery; a mismatch proves tampering or drift.

Verification & testing

Two properties must be proven: that the quantity is derived correctly and refuses degenerate inputs, and that the binding is stable and detects mutation.

python
import pytest

SNAP = FrozenSnapshot(
    ndc_11="00093310501", schedule="II", on_hand_at_discovery=420,
    last_sequence=91827, last_event_hash="a91f...c30d",
    discovery_ts="2026-07-16T14:05:00+00:00",
)


def test_quantity_is_derived_not_supplied():
    assert reported_quantity(SNAP, physical_count=380) == 40


def test_overage_and_zero_are_rejected():
    with pytest.raises(SnapshotBindingError):
        reported_quantity(SNAP, physical_count=500)   # overage
    with pytest.raises(SnapshotBindingError):
        reported_quantity(SNAP, physical_count=420)   # zero shortage


def test_hash_is_stable_across_calls():
    assert snapshot_hash(SNAP) == snapshot_hash(SNAP)


def test_binding_detects_mutation():
    payload = assemble_payload(SNAP, 380, "BX1234567", "SIGNIFICANT_LOSS")
    assert verify_binding(payload, SNAP) is True
    # A snapshot claiming a different on-hand must fail verification.
    tampered = FrozenSnapshot(**{**asdict(SNAP), "on_hand_at_discovery": 421})
    assert verify_binding(payload, tampered) is False

When assemble_payload runs it emits a single structured, PHI-free line:

text
INFO pharmacy.reporting.form106.snapshot form106_bound ndc=00093310501 \
     schedule=II qty=40 seq=91827 snapshot_hash=7d1e9c4b...af02

That line contains the identifiers, the derived quantity, the source sequence, and the binding hash — everything needed to correlate the filing to the ledger, and nothing that identifies a patient.

Gotchas & compliance pitfalls

  • Do not accept the quantity as input. The instant a caller can pass reported_quantity=40, the number is no longer provably the ledger’s. Derive it from on_hand_at_discovery − physical_count so it is reproducible.
  • Canonicalization must be deterministic. json.dumps without sort_keys=True and fixed separators can reorder keys or add spaces, changing the hash for an identical snapshot. Pin both, and never hash a Python repr or an unordered dict.
  • Snapshot must truly be frozen. If any field of the snapshot can change after discovery, the hash binds to a moving target. Use frozen=True dataclasses and store the snapshot in WORM storage alongside the payload.
  • The hash is the idempotency key. A retried filing with the same snapshot_hash is the same report, not a second loss. Deduplicate on it so a network retry never doubles the reported quantity.
  • An overage is not a loss. A physical count above ledger on-hand is a positive discrepancy that routes to overage investigation, not a Form 106 with a negative quantity. The guard raises rather than filing nonsense.
  • Keep discovery time and generation time distinct. discovery_ts comes from the snapshot; generated_ts is when the payload was built. Conflating them erases the timeline the report is supposed to document.

Frequently Asked Questions

Why hash the whole snapshot instead of just the reported quantity?

Hashing only the quantity would let someone swap in a different snapshot that happens to yield the same number, or alter the schedule or NDC without detection. Hashing the entire canonical snapshot binds the report to the complete ledger state — NDC, schedule, on-hand, sequence, and discovery time — so any change to any field breaks verification.

Can I regenerate the payload later and get the same hash?

Yes, as long as the snapshot is unchanged and canonicalization is deterministic. generated_ts will differ on regeneration, but it is not part of the snapshot, so snapshot_hash stays identical. That reproducibility is what makes the binding a valid non-repudiation check.

What if the physical count arrives after the snapshot was frozen?

That is the normal case: the snapshot fixes the ledger’s on-hand at discovery, and the physical count is gathered during the investigation. The recipe subtracts the count from the frozen on-hand, so the timeline is preserved — the report still reflects the discovery-time balance regardless of when counting finishes.

How does this satisfy the retention requirement?

Persist the frozen snapshot and the bound payload together to WORM storage. 21 CFR § 1304.04 requires the report and its supporting records to remain readily retrievable; keeping the snapshot means the quantity can always be re-derived and the hash re-verified for the full retention horizon.