From 3a517781c75da09b637d9e5be1bf97058d41a3a3 Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Sat, 31 Oct 2026 01:48:42 -0700 Subject: [PATCH 1/6] fix(ingestion): give EDI events the shipping KDEs validation requires Every EDI 856 was rejected in ingest_events. REQUIRED_KDES_BY_CTE[SHIPPING] requires reference_document, tlc_source_reference and ship_date, but: - both EDI routes stored the BOL only as reference_document_number; - tlc_source_reference was only copied for 944/945, and the legacy /api/v1/ingest/edi route had no way to accept it; - /document set ship_date/receive_date only inside the 944/945 branch, so 856/850/810/861 events had no CTE date KDE. Now both routes set reference_document from the extracted BOL (keeping reference_document_number), the legacy route accepts tlc_source_reference, /document copies it to every set, and the date KDE is chosen by CTE type for every set. An 856 without tlc_source_reference is still rejected, but for that one field only: X12 has no place for it. test_r3_restored_fixtures_e2e asserted the old behavior (ship_date as an open gap on the 856 fixture); it now expects only tlc_source_reference. Co-Authored-By: Claude Opus 5.5 --- .../ingestion/app/edi_ingestion/routes.py | 29 ++++-- .../tests/test_edi_856_required_kdes.py | 88 +++++++++++++++++++ .../tests/test_r3_restored_fixtures_e2e.py | 4 +- 3 files changed, 112 insertions(+), 9 deletions(-) create mode 100644 services/ingestion/tests/test_edi_856_required_kdes.py diff --git a/services/ingestion/app/edi_ingestion/routes.py b/services/ingestion/app/edi_ingestion/routes.py index 8d36158ff..d5abeec1c 100644 --- a/services/ingestion/app/edi_ingestion/routes.py +++ b/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. diff --git a/services/ingestion/tests/test_edi_856_required_kdes.py b/services/ingestion/tests/test_edi_856_required_kdes.py new file mode 100644 index 000000000..d04a8f795 --- /dev/null +++ b/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" + ] diff --git a/services/ingestion/tests/test_r3_restored_fixtures_e2e.py b/services/ingestion/tests/test_r3_restored_fixtures_e2e.py index fb062a9f2..0bb311290 100644 --- a/services/ingestion/tests/test_r3_restored_fixtures_e2e.py +++ b/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 From 1073577659981d54890ae417c93b6b5da09b9b17 Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:19:01 -0800 Subject: [PATCH 2/6] feat(edi): store shipments missing a TLC source reference with an open gap X12 has no element for tlc_source_reference (21 CFR 1.1340(a)(7)), so an 856 sent without the form field was rejected and lost. Product decision: store it, and record that the KDE is outstanding. ingest_events takes a server-side deferred_kdes argument, set only by the EDI routes, for shipping and this one KDE. An event missing only deferred KDEs is stored, reports them as open_gaps, and opens a missing_kde gap without closing the lot's existing gaps. Any other missing KDE still rejects, and receiving keeps the strict gate. The exemption is never read from the payload, which the sender controls. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> --- .../ingestion/app/edi_ingestion/routes.py | 20 +- services/ingestion/app/webhook_compat.py | 25 ++- services/ingestion/app/webhook_models.py | 4 + .../tests/test_edi_856_required_kdes.py | 18 +- .../tests/test_edi_date_reject_resend.py | 2 +- .../ingestion/tests/test_edi_deferred_kdes.py | 187 ++++++++++++++++++ .../test_edi_wire_format_2974_2976_2978.py | 2 +- 7 files changed, 247 insertions(+), 11 deletions(-) create mode 100644 services/ingestion/tests/test_edi_deferred_kdes.py diff --git a/services/ingestion/app/edi_ingestion/routes.py b/services/ingestion/app/edi_ingestion/routes.py index d5abeec1c..050522dfe 100644 --- a/services/ingestion/app/edi_ingestion/routes.py +++ b/services/ingestion/app/edi_ingestion/routes.py @@ -74,6 +74,12 @@ 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. @@ -519,7 +525,11 @@ 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" + 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"), @@ -744,6 +754,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( @@ -1051,7 +1062,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 every event's 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"), @@ -1464,6 +1479,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( diff --git a/services/ingestion/app/webhook_compat.py b/services/ingestion/app/webhook_compat.py index 74ced5e92..a0343c492 100644 --- a/services/ingestion/app/webhook_compat.py +++ b/services/ingestion/app/webhook_compat.py @@ -20,7 +20,7 @@ import logging import uuid from dataclasses import dataclass, field -from typing import Any, Dict, List, Optional, Union +from typing import Any, Dict, List, Mapping, Optional, Union from fastapi import Depends, Header, HTTPException, Request from shared.api_key_store import APIKeyResponse @@ -36,7 +36,7 @@ from shared.tenant_context import TenantContext, resolve_tenant_context from .authz import IngestionPrincipal, get_ingestion_principal -from .webhook_models import EventResult, IngestEvent, IngestResponse, WebhookPayload +from .webhook_models import EventResult, IngestEvent, IngestResponse, WebhookCTEType, WebhookPayload from .webhook_router_v2 import ( _check_rate_limit, _generate_alerts, @@ -237,6 +237,7 @@ async def ingest_events( x_regengine_api_key: Optional[str] = None, principal: Optional[IngestionPrincipal] = None, source_file: Optional[SourceFileReceipt] = None, + deferred_kdes: Optional[Mapping[WebhookCTEType, frozenset[str]]] = None, ) -> IngestResponse: """Internal ingestion helper — not an HTTP route. @@ -256,6 +257,12 @@ async def ingest_events( row is processed, retains the file verbatim, stamps ``ingestion_run_id`` on every canonical event, and closes the run with counts covering every row of the file. The run id is returned on the response. + + ``deferred_kdes`` names, per CTE, required KDEs the source format cannot + carry. An event missing only those is stored, with an open data gap and + ``open_gaps`` on its result, instead of being rejected. Only server-side + callers set this. Never derive it from the payload, which the sender + controls: that would let a caller switch off the KDE gate. """ if not principal_tenant_id: raise PermissionError("ingest_events requires principal_tenant_id from authenticated context") @@ -329,7 +336,9 @@ async def ingest_events( seen_in_batch.add(dedup_key) errors = _validate_event_kdes(event) - if errors: + open_gaps = gaps_from_kde_errors(errors) + deferrable = deferred_kdes.get(event.cte_type, frozenset()) if errors and deferred_kdes else frozenset() + if errors and not (open_gaps and all(gap.gap_key in deferrable for gap in open_gaps)): results.append( EventResult( traceability_lot_code=event.traceability_lot_code, @@ -344,7 +353,7 @@ async def ingest_events( order=event_index, lot_code=event.traceability_lot_code, cte_type=event.cte_type.value, - gaps=gaps_from_kde_errors(errors), + gaps=open_gaps, ) ) continue @@ -495,16 +504,20 @@ async def ingest_events( event_id=store_result.event_id, sha256_hash=store_result.sha256_hash, chain_hash=store_result.chain_hash, + open_gaps=[gap.gap_key for gap in open_gaps], ) ) accepted += 1 + # A record with deferred KDEs is stored but incomplete: it + # opens its gaps and must not close the lot's existing ones. gap_outcomes.append( ValidationOutcome( order=event_index, lot_code=event.traceability_lot_code, cte_type=event.cte_type.value, - closing_record_id=store_result.event_id, - closing_record_hash=store_result.sha256_hash, + gaps=open_gaps, + closing_record_id=None if open_gaps else store_result.event_id, + closing_record_hash=None if open_gaps else store_result.sha256_hash, ) ) _publish_graph_sync(store_result.event_id, event, tenant_id) diff --git a/services/ingestion/app/webhook_models.py b/services/ingestion/app/webhook_models.py index 08017e288..9b17d4101 100644 --- a/services/ingestion/app/webhook_models.py +++ b/services/ingestion/app/webhook_models.py @@ -329,6 +329,10 @@ class EventResult(BaseModel): sha256_hash: Optional[str] = None chain_hash: Optional[str] = None errors: List[str] = Field(default_factory=list) + open_gaps: List[str] = Field( + default_factory=list, + description="Required KDEs this accepted event still lacks; each has an open data gap", + ) class IngestResponse(BaseModel): diff --git a/services/ingestion/tests/test_edi_856_required_kdes.py b/services/ingestion/tests/test_edi_856_required_kdes.py index d04a8f795..adb67a1bb 100644 --- a/services/ingestion/tests/test_edi_856_required_kdes.py +++ b/services/ingestion/tests/test_edi_856_required_kdes.py @@ -25,7 +25,7 @@ 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_models import IngestResponse, WebhookCTEType, 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" @@ -86,3 +86,19 @@ def test_856_without_tlc_source_reference_only_misses_that_kde(route: str, captu assert _validate_event_kdes(event) == [ "Missing required KDE 'tlc_source_reference' for shipping CTE" ] + + +@pytest.mark.parametrize("route", ["", "/document"]) +def test_both_routes_defer_only_shipping_tlc_source_reference(route: str, monkeypatch: pytest.MonkeyPatch, captured) -> None: + seen: list[dict] = [] + + async def _capture_kwargs(payload: WebhookPayload, **kwargs) -> IngestResponse: + seen.append(kwargs) + return IngestResponse(accepted=0, rejected=0, total=len(payload.events), events=[]) + + monkeypatch.setattr(edi_ingestion_routes, "ingest_events", _capture_kwargs) + + assert _post(route, {"traceability_lot_code": "LOT-856-KDE"}).status_code == 201 + assert seen + for kwargs in seen: + assert kwargs["deferred_kdes"] == {WebhookCTEType.SHIPPING: frozenset({"tlc_source_reference"})} diff --git a/services/ingestion/tests/test_edi_date_reject_resend.py b/services/ingestion/tests/test_edi_date_reject_resend.py index 2e3af20a5..32cb22ad6 100644 --- a/services/ingestion/tests/test_edi_date_reject_resend.py +++ b/services/ingestion/tests/test_edi_date_reject_resend.py @@ -66,7 +66,7 @@ def ingested(monkeypatch: pytest.MonkeyPatch) -> list: """Stub the canonical write; capture each accepted payload.""" calls: list = [] - async def _fake_ingest_events(payload, *, principal_tenant_id=None, x_regengine_api_key=None): + async def _fake_ingest_events(payload, *, principal_tenant_id=None, x_regengine_api_key=None, **_kwargs): calls.append(payload) return IngestResponse( accepted=len(payload.events), diff --git a/services/ingestion/tests/test_edi_deferred_kdes.py b/services/ingestion/tests/test_edi_deferred_kdes.py new file mode 100644 index 000000000..7aa502ada --- /dev/null +++ b/services/ingestion/tests/test_edi_deferred_kdes.py @@ -0,0 +1,187 @@ +"""EDI shipments without a TLC source reference are stored with an open gap. + +X12 has no element for ``tlc_source_reference``. The EDI routes defer that one +KDE on shipping events: the event is stored, its result names the gap, and a +``missing_kde`` gap stays open. Everything else keeps the strict gate, and the +exemption must be unreachable from anything the sender controls. +""" + +from __future__ import annotations + +import asyncio +from types import SimpleNamespace +from typing import Any + +import pytest + +from services.ingestion.app import webhook_compat as wc +from services.ingestion.app.edi_ingestion import routes as edi_routes +from services.ingestion.app.webhook_models import IngestEvent, WebhookCTEType, WebhookPayload +from shared.data_gap_tracking import GAP_KIND_MISSING_KDE, Gap + +TENANT = "11111111-1111-1111-1111-111111111111" +TLC = "00012345678901-LOT001" +EVENT_HASH = "a" * 64 +TLC_SOURCE_GAP = Gap(GAP_KIND_MISSING_KDE, "tlc_source_reference") + + +class _Session: + def begin_nested(self) -> "_Session": + return self + + def commit(self) -> None: + return None + + def rollback(self) -> None: + return None + + def close(self) -> None: + return None + + +@pytest.fixture() +def harness(monkeypatch: pytest.MonkeyPatch) -> SimpleNamespace: + state = SimpleNamespace(stored=[], outcomes=[]) + + class _Persistence: + def __init__(self, db: Any) -> None: + return None + + def store_event(self, **kwargs: Any) -> SimpleNamespace: + state.stored.append(kwargs) + return SimpleNamespace(event_id="ev-1", sha256_hash=EVENT_HASH, chain_hash="c" * 64) + + def _no_canonical(*_args: Any, **_kwargs: Any) -> None: + raise ValueError("canonical write not under test") + + def _sessions(): + yield _Session() + + import shared.canonical_event as canonical_event + import shared.cte_persistence as cte_persistence + + monkeypatch.setattr(cte_persistence, "CTEPersistence", _Persistence) + monkeypatch.setattr(canonical_event, "normalize_webhook_event", _no_canonical) + monkeypatch.setattr(wc, "_get_db_session", _sessions) + monkeypatch.setattr(wc, "_check_rate_limit", lambda _tenant: None) + monkeypatch.setattr(wc, "_verify_api_key_sync", lambda **_kwargs: None) + monkeypatch.setattr(wc, "_generate_alerts", lambda _event: []) + monkeypatch.setattr(wc, "_publish_graph_sync", lambda *_args: None) + monkeypatch.setattr(wc, "record_outcomes", lambda _db, _tenant, outcomes: state.outcomes.extend(outcomes)) + return state + + +def _shipment(**kde_overrides: Any) -> IngestEvent: + kdes = { + "ship_date": "2026-03-10", + "ship_from_location": "Valley Fresh Farms", + "ship_to_location": "Metro Distribution Center", + "reference_document": "BOL-9001", + **kde_overrides, + } + return IngestEvent( + cte_type=WebhookCTEType.SHIPPING, + traceability_lot_code=TLC, + product_description="Romaine 12CT", + quantity=200, + unit_of_measure="cases", + timestamp="2026-03-10T12:00:00Z", + location_name="Valley Fresh Farms", + kdes={key: value for key, value in kdes.items() if value is not None}, + ) + + +def _receipt() -> IngestEvent: + return IngestEvent( + cte_type=WebhookCTEType.RECEIVING, + traceability_lot_code=TLC, + product_description="Romaine 12CT", + quantity=200, + unit_of_measure="cases", + timestamp="2026-03-10T12:00:00Z", + location_name="Metro Distribution Center", + kdes={ + "receive_date": "2026-03-10", + "receiving_location": "Metro Distribution Center", + "immediate_previous_source": "Valley Fresh Farms", + "reference_document": "BOL-9001", + }, + ) + + +def _ingest(event: IngestEvent, *, source: str = "edi_856_inbound", **kwargs: Any): + payload = WebhookPayload(tenant_id=TENANT, source=source, events=[event]) + return asyncio.run( + wc.ingest_events( + payload, + principal_tenant_id=TENANT, + x_regengine_api_key="test-key", # pragma: allowlist secret — synthetic test fixture only + **kwargs, + ) + ) + + +def test_deferred_kde_stores_event_and_keeps_gap_open(harness) -> None: + response = _ingest(_shipment(), deferred_kdes=edi_routes._EDI_DEFERRED_KDES) + + assert (response.accepted, response.rejected) == (1, 0) + assert response.events[0].open_gaps == ["tlc_source_reference"] + assert len(harness.stored) == 1 + + outcome = harness.outcomes[0] + assert outcome.gaps == (TLC_SOURCE_GAP,) + # A receipt would close every open gap on this lot, including its own. + assert outcome.closing_record_id is None + assert outcome.closing_record_hash is None + + +def test_supplied_reference_stores_event_and_closes_gaps(harness) -> None: + response = _ingest( + _shipment(tlc_source_reference="PACK-0310"), + deferred_kdes=edi_routes._EDI_DEFERRED_KDES, + ) + + assert response.accepted == 1 + assert response.events[0].open_gaps == [] + outcome = harness.outcomes[0] + assert outcome.gaps == () + assert (outcome.closing_record_id, outcome.closing_record_hash) == ("ev-1", EVENT_HASH) + + +def test_another_missing_kde_still_rejects(harness) -> None: + response = _ingest(_shipment(reference_document=None), deferred_kdes=edi_routes._EDI_DEFERRED_KDES) + + assert (response.accepted, response.rejected) == (0, 1) + assert harness.stored == [] + assert set(harness.outcomes[0].gaps) == { + TLC_SOURCE_GAP, + Gap(GAP_KIND_MISSING_KDE, "reference_document"), + } + + +def test_receiving_is_not_deferred(harness) -> None: + response = _ingest(_receipt(), deferred_kdes=edi_routes._EDI_DEFERRED_KDES) + + assert (response.accepted, response.rejected) == (0, 1) + assert harness.stored == [] + + +def test_error_with_no_gap_key_still_rejects(harness, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(wc, "_validate_event_kdes", lambda _event: [" "]) + + response = _ingest(_shipment(), deferred_kdes=edi_routes._EDI_DEFERRED_KDES) + + assert (response.accepted, response.rejected) == (0, 1) + assert harness.stored == [] + + +def test_sender_cannot_claim_the_exemption(harness) -> None: + """EDI-looking source and KDEs change nothing without the server-side argument.""" + response = _ingest( + _shipment(edi_transaction_set="856", deferred_kdes=["tlc_source_reference"]), + source="edi_856_inbound", + ) + + assert (response.accepted, response.rejected) == (0, 1) + assert harness.stored == [] + assert response.events[0].open_gaps == [] diff --git a/services/ingestion/tests/test_edi_wire_format_2974_2976_2978.py b/services/ingestion/tests/test_edi_wire_format_2974_2976_2978.py index a69e42f86..f959a5847 100644 --- a/services/ingestion/tests/test_edi_wire_format_2974_2976_2978.py +++ b/services/ingestion/tests/test_edi_wire_format_2974_2976_2978.py @@ -140,7 +140,7 @@ def client(monkeypatch: pytest.MonkeyPatch, captured_payload: dict) -> TestClien auth_mode="test", ) - async def _fake_ingest_events(payload, *, principal_tenant_id=None, x_regengine_api_key=None): + async def _fake_ingest_events(payload, *, principal_tenant_id=None, x_regengine_api_key=None, **_kwargs): captured_payload["payload"] = payload return IngestResponse( accepted=len(payload.events), From 7dd293b470382c24d30e7cf750dd5b65034823f7 Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:31:56 -0800 Subject: [PATCH 3/6] fix(edi): commit deferred gaps with the event, refuse a shared TLC source Review findings on the deferred KDE path: - The gap was written after commit by the fail-open record_outcomes, so open_gaps could name a gap that was never recorded. Write it in the event's savepoint. If it can't be written, the event is not accepted. - The tlc_source_reference form field is a single value copied to every event, so a document with several lots could link a lot to a source that never assigned its code. Refuse that with a 422 before the ISA13 is reserved. Without the field those lots are stored with open gaps. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> --- .../ingestion/app/edi_ingestion/routes.py | 48 ++++++++++++++--- services/ingestion/app/webhook_compat.py | 20 ++++++- .../tests/test_edi_856_required_kdes.py | 53 +++++++++++++++++++ .../ingestion/tests/test_edi_deferred_kdes.py | 40 ++++++++++++-- services/shared/data_gap_tracking.py | 21 ++++++++ 5 files changed, 169 insertions(+), 13 deletions(-) diff --git a/services/ingestion/app/edi_ingestion/routes.py b/services/ingestion/app/edi_ingestion/routes.py index 050522dfe..3927f1433 100644 --- a/services/ingestion/app/edi_ingestion/routes.py +++ b/services/ingestion/app/edi_ingestion/routes.py @@ -26,7 +26,7 @@ import importlib import logging import os -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 @@ -92,6 +92,38 @@ 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, + extractions: Iterable[dict[str, Any]], + traceability_lot_code: str, +) -> None: + """Refuse one TLC source reference across several 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. + """ + if not (tlc_source_reference and tlc_source_reference.strip()): + return + lots: set[str] = set() + for extracted in extractions: + for line in extracted.get("line_items") or [{}]: + lot = line.get("lot_number") + lots.add(lot.strip() if isinstance(lot, str) and lot.strip() else traceability_lot_code) + if len(lots) > 1: + raise HTTPException( + status_code=422, + detail={ + "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." + ), + "lot_count": len(lots), + }, + ) + + def _decode_edi_bytes(raw_bytes: bytes) -> str: """Decode an EDI upload without silent data loss. @@ -625,19 +657,18 @@ 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, [extracted], traceability_lot_code) + + _enforce_envelope_integrity(extracted, sender_tenant_id, "856", before_reserve=_preflight) lines = extracted.get("line_items") or [] if len(lines) > 1: @@ -1200,6 +1231,7 @@ def _preflight_all_sets() -> None: partner_id=x_partner_id, source=source, ) + _reject_shared_tlc_source_for_distinct_lots(tlc_source_reference, extractions, traceability_lot_code) # Event-date rejections run after the allowlist and before the ISA13 # is reserved, so a corrected resend is not refused as a 409 replay. diff --git a/services/ingestion/app/webhook_compat.py b/services/ingestion/app/webhook_compat.py index a0343c492..4910f39fa 100644 --- a/services/ingestion/app/webhook_compat.py +++ b/services/ingestion/app/webhook_compat.py @@ -29,11 +29,13 @@ ValidationOutcome, gap_from_rule_reason, gaps_from_kde_errors, + open_gaps_for_stored_record, record_outcomes, ) from shared.database import get_db_session as _get_db_session from shared.permissions import has_permission from shared.tenant_context import TenantContext, resolve_tenant_context +from sqlalchemy.exc import SQLAlchemyError from .authz import IngestionPrincipal, get_ingestion_principal from .webhook_models import EventResult, IngestEvent, IngestResponse, WebhookCTEType, WebhookPayload @@ -494,6 +496,20 @@ async def ingest_events( ) continue + # An incomplete record and its gap commit together: if the + # gap can't be written, the event is not accepted. + if open_gaps: + try: + open_gaps_for_stored_record( + db_session, + tenant_id, + lot_code=event.traceability_lot_code, + cte_type=event.cte_type.value, + gaps=open_gaps, + ) + except SQLAlchemyError as gap_err: + raise RuntimeError("deferred gap write failed") from gap_err + # Accepted path. Savepoint commits the primary CTE write. savepoint.commit() results.append( @@ -509,7 +525,9 @@ async def ingest_events( ) accepted += 1 # A record with deferred KDEs is stored but incomplete: it - # opens its gaps and must not close the lot's existing ones. + # must not close the lot's existing gaps. Its gaps stay queued + # as well, so a complete record earlier in this batch can't + # close them when outcomes are applied in arrival order. gap_outcomes.append( ValidationOutcome( order=event_index, diff --git a/services/ingestion/tests/test_edi_856_required_kdes.py b/services/ingestion/tests/test_edi_856_required_kdes.py index adb67a1bb..3d33d30c4 100644 --- a/services/ingestion/tests/test_edi_856_required_kdes.py +++ b/services/ingestion/tests/test_edi_856_required_kdes.py @@ -102,3 +102,56 @@ async def _capture_kwargs(payload: WebhookPayload, **kwargs) -> IngestResponse: assert seen for kwargs in seen: assert kwargs["deferred_kdes"] == {WebhookCTEType.SHIPPING: frozenset({"tlc_source_reference"})} + + +def _two_lot_asn() -> bytes: + raw = _ASN_856.read_text(encoding="utf-8") + line = "LIN**SK*ROMAINE-12CT~SN1**200*CA~" + assert line in raw + two_lots = raw.replace( + line, + "LIN**SK*ROMAINE-12CT~SN1**100*CA~REF*LT*LOT-A~LIN**SK*ROMAINE-12CT~SN1**100*CA~REF*LT*LOT-B~", + ) + return two_lots.replace("SE*10*0001", "SE*14*0001").encode() + + +def _post_two_lots(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", _two_lot_asn(), "application/edi-x12")}, + ) + + +@pytest.mark.parametrize("route", ["", "/document"]) +def test_one_reference_cannot_cover_distinct_lots(route: str, monkeypatch: pytest.MonkeyPatch, captured) -> None: + claims: list[tuple] = [] + + def _claim(*args, **kwargs): + claims.append(args) + return False, None + + monkeypatch.setattr(edi_ingestion_routes, "check_and_record_interchange", _claim) + + response = _post_two_lots(route, {"traceability_lot_code": "LOT-856-KDE", "tlc_source_reference": "PACK-0310"}) + + assert response.status_code == 422, response.text + assert response.json()["detail"]["error"] == "edi_tlc_source_reference_with_multiple_lots" + assert captured == [] + assert claims == [], "a refused document must not reserve its ISA13" + + +@pytest.mark.parametrize("route", ["", "/document"]) +def test_distinct_lots_without_a_reference_are_ingested(route: str, captured) -> None: + response = _post_two_lots(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 {event.traceability_lot_code for event in events} == {"LOT-A", "LOT-B"} + assert all(event.kdes.get("tlc_source_reference") is None for event in events) diff --git a/services/ingestion/tests/test_edi_deferred_kdes.py b/services/ingestion/tests/test_edi_deferred_kdes.py index 7aa502ada..344d3e43e 100644 --- a/services/ingestion/tests/test_edi_deferred_kdes.py +++ b/services/ingestion/tests/test_edi_deferred_kdes.py @@ -13,6 +13,7 @@ from typing import Any import pytest +from sqlalchemy.exc import ProgrammingError from services.ingestion.app import webhook_compat as wc from services.ingestion.app.edi_ingestion import routes as edi_routes @@ -25,9 +26,29 @@ TLC_SOURCE_GAP = Gap(GAP_KIND_MISSING_KDE, "tlc_source_reference") +class _Savepoint: + def __init__(self, state: SimpleNamespace) -> None: + self.state = state + self.stored_before = len(state.stored) + + def commit(self) -> None: + return None + + def rollback(self) -> None: + del self.state.stored[self.stored_before :] + + class _Session: - def begin_nested(self) -> "_Session": - return self + def __init__(self, state: SimpleNamespace) -> None: + self.state = state + + def begin_nested(self) -> "_Savepoint": + return _Savepoint(self.state) + + def execute(self, _statement: Any, params: dict[str, Any]) -> None: + if self.state.gap_write_error is not None: + raise self.state.gap_write_error + self.state.inline_gaps.append(params["key"]) def commit(self) -> None: return None @@ -41,7 +62,7 @@ def close(self) -> None: @pytest.fixture() def harness(monkeypatch: pytest.MonkeyPatch) -> SimpleNamespace: - state = SimpleNamespace(stored=[], outcomes=[]) + state = SimpleNamespace(stored=[], outcomes=[], inline_gaps=[], gap_write_error=None) class _Persistence: def __init__(self, db: Any) -> None: @@ -55,7 +76,7 @@ def _no_canonical(*_args: Any, **_kwargs: Any) -> None: raise ValueError("canonical write not under test") def _sessions(): - yield _Session() + yield _Session(state) import shared.canonical_event as canonical_event import shared.cte_persistence as cte_persistence @@ -127,6 +148,7 @@ def test_deferred_kde_stores_event_and_keeps_gap_open(harness) -> None: assert (response.accepted, response.rejected) == (1, 0) assert response.events[0].open_gaps == ["tlc_source_reference"] assert len(harness.stored) == 1 + assert harness.inline_gaps == ["tlc_source_reference"] outcome = harness.outcomes[0] assert outcome.gaps == (TLC_SOURCE_GAP,) @@ -135,6 +157,16 @@ def test_deferred_kde_stores_event_and_keeps_gap_open(harness) -> None: assert outcome.closing_record_hash is None +def test_event_is_not_accepted_when_its_gap_cannot_be_written(harness) -> None: + harness.gap_write_error = ProgrammingError("INSERT", {}, Exception("permission denied")) + + response = _ingest(_shipment(), deferred_kdes=edi_routes._EDI_DEFERRED_KDES) + + assert (response.accepted, response.rejected) == (0, 1) + assert response.events[0].open_gaps == [] + assert harness.stored == [], "the stored event must roll back with its gap" + + def test_supplied_reference_stores_event_and_closes_gaps(harness) -> None: response = _ingest( _shipment(tlc_source_reference="PACK-0310"), diff --git a/services/shared/data_gap_tracking.py b/services/shared/data_gap_tracking.py index f9a66f986..b78b81bf0 100644 --- a/services/shared/data_gap_tracking.py +++ b/services/shared/data_gap_tracking.py @@ -252,6 +252,27 @@ def _apply_outcome(db_session: Any, tenant_id: str, outcome: ValidationOutcome) ) +def open_gaps_for_stored_record( + db_session: Any, + tenant_id: str, + *, + lot_code: str, + cte_type: str, + gaps: tuple[Gap, ...], +) -> None: + """Open gaps inside the record's own transaction. Raises on failure. + + Gap logging is normally fail-open, because a failed record is not stored + anyway. A record stored with a known gap is different: if the gap is lost, + nothing shows the record is incomplete. Commit the two together. + """ + _apply_outcome( + db_session, + tenant_id, + ValidationOutcome(order=0, lot_code=lot_code, cte_type=cte_type, gaps=gaps), + ) + + def record_outcomes(db_session: Any, tenant_id: str, outcomes: Sequence[ValidationOutcome]) -> None: """Apply queued outcomes in arrival order, then commit. Never raises. From 0053d5f6ebc1332687707c8e72fae59f8b8961de Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:43:36 -0800 Subject: [PATCH 4/6] fix(edi): keep multi-lot receipts ingestible, record the lot refusal Review findings on the previous commit: - The refusal covered receiving, which requires the field and has no open gap to fall back on, so a multi-lot 944 could not be ingested at all. Limit it to shipping sets. - The refusal left no audit trail. Record it like other EDI rejections. - Replaying every deferred gap after commit could reopen a gap that another request had just closed. Requeue only to follow a close queued earlier in the same batch. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> --- .../ingestion/app/edi_ingestion/routes.py | 76 ++++++++++++++----- services/ingestion/app/webhook_compat.py | 38 ++++++---- .../tests/test_edi_856_required_kdes.py | 18 +++++ .../ingestion/tests/test_edi_deferred_kdes.py | 36 +++++++-- 4 files changed, 131 insertions(+), 37 deletions(-) diff --git a/services/ingestion/app/edi_ingestion/routes.py b/services/ingestion/app/edi_ingestion/routes.py index 3927f1433..90a491da7 100644 --- a/services/ingestion/app/edi_ingestion/routes.py +++ b/services/ingestion/app/edi_ingestion/routes.py @@ -93,35 +93,61 @@ def _record_edi_rejection(**kwargs): def _reject_shared_tlc_source_for_distinct_lots( + *, tlc_source_reference: str | None, - extractions: Iterable[dict[str, Any]], + 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 lots. + """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. + 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 extracted in extractions: + for _, extracted in shipping_sets: for line in extracted.get("line_items") or [{}]: lot = line.get("lot_number") lots.add(lot.strip() if isinstance(lot, str) and lot.strip() else traceability_lot_code) - if len(lots) > 1: - raise HTTPException( - status_code=422, - detail={ - "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." - ), - "lot_count": len(lots), - }, - ) + 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"], + }, + ) def _decode_edi_bytes(raw_bytes: bytes) -> str: @@ -666,7 +692,14 @@ def _preflight() -> None: partner_id=x_partner_id, source=source, ) - _reject_shared_tlc_source_for_distinct_lots(tlc_source_reference, [extracted], traceability_lot_code) + _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) @@ -1231,7 +1264,14 @@ def _preflight_all_sets() -> None: partner_id=x_partner_id, source=source, ) - _reject_shared_tlc_source_for_distinct_lots(tlc_source_reference, extractions, traceability_lot_code) + _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. diff --git a/services/ingestion/app/webhook_compat.py b/services/ingestion/app/webhook_compat.py index 4910f39fa..7cfa15f20 100644 --- a/services/ingestion/app/webhook_compat.py +++ b/services/ingestion/app/webhook_compat.py @@ -316,6 +316,7 @@ async def ingest_events( seen_in_batch: set[str] = set() # Data gap time to fix: written after the ingest commit, fail-open. gap_outcomes: list[ValidationOutcome] = [] + closes_queued: set[tuple[str, str]] = set() try: if source_file is not None: @@ -524,20 +525,31 @@ async def ingest_events( ) ) accepted += 1 - # A record with deferred KDEs is stored but incomplete: it - # must not close the lot's existing gaps. Its gaps stay queued - # as well, so a complete record earlier in this batch can't - # close them when outcomes are applied in arrival order. - gap_outcomes.append( - ValidationOutcome( - order=event_index, - lot_code=event.traceability_lot_code, - cte_type=event.cte_type.value, - gaps=open_gaps, - closing_record_id=None if open_gaps else store_result.event_id, - closing_record_hash=None if open_gaps else store_result.sha256_hash, + # A record with deferred KDEs is stored but incomplete, so it + # must not close the lot's existing gaps. Its gaps are already + # written; requeue them only to follow a close queued earlier + # in this batch, as a replay could reopen another request's. + gap_scope = (event.traceability_lot_code.strip(), event.cte_type.value) + if not open_gaps: + closes_queued.add(gap_scope) + gap_outcomes.append( + ValidationOutcome( + order=event_index, + lot_code=event.traceability_lot_code, + cte_type=event.cte_type.value, + closing_record_id=store_result.event_id, + closing_record_hash=store_result.sha256_hash, + ) + ) + elif gap_scope in closes_queued: + gap_outcomes.append( + ValidationOutcome( + order=event_index, + lot_code=event.traceability_lot_code, + cte_type=event.cte_type.value, + gaps=open_gaps, + ) ) - ) _publish_graph_sync(store_result.event_id, event, tenant_id) # Non-blocking failures → exception queue. Separate diff --git a/services/ingestion/tests/test_edi_856_required_kdes.py b/services/ingestion/tests/test_edi_856_required_kdes.py index 3d33d30c4..6ac091fe9 100644 --- a/services/ingestion/tests/test_edi_856_required_kdes.py +++ b/services/ingestion/tests/test_edi_856_required_kdes.py @@ -22,6 +22,7 @@ pytest.importorskip("fastapi") +import app.edi_ingestion.rejection_log as rejection_log # noqa: E402 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 @@ -131,6 +132,7 @@ def _post_two_lots(route: str, data: dict[str, str]): @pytest.mark.parametrize("route", ["", "/document"]) def test_one_reference_cannot_cover_distinct_lots(route: str, monkeypatch: pytest.MonkeyPatch, captured) -> None: + rejection_log.reset_edi_rejections() claims: list[tuple] = [] def _claim(*args, **kwargs): @@ -145,6 +147,9 @@ def _claim(*args, **kwargs): assert response.json()["detail"]["error"] == "edi_tlc_source_reference_with_multiple_lots" assert captured == [] assert claims == [], "a refused document must not reserve its ISA13" + recorded = rejection_log.list_edi_rejections(_TENANT) + assert [record["reason"] for record in recorded] == ["edi_tlc_source_reference_with_multiple_lots"] + assert response.json()["detail"]["rejection_id"] == recorded[0]["rejection_id"] @pytest.mark.parametrize("route", ["", "/document"]) @@ -155,3 +160,16 @@ def test_distinct_lots_without_a_reference_are_ingested(route: str, captured) -> events = [event for payload in captured for event in payload.events] assert {event.traceability_lot_code for event in events} == {"LOT-A", "LOT-B"} assert all(event.kdes.get("tlc_source_reference") is None for event in events) + + +@pytest.mark.parametrize("transaction_set", ["861", "944"]) +def test_multi_lot_receipt_may_share_a_reference(transaction_set: str) -> None: + """Receiving requires the KDE, so refusing it would leave no way in.""" + edi_ingestion_routes._reject_shared_tlc_source_for_distinct_lots( + tlc_source_reference="PACK-0310", + transaction_sets=[(transaction_set, {"line_items": [{"lot_number": "LOT-A"}, {"lot_number": "LOT-B"}]})], + traceability_lot_code="LOT-FORM", + sender_tenant_id=_TENANT, + partner_id=None, + source="edi_inbound", + ) diff --git a/services/ingestion/tests/test_edi_deferred_kdes.py b/services/ingestion/tests/test_edi_deferred_kdes.py index 344d3e43e..942582cbd 100644 --- a/services/ingestion/tests/test_edi_deferred_kdes.py +++ b/services/ingestion/tests/test_edi_deferred_kdes.py @@ -131,7 +131,11 @@ def _receipt() -> IngestEvent: def _ingest(event: IngestEvent, *, source: str = "edi_856_inbound", **kwargs: Any): - payload = WebhookPayload(tenant_id=TENANT, source=source, events=[event]) + return _ingest_batch([event], source=source, **kwargs) + + +def _ingest_batch(events: list[IngestEvent], *, source: str = "edi_856_inbound", **kwargs: Any): + payload = WebhookPayload(tenant_id=TENANT, source=source, events=events) return asyncio.run( wc.ingest_events( payload, @@ -150,11 +154,31 @@ def test_deferred_kde_stores_event_and_keeps_gap_open(harness) -> None: assert len(harness.stored) == 1 assert harness.inline_gaps == ["tlc_source_reference"] - outcome = harness.outcomes[0] - assert outcome.gaps == (TLC_SOURCE_GAP,) - # A receipt would close every open gap on this lot, including its own. - assert outcome.closing_record_id is None - assert outcome.closing_record_hash is None + # Nothing queued: no receipt to close the lot's gaps, and no replay that + # could reopen a gap another request closes after this commit. + assert harness.outcomes == [] + + +def test_batch_keeps_arrival_order_for_a_lot(harness) -> None: + """A later incomplete record keeps its gap; a later complete one closes it.""" + complete = _shipment(tlc_source_reference="PACK-0310") + incomplete = _shipment().model_copy(update={"timestamp": "2026-03-10T13:00:00Z"}) + + _ingest_batch([complete, incomplete], deferred_kdes=edi_routes._EDI_DEFERRED_KDES) + + assert [outcome.gaps for outcome in harness.outcomes] == [(), (TLC_SOURCE_GAP,)] + assert harness.outcomes[0].closing_record_id == "ev-1" + assert harness.outcomes[1].closing_record_id is None + + +def test_incomplete_then_complete_in_a_batch_closes_the_gap(harness) -> None: + incomplete = _shipment() + complete = _shipment(tlc_source_reference="PACK-0310").model_copy(update={"timestamp": "2026-03-10T13:00:00Z"}) + + _ingest_batch([incomplete, complete], deferred_kdes=edi_routes._EDI_DEFERRED_KDES) + + assert harness.inline_gaps == ["tlc_source_reference"] + assert [outcome.closing_record_id for outcome in harness.outcomes] == ["ev-1"] def test_event_is_not_accepted_when_its_gap_cannot_be_written(harness) -> None: From 785dbb91938da8c30391f1414116a7964373b51f Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Wed, 4 Nov 2026 22:29:06 -0800 Subject: [PATCH 5/6] fix(edi): compare stored TLCs when refusing a shared TLC source A 945 lot that isn't production-shaped is stored with the line's GTIN or UPC as a prefix, so one short lot on two items is two TLCs. The guard compared bare lots, counted one, and let one reference cover both. Move the rebuild into a helper shared by the validator and the guard so the two can't disagree. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> --- .../ingestion/app/edi_ingestion/routes.py | 12 +++-- .../ingestion/app/edi_ingestion/validation.py | 44 ++++++++++++------- .../tests/test_edi_856_required_kdes.py | 30 ++++++++++++- 3 files changed, 66 insertions(+), 20 deletions(-) diff --git a/services/ingestion/app/edi_ingestion/routes.py b/services/ingestion/app/edi_ingestion/routes.py index 31852f159..c18cf4173 100644 --- a/services/ingestion/app/edi_ingestion/routes.py +++ b/services/ingestion/app/edi_ingestion/routes.py @@ -73,7 +73,7 @@ _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") @@ -124,10 +124,16 @@ def _reject_shared_tlc_source_for_distinct_lots( if _CTE_TYPE_BY_SET.get(transaction_set, WebhookCTEType.SHIPPING) == WebhookCTEType.SHIPPING ] lots: set[str] = set() - for _, extracted in shipping_sets: + for transaction_set, extracted in shipping_sets: for line in extracted.get("line_items") or [{}]: lot = line.get("lot_number") - lots.add(lot.strip() if isinstance(lot, str) and lot.strip() else traceability_lot_code) + 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" diff --git a/services/ingestion/app/edi_ingestion/validation.py b/services/ingestion/app/edi_ingestion/validation.py index c2a848235..198b6ef9c 100644 --- a/services/ingestion/app/edi_ingestion/validation.py +++ b/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. diff --git a/services/ingestion/tests/test_edi_856_required_kdes.py b/services/ingestion/tests/test_edi_856_required_kdes.py index 6ac091fe9..4a5222667 100644 --- a/services/ingestion/tests/test_edi_856_required_kdes.py +++ b/services/ingestion/tests/test_edi_856_required_kdes.py @@ -14,7 +14,7 @@ from pathlib import Path import pytest -from fastapi import FastAPI +from fastapi import FastAPI, HTTPException from fastapi.testclient import TestClient service_dir = Path(__file__).parent.parent @@ -173,3 +173,31 @@ def test_multi_lot_receipt_may_share_a_reference(transaction_set: str) -> None: partner_id=None, source="edi_inbound", ) + + +def _check_945_lines(*lines: dict) -> None: + edi_ingestion_routes._reject_shared_tlc_source_for_distinct_lots( + tlc_source_reference="PACK-0310", + transaction_sets=[("945", {"line_items": list(lines)})], + traceability_lot_code="LOT-FORM", + sender_tenant_id=_TENANT, + partner_id=None, + source="edi_inbound", + ) + + +def test_one_short_lot_on_two_gtins_is_two_tlcs() -> None: + with pytest.raises(HTTPException) as refused: + _check_945_lines( + {"lot_number": "A1", "gtin": "00012345678901"}, + {"lot_number": "A1", "gtin": "00012345678902"}, + ) + + assert refused.value.detail["error"] == "edi_tlc_source_reference_with_multiple_lots" + + +def test_one_short_lot_on_one_gtin_is_one_tlc() -> None: + _check_945_lines( + {"lot_number": "A1", "gtin": "00012345678901"}, + {"lot_number": "A1", "gtin": "00012345678901"}, + ) From 3c0507eae52b163453e70a74eb64e75395181b2a Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Wed, 4 Nov 2026 23:32:34 -0800 Subject: [PATCH 6/6] fix(edi): don't reject deferred gaps under RULES_ENGINE_ENFORCE=all The deferral only skipped the KDE gate. The seeded shipping rule for tlc_source_reference then failed as a warning, and all rejects on warnings, so these shipments could never be stored in that mode. should_reject takes deferred_kdes. A non-critical failure whose missing fields all belong to a deferred KDE no longer rejects. Every other failure still does, and it fails closed if those results don't explain the failure count. The verdict stays fail, so the exception queue still records it. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> --- docs/architecture/RULES_ENFORCEMENT.md | 2 + services/ingestion/app/webhook_compat.py | 8 +- services/shared/rules/enforcement.py | 49 ++++++++-- .../test_rules_enforcement_deferred_kdes.py | 96 +++++++++++++++++++ 4 files changed, 146 insertions(+), 9 deletions(-) create mode 100644 services/shared/tests/test_rules_enforcement_deferred_kdes.py diff --git a/docs/architecture/RULES_ENFORCEMENT.md b/docs/architecture/RULES_ENFORCEMENT.md index cd88f2ac5..7a03b97e3 100644 --- a/docs/architecture/RULES_ENFORCEMENT.md +++ b/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." diff --git a/services/ingestion/app/webhook_compat.py b/services/ingestion/app/webhook_compat.py index 7cfa15f20..131105545 100644 --- a/services/ingestion/app/webhook_compat.py +++ b/services/ingestion/app/webhook_compat.py @@ -441,7 +441,13 @@ async def ingest_events( if summary is not None: from shared.rules.enforcement import current_mode, should_reject - should, reject_reason = should_reject(summary) + if open_gaps: + should, reject_reason = should_reject( + summary, + deferred_kdes=frozenset(gap.gap_key for gap in open_gaps), + ) + else: + should, reject_reason = should_reject(summary) if should: # Finding B — durable rejection audit row, # written via its own session BEFORE the diff --git a/services/shared/rules/enforcement.py b/services/shared/rules/enforcement.py index 8212857f5..33da05b8f 100644 --- a/services/shared/rules/enforcement.py +++ b/services/shared/rules/enforcement.py @@ -27,9 +27,13 @@ from __future__ import annotations import os -from typing import Optional, Tuple +from typing import FrozenSet, Optional, Sequence, Tuple -from .types import EvaluationSummary +from .types import EvaluationSummary, RuleEvaluationResult + +# Other fields a rule accepts for the same KDE. A rule that fails on a deferred +# KDE reports these as missing too, so they don't make the failure a new one. +_KDE_ALTERNATE_FIELDS = {"tlc_source_reference": ("kdes.tlc_source_gln",)} class EnforcementMode: @@ -155,7 +159,22 @@ def current_mode() -> str: return raw -def should_reject(summary: EvaluationSummary) -> Tuple[bool, Optional[str]]: +def _is_deferred_failure(result: RuleEvaluationResult, deferred_kdes: FrozenSet[str]) -> bool: + """True when a non-critical failure comes only from deferred KDEs.""" + if result.result != "fail" or result.severity == "critical": + return False + allowed = {f"kdes.{kde}" for kde in deferred_kdes} + for kde in deferred_kdes: + allowed.update(_KDE_ALTERNATE_FIELDS.get(kde, ())) + missing = {entry.get("field") for entry in result.evidence_fields_inspected if not entry.get("present")} + return bool(missing) and missing <= allowed + + +def should_reject( + summary: EvaluationSummary, + *, + deferred_kdes: FrozenSet[str] = frozenset(), +) -> Tuple[bool, Optional[str]]: """Return ``(reject, reason)`` for a rules evaluation. ``reject``: True means the caller should reject the event (typically @@ -163,6 +182,12 @@ def should_reject(summary: EvaluationSummary) -> Tuple[bool, Optional[str]]: ``reason``: a short human-readable string suitable for the HTTP response body when rejecting. ``None`` when not rejecting. + + ``deferred_kdes`` names KDEs the caller stores with an open data gap. + Under ``all``, a warning that fails only because of them does not + reject, or the gap could never be stored. Only server-side callers may + set this. The verdict stays ``fail``, so the exception queue still sees + it, and every other failure on the event still rejects. """ mode = current_mode() if mode == EnforcementMode.OFF: @@ -181,9 +206,13 @@ def should_reject(summary: EvaluationSummary) -> Tuple[bool, Optional[str]]: return True, _first_failure_reason(summary) return False, None if mode == EnforcementMode.ALL: - if summary.compliant is False: - return True, _first_failure_reason(summary) - return False, None + if summary.compliant is not False: + return False, None + deferred = [result for result in summary.results if _is_deferred_failure(result, deferred_kdes)] + # Fail closed: accept only if these results explain every failure. + if deferred and summary.errored == 0 and len(deferred) == summary.failed: + return False, None + return True, _first_failure_reason(summary, skip=deferred) return False, None @@ -262,9 +291,13 @@ def handle_preeval_exception(err: BaseException, *, where: str) -> None: ) from err -def _first_failure_reason(summary: EvaluationSummary) -> str: +def _first_failure_reason( + summary: EvaluationSummary, + *, + skip: Sequence[RuleEvaluationResult] = (), +) -> str: for result in summary.critical_failures or summary.results: - if result.result in ("fail", "error"): + if result.result in ("fail", "error") and not any(result is skipped for skipped in skip): rule_id = result.rule_id or "unknown_rule" why = result.why_failed or "failed" return f"{rule_id}: {why}" diff --git a/services/shared/tests/test_rules_enforcement_deferred_kdes.py b/services/shared/tests/test_rules_enforcement_deferred_kdes.py new file mode 100644 index 000000000..ecbf22ba7 --- /dev/null +++ b/services/shared/tests/test_rules_enforcement_deferred_kdes.py @@ -0,0 +1,96 @@ +"""Under ``all``, a warning caused only by a deferred KDE does not reject. + +EDI shipments are stored with an open data gap when X12 can't carry the TLC +source reference. The seeded rule for that KDE is a warning, and ``all`` +rejects on warnings, so without this the gap could never be stored. +""" + +from __future__ import annotations + +import pytest + +from shared.rules.enforcement import should_reject +from shared.rules.types import EvaluationSummary, RuleEvaluationResult + +DEFERRED = frozenset({"tlc_source_reference"}) + + +def _field(name: str, present: bool = False) -> dict: + return {"field": name, "value": None, "present": present} + + +def _tlc_source_warning(severity: str = "warning") -> RuleEvaluationResult: + return RuleEvaluationResult( + rule_id="tlc-source", + severity=severity, + result="fail", + why_failed="missing TLC source", + evidence_fields_inspected=[_field("kdes.tlc_source_reference"), _field("kdes.tlc_source_gln")], + ) + + +def _other_warning() -> RuleEvaluationResult: + return RuleEvaluationResult( + rule_id="other", + severity="warning", + result="fail", + why_failed="other failure", + evidence_fields_inspected=[_field("kdes.carrier")], + ) + + +def _summary(*results: RuleEvaluationResult, errored: int = 0) -> EvaluationSummary: + failed = sum(1 for result in results if result.result == "fail") + return EvaluationSummary( + total_rules=len(results) + 1, + passed=1, + failed=failed, + errored=errored, + results=list(results), + critical_failures=[result for result in results if result.severity == "critical"], + ) + + +@pytest.fixture(autouse=True) +def _all_mode(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("RULES_ENGINE_ENFORCE", "all") + + +def test_deferred_failure_does_not_reject_and_stays_noncompliant() -> None: + summary = _summary(_tlc_source_warning()) + + assert should_reject(summary, deferred_kdes=DEFERRED) == (False, None) + assert summary.compliant is False + + +def test_without_deferred_kdes_the_warning_still_rejects() -> None: + assert should_reject(_summary(_tlc_source_warning()))[0] is True + + +def test_another_failure_still_rejects_and_is_the_reason() -> None: + reject, reason = should_reject(_summary(_tlc_source_warning(), _other_warning()), deferred_kdes=DEFERRED) + + assert reject is True + assert reason == "other: other failure" + + +def test_critical_rule_on_the_same_kde_still_rejects() -> None: + assert should_reject(_summary(_tlc_source_warning("critical")), deferred_kdes=DEFERRED)[0] is True + + +def test_rule_needing_another_missing_field_still_rejects() -> None: + result = _tlc_source_warning() + result.evidence_fields_inspected.append(_field("kdes.carrier")) + + assert should_reject(_summary(result), deferred_kdes=DEFERRED)[0] is True + + +def test_evaluator_error_still_rejects() -> None: + assert should_reject(_summary(_tlc_source_warning(), errored=1), deferred_kdes=DEFERRED)[0] is True + + +def test_unexplained_failure_count_fails_closed() -> None: + summary = _summary(_tlc_source_warning()) + summary.failed = 2 + + assert should_reject(summary, deferred_kdes=DEFERRED)[0] is True