From 5c0f171e606e993c5d694eccae758624ec419032 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Wed, 28 Oct 2026 08:20:48 -0700
Subject: [PATCH 1/5] fix(ingestion): EPCIS export Preview labels (H5)

Add X-RegEngine-Preview and exportMetadata.preview/previewNotice on every
EPCIS (and FDA) export response. Mark /formats epcis as status preview and
remove unconfirmed named retailer claims. Spec section 7 test 10 covered
with synthetic harvest fixture path only (A-2165).

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 services/ingestion/app/epcis_export.py        |  25 ++-
 .../tests/test_epcis_export_preview.py        | 157 ++++++++++++++++++
 2 files changed, 175 insertions(+), 7 deletions(-)
 create mode 100644 services/ingestion/tests/test_epcis_export_preview.py

diff --git a/services/ingestion/app/epcis_export.py b/services/ingestion/app/epcis_export.py
index d5a1f0859..56dc25389 100644
--- a/services/ingestion/app/epcis_export.py
+++ b/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
@@ -483,8 +483,9 @@ 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."
+        "Generates a GS1 EPCIS 2.0 JSON-LD document containing traceability "
+        "events (Preview). Validate with your trading partner before relying "
+        "on it for compliance."
     ),
 )
 async def export_epcis(
@@ -625,6 +626,8 @@ def _skip(row: dict, reason: str) -> None:
                         "valid",
                         "errors",
                         "checkedAt",
+                        "preview",
+                        "previewNotice",
                     )
                 },
             },
@@ -646,6 +649,11 @@ def _skip(row: dict, reason: str) -> None:
             "dataSource": data_source,
             "integrityVerified": data_source == "tenant",
             "chainHashVerified": data_source == "tenant",
+            "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 {}),
         },
     }
@@ -681,6 +689,7 @@ def _skip(row: dict, reason: str) -> None:
         "X-RegEngine-Skipped-Events-Count": str(skipped),
         "X-RegEngine-Integrity-Verified": "true" if data_source == "tenant" else "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"
@@ -757,6 +766,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"
@@ -780,9 +790,10 @@ 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",
diff --git a/services/ingestion/tests/test_epcis_export_preview.py b/services/ingestion/tests/test_epcis_export_preview.py
new file mode 100644
index 000000000..6c44eb414
--- /dev/null
+++ b/services/ingestion/tests/test_epcis_export_preview.py
@@ -0,0 +1,157 @@
+"""H5 Preview labels for EPCIS export (spec section 6 / section 7 test 10).
+
+Synthetic fixtures only. No partner data. EPCIS stays Preview.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+from uuid import uuid4
+
+import pytest
+from fastapi import FastAPI, Request
+from fastapi.testclient import TestClient
+
+_ingestion_dir = Path(__file__).resolve().parent.parent
+if str(_ingestion_dir) not in sys.path:
+    sys.path.insert(0, str(_ingestion_dir))
+
+_services_dir = _ingestion_dir.parent
+if str(_services_dir) not in sys.path:
+    sys.path.insert(0, str(_services_dir))
+
+pytest.importorskip("fastapi")
+
+import app.epcis_export as epcis_export
+from app.epcis_export import router as epcis_export_router
+from app.webhook_compat import _verify_api_key, require_tenant_context
+from shared.tenant_context import TenantContext
+
+FIXTURES = Path(__file__).resolve().parent / "fixtures"
+HARVEST_CSV = FIXTURES / "01_harvesting.csv"
+PREVIEW_NOTICE = (
+    "EPCIS export is in Preview. Validate with your trading partner "
+    "before relying on it for compliance."
+)
+NAMED_RETAILERS = ("Walmart", "Kroger", "Costco", "Target")
+
+
+@pytest.fixture()
+def client() -> TestClient:
+    app = FastAPI()
+    app.include_router(epcis_export_router)
+
+    async def _tenant_ctx(request: Request) -> TenantContext:
+        tenant_id = request.path_params.get("tenant_id")
+        if not tenant_id and request.method in {"POST", "PUT", "PATCH"}:
+            try:
+                body = await request.json()
+            except Exception:
+                body = {}
+            if isinstance(body, dict):
+                tenant_id = body.get("tenant_id") or body.get("tenantId")
+        return TenantContext(
+            tenant_id=tenant_id or str(uuid4()),
+            principal_kind="test",
+            principal_id=uuid4().hex,
+        )
+
+    app.dependency_overrides[_verify_api_key] = lambda: None
+    app.dependency_overrides[require_tenant_context] = _tenant_ctx
+    return TestClient(app)
+
+
+def _empty_rows(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(epcis_export, "_load_export_rows", lambda *_a, **_k: [])
+
+
+class TestPreviewHeaderAndMetadata:
+    """Spec section 7 test 10: Preview header and metadata on every response."""
+
+    def test_preview_header_on_empty_export(
+        self, client: TestClient, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        _empty_rows(monkeypatch)
+        response = client.post(
+            "/api/v1/export/epcis",
+            json={"tenant_id": str(uuid4())},
+        )
+        assert response.status_code == 200
+        assert response.headers.get("X-RegEngine-Preview") == "true"
+        meta = response.json()["regengine:exportMetadata"]
+        assert meta["preview"] is True
+        assert meta["previewNotice"] == PREVIEW_NOTICE
+        assert meta["eventsCount"] == 0
+
+    def test_preview_header_on_success(
+        self, client: TestClient, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        # Synthetic harvest lot string only (fixture path under tests/fixtures).
+        assert HARVEST_CSV.is_file()
+        lot = HARVEST_CSV.read_text().splitlines()[1].split(",")[0]
+        assert lot.startswith("LOT-")
+
+        def _one_row(*_a: object, **_k: object) -> list[dict]:
+            return [
+                {
+                    "event_type": "harvesting",
+                    "epc_class": "https://id.gs1.org/01/00614141999996/10/" + lot,
+                    "traceability_lot_code": lot,
+                    "quantity": 5000,
+                    "unit_of_measure": "lbs",
+                    "event_timestamp": "2026-03-12T12:00:00+00:00",
+                    "location_gln": "0061414100014",
+                    "sha256_hash": None,
+                    "epcis_event_type": "ObjectEvent",
+                    "epcis_action": "ADD",
+                    "epcis_biz_step": None,
+                }
+            ]
+
+        monkeypatch.setattr(epcis_export, "_load_export_rows", _one_row)
+        response = client.post(
+            "/api/v1/export/epcis",
+            json={"tenant_id": str(uuid4())},
+        )
+        assert response.status_code == 200
+        assert response.headers.get("X-RegEngine-Preview") == "true"
+        body = response.json()
+        meta = body["regengine:exportMetadata"]
+        assert meta["preview"] is True
+        assert meta["previewNotice"] == PREVIEW_NOTICE
+        assert meta["eventsCount"] >= 1
+        # preview keys registered in @context
+        ctx = body["@context"]
+        assert isinstance(ctx, list) and len(ctx) >= 2
+        assert ctx[1].get("preview") == "regengine:preview"
+        assert ctx[1].get("previewNotice") == "regengine:previewNotice"
+
+    def test_preview_header_on_fda_empty(
+        self, client: TestClient, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        _empty_rows(monkeypatch)
+        response = client.post(
+            "/api/v1/export/fda",
+            json={"tenant_id": str(uuid4())},
+        )
+        assert response.status_code == 200
+        assert response.headers.get("X-RegEngine-Preview") == "true"
+
+
+class TestFormatsPreviewStatus:
+    def test_formats_epcis_status_preview(self, client: TestClient) -> None:
+        response = client.get("/api/v1/export/formats")
+        assert response.status_code == 200
+        formats = {f["id"]: f for f in response.json()["formats"]}
+        epcis = formats["epcis"]
+        assert epcis["status"] == "preview"
+        assert "Preview" in epcis["description"]
+        assert epcis["retailers"] == []
+
+    def test_formats_has_no_named_retailers(self, client: TestClient) -> None:
+        response = client.get("/api/v1/export/formats")
+        blob = json.dumps(response.json())
+        for name in NAMED_RETAILERS:
+            assert name not in blob

From aaaa6fbf98f70fb89dfb581f5ef8e9b1838a3980 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Wed, 28 Oct 2026 08:44:08 -0700
Subject: [PATCH 2/5] fix(epcis): Preview labels, harvest
 creating_class_instance, generated test IDs

Fold P1s into H5: label graph EPCIS Preview (header + exportMetadata),
map Harvest bizStep to creating_class_instance on ingestion and graph,
and rewrite both test_epcis_export_gs1_2 suites to use generated GTIN,
GLN, and lot values with check-digit round-trip checks.

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 services/graph/app/routers/fsma/compliance.py | 14 +++-
 .../graph/tests/test_epcis_export_gs1_2.py    | 79 +++++++++++++------
 services/ingestion/app/epcis_export.py        |  2 +-
 .../tests/test_epcis_export_gs1_2.py          | 74 ++++++++++++-----
 4 files changed, 123 insertions(+), 46 deletions(-)

diff --git a/services/graph/app/routers/fsma/compliance.py b/services/graph/app/routers/fsma/compliance.py
index 511a9e550..f17c162a6 100644
--- a/services/graph/app/routers/fsma/compliance.py
+++ b/services/graph/app/routers/fsma/compliance.py
@@ -273,6 +273,7 @@ async def export_epcis(
 
     Event Types Mapped:
     - CREATION → epcis:ObjectEvent (bizStep: commissioning)
+    - HARVESTING → epcis:ObjectEvent (bizStep: creating_class_instance)
     - RECEIVING → epcis:ObjectEvent (bizStep: receiving)
     - SHIPPING → epcis:ObjectEvent (bizStep: shipping)
     - TRANSFORMATION → epcis:TransformationEvent
@@ -345,13 +346,13 @@ def location(value):
                 return f"https://id.gs1.org/414/{value}"
             return epc_id(value, "sgln")
 
-        # Phase A: harvesting maps to CBV collecting (kept intentionally).
+        # Harvesting commissions a class instance (CBV creating_class_instance).
         BIZ_STEP_MAP = {
             "CREATION": "commissioning",
             "RECEIVING": "receiving",
             "SHIPPING": "shipping",
             "TRANSFORMATION": "commissioning",
-            "HARVESTING": "collecting",
+            "HARVESTING": "creating_class_instance",
             "COOLING": "storing",
             "INITIAL_PACKING": "packing",
             "FIRST_LAND_BASED_RECEIVING": "receiving",
@@ -474,6 +475,11 @@ def _skip(reason: str) -> None:
                 "eventsCount": len(epcis_events),
                 "skippedEventsCount": skipped_total,
                 "skippedEventsByReason": skipped_by_reason,
+                "preview": True,
+                "previewNotice": (
+                    "EPCIS export is in Preview. Validate with your trading partner "
+                    "before relying on it for compliance."
+                ),
             },
         }
 
@@ -493,6 +499,7 @@ def _skip(reason: str) -> None:
                     "Content-Disposition": f"attachment; filename={filename}",
                     "X-RegEngine-Events-Count": str(len(epcis_events)),
                     "X-RegEngine-Skipped-Events-Count": skip_header,
+                    "X-RegEngine-Preview": "true",
                 },
             )
         else:
@@ -508,6 +515,7 @@ def _skip(reason: str) -> None:
                     "Content-Disposition": f"attachment; filename={filename}",
                     "X-RegEngine-Events-Count": str(len(epcis_events)),
                     "X-RegEngine-Skipped-Events-Count": skip_header,
+                    "X-RegEngine-Preview": "true",
                 },
             )
 
@@ -569,7 +577,7 @@ def add_ilmd(parent, event):
                 continue
             add(node, f"{{{key[: separator + 1]}}}{key[separator + 1 :]}", value)
 
-    _CBV_BIZSTEPS = {"shipping", "receiving", "commissioning", "collecting", "storing", "packing"}
+    _CBV_BIZSTEPS = {"shipping", "receiving", "commissioning", "collecting", "creating_class_instance", "storing", "packing"}
     _CBV_DISPOSITIONS = {"in_transit", "in_progress"}
 
     def cbv_value(field, value):
diff --git a/services/graph/tests/test_epcis_export_gs1_2.py b/services/graph/tests/test_epcis_export_gs1_2.py
index 71b577e18..8ce6e4799 100644
--- a/services/graph/tests/test_epcis_export_gs1_2.py
+++ b/services/graph/tests/test_epcis_export_gs1_2.py
@@ -15,7 +15,6 @@
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[3]))
 
-from services.graph.app import fsma_utils  # noqa: E402
 from services.graph.app.routers.fsma import compliance  # noqa: E402
 
 FIXTURES = Path(__file__).resolve().parents[2] / "ingestion" / "tests" / "fixtures" / "epcis_gs1"
@@ -55,6 +54,26 @@ def location_identifier():
     return f"{digits[:7]}.{digits[7:]}.{uuid.uuid4().hex}"
 
 
+def _gs1_with_check(data_digits: str) -> str:
+    from services.shared.fsma_validation import calculate_gs1_check_digit
+
+    return data_digits + str(calculate_gs1_check_digit(data_digits))
+
+
+def generated_gtin14() -> str:
+    base = str(uuid.uuid4().int % (10**13)).zfill(13)
+    return _gs1_with_check(base)
+
+
+def generated_gln13() -> str:
+    base = str(uuid.uuid4().int % (10**12)).zfill(12)
+    return _gs1_with_check(base)
+
+
+def generated_lot() -> str:
+    return f"SYN-{uuid.uuid4().hex[:12]}"
+
+
 def graph_row(kind):
     now = datetime.now(timezone.utc)
     row = {
@@ -74,16 +93,17 @@ def graph_row(kind):
 
 
 async def export(format, rows):
-    async def query(client, start, end, tenant):
+    async def query(client, start, end, tenant, **_kwargs):
         assert tenant == str(TENANT)
         return rows
 
     today = datetime.now(timezone.utc).date().isoformat()
     # Bypass slowapi so the suite can call the endpoint many times in one process.
+    # Patch the name bound in compliance (from-import), not fsma_utils.
     endpoint = getattr(compliance.export_epcis, "__wrapped__", compliance.export_epcis)
     with patch.object(compliance, "Neo4jClient", FakeNeo4j), patch.object(
         compliance, "reconcile_tenant_scope", return_value=str(TENANT)
-    ), patch.object(fsma_utils, "query_events_by_range", query):
+    ), patch.object(compliance, "query_events_by_range", query):
         response = await endpoint(
             Request({"type": "http", "method": "GET", "path": "/export/epcis", "query_string": b"", "headers": []}),
             today, today, format, TENANT, object(),
@@ -123,6 +143,8 @@ def test_empty_export_has_no_events_in_either_format():
     document = json.loads(asyncio.run(export("json-ld", [])))
     xml = asyncio.run(export("xml", []))
     assert document["epcisBody"]["eventList"] == []
+    assert document["regengine:exportMetadata"]["preview"] is True
+    assert "Preview" in document["regengine:exportMetadata"]["previewNotice"]
     jsonschema = pytest.importorskip("jsonschema")
     jsonschema.Draft7Validator(
         json.loads(schema_file("EPCIS-JSON-Schema.json").read_text())
@@ -140,26 +162,36 @@ def test_missing_event_time_is_skipped():
     assert document["epcisBody"]["eventList"] == []
 
 
-def test_gs1_location_check_digit_accepts_known_valid_gln():
-    """Graph location helper must use GS1 weight-3-from-right."""
+def test_gs1_location_check_digit_round_trip_for_generated_gln():
+    """Graph location helper must use GS1 weight-3-from-right on generated GLNs."""
+    from services.shared.fsma_validation import verify_gs1_check_digit
+
+    gln = generated_gln13()
+    assert len(gln) == 13 and gln.isdigit()
+    assert verify_gs1_check_digit(gln)
+    bad = gln[:-1] + ("0" if gln[-1] != "0" else "1")
+    assert not verify_gs1_check_digit(bad)
     row = graph_row("RECEIVING")
-    row["location_gln"] = "0614141000005"
+    row["location_gln"] = gln
     row["tlc"] = epc_identifier()
     document = json.loads(asyncio.run(export("json-ld", [row])))
     assert document["epcisBody"]["eventList"]
-    assert document["epcisBody"]["eventList"][0]["readPoint"]["id"] == "https://id.gs1.org/414/0614141000005"
+    assert document["epcisBody"]["eventList"][0]["readPoint"]["id"] == f"https://id.gs1.org/414/{gln}"
 
 
 def test_graph_exports_raw_tlc_with_lot_gtin_as_digital_link():
-    """Real FSMA lots use free-text TLC + Lot.gtin — must not be silently dropped."""
+    """Real FSMA lots use free-text TLC + Lot.gtin; must not be silently dropped."""
+    gtin = generated_gtin14()
+    gln = generated_gln13()
+    lot = generated_lot()
     row = {
         "type": "SHIPPING",
         "event_date": datetime.now(timezone.utc).date().isoformat(),
         "event_time": datetime.now(timezone.utc).time().isoformat(timespec="seconds"),
-        "tlc": "LOT-2024-001",
-        "gtin": "09506000134352",
-        "location_gln": "0614141000005",
-        "ship_from_gln": "0614141000005",
+        "tlc": lot,
+        "gtin": gtin,
+        "location_gln": gln,
+        "ship_from_gln": gln,
         "ship_to_gln": location_identifier(),
         "quantity": 12,
         "unit_of_measure": "KGM",
@@ -170,38 +202,41 @@ def test_graph_exports_raw_tlc_with_lot_gtin_as_digital_link():
     assert len(events) == 1
     event = events[0]
     assert event["type"] == "ObjectEvent"
-    assert event["quantityList"][0]["epcClass"] == "https://id.gs1.org/01/09506000134352/10/LOT-2024-001"
+    assert event["quantityList"][0]["epcClass"] == f"https://id.gs1.org/01/{gtin}/10/{lot}"
     assert event["bizStep"] == "shipping"
     assert document["regengine:exportMetadata"]["skippedEventsCount"] == 0
 
 
 def test_graph_exports_transformation_with_input_tlc_and_gtin():
+    gtin = generated_gtin14()
+    out_lot = generated_lot()
+    in_lot = generated_lot()
     row = {
         "type": "TRANSFORMATION",
         "event_date": datetime.now(timezone.utc).date().isoformat(),
         "event_time": datetime.now(timezone.utc).time().isoformat(timespec="seconds"),
-        "tlc": "OUT-LOT-9",
-        "gtin": "09506000134352",
-        "input_tlc": "IN-LOT-9",
-        "input_gtin": "09506000134352",
+        "tlc": out_lot,
+        "gtin": gtin,
+        "input_tlc": in_lot,
+        "input_gtin": gtin,
         "location_gln": location_identifier(),
-        "event_id": "evt-bare-id-123",
+        "event_id": f"evt-bare-{uuid.uuid4().hex[:12]}",
         "quantity": 3,
         "unit_of_measure": "KGM",
     }
     document = json.loads(asyncio.run(export("json-ld", [row])))
     event = document["epcisBody"]["eventList"][0]
     assert event["type"] == "TransformationEvent"
-    assert event["outputQuantityList"][0]["epcClass"].endswith("/10/OUT-LOT-9")
-    assert event["inputQuantityList"][0]["epcClass"].endswith("/10/IN-LOT-9")
+    assert event["outputQuantityList"][0]["epcClass"].endswith(f"/10/{out_lot}")
+    assert event["inputQuantityList"][0]["epcClass"].endswith(f"/10/{in_lot}")
     assert event["transformationID"].startswith("urn:uuid:")
 
 
-def test_graph_harvesting_uses_collecting_bizstep():
+def test_graph_harvesting_uses_creating_class_instance_bizstep():
     row = graph_row("HARVESTING")
     # graph_row uses split SGTIN-shaped tlc
     document = json.loads(asyncio.run(export("json-ld", [row])))
-    assert document["epcisBody"]["eventList"][0]["bizStep"] == "collecting"
+    assert document["epcisBody"]["eventList"][0]["bizStep"] == "creating_class_instance"
 
 
 def test_missing_event_time_records_skip_reason():
diff --git a/services/ingestion/app/epcis_export.py b/services/ingestion/app/epcis_export.py
index 56dc25389..c21649a75 100644
--- a/services/ingestion/app/epcis_export.py
+++ b/services/ingestion/app/epcis_export.py
@@ -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",
 }
diff --git a/services/ingestion/tests/test_epcis_export_gs1_2.py b/services/ingestion/tests/test_epcis_export_gs1_2.py
index dbf4c49c5..ce3c71867 100644
--- a/services/ingestion/tests/test_epcis_export_gs1_2.py
+++ b/services/ingestion/tests/test_epcis_export_gs1_2.py
@@ -33,6 +33,32 @@ def class_uri():
     return f"urn:epc:class:lgtin:{digits[:7]}.{digits[7:]}.{uuid4().hex}"
 
 
+def _gs1_with_check(data_digits: str) -> str:
+    """Append GS1 mod-10 check digit (weight 3 from the right)."""
+    from shared.fsma_validation import calculate_gs1_check_digit
+
+    return data_digits + str(calculate_gs1_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(
@@ -113,30 +139,38 @@ def loader(url, options=None):
     assert "https://regengine.co/ns/metadata" in expanded[0]
 
 
-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")
+def test_gs1_check_digit_round_trip_for_generated_ids():
+    """Generated GTIN/GLN/SSCC pass structure + check-digit round-trip."""
+    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 exporter._gs1_check_digit_valid(gtin)
+    assert exporter._gs1_check_digit_valid(gln)
+    assert exporter._gs1_check_digit_valid(sscc)
+    # Flipping the check digit must fail
+    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]):
@@ -145,10 +179,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"]
 
@@ -162,7 +196,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,
@@ -186,7 +220,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,
@@ -209,8 +243,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",
@@ -219,7 +253,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,
@@ -230,4 +264,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"

From 498b7eb792b854f117ec0e8caaa5c8f6bc9ce595 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Wed, 28 Oct 2026 08:44:55 -0700
Subject: [PATCH 3/5] revert(graph): leave EPCIS graph export to Issue Engineer

Restore compliance.py and test_epcis_export_gs1_2.py to origin/main.
#3716 keeps ingestion-only Preview, Harvest creating_class_instance, and
generated GS1 test IDs.

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 services/graph/app/routers/fsma/compliance.py | 14 +---
 .../graph/tests/test_epcis_export_gs1_2.py    | 79 ++++++-------------
 2 files changed, 25 insertions(+), 68 deletions(-)

diff --git a/services/graph/app/routers/fsma/compliance.py b/services/graph/app/routers/fsma/compliance.py
index f17c162a6..511a9e550 100644
--- a/services/graph/app/routers/fsma/compliance.py
+++ b/services/graph/app/routers/fsma/compliance.py
@@ -273,7 +273,6 @@ async def export_epcis(
 
     Event Types Mapped:
     - CREATION → epcis:ObjectEvent (bizStep: commissioning)
-    - HARVESTING → epcis:ObjectEvent (bizStep: creating_class_instance)
     - RECEIVING → epcis:ObjectEvent (bizStep: receiving)
     - SHIPPING → epcis:ObjectEvent (bizStep: shipping)
     - TRANSFORMATION → epcis:TransformationEvent
@@ -346,13 +345,13 @@ def location(value):
                 return f"https://id.gs1.org/414/{value}"
             return epc_id(value, "sgln")
 
-        # Harvesting commissions a class instance (CBV creating_class_instance).
+        # Phase A: harvesting maps to CBV collecting (kept intentionally).
         BIZ_STEP_MAP = {
             "CREATION": "commissioning",
             "RECEIVING": "receiving",
             "SHIPPING": "shipping",
             "TRANSFORMATION": "commissioning",
-            "HARVESTING": "creating_class_instance",
+            "HARVESTING": "collecting",
             "COOLING": "storing",
             "INITIAL_PACKING": "packing",
             "FIRST_LAND_BASED_RECEIVING": "receiving",
@@ -475,11 +474,6 @@ def _skip(reason: str) -> None:
                 "eventsCount": len(epcis_events),
                 "skippedEventsCount": skipped_total,
                 "skippedEventsByReason": skipped_by_reason,
-                "preview": True,
-                "previewNotice": (
-                    "EPCIS export is in Preview. Validate with your trading partner "
-                    "before relying on it for compliance."
-                ),
             },
         }
 
@@ -499,7 +493,6 @@ def _skip(reason: str) -> None:
                     "Content-Disposition": f"attachment; filename={filename}",
                     "X-RegEngine-Events-Count": str(len(epcis_events)),
                     "X-RegEngine-Skipped-Events-Count": skip_header,
-                    "X-RegEngine-Preview": "true",
                 },
             )
         else:
@@ -515,7 +508,6 @@ def _skip(reason: str) -> None:
                     "Content-Disposition": f"attachment; filename={filename}",
                     "X-RegEngine-Events-Count": str(len(epcis_events)),
                     "X-RegEngine-Skipped-Events-Count": skip_header,
-                    "X-RegEngine-Preview": "true",
                 },
             )
 
@@ -577,7 +569,7 @@ def add_ilmd(parent, event):
                 continue
             add(node, f"{{{key[: separator + 1]}}}{key[separator + 1 :]}", value)
 
-    _CBV_BIZSTEPS = {"shipping", "receiving", "commissioning", "collecting", "creating_class_instance", "storing", "packing"}
+    _CBV_BIZSTEPS = {"shipping", "receiving", "commissioning", "collecting", "storing", "packing"}
     _CBV_DISPOSITIONS = {"in_transit", "in_progress"}
 
     def cbv_value(field, value):
diff --git a/services/graph/tests/test_epcis_export_gs1_2.py b/services/graph/tests/test_epcis_export_gs1_2.py
index 8ce6e4799..71b577e18 100644
--- a/services/graph/tests/test_epcis_export_gs1_2.py
+++ b/services/graph/tests/test_epcis_export_gs1_2.py
@@ -15,6 +15,7 @@
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[3]))
 
+from services.graph.app import fsma_utils  # noqa: E402
 from services.graph.app.routers.fsma import compliance  # noqa: E402
 
 FIXTURES = Path(__file__).resolve().parents[2] / "ingestion" / "tests" / "fixtures" / "epcis_gs1"
@@ -54,26 +55,6 @@ def location_identifier():
     return f"{digits[:7]}.{digits[7:]}.{uuid.uuid4().hex}"
 
 
-def _gs1_with_check(data_digits: str) -> str:
-    from services.shared.fsma_validation import calculate_gs1_check_digit
-
-    return data_digits + str(calculate_gs1_check_digit(data_digits))
-
-
-def generated_gtin14() -> str:
-    base = str(uuid.uuid4().int % (10**13)).zfill(13)
-    return _gs1_with_check(base)
-
-
-def generated_gln13() -> str:
-    base = str(uuid.uuid4().int % (10**12)).zfill(12)
-    return _gs1_with_check(base)
-
-
-def generated_lot() -> str:
-    return f"SYN-{uuid.uuid4().hex[:12]}"
-
-
 def graph_row(kind):
     now = datetime.now(timezone.utc)
     row = {
@@ -93,17 +74,16 @@ def graph_row(kind):
 
 
 async def export(format, rows):
-    async def query(client, start, end, tenant, **_kwargs):
+    async def query(client, start, end, tenant):
         assert tenant == str(TENANT)
         return rows
 
     today = datetime.now(timezone.utc).date().isoformat()
     # Bypass slowapi so the suite can call the endpoint many times in one process.
-    # Patch the name bound in compliance (from-import), not fsma_utils.
     endpoint = getattr(compliance.export_epcis, "__wrapped__", compliance.export_epcis)
     with patch.object(compliance, "Neo4jClient", FakeNeo4j), patch.object(
         compliance, "reconcile_tenant_scope", return_value=str(TENANT)
-    ), patch.object(compliance, "query_events_by_range", query):
+    ), patch.object(fsma_utils, "query_events_by_range", query):
         response = await endpoint(
             Request({"type": "http", "method": "GET", "path": "/export/epcis", "query_string": b"", "headers": []}),
             today, today, format, TENANT, object(),
@@ -143,8 +123,6 @@ def test_empty_export_has_no_events_in_either_format():
     document = json.loads(asyncio.run(export("json-ld", [])))
     xml = asyncio.run(export("xml", []))
     assert document["epcisBody"]["eventList"] == []
-    assert document["regengine:exportMetadata"]["preview"] is True
-    assert "Preview" in document["regengine:exportMetadata"]["previewNotice"]
     jsonschema = pytest.importorskip("jsonschema")
     jsonschema.Draft7Validator(
         json.loads(schema_file("EPCIS-JSON-Schema.json").read_text())
@@ -162,36 +140,26 @@ def test_missing_event_time_is_skipped():
     assert document["epcisBody"]["eventList"] == []
 
 
-def test_gs1_location_check_digit_round_trip_for_generated_gln():
-    """Graph location helper must use GS1 weight-3-from-right on generated GLNs."""
-    from services.shared.fsma_validation import verify_gs1_check_digit
-
-    gln = generated_gln13()
-    assert len(gln) == 13 and gln.isdigit()
-    assert verify_gs1_check_digit(gln)
-    bad = gln[:-1] + ("0" if gln[-1] != "0" else "1")
-    assert not verify_gs1_check_digit(bad)
+def test_gs1_location_check_digit_accepts_known_valid_gln():
+    """Graph location helper must use GS1 weight-3-from-right."""
     row = graph_row("RECEIVING")
-    row["location_gln"] = gln
+    row["location_gln"] = "0614141000005"
     row["tlc"] = epc_identifier()
     document = json.loads(asyncio.run(export("json-ld", [row])))
     assert document["epcisBody"]["eventList"]
-    assert document["epcisBody"]["eventList"][0]["readPoint"]["id"] == f"https://id.gs1.org/414/{gln}"
+    assert document["epcisBody"]["eventList"][0]["readPoint"]["id"] == "https://id.gs1.org/414/0614141000005"
 
 
 def test_graph_exports_raw_tlc_with_lot_gtin_as_digital_link():
-    """Real FSMA lots use free-text TLC + Lot.gtin; must not be silently dropped."""
-    gtin = generated_gtin14()
-    gln = generated_gln13()
-    lot = generated_lot()
+    """Real FSMA lots use free-text TLC + Lot.gtin — must not be silently dropped."""
     row = {
         "type": "SHIPPING",
         "event_date": datetime.now(timezone.utc).date().isoformat(),
         "event_time": datetime.now(timezone.utc).time().isoformat(timespec="seconds"),
-        "tlc": lot,
-        "gtin": gtin,
-        "location_gln": gln,
-        "ship_from_gln": gln,
+        "tlc": "LOT-2024-001",
+        "gtin": "09506000134352",
+        "location_gln": "0614141000005",
+        "ship_from_gln": "0614141000005",
         "ship_to_gln": location_identifier(),
         "quantity": 12,
         "unit_of_measure": "KGM",
@@ -202,41 +170,38 @@ def test_graph_exports_raw_tlc_with_lot_gtin_as_digital_link():
     assert len(events) == 1
     event = events[0]
     assert event["type"] == "ObjectEvent"
-    assert event["quantityList"][0]["epcClass"] == f"https://id.gs1.org/01/{gtin}/10/{lot}"
+    assert event["quantityList"][0]["epcClass"] == "https://id.gs1.org/01/09506000134352/10/LOT-2024-001"
     assert event["bizStep"] == "shipping"
     assert document["regengine:exportMetadata"]["skippedEventsCount"] == 0
 
 
 def test_graph_exports_transformation_with_input_tlc_and_gtin():
-    gtin = generated_gtin14()
-    out_lot = generated_lot()
-    in_lot = generated_lot()
     row = {
         "type": "TRANSFORMATION",
         "event_date": datetime.now(timezone.utc).date().isoformat(),
         "event_time": datetime.now(timezone.utc).time().isoformat(timespec="seconds"),
-        "tlc": out_lot,
-        "gtin": gtin,
-        "input_tlc": in_lot,
-        "input_gtin": gtin,
+        "tlc": "OUT-LOT-9",
+        "gtin": "09506000134352",
+        "input_tlc": "IN-LOT-9",
+        "input_gtin": "09506000134352",
         "location_gln": location_identifier(),
-        "event_id": f"evt-bare-{uuid.uuid4().hex[:12]}",
+        "event_id": "evt-bare-id-123",
         "quantity": 3,
         "unit_of_measure": "KGM",
     }
     document = json.loads(asyncio.run(export("json-ld", [row])))
     event = document["epcisBody"]["eventList"][0]
     assert event["type"] == "TransformationEvent"
-    assert event["outputQuantityList"][0]["epcClass"].endswith(f"/10/{out_lot}")
-    assert event["inputQuantityList"][0]["epcClass"].endswith(f"/10/{in_lot}")
+    assert event["outputQuantityList"][0]["epcClass"].endswith("/10/OUT-LOT-9")
+    assert event["inputQuantityList"][0]["epcClass"].endswith("/10/IN-LOT-9")
     assert event["transformationID"].startswith("urn:uuid:")
 
 
-def test_graph_harvesting_uses_creating_class_instance_bizstep():
+def test_graph_harvesting_uses_collecting_bizstep():
     row = graph_row("HARVESTING")
     # graph_row uses split SGTIN-shaped tlc
     document = json.loads(asyncio.run(export("json-ld", [row])))
-    assert document["epcisBody"]["eventList"][0]["bizStep"] == "creating_class_instance"
+    assert document["epcisBody"]["eventList"][0]["bizStep"] == "collecting"
 
 
 def test_missing_event_time_records_skip_reason():

From 9891e1dd725bb8a0439f2e1978eb95d8a49cc62d Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Wed, 28 Oct 2026 09:01:30 -0700
Subject: [PATCH 4/5] fix(ingestion): DeepSeek #3716 follow-ups for Preview
 EPCIS

Non-circular GS1 mod-10 tests with published examples; generated GTIN in
preview success path; run metadata namespace test without pyld (not in
lock); mark FDA /formats as preview.

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 services/ingestion/app/epcis_export.py        |  3 +-
 .../tests/test_epcis_export_gs1_2.py          | 74 +++++++++++++------
 .../tests/test_epcis_export_preview.py        | 27 ++++++-
 3 files changed, 80 insertions(+), 24 deletions(-)

diff --git a/services/ingestion/app/epcis_export.py b/services/ingestion/app/epcis_export.py
index c21649a75..d4e8675f4 100644
--- a/services/ingestion/app/epcis_export.py
+++ b/services/ingestion/app/epcis_export.py
@@ -798,8 +798,9 @@ async def list_export_formats():
             {
                 "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"],
             },
         ]
diff --git a/services/ingestion/tests/test_epcis_export_gs1_2.py b/services/ingestion/tests/test_epcis_export_gs1_2.py
index ce3c71867..a15340a98 100644
--- a/services/ingestion/tests/test_epcis_export_gs1_2.py
+++ b/services/ingestion/tests/test_epcis_export_gs1_2.py
@@ -33,11 +33,24 @@ def class_uri():
     return f"urn:epc:class:lgtin:{digits[:7]}.{digits[7:]}.{uuid4().hex}"
 
 
-def _gs1_with_check(data_digits: str) -> str:
-    """Append GS1 mod-10 check digit (weight 3 from the right)."""
-    from shared.fsma_validation import calculate_gs1_check_digit
+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
 
-    return data_digits + str(calculate_gs1_check_digit(data_digits))
+
+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:
@@ -109,6 +122,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(),
@@ -125,32 +139,50 @@ 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 = []
-
-    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_round_trip_for_generated_ids():
-    """Generated GTIN/GLN/SSCC pass structure + check-digit round-trip."""
+    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 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 textbook examples (data + expected check digit).
+    published = (
+        ("0950600013435", "2", 14),   # GTIN-14
+        ("061414100000", "5", 13),    # GLN-13
+        ("400638133393", "1", 13),    # EAN-13
+        ("10614141123456789", "7", 18),  # SSCC-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)
+
+    # 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)
-    # Flipping the check digit must fail
     bad_gtin = gtin[:-1] + ("0" if gtin[-1] != "0" else "1")
     assert not exporter._gs1_check_digit_valid(bad_gtin)
 
diff --git a/services/ingestion/tests/test_epcis_export_preview.py b/services/ingestion/tests/test_epcis_export_preview.py
index 6c44eb414..9194f6e5a 100644
--- a/services/ingestion/tests/test_epcis_export_preview.py
+++ b/services/ingestion/tests/test_epcis_export_preview.py
@@ -38,6 +38,19 @@
 NAMED_RETAILERS = ("Walmart", "Kroger", "Costco", "Target")
 
 
+def _independent_gs1_mod10_check_digit(data_digits: str) -> int:
+    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 generated_gtin14() -> str:
+    base = str(uuid4().int % (10**13)).zfill(13)
+    return base + str(_independent_gs1_mod10_check_digit(base))
+
+
+
 @pytest.fixture()
 def client() -> TestClient:
     app = FastAPI()
@@ -93,16 +106,18 @@ def test_preview_header_on_success(
         lot = HARVEST_CSV.read_text().splitlines()[1].split(",")[0]
         assert lot.startswith("LOT-")
 
+        gtin = generated_gtin14()
+
         def _one_row(*_a: object, **_k: object) -> list[dict]:
             return [
                 {
                     "event_type": "harvesting",
-                    "epc_class": "https://id.gs1.org/01/00614141999996/10/" + lot,
+                    "epc_class": f"https://id.gs1.org/01/{gtin}/10/" + lot,
                     "traceability_lot_code": lot,
                     "quantity": 5000,
                     "unit_of_measure": "lbs",
                     "event_timestamp": "2026-03-12T12:00:00+00:00",
-                    "location_gln": "0061414100014",
+                    "location_gln": "0061414100014",  # from fixtures/01_harvesting.csv
                     "sha256_hash": None,
                     "epcis_event_type": "ObjectEvent",
                     "epcis_action": "ADD",
@@ -150,6 +165,14 @@ def test_formats_epcis_status_preview(self, client: TestClient) -> None:
         assert "Preview" in epcis["description"]
         assert epcis["retailers"] == []
 
+    def test_formats_fda_status_preview(self, client: TestClient) -> None:
+        response = client.get("/api/v1/export/formats")
+        assert response.status_code == 200
+        formats = {f["id"]: f for f in response.json()["formats"]}
+        fda = formats["fda"]
+        assert fda["status"] == "preview"
+        assert "Preview" in fda["description"]
+
     def test_formats_has_no_named_retailers(self, client: TestClient) -> None:
         response = client.get("/api/v1/export/formats")
         blob = json.dumps(response.json())

From 8e526064f0f1240d3f476f110be1efa69f98ade1 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Wed, 28 Oct 2026 09:14:14 -0700
Subject: [PATCH 5/5] test(ingestion): cite GS1 docs for check-digit published
 examples
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Keep GTIN 09506000134352 and GLN 0614141000005 with document+section
citations verified on ref.gs1.org. Drop uncited 4006381333931 and
106141411234567897; replace with Gen Specs §7.9.1 Table 7-9 SSCC
376104250021234569. Addresses DeepSeek FAIL on #3716.

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 .../ingestion/tests/test_epcis_export_gs1_2.py  | 17 ++++++++++++-----
 1 file changed, 12 insertions(+), 5 deletions(-)

diff --git a/services/ingestion/tests/test_epcis_export_gs1_2.py b/services/ingestion/tests/test_epcis_export_gs1_2.py
index a15340a98..6ed7f588e 100644
--- a/services/ingestion/tests/test_epcis_export_gs1_2.py
+++ b/services/ingestion/tests/test_epcis_export_gs1_2.py
@@ -153,12 +153,19 @@ def test_gs1_check_digit_independent_mod10_and_published_examples():
     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 textbook examples (data + expected check digit).
+    # Published GS1 examples (data + expected check digit), each cited from a
+    # GS1 document on ref.gs1.org / gs1.org — no third-party pages.
     published = (
-        ("0950600013435", "2", 14),   # GTIN-14
-        ("061414100000", "5", 13),    # GLN-13
-        ("400638133393", "1", 13),    # EAN-13
-        ("10614141123456789", "7", 18),  # SSCC-18
+        # 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)