fix(shared): SEL-598 lossless lot identity / GTIN-14 Digital Link (H2)

services/ingestion/app/epcis_export.py

@@ -19,6 +19,7 @@
 from fastapi.responses import JSONResponse, StreamingResponse
 from pydantic import BaseModel, Field
 from shared.database import get_db_safe
+from shared.epcis_lot_identity import is_full_class_uri, lot_token_only
 from shared.fsma_validation import normalize_gtin, verify_gs1_check_digit
 from shared.tenant_context import TenantContext
 from sqlalchemy import text
@@ -46,6 +47,53 @@ class ExportStorageUnavailable(RuntimeError):
     """The event store could not be read; never report this as "no data"."""
 
 
+def _lot_code_filter(lot_code: str | None) -> tuple[str | None, dict]:
+    """SQL predicate for an export lot filter that reads both TLC shapes (SEL-598).
+
+    Rows may store the bare lot token (pre-H2, or ``LOT_IDENTITY_LOSSLESS``
+    off) or the full LGTIN / GS1 Digital Link class URI (flag on). A bare
+    filter also matches URIs whose lot segment ends with that token; a URI
+    filter also matches the bare token. ``_row_matches_lot`` then keeps
+    only rows whose lot token is exactly the requested one.
+    """
+    if not lot_code:
+        return None, {}
+    params: dict = {"lot_code": lot_code}
+    clauses = ["e.traceability_lot_code = :lot_code"]
+    token = lot_token_only(lot_code)
+    if token and token != lot_code:
+        # Full URI requested: also match rows stored with the bare token.
+        params["lot_token"] = token
+        clauses.append("e.traceability_lot_code = :lot_token")
+    elif token:
+        # Bare lot requested: also match full class URIs carrying it.
+        suffixes = sorted({f"/10/{token}", f"/10/{quote(token, safe='-._~')}", f".{token}", f".{quote(token, safe='-._~')}"})
+        for index, suffix in enumerate(suffixes):
+            name = f"lot_suffix_{index}"
+            params[name] = suffix
+            clauses.append(
+                "((lower(e.traceability_lot_code) LIKE 'https://id.gs1.org/01/%'"
+                " OR lower(e.traceability_lot_code) LIKE 'urn:epc:class:lgtin:%')"
+                f" AND right(e.traceability_lot_code, char_length(:{name})) = :{name})"
+            )
+    return "(" + " OR ".join(clauses) + ")", params
+
+
+def _row_matches_lot(stored: object, lot_code: str) -> bool:
+    """Exact lot match across bare-token and full-class-URI storage."""
+    if not isinstance(stored, str):
+        return False
+    if stored == lot_code:
+        return True
+    wanted = lot_token_only(lot_code)
+    if not wanted:
+        return False
+    if is_full_class_uri(lot_code):
+        # A URI filter matches its own URI (above) or a bare historical row.
+        return stored == wanted
+    return lot_token_only(stored) == wanted and (stored == wanted or is_full_class_uri(stored))
+
+
 def _query_tenant_events(
     tenant_id: str, lot_code: str | None, date_from: str | None, date_to: str | None
 ) -> list[dict]:
@@ -59,9 +107,10 @@ def _query_tenant_events(
     try:
         where = []
         params: dict = {"tenant_id": tenant_id}
-        if lot_code:
-            where.append("e.traceability_lot_code = :lot_code")
-            params["lot_code"] = lot_code
+        lot_predicate, lot_params = _lot_code_filter(lot_code)
+        if lot_predicate:
+            where.append(lot_predicate)
+            params.update(lot_params)
         if date_from:
             where.append("e.event_timestamp >= :date_from")
             params["date_from"] = date_from
@@ -117,12 +166,15 @@ def _query_tenant_events(
             """),
             {**params, "row_limit": _MAX_EXPORT_EVENTS + 1},
         ).fetchall()
-        return [dict(row._mapping) for row in rows]
+        result = [dict(row._mapping) for row in rows]
     except Exception as exc:
         logger.error("db_query_failed_for_export", extra={"error": str(exc)})
         raise ExportStorageUnavailable(str(exc)) from exc
     finally:
         db.close()
+    if lot_code:
+        result = [row for row in result if _row_matches_lot(row.get("traceability_lot_code"), lot_code)]
+    return result
 
 
 def _load_export_rows(tenant_id: str, request: "ExportRequest") -> list[dict]:
@@ -332,13 +384,20 @@ def _valid_class_uri(value: object) -> bool:
 def _epc_class(row: dict) -> str | None:
     """Use an existing class URI, product_id, or a GTIN Digital Link.
 
+    SEL-598 / H2: if ``traceability_lot_code`` (or ``epc_class`` /
+    ``product_id``) already holds a full LGTIN / Digital Link, emit it
+    unchanged. Never pass a full class URI as the ``{lot}`` argument when
+    rebuilding from GTIN + lot token.
+
     GTIN/GLN digits do not reveal the company-prefix split required by EPC
-    URNs. A lot code alone cannot identify an EPC class — callers must
+    URNs. A lot code alone cannot identify an EPC class: callers must
     persist ``epc_class``, ``product_id`` (class URI), ``gtin``, or an
     embedded ``epcis_document_json`` quantity class.
     """
-    for key in ("epc_class", "product_id"):
+    for key in ("epc_class", "product_id", "traceability_lot_code", "canonical_epc_class"):
         value = _kde_scalar(row.get(key))
+        # Strict export gate only — do not emit URIs the lenient parser accepts
+        # but `_valid_class_uri` rejects (Codex P2 / CodeRabbit).
         if _valid_class_uri(value):
             return value
     raw = row.get("epcis_document_json")
@@ -354,7 +413,16 @@ def _epc_class(row: dict) -> str | None:
         except (TypeError, ValueError):
             pass
     gtin_raw = _kde_scalar(row.get("gtin")) or _kde_scalar(row.get("product_id"))
-    lot = _kde_scalar(row.get("traceability_lot_code"))
+    raw_lot = _kde_scalar(row.get("lot_token")) or _kde_scalar(row.get("traceability_lot_code"))
+    # Never pass a full URI as the lot argument (SEL-598).
+    lot = lot_token_only(raw_lot) or (
+        raw_lot
+        if raw_lot
+        and not is_full_class_uri(raw_lot)
+        and "://" not in raw_lot
+        and not str(raw_lot).lower().startswith("urn:epc:")
+        else None
+    )
     digits = re.sub(r"\D", "", gtin_raw)
     if digits and lot:
         try:

services/ingestion/tests/test_epcis_export_1342.py

@@ -117,8 +117,12 @@ def test_optional_filters_preserve_tenant_scope(
         sql, params = session.calls[0][1]
         expected_where = "WHERE e.tenant_id = :tenant_id"
         expected_params = {"tenant_id": tenant_id, "row_limit": epcis_export._MAX_EXPORT_EVENTS + 1}
+        if lot_code is not None:
+            # SEL-598 dual-read: bare and full-class-URI TLCs both match.
+            lot_predicate, lot_params = epcis_export._lot_code_filter(lot_code)
+            expected_where += f" AND {lot_predicate}"
+            expected_params.update(lot_params)
         for name, value, predicate in (
-            ("lot_code", lot_code, "e.traceability_lot_code = :lot_code"),
             ("date_from", date_from, "e.event_timestamp >= :date_from"),
             ("date_to", date_to, "e.event_timestamp < :date_to"),
         ):

services/shared/canonical_event.py

@@ -32,6 +32,12 @@
 from uuid import UUID, uuid4
 
 from pydantic import BaseModel, Field, field_validator
+from shared.epcis_lot_identity import (
+    lossless_enabled,
+    lot_token_only,
+    normalize_epcis_lot,
+    parse_class_uri,
+)
 
 logger = logging.getLogger("canonical-event")
 
@@ -1038,8 +1044,7 @@ def _resolve_bizstep_to_cte_type(biz_step: Any) -> CTEType:
     raw = "" if biz_step is None else str(biz_step).strip()
     if not raw:
         raise UnmappedBizStepError(
-            "EPCIS event has no bizStep — cannot derive an FSMA 204 CTE type. "
-            "Supply a CBV or FSMA bizStep URI. (#2815)"
+            "EPCIS event has no bizStep — cannot derive an FSMA 204 CTE type. Supply a CBV or FSMA bizStep URI. (#2815)"
         )
 
     lowered = raw.lower()
@@ -1097,8 +1102,13 @@ def _epcis_list(epcis_data: Dict[str, Any], key: str) -> List[Any]:
     return list(value) if isinstance(value, list) else []
 
 
-def _epcis_traceability_lot_code(epcis_data: Dict[str, Any]) -> Tuple[str, Optional[str]]:
-    """Resolve ``(traceability_lot_code, gtin)`` for an EPCIS event.
+def _legacy_epcis_traceability_lot_code(epcis_data: Dict[str, Any]) -> Tuple[str, Optional[str]]:
+    """Pre-SEL-598 resolver, used while ``LOT_IDENTITY_LOSSLESS`` is off.
+
+    Kept byte-for-byte so flag-off writes keep the same TLC, GTIN KDE, event
+    hash and idempotency key as before H2.
+
+    Resolve ``(traceability_lot_code, gtin)`` for an EPCIS event.
 
     Precedence: ILMD ``fsma:traceabilityLotCode`` / ``cbvmda:lotNumber`` (the
     same order as ``epcis/extraction._extract_lot_data``), then the lot of an
@@ -1148,7 +1158,7 @@ def _epcis_traceability_lot_code(epcis_data: Dict[str, Any]) -> Tuple[str, Optio
     )
 
 
-def _epcis_transformation_inputs(epcis_data: Dict[str, Any]) -> List[Dict[str, Any]]:
+def _legacy_epcis_transformation_inputs(epcis_data: Dict[str, Any]) -> List[Dict[str, Any]]:
     """Per-input lot, quantity and unit for a TransformationEvent.
 
     ``inputQuantityList`` carries class-level (LGTIN) inputs with an explicit
@@ -1187,6 +1197,157 @@ def _epcis_transformation_inputs(epcis_data: Dict[str, Any]) -> List[Dict[str, A
     return inputs
 
 
+def _epcis_traceability_lot_code(epcis_data: Dict[str, Any]) -> Tuple[str, Optional[str]]:
+    """Resolve ``(traceability_lot_code, gtin)`` for an EPCIS event.
+
+    SEL-598 / H2: when ``LOT_IDENTITY_LOSSLESS`` is enabled (opt-in), the
+    stored ``traceability_lot_code`` is the full LGTIN or GS1 Digital Link
+    class URI when present. The bare lot token remains available via
+    ``lot_token_only`` / ``normalize_epcis_lot(...).lot_token``.
+
+    Precedence (H2 / SEL-598, intentional vs legacy extraction): a full class
+    URI among quantity classes / EPCs first (canonical stored identity), then
+    ILMD ``fsma:traceabilityLotCode`` / ``cbvmda:lotNumber`` as the bare partner
+    TLC when no class URI is present (preserve bare TLC — do not rebuild a
+    Digital Link from an SGTIN-derived GTIN at this boundary), then SGTIN
+    serial, then flat top-level keys. Legacy ``epcis/extraction`` prefers ILMD
+    first; canonical + export agree on class-URI-first after H2. Raises
+    ``ValueError`` when nothing identifies the lot.
+
+    With the flag off (default) this is the pre-H2 resolver unchanged: the
+    stored TLC, GTIN KDE, event hash and idempotency key do not move.
+    """
+    if not lossless_enabled():
+        return _legacy_epcis_traceability_lot_code(epcis_data)
+    ilmd = epcis_data.get("ilmd")
+    if not isinstance(ilmd, dict):
+        ilmd = {}
+    epcs = _epcis_list(epcis_data, "epcList") + _epcis_list(epcis_data, "outputEPCList")
+    extension = epcis_data.get("extension")
+    extension = extension if isinstance(extension, dict) else {}
+    classes = [
+        q.get("epcClass")
+        for q in (
+            _epcis_list(epcis_data, "outputQuantityList")
+            + _epcis_list(epcis_data, "quantityList")
+            + (list(extension.get("quantityList") or []) if isinstance(extension.get("quantityList"), list) else [])
+        )
+        if isinstance(q, dict) and q.get("epcClass")
+    ]
+    # Real GTIN-14 only (never the EPC company.item fragment from _epc_gtin).
+    gtin = None
+    for e in list(classes) + list(epcs):
+        parsed = parse_class_uri(e)
+        if parsed and parsed.gtin:
+            gtin = parsed.gtin
+            break
+        identity = normalize_epcis_lot(e, prefer_full=False)
+        if identity and identity.gtin:
+            gtin = identity.gtin
+            break
+
+    # Prefer full class URI from EPC / quantity class (SEL-598 canonical).
+    for epc in list(classes) + list(epcs):
+        parsed = parse_class_uri(epc)
+        if parsed:
+            if lossless_enabled():
+                return parsed.canonical_id, parsed.gtin or gtin
+            return parsed.lot_token, parsed.gtin or gtin
+
+    for key in ("fsma:traceabilityLotCode", "traceabilityLotCode", "cbvmda:lotNumber", "lotNumber"):
+        value = ilmd.get(key)
+        if value is not None and str(value).strip():
+            # Preserve bare partner TLC; do not rebuild Digital Link here.
+            identity = normalize_epcis_lot(value)
+            if identity:
+                return identity.canonical_id, identity.gtin or gtin
+            return str(value).strip(), gtin
+
+    for epc in epcs:
+        if "sgtin" in str(epc).lower():
+            identity = normalize_epcis_lot(epc)
+            if identity:
+                return identity.canonical_id, identity.gtin or gtin
+            segment = _epc_last_segment(epc)
+            if segment:
+                return segment, gtin
+
+    for key in ("traceability_lot_code", "lotNumber"):
+        value = epcis_data.get(key)
+        if value is not None and str(value).strip():
+            # Preserve bare flat TLC (same boundary rule as ILMD).
+            identity = normalize_epcis_lot(value)
+            if identity:
+                return identity.canonical_id, identity.gtin or gtin
+            return str(value).strip(), gtin
+
+    raise ValueError(
+        "EPCIS event carries no lot identifier (ILMD traceabilityLotCode or lotNumber, an LGTIN or SGTIN "
+        "EPC, or traceability_lot_code); refusing to fabricate one"
+    )
+
+
+def _epcis_transformation_inputs(epcis_data: Dict[str, Any]) -> List[Dict[str, Any]]:
+    """Per-input lot, quantity and unit for a TransformationEvent.
+
+    ``inputQuantityList`` carries class-level (LGTIN) inputs with an explicit
+    quantity; a quantityElement without ``uom`` is a count of instances in
+    EPCIS, recorded as ``each``. ``inputEPCList`` carries instance EPCs, each
+    of which is exactly one item. Lots are not merged or summed.
+    """
+    if not lossless_enabled():
+        return _legacy_epcis_transformation_inputs(epcis_data)
+    inputs: List[Dict[str, Any]] = []
+    seen: set = set()
+    for element in _epcis_list(epcis_data, "inputQuantityList"):
+        if not isinstance(element, dict):
+            continue
+        epc_class = element.get("epcClass")
+        identity = normalize_epcis_lot(epc_class)
+        if identity is not None:
+            lot_key = identity.canonical_id if lossless_enabled() else identity.lot_token
+            lot_token = identity.lot_token
+        else:
+            lot_token = _epc_last_segment(epc_class)
+            lot_key = lot_token
+        dedup_key = ("class", lot_key)
+        if not lot_token or dedup_key in seen:
+            continue
+        raw_quantity = element.get("quantity")
+        try:
+            quantity = float(raw_quantity) if raw_quantity is not None else None
+        except (TypeError, ValueError):
+            quantity = None
+        inputs.append(
+            {
+                "traceability_lot_code": lot_key,
+                "lot_token": lot_token,
+                "quantity": quantity,
+                "unit_of_measure": (element.get("uom") or "each") if quantity is not None else None,
+                "epc_class": epc_class,
+            }
+        )
+        seen.add(dedup_key)
+    for epc in _epcis_list(epcis_data, "inputEPCList"):
+        identity = normalize_epcis_lot(epc)
+        segment = identity.lot_token if identity else _epc_last_segment(epc)
+        dedup_key = ("epc", segment)
+        if not segment or dedup_key in seen:
+            continue
+        lot_key = identity.canonical_id if (identity and lossless_enabled()) else segment
+        inputs.append(
+            {
+                "traceability_lot_code": lot_key,
+                "lot_token": segment,
+                "quantity": 1.0,
+                "unit_of_measure": "each",
+                "epc": str(epc),
+            }
+        )
+        seen.add(dedup_key)
+    return inputs
+
+
 def normalize_epcis_event(
     epcis_data: Dict[str, Any],
     tenant_id: str,