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

services/ingestion/app/edi_ingestion/routes.py

@@ -518,6 +518,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"),
@@ -698,6 +701,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",
@@ -882,6 +888,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,
     }
 
@@ -995,18 +1005,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,
@@ -1040,7 +1051,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"),
@@ -1184,12 +1195,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
@@ -1213,7 +1225,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,88 @@
+"""EDI 856 events must carry the KDEs that shipping 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.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]):
+    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": ("asn_856.edi", _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"
+        ]

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