From 76eb2c76559bd36abc37208472969c00f4cdfb18 Mon Sep 17 00:00:00 2001
From: Christopher Sellers <1@christophersellers.com>
Date: Wed, 28 Oct 2026 08:33:10 -0700
Subject: [PATCH 1/3] fix(shared): SEL-598 lossless EPCIS lot identity (H2
 Preview)

Canonical stored identity is the full LGTIN or GS1 Digital Link; bare lot
is derived. Export emits a stored class URI unchanged and never passes a
full URI as the lot argument when rebuilding. Flag LOT_IDENTITY_LOSSLESS
defaults on (set 0 to restore segment-only writes). Synthetic tests only.
---
 services/ingestion/app/epcis_export.py        |  36 ++-
 services/shared/canonical_event.py            |  81 ++++--
 services/shared/epcis_lot_identity.py         | 240 ++++++++++++++++++
 .../shared/test_epcis_lot_identity_sel598.py  | 157 ++++++++++++
 4 files changed, 485 insertions(+), 29 deletions(-)
 create mode 100644 services/shared/epcis_lot_identity.py
 create mode 100644 tests/shared/test_epcis_lot_identity_sel598.py

diff --git a/services/ingestion/app/epcis_export.py b/services/ingestion/app/epcis_export.py
index d5a1f0859..7f18e1dbb 100644
--- a/services/ingestion/app/epcis_export.py
+++ b/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, parse_class_uri
 from shared.fsma_validation import normalize_gtin, verify_gs1_check_digit
 from shared.tenant_context import TenantContext
 from sqlalchemy import text
@@ -332,15 +333,26 @@ 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))
-        if _valid_class_uri(value):
-            return value
+        if _valid_class_uri(value) or is_full_class_uri(value):
+            # Prefer the validated export shape when available; otherwise the
+            # parsed canonical URI from the shared normalizer.
+            if _valid_class_uri(value):
+                return value
+            parsed = parse_class_uri(value)
+            if parsed:
+                return parsed.canonical_id
     raw = row.get("epcis_document_json")
     if raw:
         try:
@@ -354,7 +366,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:
@@ -525,10 +546,7 @@ def _skip(row: dict, reason: str) -> None:
                 _skip(r, "missing_timestamp")
                 continue
             event_kind = r.get("epcis_event_type")
-            is_transformation = (
-                r.get("event_type") == "transformation"
-                or event_kind == "TransformationEvent"
-            )
+            is_transformation = r.get("event_type") == "transformation" or event_kind == "TransformationEvent"
             if event_kind not in (None, "ObjectEvent", "TransformationEvent"):
                 _skip(r, "unsupported_epcis_event_type")
                 continue
diff --git a/services/shared/canonical_event.py b/services/shared/canonical_event.py
index 193a8bca7..c50e9679c 100644
--- a/services/shared/canonical_event.py
+++ b/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()
@@ -1100,12 +1105,15 @@ def _epcis_list(epcis_data: Dict[str, Any], key: str) -> List[Any]:
 def _epcis_traceability_lot_code(epcis_data: Dict[str, Any]) -> Tuple[str, Optional[str]]:
     """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
-    LGTIN among the event's own EPCs or class-level quantities (output side for
-    a transformation), then an SGTIN serial, then the flattened top-level keys.
-    Raises ``ValueError`` when nothing identifies the lot: the previous
-    ``"UNKNOWN"`` placeholder fabricated a lot code.
+    SEL-598 / H2: when ``LOT_IDENTITY_LOSSLESS`` is enabled (default), 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: a full class URI among EPCs / quantity classes, then ILMD
+    ``fsma:traceabilityLotCode`` / ``cbvmda:lotNumber`` (same order as
+    ``epcis/extraction._extract_lot_data``), then SGTIN serial, then flat
+    top-level keys. Raises ``ValueError`` when nothing identifies the lot.
     """
     ilmd = epcis_data.get("ilmd")
     if not isinstance(ilmd, dict):
@@ -1124,24 +1132,39 @@ def _epcis_traceability_lot_code(epcis_data: Dict[str, Any]) -> Tuple[str, Optio
     ]
     gtin = next((g for g in (_epc_gtin(e) for e in epcs + classes) if g), None)
 
+    # 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():
+            identity = normalize_epcis_lot(value, gtin=gtin)
+            if identity:
+                return identity.canonical_id, identity.gtin or gtin
             return str(value).strip(), gtin
-    for epc in epcs + classes:
-        if "lgtin" in str(epc).lower():
-            segment = _epc_last_segment(epc)
-            if segment:
-                return segment, gtin
+
     for epc in epcs:
         if "sgtin" in str(epc).lower():
+            identity = normalize_epcis_lot(epc, gtin=gtin)
+            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():
+            identity = normalize_epcis_lot(value, gtin=gtin)
+            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"
@@ -1161,8 +1184,15 @@ def _epcis_transformation_inputs(epcis_data: Dict[str, Any]) -> List[Dict[str, A
     for element in _epcis_list(epcis_data, "inputQuantityList"):
         if not isinstance(element, dict):
             continue
-        lot = _epc_last_segment(element.get("epcClass"))
-        if not lot or lot in seen:
+        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
+        if not lot_token or lot_token in seen:
             continue
         raw_quantity = element.get("quantity")
         try:
@@ -1171,18 +1201,29 @@ def _epcis_transformation_inputs(epcis_data: Dict[str, Any]) -> List[Dict[str, A
             quantity = None
         inputs.append(
             {
-                "traceability_lot_code": lot,
+                "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": element.get("epcClass"),
+                "epc_class": epc_class,
             }
         )
-        seen.add(lot)
+        seen.add(lot_token)
     for epc in _epcis_list(epcis_data, "inputEPCList"):
-        segment = _epc_last_segment(epc)
+        identity = normalize_epcis_lot(epc)
+        segment = identity.lot_token if identity else _epc_last_segment(epc)
         if not segment or segment in seen:
             continue
-        inputs.append({"traceability_lot_code": segment, "quantity": 1.0, "unit_of_measure": "each", "epc": str(epc)})
+        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(segment)
     return inputs
 
diff --git a/services/shared/epcis_lot_identity.py b/services/shared/epcis_lot_identity.py
new file mode 100644
index 000000000..c609e6af9
--- /dev/null
+++ b/services/shared/epcis_lot_identity.py
@@ -0,0 +1,240 @@
+"""SEL-598 lossless EPCIS lot identity (H2).
+
+Canonical stored identity is the full LGTIN or GS1 Digital Link (AI 01 + 10).
+The bare lot token is derived for UI, gap reports, and export rebuild paths.
+Round-trip acceptance is exact equality on the canonical full ID.
+
+Feature flag: ``LOT_IDENTITY_LOSSLESS`` (default enabled). Set to ``0`` /
+``false`` to restore segment-only canonical writes (rollback).
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from dataclasses import dataclass
+from typing import Any, Optional
+from urllib.parse import unquote
+
+_LGTIN_RE = re.compile(r"^urn:epc:class:lgtin:(\d+)\.(\d+)\.(.+)$", re.IGNORECASE)
+_SGTIN_RE = re.compile(r"^urn:epc:id:sgtin:(\d+)\.(\d+)\.(.+)$", re.IGNORECASE)
+_DL_CLASS_RE = re.compile(
+    r"^https://id\.gs1\.org/01/(\d{8,14})/10/([^/?#]+)$",
+    re.IGNORECASE,
+)
+
+
+def lossless_enabled() -> bool:
+    raw = (os.environ.get("LOT_IDENTITY_LOSSLESS") or "1").strip().lower()
+    return raw not in {"0", "false", "no", "off"}
+
+
+@dataclass(frozen=True)
+class LotIdentity:
+    """Normalized lot identity for ingest, canonical store, and export."""
+
+    canonical_id: str
+    lot_token: str
+    gtin: Optional[str]
+    is_full_class_uri: bool
+
+
+def _company_item_to_gtin14(company: str, item: str) -> Optional[str]:
+    """Rebuild a 14-digit GTIN body from EPC company.item when lengths sum to 13."""
+    if not company.isdigit() or not item.isdigit():
+        return None
+    if len(company) + len(item) != 13:
+        return None
+    body = company + item
+    # Check digit for GTIN-14 with leading zero indicator already in body length 13
+    digits = "0" + body if len(body) == 13 else body
+    if len(digits) != 14 or not digits.isdigit():
+        return None
+    return digits
+
+
+def _gs1_check_digit(digits13: str) -> str:
+    total = 0
+    for i, ch in enumerate(digits13):
+        n = int(ch)
+        total += n * (3 if (i % 2 == 0) else 1)
+    return str((10 - (total % 10)) % 10)
+
+
+def normalize_gtin14(raw: str) -> Optional[str]:
+    digits = re.sub(r"\D", "", raw or "")
+    if not digits:
+        return None
+    if len(digits) < 8 or len(digits) > 14:
+        return None
+    digits = digits.zfill(14)
+    body, check = digits[:-1], digits[-1]
+    if _gs1_check_digit(body) != check:
+        # Still return zero-padded digits for dual-read rebuild; caller may validate
+        return digits
+    return digits
+
+
+def parse_class_uri(value: object) -> Optional[LotIdentity]:
+    """Parse a full LGTIN or Digital Link class URI into LotIdentity."""
+    if not isinstance(value, str):
+        return None
+    text = value.strip()
+    if not text or "0000000.000000" in text:
+        return None
+
+    m = _LGTIN_RE.match(text)
+    if m:
+        company, item, lot_raw = m.group(1), m.group(2), m.group(3)
+        lot = unquote(lot_raw)
+        if not lot:
+            return None
+        gtin = _company_item_to_gtin14(company, item)
+        return LotIdentity(
+            canonical_id=text,
+            lot_token=lot,
+            gtin=gtin,
+            is_full_class_uri=True,
+        )
+
+    m = _DL_CLASS_RE.match(text)
+    if m:
+        gtin_raw, lot_raw = m.group(1), m.group(2)
+        lot = unquote(lot_raw).strip()
+        if not lot:
+            return None
+        gtin = normalize_gtin14(gtin_raw) or gtin_raw.zfill(14)
+        return LotIdentity(
+            canonical_id=text,
+            lot_token=lot,
+            gtin=gtin,
+            is_full_class_uri=True,
+        )
+
+    return None
+
+
+def is_full_class_uri(value: object) -> bool:
+    parsed = parse_class_uri(value)
+    return bool(parsed and parsed.is_full_class_uri)
+
+
+def lot_token_only(value: object) -> Optional[str]:
+    """Derived bare lot segment; never treat a full class URI as the token."""
+    parsed = parse_class_uri(value)
+    if parsed:
+        return parsed.lot_token
+    if not isinstance(value, str):
+        return None
+    text = value.strip()
+    if not text:
+        return None
+    # Legacy LGTIN/SGTIN segment helper for non-class schemes already truncated
+    lowered = text.lower()
+    if "lgtin" in lowered or "sgtin" in lowered:
+        parts = text.rsplit(".", 1)
+        if len(parts) == 2 and parts[1]:
+            return unquote(parts[1])
+        return None
+    # Reject values that look like URIs but failed class parse
+    if "://" in text or text.lower().startswith("urn:epc:"):
+        return None
+    return text
+
+
+def normalize_epcis_lot(
+    value: object,
+    *,
+    gtin: Optional[str] = None,
+    prefer_full: Optional[bool] = None,
+) -> Optional[LotIdentity]:
+    """Normalize any lot-ish value into LotIdentity.
+
+    When ``prefer_full`` is None, uses ``lossless_enabled()``.
+    Dual-read: bare lot + GTIN rebuilds a Digital Link canonical ID.
+    """
+    use_full = lossless_enabled() if prefer_full is None else prefer_full
+
+    parsed = parse_class_uri(value)
+    if parsed:
+        if use_full:
+            return parsed
+        return LotIdentity(
+            canonical_id=parsed.lot_token,
+            lot_token=parsed.lot_token,
+            gtin=parsed.gtin or (normalize_gtin14(gtin) if gtin else None),
+            is_full_class_uri=False,
+        )
+
+    # SGTIN is instance-level; expose serial as token but not a class URI
+    if isinstance(value, str):
+        sm = _SGTIN_RE.match(value.strip())
+        if sm:
+            serial = unquote(sm.group(3))
+            g = _company_item_to_gtin14(sm.group(1), sm.group(2))
+            return LotIdentity(
+                canonical_id=serial if use_full else serial,
+                lot_token=serial,
+                gtin=g or (normalize_gtin14(gtin) if gtin else None),
+                is_full_class_uri=False,
+            )
+
+    token = lot_token_only(value)
+    if not token:
+        return None
+
+    gtin14 = normalize_gtin14(gtin) if gtin else None
+    if use_full and gtin14:
+        # Rebuild Digital Link; do not invent GTIN when absent
+        from urllib.parse import quote
+
+        canonical = f"https://id.gs1.org/01/{gtin14}/10/{quote(token, safe='-._~')}"
+        return LotIdentity(
+            canonical_id=canonical,
+            lot_token=token,
+            gtin=gtin14,
+            is_full_class_uri=True,
+        )
+
+    return LotIdentity(
+        canonical_id=token,
+        lot_token=token,
+        gtin=gtin14,
+        is_full_class_uri=False,
+    )
+
+
+def resolve_stored_lot(
+    *,
+    epc_or_class: Any = None,
+    ilmd_tlc: Any = None,
+    flat_tlc: Any = None,
+    gtin: Optional[str] = None,
+) -> LotIdentity:
+    """Resolve identity with SEL-598 precedence for canonical write.
+
+    Prefer a full class URI from EPC/quantity class, then ILMD TLC when it is
+    already a full URI, then rebuild from GTIN + derived token, else bare token.
+    """
+    for candidate in (epc_or_class, ilmd_tlc, flat_tlc):
+        parsed = parse_class_uri(candidate)
+        if parsed and lossless_enabled():
+            return parsed
+        if parsed and not lossless_enabled():
+            return LotIdentity(
+                canonical_id=parsed.lot_token,
+                lot_token=parsed.lot_token,
+                gtin=parsed.gtin or (normalize_gtin14(gtin) if gtin else None),
+                is_full_class_uri=False,
+            )
+
+    # ILMD / flat may carry bare lot while EPC carries the class URI (handled above)
+    for candidate in (ilmd_tlc, flat_tlc, epc_or_class):
+        identity = normalize_epcis_lot(candidate, gtin=gtin)
+        if identity:
+            return identity
+
+    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"
+    )
diff --git a/tests/shared/test_epcis_lot_identity_sel598.py b/tests/shared/test_epcis_lot_identity_sel598.py
new file mode 100644
index 000000000..829256a22
--- /dev/null
+++ b/tests/shared/test_epcis_lot_identity_sel598.py
@@ -0,0 +1,157 @@
+"""SEL-598 / H2: lossless EPCIS lot identity (structural + synthetic URIs).
+
+Synthetic identifiers only (A-2165). No partner data. No compliance claims.
+"""
+
+from __future__ import annotations
+
+import os
+
+import pytest
+from shared.canonical_event import (
+    _epcis_traceability_lot_code,
+    _epcis_transformation_inputs,
+)
+from shared.epcis_lot_identity import (
+    is_full_class_uri,
+    lossless_enabled,
+    lot_token_only,
+    normalize_epcis_lot,
+    parse_class_uri,
+)
+
+# Synthetic GS1-shaped IDs (not partner data). GTIN 00614141123452 check digit
+# is illustrative; normalize_epcis_lot rebuilds Digital Links from digits.
+SYN_GTIN14 = "00614141123452"
+SYN_LOT = "LOT-ROM-2026-0312A"
+SYN_LGTIN = f"urn:epc:class:lgtin:0614141.123452.{SYN_LOT}"
+SYN_DL = f"https://id.gs1.org/01/{SYN_GTIN14}/10/{SYN_LOT}"
+SYN_LGTIN_OTHER_GTIN = f"urn:epc:class:lgtin:0614141.999990.{SYN_LOT}"
+
+
+def test_lgtin_canonical_not_truncated_on_parse():
+    parsed = parse_class_uri(SYN_LGTIN)
+    assert parsed is not None
+    assert parsed.is_full_class_uri is True
+    assert parsed.canonical_id == SYN_LGTIN
+    assert parsed.lot_token == SYN_LOT
+    assert parsed.canonical_id != SYN_LOT
+
+
+def test_digital_link_canonical_not_truncated_on_parse():
+    parsed = parse_class_uri(SYN_DL)
+    assert parsed is not None
+    assert parsed.canonical_id == SYN_DL
+    assert parsed.lot_token == SYN_LOT
+
+
+def test_full_uri_not_passed_as_lot_arg_to_epc_class():
+    """lot_token_only must refuse to treat a class URI as the lot segment."""
+    assert lot_token_only(SYN_DL) == SYN_LOT
+    assert lot_token_only(SYN_LGTIN) == SYN_LOT
+    assert is_full_class_uri(SYN_DL)
+    # Rebuild path: normalize with GTIN + accidental full URI still yields token
+    identity = normalize_epcis_lot(SYN_DL, gtin=SYN_GTIN14)
+    assert identity is not None
+    assert identity.lot_token == SYN_LOT
+    assert "://" not in identity.lot_token
+    assert not identity.lot_token.lower().startswith("urn:epc:")
+
+
+def test_same_lot_different_gtin_are_distinct():
+    a = parse_class_uri(SYN_LGTIN)
+    b = parse_class_uri(SYN_LGTIN_OTHER_GTIN)
+    assert a is not None and b is not None
+    assert a.lot_token == b.lot_token == SYN_LOT
+    assert a.canonical_id != b.canonical_id
+
+
+def test_both_pass_fixture_round_trip_full_id():
+    """Synthetic both-pass: full Digital Link survives normalize → token → rebuild."""
+    first = normalize_epcis_lot(SYN_DL)
+    assert first is not None
+    assert first.canonical_id == SYN_DL
+    rebuilt = normalize_epcis_lot(first.lot_token, gtin=first.gtin or SYN_GTIN14)
+    assert rebuilt is not None
+    assert rebuilt.is_full_class_uri is True
+    # Exact equality on full Digital Link form
+    assert rebuilt.canonical_id == SYN_DL or rebuilt.lot_token == SYN_LOT and rebuilt.gtin == first.gtin
+
+
+def test_lenient_only_fixture_preserves_raw_tlc_and_gtin():
+    """Synthetic lenient-only: bare TLC + GTIN rebuilds without inventing GTIN."""
+    bare = normalize_epcis_lot(SYN_LOT, gtin=SYN_GTIN14)
+    assert bare is not None
+    assert bare.lot_token == SYN_LOT
+    assert bare.gtin is not None
+    assert SYN_LOT in bare.canonical_id
+    # Without GTIN, do not invent a class URI
+    no_gtin = normalize_epcis_lot(SYN_LOT, gtin=None)
+    assert no_gtin is not None
+    assert no_gtin.canonical_id == SYN_LOT
+    assert no_gtin.is_full_class_uri is False
+
+
+def test_canonical_path_preserves_full_lgtin_when_lossless(monkeypatch):
+    monkeypatch.delenv("LOT_IDENTITY_LOSSLESS", raising=False)
+    assert lossless_enabled() is True
+    event = {
+        "quantityList": [{"epcClass": SYN_LGTIN, "quantity": 1, "uom": "EA"}],
+        "ilmd": {},
+    }
+    tlc, gtin = _epcis_traceability_lot_code(event)
+    assert tlc == SYN_LGTIN
+    assert tlc != SYN_LOT
+
+
+def test_canonical_path_segment_only_when_flag_off(monkeypatch):
+    monkeypatch.setenv("LOT_IDENTITY_LOSSLESS", "0")
+    assert lossless_enabled() is False
+    event = {
+        "quantityList": [{"epcClass": SYN_LGTIN, "quantity": 1, "uom": "EA"}],
+        "ilmd": {},
+    }
+    tlc, _gtin = _epcis_traceability_lot_code(event)
+    assert tlc == SYN_LOT
+
+
+def test_transformation_inputs_keep_full_class_uri(monkeypatch):
+    monkeypatch.delenv("LOT_IDENTITY_LOSSLESS", raising=False)
+    event = {
+        "inputQuantityList": [
+            {"epcClass": SYN_LGTIN, "quantity": 2.5, "uom": "KGM"},
+        ],
+    }
+    inputs = _epcis_transformation_inputs(event)
+    assert len(inputs) == 1
+    assert inputs[0]["traceability_lot_code"] == SYN_LGTIN
+    assert inputs[0]["lot_token"] == SYN_LOT
+    assert inputs[0]["epc_class"] == SYN_LGTIN
+
+
+def test_export_epc_class_emits_stored_full_uri_unchanged():
+    from ingestion.app.epcis_export import _epc_class
+
+    row = {
+        "traceability_lot_code": SYN_DL,
+        "gtin": SYN_GTIN14,
+    }
+    assert _epc_class(row) == SYN_DL
+
+
+def test_export_epc_class_does_not_embed_full_uri_as_lot():
+    from ingestion.app.epcis_export import _epc_class
+
+    # Stored TLC accidentally holds a full URI while epc_class is empty;
+    # rebuild must not produce .../10/https%3A%2F%2F...
+    row = {
+        "traceability_lot_code": SYN_DL,
+        "gtin": SYN_GTIN14,
+        "epc_class": None,
+        "product_id": None,
+    }
+    out = _epc_class(row)
+    assert out == SYN_DL
+    assert out is not None
+    assert "%3A" not in out
+    assert out.count("https://") == 1

From ebd8ea3b1cd3c29431b66fdb4fa26d4d3873decc Mon Sep 17 00:00:00 2001
From: Christopher Sellers <1@christophersellers.com>
Date: Wed, 28 Oct 2026 12:36:21 -0700
Subject: [PATCH 2/3] fix(shared): SEL-598 review findings for lossless lot
 identity

Reconstruct GTIN-14 from EPC company.item with indicator + check digit;
preserve bare ILMD/flat TLC (no fabricated Digital Link from SGTIN GTIN);
export only via _valid_class_uri; dedupe transformation inputs by lot_key;
tighten round-trip and add regression tests.

Signed-off-by: Christopher Sellers <1@christophersellers.com>
---
 services/ingestion/app/epcis_export.py        | 14 ++---
 services/shared/canonical_event.py            | 42 ++++++++++-----
 services/shared/epcis_lot_identity.py         | 20 ++++---
 .../shared/test_epcis_lot_identity_sel598.py  | 54 ++++++++++++++++++-
 4 files changed, 101 insertions(+), 29 deletions(-)

diff --git a/services/ingestion/app/epcis_export.py b/services/ingestion/app/epcis_export.py
index 7f18e1dbb..f28ab7f87 100644
--- a/services/ingestion/app/epcis_export.py
+++ b/services/ingestion/app/epcis_export.py
@@ -19,7 +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, parse_class_uri
+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
@@ -345,14 +345,10 @@ def _epc_class(row: dict) -> str | None:
     """
     for key in ("epc_class", "product_id", "traceability_lot_code", "canonical_epc_class"):
         value = _kde_scalar(row.get(key))
-        if _valid_class_uri(value) or is_full_class_uri(value):
-            # Prefer the validated export shape when available; otherwise the
-            # parsed canonical URI from the shared normalizer.
-            if _valid_class_uri(value):
-                return value
-            parsed = parse_class_uri(value)
-            if parsed:
-                return parsed.canonical_id
+        # 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")
     if raw:
         try:
diff --git a/services/shared/canonical_event.py b/services/shared/canonical_event.py
index c50e9679c..b33c48b9e 100644
--- a/services/shared/canonical_event.py
+++ b/services/shared/canonical_event.py
@@ -1110,10 +1110,14 @@ def _epcis_traceability_lot_code(epcis_data: Dict[str, Any]) -> Tuple[str, Optio
     class URI when present. The bare lot token remains available via
     ``lot_token_only`` / ``normalize_epcis_lot(...).lot_token``.
 
-    Precedence: a full class URI among EPCs / quantity classes, then ILMD
-    ``fsma:traceabilityLotCode`` / ``cbvmda:lotNumber`` (same order as
-    ``epcis/extraction._extract_lot_data``), then SGTIN serial, then flat
-    top-level keys. Raises ``ValueError`` when nothing identifies the lot.
+    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.
     """
     ilmd = epcis_data.get("ilmd")
     if not isinstance(ilmd, dict):
@@ -1130,7 +1134,17 @@ class URI when present. The bare lot token remains available via
         )
         if isinstance(q, dict) and q.get("epcClass")
     ]
-    gtin = next((g for g in (_epc_gtin(e) for e in epcs + classes) if g), None)
+    # 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):
@@ -1143,14 +1157,15 @@ class URI when present. The bare lot token remains available via
     for key in ("fsma:traceabilityLotCode", "traceabilityLotCode", "cbvmda:lotNumber", "lotNumber"):
         value = ilmd.get(key)
         if value is not None and str(value).strip():
-            identity = normalize_epcis_lot(value, gtin=gtin)
+            # 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, gtin=gtin)
+            identity = normalize_epcis_lot(epc)
             if identity:
                 return identity.canonical_id, identity.gtin or gtin
             segment = _epc_last_segment(epc)
@@ -1160,7 +1175,8 @@ class URI when present. The bare lot token remains available via
     for key in ("traceability_lot_code", "lotNumber"):
         value = epcis_data.get(key)
         if value is not None and str(value).strip():
-            identity = normalize_epcis_lot(value, gtin=gtin)
+            # 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
@@ -1192,7 +1208,8 @@ def _epcis_transformation_inputs(epcis_data: Dict[str, Any]) -> List[Dict[str, A
         else:
             lot_token = _epc_last_segment(epc_class)
             lot_key = lot_token
-        if not lot_token or lot_token in seen:
+        dedup_key = ("class", lot_key)
+        if not lot_token or dedup_key in seen:
             continue
         raw_quantity = element.get("quantity")
         try:
@@ -1208,11 +1225,12 @@ def _epcis_transformation_inputs(epcis_data: Dict[str, Any]) -> List[Dict[str, A
                 "epc_class": epc_class,
             }
         )
-        seen.add(lot_token)
+        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)
-        if not segment or segment in seen:
+        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(
@@ -1224,7 +1242,7 @@ def _epcis_transformation_inputs(epcis_data: Dict[str, Any]) -> List[Dict[str, A
                 "epc": str(epc),
             }
         )
-        seen.add(segment)
+        seen.add(dedup_key)
     return inputs
 
 
diff --git a/services/shared/epcis_lot_identity.py b/services/shared/epcis_lot_identity.py
index c609e6af9..e4f831322 100644
--- a/services/shared/epcis_lot_identity.py
+++ b/services/shared/epcis_lot_identity.py
@@ -40,17 +40,23 @@ class LotIdentity:
 
 
 def _company_item_to_gtin14(company: str, item: str) -> Optional[str]:
-    """Rebuild a 14-digit GTIN body from EPC company.item when lengths sum to 13."""
-    if not company.isdigit() or not item.isdigit():
+    """Rebuild GTIN-14 from EPC Pure Identity company.item (SGTIN/LGTIN).
+
+    Company prefix + item reference must sum to 13 digits. The first digit of
+    the item reference is the Indicator; the remaining item digits follow the
+    company prefix; then append the GS1 check digit.
+    Example: ``0614141.123452`` → ``10614141234520`` (not ``00614141123452``).
+    """
+    if not company.isdigit() or not item.isdigit() or not item:
         return None
     if len(company) + len(item) != 13:
         return None
-    body = company + item
-    # Check digit for GTIN-14 with leading zero indicator already in body length 13
-    digits = "0" + body if len(body) == 13 else body
-    if len(digits) != 14 or not digits.isdigit():
+    indicator = item[0]
+    item_without_indicator = item[1:]
+    digits13 = indicator + company + item_without_indicator
+    if len(digits13) != 13:
         return None
-    return digits
+    return digits13 + _gs1_check_digit(digits13)
 
 
 def _gs1_check_digit(digits13: str) -> str:
diff --git a/tests/shared/test_epcis_lot_identity_sel598.py b/tests/shared/test_epcis_lot_identity_sel598.py
index 829256a22..21f43591c 100644
--- a/tests/shared/test_epcis_lot_identity_sel598.py
+++ b/tests/shared/test_epcis_lot_identity_sel598.py
@@ -23,6 +23,8 @@
 # Synthetic GS1-shaped IDs (not partner data). GTIN 00614141123452 check digit
 # is illustrative; normalize_epcis_lot rebuilds Digital Links from digits.
 SYN_GTIN14 = "00614141123452"
+# EPC Pure Identity 0614141.123452 → indicator 1 + check digit
+SYN_GTIN_FROM_LGTIN = "10614141234520"
 SYN_LOT = "LOT-ROM-2026-0312A"
 SYN_LGTIN = f"urn:epc:class:lgtin:0614141.123452.{SYN_LOT}"
 SYN_DL = f"https://id.gs1.org/01/{SYN_GTIN14}/10/{SYN_LOT}"
@@ -75,7 +77,7 @@ def test_both_pass_fixture_round_trip_full_id():
     assert rebuilt is not None
     assert rebuilt.is_full_class_uri is True
     # Exact equality on full Digital Link form
-    assert rebuilt.canonical_id == SYN_DL or rebuilt.lot_token == SYN_LOT and rebuilt.gtin == first.gtin
+    assert rebuilt.canonical_id == SYN_DL
 
 
 def test_lenient_only_fixture_preserves_raw_tlc_and_gtin():
@@ -155,3 +157,53 @@ def test_export_epc_class_does_not_embed_full_uri_as_lot():
     assert out is not None
     assert "%3A" not in out
     assert out.count("https://") == 1
+
+
+def test_lgtin_company_item_rebuilds_indicator_and_check_digit():
+    parsed = parse_class_uri(SYN_LGTIN)
+    assert parsed is not None
+    assert parsed.gtin == SYN_GTIN_FROM_LGTIN
+    # Must not use the naive 0+company+item concatenation
+    assert parsed.gtin != "00614141123452"
+
+
+def test_bare_ilmd_tlc_preserved_with_sgtin_epc(monkeypatch):
+    """SGTIN + bare ILMD lot must not fabricate a Digital Link TLC."""
+    monkeypatch.delenv("LOT_IDENTITY_LOSSLESS", raising=False)
+    sgtin = "urn:epc:id:sgtin:0614141.123452.SERIAL9"
+    event = {
+        "epcList": [sgtin],
+        "ilmd": {"cbvmda:lotNumber": SYN_LOT},
+    }
+    tlc, gtin = _epcis_traceability_lot_code(event)
+    assert tlc == SYN_LOT
+    assert "://" not in tlc
+    assert gtin == SYN_GTIN_FROM_LGTIN
+
+
+def test_transformation_inputs_same_lot_different_gtin_kept(monkeypatch):
+    monkeypatch.delenv("LOT_IDENTITY_LOSSLESS", raising=False)
+    event = {
+        "inputQuantityList": [
+            {"epcClass": SYN_LGTIN, "quantity": 1, "uom": "EA"},
+            {"epcClass": SYN_LGTIN_OTHER_GTIN, "quantity": 2, "uom": "EA"},
+        ],
+    }
+    inputs = _epcis_transformation_inputs(event)
+    assert len(inputs) == 2
+    codes = {i["traceability_lot_code"] for i in inputs}
+    assert codes == {SYN_LGTIN, SYN_LGTIN_OTHER_GTIN}
+
+
+def test_export_rejects_invalid_class_uri_from_lenient_parse():
+    from ingestion.app.epcis_export import _epc_class
+
+    # Bad company+item length — lenient parser may accept; export must not emit.
+    bad = "urn:epc:class:lgtin:1.2.LOT1"
+    row = {"traceability_lot_code": bad, "gtin": None, "epc_class": None, "product_id": None}
+    assert _epc_class(row) is None
+
+    # Digital Link with invalid check digit
+    bad_dl = "https://id.gs1.org/01/09506000134351/10/LOT1"
+    row2 = {"traceability_lot_code": bad_dl, "gtin": None, "epc_class": None, "product_id": None}
+    assert _epc_class(row2) is None

From 16fc6e7b3bb141b5cea11018b81fc4385a2ed894 Mon Sep 17 00:00:00 2001
From: Christopher Sellers <1@christophersellers.com>
Date: Mon, 2 Nov 2026 23:01:58 -0800
Subject: [PATCH 3/3] fix(shared): make SEL-598 lossless lot identity opt-in
 and dual-read export lots

- LOT_IDENTITY_LOSSLESS now defaults to off. With the flag off, the
  canonical TLC and transformation-input resolvers are the pre-H2 code
  unchanged, so the stored TLC, GTIN KDE, event hash and idempotency key
  do not move and re-ingest does not duplicate events. Turning the flag
  on still needs a dual-key or migration path first (documented).
- EPCIS export lot filter reads both shapes: a bare lot also matches
  full LGTIN / GS1 Digital Link class URIs carrying it, and a URI filter
  also matches the bare historical row. Rows are re-checked by exact lot
  token after the coarse SQL match.
- lot_token_only only applies the legacy LGTIN/SGTIN segment split to
  urn:epc: values, so bare lots containing "lgtin"/"sgtin" are kept.
- Tests: flag-off parity with the pre-H2 resolvers, opt-in default,
  bare-lot keyword cases, export dual-read filter.

Addresses Grok FAIL on #3737 (738a4c7c).

Signed-off-by: Christopher Sellers <1@christophersellers.com>
---
 services/ingestion/app/epcis_export.py        |  59 +++++++++-
 .../ingestion/tests/test_epcis_export_1342.py |   6 +-
 services/shared/canonical_event.py            | 104 +++++++++++++++++-
 services/shared/epcis_lot_identity.py         |  22 ++--
 .../shared/test_epcis_lot_identity_sel598.py  |  83 +++++++++++++-
 5 files changed, 257 insertions(+), 17 deletions(-)

diff --git a/services/ingestion/app/epcis_export.py b/services/ingestion/app/epcis_export.py
index f965eaf94..cc042eaf5 100644
--- a/services/ingestion/app/epcis_export.py
+++ b/services/ingestion/app/epcis_export.py
@@ -47,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]:
@@ -60,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
@@ -118,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]:
diff --git a/services/ingestion/tests/test_epcis_export_1342.py b/services/ingestion/tests/test_epcis_export_1342.py
index 9437fffac..588adefc5 100644
--- a/services/ingestion/tests/test_epcis_export_1342.py
+++ b/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"),
         ):
diff --git a/services/shared/canonical_event.py b/services/shared/canonical_event.py
index b33c48b9e..a2083cf8d 100644
--- a/services/shared/canonical_event.py
+++ b/services/shared/canonical_event.py
@@ -1102,10 +1102,105 @@ def _epcis_list(epcis_data: Dict[str, Any], key: str) -> List[Any]:
     return list(value) if isinstance(value, list) else []
 
 
+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
+    LGTIN among the event's own EPCs or class-level quantities (output side for
+    a transformation), then an SGTIN serial, then the flattened top-level keys.
+    Raises ``ValueError`` when nothing identifies the lot: the previous
+    ``"UNKNOWN"`` placeholder fabricated a lot code.
+    """
+    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")
+    ]
+    gtin = next((g for g in (_epc_gtin(e) for e in epcs + classes) if g), None)
+
+    for key in ("fsma:traceabilityLotCode", "traceabilityLotCode", "cbvmda:lotNumber", "lotNumber"):
+        value = ilmd.get(key)
+        if value is not None and str(value).strip():
+            return str(value).strip(), gtin
+    for epc in epcs + classes:
+        if "lgtin" in str(epc).lower():
+            segment = _epc_last_segment(epc)
+            if segment:
+                return segment, gtin
+    for epc in epcs:
+        if "sgtin" in str(epc).lower():
+            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():
+            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 _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
+    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.
+    """
+    inputs: List[Dict[str, Any]] = []
+    seen: set = set()
+    for element in _epcis_list(epcis_data, "inputQuantityList"):
+        if not isinstance(element, dict):
+            continue
+        lot = _epc_last_segment(element.get("epcClass"))
+        if not lot or lot 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,
+                "quantity": quantity,
+                "unit_of_measure": (element.get("uom") or "each") if quantity is not None else None,
+                "epc_class": element.get("epcClass"),
+            }
+        )
+        seen.add(lot)
+    for epc in _epcis_list(epcis_data, "inputEPCList"):
+        segment = _epc_last_segment(epc)
+        if not segment or segment in seen:
+            continue
+        inputs.append({"traceability_lot_code": segment, "quantity": 1.0, "unit_of_measure": "each", "epc": str(epc)})
+        seen.add(segment)
+    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 (default), the
+    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``.
@@ -1118,7 +1213,12 @@ class URI when present. The bare lot token remains available via
     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 = {}
@@ -1195,6 +1295,8 @@ def _epcis_transformation_inputs(epcis_data: Dict[str, Any]) -> List[Dict[str, A
     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"):
diff --git a/services/shared/epcis_lot_identity.py b/services/shared/epcis_lot_identity.py
index e4f831322..f26d0feb5 100644
--- a/services/shared/epcis_lot_identity.py
+++ b/services/shared/epcis_lot_identity.py
@@ -4,8 +4,14 @@
 The bare lot token is derived for UI, gap reports, and export rebuild paths.
 Round-trip acceptance is exact equality on the canonical full ID.
 
-Feature flag: ``LOT_IDENTITY_LOSSLESS`` (default enabled). Set to ``0`` /
-``false`` to restore segment-only canonical writes (rollback).
+Feature flag: ``LOT_IDENTITY_LOSSLESS`` (default disabled). Set to ``1`` /
+``true`` to store the full class URI as ``traceability_lot_code``.
+
+Keep it off until a dual-key or migration path exists for the event hash and
+idempotency key. Both include the TLC, so switching a lot from its bare token
+to the full URI changes them, and re-ingesting events stored before the switch
+would create duplicates. With the flag off, canonical writes are the same as
+before H2.
 """
 
 from __future__ import annotations
@@ -25,8 +31,8 @@
 
 
 def lossless_enabled() -> bool:
-    raw = (os.environ.get("LOT_IDENTITY_LOSSLESS") or "1").strip().lower()
-    return raw not in {"0", "false", "no", "off"}
+    raw = (os.environ.get("LOT_IDENTITY_LOSSLESS") or "0").strip().lower()
+    return raw in {"1", "true", "yes", "on"}
 
 
 @dataclass(frozen=True)
@@ -135,15 +141,17 @@ def lot_token_only(value: object) -> Optional[str]:
     text = value.strip()
     if not text:
         return None
-    # Legacy LGTIN/SGTIN segment helper for non-class schemes already truncated
     lowered = text.lower()
-    if "lgtin" in lowered or "sgtin" in lowered:
+    # Legacy LGTIN/SGTIN segment helper, only for EPC URNs the strict class
+    # parse rejected. A bare lot that merely contains "lgtin"/"sgtin" is a
+    # lot token, not a URI, and must not be dropped.
+    if lowered.startswith("urn:epc:") and ("lgtin" in lowered or "sgtin" in lowered):
         parts = text.rsplit(".", 1)
         if len(parts) == 2 and parts[1]:
             return unquote(parts[1])
         return None
     # Reject values that look like URIs but failed class parse
-    if "://" in text or text.lower().startswith("urn:epc:"):
+    if "://" in text or lowered.startswith("urn:epc:"):
         return None
     return text
 
diff --git a/tests/shared/test_epcis_lot_identity_sel598.py b/tests/shared/test_epcis_lot_identity_sel598.py
index 21f43591c..1add71773 100644
--- a/tests/shared/test_epcis_lot_identity_sel598.py
+++ b/tests/shared/test_epcis_lot_identity_sel598.py
@@ -11,6 +11,8 @@
 from shared.canonical_event import (
     _epcis_traceability_lot_code,
     _epcis_transformation_inputs,
+    _legacy_epcis_traceability_lot_code,
+    _legacy_epcis_transformation_inputs,
 )
 from shared.epcis_lot_identity import (
     is_full_class_uri,
@@ -31,6 +33,13 @@
 SYN_LGTIN_OTHER_GTIN = f"urn:epc:class:lgtin:0614141.999990.{SYN_LOT}"
 
 
+
+@pytest.fixture(autouse=True)
+def _lossless_on(monkeypatch):
+    """The H2 behaviour under test is opt-in; tests that need it off say so."""
+    monkeypatch.setenv("LOT_IDENTITY_LOSSLESS", "1")
+
+
 def test_lgtin_canonical_not_truncated_on_parse():
     parsed = parse_class_uri(SYN_LGTIN)
     assert parsed is not None
@@ -95,7 +104,7 @@ def test_lenient_only_fixture_preserves_raw_tlc_and_gtin():
 
 
 def test_canonical_path_preserves_full_lgtin_when_lossless(monkeypatch):
-    monkeypatch.delenv("LOT_IDENTITY_LOSSLESS", raising=False)
+    monkeypatch.setenv("LOT_IDENTITY_LOSSLESS", "1")
     assert lossless_enabled() is True
     event = {
         "quantityList": [{"epcClass": SYN_LGTIN, "quantity": 1, "uom": "EA"}],
@@ -118,7 +127,7 @@ def test_canonical_path_segment_only_when_flag_off(monkeypatch):
 
 
 def test_transformation_inputs_keep_full_class_uri(monkeypatch):
-    monkeypatch.delenv("LOT_IDENTITY_LOSSLESS", raising=False)
+    monkeypatch.setenv("LOT_IDENTITY_LOSSLESS", "1")
     event = {
         "inputQuantityList": [
             {"epcClass": SYN_LGTIN, "quantity": 2.5, "uom": "KGM"},
@@ -169,7 +178,7 @@ def test_lgtin_company_item_rebuilds_indicator_and_check_digit():
 
 def test_bare_ilmd_tlc_preserved_with_sgtin_epc(monkeypatch):
     """SGTIN + bare ILMD lot must not fabricate a Digital Link TLC."""
-    monkeypatch.delenv("LOT_IDENTITY_LOSSLESS", raising=False)
+    monkeypatch.setenv("LOT_IDENTITY_LOSSLESS", "1")
     sgtin = "urn:epc:id:sgtin:0614141.123452.SERIAL9"
     event = {
         "epcList": [sgtin],
@@ -182,7 +191,7 @@ def test_bare_ilmd_tlc_preserved_with_sgtin_epc(monkeypatch):
 
 
 def test_transformation_inputs_same_lot_different_gtin_kept(monkeypatch):
-    monkeypatch.delenv("LOT_IDENTITY_LOSSLESS", raising=False)
+    monkeypatch.setenv("LOT_IDENTITY_LOSSLESS", "1")
     event = {
         "inputQuantityList": [
             {"epcClass": SYN_LGTIN, "quantity": 1, "uom": "EA"},
@@ -207,3 +216,69 @@ def test_export_rejects_invalid_class_uri_from_lenient_parse():
     bad_dl = "https://id.gs1.org/01/09506000134351/10/LOT1"
     row2 = {"traceability_lot_code": bad_dl, "gtin": None, "epc_class": None, "product_id": None}
     assert _epc_class(row2) is None
+
+
+def test_lossless_is_opt_in_by_default(monkeypatch):
+    monkeypatch.delenv("LOT_IDENTITY_LOSSLESS", raising=False)
+    assert lossless_enabled() is False
+    monkeypatch.setenv("LOT_IDENTITY_LOSSLESS", "true")
+    assert lossless_enabled() is True
+
+
+@pytest.mark.parametrize(
+    "event",
+    [
+        {"quantityList": [{"epcClass": SYN_LGTIN, "quantity": 1, "uom": "EA"}], "ilmd": {}},
+        {"epcList": ["urn:epc:id:sgtin:0614141.123452.SERIAL9"], "ilmd": {"cbvmda:lotNumber": SYN_LOT}},
+        {"quantityList": [{"epcClass": SYN_LGTIN_OTHER_GTIN, "quantity": 1}], "ilmd": {"fsma:traceabilityLotCode": "TLC-ILMD-1"}},
+        {"traceability_lot_code": " FLAT-LOT-9 "},
+    ],
+)
+def test_flag_off_resolver_is_pre_h2_unchanged(monkeypatch, event):
+    """Flag off must keep TLC and GTIN KDE (and so hash/idempotency) as before H2."""
+    monkeypatch.setenv("LOT_IDENTITY_LOSSLESS", "0")
+    assert _epcis_traceability_lot_code(event) == _legacy_epcis_traceability_lot_code(event)
+
+
+def test_flag_off_transformation_inputs_are_pre_h2_unchanged(monkeypatch):
+    monkeypatch.setenv("LOT_IDENTITY_LOSSLESS", "0")
+    event = {
+        "inputQuantityList": [
+            {"epcClass": SYN_LGTIN, "quantity": 1, "uom": "EA"},
+            {"epcClass": SYN_LGTIN_OTHER_GTIN, "quantity": 2, "uom": "EA"},
+        ],
+        "inputEPCList": ["urn:epc:id:sgtin:0614141.123452.SERIAL9"],
+    }
+    assert _epcis_transformation_inputs(event) == _legacy_epcis_transformation_inputs(event)
+
+
+@pytest.mark.parametrize("bare", ["SGTINLOT-7", "PLGTIN.A", "lot-lgtin"])
+def test_lot_token_only_keeps_bare_lot_containing_gtin_words(bare):
+    assert lot_token_only(bare) == bare
+
+
+def test_export_lot_filter_bare_lot_reads_both_shapes():
+    from ingestion.app.epcis_export import _lot_code_filter, _row_matches_lot
+
+    predicate, params = _lot_code_filter(SYN_LOT)
+    assert "e.traceability_lot_code = :lot_code" in predicate
+    assert f"/10/{SYN_LOT}" in params.values()
+    assert f".{SYN_LOT}" in params.values()
+    assert _row_matches_lot(SYN_LOT, SYN_LOT)
+    assert _row_matches_lot(SYN_DL, SYN_LOT)
+    assert _row_matches_lot(SYN_LGTIN, SYN_LOT)
+    # Same suffix, different lot: excluded after the coarse SQL match.
+    assert not _row_matches_lot(f"urn:epc:class:lgtin:0614141.123452.X.{SYN_LOT}", SYN_LOT)
+    assert not _row_matches_lot("OTHER-LOT", SYN_LOT)
+
+
+def test_export_lot_filter_full_uri_also_reads_bare_rows():
+    from ingestion.app.epcis_export import _lot_code_filter, _row_matches_lot
+
+    predicate, params = _lot_code_filter(SYN_DL)
+    assert params == {"lot_code": SYN_DL, "lot_token": SYN_LOT}
+    assert "e.traceability_lot_code = :lot_token" in predicate
+    assert _row_matches_lot(SYN_DL, SYN_DL)
+    assert _row_matches_lot(SYN_LOT, SYN_DL)
+    # A different GTIN's URI with the same lot is not the requested class.
+    assert not _row_matches_lot(SYN_LGTIN_OTHER_GTIN, SYN_DL)