fix(fsma): strict TLC mismatch is warning, never hard reject

services/ingestion/app/edi_ingestion/routes.py

@@ -1240,6 +1240,8 @@ def _preflight_all_sets() -> None:
             )
             if slice_set in ("856", "861", "944", "945"):
                 slice_extracted["traceability_lot_code"] = validated["tlc"]
+            if validated.get("warnings"):
+                slice_extracted["fsma_validation_warnings"] = validated["warnings"]
             slice_extracted["fsma_validation_status"] = "passed"
         except ValidationError as ve:
             errors = ve.errors(include_context=False) if slice_set in ("856", "861", "944", "945") else ve.errors()
@@ -1312,6 +1314,7 @@ def _summarize(slice_set: str, slice_extracted: dict[str, Any]) -> dict[str, Any
             "product_description": slice_extracted.get("product_description"),
             "reference_document_number": slice_extracted.get("reference_document_number"),
             "fsma_validation_status": slice_extracted.get("fsma_validation_status"),
+            "fsma_validation_warnings": slice_extracted.get("fsma_validation_warnings", []),
         }
         if slice_set in ("856", "861", "944", "945"):
             summary.update(

services/ingestion/app/edi_ingestion/validation.py

@@ -39,7 +39,7 @@ def _validate_edi_as_fsma_event(
     forced every caller to treat ``None`` as "maybe-failed, maybe-
     unavailable" and masked schema errors in the audit trail.
     """
-    from shared.schemas import FSMAEvent, FSMAEventType
+    from shared.schemas import _TLC_PRODUCTION_RE, FSMAEvent, FSMAEventType
 
     type_map = {
         "856": FSMAEventType.SHIPPING,
@@ -56,18 +56,26 @@ def _validate_edi_as_fsma_event(
             lot_number=extracted.get("lot_number"),
             quantity=extracted.get("quantity"),
         )
-        tlc = line.lot_number
-        # Follow the shared EDI schema TLC rule; originator lot versus
-        # GTIN-prefixed TLC remains a founder decision (#1104).
+        tlc = line.lot_number.strip()
+        # A-2170: a whitespace-only lot strips empty. Reject before the
+        # GTIN/UPC rebuild so we never emit a production-shaped "GTIN-"
+        # TLC that FSMAEvent would accept, collapsing blank lots onto one id.
+        if not tlc:
+            _WarehouseLineRequiredKDEs(lot_number=tlc, quantity=line.quantity)
+        # Pattern misses are advisory, so check explicitly to retain the
+        # warehouse GTIN/UPC rebuild for both accepted and short lots.
         try:
-            FSMAEvent.validate_tlc_format(tlc)
+            tlc = FSMAEvent.validate_tlc_format(tlc)
+            rebuild = not _TLC_PRODUCTION_RE.fullmatch(tlc)
         except ValueError:
+            rebuild = True
+        if rebuild:
             for item_id, lengths in (
                 (extracted.get("gtin"), (8, 12, 13, 14)),
                 (extracted.get("upc"), (12, 13)),
             ):
                 if isinstance(item_id, str) and item_id.isascii() and item_id.isdigit() and len(item_id) in lengths:
-                    tlc = f"{item_id.zfill(14)}-{line.lot_number}"
+                    tlc = f"{item_id.zfill(14)}-{tlc}"
                     break
     elif transaction_set in ("856", "861"):
         # The line lot is the TLC verbatim. Do not prefix a GTIN (#1104).

services/ingestion/tests/test_edi_wire_format_2974_2976_2978.py

@@ -197,8 +197,9 @@ def _post(
 # ---------------------------------------------------------------------------
 
 
+@pytest.mark.parametrize("tlc", (_VALID_FSMA_TLC, "A" * 3))
 def test_2976_multi_st_interchange_emits_one_event_per_transaction_set(
-    client: TestClient, captured_payload: dict
+    client: TestClient, captured_payload: dict, tlc: str
 ) -> None:
     """A batched interchange must produce one CTE per ST/SE pair with no
     cross-shipment field bleed.
@@ -236,7 +237,7 @@ def test_2976_multi_st_interchange_emits_one_event_per_transaction_set(
         isa13="000029761",
     )
 
-    response = _post(client, raw)
+    response = _post(client, raw, extra_data={"traceability_lot_code": tlc})
     assert response.status_code == 201, response.text
 
     payload = captured_payload["payload"]
@@ -262,6 +263,10 @@ def test_2976_multi_st_interchange_emits_one_event_per_transaction_set(
 
     body = response.json()
     assert body["extracted"]["transaction_set_count"] == 3
+    if tlc == "A" * 3:
+        for transaction in body["extracted"]["transaction_sets"]:
+            assert transaction["fsma_validation_status"] == "passed"
+            assert any("production pattern" in warning for warning in transaction["fsma_validation_warnings"])
 
 
 def test_2976_ge01_disagreeing_with_st_count_is_rejected(client: TestClient, captured_payload: dict) -> None:
@@ -558,8 +563,8 @@ def test_2976_batched_interchange_never_half_lands(client: TestClient, captured_
         "/api/v1/ingest/edi/document",
         params={"strict": "false"},
         data={
-            # Fails FSMAEvent.validate_tlc_format for every set.
-            "traceability_lot_code": "not-a-gtin-tlc",
+            # Below the shared TLC minimum for every set; pattern misses only warn.
+            "traceability_lot_code": "A" * 2,
             "tenant_id": TEST_TENANT_ID,
         },
         files={"file": ("batch.edi", raw, "application/edi-x12")},
@@ -670,3 +675,59 @@ def test_scalar_overrides_still_work_for_a_single_transaction_set(client: TestCl
 
     assert response.status_code in (200, 201), response.text
     assert captured_payload["payload"].events[0].quantity == 42.0
+
+
+@pytest.mark.parametrize("transaction_set", ("856", "861", "944", "945"))
+def test_tlc_pattern_miss_is_accepted_with_persisted_warning(transaction_set):
+    validated = edi_ingestion_routes._validate_edi_as_fsma_event(
+        {"lot_number": " \t" + "A" * 3 + "\n", "quantity": 1},
+        transaction_set,
+        "A" * 3,
+    )
+    assert validated["tlc"] == "A" * 3
+    assert any("production pattern" in warning for warning in validated["warnings"])
+
+
+@pytest.mark.parametrize("transaction_set", ("944", "945"))
+@pytest.mark.parametrize("lot", ("A" * 3, "0" * 14 + "A", "A"))
+def test_warehouse_tlc_prefix_rebuild_is_reachable(transaction_set, lot):
+    validated = edi_ingestion_routes._validate_edi_as_fsma_event(
+        {"lot_number": lot, "quantity": 1, "gtin": "0" * 14},
+        transaction_set,
+        lot,
+    )
+    expected = lot if lot.startswith("0" * 14) else "0" * 14 + "-" + lot.strip()
+    assert validated["tlc"] == expected
+    assert validated["warnings"] == []
+
+
+@pytest.mark.parametrize("transaction_set", ("944", "945"))
+@pytest.mark.parametrize("item_id", ("A" * 14, "0" * 3))
+def test_invalid_warehouse_prefix_keeps_lot_and_warning(transaction_set, item_id):
+    validated = edi_ingestion_routes._validate_edi_as_fsma_event(
+        {"lot_number": "A" * 3, "quantity": 1, "gtin": item_id, "upc": item_id},
+        transaction_set,
+        "A" * 3,
+    )
+    assert validated["tlc"] == "A" * 3
+    assert any("production pattern" in warning for warning in validated["warnings"])
+
+
+@pytest.mark.parametrize("transaction_set", ("944", "945"))
+@pytest.mark.parametrize("lot", ("", " ", "   ", "\t"))
+def test_warehouse_stripped_empty_lot_rejected_before_rebuild(transaction_set, lot):
+    """A-2170: empty or whitespace-only 944/945 lots reject like a missing lot.
+
+    Must fail before the GTIN/UPC rebuild; otherwise a blank lot becomes
+    ``{gtin}-`` and FSMAEvent accepts it with no warning.
+    """
+    from pydantic import ValidationError
+
+    with pytest.raises(ValidationError) as excinfo:
+        edi_ingestion_routes._validate_edi_as_fsma_event(
+            {"lot_number": lot, "quantity": 1, "gtin": "0" * 14},
+            transaction_set,
+            "caller-tlc",
+        )
+    assert any(err["loc"][-1] == "lot_number" for err in excinfo.value.errors())
+

services/ingestion/tests/test_epcis_cleartext_logging.py

@@ -117,9 +117,12 @@ def test_fsma_rejection_logs_omit_private_values(ingest, warning, monkeypatch, c
     monkeypatch.setattr(pmod, "_fsma_strict_mode", lambda: True)
     monkeypatch.setattr(pmod, "_epcis_store", {})
     monkeypatch.setattr(pmod, "_epcis_idempotency_index", {})
+    # TLC production-pattern misses are warnings now, so a valid-length TLC alone
+    # no longer triggers fsma_validation_failed. Blank eventTime still hard-fails
+    # FSMAEvent while keeping TLC in the audited payload for cleartext checks.
     event = {
         "type": "ObjectEvent",
-        "eventTime": datetime.now(timezone.utc).isoformat(),
+        "eventTime": "   ",
         "action": "OBSERVE",
         "bizStep": "receiving",
         "ilmd": {"fsma:traceabilityLotCode": TLC},
@@ -133,7 +136,7 @@ def test_fsma_rejection_logs_omit_private_values(ingest, warning, monkeypatch, c
     assert warning in [record.getMessage() for record in caplog.records]
     assert audited[0][1] == TENANT
     assert audited[0][2]["tlc"] == TLC
-    assert any(error["loc"] == ("tlc",) for error in audited[0][0])
+    assert any(error["loc"] == ("event_time",) for error in audited[0][0])
     _assert_clean(caplog)
 
 
@@ -161,16 +164,20 @@ def _raise(db_session, tenant_id, prepared):
 def test_pydantic_validation_failure_log_omits_private_values(monkeypatch, caplog):
     audited = []
     monkeypatch.setattr(vmod, "_audit_log_validation_failure", lambda e, t, n: audited.append((e, t, n)))
+    # Valid-length TLC is accepted with a production-pattern warning; blank
+    # event_time still hard-fails while keeping TLC in the audited payload.
     normalized = {
         "event_type": "receiving",
         "tlc": TLC,
-        "event_time": datetime.now(timezone.utc).isoformat(),
+        "event_time": "   ",
     }
     with caplog.at_level(logging.WARNING):
         result = vmod._validate_as_fsma_event(normalized, tenant_id=TENANT)
 
     assert result is None
     assert "fsma_event_validation_failed" in _log_text(caplog)
     assert audited[0][1:] == (TENANT, normalized)
-    assert any(error["loc"] == ("tlc",) and error["type"] == "value_error" for error in audited[0][0])
+    assert any(
+        error["loc"] == ("event_time",) and error["type"] == "value_error" for error in audited[0][0]
+    )
     _assert_clean(caplog)

services/ingestion/tests/test_r3_restored_fixtures_e2e.py

@@ -38,7 +38,6 @@
 from app.edi_ingestion.validation import _validate_edi_as_fsma_event  # noqa: E402
 from app.fda_export_service import FDA_COLUMNS, _generate_csv  # noqa: E402
 from app.webhook_router_v2 import _validate_event_kdes  # noqa: E402
-from pydantic import ValidationError  # noqa: E402
 from shared.data_gap_tracking import (  # noqa: E402
     GAP_KIND_MISSING_KDE,
     decide_transition,
@@ -102,9 +101,7 @@ def test_csv_harvesting_fixture_end_to_end():
 
     headers, rows = _parse_sheet(sheet)
     assert headers == FDA_COLUMNS
-    assert [r["Traceability Lot Code (TLC)"] for r in rows] == [
-        s["traceability_lot_code"] for s in source
-    ]
+    assert [r["Traceability Lot Code (TLC)"] for r in rows] == [s["traceability_lot_code"] for s in source]
     for row, src in zip(rows, source):
         assert row["Event Type (CTE)"].lower() == "harvesting"
         assert row["Product Description"] == src["product_description"]
@@ -133,12 +130,11 @@ def test_edi_856_fixture_end_to_end():
     merged = {**header, **lines[0], "edi_line_index": 1}
     tlc = extracted["asn_number"]
 
-    with pytest.raises(ValidationError):
-        _validate_edi_as_fsma_event(merged, "856", tlc, None)
+    validated = _validate_edi_as_fsma_event(merged, "856", tlc, None)
+    assert validated["tlc"] == tlc.strip()
+    assert any("production pattern" in warning for warning in validated["warnings"])
 
-    event = edi_routes._build_ingest_event_for_set(
-        "856", merged, tlc, None, None, None, None, segments=segments
-    )
+    event = edi_routes._build_ingest_event_for_set("856", merged, tlc, None, None, None, None, segments=segments)
     assert event.cte_type.value == "shipping"
     assert event.quantity == lines[0]["quantity"]
     assert event.kdes["ship_from_gln"] == expected["ship_from_gln"]

services/shared/fsma_validation.py

@@ -17,7 +17,7 @@
 detect_identifier_type  – auto-detect identifier kind from format
 validate_gln            – GLN-13 with GS1 check digit
 validate_gtin           – GTIN-8/12/13/14 with GS1 check digit
-validate_tlc            – Traceability Lot Code (GTIN-14 prefix in strict mode)
+validate_tlc            – Traceability Lot Code (GTIN-14 pattern warning in strict mode)
 validate_sscc           – SSCC-18 with GS1 check digit
 validate_identifier     – auto-detecting dispatcher
 validate_batch          – batch validation helper
@@ -187,14 +187,14 @@ def validate_tlc(tlc: str, *, strict: bool = False) -> ValidationResult:
     """
     Validate Traceability Lot Code (TLC).
 
-    A production TLC must match the GTIN-14 prefix pattern:
+    Strict mode checks the production GTIN-14 prefix pattern:
     ``^\\d{14}[A-Za-z0-9\\-\\.]+$`` (14 numeric digits followed by an
-    alphanumeric lot suffix).  In non-strict mode the format is more
-    permissive (minimum 3 chars, alphanumeric + common separators).
+    alphanumeric lot suffix). Strict mismatches are warnings, not failures.
+    Both modes require at least 3 characters and warn on unusual characters.
 
     Args:
         tlc: Raw TLC string to validate.
-        strict: When True, enforce the production GTIN-14 prefix pattern.
+        strict: When True, warn on a production GTIN-14 prefix pattern mismatch.
     """
     raw = str(tlc) if tlc else ""
     cleaned = raw.strip()
@@ -209,28 +209,14 @@ def validate_tlc(tlc: str, *, strict: bool = False) -> ValidationResult:
 
     warnings: List[str] = []
 
-    if strict:
-        # Production mode: GTIN-14 prefix + alphanumeric lot suffix
-        if not _TLC_PRODUCTION_PATTERN.match(cleaned):
-            return ValidationResult(
-                is_valid=False,
-                identifier_type=IdentifierType.TLC,
-                original_value=raw,
-                errors=[
-                    ValidationError(
-                        field="tlc",
-                        message=(
-                            "TLC does not match production pattern "
-                            "(expected 14-digit GTIN prefix + alphanumeric lot suffix, "
-                            "e.g. '00012345678901-LotA')"
-                        ),
-                    )
-                ],
-            )
-    else:
-        # Permissive mode: reject obviously invalid characters
-        if not re.match(r"^[A-Za-z0-9\-\._/]+$", cleaned):
-            warnings.append("TLC contains unusual characters; consider strict validation")
+    if strict and not _TLC_PRODUCTION_PATTERN.match(cleaned):
+        warnings.append(
+            "Warning: TLC does not match production pattern "
+            "(14-digit GTIN prefix + alphanumeric lot suffix); this is not a validation failure"
+        )
+
+    if not re.match(r"^[A-Za-z0-9\-\._/]+$", cleaned):
+        warnings.append("TLC contains unusual characters; consider strict validation")
 
     # Warn on very long TLCs
     if len(cleaned) > 50:

services/shared/schemas.py

@@ -11,7 +11,9 @@
 from urllib.parse import unquote
 from uuid import UUID, uuid4
 
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, Field, field_validator, model_validator
+
+from .fsma_validation import validate_tlc
 
 # Example constants to avoid duplication in schema examples
 EX_SUBJECT = "food facilities on the FTL"
@@ -372,7 +374,7 @@ class FSMAEvent(BaseModel):
     model before being emitted to Kafka.
 
     Required KDEs (rejection triggers):
-      - ``tlc``        – GTIN-14 + lot, LGTIN class URN, or GS1 Digital Link
+      - ``tlc``        – at least 3 characters after stripping whitespace
       - ``event_time`` – ISO 8601 timestamp (required)
     """
 
@@ -386,11 +388,8 @@ class FSMAEvent(BaseModel):
     )
     tlc: str = Field(
         ...,
-        description="Traceability Lot Code (GTIN-14 + lot, LGTIN class URN, or GS1 Digital Link)",
-        # 15 is the GTIN-14 form (14 digits + 1 lot character). An LGTIN class
-        # URN is at least 36 characters and a GS1 Digital Link at least 31,
-        # so this floor does not reject either lot-level form.
-        min_length=15,
+        description="Traceability Lot Code (production-pattern mismatches warn only)",
+        min_length=3,
     )
     product_description: str = Field(
         default="Unknown Product",
@@ -409,6 +408,8 @@ class FSMAEvent(BaseModel):
     reference_document_number: Optional[str] = Field(None, description="Source document reference number")
     tenant_id: Optional[str] = Field(None, description="Tenant identifier for multi-tenancy")
 
+    warnings: List[str] = Field(default_factory=list)
+
     # Legacy fields retained for backward compatibility with document extraction
     date: Optional[str] = Field(None, description="Event date in YYYY-MM-DD format")
     cte_type: Optional[CTEType] = Field(None, description="Legacy CTE type (use event_type instead)")
@@ -424,18 +425,23 @@ class FSMAEvent(BaseModel):
     @field_validator("tlc")
     @classmethod
     def validate_tlc_format(cls, v: str) -> str:
-        """Accept GTIN-14 + lot, an LGTIN class URN, or a GS1 Digital Link."""
-        if _TLC_PRODUCTION_RE.fullmatch(v):
-            return v
-        for pattern in (_TLC_LGTIN_RE, _TLC_DIGITAL_LINK_RE):
-            matched = pattern.fullmatch(v)
-            if matched is not None and _decoded_ai10_lot_within_limit(matched.group("lot")):
-                return v
-        raise ValueError(
-            f"TLC '{v}' does not match FSMA 204 production pattern "
-            "(accepted formats: a fourteen-digit GTIN prefix followed by a lot "
-            "suffix, an LGTIN class URN, or a GS1 Digital Link)"
-        )
+        """Strip surrounding whitespace; reject blank or short TLCs.
+
+        Production-pattern advisories are collected on each model instance.
+        """
+        cleaned = v.strip()
+        if not cleaned:
+            raise ValueError("TLC must not be blank")
+        if len(cleaned) < 3:
+            raise ValueError("TLC must be at least 3 characters")
+        return cleaned
+
+    @model_validator(mode="after")
+    def collect_tlc_warnings(self) -> "FSMAEvent":
+        for warning in validate_tlc(self.tlc, strict=True).warnings:
+            if warning not in self.warnings:
+                self.warnings.append(warning)
+        return self
 
     @field_validator("event_time")
     @classmethod