fix(ingestion): give EDI events the shipping KDEs validation requires

services/ingestion/app/edi_ingestion/extractors.py

@@ -5,7 +5,7 @@
 
 from .constants import _FROM_ENTITY_CODES, _TO_ENTITY_CODES
 from .parser import _first_segment
-from .utils import _safe_float
+from .utils import _safe_float, _valid_gln_or_none
 
 
 def _extract_envelope_ids(segments: list[list[str]]) -> dict[str, Any]:
@@ -503,6 +503,7 @@ def _extract_861_fields(segments: list[list[str]]) -> dict[str, Any]:
     data["reference_document_number"] = data["receiving_advice_number"]
 
     items = _ItemLines()
+    has_explicit_ship_from = False
     for segment in segments:
         if not segment:
             continue
@@ -512,12 +513,26 @@ def _extract_861_fields(segments: list[list[str]]) -> dict[str, Any]:
             entity_code = segment[1].upper() if len(segment) > 1 else ""
             name = segment[2] if len(segment) > 2 and segment[2] else None
             location_id = segment[4] if len(segment) > 4 and segment[4] else None
-            if entity_code in _FROM_ENTITY_CODES:
+            # Partners also send GLNs under qualifier 92. Accept a value only
+            # when it is itself a valid GLN, without sanitizing an internal ID.
+            gln = (
+                _valid_gln_or_none(location_id)
+                if location_id and location_id.isascii() and location_id.isdigit()
+                else None
+            )
+            if entity_code == "SF" and (name or gln):
+                if not has_explicit_ship_from or (name and not data["ship_from_name"]):
+                    # An actual ship-from overrides SU/SH; a later named SF
+                    # improves an identifier-only SF without mixing parties.
+                    has_explicit_ship_from = True
+                    data["ship_from_name"] = name
+                    data["ship_from_gln"] = gln
+            elif entity_code in _FROM_ENTITY_CODES and not has_explicit_ship_from:
                 data["ship_from_name"] = data["ship_from_name"] or name
-                data["ship_from_gln"] = data["ship_from_gln"] or location_id
+                data["ship_from_gln"] = data["ship_from_gln"] or gln
             elif entity_code in _TO_ENTITY_CODES:
                 data["receiving_location_name"] = data["receiving_location_name"] or name
-                data["receiving_location_gln"] = data["receiving_location_gln"] or location_id
+                data["receiving_location_gln"] = data["receiving_location_gln"] or gln
 
         elif seg_id == "RCD":
             # RCD[5] is the condition code this extractor has always read.

services/ingestion/app/edi_ingestion/routes.py

@@ -576,6 +576,9 @@ 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="Optional TLC source reference (where the TLC was assigned) copied to each event's KDEs"
+    ),
     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"),
@@ -756,6 +759,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",
@@ -940,6 +946,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,
     }
 
@@ -983,6 +993,7 @@ def _build_ingest_event_for_set(
                 "receiving_advice_number": extracted.get("receiving_advice_number"),
                 "po_number": extracted.get("po_number"),
                 "ship_from_location": extracted.get("ship_from_name"),
+                "immediate_previous_source": extracted.get("ship_from_name") or extracted.get("ship_from_gln"),
                 "receiving_location": extracted.get("receiving_location_name"),
                 "receiving_location_gln": extracted.get("receiving_location_gln"),
                 "condition_code": extracted.get("condition_code"),
@@ -1053,18 +1064,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 +1111,7 @@ 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="Optional TLC source reference copied to every event's KDEs"
     ),
     tenant_id: Optional[str] = Form(None, description="Optional tenant override"),
     source: str = Form("edi_inbound", description="Source system identifier"),
@@ -1243,12 +1255,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 +1285,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.

services/ingestion/tests/test_edi_856_required_kdes.py

@@ -0,0 +1,209 @@
+"""EDI 856 and 861 events must carry the KDEs that validation requires.
+
+Both EDI routes built 856 events with ``reference_document_number`` only and
+never set ``tlc_source_reference``, while ``REQUIRED_KDES_BY_CTE[SHIPPING]``
+requires ``reference_document`` and ``tlc_source_reference``. Every 856 was
+therefore rejected in ``ingest_events``. These checks post the recorded ASN
+fixture through each route and run the captured events through the real
+KDE validator.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+service_dir = Path(__file__).parent.parent
+sys.path.insert(0, str(service_dir))
+
+pytest.importorskip("fastapi")
+
+import app.edi_ingestion.routes as edi_ingestion_routes  # noqa: E402
+from app.authz import IngestionPrincipal, get_ingestion_principal  # noqa: E402
+from app.edi_ingestion import router as edi_router  # noqa: E402
+from app.edi_ingestion.extractors import _extract_861_fields  # noqa: E402
+from app.edi_ingestion.parser import _parse_x12_segments  # noqa: E402
+from app.webhook_models import IngestResponse, WebhookPayload  # noqa: E402
+from app.webhook_router_v2 import _validate_event_kdes  # noqa: E402
+
+_ASN_856 = Path(__file__).parent / "fixtures" / "edi_corpus" / "asn_856.edi"
+_TENANT = "11111111-1111-1111-1111-111111111111"
+
+
+@pytest.fixture()
+def captured(monkeypatch: pytest.MonkeyPatch) -> list[WebhookPayload]:
+    payloads: list[WebhookPayload] = []
+
+    async def _capture(payload: WebhookPayload, **_kwargs) -> IngestResponse:
+        payloads.append(payload)
+        return IngestResponse(accepted=0, rejected=0, total=len(payload.events), events=[])
+
+    monkeypatch.setattr(edi_ingestion_routes, "ingest_events", _capture)
+    monkeypatch.setattr(edi_ingestion_routes, "_resolve_principal_tenant_id", lambda *args: _TENANT)
+    monkeypatch.setattr(edi_ingestion_routes, "check_and_record_interchange", lambda *args, **kwargs: (False, None))
+    monkeypatch.delenv("EDI_REQUIRE_PARTNER_ID", raising=False)
+    monkeypatch.delenv("EDI_PARTNER_ALLOWLIST", raising=False)
+    return payloads
+
+
+def _post(route: str, data: dict[str, str], body: bytes | None = None):
+    app = FastAPI()
+    app.include_router(edi_router)
+    app.dependency_overrides[get_ingestion_principal] = lambda: IngestionPrincipal(
+        key_id="test", scopes=["edi.ingest"], tenant_id=_TENANT
+    )
+    with TestClient(app) as client:
+        return client.post(
+            f"/api/v1/ingest/edi{route}",
+            data=data,
+            files={"file": ("document.edi", body or _ASN_856.read_bytes(), "application/edi-x12")},
+        )
+
+
+@pytest.mark.parametrize("route", ["", "/document"])
+def test_856_with_tlc_source_reference_passes_kde_validation(route: str, captured) -> None:
+    response = _post(route, {"traceability_lot_code": "LOT-856-KDE", "tlc_source_reference": " PACK-0310 "})
+
+    assert response.status_code == 201, response.text
+    events = [event for payload in captured for event in payload.events]
+    assert events
+    for event in events:
+        assert event.kdes["reference_document"] == "BOL-9001"
+        assert event.kdes["tlc_source_reference"] == "PACK-0310"
+        assert _validate_event_kdes(event) == []
+
+
+@pytest.mark.parametrize("route", ["", "/document"])
+def test_856_without_tlc_source_reference_only_misses_that_kde(route: str, captured) -> None:
+    response = _post(route, {"traceability_lot_code": "LOT-856-KDE"})
+
+    assert response.status_code == 201, response.text
+    events = [event for payload in captured for event in payload.events]
+    assert events
+    for event in events:
+        assert _validate_event_kdes(event) == ["Missing required KDE 'tlc_source_reference' for shipping CTE"]
+
+
+def _receiving_advice_861(*ship_from: str) -> bytes:
+    body = [
+        "ST*861*0001",
+        "BRA*RA-861-1*20260311*00",
+        *ship_from,
+        "N1*ST*Metro Distribution Center*92*6141410000069",
+        "RCD**200*CA",
+        "REF*LT*00012345678901-LOT861",
+        f"SE*{len(ship_from) + 6}*0001",
+    ]
+    segments = [
+        "ISA*00*          *00*          *ZZ*SENDERID       *ZZ*RECEIVERID     *260311*1200*U*00401*000000861*0*P*>",
+        "GS*RE*SENDERID*RECEIVERID*20260311*1200*1*X*004010",
+        *body,
+        "GE*1*1",
+        "IEA*1*000000861",
+    ]
+    return ("~".join(segments) + "~").encode()
+
+
+@pytest.mark.parametrize(
+    ("ship_from", "expected"),
+    [
+        ("N1*SF*Valley Fresh Farms*92*6141410000014", "Valley Fresh Farms"),
+        ("N1*SF**92*6141410000014", "6141410000014"),
+    ],
+)
+def test_861_takes_immediate_previous_source_from_ship_from(ship_from: str, expected: str, captured) -> None:
+    response = _post(
+        "/document",
+        {"traceability_lot_code": "LOT-861-KDE", "tlc_source_reference": "PACK-0311"},
+        _receiving_advice_861(ship_from),
+    )
+
+    assert response.status_code == 201, response.text
+    events = [event for payload in captured for event in payload.events]
+    assert events
+    for event in events:
+        assert event.kdes["immediate_previous_source"] == expected
+        assert _validate_event_kdes(event) == []
+
+
+def test_861_without_ship_from_only_misses_immediate_previous_source(captured) -> None:
+    response = _post(
+        "/document",
+        {"traceability_lot_code": "LOT-861-KDE", "tlc_source_reference": "PACK-0311"},
+        _receiving_advice_861(),
+    )
+
+    assert response.status_code == 201, response.text
+    events = [event for payload in captured for event in payload.events]
+    assert events
+    for event in events:
+        assert "immediate_previous_source" not in event.kdes
+        assert _validate_event_kdes(event) == ["Missing required KDE 'immediate_previous_source' for receiving CTE"]
+
+
+_SUPPLIER = "N1*SU*Fresh Supply Co*92*0614141000005"
+_SHIP_FROM = "N1*SF*Valley Fresh Farms*92*6141410000014"
+
+
+@pytest.mark.parametrize(
+    ("parties", "expected_source", "expected_gln"),
+    [
+        ((_SUPPLIER, _SHIP_FROM), "Valley Fresh Farms", "6141410000014"),
+        ((_SHIP_FROM, _SUPPLIER), "Valley Fresh Farms", "6141410000014"),
+        ((_SUPPLIER,), "Fresh Supply Co", "0614141000005"),
+        ((_SUPPLIER, "N1*SF"), "Fresh Supply Co", "0614141000005"),
+    ],
+)
+def test_861_prefers_the_ship_from_party_over_a_supplier(
+    parties: tuple[str, ...], expected_source: str, expected_gln: str, captured
+) -> None:
+    response = _post(
+        "/document",
+        {"traceability_lot_code": "LOT-861-KDE", "tlc_source_reference": "PACK-0311"},
+        _receiving_advice_861(*parties),
+    )
+
+    assert response.status_code == 201, response.text
+    events = [event for payload in captured for event in payload.events]
+    assert events
+    for event in events:
+        assert event.kdes["immediate_previous_source"] == expected_source
+        assert event.kdes["ship_from_location"] == expected_source
+        assert _validate_event_kdes(event) == []
+    assert _extract_861_fields(_parse_x12_segments(_receiving_advice_861(*parties).decode()))["ship_from_gln"] == (
+        expected_gln
+    )
+
+
+@pytest.mark.parametrize(
+    "parties",
+    [
+        ("N1*SF**92*6141410000014", _SHIP_FROM),
+        (_SHIP_FROM, "N1*SF**92*6141410000014"),
+    ],
+)
+def test_861_named_ship_from_wins_over_identifier_only(parties, captured):
+    """Do not discard an available SF name because its identifier came first."""
+    response = _post(
+        "/document",
+        {"traceability_lot_code": "LOT-861-KDE", "tlc_source_reference": "PACK-0311"},
+        _receiving_advice_861(*parties),
+    )
+    assert response.status_code == 201, response.text
+    event = captured[0].events[0]
+    assert event.kdes["immediate_previous_source"] == "Valley Fresh Farms"
+    assert _validate_event_kdes(event) == []
+
+
+@pytest.mark.parametrize("identifier", ["WAREHOUSE-1", "6141410000015", "X6141410000014"])
+@pytest.mark.parametrize("with_supplier", [False, True])
+def test_861_invalid_identifier_is_not_a_gln(identifier, with_supplier):
+    """Invalid identifier-only SF must neither become a GLN nor erase a name."""
+    parties = ((_SUPPLIER,) if with_supplier else ()) + (f"N1*SF**92*{identifier}",)
+    extracted = _extract_861_fields(_parse_x12_segments(_receiving_advice_861(*parties).decode()))
+    assert extracted["ship_from_name"] == ("Fresh Supply Co" if with_supplier else None)
+    assert extracted["ship_from_gln"] == ("0614141000005" if with_supplier else None)

services/ingestion/tests/test_r3_restored_fixtures_e2e.py

@@ -144,7 +144,9 @@ def test_edi_856_fixture_end_to_end():
     transition, gaps = gaps_by_tlc[tlc]
     assert transition == "open"
     assert {g.gap_kind for g in gaps} == {GAP_KIND_MISSING_KDE}
-    assert {"ship_date", "tlc_source_reference"} <= {g.gap_key for g in gaps}
+    # The 856 carries its ship date (DTM) and BOL; only the TLC source
+    # reference has no X12 home, so it stays a gap unless the caller sends it.
+    assert {g.gap_key for g in gaps} == {"tlc_source_reference"}
 
     headers, rows = _parse_sheet(sheet)
     assert headers == FDA_COLUMNS