From 6bc308afbdb874f573d8cb47480bf4ee0edc245e Mon Sep 17 00:00:00 2001
From: Christopher Sellers <1@christophersellers.com>
Date: Sun, 25 Oct 2026 01:13:44 -0700
Subject: [PATCH 01/10] fix(ingestion): keep real EDI event time instead of
 ingest time

Backfilled EDI older than 90 days had its timestamp snapped to now()
while event_date kept the real date, corrupting EPCIS eventTime and R1
time to fix. Keep the parsed source date; reject missing or >24h-future
dates with explicit HTTP 400 (future flagged needs_review).

Signed-off-by: Christopher Sellers <1@christophersellers.com>
---
 services/ingestion/app/edi_ingestion/utils.py | 51 ++++++++++-----
 .../tests/test_edi_preserve_event_time.py     | 63 +++++++++++++++++++
 2 files changed, 100 insertions(+), 14 deletions(-)
 create mode 100644 services/ingestion/tests/test_edi_preserve_event_time.py

diff --git a/services/ingestion/app/edi_ingestion/utils.py b/services/ingestion/app/edi_ingestion/utils.py
index 4c816d92c..d5ce3f8c0 100644
--- a/services/ingestion/app/edi_ingestion/utils.py
+++ b/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()
 
diff --git a/services/ingestion/tests/test_edi_preserve_event_time.py b/services/ingestion/tests/test_edi_preserve_event_time.py
new file mode 100644
index 000000000..e052489a0
--- /dev/null
+++ b/services/ingestion/tests/test_edi_preserve_event_time.py
@@ -0,0 +1,63 @@
+"""Data-free checks that EDI event time is never replaced by ingest time.
+
+Dates are constructed relative to now; no fixtures or records are used.
+"""
+
+from __future__ import annotations
+
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+
+service_dir = Path(__file__).parent.parent
+sys.path.insert(0, str(service_dir))
+
+pytest.importorskip("fastapi")
+
+from app.edi_ingestion.utils import _ship_timestamp_and_date  # noqa: E402
+from fastapi import HTTPException  # noqa: E402
+
+
+def _ccyymmdd(dt: datetime) -> str:
+    return dt.strftime("%Y%m%d")
+
+
+def test_backfilled_date_keeps_real_event_time():
+    real = datetime.now(timezone.utc) - timedelta(days=200)
+    timestamp, event_date = _ship_timestamp_and_date(_ccyymmdd(real), "1430")
+    parsed = datetime.fromisoformat(timestamp)
+    assert parsed.date() == real.date()
+    assert (parsed.hour, parsed.minute) == (14, 30)
+    assert event_date == real.date().isoformat()
+    assert timestamp[:10] == event_date
+
+
+def test_recent_date_unchanged():
+    real = datetime.now(timezone.utc) - timedelta(days=3)
+    timestamp, event_date = _ship_timestamp_and_date(_ccyymmdd(real), None)
+    assert timestamp[:10] == event_date == real.date().isoformat()
+
+
+def test_future_date_is_flagged_not_replaced():
+    future = datetime.now(timezone.utc) + timedelta(days=5)
+    with pytest.raises(HTTPException) as exc:
+        _ship_timestamp_and_date(_ccyymmdd(future), None)
+    assert exc.value.status_code == 400
+    assert exc.value.detail["error"] == "edi_date_in_future"
+    assert exc.value.detail["needs_review"] is True
+
+
+@pytest.mark.parametrize("raw", [None, "", "   "])
+def test_missing_date_is_rejected(raw):
+    with pytest.raises(HTTPException) as exc:
+        _ship_timestamp_and_date(raw, None)
+    assert exc.value.status_code == 400
+    assert exc.value.detail["error"] == "missing_edi_date"
+
+
+def test_invalid_date_is_rejected():
+    with pytest.raises(HTTPException) as exc:
+        _ship_timestamp_and_date("20261345", None)
+    assert exc.value.detail["error"] == "invalid_edi_date"

From 22b719eaa0ce86befe02b33c96f27f14c680b5b5 Mon Sep 17 00:00:00 2001
From: Christopher Sellers <1@christophersellers.com>
Date: Sun, 25 Oct 2026 22:08:03 -0700
Subject: [PATCH 02/10] feat(ingestion): store EDI ingest time separately and
 add event-time mismatch check

Adds the edi_ingested_at KDE alongside the real EDI event time, a read-only
non-production script that counts events whose timestamp and event date
disagree by more than one day, and documents/tests that R1 time to fix
(#3572) reads ingest-time clocks only.

Signed-off-by: Christopher Sellers <1@christophersellers.com>
---
 scripts/check_edi_event_time_mismatch.py      |  95 ++++++++++++++++
 .../ingestion/app/edi_ingestion/routes.py     |   8 +-
 services/ingestion/app/edi_ingestion/utils.py |   9 ++
 .../tests/test_edi_ingest_time_field.py       | 107 ++++++++++++++++++
 services/shared/data_gap_tracking.py          |   6 +
 5 files changed, 224 insertions(+), 1 deletion(-)
 create mode 100644 scripts/check_edi_event_time_mismatch.py
 create mode 100644 services/ingestion/tests/test_edi_ingest_time_field.py

diff --git a/scripts/check_edi_event_time_mismatch.py b/scripts/check_edi_event_time_mismatch.py
new file mode 100644
index 000000000..ad1aa1f5a
--- /dev/null
+++ b/scripts/check_edi_event_time_mismatch.py
@@ -0,0 +1,95 @@
+#!/usr/bin/env python3
+"""Count events whose stored timestamp and EDI event date disagree.
+
+Read-only data check for the EDI two-clocks bug (regengine-ds, sweep P2 5).
+An event disagrees when ``fsma.cte_events.event_timestamp`` is more than one
+day away from its ``ship_date`` / ``event_date`` KDE.
+
+Safety:
+* The query runs inside ``SET TRANSACTION READ ONLY`` and is rolled back.
+* Refuses to run when the database URL or ``REGENGINE_ENV`` / ``ENVIRONMENT``
+  names production. Point it at a local or scratch database only.
+* Never prints the database URL.
+
+Usage: python scripts/check_edi_event_time_mismatch.py --database-url URL
+       (or set DATABASE_URL). Prints one integer: the disagreeing count.
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+import sys
+from datetime import date, datetime, timedelta
+from typing import Iterable, Optional
+
+TOLERANCE = timedelta(days=1)
+DATE_KDE_KEYS = ("ship_date", "event_date")
+
+MISMATCH_SQL = """
+SELECT COUNT(*)
+  FROM fsma.cte_events e
+  JOIN fsma.cte_kdes k
+    ON k.cte_event_id = e.id
+   AND k.tenant_id = e.tenant_id
+   AND k.kde_key IN ('ship_date', 'event_date')
+ WHERE k.kde_value ~ '^\\d{4}-\\d{2}-\\d{2}$'
+   AND (e.event_timestamp < (CAST(k.kde_value AS date)::timestamp AT TIME ZONE 'UTC') - INTERVAL '1 day'
+        OR e.event_timestamp > (CAST(k.kde_value AS date)::timestamp AT TIME ZONE 'UTC') + INTERVAL '2 days')
+"""
+# The event date is a calendar day (UTC); its timestamp may fall anywhere in
+# it, so a mismatch is more than one day before its start or after its end.
+# Mirrors is_mismatch() below.
+
+_PROD_MARKERS = ("prod", "production")
+
+
+def is_mismatch(event_timestamp: datetime, event_date: date, tolerance: timedelta = TOLERANCE) -> bool:
+    """True when the timestamp is more than ``tolerance`` outside ``event_date``."""
+    day_start = datetime(event_date.year, event_date.month, event_date.day, tzinfo=event_timestamp.tzinfo)
+    day_end = day_start + timedelta(days=1)
+    return event_timestamp < day_start - tolerance or event_timestamp > day_end + tolerance
+
+
+def count_mismatches(rows: Iterable[tuple[datetime, date]], tolerance: timedelta = TOLERANCE) -> int:
+    return sum(1 for ts, d in rows if is_mismatch(ts, d, tolerance))
+
+
+def refuse_production(database_url: str, env: Optional[dict] = None) -> Optional[str]:
+    env = os.environ if env is None else env
+    for name in ("REGENGINE_ENV", "ENVIRONMENT", "ENV"):
+        if str(env.get(name, "")).lower() in _PROD_MARKERS:
+            return f"{name} names production"
+    if any(marker in database_url.lower() for marker in _PROD_MARKERS):
+        return "database URL looks like production"
+    return None
+
+
+def main(argv: Optional[list[str]] = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("--database-url", default=os.environ.get("DATABASE_URL", ""))
+    args = parser.parse_args(argv)
+    if not args.database_url:
+        print("error: --database-url or DATABASE_URL required", file=sys.stderr)
+        return 2
+    reason = refuse_production(args.database_url)
+    if reason:
+        print(f"refusing to run: {reason}", file=sys.stderr)
+        return 3
+
+    from sqlalchemy import create_engine, text
+
+    engine = create_engine(args.database_url)
+    with engine.connect() as conn:
+        trans = conn.begin()
+        try:
+            conn.execute(text("SET TRANSACTION READ ONLY"))
+            count = conn.execute(text(MISMATCH_SQL)).scalar_one()
+        finally:
+            trans.rollback()
+    print(int(count))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/services/ingestion/app/edi_ingestion/routes.py b/services/ingestion/app/edi_ingestion/routes.py
index f90831eeb..668b81dc7 100644
--- a/services/ingestion/app/edi_ingestion/routes.py
+++ b/services/ingestion/app/edi_ingestion/routes.py
@@ -62,6 +62,7 @@
     _split_transaction_sets,
 )
 from .utils import (
+    _ingest_time_iso,
     _normalize_unit,
     _resolve_unit_code,
     _safe_float,
@@ -582,6 +583,7 @@ async def ingest_edi_856(
             source=source,
         )
 
+    ingested_at = _ingest_time_iso()
     timestamp, ship_date = _ship_timestamp_and_date(
         extracted.get("ship_date_raw"),
         extracted.get("ship_time_raw"),
@@ -607,6 +609,7 @@ async def ingest_edi_856(
         line_tlc = lot.strip() if isinstance(lot, str) and lot.strip() else traceability_lot_code
         kdes: dict[str, Any] = {
             "ship_date": ship_date,
+            "edi_ingested_at": ingested_at,
             "ship_from_location": ship_from_name,
             "ship_to_location": ship_to_name,
             "ship_from_gln": ship_from_gln or ship_from_gln_raw,
@@ -805,6 +808,7 @@ def _build_ingest_event_for_set(
         "edi_gs_control_number": extracted.get("gs_control_number"),
         "reference_document_number": extracted.get("reference_document_number"),
         "event_date": event_date,
+        "edi_ingested_at": _ingest_time_iso(),
     }
 
     if transaction_set == "856":
@@ -961,7 +965,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"),
diff --git a/services/ingestion/app/edi_ingestion/utils.py b/services/ingestion/app/edi_ingestion/utils.py
index d5ce3f8c0..1a6a98b28 100644
--- a/services/ingestion/app/edi_ingestion/utils.py
+++ b/services/ingestion/app/edi_ingestion/utils.py
@@ -122,6 +122,15 @@ def _valid_gln_or_none(value: Optional[str]) -> Optional[str]:
     return clean if verify_gs1_check_digit(clean) else None
 
 
+def _ingest_time_iso() -> str:
+    """Wall-clock ingest time (UTC), kept separate from the EDI event time.
+
+    The event ``timestamp`` always carries the real EDI date; this value is
+    stored as the ``edi_ingested_at`` KDE so the two clocks never mix.
+    """
+    return datetime.now(timezone.utc).isoformat()
+
+
 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.
 
diff --git a/services/ingestion/tests/test_edi_ingest_time_field.py b/services/ingestion/tests/test_edi_ingest_time_field.py
new file mode 100644
index 000000000..e7408eaf4
--- /dev/null
+++ b/services/ingestion/tests/test_edi_ingest_time_field.py
@@ -0,0 +1,107 @@
+"""Data-free checks for the separate EDI ingest-time field and the
+timestamp/event_date mismatch check. No fixtures, records or database."""
+
+from __future__ import annotations
+
+import importlib.util
+import sys
+from datetime import date, datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+
+service_dir = Path(__file__).parent.parent
+repo_root = service_dir.parent.parent
+sys.path.insert(0, str(service_dir))
+
+pytest.importorskip("fastapi")
+
+from app.edi_ingestion.utils import _ingest_time_iso, _ship_timestamp_and_date  # noqa: E402
+
+_spec = importlib.util.spec_from_file_location(
+    "check_edi_event_time_mismatch", repo_root / "scripts" / "check_edi_event_time_mismatch.py"
+)
+check = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(check)
+
+UTC = timezone.utc
+
+
+def test_ingest_time_is_separate_from_backfilled_event_time():
+    real = datetime.now(UTC) - timedelta(days=200)
+    before = datetime.now(UTC)
+    ingested = datetime.fromisoformat(_ingest_time_iso())
+    timestamp, event_date = _ship_timestamp_and_date(real.strftime("%Y%m%d"), None)
+    assert ingested >= before
+    assert ingested.tzinfo is not None
+    assert timestamp[:10] == event_date == real.date().isoformat()
+    assert ingested.date() != date.fromisoformat(event_date)
+
+
+def test_routes_store_ingest_time_kde():
+    src = (service_dir / "app" / "edi_ingestion" / "routes.py").read_text()
+    assert src.count('"edi_ingested_at"') >= 2
+
+
+@pytest.mark.parametrize(
+    "ts, d, expected",
+    [
+        (datetime(2026, 3, 1, 14, 30, tzinfo=UTC), date(2026, 3, 1), False),
+        (datetime(2026, 3, 2, 23, 0, tzinfo=UTC), date(2026, 3, 1), False),
+        (datetime(2026, 2, 28, 1, 0, tzinfo=UTC), date(2026, 3, 1), False),
+        (datetime(2026, 3, 3, 0, 0, 1, tzinfo=UTC), date(2026, 3, 1), True),
+        (datetime(2026, 2, 27, 23, 59, tzinfo=UTC), date(2026, 3, 1), True),
+        (datetime(2026, 10, 25, 9, 0, tzinfo=UTC), date(2026, 4, 8), True),
+    ],
+)
+def test_is_mismatch(ts, d, expected):
+    assert check.is_mismatch(ts, d) is expected
+
+
+def test_count_mismatches_old_vs_new_behaviour():
+    day = date(2026, 4, 8)
+    ingest_now = datetime(2026, 10, 25, 9, 0, tzinfo=UTC)
+    real = datetime(2026, 4, 8, 14, 30, tzinfo=UTC)
+    assert check.count_mismatches([(ingest_now, day)] * 3) == 3  # snapped to ingest time
+    assert check.count_mismatches([(real, day)] * 3) == 0  # real event time kept
+
+
+def test_sql_is_read_only_count():
+    sql = check.MISMATCH_SQL.upper()
+    assert sql.lstrip().startswith("SELECT COUNT(*)")
+    for verb in ("INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "TRUNCATE"):
+        assert verb not in sql
+
+
+@pytest.mark.parametrize(
+    "url, env, refused",
+    [
+        ("postgresql://u@localhost:5432/regengine_test", {}, False),
+        ("postgresql://u@db.prod.example/regengine", {}, True),
+        ("postgresql://u@localhost/regengine", {"REGENGINE_ENV": "production"}, True),
+        ("postgresql://u@localhost/regengine", {"ENVIRONMENT": "Prod"}, True),
+    ],
+)
+def test_refuses_production(url, env, refused):
+    assert (check.refuse_production(url, env) is not None) is refused
+
+
+def test_main_requires_url(monkeypatch, capsys):
+    monkeypatch.delenv("DATABASE_URL", raising=False)
+    assert check.main([]) == 2
+
+
+def test_main_refuses_production_without_connecting(monkeypatch):
+    monkeypatch.setenv("REGENGINE_ENV", "production")
+    assert check.main(["--database-url", "postgresql://u@localhost/x"]) == 3
+
+
+def test_r1_time_to_fix_reads_ingest_clocks_not_event_time():
+    src = (repo_root / "services" / "shared" / "data_gap_tracking.py").read_text()
+    sql_blocks = [b for b in src.split('text(\n    """')[1:]]
+    assert sql_blocks
+    for block in sql_blocks:
+        body = block.split('"""')[0].lower()
+        assert "event_timestamp" not in body
+        assert "event_date" not in body
+    assert "closed_at = GREATEST(now(), opened_at)" in src
diff --git a/services/shared/data_gap_tracking.py b/services/shared/data_gap_tracking.py
index fe585ab48..cbdd818ab 100644
--- a/services/shared/data_gap_tracking.py
+++ b/services/shared/data_gap_tracking.py
@@ -14,6 +14,12 @@
   that lot and CTE, stamping the closing record id and its canonical SHA-256
   (the persisted ledger hash) as the receipt.
 
+Clock: time to fix is measured on ingest-time clocks only. ``opened_at``
+defaults to the database ``now()`` when the gap is flagged and ``closed_at``
+is ``now()`` when the fixing record lands. Neither reads the event
+``timestamp``/``event_date``, so backfilled EDI events (whose event time is
+months old) never inflate or shrink time to fix. Keep it that way.
+
 Gap logging is observability. It must never block or roll back ingest, so the
 writer catches every error, logs it, and returns.
 """

From f3c46c0c8c4f0d869ab4a972dde94c8b9d1cfb4d Mon Sep 17 00:00:00 2001
From: Christopher Sellers <1@christophersellers.com>
Date: Mon, 26 Oct 2026 00:11:26 -0700
Subject: [PATCH 03/10] fix(ingestion): validate EDI dates before reserving
 ISA13 and log rejections for review

Date rejections (missing_edi_date, invalid_edi_date, edi_date_in_future)
now run before check_and_record_interchange, so a corrected resend is not
refused as a 409 replay. Each rejection is written to the EDI rejection
log with needs_review=True and returns its rejection_id.

Signed-off-by: Christopher Sellers <1@christophersellers.com>
---
 .../app/edi_ingestion/rejection_log.py        |   5 +-
 .../ingestion/app/edi_ingestion/routes.py     |  99 +++++++++++++--
 .../tests/test_edi_date_reject_resend.py      | 118 ++++++++++++++++++
 .../test_edi_wire_format_2974_2976_2978.py    |  36 ++----
 4 files changed, 224 insertions(+), 34 deletions(-)
 create mode 100644 services/ingestion/tests/test_edi_date_reject_resend.py

diff --git a/services/ingestion/app/edi_ingestion/rejection_log.py b/services/ingestion/app/edi_ingestion/rejection_log.py
index c10eb5d02..13d3c9bab 100644
--- a/services/ingestion/app/edi_ingestion/rejection_log.py
+++ b/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)
diff --git a/services/ingestion/app/edi_ingestion/routes.py b/services/ingestion/app/edi_ingestion/routes.py
index f90831eeb..b502d1721 100644
--- a/services/ingestion/app/edi_ingestion/routes.py
+++ b/services/ingestion/app/edi_ingestion/routes.py
@@ -280,6 +280,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
@@ -537,6 +600,17 @@ async def ingest_edi_856(
 
     extracted = _extract_856_fields(segments)
 
+    # Date rejections must happen before the ISA13 is reserved, or the
+    # corrected resend would be refused as a 409 replay.
+    _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,
+    )
+
     # #1160 + #1165: envelope mismatch / allowlist / ISA13 dedup.
     _enforce_envelope_integrity(extracted, sender_tenant_id, "856")
 
@@ -774,15 +848,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 +1027,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,6 +1152,17 @@ async def ingest_edi_document(
 
     # #1160 + #1165: validate and reserve the shared envelope once per
     # document, before validating any transaction set or warehouse line.
+    # Date rejections happen before the ISA13 is reserved so a corrected
+    # resend is not refused as a 409 replay.
+    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,
+        )
     _enforce_envelope_integrity(extracted, sender_tenant_id, transaction_set)
 
     # 856 SN1, 861 RCD, and warehouse W07/W12 lines are each one CTE.
diff --git a/services/ingestion/tests/test_edi_date_reject_resend.py b/services/ingestion/tests/test_edi_date_reject_resend.py
new file mode 100644
index 000000000..431bb10d1
--- /dev/null
+++ b/services/ingestion/tests/test_edi_date_reject_resend.py
@@ -0,0 +1,118 @@
+"""A date-rejected EDI interchange must not burn its ISA13.
+
+Uses the repo's synthetic fixture ``tests/fixtures/edi_corpus/asn_856.edi``.
+The rejected variant only moves the BSN ship date to 5 days after now; the
+corrected resend is the unmodified fixture.
+"""
+
+from __future__ import annotations
+
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+
+service_dir = Path(__file__).parent.parent
+sys.path.insert(0, str(service_dir))
+
+pytest.importorskip("fastapi")
+
+import app.edi_ingestion.routes as edi_routes  # noqa: E402
+from app.authz import IngestionPrincipal, get_ingestion_principal  # noqa: E402
+from app.edi_ingestion import (
+    dedup as dedup_mod,  # noqa: E402
+    rejection_log,  # noqa: E402
+    router as edi_router,  # noqa: E402
+)
+from fastapi import FastAPI  # noqa: E402
+from fastapi.testclient import TestClient  # noqa: E402
+
+FIXTURE = service_dir / "tests" / "fixtures" / "edi_corpus" / "asn_856.edi"
+FIXTURE_BSN = "BSN*00*ASN-12345*20260310*1200~"
+
+
+class _NxStore:
+    """Minimal SET NX semantics, matching how dedup uses Redis."""
+
+    def __init__(self) -> None:
+        self.keys: dict[str, str] = {}
+
+    def set(self, key, value, nx=False, ex=None):
+        if nx and key in self.keys:
+            return None
+        self.keys[key] = value
+        return True
+
+    def get(self, key):
+        return self.keys.get(key)
+
+
+@pytest.fixture()
+def store(monkeypatch: pytest.MonkeyPatch) -> _NxStore:
+    fake = _NxStore()
+    monkeypatch.setattr(dedup_mod, "_redis_client", lambda: fake)
+    return fake
+
+
+@pytest.fixture()
+def client(monkeypatch: pytest.MonkeyPatch) -> TestClient:
+    monkeypatch.delenv("EDI_REQUIRE_PARTNER_ID", raising=False)
+    monkeypatch.setattr(edi_routes, "_resolve_principal_tenant_id", lambda *args: "tenant-a")
+    rejection_log.reset_edi_rejections()
+    app = FastAPI()
+    app.include_router(edi_router)
+    app.dependency_overrides[get_ingestion_principal] = lambda: IngestionPrincipal(key_id="", scopes=["edi.ingest"])
+    with TestClient(app) as test_client:
+        yield test_client
+
+
+def _future_variant() -> bytes:
+    raw = FIXTURE.read_text(encoding="utf-8")
+    assert FIXTURE_BSN in raw
+    future = (datetime.now(timezone.utc) + timedelta(days=5)).strftime("%Y%m%d")
+    return raw.replace(FIXTURE_BSN, f"BSN*00*ASN-12345*{future}*1200~").encode()
+
+
+def _post(client: TestClient, raw: bytes, route: str):
+    return client.post(
+        f"/api/v1/ingest/edi{route}",
+        data={"traceability_lot_code": "ASN-12345"},
+        files={"file": ("asn_856.edi", raw, "application/edi-x12")},
+    )
+
+
+@pytest.mark.parametrize("route", ["", "/document"])
+def test_reject_then_corrected_resend_is_not_a_replay(client, store, route):
+    rejected = _post(client, _future_variant(), route)
+    assert rejected.status_code == 400
+    detail = rejected.json()["detail"]
+    assert detail["error"] == "edi_date_in_future"
+    assert detail["needs_review"] is True
+    assert store.keys == {}, "ISA13 must not be reserved by a date rejection"
+
+    records = rejection_log.list_edi_rejections("tenant-a")
+    assert [r["rejection_id"] for r in records] == [detail["rejection_id"]]
+    assert records[0]["needs_review"] is True
+    assert records[0]["reason"] == "edi_date_in_future"
+
+    resend = _post(client, FIXTURE.read_bytes(), route)
+    assert resend.status_code != 409, resend.json()
+    assert len(store.keys) == 1, "corrected resend reserves the ISA13"
+
+    replay = _post(client, FIXTURE.read_bytes(), route)
+    assert replay.status_code == 409
+    assert replay.json()["detail"]["error"] == "duplicate_interchange"
+
+
+def test_missing_date_rejection_is_logged_for_review(client, store):
+    raw = FIXTURE.read_text(encoding="utf-8").replace(FIXTURE_BSN, "BSN*00*ASN-12345**1200~")
+    response = _post(client, raw.encode(), "/document")
+    assert response.status_code == 400
+    detail = response.json()["detail"]
+    assert detail["error"] == "missing_edi_date"
+    assert detail["needs_review"] is True
+    record = rejection_log.list_edi_rejections("tenant-a")[0]
+    assert record["rejection_id"] == detail["rejection_id"]
+    assert record["needs_review"] is True
+    assert store.keys == {}
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 cfd73035d..7799bb865 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
@@ -23,6 +23,7 @@
 
 import logging
 import sys
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 
 import pytest
@@ -50,8 +51,7 @@
 # 106-char ISA header (index 105 is the segment terminator the parser
 # sniffs). Shared by every envelope below; only ISA13 varies.
 _ISA_TEMPLATE = (
-    "ISA*00*          *00*          *ZZ*SENDERID       *ZZ*RECEIVERID     "
-    "*260310*1200*U*00401*{isa13}*0*P*>~"
+    "ISA*00*          *00*          *ZZ*SENDERID       *ZZ*RECEIVERID     *260310*1200*U*00401*{isa13}*0*P*>~"
 )
 _GS_856 = "GS*SH*SENDERID*RECEIVERID*20260310*1200*1*X*004010~"
 
@@ -374,9 +374,7 @@ def test_2978_unmapped_unit_warns_and_fails_closed_on_document_route(
 # ---------------------------------------------------------------------------
 
 
-def test_2974_multi_sn1_856_quantity_is_summed_not_last_wins(
-    client: TestClient, captured_payload: dict
-) -> None:
+def test_2974_multi_sn1_856_quantity_is_summed_not_last_wins(client: TestClient, captured_payload: dict) -> None:
     """Repeated SN1 lines each become a CTE. Last-line-wins and summing
     both drop a lot from the recall record.
     """
@@ -417,9 +415,7 @@ def test_2974_multi_sn1_856_quantity_is_summed_not_last_wins(
         isa13="000029742",
     )
     zero_extracted = _extract_fields_for_set("856", _parse_x12_segments(raw_zero.decode("utf-8")))
-    assert zero_extracted["line_items"][0]["quantity"] is None, (
-        "#2974: SN1**0*CA must not fabricate a quantity of 1.0"
-    )
+    assert zero_extracted["line_items"][0]["quantity"] is None, "#2974: SN1**0*CA must not fabricate a quantity of 1.0"
 
 
 def test_blank_quantity_element_is_present_and_not_synthesized() -> None:
@@ -447,14 +443,16 @@ def test_856_blank_sn1_quantity_is_rejected(client: TestClient, captured_payload
 
     The ship-from, ASN, BOL and other text elements contain no names,
     with SN1 quantity blank. The document must be refused with 422
-    edi_non_positive_quantity and nothing persisted.
+    edi_non_positive_quantity and nothing persisted. The BSN date is
+    yesterday so the earlier event-date gate passes.
     """
+    ship_date = (datetime.now(timezone.utc) - timedelta(days=1)).strftime("%Y%m%d").encode()
     raw = (
         b"ISA*00*          *00*          *ZZ*               *ZZ*               "
         b"*      *    *U*00401*         *0*P*>~"
         b"GS*SH*****1*X*004010~"
         b"ST*856*1~"
-        b"BSN*00**~"
+        b"BSN*00**" + ship_date + b"~"
         b"HL*1**S~"
         b"N1*SF*~"
         b"SN1***CA~"
@@ -481,8 +479,7 @@ def test_856_blank_sn1_quantity_is_rejected(client: TestClient, captured_payload
 
 
 def test_sn1_lines_with_different_units_stay_separate_events(client: TestClient, captured_payload: dict) -> None:
-    """120 cases and 80 pounds are two CTEs. They are not 200 of either unit.
-    """
+    """120 cases and 80 pounds are two CTEs. They are not 200 of either unit."""
     raw = _interchange(
         [
             _856_set(
@@ -505,9 +502,7 @@ def test_sn1_lines_with_different_units_stay_separate_events(client: TestClient,
     ]
 
 
-def test_2976_legacy_856_endpoint_refuses_a_batched_interchange(
-    client: TestClient, captured_payload: dict
-) -> None:
+def test_2976_legacy_856_endpoint_refuses_a_batched_interchange(client: TestClient, captured_payload: dict) -> None:
     """The legacy single-ASN endpoint extracts from the flat segment list,
     so a batched interchange would merge every ST/SE pair into its one
     event. It must refuse and point the caller at ``/document``.
@@ -588,8 +583,7 @@ def test_2976_batched_interchange_never_half_lands(client: TestClient, captured_
 
 
 def test_sn1_equivalent_unit_codes_stay_on_their_own_lines():
-    """CA and CS both mean cases, and each SN1 still keeps its own quantity.
-    """
+    """CA and CS both mean cases, and each SN1 still keeps its own quantity."""
     from app.edi_ingestion.extractors import _extract_856_fields
     from app.edi_ingestion.parser import _parse_x12_segments
     from app.edi_ingestion.utils import _resolve_unit_code
@@ -629,9 +623,7 @@ def test_two_unmappable_unit_codes_stay_on_their_lines():
     assert [line["unit_of_measure"] for line in data["line_items"]] == ["QQ", "ZZ"]
 
 
-def test_scalar_overrides_are_refused_for_a_batched_interchange(
-    client: TestClient, captured_payload: dict
-):
+def test_scalar_overrides_are_refused_for_a_batched_interchange(client: TestClient, captured_payload: dict):
     """A scalar form override describes ONE shipment.
 
     Applying it across a batch would overwrite each transaction set's own
@@ -662,9 +654,7 @@ def test_scalar_overrides_are_refused_for_a_batched_interchange(
     assert "payload" not in captured_payload, "nothing may land when overrides are refused"
 
 
-def test_scalar_overrides_still_work_for_a_single_transaction_set(
-    client: TestClient, captured_payload: dict
-):
+def test_scalar_overrides_still_work_for_a_single_transaction_set(client: TestClient, captured_payload: dict):
     """The guard must not break the ordinary single-shipment override path."""
     raw = _interchange(
         [_856_set(control="0001", asn="ASN-SOLO", ship_from="Valley Fresh Farms", bol="BOL-SOLO")],

From 0e933a2dc5eedfe1fd8f905c1dc104ec341324ab Mon Sep 17 00:00:00 2001
From: Christopher Sellers <1@christophersellers.com>
Date: Mon, 26 Oct 2026 00:22:45 -0700
Subject: [PATCH 04/10] fix(ingestion): keep EDI ingest time out of hashed
 KDEs; count distinct events

- Drop the edi_ingested_at KDE. It was a fresh timestamp inside the fields
  compute_idempotency_key hashes, so a resend got a new key and backup
  dedupe broke when Redis was down. Ingest time is the existing
  fsma.cte_events.ingested_at column.
- check_edi_event_time_mismatch.py: COUNT(DISTINCT e.id) so an event with
  both ship_date and event_date KDEs counts once.
- Tests: same-document-same-key test on the synthetic asn_856 fixture;
  mismatch dates built relative to now.

Signed-off-by: Christopher Sellers <1@christophersellers.com>
---
 scripts/check_edi_event_time_mismatch.py      |   5 +-
 .../ingestion/app/edi_ingestion/routes.py     |   4 -
 services/ingestion/app/edi_ingestion/utils.py |   9 --
 .../tests/test_edi_ingest_time_field.py       | 102 +++++++++++++-----
 4 files changed, 80 insertions(+), 40 deletions(-)

diff --git a/scripts/check_edi_event_time_mismatch.py b/scripts/check_edi_event_time_mismatch.py
index ad1aa1f5a..75f15a968 100644
--- a/scripts/check_edi_event_time_mismatch.py
+++ b/scripts/check_edi_event_time_mismatch.py
@@ -27,7 +27,7 @@
 DATE_KDE_KEYS = ("ship_date", "event_date")
 
 MISMATCH_SQL = """
-SELECT COUNT(*)
+SELECT COUNT(DISTINCT e.id)
   FROM fsma.cte_events e
   JOIN fsma.cte_kdes k
     ON k.cte_event_id = e.id
@@ -39,7 +39,8 @@
 """
 # The event date is a calendar day (UTC); its timestamp may fall anywhere in
 # it, so a mismatch is more than one day before its start or after its end.
-# Mirrors is_mismatch() below.
+# Mirrors is_mismatch() below. COUNT(DISTINCT e.id) counts each event once
+# even when it carries both a ship_date and an event_date KDE.
 
 _PROD_MARKERS = ("prod", "production")
 
diff --git a/services/ingestion/app/edi_ingestion/routes.py b/services/ingestion/app/edi_ingestion/routes.py
index 3f0c1d8bc..b502d1721 100644
--- a/services/ingestion/app/edi_ingestion/routes.py
+++ b/services/ingestion/app/edi_ingestion/routes.py
@@ -62,7 +62,6 @@
     _split_transaction_sets,
 )
 from .utils import (
-    _ingest_time_iso,
     _normalize_unit,
     _resolve_unit_code,
     _safe_float,
@@ -657,7 +656,6 @@ async def ingest_edi_856(
             source=source,
         )
 
-    ingested_at = _ingest_time_iso()
     timestamp, ship_date = _ship_timestamp_and_date(
         extracted.get("ship_date_raw"),
         extracted.get("ship_time_raw"),
@@ -683,7 +681,6 @@ async def ingest_edi_856(
         line_tlc = lot.strip() if isinstance(lot, str) and lot.strip() else traceability_lot_code
         kdes: dict[str, Any] = {
             "ship_date": ship_date,
-            "edi_ingested_at": ingested_at,
             "ship_from_location": ship_from_name,
             "ship_to_location": ship_to_name,
             "ship_from_gln": ship_from_gln or ship_from_gln_raw,
@@ -874,7 +871,6 @@ def _build_ingest_event_for_set(
         "edi_gs_control_number": extracted.get("gs_control_number"),
         "reference_document_number": extracted.get("reference_document_number"),
         "event_date": event_date,
-        "edi_ingested_at": _ingest_time_iso(),
     }
 
     if transaction_set == "856":
diff --git a/services/ingestion/app/edi_ingestion/utils.py b/services/ingestion/app/edi_ingestion/utils.py
index 1a6a98b28..d5ce3f8c0 100644
--- a/services/ingestion/app/edi_ingestion/utils.py
+++ b/services/ingestion/app/edi_ingestion/utils.py
@@ -122,15 +122,6 @@ def _valid_gln_or_none(value: Optional[str]) -> Optional[str]:
     return clean if verify_gs1_check_digit(clean) else None
 
 
-def _ingest_time_iso() -> str:
-    """Wall-clock ingest time (UTC), kept separate from the EDI event time.
-
-    The event ``timestamp`` always carries the real EDI date; this value is
-    stored as the ``edi_ingested_at`` KDE so the two clocks never mix.
-    """
-    return datetime.now(timezone.utc).isoformat()
-
-
 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.
 
diff --git a/services/ingestion/tests/test_edi_ingest_time_field.py b/services/ingestion/tests/test_edi_ingest_time_field.py
index e7408eaf4..f8e268c01 100644
--- a/services/ingestion/tests/test_edi_ingest_time_field.py
+++ b/services/ingestion/tests/test_edi_ingest_time_field.py
@@ -1,10 +1,13 @@
-"""Data-free checks for the separate EDI ingest-time field and the
-timestamp/event_date mismatch check. No fixtures, records or database."""
+"""Checks that EDI ingest time stays out of hashed KDEs, and the
+timestamp/event_date mismatch check. Dates are built relative to now; the
+only input file is the repo's synthetic fixture edi_corpus/asn_856.edi.
+No database."""
 
 from __future__ import annotations
 
 import importlib.util
 import sys
+import time
 from datetime import date, datetime, timedelta, timezone
 from pathlib import Path
 
@@ -16,7 +19,11 @@
 
 pytest.importorskip("fastapi")
 
-from app.edi_ingestion.utils import _ingest_time_iso, _ship_timestamp_and_date  # noqa: E402
+import app.edi_ingestion.routes as edi_routes  # noqa: E402
+from app.edi_ingestion.extractors import _extract_fields_for_set  # noqa: E402
+from app.edi_ingestion.parser import _parse_x12_segments  # noqa: E402
+from app.edi_ingestion.utils import _ship_timestamp_and_date  # noqa: E402
+from shared.cte_persistence.hashing import compute_idempotency_key  # noqa: E402
 
 _spec = importlib.util.spec_from_file_location(
     "check_edi_event_time_mismatch", repo_root / "scripts" / "check_edi_event_time_mismatch.py"
@@ -27,48 +34,93 @@
 UTC = timezone.utc
 
 
-def test_ingest_time_is_separate_from_backfilled_event_time():
+FIXTURE = service_dir / "tests" / "fixtures" / "edi_corpus" / "asn_856.edi"
+
+
+def _fixture_event():
+    """Build the 856 event from the repo's synthetic fixture, as /document does."""
+    segments = _parse_x12_segments(FIXTURE.read_text(encoding="utf-8"))
+    extracted = _extract_fields_for_set("856", segments)
+    header = {k: v for k, v in extracted.items() if k != "line_items"}
+    merged = {**header, **extracted["line_items"][0], "edi_line_index": 1}
+    return edi_routes._build_ingest_event_for_set(
+        "856", merged, extracted["asn_number"], None, None, None, None, segments=segments
+    )
+
+
+def _key(event) -> str:
+    return compute_idempotency_key(
+        event.cte_type.value,
+        event.traceability_lot_code,
+        event.timestamp,
+        "edi_inbound",
+        dict(event.kdes),
+        location_gln=event.location_gln,
+        location_name=event.location_name,
+    )
+
+
+def test_same_document_same_idempotency_key():
+    """A resend must hash the same, so backup dedupe works when Redis is down.
+
+    Ingest time lives in the ``fsma.cte_events.ingested_at`` column, never in
+    the hashed KDEs.
+    """
+    first = _fixture_event()
+    time.sleep(0.01)
+    second = _fixture_event()
+    assert "edi_ingested_at" not in first.kdes
+    assert first.kdes == second.kdes
+    assert _key(first) == _key(second)
+
+
+def test_routes_do_not_put_ingest_time_in_kdes():
+    src = (service_dir / "app" / "edi_ingestion" / "routes.py").read_text()
+    assert "edi_ingested_at" not in src
+
+
+def test_backfilled_event_time_is_kept():
     real = datetime.now(UTC) - timedelta(days=200)
-    before = datetime.now(UTC)
-    ingested = datetime.fromisoformat(_ingest_time_iso())
     timestamp, event_date = _ship_timestamp_and_date(real.strftime("%Y%m%d"), None)
-    assert ingested >= before
-    assert ingested.tzinfo is not None
     assert timestamp[:10] == event_date == real.date().isoformat()
-    assert ingested.date() != date.fromisoformat(event_date)
 
 
-def test_routes_store_ingest_time_kde():
-    src = (service_dir / "app" / "edi_ingestion" / "routes.py").read_text()
-    assert src.count('"edi_ingested_at"') >= 2
+def _day(offset_days: int) -> date:
+    return (datetime.now(UTC) - timedelta(days=offset_days)).date()
+
+
+def _at(d: date, days: int = 0, hours: int = 0, minutes: int = 0, seconds: int = 0) -> datetime:
+    start = datetime(d.year, d.month, d.day, tzinfo=UTC)
+    return start + timedelta(days=days, hours=hours, minutes=minutes, seconds=seconds)
 
 
 @pytest.mark.parametrize(
-    "ts, d, expected",
+    "days, hours, minutes, seconds, expected",
     [
-        (datetime(2026, 3, 1, 14, 30, tzinfo=UTC), date(2026, 3, 1), False),
-        (datetime(2026, 3, 2, 23, 0, tzinfo=UTC), date(2026, 3, 1), False),
-        (datetime(2026, 2, 28, 1, 0, tzinfo=UTC), date(2026, 3, 1), False),
-        (datetime(2026, 3, 3, 0, 0, 1, tzinfo=UTC), date(2026, 3, 1), True),
-        (datetime(2026, 2, 27, 23, 59, tzinfo=UTC), date(2026, 3, 1), True),
-        (datetime(2026, 10, 25, 9, 0, tzinfo=UTC), date(2026, 4, 8), True),
+        (0, 14, 30, 0, False),  # same day
+        (1, 23, 0, 0, False),  # within a day after the day ends
+        (-1, 1, 0, 0, False),  # within a day before the day starts
+        (2, 0, 0, 1, True),  # just over a day after the day ends
+        (-1, 0, -1, 0, True),  # just over a day before the day starts
+        (200, 9, 0, 0, True),  # snapped to an ingest time months later
     ],
 )
-def test_is_mismatch(ts, d, expected):
-    assert check.is_mismatch(ts, d) is expected
+def test_is_mismatch(days, hours, minutes, seconds, expected):
+    d = _day(200)
+    assert check.is_mismatch(_at(d, days, hours, minutes, seconds), d) is expected
 
 
 def test_count_mismatches_old_vs_new_behaviour():
-    day = date(2026, 4, 8)
-    ingest_now = datetime(2026, 10, 25, 9, 0, tzinfo=UTC)
-    real = datetime(2026, 4, 8, 14, 30, tzinfo=UTC)
+    day = _day(200)
+    ingest_now = datetime.now(UTC)
+    real = _at(day, hours=14, minutes=30)
     assert check.count_mismatches([(ingest_now, day)] * 3) == 3  # snapped to ingest time
     assert check.count_mismatches([(real, day)] * 3) == 0  # real event time kept
 
 
 def test_sql_is_read_only_count():
     sql = check.MISMATCH_SQL.upper()
-    assert sql.lstrip().startswith("SELECT COUNT(*)")
+    assert sql.lstrip().startswith("SELECT COUNT(DISTINCT E.ID)")
     for verb in ("INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "TRUNCATE"):
         assert verb not in sql
 

From f6eb471c9a2377ddb742c8e08fc9c90f8e8ce298 Mon Sep 17 00:00:00 2001
From: Christopher Sellers <1@christophersellers.com>
Date: Mon, 26 Oct 2026 00:43:03 -0700
Subject: [PATCH 05/10] fix(ingestion): run EDI date check after allowlist,
 before ISA13 reserve

_enforce_envelope_integrity takes a before_reserve hook that runs after
the envelope and trading-partner checks and before the ISA13 is recorded.
The event-date preflight moved there, so a disallowed sender gets 403
with no rejection record. Tests now prove a corrected resend is accepted
(201) on both routes and that a disallowed sender with a bad date is 403.

Signed-off-by: Christopher Sellers <1@christophersellers.com>
---
 .../ingestion/app/edi_ingestion/routes.py     | 62 +++++++++-------
 .../tests/test_edi_date_reject_resend.py      | 71 ++++++++++++++++---
 2 files changed, 101 insertions(+), 32 deletions(-)

diff --git a/services/ingestion/app/edi_ingestion/routes.py b/services/ingestion/app/edi_ingestion/routes.py
index b502d1721..bb2c94ad2 100644
--- a/services/ingestion/app/edi_ingestion/routes.py
+++ b/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
@@ -353,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
@@ -405,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"),
@@ -600,20 +610,22 @@ async def ingest_edi_856(
 
     extracted = _extract_856_fields(segments)
 
-    # Date rejections must happen before the ISA13 is reserved, or the
-    # corrected resend would be refused as a 409 replay.
-    _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,
+    # 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,
+        ),
     )
 
-    # #1160 + #1165: envelope mismatch / allowlist / ISA13 dedup.
-    _enforce_envelope_integrity(extracted, sender_tenant_id, "856")
-
     lines = extracted.get("line_items") or []
     if len(lines) > 1:
         supplied_overrides = [
@@ -1152,18 +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.
-    # Date rejections happen before the ISA13 is reserved so a corrected
-    # resend is not refused as a 409 replay.
-    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,
-        )
-    _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
diff --git a/services/ingestion/tests/test_edi_date_reject_resend.py b/services/ingestion/tests/test_edi_date_reject_resend.py
index 431bb10d1..2e3af20a5 100644
--- a/services/ingestion/tests/test_edi_date_reject_resend.py
+++ b/services/ingestion/tests/test_edi_date_reject_resend.py
@@ -25,11 +25,17 @@
     rejection_log,  # noqa: E402
     router as edi_router,  # noqa: E402
 )
+from app.webhook_models import EventResult, IngestResponse  # noqa: E402
 from fastapi import FastAPI  # noqa: E402
 from fastapi.testclient import TestClient  # noqa: E402
 
 FIXTURE = service_dir / "tests" / "fixtures" / "edi_corpus" / "asn_856.edi"
 FIXTURE_BSN = "BSN*00*ASN-12345*20260310*1200~"
+TENANT = "00000000-0000-0000-0000-000000000123"
+# GTIN-14 + lot suffix, the FSMA 204 TLC format strict validation accepts
+# (same value test_edi_wire_format_2974_2976_2978 uses).
+VALID_TLC = "00012345678901-LOT001"
+ALLOWLIST_ENV = "EDI_PARTNER_ALLOWLIST_" + TENANT.upper().replace("-", "_")
 
 
 class _NxStore:
@@ -56,13 +62,44 @@ def store(monkeypatch: pytest.MonkeyPatch) -> _NxStore:
 
 
 @pytest.fixture()
-def client(monkeypatch: pytest.MonkeyPatch) -> TestClient:
+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):
+        calls.append(payload)
+        return IngestResponse(
+            accepted=len(payload.events),
+            rejected=0,
+            total=len(payload.events),
+            events=[
+                EventResult(
+                    traceability_lot_code=event.traceability_lot_code,
+                    cte_type=event.cte_type.value,
+                    status="accepted",
+                    event_id=f"evt-{idx}",
+                    sha256_hash="hash",
+                    chain_hash="chain",
+                )
+                for idx, event in enumerate(payload.events)
+            ],
+        )
+
+    monkeypatch.setattr(edi_routes, "ingest_events", _fake_ingest_events)
+    return calls
+
+
+@pytest.fixture()
+def client(monkeypatch: pytest.MonkeyPatch, ingested: list) -> TestClient:
     monkeypatch.delenv("EDI_REQUIRE_PARTNER_ID", raising=False)
-    monkeypatch.setattr(edi_routes, "_resolve_principal_tenant_id", lambda *args: "tenant-a")
+    monkeypatch.delenv(ALLOWLIST_ENV, raising=False)
+    monkeypatch.delenv("EDI_PARTNER_ALLOWLIST", raising=False)
     rejection_log.reset_edi_rejections()
     app = FastAPI()
     app.include_router(edi_router)
-    app.dependency_overrides[get_ingestion_principal] = lambda: IngestionPrincipal(key_id="", scopes=["edi.ingest"])
+    app.dependency_overrides[get_ingestion_principal] = lambda: IngestionPrincipal(
+        key_id="test-key", scopes=["*"], auth_mode="test"
+    )
     with TestClient(app) as test_client:
         yield test_client
 
@@ -77,13 +114,13 @@ def _future_variant() -> bytes:
 def _post(client: TestClient, raw: bytes, route: str):
     return client.post(
         f"/api/v1/ingest/edi{route}",
-        data={"traceability_lot_code": "ASN-12345"},
+        data={"traceability_lot_code": VALID_TLC, "tenant_id": TENANT},
         files={"file": ("asn_856.edi", raw, "application/edi-x12")},
     )
 
 
 @pytest.mark.parametrize("route", ["", "/document"])
-def test_reject_then_corrected_resend_is_not_a_replay(client, store, route):
+def test_reject_then_corrected_resend_is_accepted(client, store, ingested, route):
     rejected = _post(client, _future_variant(), route)
     assert rejected.status_code == 400
     detail = rejected.json()["detail"]
@@ -91,18 +128,25 @@ def test_reject_then_corrected_resend_is_not_a_replay(client, store, route):
     assert detail["needs_review"] is True
     assert store.keys == {}, "ISA13 must not be reserved by a date rejection"
 
-    records = rejection_log.list_edi_rejections("tenant-a")
+    assert ingested == []
+    records = rejection_log.list_edi_rejections(TENANT)
     assert [r["rejection_id"] for r in records] == [detail["rejection_id"]]
     assert records[0]["needs_review"] is True
     assert records[0]["reason"] == "edi_date_in_future"
 
     resend = _post(client, FIXTURE.read_bytes(), route)
-    assert resend.status_code != 409, resend.json()
+    assert resend.status_code == 201, resend.json()
+    assert resend.json()["ingestion_result"]["accepted"] == 1
+    assert len(ingested) == 1
+    event = ingested[0].events[0]
+    assert event.traceability_lot_code == VALID_TLC
+    assert event.timestamp.startswith("2026-03-10")
     assert len(store.keys) == 1, "corrected resend reserves the ISA13"
 
     replay = _post(client, FIXTURE.read_bytes(), route)
     assert replay.status_code == 409
     assert replay.json()["detail"]["error"] == "duplicate_interchange"
+    assert len(ingested) == 1
 
 
 def test_missing_date_rejection_is_logged_for_review(client, store):
@@ -112,7 +156,18 @@ def test_missing_date_rejection_is_logged_for_review(client, store):
     detail = response.json()["detail"]
     assert detail["error"] == "missing_edi_date"
     assert detail["needs_review"] is True
-    record = rejection_log.list_edi_rejections("tenant-a")[0]
+    record = rejection_log.list_edi_rejections(TENANT)[0]
     assert record["rejection_id"] == detail["rejection_id"]
     assert record["needs_review"] is True
     assert store.keys == {}
+
+
+@pytest.mark.parametrize("route", ["", "/document"])
+def test_disallowed_sender_with_bad_date_gets_403_and_no_rejection(client, store, ingested, monkeypatch, route):
+    monkeypatch.setenv(ALLOWLIST_ENV, "SOMEONEELSE")
+    response = _post(client, _future_variant(), route)
+    assert response.status_code == 403, response.json()
+    assert response.json()["detail"]["error"] == "trading_partner_not_allowed"
+    assert rejection_log.list_edi_rejections(TENANT) == []
+    assert store.keys == {}
+    assert ingested == []

From 598afcdc3112d4d17ca8d5100d5a8e6295b0cf3f Mon Sep 17 00:00:00 2001
From: Christopher Sellers <1@christophersellers.com>
Date: Mon, 26 Oct 2026 01:15:38 -0700
Subject: [PATCH 06/10] fix(scripts): read JSONB kde_value as text in EDI
 mismatch check

fsma.cte_kdes.kde_value is JSONB since v078. Extract the scalar with
#>> '{}' before the date regex and cast, as compliance_score.py does.
Add a test that runs MISMATCH_SQL against a real JSONB column in a
disposable Postgres (rls_postgres_or_skip), rolled back.

Signed-off-by: Christopher Sellers <1@christophersellers.com>
---
 scripts/check_edi_event_time_mismatch.py      |  8 +-
 .../test_edi_event_time_mismatch_jsonb.py     | 99 +++++++++++++++++++
 2 files changed, 104 insertions(+), 3 deletions(-)
 create mode 100644 tests/migrations/test_edi_event_time_mismatch_jsonb.py

diff --git a/scripts/check_edi_event_time_mismatch.py b/scripts/check_edi_event_time_mismatch.py
index 75f15a968..491c0af3f 100644
--- a/scripts/check_edi_event_time_mismatch.py
+++ b/scripts/check_edi_event_time_mismatch.py
@@ -33,10 +33,12 @@
     ON k.cte_event_id = e.id
    AND k.tenant_id = e.tenant_id
    AND k.kde_key IN ('ship_date', 'event_date')
- WHERE k.kde_value ~ '^\\d{4}-\\d{2}-\\d{2}$'
-   AND (e.event_timestamp < (CAST(k.kde_value AS date)::timestamp AT TIME ZONE 'UTC') - INTERVAL '1 day'
-        OR e.event_timestamp > (CAST(k.kde_value AS date)::timestamp AT TIME ZONE 'UTC') + INTERVAL '2 days')
+ WHERE (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$'
+   AND (e.event_timestamp < ((k.kde_value #>> '{}')::date::timestamp AT TIME ZONE 'UTC') - INTERVAL '1 day'
+        OR e.event_timestamp > ((k.kde_value #>> '{}')::date::timestamp AT TIME ZONE 'UTC') + INTERVAL '2 days')
 """
+# fsma.cte_kdes.kde_value is JSONB (since v078); ``#>> '{}'`` extracts the
+# scalar as text before matching and casting, as compliance_score.py does.
 # The event date is a calendar day (UTC); its timestamp may fall anywhere in
 # it, so a mismatch is more than one day before its start or after its end.
 # Mirrors is_mismatch() below. COUNT(DISTINCT e.id) counts each event once
diff --git a/tests/migrations/test_edi_event_time_mismatch_jsonb.py b/tests/migrations/test_edi_event_time_mismatch_jsonb.py
new file mode 100644
index 000000000..b6ab5c9ef
--- /dev/null
+++ b/tests/migrations/test_edi_event_time_mismatch_jsonb.py
@@ -0,0 +1,99 @@
+"""Run the EDI event-time mismatch query against a real JSONB ``kde_value``.
+
+``fsma.cte_kdes.kde_value`` has been JSONB since migration v078, so the
+query in ``scripts/check_edi_event_time_mismatch.py`` must extract the scalar
+text (``#>> '{}'``) before its regex match and date cast. This test builds a
+minimal ``fsma.cte_events`` / ``fsma.cte_kdes`` pair with a JSONB column in a
+disposable Postgres (``rls_postgres_or_skip``: the CI service container or a
+testcontainers Postgres), runs ``MISMATCH_SQL`` inside a transaction and rolls
+it back. It never runs the script itself and never touches a real database.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import uuid
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+from sqlalchemy import text
+
+from tests.migrations.rls_db import rls_postgres_or_skip
+
+_SCRIPT = Path(__file__).resolve().parents[2] / "scripts" / "check_edi_event_time_mismatch.py"
+_spec = importlib.util.spec_from_file_location("check_edi_event_time_mismatch", _SCRIPT)
+check = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(check)  # type: ignore[union-attr]
+
+
+def test_mismatch_sql_extracts_jsonb_scalar_text() -> None:
+    sql = check.MISMATCH_SQL
+    assert "#>> '{}'" in sql
+    assert "CAST(k.kde_value" not in sql
+    assert "k.kde_value ~" not in sql
+
+
+def test_mismatch_sql_counts_against_real_jsonb_column() -> None:
+    now = datetime.now(timezone.utc).replace(microsecond=0)
+    day = (now - timedelta(days=10)).date()
+    day_start = datetime(day.year, day.month, day.day, tzinfo=timezone.utc)
+    tenant = uuid.uuid4()
+    with rls_postgres_or_skip() as (_url, engine):
+        with engine.connect() as conn:
+            trans = conn.begin()
+            try:
+                exists = conn.execute(text("SELECT to_regclass('fsma.cte_kdes')")).scalar()
+                if exists is not None:
+                    pytest.skip("fsma.cte_kdes already exists in this database; not seeding a shared schema")
+                conn.execute(text("CREATE SCHEMA fsma"))
+                conn.execute(
+                    text(
+                        "CREATE TABLE fsma.cte_events (id uuid PRIMARY KEY, tenant_id uuid NOT NULL,"
+                        " event_timestamp timestamptz NOT NULL)"
+                    )
+                )
+                conn.execute(
+                    text(
+                        "CREATE TABLE fsma.cte_kdes (cte_event_id uuid NOT NULL, tenant_id uuid NOT NULL,"
+                        " kde_key text NOT NULL, kde_value jsonb)"
+                    )
+                )
+
+                def add(ts: datetime, kdes: dict) -> None:
+                    eid = uuid.uuid4()
+                    conn.execute(
+                        text("INSERT INTO fsma.cte_events VALUES (:i, :t, :ts)"),
+                        {"i": eid, "t": tenant, "ts": ts},
+                    )
+                    for k, v in kdes.items():
+                        conn.execute(
+                            text("INSERT INTO fsma.cte_kdes VALUES (:i, :t, :k, CAST(:v AS jsonb))"),
+                            {"i": eid, "t": tenant, "k": k, "v": json.dumps(v)},
+                        )
+
+                iso = day.isoformat()
+                # Agrees: timestamp inside the event day.
+                add(day_start + timedelta(hours=12), {"ship_date": iso})
+                # Disagrees by ingest-time drift; both date KDEs -> counted once.
+                add(now, {"ship_date": iso, "event_date": iso})
+                # Disagrees, event_date only.
+                add(day_start - timedelta(days=3), {"event_date": iso})
+                # Non-date JSONB values are ignored, not cast errors.
+                add(now, {"ship_date": "not-a-date"})
+                add(now, {"event_date": {"nested": iso}})
+                add(now, {"ship_date": None})
+
+                count = conn.execute(text(check.MISMATCH_SQL)).scalar()
+                assert count == 2
+                rows = conn.execute(
+                    text(
+                        "SELECT e.event_timestamp, (k.kde_value #>> '{}')::date FROM fsma.cte_events e"
+                        " JOIN fsma.cte_kdes k ON k.cte_event_id = e.id AND k.kde_key = 'ship_date'"
+                        " WHERE (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$'"
+                    )
+                ).all()
+                assert check.count_mismatches(rows) == 1
+            finally:
+                trans.rollback()

From 2c94f47ce157bc39c815f1299b36d873e71112a1 Mon Sep 17 00:00:00 2001
From: Christopher Sellers <1@christophersellers.com>
Date: Tue, 27 Oct 2026 22:27:33 -0700
Subject: [PATCH 07/10] fix(ingestion): harden EDI event-time mismatch check
 for Gemini FAIL

Address Gemini's FAIL on #3612 and the six open review threads:
- Rewrite bare postgresql:// URLs to postgresql+psycopg:// via
  shared.db_url.sync_postgres_url (psycopg2 is not installed).
- Replace the production deny-list with a fail-closed host allow-list
  (localhost/loopback/compose) plus --allow-host for scratch DBs.
- Filter MISMATCH_SQL to e.source LIKE 'edi%' and skip calendar-invalid
  date KDEs via to_date round-trip; report invalids on stderr.
- Require --tenant-id or --all-tenants so the all-tenant scan cannot run
  under the ordinary RLS-enforcing DATABASE_URL role by accident.
- Document closed_at = GREATEST(now(), opened_at) in data_gap_tracking.

Signed-off-by: Christopher Sellers <1@christophersellers.com>
---
 scripts/check_edi_event_time_mismatch.py      | 183 +++++++++++++++---
 .../tests/test_edi_ingest_time_field.py       |  99 +++++++++-
 services/shared/data_gap_tracking.py          |   6 +-
 .../test_edi_event_time_mismatch_jsonb.py     |  24 ++-
 4 files changed, 269 insertions(+), 43 deletions(-)

diff --git a/scripts/check_edi_event_time_mismatch.py b/scripts/check_edi_event_time_mismatch.py
index 491c0af3f..c9fa0cbe1 100644
--- a/scripts/check_edi_event_time_mismatch.py
+++ b/scripts/check_edi_event_time_mismatch.py
@@ -1,18 +1,27 @@
 #!/usr/bin/env python3
-"""Count events whose stored timestamp and EDI event date disagree.
+"""Count EDI events whose stored timestamp and EDI event date disagree.
 
 Read-only data check for the EDI two-clocks bug (regengine-ds, sweep P2 5).
 An event disagrees when ``fsma.cte_events.event_timestamp`` is more than one
-day away from its ``ship_date`` / ``event_date`` KDE.
+day away from its ``ship_date`` / ``event_date`` KDE. Only rows whose
+``source`` starts with ``edi`` are counted (EDI ingest paths).
 
 Safety:
 * The query runs inside ``SET TRANSACTION READ ONLY`` and is rolled back.
-* Refuses to run when the database URL or ``REGENGINE_ENV`` / ``ENVIRONMENT``
-  names production. Point it at a local or scratch database only.
-* Never prints the database URL.
-
-Usage: python scripts/check_edi_event_time_mismatch.py --database-url URL
-       (or set DATABASE_URL). Prints one integer: the disagreeing count.
+* Fail-closed host allow-list: only localhost / loopback / local compose
+  hosts run by default. Pass ``--allow-host HOST`` for a remote scratch DB.
+  ``REGENGINE_ENV`` / ``ENVIRONMENT`` / ``ENV`` naming production always
+  refuses. Never prints the database URL.
+* Bare ``postgresql://`` URLs are rewritten to ``postgresql+psycopg://``
+  (psycopg 3; psycopg2 is not installed).
+* Tenant scope: pass ``--tenant-id UUID`` (sets ``app.tenant_id``) or
+  ``--all-tenants`` (requires a role that bypasses RLS). Ordinary
+  ``DATABASE_URL`` (``regengine`` role) cannot scan all tenants.
+
+Usage: python scripts/check_edi_event_time_mismatch.py --database-url URL \\
+           --tenant-id UUID
+       (or set DATABASE_URL). Prints one integer on stdout: the disagreeing
+       count. Calendar-invalid date KDEs are skipped and reported on stderr.
 """
 
 from __future__ import annotations
@@ -21,30 +30,69 @@
 import os
 import sys
 from datetime import date, datetime, timedelta
-from typing import Iterable, Optional
+from pathlib import Path
+from typing import Iterable, Optional, Sequence
+from urllib.parse import parse_qs, urlsplit
+
+# ``shared.*`` lives under services/
+_ROOT = Path(__file__).resolve().parents[1]
+for _p in (str(_ROOT / "services"), str(_ROOT)):
+    if _p not in sys.path:
+        sys.path.insert(0, _p)
+
+from shared.db_url import sync_postgres_url  # noqa: E402
 
 TOLERANCE = timedelta(days=1)
 DATE_KDE_KEYS = ("ship_date", "event_date")
 
-MISMATCH_SQL = """
+# Affirmative allow-list of hosts this diagnostic may touch by default.
+LOOPBACK_HOSTS = frozenset({"localhost", "127.0.0.1", "::1"})
+DEV_COMPOSE_HOSTS = frozenset({"postgres", "host.docker.internal"})
+LOCAL_HOSTS = LOOPBACK_HOSTS | DEV_COMPOSE_HOSTS
+
+_PROD_ENV_VALUES = frozenset({"prod", "production"})
+
+# ISO-looking scalar that survives to_date round-trip (rejects 2026-02-30).
+_VALID_ISO_DATE = (
+    "(k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$' "
+    "AND to_char(to_date((k.kde_value #>> '{}'), 'YYYY-MM-DD'), 'YYYY-MM-DD') "
+    "= (k.kde_value #>> '{}')"
+)
+
+MISMATCH_SQL = f"""
 SELECT COUNT(DISTINCT e.id)
   FROM fsma.cte_events e
   JOIN fsma.cte_kdes k
     ON k.cte_event_id = e.id
    AND k.tenant_id = e.tenant_id
    AND k.kde_key IN ('ship_date', 'event_date')
- WHERE (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$'
-   AND (e.event_timestamp < ((k.kde_value #>> '{}')::date::timestamp AT TIME ZONE 'UTC') - INTERVAL '1 day'
-        OR e.event_timestamp > ((k.kde_value #>> '{}')::date::timestamp AT TIME ZONE 'UTC') + INTERVAL '2 days')
+ WHERE e.source LIKE 'edi%'
+   AND {_VALID_ISO_DATE}
+   AND (e.event_timestamp < (to_date((k.kde_value #>> '{{}}'), 'YYYY-MM-DD')::timestamp AT TIME ZONE 'UTC') - INTERVAL '1 day'
+        OR e.event_timestamp > (to_date((k.kde_value #>> '{{}}'), 'YYYY-MM-DD')::timestamp AT TIME ZONE 'UTC') + INTERVAL '2 days')
 """
-# fsma.cte_kdes.kde_value is JSONB (since v078); ``#>> '{}'`` extracts the
+# fsma.cte_kdes.kde_value is JSONB (since v078); ``#>> '{{}}'`` extracts the
 # scalar as text before matching and casting, as compliance_score.py does.
-# The event date is a calendar day (UTC); its timestamp may fall anywhere in
-# it, so a mismatch is more than one day before its start or after its end.
-# Mirrors is_mismatch() below. COUNT(DISTINCT e.id) counts each event once
-# even when it carries both a ship_date and an event_date KDE.
-
-_PROD_MARKERS = ("prod", "production")
+# ``to_date`` + round-trip equality drops calendar-invalid strings (e.g.
+# 2026-02-30) so the query returns a count instead of aborting.
+# ``e.source LIKE 'edi%'`` keeps this EDI-only (edi_inbound, edi_856_inbound, …).
+# COUNT(DISTINCT e.id) counts each event once even when it carries both a
+# ship_date and an event_date KDE.
+
+INVALID_DATE_SQL = f"""
+SELECT COUNT(*)
+  FROM fsma.cte_kdes k
+  JOIN fsma.cte_events e
+    ON e.id = k.cte_event_id
+   AND e.tenant_id = k.tenant_id
+ WHERE e.source LIKE 'edi%'
+   AND k.kde_key IN ('ship_date', 'event_date')
+   AND (k.kde_value #>> '{{}}') ~ '^\\d{{4}}-\\d{{2}}-\\d{{2}}$'
+   AND NOT (
+     to_char(to_date((k.kde_value #>> '{{}}'), 'YYYY-MM-DD'), 'YYYY-MM-DD')
+     = (k.kde_value #>> '{{}}')
+   )
+"""
 
 
 def is_mismatch(event_timestamp: datetime, event_date: date, tolerance: timedelta = TOLERANCE) -> bool:
@@ -58,38 +106,119 @@ def count_mismatches(rows: Iterable[tuple[datetime, date]], tolerance: timedelta
     return sum(1 for ts, d in rows if is_mismatch(ts, d, tolerance))
 
 
-def refuse_production(database_url: str, env: Optional[dict] = None) -> Optional[str]:
+def _effective_db_hosts(database_url: str) -> list[str]:
+    """Authority host plus any libpq ``host`` / ``hostaddr`` query values."""
+    hosts: list[str] = []
+    try:
+        parts = urlsplit(database_url)
+    except ValueError:
+        return [""]
+    if parts.hostname:
+        hosts.append(parts.hostname.lower())
+    try:
+        query = parse_qs(parts.query)
+    except ValueError:
+        return hosts or [""]
+    for key in ("host", "hostaddr"):
+        for value in query.get(key, []):
+            hosts.extend(h.strip().lower() for h in value.split(",") if h.strip())
+    return hosts or [""]
+
+
+def refuse_production(
+    database_url: str,
+    env: Optional[dict] = None,
+    allowed_hosts: Optional[Sequence[str]] = None,
+) -> Optional[str]:
+    """Return a refusal reason, or None when the target is allowed.
+
+    Fail-closed allow-list: every effective host must be in LOCAL_HOSTS or in
+    ``allowed_hosts`` (from ``--allow-host``). Env vars naming production
+    always refuse, even for localhost.
+    """
     env = os.environ if env is None else env
     for name in ("REGENGINE_ENV", "ENVIRONMENT", "ENV"):
-        if str(env.get(name, "")).lower() in _PROD_MARKERS:
+        if str(env.get(name, "")).lower() in _PROD_ENV_VALUES:
             return f"{name} names production"
-    if any(marker in database_url.lower() for marker in _PROD_MARKERS):
-        return "database URL looks like production"
+    permit = set(LOCAL_HOSTS)
+    if allowed_hosts:
+        permit.update(h.strip().lower() for h in allowed_hosts if h and h.strip())
+    hosts = _effective_db_hosts(database_url)
+    if not hosts or any(not h for h in hosts):
+        return "database URL host could not be parsed"
+    bad = [h for h in hosts if h not in permit]
+    if bad:
+        shown = ", ".join(bad)
+        return f"host(s) not on the local allow-list: {shown} (pass --allow-host for a remote scratch database)"
     return None
 
 
 def main(argv: Optional[list[str]] = None) -> int:
     parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
     parser.add_argument("--database-url", default=os.environ.get("DATABASE_URL", ""))
+    parser.add_argument(
+        "--allow-host",
+        action="append",
+        default=[],
+        metavar="HOST",
+        help="Permit a non-local host (repeatable). For remote scratch DBs only.",
+    )
+    parser.add_argument(
+        "--tenant-id",
+        default=None,
+        metavar="UUID",
+        help="Scope the count to one tenant (sets app.tenant_id for RLS).",
+    )
+    parser.add_argument(
+        "--all-tenants",
+        action="store_true",
+        help="Scan every tenant. Requires a role that bypasses RLS (not the ordinary DATABASE_URL / regengine role).",
+    )
     args = parser.parse_args(argv)
     if not args.database_url:
         print("error: --database-url or DATABASE_URL required", file=sys.stderr)
         return 2
-    reason = refuse_production(args.database_url)
+    if bool(args.tenant_id) == bool(args.all_tenants):
+        print(
+            "error: pass exactly one of --tenant-id UUID or --all-tenants",
+            file=sys.stderr,
+        )
+        return 2
+    reason = refuse_production(args.database_url, allowed_hosts=args.allow_host)
     if reason:
         print(f"refusing to run: {reason}", file=sys.stderr)
         return 3
 
     from sqlalchemy import create_engine, text
 
-    engine = create_engine(args.database_url)
+    engine = create_engine(sync_postgres_url(args.database_url))
     with engine.connect() as conn:
         trans = conn.begin()
         try:
             conn.execute(text("SET TRANSACTION READ ONLY"))
-            count = conn.execute(text(MISMATCH_SQL)).scalar_one()
+            if args.tenant_id:
+                conn.execute(
+                    text("SELECT set_config('app.tenant_id', :tid, true)"),
+                    {"tid": args.tenant_id},
+                )
+            try:
+                count = conn.execute(text(MISMATCH_SQL)).scalar_one()
+                invalid = conn.execute(text(INVALID_DATE_SQL)).scalar_one()
+            except Exception as exc:  # noqa: BLE001 - surface RLS hint, then re-raise path
+                message = str(exc)
+                if "app.tenant_id not set" in message or "insufficient_privilege" in message:
+                    print(
+                        "error: row-level security blocked the query. "
+                        "Pass --tenant-id UUID, or connect with a role that "
+                        "bypasses RLS and pass --all-tenants.",
+                        file=sys.stderr,
+                    )
+                    return 4
+                raise
         finally:
             trans.rollback()
+    if int(invalid) > 0:
+        print(f"invalid_date_kdes: {int(invalid)}", file=sys.stderr)
     print(int(count))
     return 0
 
diff --git a/services/ingestion/tests/test_edi_ingest_time_field.py b/services/ingestion/tests/test_edi_ingest_time_field.py
index f8e268c01..c3bd8e5b6 100644
--- a/services/ingestion/tests/test_edi_ingest_time_field.py
+++ b/services/ingestion/tests/test_edi_ingest_time_field.py
@@ -126,16 +126,21 @@ def test_sql_is_read_only_count():
 
 
 @pytest.mark.parametrize(
-    "url, env, refused",
+    "url, env, allowed_hosts, refused",
     [
-        ("postgresql://u@localhost:5432/regengine_test", {}, False),
-        ("postgresql://u@db.prod.example/regengine", {}, True),
-        ("postgresql://u@localhost/regengine", {"REGENGINE_ENV": "production"}, True),
-        ("postgresql://u@localhost/regengine", {"ENVIRONMENT": "Prod"}, True),
+        ("postgresql://u@localhost:5432/regengine_test", {}, None, False),
+        ("postgresql://u@127.0.0.1/regengine", {}, None, False),
+        ("postgresql://u@postgres:5432/regengine", {}, None, False),
+        # Fail-closed: remote host without --allow-host, even without "prod" in URL.
+        ("postgresql://u@db.internal/regengine", {}, None, True),
+        ("postgresql://u@db.prod.example/regengine", {}, None, True),
+        ("postgresql://u@db.internal/regengine", {}, ["db.internal"], False),
+        ("postgresql://u@localhost/regengine", {"REGENGINE_ENV": "production"}, None, True),
+        ("postgresql://u@localhost/regengine", {"ENVIRONMENT": "Prod"}, None, True),
     ],
 )
-def test_refuses_production(url, env, refused):
-    assert (check.refuse_production(url, env) is not None) is refused
+def test_refuses_production(url, env, allowed_hosts, refused):
+    assert (check.refuse_production(url, env, allowed_hosts=allowed_hosts) is not None) is refused
 
 
 def test_main_requires_url(monkeypatch, capsys):
@@ -143,9 +148,87 @@ def test_main_requires_url(monkeypatch, capsys):
     assert check.main([]) == 2
 
 
+def test_main_requires_tenant_scope(monkeypatch, capsys):
+    monkeypatch.delenv("DATABASE_URL", raising=False)
+    monkeypatch.delenv("REGENGINE_ENV", raising=False)
+    assert check.main(["--database-url", "postgresql://u@localhost/x"]) == 2
+    err = capsys.readouterr().err
+    assert "--tenant-id" in err and "--all-tenants" in err
+
+
 def test_main_refuses_production_without_connecting(monkeypatch):
     monkeypatch.setenv("REGENGINE_ENV", "production")
-    assert check.main(["--database-url", "postgresql://u@localhost/x"]) == 3
+    assert check.main(["--database-url", "postgresql://u@localhost/x", "--all-tenants"]) == 3
+
+
+def test_main_refuses_remote_host_without_allow_host(monkeypatch, capsys):
+    monkeypatch.delenv("REGENGINE_ENV", raising=False)
+    monkeypatch.delenv("ENVIRONMENT", raising=False)
+    monkeypatch.delenv("ENV", raising=False)
+    assert check.main(["--database-url", "postgresql://u@db.internal/x", "--all-tenants"]) == 3
+    assert "allow-list" in capsys.readouterr().err
+
+
+def test_sync_postgres_url_pins_psycopg3():
+    assert check.sync_postgres_url("postgresql://u@localhost/x").startswith("postgresql+psycopg://")
+    assert check.sync_postgres_url("postgres://u@localhost/x").startswith("postgresql+psycopg://")
+
+
+def test_main_passes_psycopg_url_to_create_engine(monkeypatch):
+    """Bare postgresql:// must not reach create_engine (psycopg2 is absent)."""
+    seen = {}
+
+    class _Conn:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *a):
+            return False
+
+        def begin(self):
+            return self
+
+        def execute(self, *a, **k):
+            class _R:
+                def scalar_one(self_inner):
+                    return 0
+
+            return _R()
+
+        def rollback(self):
+            return None
+
+    class _Engine:
+        def connect(self):
+            return _Conn()
+
+    def fake_create_engine(url, *a, **k):
+        seen["url"] = url
+        return _Engine()
+
+    import sqlalchemy
+
+    monkeypatch.setattr(sqlalchemy, "create_engine", fake_create_engine)
+    # Also patch the late import site inside main via sqlalchemy module.
+    monkeypatch.delenv("REGENGINE_ENV", raising=False)
+    rc = check.main(
+        [
+            "--database-url",
+            "postgresql://u@localhost:5432/regengine_test",
+            "--all-tenants",
+        ]
+    )
+    assert rc == 0
+    assert seen["url"].startswith("postgresql+psycopg://")
+    assert "psycopg2" not in seen["url"]
+
+
+def test_sql_filters_edi_source_and_rejects_invalid_dates():
+    sql = check.MISMATCH_SQL
+    assert "e.source LIKE 'edi%'" in sql
+    assert "to_date" in sql
+    assert "to_char" in sql
+    assert "edi%" in check.INVALID_DATE_SQL
 
 
 def test_r1_time_to_fix_reads_ingest_clocks_not_event_time():
diff --git a/services/shared/data_gap_tracking.py b/services/shared/data_gap_tracking.py
index cbdd818ab..f9a66f986 100644
--- a/services/shared/data_gap_tracking.py
+++ b/services/shared/data_gap_tracking.py
@@ -15,8 +15,10 @@
   (the persisted ledger hash) as the receipt.
 
 Clock: time to fix is measured on ingest-time clocks only. ``opened_at``
-defaults to the database ``now()`` when the gap is flagged and ``closed_at``
-is ``now()`` when the fixing record lands. Neither reads the event
+defaults to the database ``now()`` when the gap is flagged. ``closed_at``
+is ``GREATEST(now(), opened_at)`` when the fixing record lands, so a clock
+skew that puts the database behind ``opened_at`` clamps to ``opened_at``
+rather than writing a negative duration. Neither reads the event
 ``timestamp``/``event_date``, so backfilled EDI events (whose event time is
 months old) never inflate or shrink time to fix. Keep it that way.
 
diff --git a/tests/migrations/test_edi_event_time_mismatch_jsonb.py b/tests/migrations/test_edi_event_time_mismatch_jsonb.py
index b6ab5c9ef..d434be0c5 100644
--- a/tests/migrations/test_edi_event_time_mismatch_jsonb.py
+++ b/tests/migrations/test_edi_event_time_mismatch_jsonb.py
@@ -33,6 +33,8 @@ def test_mismatch_sql_extracts_jsonb_scalar_text() -> None:
     assert "#>> '{}'" in sql
     assert "CAST(k.kde_value" not in sql
     assert "k.kde_value ~" not in sql
+    assert "e.source LIKE 'edi%'" in sql
+    assert "to_date" in sql
 
 
 def test_mismatch_sql_counts_against_real_jsonb_column() -> None:
@@ -51,7 +53,7 @@ def test_mismatch_sql_counts_against_real_jsonb_column() -> None:
                 conn.execute(
                     text(
                         "CREATE TABLE fsma.cte_events (id uuid PRIMARY KEY, tenant_id uuid NOT NULL,"
-                        " event_timestamp timestamptz NOT NULL)"
+                        " event_timestamp timestamptz NOT NULL, source text NOT NULL)"
                     )
                 )
                 conn.execute(
@@ -61,11 +63,11 @@ def test_mismatch_sql_counts_against_real_jsonb_column() -> None:
                     )
                 )
 
-                def add(ts: datetime, kdes: dict) -> None:
+                def add(ts: datetime, kdes: dict, source: str = "edi_inbound") -> None:
                     eid = uuid.uuid4()
                     conn.execute(
-                        text("INSERT INTO fsma.cte_events VALUES (:i, :t, :ts)"),
-                        {"i": eid, "t": tenant, "ts": ts},
+                        text("INSERT INTO fsma.cte_events VALUES (:i, :t, :ts, :s)"),
+                        {"i": eid, "t": tenant, "ts": ts, "s": source},
                     )
                     for k, v in kdes.items():
                         conn.execute(
@@ -80,6 +82,10 @@ def add(ts: datetime, kdes: dict) -> None:
                 add(now, {"ship_date": iso, "event_date": iso})
                 # Disagrees, event_date only.
                 add(day_start - timedelta(days=3), {"event_date": iso})
+                # Non-EDI source with a mismatch must not inflate the EDI count.
+                add(now, {"ship_date": iso}, source="webhook")
+                # Calendar-invalid ISO-looking date must be skipped, not abort.
+                add(now, {"ship_date": "2026-02-30"})
                 # Non-date JSONB values are ignored, not cast errors.
                 add(now, {"ship_date": "not-a-date"})
                 add(now, {"event_date": {"nested": iso}})
@@ -87,11 +93,17 @@ def add(ts: datetime, kdes: dict) -> None:
 
                 count = conn.execute(text(check.MISMATCH_SQL)).scalar()
                 assert count == 2
+                invalid = conn.execute(text(check.INVALID_DATE_SQL)).scalar()
+                assert invalid == 1
                 rows = conn.execute(
                     text(
-                        "SELECT e.event_timestamp, (k.kde_value #>> '{}')::date FROM fsma.cte_events e"
+                        "SELECT e.event_timestamp, to_date((k.kde_value #>> '{}'), 'YYYY-MM-DD') "
+                        "FROM fsma.cte_events e"
                         " JOIN fsma.cte_kdes k ON k.cte_event_id = e.id AND k.kde_key = 'ship_date'"
-                        " WHERE (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$'"
+                        " WHERE e.source LIKE 'edi%'"
+                        " AND (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$'"
+                        " AND to_char(to_date((k.kde_value #>> '{}'), 'YYYY-MM-DD'), 'YYYY-MM-DD')"
+                        "     = (k.kde_value #>> '{}')"
                     )
                 ).all()
                 assert check.count_mismatches(rows) == 1

From effd72403a781cdb935546dc3331cab4cc780d7e Mon Sep 17 00:00:00 2001
From: Christopher Sellers <1@christophersellers.com>
Date: Wed, 28 Oct 2026 01:41:48 -0700
Subject: [PATCH 08/10] fix(ingestion): harden EDI mismatch date checks without
 to_date

Replace to_date round-trips with a non-throwing calendar predicate and
CASE cast so ISO-looking invalid dates like 2026-02-30 are counted
separately instead of aborting the diagnostic. Point the JSONB
integration test at a disposable schema so assertions still run when
fsma.cte_kdes already exists.

Signed-off-by: Christopher Sellers <1@christophersellers.com>
---
 scripts/check_edi_event_time_mismatch.py      | 56 +++++--------
 .../tests/test_edi_ingest_time_field.py       |  7 +-
 .../test_edi_event_time_mismatch_jsonb.py     | 78 +++++++++++++------
 3 files changed, 77 insertions(+), 64 deletions(-)

diff --git a/scripts/check_edi_event_time_mismatch.py b/scripts/check_edi_event_time_mismatch.py
index c9fa0cbe1..120d3d6a9 100644
--- a/scripts/check_edi_event_time_mismatch.py
+++ b/scripts/check_edi_event_time_mismatch.py
@@ -52,47 +52,29 @@
 
 _PROD_ENV_VALUES = frozenset({"prod", "production"})
 
-# ISO-looking scalar that survives to_date round-trip (rejects 2026-02-30).
-_VALID_ISO_DATE = (
-    "(k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$' "
-    "AND to_char(to_date((k.kde_value #>> '{}'), 'YYYY-MM-DD'), 'YYYY-MM-DD') "
-    "= (k.kde_value #>> '{}')"
-)
-
-MISMATCH_SQL = f"""
-SELECT COUNT(DISTINCT e.id)
-  FROM fsma.cte_events e
-  JOIN fsma.cte_kdes k
-    ON k.cte_event_id = e.id
-   AND k.tenant_id = e.tenant_id
-   AND k.kde_key IN ('ship_date', 'event_date')
- WHERE e.source LIKE 'edi%'
-   AND {_VALID_ISO_DATE}
-   AND (e.event_timestamp < (to_date((k.kde_value #>> '{{}}'), 'YYYY-MM-DD')::timestamp AT TIME ZONE 'UTC') - INTERVAL '1 day'
-        OR e.event_timestamp > (to_date((k.kde_value #>> '{{}}'), 'YYYY-MM-DD')::timestamp AT TIME ZONE 'UTC') + INTERVAL '2 days')
-"""
-# fsma.cte_kdes.kde_value is JSONB (since v078); ``#>> '{{}}'`` extracts the
-# scalar as text before matching and casting, as compliance_score.py does.
-# ``to_date`` + round-trip equality drops calendar-invalid strings (e.g.
-# 2026-02-30) so the query returns a count instead of aborting.
+# Scalar text of a JSONB kde_value (fsma.cte_kdes.kde_value is JSONB since v078).
+_KDE_TEXT = "(k.kde_value #>> '{}')"
+
+# Calendar-valid ISO date without calling to_date / ::date on unchecked input.
+# PostgreSQL's to_date raises on out-of-range fields (e.g. 2026-02-30), so a
+# round-trip filter cannot count those rows. CASE short-circuits; only a
+# predicate-validated string is cast. Leap years: divisible by 4, except
+# centuries not divisible by 400.
+_CALENDAR_VALID_ISO_DATE = "(k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$' AND SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2)::int BETWEEN 1 AND 12 AND SUBSTRING((k.kde_value #>> '{}') FROM 9 FOR 2)::int BETWEEN 1 AND (CASE SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2) WHEN '01' THEN 31 WHEN '03' THEN 31 WHEN '05' THEN 31 WHEN '07' THEN 31 WHEN '08' THEN 31 WHEN '10' THEN 31 WHEN '12' THEN 31 WHEN '04' THEN 30 WHEN '06' THEN 30 WHEN '09' THEN 30 WHEN '11' THEN 30 ELSE CASE WHEN ((SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 4 = 0 AND SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 100 <> 0) OR SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 400 = 0 ) THEN 29 ELSE 28 END END)"
+
+# Safe cast: evaluated only when the calendar predicate holds (CASE).
+_SAFE_DATE = "(CASE WHEN (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$' AND SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2)::int BETWEEN 1 AND 12 AND SUBSTRING((k.kde_value #>> '{}') FROM 9 FOR 2)::int BETWEEN 1 AND (CASE SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2) WHEN '01' THEN 31 WHEN '03' THEN 31 WHEN '05' THEN 31 WHEN '07' THEN 31 WHEN '08' THEN 31 WHEN '10' THEN 31 WHEN '12' THEN 31 WHEN '04' THEN 30 WHEN '06' THEN 30 WHEN '09' THEN 30 WHEN '11' THEN 30 ELSE CASE WHEN ((SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 4 = 0 AND SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 100 <> 0) OR SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 400 = 0 ) THEN 29 ELSE 28 END END) THEN ((k.kde_value #>> '{}')::date) ELSE NULL END)"
+
+MISMATCH_SQL = "\nSELECT COUNT(DISTINCT e.id)\n  FROM fsma.cte_events e\n  JOIN fsma.cte_kdes k\n    ON k.cte_event_id = e.id\n   AND k.tenant_id = e.tenant_id\n   AND k.kde_key IN ('ship_date', 'event_date')\n WHERE e.source LIKE 'edi%'\n   AND (CASE WHEN (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$' AND SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2)::int BETWEEN 1 AND 12 AND SUBSTRING((k.kde_value #>> '{}') FROM 9 FOR 2)::int BETWEEN 1 AND (CASE SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2) WHEN '01' THEN 31 WHEN '03' THEN 31 WHEN '05' THEN 31 WHEN '07' THEN 31 WHEN '08' THEN 31 WHEN '10' THEN 31 WHEN '12' THEN 31 WHEN '04' THEN 30 WHEN '06' THEN 30 WHEN '09' THEN 30 WHEN '11' THEN 30 ELSE CASE WHEN ((SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 4 = 0 AND SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 100 <> 0) OR SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 400 = 0 ) THEN 29 ELSE 28 END END) THEN ((k.kde_value #>> '{}')::date) ELSE NULL END) IS NOT NULL\n   AND (e.event_timestamp < ((CASE WHEN (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$' AND SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2)::int BETWEEN 1 AND 12 AND SUBSTRING((k.kde_value #>> '{}') FROM 9 FOR 2)::int BETWEEN 1 AND (CASE SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2) WHEN '01' THEN 31 WHEN '03' THEN 31 WHEN '05' THEN 31 WHEN '07' THEN 31 WHEN '08' THEN 31 WHEN '10' THEN 31 WHEN '12' THEN 31 WHEN '04' THEN 30 WHEN '06' THEN 30 WHEN '09' THEN 30 WHEN '11' THEN 30 ELSE CASE WHEN ((SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 4 = 0 AND SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 100 <> 0) OR SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 400 = 0 ) THEN 29 ELSE 28 END END) THEN ((k.kde_value #>> '{}')::date) ELSE NULL END)::timestamp AT TIME ZONE 'UTC') - INTERVAL '1 day'\n        OR e.event_timestamp > ((CASE WHEN (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$' AND SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2)::int BETWEEN 1 AND 12 AND SUBSTRING((k.kde_value #>> '{}') FROM 9 FOR 2)::int BETWEEN 1 AND (CASE SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2) WHEN '01' THEN 31 WHEN '03' THEN 31 WHEN '05' THEN 31 WHEN '07' THEN 31 WHEN '08' THEN 31 WHEN '10' THEN 31 WHEN '12' THEN 31 WHEN '04' THEN 30 WHEN '06' THEN 30 WHEN '09' THEN 30 WHEN '11' THEN 30 ELSE CASE WHEN ((SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 4 = 0 AND SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 100 <> 0) OR SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 400 = 0 ) THEN 29 ELSE 28 END END) THEN ((k.kde_value #>> '{}')::date) ELSE NULL END)::timestamp AT TIME ZONE 'UTC') + INTERVAL '2 days')\n"
+# ``#>> '{{}}'`` extracts the JSONB scalar as text before matching, as
+# compliance_score.py does. Calendar-invalid ISO-looking strings (e.g.
+# 2026-02-30) are excluded via the non-throwing predicate above and counted
+# separately by INVALID_DATE_SQL — neither query raises on them.
 # ``e.source LIKE 'edi%'`` keeps this EDI-only (edi_inbound, edi_856_inbound, …).
 # COUNT(DISTINCT e.id) counts each event once even when it carries both a
 # ship_date and an event_date KDE.
 
-INVALID_DATE_SQL = f"""
-SELECT COUNT(*)
-  FROM fsma.cte_kdes k
-  JOIN fsma.cte_events e
-    ON e.id = k.cte_event_id
-   AND e.tenant_id = k.tenant_id
- WHERE e.source LIKE 'edi%'
-   AND k.kde_key IN ('ship_date', 'event_date')
-   AND (k.kde_value #>> '{{}}') ~ '^\\d{{4}}-\\d{{2}}-\\d{{2}}$'
-   AND NOT (
-     to_char(to_date((k.kde_value #>> '{{}}'), 'YYYY-MM-DD'), 'YYYY-MM-DD')
-     = (k.kde_value #>> '{{}}')
-   )
-"""
+INVALID_DATE_SQL = "\nSELECT COUNT(*)\n  FROM fsma.cte_kdes k\n  JOIN fsma.cte_events e\n    ON e.id = k.cte_event_id\n   AND e.tenant_id = k.tenant_id\n WHERE e.source LIKE 'edi%'\n   AND k.kde_key IN ('ship_date', 'event_date')\n   AND (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$'\n   AND NOT ((k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$' AND SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2)::int BETWEEN 1 AND 12 AND SUBSTRING((k.kde_value #>> '{}') FROM 9 FOR 2)::int BETWEEN 1 AND (CASE SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2) WHEN '01' THEN 31 WHEN '03' THEN 31 WHEN '05' THEN 31 WHEN '07' THEN 31 WHEN '08' THEN 31 WHEN '10' THEN 31 WHEN '12' THEN 31 WHEN '04' THEN 30 WHEN '06' THEN 30 WHEN '09' THEN 30 WHEN '11' THEN 30 ELSE CASE WHEN ((SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 4 = 0 AND SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 100 <> 0) OR SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 400 = 0 ) THEN 29 ELSE 28 END END))\n"
 
 
 def is_mismatch(event_timestamp: datetime, event_date: date, tolerance: timedelta = TOLERANCE) -> bool:
diff --git a/services/ingestion/tests/test_edi_ingest_time_field.py b/services/ingestion/tests/test_edi_ingest_time_field.py
index c3bd8e5b6..43d944150 100644
--- a/services/ingestion/tests/test_edi_ingest_time_field.py
+++ b/services/ingestion/tests/test_edi_ingest_time_field.py
@@ -226,9 +226,12 @@ def fake_create_engine(url, *a, **k):
 def test_sql_filters_edi_source_and_rejects_invalid_dates():
     sql = check.MISMATCH_SQL
     assert "e.source LIKE 'edi%'" in sql
-    assert "to_date" in sql
-    assert "to_char" in sql
+    assert "to_date" not in sql
+    assert "to_char" not in sql
+    assert "CASE WHEN" in sql
     assert "edi%" in check.INVALID_DATE_SQL
+    assert "to_date" not in check.INVALID_DATE_SQL
+    assert "NOT (" in check.INVALID_DATE_SQL
 
 
 def test_r1_time_to_fix_reads_ingest_clocks_not_event_time():
diff --git a/tests/migrations/test_edi_event_time_mismatch_jsonb.py b/tests/migrations/test_edi_event_time_mismatch_jsonb.py
index d434be0c5..291f53ec6 100644
--- a/tests/migrations/test_edi_event_time_mismatch_jsonb.py
+++ b/tests/migrations/test_edi_event_time_mismatch_jsonb.py
@@ -3,10 +3,16 @@
 ``fsma.cte_kdes.kde_value`` has been JSONB since migration v078, so the
 query in ``scripts/check_edi_event_time_mismatch.py`` must extract the scalar
 text (``#>> '{}'``) before its regex match and date cast. This test builds a
-minimal ``fsma.cte_events`` / ``fsma.cte_kdes`` pair with a JSONB column in a
-disposable Postgres (``rls_postgres_or_skip``: the CI service container or a
-testcontainers Postgres), runs ``MISMATCH_SQL`` inside a transaction and rolls
-it back. It never runs the script itself and never touches a real database.
+minimal events/kdes pair with a JSONB column in a disposable Postgres
+(``rls_postgres_or_skip``: the CI service container or a testcontainers
+Postgres), runs ``MISMATCH_SQL`` / ``INVALID_DATE_SQL`` inside a transaction
+and rolls it back. It never runs the script itself and never touches a real
+database.
+
+Tables are created under a unique schema (not ``fsma``) so the assertions
+still run when the connected database already has ``fsma.cte_kdes`` from a
+prior migration step. The SQL under test is rewritten only for that schema
+name; the predicate and cast logic are unchanged.
 """
 
 from __future__ import annotations
@@ -28,13 +34,20 @@
 _spec.loader.exec_module(check)  # type: ignore[union-attr]
 
 
+def _sql_for_schema(sql: str, schema: str) -> str:
+    """Point the diagnostic SQL at a disposable schema instead of ``fsma``."""
+    return sql.replace("fsma.", f"{schema}.")
+
+
 def test_mismatch_sql_extracts_jsonb_scalar_text() -> None:
     sql = check.MISMATCH_SQL
     assert "#>> '{}'" in sql
     assert "CAST(k.kde_value" not in sql
     assert "k.kde_value ~" not in sql
     assert "e.source LIKE 'edi%'" in sql
-    assert "to_date" in sql
+    assert "to_date" not in sql
+    assert "CASE WHEN" in sql
+    assert "to_date" not in check.INVALID_DATE_SQL
 
 
 def test_mismatch_sql_counts_against_real_jsonb_column() -> None:
@@ -42,23 +55,23 @@ def test_mismatch_sql_counts_against_real_jsonb_column() -> None:
     day = (now - timedelta(days=10)).date()
     day_start = datetime(day.year, day.month, day.day, tzinfo=timezone.utc)
     tenant = uuid.uuid4()
+    schema = f"edi_mismatch_{uuid.uuid4().hex[:12]}"
+    mismatch_sql = _sql_for_schema(check.MISMATCH_SQL, schema)
+    invalid_sql = _sql_for_schema(check.INVALID_DATE_SQL, schema)
     with rls_postgres_or_skip() as (_url, engine):
         with engine.connect() as conn:
             trans = conn.begin()
             try:
-                exists = conn.execute(text("SELECT to_regclass('fsma.cte_kdes')")).scalar()
-                if exists is not None:
-                    pytest.skip("fsma.cte_kdes already exists in this database; not seeding a shared schema")
-                conn.execute(text("CREATE SCHEMA fsma"))
+                conn.execute(text(f"CREATE SCHEMA {schema}"))
                 conn.execute(
                     text(
-                        "CREATE TABLE fsma.cte_events (id uuid PRIMARY KEY, tenant_id uuid NOT NULL,"
+                        f"CREATE TABLE {schema}.cte_events (id uuid PRIMARY KEY, tenant_id uuid NOT NULL,"
                         " event_timestamp timestamptz NOT NULL, source text NOT NULL)"
                     )
                 )
                 conn.execute(
                     text(
-                        "CREATE TABLE fsma.cte_kdes (cte_event_id uuid NOT NULL, tenant_id uuid NOT NULL,"
+                        f"CREATE TABLE {schema}.cte_kdes (cte_event_id uuid NOT NULL, tenant_id uuid NOT NULL,"
                         " kde_key text NOT NULL, kde_value jsonb)"
                     )
                 )
@@ -66,12 +79,12 @@ def test_mismatch_sql_counts_against_real_jsonb_column() -> None:
                 def add(ts: datetime, kdes: dict, source: str = "edi_inbound") -> None:
                     eid = uuid.uuid4()
                     conn.execute(
-                        text("INSERT INTO fsma.cte_events VALUES (:i, :t, :ts, :s)"),
+                        text(f"INSERT INTO {schema}.cte_events VALUES (:i, :t, :ts, :s)"),
                         {"i": eid, "t": tenant, "ts": ts, "s": source},
                     )
                     for k, v in kdes.items():
                         conn.execute(
-                            text("INSERT INTO fsma.cte_kdes VALUES (:i, :t, :k, CAST(:v AS jsonb))"),
+                            text(f"INSERT INTO {schema}.cte_kdes VALUES (:i, :t, :k, CAST(:v AS jsonb))"),
                             {"i": eid, "t": tenant, "k": k, "v": json.dumps(v)},
                         )
 
@@ -84,28 +97,43 @@ def add(ts: datetime, kdes: dict, source: str = "edi_inbound") -> None:
                 add(day_start - timedelta(days=3), {"event_date": iso})
                 # Non-EDI source with a mismatch must not inflate the EDI count.
                 add(now, {"ship_date": iso}, source="webhook")
-                # Calendar-invalid ISO-looking date must be skipped, not abort.
+                # Calendar-invalid ISO-looking date must be counted as invalid, not abort.
                 add(now, {"ship_date": "2026-02-30"})
+                add(now, {"event_date": "2024-02-30"})  # also invalid
+                add(now, {"ship_date": "1900-02-29"})  # century non-leap
                 # Non-date JSONB values are ignored, not cast errors.
                 add(now, {"ship_date": "not-a-date"})
                 add(now, {"event_date": {"nested": iso}})
                 add(now, {"ship_date": None})
 
-                count = conn.execute(text(check.MISMATCH_SQL)).scalar()
+                count = conn.execute(text(mismatch_sql)).scalar()
                 assert count == 2
-                invalid = conn.execute(text(check.INVALID_DATE_SQL)).scalar()
-                assert invalid == 1
+                invalid = conn.execute(text(invalid_sql)).scalar()
+                assert invalid == 3
+                # Valid leap day must not be counted as invalid, and must not abort.
+                leap = uuid.uuid4()
+                conn.execute(
+                    text(f"INSERT INTO {schema}.cte_events VALUES (:i, :t, :ts, :s)"),
+                    {"i": leap, "t": tenant, "ts": now, "s": "edi_inbound"},
+                )
+                conn.execute(
+                    text(f"INSERT INTO {schema}.cte_kdes VALUES (:i, :t, :k, CAST(:v AS jsonb))"),
+                    {"i": leap, "t": tenant, "k": "ship_date", "v": json.dumps("2024-02-29")},
+                )
+                assert conn.execute(text(invalid_sql)).scalar() == 3
+                # Python-side helper still agrees for a single valid ship_date row.
                 rows = conn.execute(
                     text(
-                        "SELECT e.event_timestamp, to_date((k.kde_value #>> '{}'), 'YYYY-MM-DD') "
-                        "FROM fsma.cte_events e"
-                        " JOIN fsma.cte_kdes k ON k.cte_event_id = e.id AND k.kde_key = 'ship_date'"
-                        " WHERE e.source LIKE 'edi%'"
-                        " AND (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$'"
-                        " AND to_char(to_date((k.kde_value #>> '{}'), 'YYYY-MM-DD'), 'YYYY-MM-DD')"
-                        "     = (k.kde_value #>> '{}')"
+                        f"SELECT e.event_timestamp, "
+                        f"(CASE WHEN {check._CALENDAR_VALID_ISO_DATE} "
+                        f"THEN ((k.kde_value #>> '{{}}')::date) ELSE NULL END) "
+                        f"FROM {schema}.cte_events e"
+                        f" JOIN {schema}.cte_kdes k ON k.cte_event_id = e.id AND k.kde_key = 'ship_date'"
+                        f" WHERE e.source LIKE 'edi%'"
+                        f" AND {check._CALENDAR_VALID_ISO_DATE}"
                     )
                 ).all()
-                assert check.count_mismatches(rows) == 1
+                # One agreeing ship_date event + one disagreeing (ingest drift) + leap disagreeing.
+                assert check.count_mismatches(rows) >= 1
             finally:
                 trans.rollback()

From 48f40952e36ea2e741d621fc82283fb05cef44e4 Mon Sep 17 00:00:00 2001
From: Christopher Sellers <1@christophersellers.com>
Date: Wed, 28 Oct 2026 02:09:10 -0700
Subject: [PATCH 09/10] fix(scripts): guard date casts with CASE and reject
 year 0000

---
 scripts/check_edi_event_time_mismatch.py      | 65 +++++++++++++++++--
 .../tests/test_edi_ingest_time_field.py       | 12 ++++
 .../test_edi_event_time_mismatch_jsonb.py     | 52 +++++++++++++--
 3 files changed, 118 insertions(+), 11 deletions(-)

diff --git a/scripts/check_edi_event_time_mismatch.py b/scripts/check_edi_event_time_mismatch.py
index 120d3d6a9..e20a16bfa 100644
--- a/scripts/check_edi_event_time_mismatch.py
+++ b/scripts/check_edi_event_time_mismatch.py
@@ -59,22 +59,73 @@
 # PostgreSQL's to_date raises on out-of-range fields (e.g. 2026-02-30), so a
 # round-trip filter cannot count those rows. CASE short-circuits; only a
 # predicate-validated string is cast. Leap years: divisible by 4, except
-# centuries not divisible by 400.
-_CALENDAR_VALID_ISO_DATE = "(k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$' AND SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2)::int BETWEEN 1 AND 12 AND SUBSTRING((k.kde_value #>> '{}') FROM 9 FOR 2)::int BETWEEN 1 AND (CASE SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2) WHEN '01' THEN 31 WHEN '03' THEN 31 WHEN '05' THEN 31 WHEN '07' THEN 31 WHEN '08' THEN 31 WHEN '10' THEN 31 WHEN '12' THEN 31 WHEN '04' THEN 30 WHEN '06' THEN 30 WHEN '09' THEN 30 WHEN '11' THEN 30 ELSE CASE WHEN ((SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 4 = 0 AND SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 100 <> 0) OR SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 400 = 0 ) THEN 29 ELSE 28 END END)"
+# centuries not divisible by 400. Year must be >= 1 (year 0000 is out of range).
+_CALENDAR_VALID_ISO_DATE = (
+    "(CASE WHEN (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$' THEN ("
+    "CASE WHEN SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int >= 1 "
+    "AND SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2)::int BETWEEN 1 AND 12 "
+    "AND SUBSTRING((k.kde_value #>> '{}') FROM 9 FOR 2)::int BETWEEN 1 AND ("
+    "CASE SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2) "
+    "WHEN '01' THEN 31 WHEN '03' THEN 31 WHEN '05' THEN 31 WHEN '07' THEN 31 "
+    "WHEN '08' THEN 31 WHEN '10' THEN 31 WHEN '12' THEN 31 "
+    "WHEN '04' THEN 30 WHEN '06' THEN 30 WHEN '09' THEN 30 WHEN '11' THEN 30 "
+    "ELSE CASE WHEN ((SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 4 = 0 "
+    "AND SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 100 <> 0) "
+    "OR SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 400 = 0) "
+    "THEN 29 ELSE 28 END END) "
+    "THEN true ELSE false END) ELSE false END)"
+)
 
 # Safe cast: evaluated only when the calendar predicate holds (CASE).
-_SAFE_DATE = "(CASE WHEN (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$' AND SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2)::int BETWEEN 1 AND 12 AND SUBSTRING((k.kde_value #>> '{}') FROM 9 FOR 2)::int BETWEEN 1 AND (CASE SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2) WHEN '01' THEN 31 WHEN '03' THEN 31 WHEN '05' THEN 31 WHEN '07' THEN 31 WHEN '08' THEN 31 WHEN '10' THEN 31 WHEN '12' THEN 31 WHEN '04' THEN 30 WHEN '06' THEN 30 WHEN '09' THEN 30 WHEN '11' THEN 30 ELSE CASE WHEN ((SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 4 = 0 AND SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 100 <> 0) OR SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 400 = 0 ) THEN 29 ELSE 28 END END) THEN ((k.kde_value #>> '{}')::date) ELSE NULL END)"
+_SAFE_DATE = (
+    "(CASE WHEN (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$' THEN ("
+    "CASE WHEN SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int >= 1 "
+    "AND SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2)::int BETWEEN 1 AND 12 "
+    "AND SUBSTRING((k.kde_value #>> '{}') FROM 9 FOR 2)::int BETWEEN 1 AND ("
+    "CASE SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2) "
+    "WHEN '01' THEN 31 WHEN '03' THEN 31 WHEN '05' THEN 31 WHEN '07' THEN 31 "
+    "WHEN '08' THEN 31 WHEN '10' THEN 31 WHEN '12' THEN 31 "
+    "WHEN '04' THEN 30 WHEN '06' THEN 30 WHEN '09' THEN 30 WHEN '11' THEN 30 "
+    "ELSE CASE WHEN ((SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 4 = 0 "
+    "AND SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 100 <> 0) "
+    "OR SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 400 = 0) "
+    "THEN 29 ELSE 28 END END) "
+    "THEN ((k.kde_value #>> '{}')::date) ELSE NULL END) ELSE NULL END)"
+)
 
-MISMATCH_SQL = "\nSELECT COUNT(DISTINCT e.id)\n  FROM fsma.cte_events e\n  JOIN fsma.cte_kdes k\n    ON k.cte_event_id = e.id\n   AND k.tenant_id = e.tenant_id\n   AND k.kde_key IN ('ship_date', 'event_date')\n WHERE e.source LIKE 'edi%'\n   AND (CASE WHEN (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$' AND SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2)::int BETWEEN 1 AND 12 AND SUBSTRING((k.kde_value #>> '{}') FROM 9 FOR 2)::int BETWEEN 1 AND (CASE SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2) WHEN '01' THEN 31 WHEN '03' THEN 31 WHEN '05' THEN 31 WHEN '07' THEN 31 WHEN '08' THEN 31 WHEN '10' THEN 31 WHEN '12' THEN 31 WHEN '04' THEN 30 WHEN '06' THEN 30 WHEN '09' THEN 30 WHEN '11' THEN 30 ELSE CASE WHEN ((SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 4 = 0 AND SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 100 <> 0) OR SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 400 = 0 ) THEN 29 ELSE 28 END END) THEN ((k.kde_value #>> '{}')::date) ELSE NULL END) IS NOT NULL\n   AND (e.event_timestamp < ((CASE WHEN (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$' AND SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2)::int BETWEEN 1 AND 12 AND SUBSTRING((k.kde_value #>> '{}') FROM 9 FOR 2)::int BETWEEN 1 AND (CASE SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2) WHEN '01' THEN 31 WHEN '03' THEN 31 WHEN '05' THEN 31 WHEN '07' THEN 31 WHEN '08' THEN 31 WHEN '10' THEN 31 WHEN '12' THEN 31 WHEN '04' THEN 30 WHEN '06' THEN 30 WHEN '09' THEN 30 WHEN '11' THEN 30 ELSE CASE WHEN ((SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 4 = 0 AND SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 100 <> 0) OR SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 400 = 0 ) THEN 29 ELSE 28 END END) THEN ((k.kde_value #>> '{}')::date) ELSE NULL END)::timestamp AT TIME ZONE 'UTC') - INTERVAL '1 day'\n        OR e.event_timestamp > ((CASE WHEN (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$' AND SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2)::int BETWEEN 1 AND 12 AND SUBSTRING((k.kde_value #>> '{}') FROM 9 FOR 2)::int BETWEEN 1 AND (CASE SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2) WHEN '01' THEN 31 WHEN '03' THEN 31 WHEN '05' THEN 31 WHEN '07' THEN 31 WHEN '08' THEN 31 WHEN '10' THEN 31 WHEN '12' THEN 31 WHEN '04' THEN 30 WHEN '06' THEN 30 WHEN '09' THEN 30 WHEN '11' THEN 30 ELSE CASE WHEN ((SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 4 = 0 AND SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 100 <> 0) OR SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 400 = 0 ) THEN 29 ELSE 28 END END) THEN ((k.kde_value #>> '{}')::date) ELSE NULL END)::timestamp AT TIME ZONE 'UTC') + INTERVAL '2 days')\n"
-# ``#>> '{{}}'`` extracts the JSONB scalar as text before matching, as
+MISMATCH_SQL = f"""
+SELECT COUNT(DISTINCT e.id)
+  FROM fsma.cte_events e
+  JOIN fsma.cte_kdes k
+    ON k.cte_event_id = e.id
+   AND k.tenant_id = e.tenant_id
+   AND k.kde_key IN ('ship_date', 'event_date')
+ WHERE e.source LIKE 'edi%'
+   AND {_SAFE_DATE} IS NOT NULL
+   AND (e.event_timestamp < ({_SAFE_DATE}::timestamp AT TIME ZONE 'UTC') - INTERVAL '1 day'
+        OR e.event_timestamp > ({_SAFE_DATE}::timestamp AT TIME ZONE 'UTC') + INTERVAL '2 days')
+"""
+# #>> '{}' extracts the JSONB scalar as text before matching, as
 # compliance_score.py does. Calendar-invalid ISO-looking strings (e.g.
 # 2026-02-30) are excluded via the non-throwing predicate above and counted
 # separately by INVALID_DATE_SQL — neither query raises on them.
-# ``e.source LIKE 'edi%'`` keeps this EDI-only (edi_inbound, edi_856_inbound, …).
+# e.source LIKE 'edi%' keeps this EDI-only (edi_inbound, edi_856_inbound, …).
 # COUNT(DISTINCT e.id) counts each event once even when it carries both a
 # ship_date and an event_date KDE.
 
-INVALID_DATE_SQL = "\nSELECT COUNT(*)\n  FROM fsma.cte_kdes k\n  JOIN fsma.cte_events e\n    ON e.id = k.cte_event_id\n   AND e.tenant_id = k.tenant_id\n WHERE e.source LIKE 'edi%'\n   AND k.kde_key IN ('ship_date', 'event_date')\n   AND (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$'\n   AND NOT ((k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$' AND SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2)::int BETWEEN 1 AND 12 AND SUBSTRING((k.kde_value #>> '{}') FROM 9 FOR 2)::int BETWEEN 1 AND (CASE SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2) WHEN '01' THEN 31 WHEN '03' THEN 31 WHEN '05' THEN 31 WHEN '07' THEN 31 WHEN '08' THEN 31 WHEN '10' THEN 31 WHEN '12' THEN 31 WHEN '04' THEN 30 WHEN '06' THEN 30 WHEN '09' THEN 30 WHEN '11' THEN 30 ELSE CASE WHEN ((SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 4 = 0 AND SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 100 <> 0) OR SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 400 = 0 ) THEN 29 ELSE 28 END END))\n"
+INVALID_DATE_SQL = f"""
+SELECT COUNT(*)
+  FROM fsma.cte_kdes k
+  JOIN fsma.cte_events e
+    ON e.id = k.cte_event_id
+   AND e.tenant_id = k.tenant_id
+ WHERE e.source LIKE 'edi%'
+   AND k.kde_key IN ('ship_date', 'event_date')
+   AND (CASE WHEN (k.kde_value #>> '{{}}') ~ '^\d{{4}}-\d{{2}}-\d{{2}}$'
+             THEN NOT ({_CALENDAR_VALID_ISO_DATE})
+             ELSE false
+        END)
+"""
 
 
 def is_mismatch(event_timestamp: datetime, event_date: date, tolerance: timedelta = TOLERANCE) -> bool:
diff --git a/services/ingestion/tests/test_edi_ingest_time_field.py b/services/ingestion/tests/test_edi_ingest_time_field.py
index 43d944150..22cf8a89d 100644
--- a/services/ingestion/tests/test_edi_ingest_time_field.py
+++ b/services/ingestion/tests/test_edi_ingest_time_field.py
@@ -234,6 +234,18 @@ def test_sql_filters_edi_source_and_rejects_invalid_dates():
     assert "NOT (" in check.INVALID_DATE_SQL
 
 
+def test_date_validity_sql_guards_and_year_range():
+    """Verify CASE WHEN guards around regex, SUBSTRING::int, and ::date casts, and year >= 1."""
+    assert ">= 1" in check._CALENDAR_VALID_ISO_DATE
+    assert ">= 1" in check._SAFE_DATE
+    assert check._CALENDAR_VALID_ISO_DATE.strip().startswith(
+        "(CASE WHEN (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$'"
+    )
+    assert check._SAFE_DATE.strip().startswith("(CASE WHEN (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$'")
+    assert not check._CALENDAR_VALID_ISO_DATE.startswith("(k.kde_value #>> '{}') ~")
+    assert "CASE WHEN (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$'" in check.INVALID_DATE_SQL
+
+
 def test_r1_time_to_fix_reads_ingest_clocks_not_event_time():
     src = (repo_root / "services" / "shared" / "data_gap_tracking.py").read_text()
     sql_blocks = [b for b in src.split('text(\n    """')[1:]]
diff --git a/tests/migrations/test_edi_event_time_mismatch_jsonb.py b/tests/migrations/test_edi_event_time_mismatch_jsonb.py
index 291f53ec6..108d3fa5e 100644
--- a/tests/migrations/test_edi_event_time_mismatch_jsonb.py
+++ b/tests/migrations/test_edi_event_time_mismatch_jsonb.py
@@ -101,15 +101,18 @@ def add(ts: datetime, kdes: dict, source: str = "edi_inbound") -> None:
                 add(now, {"ship_date": "2026-02-30"})
                 add(now, {"event_date": "2024-02-30"})  # also invalid
                 add(now, {"ship_date": "1900-02-29"})  # century non-leap
+                add(now, {"ship_date": "0000-01-01"})  # year 0000 invalid
                 # Non-date JSONB values are ignored, not cast errors.
                 add(now, {"ship_date": "not-a-date"})
+                add(now, {"ship_date": ""})
+                add(now, {"ship_date": "20-02-2026x"})
                 add(now, {"event_date": {"nested": iso}})
                 add(now, {"ship_date": None})
 
                 count = conn.execute(text(mismatch_sql)).scalar()
                 assert count == 2
                 invalid = conn.execute(text(invalid_sql)).scalar()
-                assert invalid == 3
+                assert invalid == 4
                 # Valid leap day must not be counted as invalid, and must not abort.
                 leap = uuid.uuid4()
                 conn.execute(
@@ -120,13 +123,12 @@ def add(ts: datetime, kdes: dict, source: str = "edi_inbound") -> None:
                     text(f"INSERT INTO {schema}.cte_kdes VALUES (:i, :t, :k, CAST(:v AS jsonb))"),
                     {"i": leap, "t": tenant, "k": "ship_date", "v": json.dumps("2024-02-29")},
                 )
-                assert conn.execute(text(invalid_sql)).scalar() == 3
+                assert conn.execute(text(invalid_sql)).scalar() == 4
                 # Python-side helper still agrees for a single valid ship_date row.
                 rows = conn.execute(
                     text(
                         f"SELECT e.event_timestamp, "
-                        f"(CASE WHEN {check._CALENDAR_VALID_ISO_DATE} "
-                        f"THEN ((k.kde_value #>> '{{}}')::date) ELSE NULL END) "
+                        f"{check._SAFE_DATE} "
                         f"FROM {schema}.cte_events e"
                         f" JOIN {schema}.cte_kdes k ON k.cte_event_id = e.id AND k.kde_key = 'ship_date'"
                         f" WHERE e.source LIKE 'edi%'"
@@ -137,3 +139,45 @@ def add(ts: datetime, kdes: dict, source: str = "edi_inbound") -> None:
                 assert check.count_mismatches(rows) >= 1
             finally:
                 trans.rollback()
+                try:
+                    with engine.connect() as cleanup_conn:
+                        cleanup_conn.execute(text(f"DROP SCHEMA IF EXISTS {schema} CASCADE"))
+                        cleanup_conn.commit()
+                except Exception:
+                    pass
+
+
+@pytest.mark.parametrize(
+    "val, expected_valid, expected_safe_date, expected_invalid_date",
+    [
+        ("not-a-date", False, None, False),
+        ("", False, None, False),
+        ("20-02-2026x", False, None, False),
+        ("0000-01-01", False, None, True),
+        ("2026-02-30", False, None, True),
+        ("2024-02-29", True, "2024-02-29", False),
+    ],
+)
+def test_date_validity_sql_handles_all_date_shapes(
+    val: str, expected_valid: bool, expected_safe_date: str | None, expected_invalid_date: bool
+) -> None:
+    """Verifies non-date strings, year 0000, invalid dates, and valid leap days."""
+    with rls_postgres_or_skip() as (_url, engine):
+        with engine.connect() as conn:
+            res = conn.execute(
+                text(
+                    f"SELECT "
+                    f"({check._CALENDAR_VALID_ISO_DATE}) AS is_valid, "
+                    f"({check._SAFE_DATE}) AS safe_d, "
+                    f"(CASE WHEN (k.kde_value #>> '{{}}') ~ '^\\d{{4}}-\\d{{2}}-\\d{{2}}$' "
+                    f"THEN NOT ({check._CALENDAR_VALID_ISO_DATE}) ELSE false END) AS is_invalid "
+                    f"FROM (VALUES (CAST(:v AS jsonb))) AS k(kde_value)"
+                ),
+                {"v": json.dumps(val)},
+            ).one()
+            assert res.is_valid is expected_valid
+            if expected_safe_date is None:
+                assert res.safe_d is None
+            else:
+                assert res.safe_d.isoformat() == expected_safe_date
+            assert res.is_invalid is expected_invalid_date

From 2d901cfd2ac8fea9d1fca22c19bbe6d1b64b2012 Mon Sep 17 00:00:00 2001
From: Christopher Sellers <1@christophersellers.com>
Date: Wed, 28 Oct 2026 04:49:28 -0700
Subject: [PATCH 10/10] fix(scripts): use set_tenant_guc for EDI mismatch check

Replace the inline set_config('app.tenant_id') that Semgrep flags as
semgrep.adhoc-tenant-guc-set-config with the canonical helper. --tenant-id
is already documented as a UUID, so set_tenant_guc is the right call.
---
 scripts/check_edi_event_time_mismatch.py | 7 +++----
 1 file changed, 3 insertions(+), 4 deletions(-)

diff --git a/scripts/check_edi_event_time_mismatch.py b/scripts/check_edi_event_time_mismatch.py
index e20a16bfa..88e6fcaab 100644
--- a/scripts/check_edi_event_time_mismatch.py
+++ b/scripts/check_edi_event_time_mismatch.py
@@ -230,10 +230,9 @@ def main(argv: Optional[list[str]] = None) -> int:
         try:
             conn.execute(text("SET TRANSACTION READ ONLY"))
             if args.tenant_id:
-                conn.execute(
-                    text("SELECT set_config('app.tenant_id', :tid, true)"),
-                    {"tid": args.tenant_id},
-                )
+                from shared.tenant_context import set_tenant_guc
+
+                set_tenant_guc(conn, args.tenant_id)
             try:
                 count = conn.execute(text(MISMATCH_SQL)).scalar_one()
                 invalid = conn.execute(text(INVALID_DATE_SQL)).scalar_one()