fix(graph): page FSMA gap/chain/orphan queries instead of silent 2000-row cap (R1)

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, Callable, Dict, List, Optional, Sequence
 
 import structlog
 from shared.fsma_rules import TimeArrowRule, TraceEvent as SharedTraceEvent
@@ -18,6 +18,99 @@
 logger = structlog.get_logger("fsma-utils")
 
 
+# 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
+
+
+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)
+
+
+_PAGE_MARKER = "// @page"
+
+
+def _hit(record: Any) -> bool:
+    """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 True
+    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`` 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)
+    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)
+    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(r for r in page if _hit(r))
+        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_ANCHORS = "l, shipping"
+_CHAIN_KEY = ("shipping.event_id", "l.tlc")
+_CHAIN_ORDER = _sort_key("event_date", "event_id", "lot_tlc")
+
+
 async def find_gaps(
     client: Neo4jClient,
     tenant_id: str,
@@ -37,9 +130,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()
 
@@ -57,21 +148,26 @@ 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,
         e.document_id as document_id,
         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
     """
 
     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:
-        result = await session.run(query, tenant_id=tenant_id)
-        async for record in result:
+        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"])
             gaps.append(
                 {
                     "event_id": record["event_id"],
@@ -140,7 +236,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 keyset-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
@@ -158,15 +255,15 @@ 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,
         shipping.document_id  AS document_id,
         l.tlc                 AS lot_tlc,
         l.product_description AS product_description
-    ORDER BY shipping.event_date
-    LIMIT 2000
     """
 
     # ------------------------------------------------------------------ #
@@ -183,21 +280,23 @@ 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
     // Sort origins per lot before aggregating: collect() preserves the
     // 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,
@@ -206,17 +305,21 @@ 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
     """
 
     violations: List[Dict[str, Any]] = []
     seen_event_ids: set = set()
 
     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,
+            anchors=_CHAIN_ANCHORS,
+            key=_CHAIN_KEY,
+            order=_CHAIN_ORDER,
+            tenant_id=tenant_id,
+        ):
             eid = record["event_id"]
             if eid in seen_event_ids:
                 continue
@@ -236,8 +339,14 @@ 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,
+            anchors=_CHAIN_ANCHORS,
+            key=_CHAIN_KEY,
+            order=_CHAIN_ORDER,
+            tenant_id=tenant_id,
+        ):
             eid = record["event_id"]
             if eid in seen_event_ids:
                 continue
@@ -308,20 +417,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,
+            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,
@@ -331,12 +441,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
-        LIMIT 2000
         """
 
-        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, anchors=_CHAIN_ANCHORS, key=_CHAIN_KEY, order=_CHAIN_ORDER, tenant_id=tenant_id
+        ):
             eid = record["event_id"]
             if eid in seen_event_ids:
                 continue
@@ -392,9 +501,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)

services/graph/app/fsma/data_quality.py

@@ -14,6 +14,7 @@
 import structlog
 
 from ..neo4j_utils import Neo4jClient
+from .chain_integrity import _run_paged, _sort_key
 from .types import DataQualityReport, KDECompletenessMetrics, OrphanLot
 
 logger = structlog.get_logger("fsma-utils")
@@ -128,36 +129,46 @@ 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,
+        coalesce(most_recent_event IS NULL
+         OR most_recent_event.event_date IS NULL
+         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,
         l.unit_of_measure as unit_of_measure,
         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
     """
 
     orphans = []
+    # Keep the first occurrence of each lot so no lot is reported twice.
+    seen_tlcs: set = set()
     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,
+            anchors="l",
+            key=("l.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"])
             # 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

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

services/graph/tests/test_fsma_module_qa.py

@@ -232,14 +232,18 @@ 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 "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
     async def test_paradox_query_sorts_origins_before_collect(self):