Parsing Regulatory Gateway Acknowledgement Receipts

Delivering a signed AS2 payload and capturing the synchronous MDN, as covered in Submitting to the FDA ESG AS2 Gateway with Python, only confirms the envelope arrived intact. Whether the filing was actually accepted arrives later, as three asynchronous acknowledgements: ACK1 confirms the gateway itself queued the message, ACK2 confirms the receiving Center picked it up, and ACK3 carries the outcome of the Center’s own technical validation of the content — the receipt that finally says accepted or rejected. This walkthrough builds the parser and reconciliation logic for all three, as the concrete implementation behind Submission Gateway Transport & Acknowledgement Handling, part of the Electronic Submission Gateway & E-Signature Automation domain.

Why naive approaches fail

Treating acknowledgements as a stream of loosely structured status emails, rather than as typed events that must reconcile to a specific transmission, produces gaps that are invisible until an inspector or a filing deadline exposes them.

  • Reading only the first acknowledgement that arrives. A pipeline that marks a sequence “submitted” as soon as ACK1 arrives has recorded only that the gateway queued the message — not that the Center accepted the content. The submission clock and any downstream reporting must wait for ACK3.
  • Matching acknowledgements by sequence number instead of message id. A sequence can be retransmitted, and its sequence number stays the same across attempts. Matching on sequence number alone conflates receipts from different transmission attempts; matching on the original AS2 Message-Id is the only reliable key.
  • Discarding an acknowledgement that does not parse cleanly. A malformed or unexpected acknowledgement is exactly the kind of anomaly compliance tooling exists to catch. Logging a parse failure and moving on turns a detectable problem into a silent gap in the audit trail.
  • Collapsing all rejection codes into one bucket. ACK3 technical rejections range from a missing checksum to a fatal structural defect. Treating every rejection the same way as a black-box error strips out the severity information a regulatory-affairs reviewer needs to decide whether to resubmit immediately or escalate.

Architecture overview

Each inbound acknowledgement passes through the same short pipeline: parse its structure, look up the transmission it claims to answer, classify its outcome, and update the transmission’s state.

Acknowledgement processing pipeline Five stages left to right: receive ACK, parse receipt, reconcile to transmission, classify outcome, and update state. Each stage feeds the next with a single arrow. Receive ACK Parse receipt Reconcile Classify Update state

Setup and configuration

No specialized parsing library is required; the acknowledgement bodies are structured text or XML depending on the gateway, so the standard library and the transmission store’s connection details are the only configuration needed.

pip install "lxml>=5.2"
"""Environment-driven configuration for acknowledgement processing."""
from __future__ import annotations

import os
from dataclasses import dataclass


def required(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        raise RuntimeError(f"Missing required environment variable: {name}")
    return value


@dataclass(frozen=True)
class AckProcessorConfig:
    """Where inbound acknowledgements are read from and how long to wait."""

    inbox_path: str          # directory or queue the gateway deposits ACKs into
    max_pending_hours: int    # service-level window before a pending ACK pages a human

    @classmethod
    def from_env(cls) -> "AckProcessorConfig":
        return cls(
            inbox_path=required("GATEWAY_ACK_INBOX_PATH"),
            max_pending_hours=int(os.environ.get("GATEWAY_ACK_MAX_PENDING_HOURS", "48")),
        )

Full working implementation

The model separates three concerns that are easy to conflate: the acknowledgement’s own structure, the transmission it reconciles against, and the classification of its outcome by severity.

"""Parse and reconcile FDA ESG-style ACK1/ACK2/ACK3 acknowledgements."""
from __future__ import annotations

import json
import logging
from dataclasses import dataclass
from datetime import datetime, timezone
from enum import Enum

from lxml import etree

logger = logging.getLogger("gateway.ack")


class AckLevel(str, Enum):
    ACK1 = "ack1"  # gateway received and queued the message
    ACK2 = "ack2"  # receiving Center picked up the message
    ACK3 = "ack3"  # Center's technical validation outcome


class AckOutcome(str, Enum):
    ACCEPTED = "accepted"
    REJECTED = "rejected"
    PENDING = "pending"


class RejectionSeverity(str, Enum):
    NONE = "none"
    WARNING = "warning"       # accepted with a note
    HIGH = "high"              # rejected, resubmission required


class ReconciliationError(Exception):
    """Raised when an acknowledgement cannot be matched to a known transmission."""


@dataclass(frozen=True)
class Acknowledgement:
    """One parsed acknowledgement receipt."""

    message_id: str
    ack_level: AckLevel
    outcome: AckOutcome
    severity: RejectionSeverity
    code: str | None
    detail: str
    received_at: datetime


def parse_ack_xml(raw_xml: bytes) -> Acknowledgement:
    """Parse a single acknowledgement document into a typed record.

    Real gateway schemas vary by authority; this parser reads the small,
    stable set of fields every acknowledgement carries and leaves anything
    authority-specific to a thin adapter layered on top.
    """
    root = etree.fromstring(raw_xml)
    message_id = _require_text(root, "originalMessageId")
    ack_level = AckLevel(_require_text(root, "ackLevel").lower())
    status_text = _require_text(root, "status").lower()
    code = _optional_text(root, "code")
    detail = _optional_text(root, "detail") or ""

    outcome = AckOutcome.ACCEPTED if status_text == "accepted" else AckOutcome.REJECTED
    severity = _classify_severity(ack_level, outcome, code)

    return Acknowledgement(
        message_id=message_id,
        ack_level=ack_level,
        outcome=outcome,
        severity=severity,
        code=code,
        detail=detail,
        received_at=datetime.now(timezone.utc),
    )


def _require_text(root: etree._Element, tag: str) -> str:
    node = root.find(tag)
    if node is None or not node.text:
        raise ValueError(f"acknowledgement missing required field: {tag}")
    return node.text.strip()


def _optional_text(root: etree._Element, tag: str) -> str | None:
    node = root.find(tag)
    return node.text.strip() if node is not None and node.text else None


# High-severity codes stop the clock and require resubmission; warning
# codes are recorded but do not block acceptance.
HIGH_SEVERITY_CODES = frozenset({"E-STRUCTURE", "E-CHECKSUM", "E-MODULE", "E-SIGNATURE"})
WARNING_CODES = frozenset({"W-DEPRECATED-FORMAT", "W-NAMING"})


def _classify_severity(ack_level: AckLevel, outcome: AckOutcome, code: str | None) -> RejectionSeverity:
    if outcome is AckOutcome.ACCEPTED:
        return RejectionSeverity.WARNING if code in WARNING_CODES else RejectionSeverity.NONE
    if ack_level is AckLevel.ACK3 and code in HIGH_SEVERITY_CODES:
        return RejectionSeverity.HIGH
    return RejectionSeverity.HIGH if outcome is AckOutcome.REJECTED else RejectionSeverity.NONE


def reconcile(ack: Acknowledgement, known_message_ids: set[str]) -> None:
    """Raise if the acknowledgement does not reference a transmission this system sent."""
    if ack.message_id not in known_message_ids:
        raise ReconciliationError(
            f"{ack.ack_level.value} references unknown message id {ack.message_id!r}"
        )


def record_ack(transmission_id: str, ack: Acknowledgement) -> None:
    event = {
        "action": "acknowledgement_received",
        "transmission_id": transmission_id,
        "message_id": ack.message_id,
        "ack_level": ack.ack_level.value,
        "outcome": ack.outcome.value,
        "severity": ack.severity.value,
        "code": ack.code,
    }
    logger.info(json.dumps(event, sort_keys=True))

A sequence is only resolved once its ACK3 has arrived and reconciled cleanly; ACK1 and ACK2 update an intermediate state that downstream reporting can surface as “in transit,” never as “filed.”

"""Determine whether a transmission is fully resolved."""
from __future__ import annotations


def is_resolved(acks_received: dict[AckLevel, Acknowledgement]) -> bool:
    ack3 = acks_received.get(AckLevel.ACK3)
    return ack3 is not None and ack3.outcome in {AckOutcome.ACCEPTED, AckOutcome.REJECTED}


def final_outcome(acks_received: dict[AckLevel, Acknowledgement]) -> AckOutcome:
    ack3 = acks_received.get(AckLevel.ACK3)
    return ack3.outcome if ack3 else AckOutcome.PENDING

Validation and edge-case handling

A few patterns cover most of the anomalies an acknowledgement stream produces in practice:

  • Out-of-order arrival. ACK2 occasionally arrives after ACK3 due to independent delivery paths inside the gateway infrastructure. The reconciliation model above stores each acknowledgement by level in a dict keyed off the transmission, so out-of-order arrival does not corrupt state — is_resolved only depends on ACK3 having arrived, regardless of when ACK1 and ACK2 showed up.
  • A duplicate acknowledgement for the same level. Redelivery at the transport layer can result in the same ACK3 arriving twice. Storing acknowledgements keyed by (message_id, ack_level) and treating a repeat as an idempotent no-op — rather than a state change — prevents a duplicate delivery from re-triggering downstream notifications.
  • A high-severity ACK3 rejection. This is not a system failure; it is the Center telling the sponsor the sequence has a defect. The correct handling routes the sequence and its specific rejection code back toward assembly or signing for correction, exactly as described in eCTD Sequence Assembly & Lifecycle Management, rather than retrying transport.
  • An acknowledgement that never arrives. If ACK3 has not arrived within the configured service-level window, that is itself an event worth raising — a stuck transmission is either a gateway-side problem or a lost receipt, and either way a human needs to look at it before the filing deadline passes.

Testing and verification

Because the classification rules carry real compliance weight — a warning should never be mistaken for a high-severity rejection, and vice versa — they deserve direct test coverage alongside the reconciliation logic.

"""Tests for acknowledgement parsing and reconciliation."""
from __future__ import annotations

import pytest


ACCEPTED_ACK3 = b"""<ack><originalMessageId>seq-0007.abc@sponsor.example</originalMessageId>
<ackLevel>ACK3</ackLevel><status>accepted</status></ack>"""

REJECTED_ACK3 = b"""<ack><originalMessageId>seq-0007.abc@sponsor.example</originalMessageId>
<ackLevel>ACK3</ackLevel><status>rejected</status><code>E-CHECKSUM</code>
<detail>leaf checksum mismatch in m3.2.p.1</detail></ack>"""


def test_parse_accepted_ack3() -> None:
    ack = parse_ack_xml(ACCEPTED_ACK3)
    assert ack.outcome == AckOutcome.ACCEPTED
    assert ack.severity == RejectionSeverity.NONE


def test_parse_rejected_ack3_is_high_severity() -> None:
    ack = parse_ack_xml(REJECTED_ACK3)
    assert ack.outcome == AckOutcome.REJECTED
    assert ack.severity == RejectionSeverity.HIGH
    assert ack.code == "E-CHECKSUM"


def test_reconcile_raises_for_unknown_message_id() -> None:
    ack = parse_ack_xml(ACCEPTED_ACK3)
    with pytest.raises(ReconciliationError):
        reconcile(ack, known_message_ids={"some-other-id@sponsor.example"})


def test_resolved_only_after_ack3() -> None:
    ack1 = parse_ack_xml(ACCEPTED_ACK3.replace(b"ACK3", b"ACK1"))
    assert not is_resolved({AckLevel.ACK1: ack1})
    ack3 = parse_ack_xml(ACCEPTED_ACK3)
    assert is_resolved({AckLevel.ACK1: ack1, AckLevel.ACK3: ack3})

The first two tests prove acceptance and rejection are classified correctly and that a checksum rejection is tagged high severity; the third proves an acknowledgement for an unrecognized message id is rejected rather than silently accepted; the fourth proves a transmission is not considered resolved until its ACK3 has actually arrived. Together they defend the two properties that matter most for an inspector: nothing is called “filed” prematurely, and no acknowledgement is dropped on the floor.