feat(edi): store shipments missing a TLC source reference with an open gap

docs/architecture/RULES_ENFORCEMENT.md

@@ -30,6 +30,8 @@ Set via `RULES_ENGINE_ENFORCE` (read at call time, no restart needed — flip +
 | `cte_only` | Reject when `summary.critical_failures` is non-empty. Warnings/info still accept. |
 | `all` | Reject on any `summary.compliant is False` (including warning-only failures). |
 
+`ingest_events` passes `deferred_kdes` for EDI shipments stored with an open data gap. Under `all`, a warning that fails only on those KDEs does not reject. The verdict stays `fail`, and any other failure on the event still rejects.
+
 Two invariants hold in every mode:
 
 1. **No-verdict never rejects.** When `summary.compliant is None` (no rules seeded for the tenant, non-FTL product, evaluator skipped), the event accepts — "we didn't evaluate" is not the same as "the event violates a rule."

services/ingestion/app/edi_ingestion/routes.py

@@ -29,7 +29,7 @@
 import logging
 import os
 import uuid
-from collections.abc import Callable
+from collections.abc import Callable, Iterable
 from typing import Any, Optional
 
 from fastapi import APIRouter, Depends, File, Form, Header, HTTPException, Query, UploadFile
@@ -73,10 +73,16 @@
     _valid_gln_or_none,
     _verify_partner_id,
 )
-from .validation import _validate_edi_as_fsma_event
+from .validation import _validate_edi_as_fsma_event, _warehouse_line_tlc
 
 logger = logging.getLogger("edi-ingestion")
 
+# X12 has no element for the TLC source reference (21 CFR 1.1340(a)(7)), so a
+# shipment arrives without it unless the uploader sends the form field. Store
+# the shipment with an open data gap rather than lose the rest of the record.
+# Receiving keeps the strict gate, as its rule for this KDE is critical.
+_EDI_DEFERRED_KDES = {WebhookCTEType.SHIPPING: frozenset({"tlc_source_reference"})}
+
 
 def _record_edi_rejection(**kwargs):
     """Resolve the live rejection-log module before writing.
@@ -93,6 +99,70 @@ def _record_edi_rejection(**kwargs):
     return rejection_log.record_edi_rejection(**kwargs)
 
 
+def _reject_shared_tlc_source_for_distinct_lots(
+    *,
+    tlc_source_reference: str | None,
+    transaction_sets: Iterable[tuple[str, dict[str, Any]]],
+    traceability_lot_code: str,
+    sender_tenant_id: str,
+    partner_id: str | None,
+    source: str,
+) -> None:
+    """Refuse one TLC source reference across several shipped lots.
+
+    The form field is a single value copied to every event. On a document with
+    several lots it could link a lot to a source that never assigned its code,
+    which is worse than the open gap the lot gets without the field. Receiving
+    has no open gap to fall back on, as the field is required there, so
+    refusing it would leave a multi-lot receipt no way in.
+    """
+    if not (tlc_source_reference and tlc_source_reference.strip()):
+        return
+    shipping_sets = [
+        (transaction_set, extracted)
+        for transaction_set, extracted in transaction_sets
+        if _CTE_TYPE_BY_SET.get(transaction_set, WebhookCTEType.SHIPPING) == WebhookCTEType.SHIPPING
+    ]
+    lots: set[str] = set()
+    for transaction_set, extracted in shipping_sets:
+        for line in extracted.get("line_items") or [{}]:
+            lot = line.get("lot_number")
+            lot = lot.strip() if isinstance(lot, str) else ""
+            if not lot:
+                lots.add(traceability_lot_code)
+            elif transaction_set == "945":
+                lots.add(_warehouse_line_tlc(lot, line))
+            else:
+                lots.add(lot)
+    if len(lots) <= 1:
+        return
+    error = "edi_tlc_source_reference_with_multiple_lots"
+    message = (
+        "tlc_source_reference applies to one lot, but this document has several. "
+        "Send each lot separately, or resubmit without the field."
+    )
+    first_set, first_extracted = shipping_sets[0]
+    record = _record_edi_rejection(
+        tenant_id=sender_tenant_id,
+        transaction_set=first_set,
+        traceability_lot_code=traceability_lot_code,
+        errors=[{"loc": ["tlc_source_reference"], "type": error, "msg": message}],
+        extracted=first_extracted,
+        partner_id=partner_id,
+        source=source,
+        reason=error,
+    )
+    raise HTTPException(
+        status_code=422,
+        detail={
+            "error": error,
+            "message": message,
+            "lot_count": len(lots),
+            "rejection_id": record["rejection_id"],
+        },
+    )
+
+
 # ISA13 claims taken by ``_enforce_envelope_integrity`` during the current
 # request, as ``release_interchange`` arguments. ``_release_isa13_unless_stored``
 # installs the list.
@@ -576,6 +646,13 @@ def _enforce_interchange_structure(
 async def ingest_edi_856(
     file: UploadFile = File(..., description="X12 EDI file (856 ASN)"),
     traceability_lot_code: str = Form(..., description="FSMA Traceability Lot Code"),
+    tlc_source_reference: Optional[str] = Form(
+        None,
+        description=(
+            "TLC source reference (where the TLC was assigned), copied to each event's KDEs. "
+            "X12 cannot carry it: a shipment without it is stored with an open data gap."
+        ),
+    ),
     tenant_id: Optional[str] = Form(None, description="Optional tenant override"),
     source: str = Form("edi_856_inbound", description="Source system identifier"),
     product_description: Optional[str] = Form(None, description="Optional product description override"),
@@ -670,19 +747,25 @@ async def ingest_edi_856(
 
     # Event-date rejections run after the allowlist and before the ISA13
     # is reserved, so a corrected resend is not refused as a 409 replay.
-    _enforce_envelope_integrity(
-        extracted,
-        sender_tenant_id,
-        "856",
-        before_reserve=lambda: _preflight_event_date(
+    def _preflight() -> None:
+        _preflight_event_date(
             transaction_set="856",
             extracted=extracted,
             sender_tenant_id=sender_tenant_id,
             traceability_lot_code=traceability_lot_code,
             partner_id=x_partner_id,
             source=source,
-        ),
-    )
+        )
+        _reject_shared_tlc_source_for_distinct_lots(
+            tlc_source_reference=tlc_source_reference,
+            transaction_sets=[("856", extracted)],
+            traceability_lot_code=traceability_lot_code,
+            sender_tenant_id=sender_tenant_id,
+            partner_id=x_partner_id,
+            source=source,
+        )
+
+    _enforce_envelope_integrity(extracted, sender_tenant_id, "856", before_reserve=_preflight)
 
     lines = extracted.get("line_items") or []
     if len(lines) > 1:
@@ -756,6 +839,9 @@ async def ingest_edi_856(
             "ship_from_gln": ship_from_gln or ship_from_gln_raw,
             "ship_to_gln": ship_to_gln or ship_to_gln_raw,
             "reference_document_number": extracted.get("reference_document_number"),
+            # Shipping validation requires these exact names (21 CFR 1.1340(a)(7)-(8)).
+            "reference_document": extracted.get("reference_document_number"),
+            "tlc_source_reference": tlc_source_reference.strip() if tlc_source_reference else None,
             "carrier": extracted.get("carrier"),
             "asn_number": extracted.get("asn_number"),
             "edi_transaction_set": "856",
@@ -796,6 +882,7 @@ async def ingest_edi_856(
         payload,
         principal_tenant_id=sender_tenant_id,
         x_regengine_api_key=x_regengine_api_key,
+        deferred_kdes=_EDI_DEFERRED_KDES,
     )
 
     return EDIIngestResponse(
@@ -940,6 +1027,10 @@ def _build_ingest_event_for_set(
         "edi_isa13": extracted.get("isa13"),
         "edi_gs_control_number": extracted.get("gs_control_number"),
         "reference_document_number": extracted.get("reference_document_number"),
+        # Shipping/receiving validation requires these exact names
+        # (21 CFR 1.1340(a)(7)-(8), 1.1345(a)(7)-(8)) for every set.
+        "reference_document": extracted.get("reference_document_number"),
+        "tlc_source_reference": extracted.get("tlc_source_reference"),
         "event_date": event_date,
     }
 
@@ -1053,18 +1144,19 @@ def _build_ingest_event_for_set(
                 "g62_date_qualifier": extracted.get("g62_date_qualifier"),
                 "receive_date_raw": extracted.get("receive_date_raw"),
                 "ship_date_raw": extracted.get("ship_date_raw"),
-                "reference_document": extracted.get("reference_document_number"),
-                "tlc_source_reference": extracted.get("tlc_source_reference"),
             }
         )
         if transaction_set == "944":
-            kdes["receive_date"] = event_date
             kdes["immediate_previous_source"] = extracted.get("ship_from_name") or extracted.get("ship_from_gln")
-        else:
-            kdes["ship_date"] = event_date
         if extracted.get("depositor_gln"):
             kdes["depositor_gln"] = extracted["depositor_gln"]
 
+    # Every set needs the CTE's date KDE, not only 944/945.
+    if cte_type == WebhookCTEType.RECEIVING:
+        kdes["receive_date"] = event_date
+    else:
+        kdes["ship_date"] = event_date
+
     return IngestEvent(
         cte_type=cte_type,
         traceability_lot_code=traceability_lot_code,
@@ -1099,7 +1191,11 @@ async def ingest_edi_document(
         description="FSMA Traceability Lot Code for 856/850/810/861; required but ignored for 944/945 event TLCs, which use each line's lot code",
     ),
     tlc_source_reference: Optional[str] = Form(
-        None, description="Optional TLC source reference copied to 944/945 KDEs"
+        None,
+        description=(
+            "TLC source reference, copied to every event's KDEs. X12 cannot carry it: "
+            "a shipment without it is stored with an open data gap."
+        ),
     ),
     tenant_id: Optional[str] = Form(None, description="Optional tenant override"),
     source: str = Form("edi_inbound", description="Source system identifier"),
@@ -1233,6 +1329,14 @@ def _preflight_all_sets() -> None:
                 partner_id=x_partner_id,
                 source=source,
             )
+        _reject_shared_tlc_source_for_distinct_lots(
+            tlc_source_reference=tlc_source_reference,
+            transaction_sets=list(zip(slice_sets, extractions)),
+            traceability_lot_code=traceability_lot_code,
+            sender_tenant_id=sender_tenant_id,
+            partner_id=x_partner_id,
+            source=source,
+        )
 
     # Event-date rejections run after the allowlist and before the ISA13
     # is reserved, so a corrected resend is not refused as a 409 replay.
@@ -1243,12 +1347,13 @@ def _preflight_all_sets() -> None:
     # the shared header is copied onto each line.
     event_inputs: list[tuple[str, dict[str, Any], list[list[str]]]] = []
     first_event_indexes: list[int] = []
+    tlc_source_ref = tlc_source_reference.strip() if tlc_source_reference else None
     for slice_set, slice_extracted, slice_segments in zip(slice_sets, extractions, set_slices):
         first_event_indexes.append(len(event_inputs))
         if slice_set in ("856", "861", "944", "945"):
             header = {key: value for key, value in slice_extracted.items() if key != "line_items"}
+            header["tlc_source_reference"] = tlc_source_ref
             if slice_set in ("944", "945"):
-                header["tlc_source_reference"] = tlc_source_reference.strip() if tlc_source_reference else None
                 date_field = "receive_date_raw" if slice_set == "944" else "ship_date_raw"
                 header["_event_timestamp"], header["_event_date"] = _ship_timestamp_and_date(
                     header.get(date_field), None
@@ -1272,7 +1377,8 @@ def _preflight_all_sets() -> None:
                     }
                 event_inputs.append((slice_set, merged, slice_segments))
         else:
-            event_inputs.append((slice_set, slice_extracted, slice_segments))
+            with_ref = {**slice_extracted, "tlc_source_reference": tlc_source_ref}
+            event_inputs.append((slice_set, with_ref, slice_segments))
 
     # #1174: FSMA validation is fail-closed by default. A schema failure
     # returns HTTP 422 with per-field errors and nothing is persisted.
@@ -1510,6 +1616,7 @@ def _summarize(slice_set: str, slice_extracted: dict[str, Any]) -> dict[str, Any
         payload,
         principal_tenant_id=sender_tenant_id,
         x_regengine_api_key=x_regengine_api_key,
+        deferred_kdes=_EDI_DEFERRED_KDES,
     )
 
     return EDIIngestResponse(

services/ingestion/app/edi_ingestion/validation.py

@@ -23,6 +23,32 @@ class _WarehouseLineRequiredKDEs(BaseModel):
     quantity: float = Field(gt=0)
 
 
+def _warehouse_line_tlc(lot: str, line: dict[str, Any]) -> str:
+    """The TLC stored for a warehouse line with a non-blank lot.
+
+    A lot that isn't production-shaped gets the line's GTIN or UPC as a prefix,
+    so one short lot on two items is two TLCs.
+    """
+    from shared.schemas import _TLC_PRODUCTION_RE, FSMAEvent
+
+    tlc = lot
+    # Pattern misses are advisory, so check explicitly to retain the
+    # warehouse GTIN/UPC rebuild for both accepted and short lots.
+    try:
+        tlc = FSMAEvent.validate_tlc_format(tlc)
+        rebuild = not _TLC_PRODUCTION_RE.fullmatch(tlc)
+    except ValueError:
+        rebuild = True
+    if rebuild:
+        for item_id, lengths in (
+            (line.get("gtin"), (8, 12, 13, 14)),
+            (line.get("upc"), (12, 13)),
+        ):
+            if isinstance(item_id, str) and item_id.isascii() and item_id.isdigit() and len(item_id) in lengths:
+                return f"{item_id.zfill(14)}-{tlc}"
+    return tlc
+
+
 def _validate_edi_as_fsma_event(
     extracted: dict[str, Any],
     transaction_set: str,
@@ -39,7 +65,7 @@ def _validate_edi_as_fsma_event(
     forced every caller to treat ``None`` as "maybe-failed, maybe-
     unavailable" and masked schema errors in the audit trail.
     """
-    from shared.schemas import _TLC_PRODUCTION_RE, FSMAEvent, FSMAEventType
+    from shared.schemas import FSMAEvent, FSMAEventType
 
     type_map = {
         "856": FSMAEventType.SHIPPING,
@@ -62,21 +88,7 @@ def _validate_edi_as_fsma_event(
         # TLC that FSMAEvent would accept, collapsing blank lots onto one id.
         if not tlc:
             _WarehouseLineRequiredKDEs(lot_number=tlc, quantity=line.quantity)
-        # Pattern misses are advisory, so check explicitly to retain the
-        # warehouse GTIN/UPC rebuild for both accepted and short lots.
-        try:
-            tlc = FSMAEvent.validate_tlc_format(tlc)
-            rebuild = not _TLC_PRODUCTION_RE.fullmatch(tlc)
-        except ValueError:
-            rebuild = True
-        if rebuild:
-            for item_id, lengths in (
-                (extracted.get("gtin"), (8, 12, 13, 14)),
-                (extracted.get("upc"), (12, 13)),
-            ):
-                if isinstance(item_id, str) and item_id.isascii() and item_id.isdigit() and len(item_id) in lengths:
-                    tlc = f"{item_id.zfill(14)}-{tlc}"
-                    break
+        tlc = _warehouse_line_tlc(tlc, extracted)
     elif transaction_set in ("856", "861"):
         # The line lot is the TLC verbatim. Do not prefix a GTIN (#1104).
         # A line with no lot keeps the caller-supplied form field.