From 747e182583baf350122fae2f88564d6e9c1ee7c1 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Sun, 25 Oct 2026 22:33:30 -0700
Subject: [PATCH 1/6] fix(graph): page FSMA gap/chain/orphan queries instead of
 silent 2000-row cap (R1)

---
 services/graph/app/fsma/chain_integrity.py    |  61 ++++---
 services/graph/app/fsma/data_quality.py       |   8 +-
 services/graph/tests/test_fsma_module_qa.py   |   8 +-
 .../graph/tests/test_graph_row_paging_r1.py   | 165 ++++++++++++++++++
 4 files changed, 211 insertions(+), 31 deletions(-)
 create mode 100644 services/graph/tests/test_graph_row_paging_r1.py

diff --git a/services/graph/app/fsma/chain_integrity.py b/services/graph/app/fsma/chain_integrity.py
index 93e8347de..704edf316 100644
--- a/services/graph/app/fsma/chain_integrity.py
+++ b/services/graph/app/fsma/chain_integrity.py
@@ -8,7 +8,7 @@
 from __future__ import annotations
 
 import time
-from typing import Any, Dict, List, Optional
+from typing import Any, AsyncIterator, Dict, List, Optional
 
 import structlog
 from shared.fsma_rules import TimeArrowRule, TraceEvent as SharedTraceEvent
@@ -18,6 +18,31 @@
 logger = structlog.get_logger("fsma-utils")
 
 
+# Rows fetched per round trip. Results are read page by page (SKIP/LIMIT on a
+# stable ORDER BY) so every matching row is reported; nothing is silently
+# dropped at a fixed cap, and each query round trip stays bounded.
+_PAGE_SIZE = 2000
+
+
+async def _run_paged(session: Any, query: str, **params: Any) -> AsyncIterator[Any]:
+    """Yield every record of ``query``, fetched ``_PAGE_SIZE`` rows at a time.
+
+    ``query`` must end with a deterministic ORDER BY (unique tiebreaker) and
+    no SKIP/LIMIT; paging is appended here.
+    """
+    paged_query = query.rstrip() + "\n    SKIP $page_skip LIMIT $page_size\n"
+    skip = 0
+    while True:
+        result = await session.run(paged_query, page_skip=skip, page_size=_PAGE_SIZE, **params)
+        fetched = 0
+        async for record in result:
+            fetched += 1
+            yield record
+        if fetched < _PAGE_SIZE:
+            return
+        skip += fetched
+
+
 async def find_gaps(
     client: Neo4jClient,
     tenant_id: str,
@@ -37,9 +62,7 @@ async def find_gaps(
             callers read every tenant's events, so callers must pass a tenant.
     """
     if not tenant_id or not str(tenant_id).strip():
-        raise ValueError(
-            "find_gaps requires a non-empty tenant_id; null-tenant queries are no longer supported"
-        )
+        raise ValueError("find_gaps requires a non-empty tenant_id; null-tenant queries are no longer supported")
 
     start_time = time.time()
 
@@ -65,13 +88,12 @@ async def find_gaps(
         CASE WHEN e.event_date IS NULL OR e.event_date = '' THEN 'missing_date' ELSE '' END +
         CASE WHEN NOT EXISTS { MATCH (e)<-[:UNDERWENT]-(l:Lot) WHERE l.tenant_id = $tenant_id } THEN ',missing_lot' ELSE '' END +
         CASE WHEN e.responsible_party_contact IS NULL OR e.responsible_party_contact = '' THEN ',missing_responsible_party_contact' ELSE '' END as gaps
-    LIMIT 2000
+    ORDER BY event_id
     """
 
     gaps = []
     async with client.session() as session:
-        result = await session.run(query, tenant_id=tenant_id)
-        async for record in result:
+        async for record in _run_paged(session, query, tenant_id=tenant_id):
             gaps.append(
                 {
                     "event_id": record["event_id"],
@@ -140,7 +162,8 @@ async def find_broken_chains(
     # ------------------------------------------------------------------ #
     # Query 1: SHIPPING with no origin event at all
     # Optimization: single-hop UNDERWENT only (no variable-length path),
-    # LIMIT 2000 caps result set on large tenants for sub-second response.
+    # Results are paged (_run_paged) instead of capped, so large tenants
+    # get every violation.
     # ------------------------------------------------------------------ #
     # #1301: Every event node joined via UNDERWENT must be tenant-scoped,
     # not just the Lot. A cross-tenant UNDERWENT edge (possible because of
@@ -165,8 +188,7 @@ async def find_broken_chains(
         shipping.document_id  AS document_id,
         l.tlc                 AS lot_tlc,
         l.product_description AS product_description
-    ORDER BY shipping.event_date
-    LIMIT 2000
+    ORDER BY shipping.event_date, shipping.event_id, l.tlc
     """
 
     # ------------------------------------------------------------------ #
@@ -206,8 +228,7 @@ async def find_broken_chains(
         l.product_description        AS product_description,
         earliest_origin_id           AS origin_event_id,
         earliest_origin_date         AS origin_date
-    ORDER BY shipping.event_date
-    LIMIT 2000
+    ORDER BY shipping.event_date, shipping.event_id, l.tlc
     """
 
     violations: List[Dict[str, Any]] = []
@@ -215,8 +236,7 @@ async def find_broken_chains(
 
     async with client.session() as session:
         # --- Missing origin violations --------------------------------
-        result = await session.run(missing_origin_query, tenant_id=tenant_id)
-        async for record in result:
+        async for record in _run_paged(session, missing_origin_query, tenant_id=tenant_id):
             eid = record["event_id"]
             if eid in seen_event_ids:
                 continue
@@ -236,8 +256,7 @@ async def find_broken_chains(
             )
 
         # --- Temporal paradox violations ------------------------------
-        result = await session.run(temporal_paradox_query, tenant_id=tenant_id)
-        async for record in result:
+        async for record in _run_paged(session, temporal_paradox_query, tenant_id=tenant_id):
             eid = record["event_id"]
             if eid in seen_event_ids:
                 continue
@@ -331,12 +350,10 @@ async def find_broken_chains(
             l.tlc                      AS lot_tlc,
             l.product_description      AS product_description,
             origins                    AS expected_predecessors
-        ORDER BY shipping.event_date
-        LIMIT 2000
+        ORDER BY shipping.event_date, shipping.event_id, l.tlc
         """
 
-        result = await session.run(crypto_gap_query, tenant_id=tenant_id)
-        async for record in result:
+        async for record in _run_paged(session, crypto_gap_query, tenant_id=tenant_id):
             eid = record["event_id"]
             if eid in seen_event_ids:
                 continue
@@ -392,9 +409,7 @@ async def find_all_gaps(
             the underlying gap queries, which require a tenant).
     """
     if not tenant_id or not str(tenant_id).strip():
-        raise ValueError(
-            "find_all_gaps requires a non-empty tenant_id; null-tenant queries are no longer supported"
-        )
+        raise ValueError("find_all_gaps requires a non-empty tenant_id; null-tenant queries are no longer supported")
 
     gaps = await find_gaps(client, tenant_id)
     broken_chains = await find_broken_chains(client, tenant_id)
diff --git a/services/graph/app/fsma/data_quality.py b/services/graph/app/fsma/data_quality.py
index c50b5e087..cae0bea25 100644
--- a/services/graph/app/fsma/data_quality.py
+++ b/services/graph/app/fsma/data_quality.py
@@ -14,6 +14,7 @@
 import structlog
 
 from ..neo4j_utils import Neo4jClient
+from .chain_integrity import _run_paged
 from .types import DataQualityReport, KDECompletenessMetrics, OrphanLot
 
 logger = structlog.get_logger("fsma-utils")
@@ -149,15 +150,12 @@ async def find_orphaned_lots(
         l.created_at as created_at,
         most_recent_event.type as last_event_type,
         most_recent_event.event_date as last_event_date
-    ORDER BY most_recent_event.event_date ASC
-    LIMIT 2000
+    ORDER BY most_recent_event.event_date ASC, l.tlc ASC
     """
 
     orphans = []
     async with client.session() as session:
-        result = await session.run(query, tenant_id=tenant_id, cutoff_date=cutoff_date)
-
-        async for record in result:
+        async for record in _run_paged(session, query, tenant_id=tenant_id, cutoff_date=cutoff_date):
             # Calculate stagnant days. The parsed date is naive, so anchor
             # it to UTC before subtracting: an aware minus naive
             # subtraction raises TypeError, which the fallback below would
diff --git a/services/graph/tests/test_fsma_module_qa.py b/services/graph/tests/test_fsma_module_qa.py
index 64f120d83..72f2db59f 100644
--- a/services/graph/tests/test_fsma_module_qa.py
+++ b/services/graph/tests/test_fsma_module_qa.py
@@ -232,14 +232,16 @@ async def test_exists_check_scopes_lot(self):
 
 class TestBrokenChainsQueries:
     @pytest.mark.asyncio
-    async def test_crypto_gap_query_has_row_cap(self):
-        """All three broken-chain queries must cap rows for large tenants."""
+    async def test_broken_chain_queries_are_paged_not_capped(self):
+        """R1: each broken-chain query fetches bounded pages, never a silent 2000-row cap."""
         client, session = _empty_client()
         await find_broken_chains(client, tenant_id=TENANT)
         assert session.run.call_count == 3
         for i, call in enumerate(session.run.call_args_list):
             cypher = _flatten(call.args[0])
-            assert "LIMIT 2000" in cypher, f"query {i} has no row cap"
+            assert "LIMIT 2000" not in cypher, f"query {i} still has a fixed row cap"
+            assert "SKIP $page_skip LIMIT $page_size" in cypher, f"query {i} is not paged"
+            assert call.kwargs["page_size"] > 0
 
     @pytest.mark.asyncio
     async def test_paradox_query_sorts_origins_before_collect(self):
diff --git a/services/graph/tests/test_graph_row_paging_r1.py b/services/graph/tests/test_graph_row_paging_r1.py
new file mode 100644
index 000000000..b6f72bf17
--- /dev/null
+++ b/services/graph/tests/test_graph_row_paging_r1.py
@@ -0,0 +1,165 @@
+"""R1: chain_integrity / data_quality must not silently cap results at 2000 rows.
+
+Business field values (TLC, product description, quantity, unit, harvest date,
+reference document) are taken verbatim from the checked-in ingestion fixture
+``services/ingestion/tests/fixtures/01_harvesting.csv``. The only generated
+value per row is an opaque synthetic test identifier (``r1-synthetic-NNNNN``)
+used to make rows distinct; it is not business data. The fake Neo4j session
+honours ``$page_skip`` / ``$page_size`` so the tests exercise real paging.
+"""
+
+from __future__ import annotations
+
+import csv
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+sys.path.insert(0, str(Path(__file__).resolve().parents[3]))
+
+from services.graph.app.fsma import chain_integrity
+from services.graph.app.fsma.chain_integrity import find_broken_chains, find_gaps
+from services.graph.app.fsma.data_quality import find_orphaned_lots
+
+TENANT = "11111111-1111-1111-1111-111111111111"
+TOTAL_ROWS = 4501  # > 2 full pages of 2000, plus a partial page
+FIXTURE = Path(__file__).resolve().parents[2] / "ingestion" / "tests" / "fixtures" / "01_harvesting.csv"
+
+
+def _fixture_rows() -> list[dict]:
+    with FIXTURE.open(newline="") as fh:
+        rows = list(csv.DictReader(fh))
+    assert rows, "fixture must not be empty"
+    return rows
+
+
+class _Result:
+    def __init__(self, rows):
+        self._rows = rows
+
+    def __aiter__(self):
+        return self._gen()
+
+    async def _gen(self):
+        for r in self._rows:
+            yield r
+
+
+class _PagingSession:
+    """Fake session: routes by query text, slices rows by SKIP/LIMIT params."""
+
+    def __init__(self, rows_for):
+        self._rows_for = rows_for
+        self.calls: list[tuple[str, dict]] = []
+
+    async def run(self, query, **params):
+        self.calls.append((query, params))
+        assert "LIMIT 2000" not in query
+        assert "SKIP $page_skip LIMIT $page_size" in query
+        rows = self._rows_for(query)
+        skip, size = params["page_skip"], params["page_size"]
+        return _Result(rows[skip : skip + size])
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        return False
+
+
+def _client(session):
+    client = MagicMock()
+    client.session = MagicMock(return_value=session)
+    return client
+
+
+def _shipping_rows(n: int) -> list[dict]:
+    fx = _fixture_rows()
+    out = []
+    for i in range(n):
+        src = fx[i % len(fx)]
+        out.append(
+            {
+                "event_id": f"r1-synthetic-{i:05d}",
+                "type": "SHIPPING",
+                "event_date": src["harvest_date"],
+                "document_id": src["reference_document"],
+                "lot_tlc": src["traceability_lot_code"],
+                "product_description": src["product_description"],
+                "merkle_hash": None,
+                "merkle_prev_hash": None,
+                "expected_predecessors": [],
+                "gaps": "missing_lot",
+            }
+        )
+    return out
+
+
+def test_page_size_is_positive():
+    assert chain_integrity._PAGE_SIZE > 0
+
+
+@pytest.mark.asyncio
+async def test_find_gaps_returns_more_than_2000_rows():
+    rows = _shipping_rows(TOTAL_ROWS)
+    session = _PagingSession(lambda q: rows)
+    gaps = await find_gaps(_client(session), tenant_id=TENANT)
+    assert len(gaps) == TOTAL_ROWS
+    assert len({g["event_id"] for g in gaps}) == TOTAL_ROWS
+    assert [c[1]["page_skip"] for c in session.calls] == [0, 2000, 4000]
+    assert all(c[1]["page_size"] == 2000 for c in session.calls)
+    assert all("ORDER BY event_id" in c[0] for c in session.calls)
+
+
+@pytest.mark.asyncio
+async def test_find_broken_chains_returns_more_than_2000_rows():
+    rows = _shipping_rows(TOTAL_ROWS)
+
+    def rows_for(q):
+        # missing_origin query only; temporal_paradox and crypto_gap empty.
+        if "NOT EXISTS" in q and "merkle_prev_hash" not in q and "origin_date" not in q:
+            return rows
+        return []
+
+    session = _PagingSession(rows_for)
+    violations = await find_broken_chains(_client(session), tenant_id=TENANT)
+    assert len(violations) == TOTAL_ROWS
+    assert {v["lot_tlc"] for v in violations} == {r["traceability_lot_code"] for r in _fixture_rows()}
+    for q, _ in session.calls:
+        assert "ORDER BY shipping.event_date, shipping.event_id, l.tlc" in q
+
+
+@pytest.mark.asyncio
+async def test_find_orphaned_lots_returns_more_than_2000_rows():
+    fx = _fixture_rows()
+    rows = []
+    for i in range(TOTAL_ROWS):
+        src = fx[i % len(fx)]
+        rows.append(
+            {
+                "tlc": src["traceability_lot_code"],
+                "product_description": src["product_description"],
+                "quantity": float(src["quantity"]),
+                "unit_of_measure": src["unit_of_measure"],
+                "created_at": src["harvest_date"],
+                "last_event_type": "RECEIVING",
+                "last_event_date": src["harvest_date"],
+            }
+        )
+    session = _PagingSession(lambda q: rows)
+    orphans = await find_orphaned_lots(_client(session), tenant_id=TENANT)
+    assert len(orphans) == TOTAL_ROWS
+    assert len(session.calls) == 3
+    assert "ORDER BY most_recent_event.event_date ASC, l.tlc ASC" in session.calls[0][0]
+
+
+@pytest.mark.asyncio
+async def test_exact_page_boundary_issues_one_empty_followup():
+    rows = _shipping_rows(2000)
+    session = _PagingSession(lambda q: rows)
+    gaps = await find_gaps(_client(session), tenant_id=TENANT)
+    assert len(gaps) == 2000
+    assert [c[1]["page_skip"] for c in session.calls] == [0, 2000]

From 79d2399653cde0e3f2cf48c744bb1a4b24bbf929 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Sun, 25 Oct 2026 22:42:52 -0700
Subject: [PATCH 2/6] test(graph): synthetic uuid rows for R1 paging tests;
 dedupe gaps and orphaned lots across pages

---
 services/graph/app/fsma/chain_integrity.py    |   6 +
 services/graph/app/fsma/data_quality.py       |   6 +
 .../graph/tests/test_graph_row_paging_r1.py   | 121 ++++++++++--------
 3 files changed, 81 insertions(+), 52 deletions(-)

diff --git a/services/graph/app/fsma/chain_integrity.py b/services/graph/app/fsma/chain_integrity.py
index 704edf316..6a64244d4 100644
--- a/services/graph/app/fsma/chain_integrity.py
+++ b/services/graph/app/fsma/chain_integrity.py
@@ -92,8 +92,14 @@ async def find_gaps(
     """
 
     gaps = []
+    # SKIP/LIMIT pages can overlap if rows shift between pages; keep the
+    # first occurrence of each event so no event is reported twice.
+    seen_event_ids: set = set()
     async with client.session() as session:
         async for record in _run_paged(session, query, tenant_id=tenant_id):
+            if record["event_id"] in seen_event_ids:
+                continue
+            seen_event_ids.add(record["event_id"])
             gaps.append(
                 {
                     "event_id": record["event_id"],
diff --git a/services/graph/app/fsma/data_quality.py b/services/graph/app/fsma/data_quality.py
index cae0bea25..eb81dd521 100644
--- a/services/graph/app/fsma/data_quality.py
+++ b/services/graph/app/fsma/data_quality.py
@@ -154,8 +154,14 @@ async def find_orphaned_lots(
     """
 
     orphans = []
+    # SKIP/LIMIT pages can overlap if rows shift between pages; keep the
+    # first occurrence of each lot so no lot is reported twice.
+    seen_tlcs: set = set()
     async with client.session() as session:
         async for record in _run_paged(session, query, tenant_id=tenant_id, cutoff_date=cutoff_date):
+            if record["tlc"] in seen_tlcs:
+                continue
+            seen_tlcs.add(record["tlc"])
             # Calculate stagnant days. The parsed date is naive, so anchor
             # it to UTC before subtracting: an aware minus naive
             # subtraction raises TypeError, which the fallback below would
diff --git a/services/graph/tests/test_graph_row_paging_r1.py b/services/graph/tests/test_graph_row_paging_r1.py
index b6f72bf17..75a03082e 100644
--- a/services/graph/tests/test_graph_row_paging_r1.py
+++ b/services/graph/tests/test_graph_row_paging_r1.py
@@ -1,17 +1,16 @@
 """R1: chain_integrity / data_quality must not silently cap results at 2000 rows.
 
-Business field values (TLC, product description, quantity, unit, harvest date,
-reference document) are taken verbatim from the checked-in ingestion fixture
-``services/ingestion/tests/fixtures/01_harvesting.csv``. The only generated
-value per row is an opaque synthetic test identifier (``r1-synthetic-NNNNN``)
-used to make rows distinct; it is not business data. The fake Neo4j session
-honours ``$page_skip`` / ``$page_size`` so the tests exercise real paging.
+Rows are fully synthetic: every id is a fresh ``uuid4`` and business fields
+are empty or ``None``. Where a sortable field needs a value, the row index is
+used. The fake Neo4j session honours ``$page_skip`` / ``$page_size`` so the
+tests exercise real paging.
 """
 
 from __future__ import annotations
 
-import csv
 import sys
+import uuid
+from collections import Counter
 from pathlib import Path
 from unittest.mock import MagicMock
 
@@ -24,16 +23,8 @@
 from services.graph.app.fsma.chain_integrity import find_broken_chains, find_gaps
 from services.graph.app.fsma.data_quality import find_orphaned_lots
 
-TENANT = "11111111-1111-1111-1111-111111111111"
+TENANT = str(uuid.uuid4())
 TOTAL_ROWS = 4501  # > 2 full pages of 2000, plus a partial page
-FIXTURE = Path(__file__).resolve().parents[2] / "ingestion" / "tests" / "fixtures" / "01_harvesting.csv"
-
-
-def _fixture_rows() -> list[dict]:
-    with FIXTURE.open(newline="") as fh:
-        rows = list(csv.DictReader(fh))
-    assert rows, "fixture must not be empty"
-    return rows
 
 
 class _Result:
@@ -76,26 +67,28 @@ def _client(session):
     return client
 
 
+def _assert_each_once(input_ids: list[str], output_ids: list[str]) -> None:
+    counts = Counter(output_ids)
+    assert set(counts) == set(input_ids)
+    assert all(n == 1 for n in counts.values())
+
+
 def _shipping_rows(n: int) -> list[dict]:
-    fx = _fixture_rows()
-    out = []
-    for i in range(n):
-        src = fx[i % len(fx)]
-        out.append(
-            {
-                "event_id": f"r1-synthetic-{i:05d}",
-                "type": "SHIPPING",
-                "event_date": src["harvest_date"],
-                "document_id": src["reference_document"],
-                "lot_tlc": src["traceability_lot_code"],
-                "product_description": src["product_description"],
-                "merkle_hash": None,
-                "merkle_prev_hash": None,
-                "expected_predecessors": [],
-                "gaps": "missing_lot",
-            }
-        )
-    return out
+    return [
+        {
+            "event_id": str(uuid.uuid4()),
+            "type": "SHIPPING",
+            "event_date": i,  # sortable: row index
+            "document_id": None,
+            "lot_tlc": str(uuid.uuid4()),
+            "product_description": "",
+            "merkle_hash": None,
+            "merkle_prev_hash": None,
+            "expected_predecessors": [],
+            "gaps": "",
+        }
+        for i in range(n)
+    ]
 
 
 def test_page_size_is_positive():
@@ -108,12 +101,24 @@ async def test_find_gaps_returns_more_than_2000_rows():
     session = _PagingSession(lambda q: rows)
     gaps = await find_gaps(_client(session), tenant_id=TENANT)
     assert len(gaps) == TOTAL_ROWS
-    assert len({g["event_id"] for g in gaps}) == TOTAL_ROWS
+    _assert_each_once([r["event_id"] for r in rows], [g["event_id"] for g in gaps])
     assert [c[1]["page_skip"] for c in session.calls] == [0, 2000, 4000]
     assert all(c[1]["page_size"] == 2000 for c in session.calls)
+    assert all(c[1]["tenant_id"] == TENANT for c in session.calls)
     assert all("ORDER BY event_id" in c[0] for c in session.calls)
 
 
+@pytest.mark.asyncio
+async def test_find_gaps_drops_rows_repeated_across_pages():
+    rows = _shipping_rows(TOTAL_ROWS)
+    # Simulate a row shifting forward between pages: page 2 repeats the
+    # last row of page 1.
+    shifted = rows[:2000] + [rows[1999]] + rows[2000:]
+    session = _PagingSession(lambda q: shifted)
+    gaps = await find_gaps(_client(session), tenant_id=TENANT)
+    _assert_each_once([r["event_id"] for r in rows], [g["event_id"] for g in gaps])
+
+
 @pytest.mark.asyncio
 async def test_find_broken_chains_returns_more_than_2000_rows():
     rows = _shipping_rows(TOTAL_ROWS)
@@ -127,39 +132,51 @@ def rows_for(q):
     session = _PagingSession(rows_for)
     violations = await find_broken_chains(_client(session), tenant_id=TENANT)
     assert len(violations) == TOTAL_ROWS
-    assert {v["lot_tlc"] for v in violations} == {r["traceability_lot_code"] for r in _fixture_rows()}
+    _assert_each_once([r["lot_tlc"] for r in rows], [v["lot_tlc"] for v in violations])
     for q, _ in session.calls:
         assert "ORDER BY shipping.event_date, shipping.event_id, l.tlc" in q
 
 
+def _orphan_rows(n: int) -> list[dict]:
+    return [
+        {
+            "tlc": str(uuid.uuid4()),
+            "product_description": "",
+            "quantity": None,
+            "unit_of_measure": "",
+            "created_at": None,
+            "last_event_type": "",
+            "last_event_date": None,
+        }
+        for _ in range(n)
+    ]
+
+
 @pytest.mark.asyncio
 async def test_find_orphaned_lots_returns_more_than_2000_rows():
-    fx = _fixture_rows()
-    rows = []
-    for i in range(TOTAL_ROWS):
-        src = fx[i % len(fx)]
-        rows.append(
-            {
-                "tlc": src["traceability_lot_code"],
-                "product_description": src["product_description"],
-                "quantity": float(src["quantity"]),
-                "unit_of_measure": src["unit_of_measure"],
-                "created_at": src["harvest_date"],
-                "last_event_type": "RECEIVING",
-                "last_event_date": src["harvest_date"],
-            }
-        )
+    rows = _orphan_rows(TOTAL_ROWS)
     session = _PagingSession(lambda q: rows)
     orphans = await find_orphaned_lots(_client(session), tenant_id=TENANT)
     assert len(orphans) == TOTAL_ROWS
+    _assert_each_once([r["tlc"] for r in rows], [o.tlc for o in orphans])
     assert len(session.calls) == 3
     assert "ORDER BY most_recent_event.event_date ASC, l.tlc ASC" in session.calls[0][0]
 
 
+@pytest.mark.asyncio
+async def test_find_orphaned_lots_drops_rows_repeated_across_pages():
+    rows = _orphan_rows(TOTAL_ROWS)
+    shifted = rows[:2000] + [rows[1999]] + rows[2000:]
+    session = _PagingSession(lambda q: shifted)
+    orphans = await find_orphaned_lots(_client(session), tenant_id=TENANT)
+    _assert_each_once([r["tlc"] for r in rows], [o.tlc for o in orphans])
+
+
 @pytest.mark.asyncio
 async def test_exact_page_boundary_issues_one_empty_followup():
     rows = _shipping_rows(2000)
     session = _PagingSession(lambda q: rows)
     gaps = await find_gaps(_client(session), tenant_id=TENANT)
     assert len(gaps) == 2000
+    _assert_each_once([r["event_id"] for r in rows], [g["event_id"] for g in gaps])
     assert [c[1]["page_skip"] for c in session.calls] == [0, 2000]

From f26bc62ae0f71fffb2c115346c54f40cffd3d8b9 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Sun, 25 Oct 2026 22:48:04 -0700
Subject: [PATCH 3/6] fix(graph): keyset-page FSMA gap/chain/orphan queries so
 mid-read deletes cannot skip rows (R1)

---
 services/graph/app/fsma/chain_integrity.py    | 107 +++++++++++++-----
 services/graph/app/fsma/data_quality.py       |  15 ++-
 services/graph/tests/test_fsma_module_qa.py   |   4 +-
 .../graph/tests/test_graph_row_paging_r1.py   |  94 +++++++++++++--
 4 files changed, 174 insertions(+), 46 deletions(-)

diff --git a/services/graph/app/fsma/chain_integrity.py b/services/graph/app/fsma/chain_integrity.py
index 6a64244d4..0a343e112 100644
--- a/services/graph/app/fsma/chain_integrity.py
+++ b/services/graph/app/fsma/chain_integrity.py
@@ -8,7 +8,7 @@
 from __future__ import annotations
 
 import time
-from typing import Any, AsyncIterator, Dict, List, Optional
+from typing import Any, AsyncIterator, Callable, Dict, List, Optional, Sequence
 
 import structlog
 from shared.fsma_rules import TimeArrowRule, TraceEvent as SharedTraceEvent
@@ -18,29 +18,79 @@
 logger = structlog.get_logger("fsma-utils")
 
 
-# Rows fetched per round trip. Results are read page by page (SKIP/LIMIT on a
-# stable ORDER BY) so every matching row is reported; nothing is silently
-# dropped at a fixed cap, and each query round trip stays bounded.
+# Rows fetched per round trip. Results are read page by page with keyset
+# pagination on a unique key, so every matching row is reported, nothing is
+# silently dropped at a fixed cap, and each round trip stays bounded. Unlike
+# SKIP/offset paging, a row deleted (or inserted) between pages cannot shift
+# the next page past a row that has not been read yet: each page starts
+# strictly after the last key already returned.
 _PAGE_SIZE = 2000
 
 
-async def _run_paged(session: Any, query: str, **params: Any) -> AsyncIterator[Any]:
+def _sort_key(*fields: str) -> Callable[[Any], tuple]:
+    """Python sort key for records: ascending, nulls last (Neo4j ASC order)."""
+
+    def key(record: Any) -> tuple:
+        return tuple((record[f] is None, "" if record[f] is None else str(record[f])) for f in fields)
+
+    return key
+
+
+def _after_condition(width: int) -> str:
+    """Cypher for ``_page_key > $after_key``, compared element by element."""
+    terms = []
+    for i in range(width):
+        equal = [f"_page_key[{j}] = $after_key[{j}]" for j in range(i)]
+        terms.append("(" + " AND ".join(equal + [f"_page_key[{i}] > $after_key[{i}]"]) + ")")
+    return " OR ".join(terms)
+
+
+async def _run_paged(
+    session: Any,
+    query: str,
+    *,
+    key: Sequence[str],
+    order: Optional[Callable[[Any], Any]] = None,
+    **params: Any,
+) -> AsyncIterator[Any]:
     """Yield every record of ``query``, fetched ``_PAGE_SIZE`` rows at a time.
 
-    ``query`` must end with a deterministic ORDER BY (unique tiebreaker) and
-    no SKIP/LIMIT; paging is appended here.
+    ``query`` must end with a RETURN and no ORDER BY / SKIP / LIMIT. ``key``
+    names returned columns that are unique together; pages are ordered by
+    them and each page starts strictly after the last key read. Records are
+    yielded once all pages are read, sorted by ``order`` when given.
     """
-    paged_query = query.rstrip() + "\n    SKIP $page_skip LIMIT $page_size\n"
-    skip = 0
+    key_expr = ", ".join(f"coalesce(toString({c}), '')" for c in key)
+    paged_query = (
+        "CALL {\n"
+        + query.rstrip()
+        + "\n}\n"
+        + f"WITH *, [{key_expr}] AS _page_key\n"
+        + f"WHERE $after_key IS NULL OR {_after_condition(len(key))}\n"
+        + "RETURN *\n"
+        + "ORDER BY _page_key[0]"
+        + "".join(f", _page_key[{i}]" for i in range(1, len(key)))
+        + "\nLIMIT $page_size\n"
+    )
+    records: List[Any] = []
+    after_key: Optional[List[str]] = None
     while True:
-        result = await session.run(paged_query, page_skip=skip, page_size=_PAGE_SIZE, **params)
-        fetched = 0
-        async for record in result:
-            fetched += 1
-            yield record
-        if fetched < _PAGE_SIZE:
-            return
-        skip += fetched
+        result = await session.run(paged_query, after_key=after_key, page_size=_PAGE_SIZE, **params)
+        page = [record async for record in result]
+        records.extend(page)
+        if len(page) < _PAGE_SIZE:
+            break
+        after_key = list(page[-1]["_page_key"])
+    if order is not None:
+        records.sort(key=order)
+    for record in records:
+        yield record
+
+
+# Broken-chain rows are (shipping event, lot) pairs: unique together. They
+# are reported in event-date order, then event id, then lot.
+_CHAIN_KEY = ("event_id", "lot_tlc")
+_CHAIN_ORDER = _sort_key("event_date", "event_id", "lot_tlc")
 
 
 async def find_gaps(
@@ -88,15 +138,13 @@ async def find_gaps(
         CASE WHEN e.event_date IS NULL OR e.event_date = '' THEN 'missing_date' ELSE '' END +
         CASE WHEN NOT EXISTS { MATCH (e)<-[:UNDERWENT]-(l:Lot) WHERE l.tenant_id = $tenant_id } THEN ',missing_lot' ELSE '' END +
         CASE WHEN e.responsible_party_contact IS NULL OR e.responsible_party_contact = '' THEN ',missing_responsible_party_contact' ELSE '' END as gaps
-    ORDER BY event_id
     """
 
     gaps = []
-    # SKIP/LIMIT pages can overlap if rows shift between pages; keep the
-    # first occurrence of each event so no event is reported twice.
+    # Keep the first occurrence of each event so no event is reported twice.
     seen_event_ids: set = set()
     async with client.session() as session:
-        async for record in _run_paged(session, query, tenant_id=tenant_id):
+        async for record in _run_paged(session, query, key=("event_id",), tenant_id=tenant_id):
             if record["event_id"] in seen_event_ids:
                 continue
             seen_event_ids.add(record["event_id"])
@@ -168,7 +216,7 @@ async def find_broken_chains(
     # ------------------------------------------------------------------ #
     # Query 1: SHIPPING with no origin event at all
     # Optimization: single-hop UNDERWENT only (no variable-length path),
-    # Results are paged (_run_paged) instead of capped, so large tenants
+    # Results are keyset-paged (_run_paged) instead of capped, so large tenants
     # get every violation.
     # ------------------------------------------------------------------ #
     # #1301: Every event node joined via UNDERWENT must be tenant-scoped,
@@ -194,7 +242,6 @@ async def find_broken_chains(
         shipping.document_id  AS document_id,
         l.tlc                 AS lot_tlc,
         l.product_description AS product_description
-    ORDER BY shipping.event_date, shipping.event_id, l.tlc
     """
 
     # ------------------------------------------------------------------ #
@@ -234,7 +281,6 @@ async def find_broken_chains(
         l.product_description        AS product_description,
         earliest_origin_id           AS origin_event_id,
         earliest_origin_date         AS origin_date
-    ORDER BY shipping.event_date, shipping.event_id, l.tlc
     """
 
     violations: List[Dict[str, Any]] = []
@@ -242,7 +288,9 @@ async def find_broken_chains(
 
     async with client.session() as session:
         # --- Missing origin violations --------------------------------
-        async for record in _run_paged(session, missing_origin_query, tenant_id=tenant_id):
+        async for record in _run_paged(
+            session, missing_origin_query, key=_CHAIN_KEY, order=_CHAIN_ORDER, tenant_id=tenant_id
+        ):
             eid = record["event_id"]
             if eid in seen_event_ids:
                 continue
@@ -262,7 +310,9 @@ async def find_broken_chains(
             )
 
         # --- Temporal paradox violations ------------------------------
-        async for record in _run_paged(session, temporal_paradox_query, tenant_id=tenant_id):
+        async for record in _run_paged(
+            session, temporal_paradox_query, key=_CHAIN_KEY, order=_CHAIN_ORDER, tenant_id=tenant_id
+        ):
             eid = record["event_id"]
             if eid in seen_event_ids:
                 continue
@@ -356,10 +406,11 @@ async def find_broken_chains(
             l.tlc                      AS lot_tlc,
             l.product_description      AS product_description,
             origins                    AS expected_predecessors
-        ORDER BY shipping.event_date, shipping.event_id, l.tlc
         """
 
-        async for record in _run_paged(session, crypto_gap_query, tenant_id=tenant_id):
+        async for record in _run_paged(
+            session, crypto_gap_query, key=_CHAIN_KEY, order=_CHAIN_ORDER, tenant_id=tenant_id
+        ):
             eid = record["event_id"]
             if eid in seen_event_ids:
                 continue
diff --git a/services/graph/app/fsma/data_quality.py b/services/graph/app/fsma/data_quality.py
index eb81dd521..f05cfa9c1 100644
--- a/services/graph/app/fsma/data_quality.py
+++ b/services/graph/app/fsma/data_quality.py
@@ -14,7 +14,7 @@
 import structlog
 
 from ..neo4j_utils import Neo4jClient
-from .chain_integrity import _run_paged
+from .chain_integrity import _run_paged, _sort_key
 from .types import DataQualityReport, KDECompletenessMetrics, OrphanLot
 
 logger = structlog.get_logger("fsma-utils")
@@ -150,15 +150,20 @@ async def find_orphaned_lots(
         l.created_at as created_at,
         most_recent_event.type as last_event_type,
         most_recent_event.event_date as last_event_date
-    ORDER BY most_recent_event.event_date ASC, l.tlc ASC
     """
 
     orphans = []
-    # SKIP/LIMIT pages can overlap if rows shift between pages; keep the
-    # first occurrence of each lot so no lot is reported twice.
+    # Keep the first occurrence of each lot so no lot is reported twice.
     seen_tlcs: set = set()
     async with client.session() as session:
-        async for record in _run_paged(session, query, tenant_id=tenant_id, cutoff_date=cutoff_date):
+        async for record in _run_paged(
+            session,
+            query,
+            key=("tlc",),
+            order=_sort_key("last_event_date", "tlc"),
+            tenant_id=tenant_id,
+            cutoff_date=cutoff_date,
+        ):
             if record["tlc"] in seen_tlcs:
                 continue
             seen_tlcs.add(record["tlc"])
diff --git a/services/graph/tests/test_fsma_module_qa.py b/services/graph/tests/test_fsma_module_qa.py
index 72f2db59f..b3f5cfdab 100644
--- a/services/graph/tests/test_fsma_module_qa.py
+++ b/services/graph/tests/test_fsma_module_qa.py
@@ -240,7 +240,9 @@ async def test_broken_chain_queries_are_paged_not_capped(self):
         for i, call in enumerate(session.run.call_args_list):
             cypher = _flatten(call.args[0])
             assert "LIMIT 2000" not in cypher, f"query {i} still has a fixed row cap"
-            assert "SKIP $page_skip LIMIT $page_size" in cypher, f"query {i} is not paged"
+            assert "LIMIT $page_size" in cypher, f"query {i} is not paged"
+            assert "SKIP" not in cypher, f"query {i} uses offset paging"
+            assert "$after_key" in cypher, f"query {i} is not keyset-paged"
             assert call.kwargs["page_size"] > 0
 
     @pytest.mark.asyncio
diff --git a/services/graph/tests/test_graph_row_paging_r1.py b/services/graph/tests/test_graph_row_paging_r1.py
index 75a03082e..50e43d35d 100644
--- a/services/graph/tests/test_graph_row_paging_r1.py
+++ b/services/graph/tests/test_graph_row_paging_r1.py
@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+import re
 import sys
 import uuid
 from collections import Counter
@@ -40,19 +41,37 @@ async def _gen(self):
 
 
 class _PagingSession:
-    """Fake session: routes by query text, slices rows by SKIP/LIMIT params."""
+    """Fake session: routes by query text and serves keyset pages.
 
-    def __init__(self, rows_for):
+    The page key columns are read from the generated Cypher; each page holds
+    the rows whose key is strictly after ``$after_key``, sorted by key.
+    ``on_page(n, rows)`` runs after page ``n`` is served, so a test can
+    delete rows between pages.
+    """
+
+    def __init__(self, rows_for, on_page=None):
         self._rows_for = rows_for
+        self._on_page = on_page
         self.calls: list[tuple[str, dict]] = []
 
     async def run(self, query, **params):
         self.calls.append((query, params))
         assert "LIMIT 2000" not in query
-        assert "SKIP $page_skip LIMIT $page_size" in query
+        assert "SKIP" not in query
+        assert "LIMIT $page_size" in query
+        cols = re.findall(r"coalesce\(toString\((\w+)\), ''\)", query)
+        assert cols, query
         rows = self._rows_for(query)
-        skip, size = params["page_skip"], params["page_size"]
-        return _Result(rows[skip : skip + size])
+
+        def key(r):
+            return ["" if r[c] is None else str(r[c]) for c in cols]
+
+        after = params["after_key"]
+        page = sorted((r for r in rows if after is None or key(r) > after), key=key)
+        page = [{**r, "_page_key": key(r)} for r in page[: params["page_size"]]]
+        if self._on_page:
+            self._on_page(len(self.calls), rows)
+        return _Result(page)
 
     async def __aenter__(self):
         return self
@@ -78,7 +97,7 @@ def _shipping_rows(n: int) -> list[dict]:
         {
             "event_id": str(uuid.uuid4()),
             "type": "SHIPPING",
-            "event_date": i,  # sortable: row index
+            "event_date": f"{i:05d}",  # sortable: row index
             "document_id": None,
             "lot_tlc": str(uuid.uuid4()),
             "product_description": "",
@@ -102,10 +121,10 @@ async def test_find_gaps_returns_more_than_2000_rows():
     gaps = await find_gaps(_client(session), tenant_id=TENANT)
     assert len(gaps) == TOTAL_ROWS
     _assert_each_once([r["event_id"] for r in rows], [g["event_id"] for g in gaps])
-    assert [c[1]["page_skip"] for c in session.calls] == [0, 2000, 4000]
+    assert len(session.calls) == 3
+    assert session.calls[0][1]["after_key"] is None
     assert all(c[1]["page_size"] == 2000 for c in session.calls)
     assert all(c[1]["tenant_id"] == TENANT for c in session.calls)
-    assert all("ORDER BY event_id" in c[0] for c in session.calls)
 
 
 @pytest.mark.asyncio
@@ -133,8 +152,8 @@ def rows_for(q):
     violations = await find_broken_chains(_client(session), tenant_id=TENANT)
     assert len(violations) == TOTAL_ROWS
     _assert_each_once([r["lot_tlc"] for r in rows], [v["lot_tlc"] for v in violations])
-    for q, _ in session.calls:
-        assert "ORDER BY shipping.event_date, shipping.event_id, l.tlc" in q
+    # Reported in event-date order (the row index here).
+    assert [v["event_date"] for v in violations] == [f"{i:05d}" for i in range(TOTAL_ROWS)]
 
 
 def _orphan_rows(n: int) -> list[dict]:
@@ -160,7 +179,7 @@ async def test_find_orphaned_lots_returns_more_than_2000_rows():
     assert len(orphans) == TOTAL_ROWS
     _assert_each_once([r["tlc"] for r in rows], [o.tlc for o in orphans])
     assert len(session.calls) == 3
-    assert "ORDER BY most_recent_event.event_date ASC, l.tlc ASC" in session.calls[0][0]
+    assert [o.tlc for o in orphans] == sorted(r["tlc"] for r in rows)
 
 
 @pytest.mark.asyncio
@@ -179,4 +198,55 @@ async def test_exact_page_boundary_issues_one_empty_followup():
     gaps = await find_gaps(_client(session), tenant_id=TENANT)
     assert len(gaps) == 2000
     _assert_each_once([r["event_id"] for r in rows], [g["event_id"] for g in gaps])
-    assert [c[1]["page_skip"] for c in session.calls] == [0, 2000]
+    assert len(session.calls) == 2
+
+
+def _delete_after_first_page(victims):
+    def on_page(n, rows):
+        if n == 1:
+            for v in victims:
+                rows.remove(v)
+
+    return on_page
+
+
+@pytest.mark.asyncio
+async def test_find_gaps_deletion_between_pages_skips_nothing():
+    rows = _shipping_rows(TOTAL_ROWS)
+    ordered = sorted(rows, key=lambda r: r["event_id"])
+    # One already-read row and one unread row vanish after page 1. With
+    # offset paging the next page would start one row late and skip a row.
+    read, unread = ordered[10], ordered[3000]
+    session = _PagingSession(lambda q: rows, on_page=_delete_after_first_page([read, unread]))
+    expected = [r["event_id"] for r in rows if r is not unread]
+    gaps = await find_gaps(_client(session), tenant_id=TENANT)
+    _assert_each_once(expected, [g["event_id"] for g in gaps])
+
+
+@pytest.mark.asyncio
+async def test_find_broken_chains_deletion_between_pages_skips_nothing():
+    rows = _shipping_rows(TOTAL_ROWS)
+    ordered = sorted(rows, key=lambda r: (r["event_id"], r["lot_tlc"]))
+    read = ordered[0]
+
+    def rows_for(q):
+        if "NOT EXISTS" in q and "merkle_prev_hash" not in q and "origin_date" not in q:
+            return rows
+        return []
+
+    unread = ordered[2500]
+    session = _PagingSession(rows_for, on_page=_delete_after_first_page([read, unread]))
+    expected = [r["event_id"] for r in rows if r is not unread]
+    violations = await find_broken_chains(_client(session), tenant_id=TENANT)
+    _assert_each_once(expected, [v["event_id"] for v in violations])
+
+
+@pytest.mark.asyncio
+async def test_find_orphaned_lots_deletion_between_pages_skips_nothing():
+    rows = _orphan_rows(TOTAL_ROWS)
+    ordered = sorted(rows, key=lambda r: r["tlc"])
+    read, unread = ordered[5], ordered[4400]
+    session = _PagingSession(lambda q: rows, on_page=_delete_after_first_page([read, unread]))
+    expected = [r["tlc"] for r in rows if r is not unread]
+    orphans = await find_orphaned_lots(_client(session), tenant_id=TENANT)
+    _assert_each_once(expected, [o.tlc for o in orphans])

From 996cdf11ffaef110281447c9dd61c739d69a6887 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Sun, 25 Oct 2026 23:13:42 -0700
Subject: [PATCH 4/6] graph: push keyset cursor and LIMIT into the match; test
 in-page dedupe

---
 services/graph/app/fsma/chain_integrity.py    | 93 ++++++++++++-------
 services/graph/app/fsma/data_quality.py       | 18 ++--
 .../graph/tests/test_graph_row_paging_r1.py   | 86 +++++++++++++++--
 3 files changed, 152 insertions(+), 45 deletions(-)

diff --git a/services/graph/app/fsma/chain_integrity.py b/services/graph/app/fsma/chain_integrity.py
index 0a343e112..fb63d3b3e 100644
--- a/services/graph/app/fsma/chain_integrity.py
+++ b/services/graph/app/fsma/chain_integrity.py
@@ -45,39 +45,52 @@ def _after_condition(width: int) -> str:
     return " OR ".join(terms)
 
 
+_PAGE_MARKER = "// @page"
+
+
+def _hit(record: Any) -> bool:
+    """True unless the row carries a false ``_hit`` flag (anchor kept only for paging)."""
+    try:
+        return record["_hit"] is not False
+    except (KeyError, LookupError):
+        return True
+
+
 async def _run_paged(
     session: Any,
     query: str,
     *,
+    anchors: str,
     key: Sequence[str],
     order: Optional[Callable[[Any], Any]] = None,
     **params: Any,
 ) -> AsyncIterator[Any]:
-    """Yield every record of ``query``, fetched ``_PAGE_SIZE`` rows at a time.
-
-    ``query`` must end with a RETURN and no ORDER BY / SKIP / LIMIT. ``key``
-    names returned columns that are unique together; pages are ordered by
-    them and each page starts strictly after the last key read. Records are
-    yielded once all pages are read, sorted by ``order`` when given.
+    """Yield every record of ``query``, fetched ``_PAGE_SIZE`` anchors at a time.
+
+    ``query`` holds a ``// @page`` line right after the MATCH that binds the
+    ``anchors`` variables. The keyset filter, ORDER BY and LIMIT are put
+    there, so each page reads only its own anchors instead of the whole match.
+    ``key`` lists Cypher expressions over the anchors that are unique
+    together. The rest of the query must carry ``_page_key`` along and return
+    exactly one row per anchor; a row whose ``_hit`` is false is used only to
+    advance the cursor and is not yielded. Records are yielded once all pages
+    are read, sorted by ``order`` when given.
     """
+    if _PAGE_MARKER not in query:
+        raise ValueError("paged query needs a // @page marker")
     key_expr = ", ".join(f"coalesce(toString({c}), '')" for c in key)
-    paged_query = (
-        "CALL {\n"
-        + query.rstrip()
-        + "\n}\n"
-        + f"WITH *, [{key_expr}] AS _page_key\n"
-        + f"WHERE $after_key IS NULL OR {_after_condition(len(key))}\n"
-        + "RETURN *\n"
-        + "ORDER BY _page_key[0]"
-        + "".join(f", _page_key[{i}]" for i in range(1, len(key)))
-        + "\nLIMIT $page_size\n"
+    page_clause = (
+        f"WITH {anchors}, [{key_expr}] AS _page_key\n"
+        f"WHERE $after_key IS NULL OR {_after_condition(len(key))}\n"
+        "ORDER BY _page_key[0]" + "".join(f", _page_key[{i}]" for i in range(1, len(key))) + "\nLIMIT $page_size"
     )
+    paged_query = query.replace(_PAGE_MARKER, page_clause, 1)
     records: List[Any] = []
     after_key: Optional[List[str]] = None
     while True:
         result = await session.run(paged_query, after_key=after_key, page_size=_PAGE_SIZE, **params)
         page = [record async for record in result]
-        records.extend(page)
+        records.extend(r for r in page if _hit(r))
         if len(page) < _PAGE_SIZE:
             break
         after_key = list(page[-1]["_page_key"])
@@ -89,7 +102,8 @@ async def _run_paged(
 
 # Broken-chain rows are (shipping event, lot) pairs: unique together. They
 # are reported in event-date order, then event id, then lot.
-_CHAIN_KEY = ("event_id", "lot_tlc")
+_CHAIN_ANCHORS = "l, shipping"
+_CHAIN_KEY = ("shipping.event_id", "l.tlc")
 _CHAIN_ORDER = _sort_key("event_date", "event_id", "lot_tlc")
 
 
@@ -130,7 +144,9 @@ async def find_gaps(
         e.responsible_party_contact IS NULL OR
         e.responsible_party_contact = ''
     )
+    // @page
     RETURN
+        _page_key,
         e.event_id as event_id,
         e.type as type,
         e.event_date as event_date,
@@ -144,7 +160,7 @@ async def find_gaps(
     # Keep the first occurrence of each event so no event is reported twice.
     seen_event_ids: set = set()
     async with client.session() as session:
-        async for record in _run_paged(session, query, key=("event_id",), tenant_id=tenant_id):
+        async for record in _run_paged(session, query, anchors="e", key=("e.event_id",), tenant_id=tenant_id):
             if record["event_id"] in seen_event_ids:
                 continue
             seen_event_ids.add(record["event_id"])
@@ -235,7 +251,9 @@ async def find_broken_chains(
         WHERE origin.type IN ['CREATION', 'RECEIVING', 'INITIAL_PACKING', 'TRANSFORMATION']
           AND origin.tenant_id = $tenant_id
     }
+    // @page
     RETURN
+        _page_key,
         shipping.event_id    AS event_id,
         shipping.type         AS type,
         shipping.event_date   AS event_date,
@@ -258,8 +276,9 @@ async def find_broken_chains(
     WHERE l.tenant_id = $tenant_id
       AND shipping.tenant_id = $tenant_id
       AND shipping.event_date IS NOT NULL
-    WITH l, shipping
-    MATCH (l)-[:UNDERWENT]->(origin:TraceEvent)
+    // @page
+    // OPTIONAL so every paged (lot, shipping) anchor yields one row.
+    OPTIONAL MATCH (l)-[:UNDERWENT]->(origin:TraceEvent)
     WHERE origin.type IN ['CREATION', 'RECEIVING', 'INITIAL_PACKING', 'TRANSFORMATION']
       AND origin.tenant_id = $tenant_id
       AND origin.event_date IS NOT NULL
@@ -267,12 +286,13 @@ async def find_broken_chains(
     // encounter order, so head(collect(...)) is the id of the earliest
     // origin matching min(origin.event_date). Without this the reported
     // origin_event_id can point at a different origin than origin_date.
-    WITH l, shipping, origin ORDER BY origin.event_date ASC
-    WITH l, shipping,
+    WITH l, shipping, _page_key, origin ORDER BY origin.event_date ASC
+    WITH l, shipping, _page_key,
          min(origin.event_date) AS earliest_origin_date,
          head(collect(origin.event_id)) AS earliest_origin_id
-    WHERE earliest_origin_date > shipping.event_date
     RETURN
+        _page_key,
+        coalesce(earliest_origin_date > shipping.event_date, false) AS _hit,
         shipping.event_id            AS event_id,
         shipping.type                AS type,
         shipping.event_date          AS event_date,
@@ -289,7 +309,12 @@ async def find_broken_chains(
     async with client.session() as session:
         # --- Missing origin violations --------------------------------
         async for record in _run_paged(
-            session, missing_origin_query, key=_CHAIN_KEY, order=_CHAIN_ORDER, tenant_id=tenant_id
+            session,
+            missing_origin_query,
+            anchors=_CHAIN_ANCHORS,
+            key=_CHAIN_KEY,
+            order=_CHAIN_ORDER,
+            tenant_id=tenant_id,
         ):
             eid = record["event_id"]
             if eid in seen_event_ids:
@@ -311,7 +336,12 @@ async def find_broken_chains(
 
         # --- Temporal paradox violations ------------------------------
         async for record in _run_paged(
-            session, temporal_paradox_query, key=_CHAIN_KEY, order=_CHAIN_ORDER, tenant_id=tenant_id
+            session,
+            temporal_paradox_query,
+            anchors=_CHAIN_ANCHORS,
+            key=_CHAIN_KEY,
+            order=_CHAIN_ORDER,
+            tenant_id=tenant_id,
         ):
             eid = record["event_id"]
             if eid in seen_event_ids:
@@ -383,20 +413,21 @@ async def find_broken_chains(
             WHERE origin.type IN ['CREATION', 'RECEIVING', 'INITIAL_PACKING']
               AND origin.tenant_id = $tenant_id
         }
-        WITH l, shipping
+        // @page
         MATCH (l)-[:UNDERWENT]->(origin:TraceEvent)
         WHERE origin.type IN ['CREATION', 'RECEIVING', 'INITIAL_PACKING']
           AND origin.tenant_id = $tenant_id
-        WITH shipping, l,
+        WITH shipping, l, _page_key,
              collect({
                  event_id: origin.event_id,
                  type: origin.type,
                  event_date: origin.event_date,
                  merkle_hash: origin.merkle_hash
              }) AS origins
-        WHERE shipping.merkle_prev_hash IS NULL
-           OR NOT any(o IN origins WHERE o.merkle_hash = shipping.merkle_prev_hash)
         RETURN
+            _page_key,
+            (shipping.merkle_prev_hash IS NULL
+             OR NOT any(o IN origins WHERE o.merkle_hash = shipping.merkle_prev_hash)) AS _hit,
             shipping.event_id          AS event_id,
             shipping.type              AS type,
             shipping.event_date        AS event_date,
@@ -409,7 +440,7 @@ async def find_broken_chains(
         """
 
         async for record in _run_paged(
-            session, crypto_gap_query, key=_CHAIN_KEY, order=_CHAIN_ORDER, tenant_id=tenant_id
+            session, crypto_gap_query, anchors=_CHAIN_ANCHORS, key=_CHAIN_KEY, order=_CHAIN_ORDER, tenant_id=tenant_id
         ):
             eid = record["event_id"]
             if eid in seen_event_ids:
diff --git a/services/graph/app/fsma/data_quality.py b/services/graph/app/fsma/data_quality.py
index f05cfa9c1..e28f6a460 100644
--- a/services/graph/app/fsma/data_quality.py
+++ b/services/graph/app/fsma/data_quality.py
@@ -129,20 +129,21 @@ async def find_orphaned_lots(
         MATCH (trans:TraceEvent)-[:CONSUMED]->(l)
         WHERE trans.tenant_id = $tenant_id
     }
+    // @page
 
     // Get the most recent event for this lot
     OPTIONAL MATCH (l)-[:UNDERWENT]->(last_event:TraceEvent)
     WHERE last_event.tenant_id = $tenant_id
-    WITH l, last_event
+    WITH l, _page_key, last_event
     ORDER BY last_event.event_date DESC
-    WITH l, collect(last_event)[0] as most_recent_event
-
-    // Filter by stagnant days
-    WHERE most_recent_event IS NULL
-       OR most_recent_event.event_date IS NULL
-       OR most_recent_event.event_date <= $cutoff_date
+    WITH l, _page_key, collect(last_event)[0] as most_recent_event
 
+    // Stagnant-days filter, as a flag so every paged lot yields one row
     RETURN
+        _page_key,
+        (most_recent_event IS NULL
+         OR most_recent_event.event_date IS NULL
+         OR most_recent_event.event_date <= $cutoff_date) AS _hit,
         l.tlc as tlc,
         l.product_description as product_description,
         l.quantity as quantity,
@@ -159,7 +160,8 @@ async def find_orphaned_lots(
         async for record in _run_paged(
             session,
             query,
-            key=("tlc",),
+            anchors="l",
+            key=("l.tlc",),
             order=_sort_key("last_event_date", "tlc"),
             tenant_id=tenant_id,
             cutoff_date=cutoff_date,
diff --git a/services/graph/tests/test_graph_row_paging_r1.py b/services/graph/tests/test_graph_row_paging_r1.py
index 50e43d35d..c0280d6a1 100644
--- a/services/graph/tests/test_graph_row_paging_r1.py
+++ b/services/graph/tests/test_graph_row_paging_r1.py
@@ -2,8 +2,8 @@
 
 Rows are fully synthetic: every id is a fresh ``uuid4`` and business fields
 are empty or ``None``. Where a sortable field needs a value, the row index is
-used. The fake Neo4j session honours ``$page_skip`` / ``$page_size`` so the
-tests exercise real paging.
+used. The fake Neo4j session honours ``$after_key`` / ``$page_size`` so the
+tests exercise real keyset paging.
 """
 
 from __future__ import annotations
@@ -46,12 +46,14 @@ class _PagingSession:
     The page key columns are read from the generated Cypher; each page holds
     the rows whose key is strictly after ``$after_key``, sorted by key.
     ``on_page(n, rows)`` runs after page ``n`` is served, so a test can
-    delete rows between pages.
+    delete rows between pages. ``dup_in_page(n, page)`` may return a changed
+    page ``n`` (for example one row served twice) before it is sent.
     """
 
-    def __init__(self, rows_for, on_page=None):
+    def __init__(self, rows_for, on_page=None, dup_in_page=None):
         self._rows_for = rows_for
         self._on_page = on_page
+        self._dup_in_page = dup_in_page
         self.calls: list[tuple[str, dict]] = []
 
     async def run(self, query, **params):
@@ -59,9 +61,19 @@ async def run(self, query, **params):
         assert "LIMIT 2000" not in query
         assert "SKIP" not in query
         assert "LIMIT $page_size" in query
-        cols = re.findall(r"coalesce\(toString\((\w+)\), ''\)", query)
-        assert cols, query
+        assert "CALL {" not in query
+        assert "// @page" not in query
+        exprs = re.findall(r"coalesce\(toString\((\w+\.\w+)\), ''\)", query)
+        assert exprs, query
         rows = self._rows_for(query)
+        # The cursor filter runs on anchor properties (``l.tlc``); map each to
+        # the returned column (``tlc``, or ``lot_tlc`` for chain rows).
+        cols = []
+        for expr in exprs:
+            col = expr.split(".")[1]
+            if rows and col not in rows[0]:
+                col = f"lot_{col}"
+            cols.append(col)
 
         def key(r):
             return ["" if r[c] is None else str(r[c]) for c in cols]
@@ -69,6 +81,8 @@ def key(r):
         after = params["after_key"]
         page = sorted((r for r in rows if after is None or key(r) > after), key=key)
         page = [{**r, "_page_key": key(r)} for r in page[: params["page_size"]]]
+        if self._dup_in_page:
+            page = self._dup_in_page(len(self.calls), page)
         if self._on_page:
             self._on_page(len(self.calls), rows)
         return _Result(page)
@@ -250,3 +264,63 @@ async def test_find_orphaned_lots_deletion_between_pages_skips_nothing():
     expected = [r["tlc"] for r in rows if r is not unread]
     orphans = await find_orphaned_lots(_client(session), tenant_id=TENANT)
     _assert_each_once(expected, [o.tlc for o in orphans])
+
+
+def _repeat_row_in_first_page(n, page):
+    # Serve row 4 a second time, inside page 1, before the cursor moves on.
+    if n == 1:
+        return page[:5] + [dict(page[4])] + page[5:]
+    return page
+
+
+@pytest.mark.asyncio
+async def test_find_gaps_drops_row_repeated_within_one_page():
+    rows = _shipping_rows(TOTAL_ROWS)
+    session = _PagingSession(lambda q: rows, dup_in_page=_repeat_row_in_first_page)
+    gaps = await find_gaps(_client(session), tenant_id=TENANT)
+    _assert_each_once([r["event_id"] for r in rows], [g["event_id"] for g in gaps])
+
+
+@pytest.mark.asyncio
+async def test_find_broken_chains_drops_row_repeated_within_one_page():
+    rows = _shipping_rows(TOTAL_ROWS)
+
+    def rows_for(q):
+        if "NOT EXISTS" in q and "merkle_prev_hash" not in q and "origin_date" not in q:
+            return rows
+        return []
+
+    session = _PagingSession(rows_for, dup_in_page=_repeat_row_in_first_page)
+    violations = await find_broken_chains(_client(session), tenant_id=TENANT)
+    _assert_each_once([r["event_id"] for r in rows], [v["event_id"] for v in violations])
+
+
+@pytest.mark.asyncio
+async def test_find_orphaned_lots_drops_row_repeated_within_one_page():
+    rows = _orphan_rows(TOTAL_ROWS)
+    session = _PagingSession(lambda q: rows, dup_in_page=_repeat_row_in_first_page)
+    orphans = await find_orphaned_lots(_client(session), tenant_id=TENANT)
+    _assert_each_once([r["tlc"] for r in rows], [o.tlc for o in orphans])
+
+
+@pytest.mark.asyncio
+async def test_non_hit_anchor_rows_advance_cursor_but_are_not_reported():
+    rows = _orphan_rows(TOTAL_ROWS)
+    for i, r in enumerate(rows):
+        r["_hit"] = i % 2 == 0
+    session = _PagingSession(lambda q: rows)
+    orphans = await find_orphaned_lots(_client(session), tenant_id=TENANT)
+    _assert_each_once([r["tlc"] for r in rows if r["_hit"]], [o.tlc for o in orphans])
+    assert len(session.calls) == 3
+
+
+@pytest.mark.asyncio
+async def test_page_filter_and_limit_sit_inside_the_match():
+    session = _PagingSession(lambda q: [])
+    await find_broken_chains(_client(session), tenant_id=TENANT)
+    await find_gaps(_client(session), tenant_id=TENANT)
+    for query, _ in session.calls:
+        assert "CALL {" not in query
+        # The cursor filter and LIMIT come before the final RETURN.
+        assert query.index("LIMIT $page_size") < query.rindex("RETURN")
+        assert query.index("$after_key") < query.rindex("RETURN")

From eaa9b8a2d54f5c7ac8447ed473c82291ed668825 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Sun, 25 Oct 2026 23:43:34 -0700
Subject: [PATCH 5/6] fix(r1): coalesce null _hit to false, strict _hit check,
 page filter and ORDER BY/LIMIT in separate WITHs

---
 services/graph/app/fsma/chain_integrity.py    | 12 ++--
 services/graph/app/fsma/data_quality.py       |  4 +-
 .../graph/tests/test_graph_row_paging_r1.py   | 59 +++++++++++++++++++
 3 files changed, 69 insertions(+), 6 deletions(-)

diff --git a/services/graph/app/fsma/chain_integrity.py b/services/graph/app/fsma/chain_integrity.py
index fb63d3b3e..3c0de5986 100644
--- a/services/graph/app/fsma/chain_integrity.py
+++ b/services/graph/app/fsma/chain_integrity.py
@@ -49,9 +49,12 @@ def _after_condition(width: int) -> str:
 
 
 def _hit(record: Any) -> bool:
-    """True unless the row carries a false ``_hit`` flag (anchor kept only for paging)."""
+    """True unless the row carries a ``_hit`` flag that is not true.
+
+    A null flag (Cypher ``null``) is not a hit; queries coalesce it to false.
+    """
     try:
-        return record["_hit"] is not False
+        return record["_hit"] is True
     except (KeyError, LookupError):
         return True
 
@@ -82,6 +85,7 @@ async def _run_paged(
     page_clause = (
         f"WITH {anchors}, [{key_expr}] AS _page_key\n"
         f"WHERE $after_key IS NULL OR {_after_condition(len(key))}\n"
+        f"WITH {anchors}, _page_key\n"
         "ORDER BY _page_key[0]" + "".join(f", _page_key[{i}]" for i in range(1, len(key))) + "\nLIMIT $page_size"
     )
     paged_query = query.replace(_PAGE_MARKER, page_clause, 1)
@@ -426,8 +430,8 @@ async def find_broken_chains(
              }) AS origins
         RETURN
             _page_key,
-            (shipping.merkle_prev_hash IS NULL
-             OR NOT any(o IN origins WHERE o.merkle_hash = shipping.merkle_prev_hash)) AS _hit,
+            coalesce(shipping.merkle_prev_hash IS NULL
+             OR NOT any(o IN origins WHERE o.merkle_hash = shipping.merkle_prev_hash), false) AS _hit,
             shipping.event_id          AS event_id,
             shipping.type              AS type,
             shipping.event_date        AS event_date,
diff --git a/services/graph/app/fsma/data_quality.py b/services/graph/app/fsma/data_quality.py
index e28f6a460..4a00e1c48 100644
--- a/services/graph/app/fsma/data_quality.py
+++ b/services/graph/app/fsma/data_quality.py
@@ -141,9 +141,9 @@ async def find_orphaned_lots(
     // Stagnant-days filter, as a flag so every paged lot yields one row
     RETURN
         _page_key,
-        (most_recent_event IS NULL
+        coalesce(most_recent_event IS NULL
          OR most_recent_event.event_date IS NULL
-         OR most_recent_event.event_date <= $cutoff_date) AS _hit,
+         OR most_recent_event.event_date <= $cutoff_date, false) AS _hit,
         l.tlc as tlc,
         l.product_description as product_description,
         l.quantity as quantity,
diff --git a/services/graph/tests/test_graph_row_paging_r1.py b/services/graph/tests/test_graph_row_paging_r1.py
index c0280d6a1..27be5db66 100644
--- a/services/graph/tests/test_graph_row_paging_r1.py
+++ b/services/graph/tests/test_graph_row_paging_r1.py
@@ -324,3 +324,62 @@ async def test_page_filter_and_limit_sit_inside_the_match():
         # The cursor filter and LIMIT come before the final RETURN.
         assert query.index("LIMIT $page_size") < query.rindex("RETURN")
         assert query.index("$after_key") < query.rindex("RETURN")
+
+
+def _hit_expr(query: str) -> str:
+    """The RETURN expression aliased ``_hit``, whitespace collapsed."""
+    ret = query[query.rindex("RETURN") :]
+    return " ".join(ret[: ret.index("AS _hit")].split())
+
+
+@pytest.mark.asyncio
+async def test_hit_flags_coalesce_cypher_null_to_false():
+    session = _PagingSession(lambda q: [])
+    await find_broken_chains(_client(session), tenant_id=TENANT)
+    await find_orphaned_lots(_client(session), tenant_id=TENANT)
+    hits = [_hit_expr(q) for q, _ in session.calls if "AS _hit" in q]
+    assert len(hits) == 3  # temporal paradox, crypto gap, orphan
+    assert all(h.split("_page_key,", 1)[1].strip().startswith("coalesce(") for h in hits)
+    assert all(h.endswith(", false)") for h in hits)
+
+
+@pytest.mark.asyncio
+async def test_null_merkle_hash_hit_is_not_a_crypto_gap():
+    # Null merkle_hash on shipping and origin: the comparison is null in
+    # Cypher, so the row must not be reported.
+    rows = _shipping_rows(5)
+    for r in rows:
+        r["_hit"] = None
+
+    def rows_for(q):
+        return rows if "merkle_prev_hash" in q else []
+
+    session = _PagingSession(rows_for)
+    violations = await find_broken_chains(_client(session), tenant_id=TENANT)
+    assert violations == []
+
+
+@pytest.mark.asyncio
+async def test_uncomparable_dates_hit_is_not_an_orphan():
+    # last_event_date empty vs a cutoff date: the comparison is null.
+    rows = _orphan_rows(5)
+    for r in rows:
+        r["last_event_date"] = ""
+        r["_hit"] = None
+    session = _PagingSession(lambda q: rows)
+    orphans = await find_orphaned_lots(_client(session), tenant_id=TENANT)
+    assert orphans == []
+
+
+@pytest.mark.asyncio
+async def test_page_clause_orders_and_limits_in_its_own_with():
+    # openCypher: WITH <items> [ORDER BY] [SKIP] [LIMIT] [WHERE]. A WHERE
+    # before ORDER BY is not valid, so filter and order use separate WITHs.
+    session = _PagingSession(lambda q: [])
+    await find_gaps(_client(session), tenant_id=TENANT)
+    for query, _ in session.calls:
+        lines = [ln.strip() for ln in query.splitlines()]
+        w = next(i for i, ln in enumerate(lines) if ln.startswith("WHERE $after_key IS NULL"))
+        assert lines[w + 1].startswith("WITH ")
+        assert lines[w + 2].startswith("ORDER BY _page_key[0]")
+        assert lines[w + 3] == "LIMIT $page_size"

From 42b2768c3206ca80cbf76d24bbda225bea775c18 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Sun, 25 Oct 2026 23:45:24 -0700
Subject: [PATCH 6/6] test(r1): orphan mock row carries _hit true like the real
 query

---
 services/graph/tests/test_data_quality.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/services/graph/tests/test_data_quality.py b/services/graph/tests/test_data_quality.py
index cbfd8fc60..0a4cc69cf 100644
--- a/services/graph/tests/test_data_quality.py
+++ b/services/graph/tests/test_data_quality.py
@@ -302,6 +302,7 @@ async def test_orphan_detection_with_results(self):
             "created_at": "2025-01-01",
             "last_event_type": "RECEIVING",
             "last_event_date": "2025-01-05",
+            "_hit": True,  # the query flags reported lots; null is not a hit
         }.get(key)
 
         # Mock async iterator that yields one record