fix(ingestion): EPCIS Preview, harvest creating_class_instance, generated test IDs

services/ingestion/app/epcis_export.py

@@ -1,9 +1,9 @@
 """
 EPCIS Export Router.
 
-Generates GS1 EPCIS 2.0 JSON-LD exports for retailer compliance
-(Walmart, Kroger, Costco). Also provides FDA 21 CFR 1.1455 sortable
-spreadsheet export format.
+Generates GS1 EPCIS 2.0 JSON-LD exports (Preview) and FDA 21 CFR 1.1455
+sortable spreadsheet export. EPCIS output is labeled Preview; validate with
+your trading partner before relying on it for compliance.
 """
 
 from __future__ import annotations
@@ -152,7 +152,7 @@ def _load_export_rows(tenant_id: str, request: "ExportRequest") -> list[dict]:
     "shipping": "urn:epcglobal:cbv:bizstep:shipping",
     "transformation": "urn:epcglobal:cbv:bizstep:commissioning",
     "initial_packing": "urn:epcglobal:cbv:bizstep:packing",
-    "harvesting": "urn:epcglobal:cbv:bizstep:collecting",
+    "harvesting": "urn:epcglobal:cbv:bizstep:creating_class_instance",
     "cooling": "urn:epcglobal:cbv:bizstep:storing",
     "first_land_based_receiving": "urn:epcglobal:cbv:bizstep:receiving",
 }
@@ -483,9 +483,10 @@ class ExportRequest(BaseModel):
     "/epcis",
     summary="Export in GS1 EPCIS 2.0 JSON-LD format",
     description=(
-        "Generates a GS1 EPCIS 2.0 compliant JSON-LD document containing all "
-        "traceability events. Compatible with Walmart, Kroger, and Costco portals. "
-        "Stored event hashes are included without cryptographic verification."
+        "Generates a GS1 EPCIS 2.0 JSON-LD document containing traceability "
+        "events (Preview). Validate with your trading partner before relying "
+        "on it for compliance. Stored event hashes are included without "
+        "cryptographic verification."
     ),
 )
 async def export_epcis(
@@ -623,6 +624,8 @@ def _skip(row: dict, reason: str) -> None:
                         "valid",
                         "errors",
                         "checkedAt",
+                        "preview",
+                        "previewNotice",
                     )
                 },
             },
@@ -645,6 +648,11 @@ def _skip(row: dict, reason: str) -> None:
             # Loading stored hashes does not verify event content or chain integrity.
             "integrityVerified": False,
             "chainHashVerified": False,
+            "preview": True,
+            "previewNotice": (
+                "EPCIS export is in Preview. Validate with your trading partner "
+                "before relying on it for compliance."
+            ),
             **({"regengine:message": "No data yet"} if not rows else {}),
         },
     }
@@ -680,6 +688,7 @@ def _skip(row: dict, reason: str) -> None:
         "X-RegEngine-Skipped-Events-Count": str(skipped),
         "X-RegEngine-Integrity-Verified": "false",
         "X-RegEngine-Data-Source": data_source,
+        "X-RegEngine-Preview": "true",
     }
     if validate:
         response_headers["X-RegEngine-Schema-Valid"] = "true" if not validation_errors else "false"
@@ -756,6 +765,7 @@ async def export_fda(
         "X-RegEngine-Events-Count": str(event_count),
         "X-RegEngine-Format": "FDA_21CFR1.1455",
         "X-RegEngine-Data-Source": data_source,
+        "X-RegEngine-Preview": "true",
     }
     if data_source == "empty":
         fda_headers["X-RegEngine-Message"] = "No data yet"
@@ -779,15 +789,17 @@ async def list_export_formats():
             {
                 "id": "epcis",
                 "name": "GS1 EPCIS 2.0 JSON-LD",
-                "description": "Industry standard for supply chain event data. Compatible with Walmart, Kroger, Costco.",
+                "description": "EPCIS 2.0 JSON-LD export (Preview).",
                 "content_type": "application/ld+json",
-                "retailers": ["Walmart", "Kroger", "Costco", "Target"],
+                "status": "preview",
+                "retailers": [],
             },
             {
                 "id": "fda",
                 "name": "FDA Sortable Spreadsheet (CSV)",
-                "description": "Sortable electronic format required by 21 CFR 1.1455 for FDA records requests.",
+                "description": "Sortable electronic format for FDA records requests (Preview). Align with 21 CFR 1.1455 after partner validation.",
                 "content_type": "text/csv",
+                "status": "preview",
                 "retailers": ["FDA"],
             },
         ]

services/ingestion/tests/test_epcis_export_gs1_2.py

@@ -34,6 +34,45 @@ def class_uri():
     return f"urn:epc:class:lgtin:{digits[:7]}.{digits[7:]}.{uuid4().hex}"
 
 
+def _independent_gs1_mod10_check_digit(data_digits: str) -> int:
+    """GS1 General Specifications mod-10, written out for the test.
+
+    From the rightmost data digit: weights alternate 3, 1, 3, 1, ...
+    Check digit is (10 - (sum % 10)) % 10. Independent of
+    shared.fsma_validation / exporter._gs1_check_digit_valid.
+    """
+    if not data_digits.isdigit():
+        raise ValueError("data digits must be numeric")
+    total = 0
+    for i, ch in enumerate(reversed(data_digits)):
+        total += (3 if i % 2 == 0 else 1) * int(ch)
+    return (10 - (total % 10)) % 10
+
+
+def _gs1_with_check(data_digits: str) -> str:
+    """Append check digit from the independent mod-10 above."""
+    return data_digits + str(_independent_gs1_mod10_check_digit(data_digits))
+
+
+def generated_gtin14() -> str:
+    base = str(uuid4().int % (10**13)).zfill(13)
+    return _gs1_with_check(base)
+
+
+def generated_gln13() -> str:
+    base = str(uuid4().int % (10**12)).zfill(12)
+    return _gs1_with_check(base)
+
+
+def generated_sscc18() -> str:
+    base = str(uuid4().int % (10**17)).zfill(17)
+    return _gs1_with_check(base)
+
+
+def generated_lot() -> str:
+    return f"SYN-{uuid4().hex[:12]}"
+
+
 def assert_conformant(document):
     jsonschema = pytest.importorskip("jsonschema")
     jsonschema.Draft7Validator(
@@ -84,6 +123,7 @@ def test_empty_export_has_no_fabricated_events():
 
 
 def test_stored_event_export_preserves_metadata_namespace():
+    """Namespace + @context terms without pyld (pyld is not in requirements.lock)."""
     event = {
         "type": "ObjectEvent",
         "eventTime": datetime.now(timezone.utc).isoformat(),
@@ -101,44 +141,77 @@ def test_stored_event_export_preserves_metadata_namespace():
     assert_conformant(document)
     assert "regengine:metadata" in document
     assert "metadata" not in document
-    jsonld = pytest.importorskip("pyld").jsonld
-    context = fixture("epcis-context.jsonld")
-    dropped = []
+    assert document["@context"][0] == CONTEXT_URL
+    assert isinstance(document["@context"][1], dict)
+    assert document["@context"][1].get("regengine") == "https://regengine.co/ns/"
+    assert "regengine:source" in document["regengine:metadata"]
+    assert "regengine:event_count" in document["regengine:metadata"]
 
-    def loader(url, options=None):
-        if url != CONTEXT_URL:
-            raise ValueError(f"Unexpected context URL: {url}")
-        return {"contextUrl": None, "documentUrl": url, "document": context}
 
-    expanded = jsonld.expand(document, options={"documentLoader": loader}, on_property_dropped=dropped.append)
-    assert not dropped
-    assert "https://regengine.co/ns/metadata" in expanded[0]
+def test_gs1_check_digit_independent_mod10_and_published_examples():
+    """Independent mod-10 + known published vectors; production must agree.
 
+    Published examples are the independent guard (not minted via production).
+    Generated IDs are minted with the test-local mod-10, then checked by the
+    exporter so a reversed weight in production would fail.
+    """
+    # Published GS1 examples (data + expected check digit), each cited from a
+    # GS1 document on ref.gs1.org / gs1.org — no third-party pages.
+    published = (
+        # GTIN-14 09506000134352 — GS1 General Specifications Release 26.0
+        # §1.2.4.4 (JSON-LD example gs1:gtin); also QR Code powered by GS1
+        # best practices (2023), Best practice 1.
+        ("0950600013435", "2", 14),
+        # GLN-13 0614141000005 — GS1 Verifiable Credentials White Paper on
+        # Data Model and Validations 1.0.1 §5.1.2 (partyGLN for GS1 US).
+        ("061414100000", "5", 13),
+        # SSCC-18 376104250021234569 — GS1 General Specifications Release 26.0
+        # §7.9.1 Table 7-9 (18-digit check digit calculation example).
+        ("37610425002123456", "9", 18),
+    )
+    for data, expected_cd, length in published:
+        assert _independent_gs1_mod10_check_digit(data) == int(expected_cd)
+        full = data + expected_cd
+        assert len(full) == length
+        assert exporter._gs1_check_digit_valid(full)
+        wrong_cd = "0" if expected_cd != "0" else "1"
+        bad = data + wrong_cd
+        assert _independent_gs1_mod10_check_digit(data) != int(wrong_cd)
+        assert not exporter._gs1_check_digit_valid(bad)
 
-def test_gs1_check_digit_weights_match_known_valid_examples():
-    """Rightmost data digit must receive weight 3 (GS1 mod-10)."""
-    # GTIN-14, GLN-13, and EAN-13 textbook examples
-    assert exporter._gs1_check_digit_valid("09506000134352")
-    assert exporter._gs1_check_digit_valid("0614141000005")
-    assert exporter._gs1_check_digit_valid("4006381333931")
-    # SSCC-18: base 10614141123456789 → check digit 7
-    assert exporter._gs1_check_digit_valid("106141411234567897")
-    assert not exporter._gs1_check_digit_valid("09506000134351")
+    # Generated structure + production agreement with independent mint.
+    gtin = generated_gtin14()
+    gln = generated_gln13()
+    sscc = generated_sscc18()
+    assert len(gtin) == 14 and gtin.isdigit()
+    assert len(gln) == 13 and gln.isdigit()
+    assert len(sscc) == 18 and sscc.isdigit()
+    assert _independent_gs1_mod10_check_digit(gtin[:-1]) == int(gtin[-1])
+    assert _independent_gs1_mod10_check_digit(gln[:-1]) == int(gln[-1])
+    assert _independent_gs1_mod10_check_digit(sscc[:-1]) == int(sscc[-1])
+    assert exporter._gs1_check_digit_valid(gtin)
+    assert exporter._gs1_check_digit_valid(gln)
+    assert exporter._gs1_check_digit_valid(sscc)
+    bad_gtin = gtin[:-1] + ("0" if gtin[-1] != "0" else "1")
+    assert not exporter._gs1_check_digit_valid(bad_gtin)
 
 
 def test_export_builds_class_from_query_gtin_and_lot():
     """Regression: query returns gtin (not a test-only epc_class column)."""
+    gtin = generated_gtin14()
+    gln = generated_gln13()
+    lot = generated_lot()
     row = {
         "event_type": "receiving",
         "epcis_event_type": "ObjectEvent",
         "epcis_action": "ADD",
         "epcis_biz_step": "receiving",
         "event_timestamp": datetime.now(timezone.utc),
-        "gtin": "09506000134352",
-        "traceability_lot_code": "SYN-LOT-001",
+        "gtin": gtin,
+        "traceability_lot_code": lot,
         "quantity": 10,
         "unit_of_measure": "lbs",
-        "location_gln": "0614141000005",
+        "location_gln": gln,
     }
     ctx = type("Tenant", (), {"tenant_id": str(uuid4())})()
     with patch.object(exporter, "_query_tenant_events", return_value=[row]):
@@ -147,10 +220,10 @@ def test_export_builds_class_from_query_gtin_and_lot():
         )
     document = json.loads(response.body)
     event = document["epcisBody"]["eventList"][0]
-    assert event["quantityList"][0]["epcClass"] == "https://id.gs1.org/01/09506000134352/10/SYN-LOT-001"
+    assert event["quantityList"][0]["epcClass"] == f"https://id.gs1.org/01/{gtin}/10/{lot}"
     assert event["quantityList"][0]["uom"] == "LBR"
     assert event["regengine:unitOfMeasure"] == "lbs"
-    assert event["readPoint"]["id"] == "https://id.gs1.org/414/0614141000005"
+    assert event["readPoint"]["id"] == f"https://id.gs1.org/414/{gln}"
     assert document["regengine:exportMetadata"]["skippedEventsCount"] == 0
     assert "regengine:message" not in document["regengine:exportMetadata"]
 
@@ -164,7 +237,7 @@ def test_export_builds_class_from_product_id_kde():
         "epcis_biz_step": None,
         "event_timestamp": datetime.now(timezone.utc),
         "product_id": epc_class,
-        "traceability_lot_code": "SYN-LOT-002",
+        "traceability_lot_code": generated_lot(),
         "quantity": None,
         "unit_of_measure": None,
         "location_gln": None,
@@ -188,7 +261,7 @@ def test_transformation_cte_emitted_as_transformation_event():
         "epcis_biz_step": "commissioning",
         "event_timestamp": datetime.now(timezone.utc),
         "epc_class": epc_class,
-        "traceability_lot_code": "OUT-LOT",
+        "traceability_lot_code": generated_lot(),
         "quantity": 5,
         "unit_of_measure": "kg",
         "sha256_hash": "a" * 64,
@@ -211,8 +284,8 @@ def test_transformation_cte_emitted_as_transformation_event():
     assert document["regengine:exportMetadata"]["skippedEventsCount"] == 0
 
 
-def test_harvesting_maps_to_collecting_bizstep():
-    """Phase A keeps harvesting → collecting."""
+def test_harvesting_maps_to_creating_class_instance_bizstep():
+    """Harvesting commissions a class instance (CBV creating_class_instance)."""
     epc_class = class_uri()
     row = {
         "event_type": "harvesting",
@@ -221,7 +294,7 @@ def test_harvesting_maps_to_collecting_bizstep():
         "epcis_biz_step": None,
         "event_timestamp": datetime.now(timezone.utc),
         "epc_class": epc_class,
-        "traceability_lot_code": "HARVEST-1",
+        "traceability_lot_code": generated_lot(),
         "quantity": 1,
         "unit_of_measure": "EA",
         "location_gln": None,
@@ -232,4 +305,4 @@ def test_harvesting_maps_to_collecting_bizstep():
             exporter.export_epcis(exporter.ExportRequest(tenant_id=ctx.tenant_id), validate=False, ctx=ctx)
         )
     document = json.loads(response.body)
-    assert document["epcisBody"]["eventList"][0]["bizStep"] == "collecting"
+    assert document["epcisBody"]["eventList"][0]["bizStep"] == "creating_class_instance"