Suspect Product Investigation Workflows

Suspect product investigation is the exception-handling discipline of DSCSA Compliance Architecture & Standards Mapping: the workflows that fire the moment a serialized unit fails verification, a scan looks tampered with, or a serial number appears where it should not. Under the Drug Supply Chain Security Act (DSCSA), a suspect product is any package or homogeneous case for which there is reason to believe it may be counterfeit, diverted, stolen, intentionally adulterated, or otherwise unfit for distribution. The operational problem this page solves is narrow and unforgiving: how do you turn an ambiguous data anomaly into a deterministic, timestamped, auditable investigation that either clears the unit or escalates it to illegitimate — without shipping suspect stock, without halting the packaging line, and without missing the statutory notification window? For supply-chain operators, serialization specialists, and compliance officers, the answer is a workflow modeled as a finite state machine where every transition emits an immutable audit event.

Figure — Suspect-to-illegitimate product investigation lifecycle: a guarded state machine where every transition emits an immutable audit event.

Regulatory Triggers & Classification Thresholds #

DSCSA mandates immediate quarantine and investigation upon detection of a suspect product. Operational triggers manifest across scanning infrastructure, transaction records, and physical inspection points, and each maps to a distinct entry condition on the investigation state machine:

• Mismatched, unreadable, or physically compromised 2D DataMatrix codes carrying the GS1 Application Identifiers (01) GTIN, (21) serial, (17) expiration, and (10) lot.
• Serial number duplication across disparate lots, trading partners, or manufacturers — the same SGTIN commissioned twice.
• Discrepancies between physical product attributes and the EPCIS ObjectEvent, AggregationEvent, or TransactionEvent history on record.
• Verification Router Service (VRS) query failures, timeouts, or a false verification response carrying a reason code such as no_match, expired, or recalled.
• Evidence of tampering, unauthorized repackaging, or relabeling detected at receiving.

Upon identification, the regulatory clock starts. When a product is determined to be illegitimate, trading partners must notify the FDA using Form FDA 3911 and alert affected immediate trading partners within 24 hours, per the official FDA guidance on the Drug Supply Chain Security Act. If the product is cleared as legitimate, quarantine is lifted and the case closes with immutable documentation. The precise delineation between suspect (investigate, do not yet notify) and illegitimate (notify within 24 hours) status governs every downstream reporting obligation, which is why the classification threshold must be encoded explicitly rather than left to operator judgment.

Foundational Concepts & State Contracts #

A production investigation workflow is a state machine with a small, closed set of states and a strict transition table. Encoding it as an enumerated contract — rather than free-text status fields — is what makes the workflow auditable and testable. Four states cover the full lifecycle: QUARANTINED (isolated, awaiting review), UNDER_INVESTIGATION (traceability query in flight), LEGITIMATE (cleared, quarantine liftable), and ILLEGITIMATE (confirmed, notification obligations active).

The data contract binds three artifacts to every case: the serialized identifier under investigation, the trigger that opened the case, and the append-only event log. Serialized identifier correctness depends on the same rules enforced by GS1 Standards Implementation — a malformed GTIN check digit is itself a valid investigation trigger, not a parser crash.

from datetime import datetime, timezone
from enum import StrEnum
from pydantic import BaseModel, Field, field_validator


class CaseState(StrEnum):
    QUARANTINED = "QUARANTINED"
    UNDER_INVESTIGATION = "UNDER_INVESTIGATION"
    LEGITIMATE = "LEGITIMATE"
    ILLEGITIMATE = "ILLEGITIMATE"


# Allowed transitions — anything absent here is rejected and logged as a defect.
TRANSITIONS: dict[CaseState, set[CaseState]] = {
    CaseState.QUARANTINED: {CaseState.UNDER_INVESTIGATION},
    CaseState.UNDER_INVESTIGATION: {CaseState.LEGITIMATE, CaseState.ILLEGITIMATE},
    CaseState.LEGITIMATE: set(),
    CaseState.ILLEGITIMATE: set(),
}


class SuspectCase(BaseModel):
    case_id: str
    gtin: str = Field(pattern=r"^\d{14}$")   # AI (01)
    serial: str = Field(min_length=1, max_length=20)  # AI (21)
    lot: str          # AI (10)
    expiry: str       # AI (17), YYMMDD
    trigger: str      # e.g. "VRS:no_match", "DUP_SERIAL", "DATAMATRIX_UNREADABLE"
    state: CaseState = CaseState.QUARANTINED
    opened_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

    @field_validator("gtin")
    @classmethod
    def gtin_check_digit(cls, v: str) -> str:
        body, check = v[:-1], int(v[-1])
        total = sum((3 if i % 2 == 0 else 1) * int(d) for i, d in enumerate(reversed(body)))
        if (10 - total % 10) % 10 != check:
            raise ValueError("GTIN check digit failed")  # itself a suspect trigger
        return v

Step-by-Step Implementation #

A production investigation workflow operates across three synchronized layers — physical containment, digital validation, and compliance documentation — implemented as ordered, individually auditable steps.

1. Quarantine and segregate the physical unit #

The first response to any trigger is physical isolation. Barcode scanning at intake generates a unique investigation case ID and applies an inventory lock in the Warehouse Management System (WMS), moving the unit to a controlled location with role-based access so it cannot be picked or shipped. This satisfies the DSCSA requirement to quarantine suspect product pending investigation.

def open_case(scan: dict) -> SuspectCase:
    case = SuspectCase(
        case_id=f"INV-{scan['gtin']}-{scan['serial']}",
        gtin=scan["gtin"], serial=scan["serial"],
        lot=scan["lot"], expiry=scan["expiry"],
        trigger=scan["trigger"],
    )
    wms.apply_inventory_lock(case.gtin, case.serial, location="QUARANTINE-CAGE-1")
    audit.append(case.case_id, "CASE_OPENED", state=case.state, trigger=case.trigger)
    return case

2. Reconstruct the chain of custody via traceability query #

With the unit locked, the orchestration layer transitions the case to UNDER_INVESTIGATION and pulls EPCIS event history, manufacturer master data, and a fresh VRS verification response to reconstruct the product’s chain of custody. Interoperable queries here depend on the standardized event structure defined by GS1 Standards Implementation and are routed through the Verification Router Service Architecture, which resolves the query to the responsible manufacturer’s endpoint and returns a signed attestation.

async def investigate(case: SuspectCase) -> CaseState:
    transition(case, CaseState.UNDER_INVESTIGATION)
    epcis_history = await epcis.query_events(case.gtin, case.serial)
    vrs = await vrs_client.verify(case.gtin, case.serial, case.lot, case.expiry)

    legitimate = (
        vrs.verified
        and not epcis_history.has_duplicate_commission()
        and epcis_history.matches(case.lot, case.expiry)
    )
    outcome = CaseState.LEGITIMATE if legitimate else CaseState.ILLEGITIMATE
    transition(case, outcome)
    return outcome

3. Enforce the transition contract #

Every state change flows through a single guarded function so no code path can move a case into an illegal state or skip the audit log. This is the enforcement point for the transition table defined earlier and the anchor for regulatory defensibility.

def transition(case: SuspectCase, target: CaseState) -> None:
    if target not in TRANSITIONS[case.state]:
        audit.append(case.case_id, "ILLEGAL_TRANSITION",
                     frm=case.state, to=target)
        raise ValueError(f"illegal transition {case.state} -> {target}")
    audit.append(case.case_id, "STATE_CHANGE", frm=case.state, to=target)
    case.state = target

4. Disposition — clear or report within 24 hours #

A LEGITIMATE outcome lifts the WMS lock and closes the case with full documentation. An ILLEGITIMATE outcome starts the statutory clock: the workflow generates the Form FDA 3911 payload, notifies every affected immediate trading partner, and reconciles inventory — all within the 24-hour window. Because the notification and reporting sequence handles sensitive commercial data, transmission runs under the controls described in Data Security & Encryption Boundaries.

async def disposition(case: SuspectCase) -> None:
    if case.state is CaseState.LEGITIMATE:
        wms.release_inventory_lock(case.gtin, case.serial)
        audit.append(case.case_id, "QUARANTINE_LIFTED")
        return
    # ILLEGITIMATE — 24-hour obligations
    deadline = case.opened_at + timedelta(hours=24)
    form_3911 = build_fda_3911(case)
    await asyncio.gather(
        fda.submit_3911(form_3911, deadline=deadline),
        notify_trading_partners(case, deadline=deadline),
    )
    audit.append(case.case_id, "ILLEGITIMATE_REPORTED", deadline=deadline.isoformat())

For deeper patterns on schema enforcement, automated discrepancy resolution, and continuous monitoring that feed triggers into this workflow, see automating DSCSA compliance gap checks with Python.

Validation & Error Handling #

Accurate suspect identification depends on rigorous parsing of the GS1 Application Identifiers before any classification logic runs. Malformed, truncated, or incorrectly formatted DataMatrix strings must be flagged as their own trigger type — never silently dropped and never allowed to raise an unhandled exception that halts the line. The rule is that bad data opens a case; it does not crash the pipeline.

When discrepancies arise between a scan and the EPCIS record, the system logs the anomaly, preserves the raw scan payload verbatim, and initiates a secondary verification cycle rather than jumping straight to an illegitimate determination. A single VRS timeout is a transient fault, not proof of counterfeiting; a confirmed no_match after retries is a substantive finding. Distinguishing the two prevents false-positive quarantines that erode operator trust in the system.

async def verify_with_retry(client, case, attempts=3):
    for n in range(attempts):
        try:
            return await client.verify(case.gtin, case.serial, case.lot, case.expiry)
        except (TimeoutError, TransportError) as exc:
            audit.append(case.case_id, "VRS_TRANSIENT_FAULT", attempt=n, error=str(exc))
            await asyncio.sleep(2 ** n)  # exponential backoff
    audit.append(case.case_id, "VRS_UNRESOLVED", attempts=attempts)
    return VerificationResult(verified=False, reason="unresolved")  # stays suspect, not illegitimate

Structural defects such as a failed GTIN check digit or a duplicate commissioning event are escalated immediately; transient transport faults are retried with backoff and, if unresolved, hold the case in the suspect state for manual review. This mirrors the dead-letter discipline used across schema validation and error handling in the ingestion tier.

Performance & Scalability Considerations #

Investigations are bursty: a single recall notice or a misconfigured line can open thousands of cases in minutes. The workflow must absorb that without unbounded fan-out against VRS endpoints or the EPCIS repository. Three controls keep it stable under load.

• Concurrency bounding. Cap simultaneous VRS lookups with an asyncio.Semaphore so a burst of cases does not exhaust the trading partner’s rate limit and trigger throttling — the same discipline applied when handling rate limits on FDA verification APIs.
• Batch traceability queries. Group EPCIS history lookups for units from the same lot into a single query where the repository supports it; a lot-level recall rarely needs one round trip per serial.
• Circuit breakers. Wrap the VRS client so that repeated endpoint failures open the breaker, park affected cases in the suspect state, and stop hammering a degraded partner — protecting the packaging line from cascading failure while the breaker half-opens on a timer.

sem = asyncio.Semaphore(16)  # bound concurrent verification calls

async def bounded_investigate(case: SuspectCase) -> CaseState:
    async with sem:
        return await investigate(case)

The throughput target is that quarantine and case-open latency stay under the line’s reject-bin dwell time, so a suspect unit is locked before it can advance — even when the digital investigation itself queues behind a rate limiter.

Audit & Compliance Checkpoints #

Every artifact this workflow produces must be reconstructable on demand by an inspector or a trading partner. That requires an append-only audit log, not a mutable status column. Each state transition, VRS response, and notification emits a signed, timestamped record; the log is hash-chained so any post-hoc alteration is detectable.

• Immutable event log. Every transition writes a record with the case ID, prior and new state, actor, and UTC timestamp. Records are never updated or deleted — a correction is a new event.
• Hash-chained integrity. Each record carries the SHA-256 hash of its predecessor, producing a tamper-evident chain that satisfies 21 CFR Part 11 electronic-record expectations.
• Raw payload preservation. The original scan and every VRS response are stored verbatim alongside the derived classification, so the reason to believe is always reproducible.
• Retention. Investigation records and the underlying transaction history are retained for a minimum of six years, aligned with DSCSA statutory requirements and the EPCIS event structure defined in GS1 Standards Implementation.

import hashlib, json

class AuditLog:
    def __init__(self):
        self._prev = "0" * 64

    def append(self, case_id: str, event: str, **fields) -> str:
        record = {
            "case_id": case_id, "event": event,
            "ts": datetime.now(timezone.utc).isoformat(),
            "prev": self._prev, **fields,
        }
        digest = hashlib.sha256(json.dumps(record, sort_keys=True).encode()).hexdigest()
        record["hash"] = digest
        store.write_append_only(record)
        self._prev = digest
        return digest

When handling cross-border shipments, the workflow must account for divergent serialization mandates (for example EU FMD or Saudi SFDA) while maintaining a unified DSCSA-compliant core and the confidentiality controls detailed in Data Security & Encryption Boundaries.

Troubleshooting #

FAQ #

What is the difference between a suspect product and an illegitimate product under DSCSA? #

A suspect product is one there is reason to believe may be counterfeit, diverted, stolen, adulterated, or otherwise unfit for distribution — it must be quarantined and investigated, but not yet reported. An illegitimate product is one confirmed to be counterfeit, diverted, stolen, or adulterated after investigation; that determination triggers the obligation to notify the FDA and immediate trading partners within 24 hours.

How fast must a confirmed illegitimate product be reported? #

Within 24 hours of determining a product is illegitimate, trading partners must notify the FDA using Form FDA 3911 and alert affected immediate trading partners. The workflow should start this clock at the moment the case transitions to the illegitimate state and parallelize the FDA filing with partner notifications to stay inside the window.

Should a VRS timeout mark a unit as illegitimate? #

No. A single timeout is a transient transport fault. Retry with exponential backoff and, if the response remains unresolved, hold the case in the suspect state for manual review. Only a substantive negative verification — such as a confirmed no_match, expired, or recalled after retries — supports an illegitimate determination.

How long must investigation records be retained? #

Investigation records and the underlying transaction history must be retained for at least six years, remain reproducible in their original EPCIS structure, and be backed by an immutable, hash-chained audit log indexed by identifier to satisfy 21 CFR Part 11.

Can the investigation workflow halt the packaging line? #

It should not. Malformed scans and verification failures open a case and lock the affected unit, but the pipeline keeps running. Physical quarantine must complete within the reject-bin dwell time; the slower digital investigation runs asynchronously behind concurrency limits and circuit breakers so a degraded VRS endpoint never stops production.

Related #

• DSCSA Compliance Architecture & Standards Mapping — the parent framework this workflow plugs into
• Automating DSCSA compliance gap checks with Python — the detection patterns that feed triggers into these investigations
• Verification Router Service Architecture — how traceability queries resolve to the responsible manufacturer
• GS1 Standards Implementation — identifier and EPCIS event correctness underpinning every query
• Data Security & Encryption Boundaries — confidentiality and 21 CFR Part 11 controls for reporting