fix(ingestion): keep real EDI event time instead of ingest time

services/ingestion/app/edi_ingestion/rejection_log.py

@@ -55,6 +55,8 @@ def record_edi_rejection(
     extracted: dict[str, Any] | None = None,
     partner_id: str | None = None,
     source: str | None = None,
+    reason: str = "fsma_validation_failed",
+    needs_review: bool = False,
 ) -> dict[str, Any]:
     """Record an EDI FSMA validation rejection.
 
@@ -78,7 +80,8 @@ def record_edi_rejection(
         # Snapshot of the extracted fields for forensic review. Drops
         # any non-serializable values defensively.
         "extracted": _scrub(extracted or {}),
-        "reason": "fsma_validation_failed",
+        "reason": reason,
+        "needs_review": needs_review,
     }
 
     store = _rejection_store_for(tenant_id)

services/ingestion/app/edi_ingestion/routes.py

@@ -26,6 +26,7 @@
 import importlib
 import logging
 import os
+from collections.abc import Callable
 from typing import Any, Optional
 
 from fastapi import APIRouter, Depends, File, Form, Header, HTTPException, Query, UploadFile
@@ -280,6 +281,69 @@ def _reject_legacy_856_placeholders(
     )
 
 
+# Event-date field (and optional time field) per transaction set.
+_EVENT_DATE_FIELDS: dict[str, tuple[str | None, str | None]] = {
+    "856": ("ship_date_raw", "ship_time_raw"),
+    "850": ("po_date_raw", None),
+    "810": ("invoice_date_raw", None),
+    "861": ("receive_date_raw", None),
+    "944": ("receive_date_raw", None),
+    "945": ("ship_date_raw", None),
+}
+
+
+def _preflight_event_date(
+    *,
+    transaction_set: str,
+    extracted: dict[str, Any],
+    sender_tenant_id: str,
+    traceability_lot_code: str,
+    partner_id: str | None,
+    source: str,
+) -> None:
+    """Reject a bad EDI event date BEFORE the ISA13 is reserved.
+
+    ``_enforce_envelope_integrity`` records the interchange control number,
+    so a date rejection raised after it would make the corrected resend a
+    409 replay. Run this first. Every rejection is written to the EDI
+    rejection log with ``needs_review=True`` and its id is returned to the
+    caller.
+    """
+    date_field, time_field = _EVENT_DATE_FIELDS.get(transaction_set, (None, None))
+    if transaction_set in ("944", "945"):
+        time_field = None
+    try:
+        _ship_timestamp_and_date(
+            extracted.get(date_field) if date_field else None,
+            extracted.get(time_field) if time_field else None,
+        )
+    except HTTPException as exc:
+        if exc.status_code != 400 or not isinstance(exc.detail, dict):
+            raise
+        detail = dict(exc.detail)
+        record = _record_edi_rejection(
+            tenant_id=sender_tenant_id,
+            transaction_set=transaction_set,
+            traceability_lot_code=traceability_lot_code,
+            errors=[
+                {
+                    "loc": [date_field or "event_date"],
+                    "type": detail.get("error"),
+                    "msg": detail.get("message"),
+                    "input": detail.get("raw"),
+                }
+            ],
+            extracted=extracted,
+            partner_id=partner_id,
+            source=source,
+            reason=str(detail.get("error") or "invalid_edi_date"),
+            needs_review=True,
+        )
+        detail["needs_review"] = True
+        detail["rejection_id"] = record["rejection_id"]
+        raise HTTPException(status_code=400, detail=detail) from exc
+
+
 router = APIRouter(prefix="/api/v1/ingest/edi", tags=["EDI Import"])
 
 _resolve_tenant_id = resolve_tenant_id
@@ -290,9 +354,15 @@ def _enforce_envelope_integrity(
     extracted: dict[str, Any],
     sender_tenant_id: str,
     transaction_set: str,
+    before_reserve: Callable[[], None] | None = None,
 ) -> None:
     """Apply #1160 + #1165 envelope validation.
 
+    ``before_reserve`` runs after the envelope and allowlist checks and
+    before the ISA13 is recorded. Content rejections (event dates) go
+    there so a disallowed sender still gets 403 and a rejected document
+    does not burn its interchange number.
+
     - Reject on ISA/GS sender/receiver mismatch (X12 tenant-smuggling).
     - Reject on GS sender not in tenant's allowed trading-partner list.
     - Check ISA13 against the interchange dedup cache and return 409
@@ -342,6 +412,9 @@ def _enforce_envelope_integrity(
             },
         )
 
+    if before_reserve is not None:
+        before_reserve()
+
     # #1165: retransmission dedup on ISA13.
     is_duplicate, _prev = check_and_record_interchange(
         sender_id=extracted.get("sender_id"),
@@ -537,8 +610,21 @@ async def ingest_edi_856(
 
     extracted = _extract_856_fields(segments)
 
-    # #1160 + #1165: envelope mismatch / allowlist / ISA13 dedup.
-    _enforce_envelope_integrity(extracted, sender_tenant_id, "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(
+            transaction_set="856",
+            extracted=extracted,
+            sender_tenant_id=sender_tenant_id,
+            traceability_lot_code=traceability_lot_code,
+            partner_id=x_partner_id,
+            source=source,
+        ),
+    )
 
     lines = extracted.get("line_items") or []
     if len(lines) > 1:
@@ -774,15 +860,7 @@ def _build_ingest_event_for_set(
         location_gln = _valid_gln_or_none(extracted.get("receiving_location_gln"))
 
     # Resolve timestamp
-    date_field_map = {
-        "856": ("ship_date_raw", "ship_time_raw"),
-        "850": ("po_date_raw", None),
-        "810": ("invoice_date_raw", None),
-        "861": ("receive_date_raw", None),
-        "944": ("receive_date_raw", None),
-        "945": ("ship_date_raw", None),
-    }
-    date_field, time_field = date_field_map.get(transaction_set, (None, None))
+    date_field, time_field = _EVENT_DATE_FIELDS.get(transaction_set, (None, None))
     if transaction_set in ("944", "945") and extracted.get("_event_timestamp"):
         timestamp, event_date = extracted["_event_timestamp"], extracted["_event_date"]
     else:
@@ -961,7 +1039,9 @@ 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"),
+    tlc_source_reference: Optional[str] = Form(
+        None, description="Optional TLC source reference copied to 944/945 KDEs"
+    ),
     tenant_id: Optional[str] = Form(None, description="Optional tenant override"),
     source: str = Form("edi_inbound", description="Source system identifier"),
     product_description: Optional[str] = Form(None, description="Optional product description override"),
@@ -1084,7 +1164,20 @@ async def ingest_edi_document(
 
     # #1160 + #1165: validate and reserve the shared envelope once per
     # document, before validating any transaction set or warehouse line.
-    _enforce_envelope_integrity(extracted, sender_tenant_id, transaction_set)
+    def _preflight_all_sets() -> None:
+        for slice_set, slice_extracted in zip(slice_sets, extractions):
+            _preflight_event_date(
+                transaction_set=slice_set,
+                extracted=slice_extracted,
+                sender_tenant_id=sender_tenant_id,
+                traceability_lot_code=traceability_lot_code,
+                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.
+    _enforce_envelope_integrity(extracted, sender_tenant_id, transaction_set, before_reserve=_preflight_all_sets)
 
     # 856 SN1, 861 RCD, and warehouse W07/W12 lines are each one CTE.
     # ST/SE extraction and envelope handling stay at document scope, then

services/ingestion/app/edi_ingestion/utils.py

@@ -125,17 +125,30 @@ def _valid_gln_or_none(value: Optional[str]) -> Optional[str]:
 def _ship_timestamp_and_date(ship_date_raw: Optional[str], ship_time_raw: Optional[str]) -> tuple[str, str]:
     """Normalize an EDI shipment date/time into an ISO timestamp and date.
 
-    #1167: unparseable dates now raise HTTP 400 rather than silently
-    falling back to ``datetime.now()`` — retransmitted archival EDI was
-    being re-stamped as today's shipment. Stale dates (> 90 days old)
-    still snap to "now" for the ingest timestamp to keep webhook model
-    bounds valid, but the original parsed date is preserved separately
-    so downstream FSMA 204 lookback sees the real event date.
+    The returned timestamp is always the real event time from the EDI
+    source, never the ingest time. Backfilled EDI older than 90 days keeps
+    its real date (the webhook model accepts historical events for the
+    FSMA 204 2-year retention backfill). Previously the timestamp snapped
+    to ``now`` for such events, which gave each event two clocks and
+    corrupted EPCIS eventTime and R1 time to fix.
+
+    Missing, unparseable, or > 24h-future dates raise HTTP 400 so the
+    event is rejected for review instead of being re-stamped (#1167).
     """
     now = datetime.now(timezone.utc)
 
-    if not ship_date_raw:
-        return now.isoformat(), now.date().isoformat()
+    if not ship_date_raw or not str(ship_date_raw).strip():
+        raise HTTPException(
+            status_code=400,
+            detail={
+                "error": "missing_edi_date",
+                "raw": ship_date_raw,
+                "message": (
+                    "EDI event date is missing. The ingest time is not "
+                    "substituted for the event time; supply the source date."
+                ),
+            },
+        )
 
     clean_date = re.sub(r"\D", "", ship_date_raw)
 
@@ -170,12 +183,22 @@ def _ship_timestamp_and_date(ship_date_raw: Optional[str], ship_time_raw: Option
                 },
             ) from exc
 
-    # Keep webhook model bounds to avoid stale/future rejection — the
-    # ingest timestamp snaps to "now" if the EDI claims a shipment
-    # older than 90 days or > 24h in the future. The parsed event_date
-    # is returned unchanged so FSMA 204 traceability sees the real date.
-    if parsed < now - timedelta(days=90) or parsed > now + timedelta(hours=24):
-        return now.isoformat(), parsed.date().isoformat()
+    # Never replace the real event time with the ingest time. Historical
+    # dates are kept as-is; a date more than 24h ahead is rejected for
+    # review rather than overwritten.
+    if parsed > now + timedelta(hours=24):
+        raise HTTPException(
+            status_code=400,
+            detail={
+                "error": "edi_date_in_future",
+                "raw": ship_date_raw,
+                "needs_review": True,
+                "message": (
+                    "EDI event date is more than 24 hours in the future; "
+                    "flagged for review instead of being replaced with now()."
+                ),
+            },
+        )
 
     return parsed.isoformat(), parsed.date().isoformat()