diff --git a/frontend/src/app/docs/api/ApiReferenceClient.tsx b/frontend/src/app/docs/api/ApiReferenceClient.tsx index 89fc0b50e..124f593b7 100644 --- a/frontend/src/app/docs/api/ApiReferenceClient.tsx +++ b/frontend/src/app/docs/api/ApiReferenceClient.tsx @@ -141,17 +141,19 @@ const API_SERVICES: { name: string; baseUrl: string; icon: React.ComponentType<{ { method: 'GET', path: '/trace/{tlc}', - description: 'Trace the transformation graph for a traceability lot code (upstream sources and downstream products). Requires the fda.read scope. Unlike the Records API, FDA endpoints take tenant_id as an explicit query parameter (it must match your key’s workspace).', + description: 'Trace the transformation graph for a traceability lot code. direction selects upstream sources (backward), downstream products (forward) or both (default). depth_exhausted is true when a lot sat at the depth limit and was not expanded, so the graph may continue past the response. Requires the fda.read scope. Unlike the Records API, FDA endpoints take tenant_id as an explicit query parameter (it must match your key’s workspace).', auth: true, headers: ['X-RegEngine-API-Key'], - params: ['tenant_id={your-workspace-uuid}'], + params: ['tenant_id={your-workspace-uuid}', 'direction=both', 'depth=5'], response: JSON.stringify({ seed_tlc: "", tenant_id: "", traversal_depth: 5, + direction: "both", + depth_exhausted: false, node_count: 1, edge_count: 0, - nodes: [{ tlc: "", is_seed: true, role: "seed", event_count: 1 }], + nodes: [{ tlc: "", is_seed: true, role: "seed", depth: 0, event_count: 1 }], edges: [], total_events: 1 }, null, 2), diff --git a/frontend/src/app/trust/page.tsx b/frontend/src/app/trust/page.tsx index ad7269059..2637f9a13 100644 --- a/frontend/src/app/trust/page.tsx +++ b/frontend/src/app/trust/page.tsx @@ -50,7 +50,7 @@ const STATUS_SUMMARY = [ }, { label: 'Preview', - detail: 'Capability is usable for evaluation in supported workspaces, but is not GA. GS1 EPCIS 2.0 export and ingest stay Preview until separately promoted.', + detail: 'Capability is usable for evaluation in supported workspaces, but has not reached general release. GS1 EPCIS 2.0 export and ingest stay Preview until separately promoted.', }, { label: 'Pilot', diff --git a/scripts/copy-lint/baseline.json b/scripts/copy-lint/baseline.json index a9272faea..6d091f71f 100644 --- a/scripts/copy-lint/baseline.json +++ b/scripts/copy-lint/baseline.json @@ -2,22 +2,12 @@ "frontend/src/app/auth/verify/page.tsx": { "empty-render": 2 }, - "frontend/src/app/blog/fda-food-traceability-requirements/page.tsx": { - "epcis-unlabeled": 1 - }, "frontend/src/app/blog/food-traceability-software/page.tsx": { - "epcis-unlabeled": 1, "sample-demo": 2 }, "frontend/src/app/blog/fsma-204-compliance-checklist-2026/page.tsx": { "invented-entity": 2 }, - "frontend/src/app/blog/fsma-204-compliance-guide/page.tsx": { - "epcis-unlabeled": 2 - }, - "frontend/src/app/blog/fsma-204-cte-kde-guide/page.tsx": { - "epcis-unlabeled": 5 - }, "frontend/src/app/contact/layout.tsx": { "sample-demo": 2 }, @@ -38,7 +28,6 @@ "empty-render": 1 }, "frontend/src/app/design-system/page.tsx": { - "invented-entity": 1, "sample-demo": 1, "sandbox": 1 }, @@ -58,9 +47,6 @@ "dash-placeholder": 3, "empty-render": 2 }, - "frontend/src/app/fsma-204/[cte]/page.tsx": { - "invented-entity": 1 - }, "frontend/src/app/identity/page.tsx": { "dash-placeholder": 2, "empty-render": 1 @@ -78,9 +64,6 @@ "frontend/src/app/legal/vulnerability-disclosure/page.tsx": { "sample-demo": 1 }, - "frontend/src/app/product/page.tsx": { - "sandbox": 2 - }, "frontend/src/app/records/page.tsx": { "dash-placeholder": 3, "empty-render": 1 @@ -100,9 +83,6 @@ "frontend/src/app/tools/tlc-validator/components/TLCValidatorClient.tsx": { "sample-demo": 1 }, - "frontend/src/app/why-regengine/page.tsx": { - "epcis-unlabeled": 1 - }, "frontend/src/components/dashboard/compliance-status-widget.tsx": { "empty-render": 1 }, @@ -133,9 +113,6 @@ "frontend/src/lib/dev-compliance-snapshots.ts": { "dash-placeholder": 2 }, - "frontend/src/lib/epcis-upload.ts": { - "epcis-unlabeled": 4 - }, "frontend/src/lib/mock-tenants.ts": { "sample-demo": 1 }, @@ -144,8 +121,5 @@ }, "frontend/src/lib/trust-subprocessors.ts": { "sample-demo": 1 - }, - "frontend/src/lib/utils.ts": { - "dash-placeholder": 1 } } diff --git a/services/ingestion/app/canonical_router.py b/services/ingestion/app/canonical_router.py index de549133a..e0cbcfe82 100644 --- a/services/ingestion/app/canonical_router.py +++ b/services/ingestion/app/canonical_router.py @@ -473,6 +473,49 @@ class TraceResponse(BaseModel): # exports should either fail-closed or re-query with a higher cap # when this flag is set. Defaults to False for existing clients. truncated: bool = False + # ``depth_exhausted=True`` means at least one lot was reached at + # ``max_depth`` and was not expanded: the graph may continue beyond + # the returned links. Reported separately from ``truncated`` so a + # client can tell the result cap from the depth limit. Conservative: + # the unexpanded lot may have no further links. Defaults to False + # for existing clients. + depth_exhausted: bool = False + # The depth limit the walk ran with, so a client reading + # ``depth_exhausted`` knows what to raise on a re-query. + max_depth: int = 5 + + +def _trace_response(store: Any, tid: str, tlc: str, direction: str, max_depth: int) -> TraceResponse: + """Run one directional walk and shape it as a ``TraceResponse``. + + Forward links carry ``output_quantity`` / ``output_unit``; backward + links carry ``input_quantity`` / ``input_unit``. Both map onto the + response's ``quantity`` / ``unit``. + """ + walk = store.trace_walk(tid, tlc, direction, max_depth=max_depth) + qty_key, unit_key = ("output_quantity", "output_unit") if direction == "forward" else ("input_quantity", "input_unit") + links = [ + TraceLink( + input_tlc=r["input_tlc"], + output_tlc=r["output_tlc"], + transformation_event_id=r["transformation_event_id"], + process_type=r.get("process_type"), + depth=r["depth"], + confidence_score=r.get("confidence_score", 1.0), + quantity=r.get(qty_key), + unit=r.get(unit_key), + ) + for r in walk.links + ] + return TraceResponse( + tlc=tlc, + direction=direction, + links=links, + total_hops=max(r["depth"] for r in walk.links) if walk.links else 0, + truncated=walk.truncated, + depth_exhausted=walk.depth_exhausted, + max_depth=walk.max_depth, + ) @router.get( @@ -493,31 +536,11 @@ def trace_forward( set_tenant_guc(db_session, tid) store = CanonicalEventStore(db_session, dual_write=False) - # #1282: trace_forward now returns (links, truncated). ``truncated`` - # means the BFS hit max_results and stopped early; surface it so - # API clients can distinguish a complete trace from a capped one. - raw_links, truncated = store.trace_forward(tid, tlc, max_depth=max_depth) - - links = [ - TraceLink( - input_tlc=r["input_tlc"], - output_tlc=r["output_tlc"], - transformation_event_id=r["transformation_event_id"], - process_type=r.get("process_type"), - depth=r["depth"], - confidence_score=r.get("confidence_score", 1.0), - quantity=r.get("output_quantity"), - unit=r.get("output_unit"), - ) - for r in raw_links - ] - return TraceResponse( - tlc=tlc, - direction="forward", - links=links, - total_hops=max(r["depth"] for r in raw_links) if raw_links else 0, - truncated=truncated, - ) + # #1282: ``truncated`` means the BFS hit max_results and stopped early; + # ``depth_exhausted`` means ``max_depth`` cut the walk short. Both are + # surfaced so API clients can distinguish a complete trace from a + # capped one. + return _trace_response(store, tid, tlc, "forward", max_depth) @router.get( @@ -538,27 +561,5 @@ def trace_backward( set_tenant_guc(db_session, tid) store = CanonicalEventStore(db_session, dual_write=False) - # #1282: trace_backward now returns (links, truncated). See - # trace_forward handler for the rationale. - raw_links, truncated = store.trace_backward(tid, tlc, max_depth=max_depth) - - links = [ - TraceLink( - input_tlc=r["input_tlc"], - output_tlc=r["output_tlc"], - transformation_event_id=r["transformation_event_id"], - process_type=r.get("process_type"), - depth=r["depth"], - confidence_score=r.get("confidence_score", 1.0), - quantity=r.get("input_quantity"), - unit=r.get("input_unit"), - ) - for r in raw_links - ] - return TraceResponse( - tlc=tlc, - direction="backward", - links=links, - total_hops=max(r["depth"] for r in raw_links) if raw_links else 0, - truncated=truncated, - ) + # See trace_forward for the ``truncated`` / ``depth_exhausted`` contract. + return _trace_response(store, tid, tlc, "backward", max_depth) diff --git a/services/ingestion/app/fda_export/queries.py b/services/ingestion/app/fda_export/queries.py index 77e3df560..83d9584d5 100644 --- a/services/ingestion/app/fda_export/queries.py +++ b/services/ingestion/app/fda_export/queries.py @@ -850,12 +850,20 @@ def fetch_trace_graph_data( tenant_id: str, tlc: str, depth: int, + direction: str = "both", ) -> dict: """Fetch the transformation trace graph for a TLC. - Returns a dict with nodes, edges, and summary stats. + ``direction`` is ``forward`` (downstream outputs), ``backward`` + (upstream inputs) or ``both`` (the union of the two directed walks). + Returns a dict with nodes, edges, summary stats, the ``direction`` + walked and ``depth_exhausted``: True when a lot was reached at + ``depth`` and not expanded, so the graph may continue past what is + returned. """ - linked_tlcs = persistence._expand_tlcs_via_transformation_links(tenant_id=tenant_id, seed_tlc=tlc, depth=depth) + expansion = persistence._expand_tlcs_detailed(tenant_id=tenant_id, seed_tlc=tlc, depth=depth, direction=direction) + linked_tlcs = expansion.tlcs + reached_via = expansion.reached_via # Fetch event counts per TLC in one query if linked_tlcs: @@ -887,13 +895,20 @@ def fetch_trace_graph_data( else: tlc_stats = {} - # Also fetch direct link edges for the graph visualization + # Edges for the graph visualization: only links whose BOTH ends are + # lots the directed walk returned, so ``nodes`` and ``edges`` describe + # the same graph. A link touching one returned lot and one excluded + # lot (a sibling branch, an opposite-direction neighbour, or a lot + # past the depth limit) used to leak in through an OR predicate and + # name lots absent from ``nodes``. + node_set = set(linked_tlcs) link_rows = db_session.execute( text(""" SELECT input_tlc, output_tlc, process_type, confidence_score FROM fsma.transformation_links WHERE tenant_id = :tid - AND (input_tlc = ANY(:tlcs) OR output_tlc = ANY(:tlcs)) + AND input_tlc = ANY(:tlcs) + AND output_tlc = ANY(:tlcs) """), {"tid": tenant_id, "tlcs": linked_tlcs}, ).fetchall() @@ -905,19 +920,32 @@ def fetch_trace_graph_data( "confidence_score": float(row[3]) if row[3] is not None else None, } for row in link_rows + # Defence in depth for persistence stubs and older catalogs: the + # SQL already restricts both ends. + if row[0] in node_set and row[1] in node_set ] + def _role(t: str) -> str: + # The role comes from the walk that reached the lot, not from a + # direct edge to the seed: a lot two hops downstream used to be + # labelled "upstream" because only seed-adjacent edges were checked. + if t == tlc: + return "seed" + via = reached_via.get(t) + if via is None: + # Persistence without reach data (older stubs): fall back to + # the seed-adjacent heuristic. + return "downstream" if any(e["input_tlc"] == tlc and e["output_tlc"] == t for e in edges) else "upstream" + if via.get("downstream") and via.get("upstream"): + return "both" + return "downstream" if via.get("downstream") else "upstream" + nodes = [ { "tlc": t, "is_seed": t == tlc, - "role": ( - "seed" - if t == tlc - else "downstream" - if any(e["input_tlc"] == tlc and e["output_tlc"] == t for e in edges) - else "upstream" - ), + "role": _role(t), + "depth": reached_via.get(t, {}).get("depth"), **tlc_stats.get(t, {"event_count": 0, "first_event": None, "last_event": None}), } for t in linked_tlcs @@ -927,6 +955,10 @@ def fetch_trace_graph_data( "seed_tlc": tlc, "tenant_id": tenant_id, "traversal_depth": depth, + "direction": expansion.direction, + # True when a lot sat at the depth limit unexpanded: the trace may + # continue past these nodes. Never silently complete (#3701 S2). + "depth_exhausted": expansion.depth_exhausted, "node_count": len(nodes), "edge_count": len(edges), "nodes": nodes, diff --git a/services/ingestion/app/fda_export/router.py b/services/ingestion/app/fda_export/router.py index 2e61fdea5..34947985a 100644 --- a/services/ingestion/app/fda_export/router.py +++ b/services/ingestion/app/fda_export/router.py @@ -1283,9 +1283,13 @@ async def get_merkle_proof( summary="Trace transformation graph for a TLC", description=( "Returns all TLCs reachable from the given lot code via " - "fsma.transformation_links — both upstream inputs and downstream outputs. " - "Also returns the full CTE event count per linked lot so the UI can " - "show the trace tree without fetching the full FDA export." + "fsma.transformation_links. ``direction`` selects upstream inputs " + "(backward), downstream outputs (forward) or the union of both walks " + "(both, the default). ``depth_exhausted`` in the response is true when " + "a lot was reached at the depth limit and not expanded, so the graph " + "may continue past what is returned. Also returns the full CTE event " + "count per linked lot so the UI can show the trace tree without " + "fetching the full FDA export." ), tags=["Traceability"], ) @@ -1296,6 +1300,11 @@ def trace_transformation_graph( tlc: str, tenant_id: str = Query(..., description="Tenant identifier"), depth: int = Query(default=5, ge=1, le=10, description="Max traversal depth"), + direction: str = Query( + default="both", + pattern="^(forward|backward|both)$", + description="forward = downstream outputs, backward = upstream inputs, both = union of the two walks", + ), principal: IngestionPrincipal = Depends(require_permission("fda.read")), _subscription=Depends(require_active_subscription), ) -> dict: @@ -1318,6 +1327,7 @@ def trace_transformation_graph( tenant_id=tenant_id, tlc=tlc, depth=depth, + direction=direction, ) except HTTPException: diff --git a/services/ingestion/tests/test_canonical_router_1342.py b/services/ingestion/tests/test_canonical_router_1342.py index 51f7135fb..b6ec5de1b 100644 --- a/services/ingestion/tests/test_canonical_router_1342.py +++ b/services/ingestion/tests/test_canonical_router_1342.py @@ -21,6 +21,7 @@ import sys from datetime import datetime, timezone from pathlib import Path +from types import SimpleNamespace from typing import Any, Dict, List, Optional from unittest.mock import MagicMock @@ -556,29 +557,49 @@ def test_forward_walk_null_created_at(self, principal: IngestionPrincipal) -> No class _TraceStoreStub: """Callable stub for ``CanonicalEventStore(...)`` used by trace - endpoints. ``trace_forward`` and ``trace_backward`` return - ``(list_of_links, truncated_bool)`` in the current store contract.""" + endpoints. The router calls ``trace_walk`` and reads ``links``, + ``truncated``, ``depth_exhausted`` and ``max_depth`` off the result; + ``trace_forward`` / ``trace_backward`` keep the older + ``(list_of_links, truncated_bool)`` contract for other callers.""" def __init__( self, forward_result: Optional[tuple] = None, backward_result: Optional[tuple] = None, + depth_exhausted: bool = False, ) -> None: self._forward = forward_result or ([], False) self._backward = backward_result or ([], False) + self._depth_exhausted = depth_exhausted self.forward_calls: List[tuple] = [] self.backward_calls: List[tuple] = [] def __call__(self, session: Any, dual_write: bool = False) -> "_TraceStoreStub": return self + def trace_walk( + self, tenant_id: str, tlc: str, direction: str, max_depth: int = 5, max_results: int = 10000 + ) -> SimpleNamespace: + links, truncated = self._forward if direction == "forward" else self._backward + (self.forward_calls if direction == "forward" else self.backward_calls).append( + (tenant_id, tlc, max_depth, max_results) + ) + return SimpleNamespace( + links=links, + truncated=truncated, + depth_exhausted=self._depth_exhausted, + direction=direction, + max_depth=max_depth, + max_results=max_results, + ) + def trace_forward(self, tenant_id: str, tlc: str, max_depth: int = 5, max_results: int = 10000) -> tuple: - self.forward_calls.append((tenant_id, tlc, max_depth, max_results)) - return self._forward + walk = self.trace_walk(tenant_id, tlc, "forward", max_depth, max_results) + return walk.links, walk.truncated def trace_backward(self, tenant_id: str, tlc: str, max_depth: int = 5, max_results: int = 10000) -> tuple: - self.backward_calls.append((tenant_id, tlc, max_depth, max_results)) - return self._backward + walk = self.trace_walk(tenant_id, tlc, "backward", max_depth, max_results) + return walk.links, walk.truncated def _patch_trace_store(monkeypatch: pytest.MonkeyPatch, stub: _TraceStoreStub) -> None: @@ -611,6 +632,8 @@ def test_empty_links_total_hops_zero(self, principal: IngestionPrincipal, monkey assert body["links"] == [] assert body["total_hops"] == 0 assert body["truncated"] is False + assert body["depth_exhausted"] is False + assert body["max_depth"] == 5 # SET LOCAL pragma was issued. sql, params = session.calls[0] assert "app.tenant_id" in sql @@ -618,6 +641,32 @@ def test_empty_links_total_hops_zero(self, principal: IngestionPrincipal, monkey # Store was called with the resolved tenant. assert stub.forward_calls == [(TENANT_ID, "TLC-EMPTY", 5, 10000)] + def test_depth_exhausted_is_surfaced_separately_from_truncated( + self, principal: IngestionPrincipal, monkeypatch: pytest.MonkeyPatch + ) -> None: + """S2 of the 2026-10-28 gap analysis: reaching ``max_depth`` must + not read as a complete trace. The flag is independent of the + result-cap flag and the response echoes the depth it ran with.""" + raw_links = [ + {"input_tlc": "TLC-A", "output_tlc": "TLC-B", "transformation_event_id": "evt-1", "depth": 1}, + {"input_tlc": "TLC-B", "output_tlc": "TLC-C", "transformation_event_id": "evt-2", "depth": 2}, + ] + stub = _TraceStoreStub(forward_result=(raw_links, False), depth_exhausted=True) + _patch_trace_store(monkeypatch, stub) + session = _FakeSession([_Result()]) + c = _client(principal, session) + resp = c.get( + self.ENDPOINT_TEMPLATE.format(tlc="TLC-A"), + params={"tenant_id": TENANT_ID, "max_depth": 2}, + ) + assert resp.status_code == 200 + body = resp.json() + assert body["truncated"] is False + assert body["depth_exhausted"] is True + assert body["max_depth"] == 2 + assert body["total_hops"] == 2 + assert stub.forward_calls == [(TENANT_ID, "TLC-A", 2, 10000)] + def test_populated_links_compute_max_depth( self, principal: IngestionPrincipal, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -695,6 +744,7 @@ def test_empty_links_total_hops_zero(self, principal: IngestionPrincipal, monkey assert body["links"] == [] assert body["total_hops"] == 0 assert body["truncated"] is False + assert body["depth_exhausted"] is False # Pragma issued. sql, _ = session.calls[0] assert "app.tenant_id" in sql diff --git a/services/ingestion/tests/test_fda_export_queries_1342.py b/services/ingestion/tests/test_fda_export_queries_1342.py index fb1775715..91917b629 100644 --- a/services/ingestion/tests/test_fda_export_queries_1342.py +++ b/services/ingestion/tests/test_fda_export_queries_1342.py @@ -119,14 +119,102 @@ def test_fetch_v2_events_empty_result() -> None: def test_trace_with_no_linked_lots_has_no_nodes_or_edges() -> None: + from shared.cte_persistence.queries import TlcExpansion + class EmptyPersistence: - def _expand_tlcs_via_transformation_links(self, **_kwargs): - return [] + def _expand_tlcs_detailed(self, **kwargs): + return TlcExpansion(tlcs=[], depth_exhausted=False, direction=kwargs["direction"], max_depth=kwargs["depth"]) session = _EmptySession() graph = q.fetch_trace_graph_data(session, EmptyPersistence(), "", "", 0) assert graph["nodes"] == [] assert graph["edges"] == [] assert graph["total_events"] == 0 + assert graph["direction"] == "both" + assert graph["depth_exhausted"] is False assert len(session.calls) == 1 assert "fsma.transformation_links" in session.calls[0][0] + + +def test_trace_roles_follow_walk_direction_and_report_depth_limit() -> None: + """S2 (gap analysis 2026-10-28): node roles come from the walk that + reached the lot, so a lot two hops downstream is ``downstream`` (it used + to read ``upstream`` because only seed-adjacent edges were checked), and + reaching the depth limit is surfaced on the response.""" + from shared.cte_persistence.queries import TlcExpansion + + class Persistence: + def __init__(self): + self.calls = [] + + def _expand_tlcs_detailed(self, **kwargs): + self.calls.append(kwargs) + return TlcExpansion( + tlcs=["SEED", "OUT1", "OUT2", "IN1"], + depth_exhausted=True, + direction=kwargs["direction"], + max_depth=kwargs["depth"], + reached_via={ + "SEED": {"depth": 0, "downstream": False, "upstream": False}, + "OUT1": {"depth": 1, "downstream": True, "upstream": False}, + "OUT2": {"depth": 2, "downstream": True, "upstream": False}, + "IN1": {"depth": 1, "downstream": False, "upstream": True}, + }, + ) + + persistence = Persistence() + session = _EmptySession() + graph = q.fetch_trace_graph_data(session, persistence, "tenant-1", "SEED", 2, direction="forward") + assert persistence.calls == [{"tenant_id": "tenant-1", "seed_tlc": "SEED", "depth": 2, "direction": "forward"}] + roles = {n["tlc"]: n["role"] for n in graph["nodes"]} + assert roles == {"SEED": "seed", "OUT1": "downstream", "OUT2": "downstream", "IN1": "upstream"} + assert {n["tlc"]: n["depth"] for n in graph["nodes"]}["OUT2"] == 2 + assert graph["direction"] == "forward" + assert graph["depth_exhausted"] is True + assert graph["node_count"] == 4 + + +def test_trace_edges_are_restricted_to_returned_nodes() -> None: + """Codex and Cursor reviews on #3713: the edge query used to select every + link touching a returned lot, so a sibling branch (A -> S) or an + opposite-direction neighbour leaked into ``edges`` and named lots absent + from ``nodes``. Both ends must now be returned lots, in SQL and again on + the rows (persistence stubs do not run the SQL).""" + from shared.cte_persistence.queries import TlcExpansion + + class Persistence: + def _expand_tlcs_detailed(self, **kwargs): + return TlcExpansion( + tlcs=["A", "B", "C"], + depth_exhausted=False, + direction=kwargs["direction"], + max_depth=kwargs["depth"], + reached_via={ + "A": {"depth": 1, "downstream": False, "upstream": True}, + "B": {"depth": 0, "downstream": False, "upstream": False}, + "C": {"depth": 1, "downstream": True, "upstream": False}, + }, + ) + + class _EdgeResult: + def fetchall(self): + # A -> S is the sibling branch the directed walk excluded. + return [("A", "B", "ship", 1.0), ("B", "C", "transform", 0.5), ("A", "S", "ship", None)] + + class EdgeSession(_EmptySession): + def execute(self, statement, params=None): + self.calls.append((str(statement), params)) + if "fsma.transformation_links" in str(statement): + return _EdgeResult() + return _EmptyResult() + + session = EdgeSession() + graph = q.fetch_trace_graph_data(session, Persistence(), "tenant-1", "B", 5) + edge_sql = next(sql for sql, _ in session.calls if "fsma.transformation_links" in sql) + assert "input_tlc = ANY(:tlcs)" in edge_sql + assert "AND output_tlc = ANY(:tlcs)" in edge_sql + assert " OR " not in edge_sql + pairs = {(e["input_tlc"], e["output_tlc"]) for e in graph["edges"]} + assert pairs == {("A", "B"), ("B", "C")} + assert graph["edge_count"] == 2 + assert {n["tlc"] for n in graph["nodes"]} == {"A", "B", "C"} diff --git a/services/ingestion/tests/test_fda_trace_route_direction.py b/services/ingestion/tests/test_fda_trace_route_direction.py new file mode 100644 index 000000000..366684d41 --- /dev/null +++ b/services/ingestion/tests/test_fda_trace_route_direction.py @@ -0,0 +1,103 @@ +"""Route-level tests for ``GET /api/v1/fda/trace/{tlc}``. + +Gap analysis 2026-10-28, S2: the trace graph route gains a ``direction`` +query parameter (forward, backward, both) and its response carries +``depth_exhausted`` so a depth-capped trace never reads as complete. No +route-level test existed for this endpoint before; the handler is driven +here with the persistence and session factory stubbed out. +""" + +from __future__ import annotations + +import importlib +import sys +from pathlib import Path + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from app.authz import IngestionPrincipal # noqa: E402 + +# ``app.fda_export`` re-exports the APIRouter as ``router``; import the module +# itself so ``fetch_trace_graph_data`` can be monkeypatched where the handler +# looks it up. +fda_router_module = importlib.import_module("app.fda_export.router") + +from tests.authz_overrides import ( # noqa: E402 + install_active_subscription_override, + install_ingestion_principal_override, +) + +TENANT_ID = "11111111-1111-1111-1111-111111111111" + + +class _FakeSession: + def close(self) -> None: # pragma: no cover - trivial + pass + + +def _client(monkeypatch: pytest.MonkeyPatch, captured: dict) -> TestClient: + app = FastAPI() + app.include_router(fda_router_module.router) + principal = IngestionPrincipal(key_id="test-key", tenant_id=TENANT_ID, scopes=["fda.read"], auth_mode="test") + install_ingestion_principal_override(app, principal) + install_active_subscription_override(app) + + import shared.database as shared_database + + monkeypatch.setattr(shared_database, "SessionLocal", lambda: _FakeSession()) + + def _fake_fetch(**kwargs): + captured.update(kwargs) + return { + "seed_tlc": kwargs["tlc"], + "tenant_id": kwargs["tenant_id"], + "traversal_depth": kwargs["depth"], + "direction": kwargs["direction"], + "depth_exhausted": True, + "node_count": 1, + "edge_count": 0, + "nodes": [{"tlc": kwargs["tlc"], "is_seed": True, "role": "seed", "depth": 0, "event_count": 0}], + "edges": [], + "total_events": 0, + } + + monkeypatch.setattr(fda_router_module, "fetch_trace_graph_data", _fake_fetch) + return TestClient(app) + + +def test_direction_defaults_to_both_and_depth_exhausted_is_surfaced(monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + client = _client(monkeypatch, captured) + resp = client.get("/api/v1/fda/trace/LOT-1", params={"tenant_id": TENANT_ID}) + assert resp.status_code == 200, resp.text + body = resp.json() + assert captured["direction"] == "both" + assert captured["depth"] == 5 + assert captured["tenant_id"] == TENANT_ID + assert body["direction"] == "both" + assert body["depth_exhausted"] is True + assert body["nodes"][0]["depth"] == 0 + + +@pytest.mark.parametrize("direction", ["forward", "backward", "both"]) +def test_direction_is_passed_through(monkeypatch: pytest.MonkeyPatch, direction: str) -> None: + captured: dict = {} + client = _client(monkeypatch, captured) + resp = client.get("/api/v1/fda/trace/LOT-1", params={"tenant_id": TENANT_ID, "direction": direction, "depth": 3}) + assert resp.status_code == 200, resp.text + assert captured["direction"] == direction + assert captured["depth"] == 3 + assert resp.json()["direction"] == direction + + +def test_invalid_direction_is_rejected_with_422(monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + client = _client(monkeypatch, captured) + resp = client.get("/api/v1/fda/trace/LOT-1", params={"tenant_id": TENANT_ID, "direction": "sideways"}) + assert resp.status_code == 422 + # The handler never ran: no persistence call was made. + assert captured == {} diff --git a/services/shared/canonical_persistence/writer.py b/services/shared/canonical_persistence/writer.py index 7d3ba8bfa..629cb59e7 100644 --- a/services/shared/canonical_persistence/writer.py +++ b/services/shared/canonical_persistence/writer.py @@ -9,6 +9,7 @@ import json import logging +from dataclasses import dataclass from datetime import datetime, timezone from typing import Any, Dict, List, Optional, Tuple from uuid import UUID, uuid4 @@ -27,6 +28,26 @@ from sqlalchemy import bindparam, text from sqlalchemy.orm import Session + +@dataclass(frozen=True) +class TraceWalk: + """Result of one breadth-first walk over ``fsma.transformation_links``. + + ``truncated`` means the walk stopped because ``max_results`` links were + collected (#1282). ``depth_exhausted`` means at least one lot was + reached at ``max_depth`` and therefore never expanded: the graph may + continue beyond the returned links. The two limits are reported + separately so a caller can tell "the cap bit" from "the depth bit"; + both mean the link list is a valid prefix, not a complete trace. + """ + + links: List[Dict[str, Any]] + truncated: bool + depth_exhausted: bool + direction: str + max_depth: int + max_results: int + logger = logging.getLogger("canonical-persistence") @@ -824,20 +845,45 @@ def _create_transformation_links(self, event: TraceabilityEvent) -> None: # exists to prevent OOM under adversarial inputs — see #1282. _TRACE_DEFAULT_MAX_RESULTS = 10_000 - def trace_forward( + _TRACE_FORWARD_SQL = text(""" + SELECT output_tlc, transformation_event_id, process_type, + output_quantity, output_unit, confidence_score + FROM fsma.transformation_links + WHERE tenant_id = :tid AND input_tlc = :tlc + """) + _TRACE_BACKWARD_SQL = text(""" + SELECT input_tlc, transformation_event_id, process_type, + input_quantity, input_unit, confidence_score + FROM fsma.transformation_links + WHERE tenant_id = :tid AND output_tlc = :tlc + """) + + def trace_walk( self, tenant_id: str, tlc: str, + direction: str, max_depth: int = 5, max_results: int = _TRACE_DEFAULT_MAX_RESULTS, - ) -> Tuple[List[Dict[str, Any]], bool]: - """Forward trace: find all output TLCs an input TLC contributed to. + ) -> TraceWalk: + """Breadth-first walk over ``fsma.transformation_links`` in one direction. + + ``direction`` is ``"forward"`` (input lot to the output lots it went + into) or ``"backward"`` (output lot to the input lots it came from). + + The walk stops on two independent limits and reports each one: - Returns ``(links, truncated)``. ``truncated=True`` means the BFS - stopped at ``max_results`` — the link list is still valid data, - just incomplete. Callers MUST check the flag and either surface - truncation to the end user or fail-closed if completeness is - required (e.g. regulator traceback export). + * ``truncated``: ``max_results`` links were collected (#1282). The + remaining BFS frontier is dropped without further queries. + * ``depth_exhausted``: at least one lot was reached at ``max_depth`` + and was therefore never expanded. It is a conservative flag: the + unexpanded lot may have no further links. It never costs an extra + query. Callers that need certainty can re-run with a higher + ``max_depth``. + + Both flags mean the link list is a valid prefix, not a complete + trace. Callers MUST surface them, or fail closed when completeness + is required (regulator traceback export). #1282: without a cap, ``results`` could grow to O(depth × fan-out) with no ceiling, letting a malicious or misconfigured tenant OOM @@ -845,46 +891,68 @@ def trace_forward( cycles but does not bound result size, so we add an explicit ``max_results`` gate distinct from ``max_depth``. """ + if direction == "forward": + stmt = self._TRACE_FORWARD_SQL + elif direction == "backward": + stmt = self._TRACE_BACKWARD_SQL + else: + raise ValueError(f"direction must be 'forward' or 'backward', got {direction!r}") + if max_depth < 1: + raise ValueError(f"max_depth must be >= 1, got {max_depth}") + results: List[Dict[str, Any]] = [] visited: set[str] = set() queue: List[Tuple[str, int]] = [(tlc, 0)] truncated = False + depth_exhausted = False while queue: current_tlc, depth = queue.pop(0) - if depth >= max_depth or current_tlc in visited: + if current_tlc in visited: + continue + if depth >= max_depth: + # A lot we have not expanded sits at the depth limit. We + # do not know whether it has further links, so report the + # limit as reached rather than silently returning a trace + # that reads as complete. + depth_exhausted = True continue visited.add(current_tlc) - rows = self.session.execute( - text(""" - SELECT output_tlc, transformation_event_id, process_type, - output_quantity, output_unit, confidence_score - FROM fsma.transformation_links - WHERE tenant_id = :tid AND input_tlc = :tlc - """), - {"tid": tenant_id, "tlc": current_tlc}, - ).fetchall() + rows = self.session.execute(stmt, {"tid": tenant_id, "tlc": current_tlc}).fetchall() for row in rows: if len(results) >= max_results: truncated = True break - output_tlc = row[0] - results.append( - { - "input_tlc": current_tlc, - "output_tlc": output_tlc, - "transformation_event_id": str(row[1]), - "process_type": row[2], - "output_quantity": float(row[3]) if row[3] else None, - "output_unit": row[4], - "confidence_score": float(row[5]) if row[5] else 1.0, - "depth": depth + 1, - } - ) - if output_tlc not in visited: - queue.append((output_tlc, depth + 1)) + neighbour = row[0] + link: Dict[str, Any] = { + "transformation_event_id": str(row[1]), + "process_type": row[2], + "confidence_score": float(row[5]) if row[5] else 1.0, + "depth": depth + 1, + } + if direction == "forward": + link.update( + { + "input_tlc": current_tlc, + "output_tlc": neighbour, + "output_quantity": float(row[3]) if row[3] else None, + "output_unit": row[4], + } + ) + else: + link.update( + { + "input_tlc": neighbour, + "output_tlc": current_tlc, + "input_quantity": float(row[3]) if row[3] else None, + "input_unit": row[4], + } + ) + results.append(link) + if neighbour not in visited: + queue.append((neighbour, depth + 1)) if truncated: # Drop remaining BFS frontier — we've already accepted we @@ -892,89 +960,59 @@ def trace_forward( # just burn budget without adding surfaced rows. break - if truncated: + if truncated or depth_exhausted: logger.warning( - "trace_forward_truncated", + f"trace_{direction}_incomplete", extra={ "tenant_id": tenant_id, "tlc": tlc, + "direction": direction, + "truncated": truncated, + "depth_exhausted": depth_exhausted, "max_results": max_results, "max_depth": max_depth, "returned": len(results), }, ) - return results, truncated + return TraceWalk( + links=results, + truncated=truncated, + depth_exhausted=depth_exhausted, + direction=direction, + max_depth=max_depth, + max_results=max_results, + ) - def trace_backward( + def trace_forward( self, tenant_id: str, tlc: str, max_depth: int = 5, max_results: int = _TRACE_DEFAULT_MAX_RESULTS, ) -> Tuple[List[Dict[str, Any]], bool]: - """Backward trace: find all input TLCs that contributed to an output TLC. + """Forward trace: find all output TLCs an input TLC contributed to. - Returns ``(links, truncated)``. See ``trace_forward`` for the - cap rationale (#1282). + Returns ``(links, truncated)`` for callers on the #1282 contract. + ``truncated`` covers the result cap only; use :meth:`trace_walk` + to also learn whether ``max_depth`` cut the walk short. """ - results: List[Dict[str, Any]] = [] - visited: set[str] = set() - queue: List[Tuple[str, int]] = [(tlc, 0)] - truncated = False + walk = self.trace_walk(tenant_id, tlc, "forward", max_depth=max_depth, max_results=max_results) + return walk.links, walk.truncated - while queue: - current_tlc, depth = queue.pop(0) - if depth >= max_depth or current_tlc in visited: - continue - visited.add(current_tlc) - - rows = self.session.execute( - text(""" - SELECT input_tlc, transformation_event_id, process_type, - input_quantity, input_unit, confidence_score - FROM fsma.transformation_links - WHERE tenant_id = :tid AND output_tlc = :tlc - """), - {"tid": tenant_id, "tlc": current_tlc}, - ).fetchall() - - for row in rows: - if len(results) >= max_results: - truncated = True - break - input_tlc = row[0] - results.append( - { - "input_tlc": input_tlc, - "output_tlc": current_tlc, - "transformation_event_id": str(row[1]), - "process_type": row[2], - "input_quantity": float(row[3]) if row[3] else None, - "input_unit": row[4], - "confidence_score": float(row[5]) if row[5] else 1.0, - "depth": depth + 1, - } - ) - if input_tlc not in visited: - queue.append((input_tlc, depth + 1)) - - if truncated: - break - - if truncated: - logger.warning( - "trace_backward_truncated", - extra={ - "tenant_id": tenant_id, - "tlc": tlc, - "max_results": max_results, - "max_depth": max_depth, - "returned": len(results), - }, - ) + def trace_backward( + self, + tenant_id: str, + tlc: str, + max_depth: int = 5, + max_results: int = _TRACE_DEFAULT_MAX_RESULTS, + ) -> Tuple[List[Dict[str, Any]], bool]: + """Backward trace: find all input TLCs that contributed to an output TLC. - return results, truncated + Returns ``(links, truncated)``; see :meth:`trace_forward`. + """ + walk = self.trace_walk(tenant_id, tlc, "backward", max_depth=max_depth, max_results=max_results) + return walk.links, walk.truncated # ------------------------------------------------------------------ # Read Path diff --git a/services/shared/cte_persistence/queries.py b/services/shared/cte_persistence/queries.py index 9399b29a4..986d0a958 100644 --- a/services/shared/cte_persistence/queries.py +++ b/services/shared/cte_persistence/queries.py @@ -10,6 +10,7 @@ from __future__ import annotations +from dataclasses import dataclass, field from datetime import date as _date, datetime as _datetime, time as _time, timedelta as _timedelta, timezone as _timezone from typing import Any, Dict, List, Optional, Tuple @@ -20,6 +21,52 @@ ProgrammingError, ) +# Directions a transformation-link walk can take. ``forward`` follows +# input lot -> output lot (what did this lot go into), ``backward`` follows +# output lot -> input lot (what went into this lot), ``both`` is the union +# of a pure forward walk and a pure backward walk from the seed. ``both`` +# is NOT an undirected walk: stepping backward to a parent and then +# forward again would pull in sibling lots that share an input but never +# touched the seed, which 21 CFR 1.1350(c) does not ask for. +TRACE_DIRECTIONS = ("forward", "backward", "both") + +# VALUES fragments seeding the recursive walk's direction column. Keyed by +# the validated direction name, so no caller-supplied text reaches SQL. +_DIRECTION_SEED_VALUES = { + "forward": "(CAST('f' AS text))", + "backward": "(CAST('b' AS text))", + "both": "(CAST('f' AS text)), (CAST('b' AS text))", +} + + +def _validate_direction(direction: str) -> str: + if direction not in TRACE_DIRECTIONS: + raise ValueError(f"direction must be one of {TRACE_DIRECTIONS}, got {direction!r}") + return direction + + +@dataclass(frozen=True) +class TlcExpansion: + """Result of expanding a seed lot through ``fsma.transformation_links``. + + ``tlcs`` always includes the seed. ``depth_exhausted`` is True when at + least one lot was reached at ``max_depth`` by some walk and therefore + not expanded in that direction: the graph may continue past the + returned set. The limit is judged per (lot, direction) before the + walks are collapsed, so a lot the backward walk found early does not + hide a forward frontier. It is conservative (the lot at the limit may + have no further links) and costs no extra query. ``reached_via`` maps + each lot to the shallowest depth it was found at and whether the + forward (downstream) or backward (upstream) walk reached it; the seed + is reached by neither. + """ + + tlcs: List[str] + depth_exhausted: bool + direction: str + max_depth: int + reached_via: Dict[str, Dict[str, Any]] = field(default_factory=dict) + def _end_bound_utc(end_date: str) -> str: """Convert a user-supplied ``YYYY-MM-DD`` end-date to a strict @@ -94,12 +141,26 @@ def _expand_tlcs_via_transformation_links( tenant_id: str, seed_tlc: str, depth: Optional[int] = None, + direction: str = "both", ) -> List[str]: """Return all TLCs reachable from seed_tlc through transformation_links. - Walks both directions (forward: seed → outputs, backward: seed → inputs) - using a breadth-first traversal capped at ``depth`` hops. Returns the - full set of TLCs including the seed itself. + Thin list-returning wrapper over :meth:`_expand_tlcs_detailed` for + callers on the original contract. ``direction`` defaults to ``both``: + the union of a forward walk (seed → outputs) and a backward walk + (seed → inputs), capped at ``depth`` hops, seed included. Whether + the depth cap was reached is only available from the detailed form. + """ + return self._expand_tlcs_detailed(tenant_id, seed_tlc, depth=depth, direction=direction).tlcs + + def _expand_tlcs_detailed( + self, + tenant_id: str, + seed_tlc: str, + depth: Optional[int] = None, + direction: str = "both", + ) -> TlcExpansion: + """Expand ``seed_tlc`` through ``fsma.transformation_links``. Args: tenant_id: RLS tenant identifier. @@ -109,8 +170,15 @@ def _expand_tlcs_via_transformation_links( in the signature, now a class attribute so callers and tests can raise/lower it without patching the signature (#1322). + direction: ``forward``, ``backward`` or ``both`` (see + ``TRACE_DIRECTIONS``). ``both`` is the union of the two + directed walks, not an undirected walk, so sibling lots + that merely share an input with the seed are excluded. - Falls back to ``[seed_tlc]`` ONLY when the transformation_links + Returns a :class:`TlcExpansion` whose ``depth_exhausted`` flag says + whether the walk reached ``depth`` with lots left unexpanded. + + Falls back to the seed alone ONLY when the transformation_links table does not exist (fresh DB bootstrapping). Every other SQL error — including transient connectivity loss mid-export — is re-raised so it surfaces in callers instead of silently @@ -124,6 +192,7 @@ def _expand_tlcs_via_transformation_links( max_depth = depth if depth is not None else self.DEFAULT_TRAVERSAL_DEPTH if max_depth < 0: raise ValueError(f"depth must be >= 0, got {max_depth}") + direction = _validate_direction(direction) try: # Note: use CAST(:tlc AS text) rather than :tlc::text — SQLAlchemy's @@ -135,29 +204,98 @@ def _expand_tlcs_via_transformation_links( # seed UNION forward UNION backward puts the forward branch inside # the non-recursive term and every execution fails with # InvalidRecursion ("recursive reference ... must not appear within - # its non-recursive term"). Both directions are therefore walked in - # one branch: join a link whichever side matches, emit the other. + # its non-recursive term"). Direction therefore travels as a + # column: the seed is emitted once per requested direction and + # each row only ever steps the way its own ``dir`` says, so a + # backward step is never followed by a forward one. + # + # ``_DIRECTION_SEED_VALUES`` holds fixed SQL fragments chosen by + # the validated direction name; no caller text is interpolated. + dir_values = _DIRECTION_SEED_VALUES[direction] rows = self.session.execute( - text(""" - WITH RECURSIVE tlc_graph(tlc, depth) AS ( - -- Seed - SELECT CAST(:tlc AS text), 0 + text(f""" + WITH RECURSIVE tlc_graph(tlc, depth, dir) AS ( + -- Seed, once per direction + SELECT CAST(:tlc AS text), 0, d.dir + FROM (VALUES {dir_values}) AS d(dir) UNION - -- Walk both directions: input → output and output → input - SELECT CASE WHEN tg.tlc = tl.input_tlc - THEN tl.output_tlc + -- Forward rows follow input → output; backward rows + -- follow output → input. Never the other way round. + SELECT CASE WHEN tg.dir = 'f' THEN tl.output_tlc ELSE tl.input_tlc END, - tg.depth + 1 + tg.depth + 1, + tg.dir FROM fsma.transformation_links tl - JOIN tlc_graph tg ON tg.tlc IN (tl.input_tlc, tl.output_tlc) + JOIN tlc_graph tg + ON (tg.dir = 'f' AND tg.tlc = tl.input_tlc) + OR (tg.dir = 'b' AND tg.tlc = tl.output_tlc) WHERE tl.tenant_id = :tid AND tg.depth < :max_depth ) - SELECT DISTINCT tlc FROM tlc_graph + -- The depth limit is judged per (lot, direction) BEFORE + -- the two walks are collapsed into one display depth: + -- a lot the forward walk only reaches at the limit was + -- never expanded forward, even when the backward walk + -- found it sooner (S -> A -> X -> S plus X -> Y, depth 2 + -- from S: X is at depth 2 forward and 1 backward, and Y + -- is missing). + SELECT tlc, + MIN(depth) AS depth, + BOOL_OR(dir = 'f') AS via_forward, + BOOL_OR(dir = 'b') AS via_backward, + BOOL_OR(depth >= :max_depth) AS at_limit + FROM ( + SELECT tlc, dir, MIN(depth) AS depth + FROM tlc_graph + GROUP BY tlc, dir + ) per_direction + GROUP BY tlc """), {"tid": tenant_id, "tlc": seed_tlc, "max_depth": max_depth}, ).fetchall() - return [r[0] for r in rows] if rows else [seed_tlc] + if not rows: + return TlcExpansion( + tlcs=[seed_tlc], + depth_exhausted=False, + direction=direction, + max_depth=max_depth, + reached_via={seed_tlc: {"depth": 0, "downstream": False, "upstream": False}}, + ) + tlcs: List[str] = [] + reached_via: Dict[str, Dict[str, Any]] = {} + deepest = 0 + at_limit_any = False + has_limit_column = False + for row in rows: + tlc = row[0] + row_depth = int(row[1]) if len(row) > 1 and row[1] is not None else 0 + via_forward = bool(row[2]) if len(row) > 2 and row[2] is not None else False + via_backward = bool(row[3]) if len(row) > 3 and row[3] is not None else False + if len(row) > 4 and row[4] is not None: + has_limit_column = True + at_limit_any = at_limit_any or bool(row[4]) + tlcs.append(tlc) + deepest = max(deepest, row_depth) + reached_via[tlc] = { + "depth": row_depth, + # The seed is reached by neither walk, whatever ``dir`` + # rows it was emitted with. + "downstream": via_forward and tlc != seed_tlc, + "upstream": via_backward and tlc != seed_tlc, + } + return TlcExpansion( + tlcs=tlcs, + # A lot a walk only found at the depth limit was never + # expanded in that direction (the recursive term stops at + # ``depth < max_depth``), so the graph may continue past it. + # ``at_limit`` is judged per direction before the walks are + # collapsed; rows without that column (older fakes) fall + # back to the collapsed depth. Conservative by design. + depth_exhausted=at_limit_any if has_limit_column else deepest >= max_depth, + direction=direction, + max_depth=max_depth, + reached_via=reached_via, + ) except (ProgrammingError, OperationalError) as e: if not _is_missing_table_error(e): # Fail closed (#1322 spirit): transient connectivity loss or @@ -193,7 +331,13 @@ def _expand_tlcs_via_transformation_links( self.session.rollback() except DatabaseError: _core.logger.debug("Rollback failed in recovery handler", exc_info=True) - return [seed_tlc] + return TlcExpansion( + tlcs=[seed_tlc], + depth_exhausted=False, + direction=direction, + max_depth=max_depth, + reached_via={seed_tlc: {"depth": 0, "downstream": False, "upstream": False}}, + ) def query_events_by_tlc( self, @@ -361,15 +505,18 @@ def _expand_tlcs_via_transformation_links_batch( tenant_id: str, seed_tlcs: List[str], depth: Optional[int] = None, + direction: str = "both", ) -> List[str]: """Batch variant of ``_expand_tlcs_via_transformation_links`` (#2515). Expands *all* ``seed_tlcs`` through ``fsma.transformation_links`` in a single ``WITH RECURSIVE`` traversal instead of one recursive-CTE round-trip per seed. The seed set is materialized as a ``VALUES`` list - feeding the recursion base case; the bidirectional walk and depth cap - are identical to the single-seed method. Returns the full reachable TLC - set (including the seeds), de-duplicated. + feeding the recursion base case; the directed walk and depth cap are + identical to the single-seed method (``both`` is the union of a + forward and a backward walk from each seed, never an undirected + walk). Returns the full reachable TLC set (including the seeds), + de-duplicated. Degrades to the seed set ONLY when ``transformation_links`` does not exist (fresh DB bootstrapping) — mirroring the single-seed fallback. @@ -384,35 +531,43 @@ def _expand_tlcs_via_transformation_links_batch( max_depth = depth if depth is not None else self.DEFAULT_TRAVERSAL_DEPTH if max_depth < 0: raise ValueError(f"depth must be >= 0, got {max_depth}") + direction = _validate_direction(direction) seeds = list(dict.fromkeys(seed_tlcs)) # preserve order, drop dupes seed_placeholders = ", ".join(f"(:seed_{i})" for i in range(len(seeds))) params: Dict[str, Any] = {"tid": tenant_id, "max_depth": max_depth} for i, s in enumerate(seeds): params[f"seed_{i}"] = s + dir_values = _DIRECTION_SEED_VALUES[direction] try: # Single recursive branch for the same reason as the single-seed # twin above: Postgres splits a recursive CTE at the LAST top-level # UNION, so a seed/forward/backward three-branch UNION fails with - # InvalidRecursion on every execution. + # InvalidRecursion on every execution. Direction travels as a + # column, exactly as in the single-seed walk. rows = self.session.execute( text(f""" WITH RECURSIVE seeds(tlc) AS ( VALUES {seed_placeholders} ), - tlc_graph(tlc, depth) AS ( - -- Seed set - SELECT CAST(tlc AS text), 0 FROM seeds + tlc_graph(tlc, depth, dir) AS ( + -- Seed set, once per direction + SELECT CAST(s.tlc AS text), 0, d.dir + FROM seeds s + CROSS JOIN (VALUES {dir_values}) AS d(dir) UNION - -- Walk both directions: input → output and output → input - SELECT CASE WHEN tg.tlc = tl.input_tlc - THEN tl.output_tlc + -- Forward rows follow input → output; backward rows + -- follow output → input. Never the other way round. + SELECT CASE WHEN tg.dir = 'f' THEN tl.output_tlc ELSE tl.input_tlc END, - tg.depth + 1 + tg.depth + 1, + tg.dir FROM fsma.transformation_links tl - JOIN tlc_graph tg ON tg.tlc IN (tl.input_tlc, tl.output_tlc) + JOIN tlc_graph tg + ON (tg.dir = 'f' AND tg.tlc = tl.input_tlc) + OR (tg.dir = 'b' AND tg.tlc = tl.output_tlc) WHERE tl.tenant_id = :tid AND tg.depth < :max_depth ) SELECT DISTINCT tlc FROM tlc_graph diff --git a/tests/shared/test_canonical_trace_cap_1282.py b/tests/shared/test_canonical_trace_cap_1282.py index 35e00c626..99dd35778 100644 --- a/tests/shared/test_canonical_trace_cap_1282.py +++ b/tests/shared/test_canonical_trace_cap_1282.py @@ -218,3 +218,89 @@ def test_max_depth_stops_before_max_results(self): links, truncated = store.trace_forward("tenant-1", "L0", max_depth=3, max_results=1000) assert len(links) == 3 assert truncated is False + + +# --------------------------------------------------------------------------- +# Gap analysis 2026-10-28, S1 — reaching max_depth is reported, not silent +# --------------------------------------------------------------------------- + + +class TestDepthExhaustedFlag_S1: + """``trace_walk`` reports ``depth_exhausted`` when a lot was reached at + ``max_depth`` and therefore never expanded. Before this flag the only + signal was ``truncated`` (the result cap), so a depth-capped trace read + as complete. The flag is conservative and costs no extra query.""" + + def test_depth_exhausted_when_chain_outruns_max_depth(self): + chain = {f"L{i}": [f"L{i + 1}"] for i in range(20)} + chain["L20"] = [] + session = _GraphSession(forward=chain) + store = CanonicalEventStore(session=session, dual_write=False, skip_chain_write=True) + + walk = store.trace_walk("tenant-1", "L0", "forward", max_depth=3, max_results=1000) + assert len(walk.links) == 3 + assert walk.truncated is False + assert walk.depth_exhausted is True + assert walk.direction == "forward" + assert walk.max_depth == 3 + # Lots at the depth limit are not expanded: exactly 3 queries (L0, L1, L2). + assert session.call_count == 3 + + def test_not_exhausted_when_graph_ends_before_max_depth(self): + session = _GraphSession(forward={"L0": ["L1"], "L1": ["L2"], "L2": []}) + store = CanonicalEventStore(session=session, dual_write=False, skip_chain_write=True) + + walk = store.trace_walk("tenant-1", "L0", "forward", max_depth=5, max_results=1000) + assert len(walk.links) == 2 + assert walk.truncated is False + assert walk.depth_exhausted is False + + def test_leaf_exactly_at_max_depth_is_reported_conservatively(self): + """L2 is reached at depth 2 == max_depth and is never queried, so + the walk cannot know it is a leaf. It must say so rather than + claim completeness.""" + session = _GraphSession(forward={"L0": ["L1"], "L1": ["L2"], "L2": []}) + store = CanonicalEventStore(session=session, dual_write=False, skip_chain_write=True) + + walk = store.trace_walk("tenant-1", "L0", "forward", max_depth=2, max_results=1000) + assert len(walk.links) == 2 + assert walk.depth_exhausted is True + + def test_backward_walk_reports_depth_exhausted(self): + session = _GraphSession(backward={"L3": ["L2"], "L2": ["L1"], "L1": ["L0"], "L0": []}) + store = CanonicalEventStore(session=session, dual_write=False, skip_chain_write=True) + + walk = store.trace_walk("tenant-1", "L3", "backward", max_depth=2, max_results=1000) + assert [l["input_tlc"] for l in walk.links] == ["L2", "L1"] + assert walk.depth_exhausted is True + assert walk.direction == "backward" + + def test_cap_and_depth_flags_are_independent(self): + """Result cap hit on the first expansion: ``truncated`` is True and, + because the frontier is dropped before any lot reaches + ``max_depth``, ``depth_exhausted`` stays False.""" + fan_out = [f"L{i}" for i in range(100)] + session = _GraphSession(forward={"ROOT": fan_out, **{c: [] for c in fan_out}}) + store = CanonicalEventStore(session=session, dual_write=False, skip_chain_write=True) + + walk = store.trace_walk("tenant-1", "ROOT", "forward", max_depth=5, max_results=10) + assert walk.truncated is True + assert walk.depth_exhausted is False + assert session.call_count == 1 + + def test_invalid_direction_rejected(self): + session = _GraphSession() + store = CanonicalEventStore(session=session, dual_write=False, skip_chain_write=True) + with pytest.raises(ValueError, match="direction"): + store.trace_walk("tenant-1", "L0", "sideways") + assert session.call_count == 0 + + def test_tuple_contract_wrappers_delegate_to_trace_walk(self): + chain = {f"L{i}": [f"L{i + 1}"] for i in range(5)} + chain["L5"] = [] + session = _GraphSession(forward=chain) + store = CanonicalEventStore(session=session, dual_write=False, skip_chain_write=True) + + links, truncated = store.trace_forward("tenant-1", "L0", max_depth=2) + assert len(links) == 2 + assert truncated is False diff --git a/tests/shared/test_cte_persistence_hardening_v2.py b/tests/shared/test_cte_persistence_hardening_v2.py index 4820d00c2..1feeb0be6 100644 --- a/tests/shared/test_cte_persistence_hardening_v2.py +++ b/tests/shared/test_cte_persistence_hardening_v2.py @@ -280,6 +280,158 @@ def _raise_runtime(sql, params): ) +# --------------------------------------------------------------------------- +# Gap analysis 2026-10-28, S2 — directed walk and depth-limit reporting +# --------------------------------------------------------------------------- + + +class TestTraversalDirection_S2: + """The expansion used to join a link whichever side matched, so a + backward step to a parent could be followed by a forward step to a + sibling lot that never touched the seed. Direction now travels as a + column in the recursive CTE and ``both`` is the union of a pure + forward and a pure backward walk. Reaching the depth limit is + reported through ``TlcExpansion.depth_exhausted``.""" + + @staticmethod + def _cte_sql(session: FakeSession) -> str: + return next(sql for sql, _ in session.calls if "RECURSIVE tlc_graph" in sql) + + def test_default_direction_is_both_and_seeds_both_rows(self): + session = FakeSession() + p = CTEPersistence(session=session) + result = p._expand_tlcs_detailed(tenant_id=TENANT_A, seed_tlc="TLC-1") + assert result.direction == "both" + sql = self._cte_sql(session) + assert "CAST('f' AS text)" in sql and "CAST('b' AS text)" in sql + # Forward rows only ever join on input_tlc, backward rows only on output_tlc. + assert "tg.dir = 'f' AND tg.tlc = tl.input_tlc" in sql + assert "tg.dir = 'b' AND tg.tlc = tl.output_tlc" in sql + # The undirected join is gone. + assert "tg.tlc IN (tl.input_tlc, tl.output_tlc)" not in sql + + def test_forward_only_seeds_forward_direction(self): + session = FakeSession() + p = CTEPersistence(session=session) + p._expand_tlcs_via_transformation_links(tenant_id=TENANT_A, seed_tlc="TLC-1", direction="forward") + sql = self._cte_sql(session) + assert "CAST('f' AS text)" in sql + assert "CAST('b' AS text)" not in sql + + def test_backward_only_seeds_backward_direction(self): + session = FakeSession() + p = CTEPersistence(session=session) + p._expand_tlcs_via_transformation_links(tenant_id=TENANT_A, seed_tlc="TLC-1", direction="backward") + sql = self._cte_sql(session) + assert "CAST('b' AS text)" in sql + assert "CAST('f' AS text)" not in sql + + def test_invalid_direction_is_rejected_before_sql(self): + session = FakeSession() + p = CTEPersistence(session=session) + with pytest.raises(ValueError, match="direction"): + p._expand_tlcs_detailed(tenant_id=TENANT_A, seed_tlc="TLC-1", direction="sideways") + assert not session.calls + + def test_batch_walk_is_directed_too(self): + session = FakeSession() + p = CTEPersistence(session=session) + p._expand_tlcs_via_transformation_links_batch(tenant_id=TENANT_A, seed_tlcs=["TLC-1", "TLC-2"], direction="forward") + sql = next(sql for sql, _ in session.calls if "tlc_graph(tlc, depth, dir)" in sql) + assert "CROSS JOIN (VALUES (CAST('f' AS text)))" in sql + assert "tg.tlc IN (tl.input_tlc, tl.output_tlc)" not in sql + + def test_depth_exhausted_when_a_lot_sits_at_the_limit(self): + """Rows come back as (tlc, depth, via_forward, via_backward). A lot + at depth == max_depth was never expanded, so the expansion must + say the limit was reached.""" + session = FakeSession() + session.add_rule( + r"FROM\s+tlc_graph", + _FakeResult(rows=[("TLC-1", 0, True, True), ("TLC-2", 1, True, False), ("TLC-3", 2, True, False)]), + ) + p = CTEPersistence(session=session) + result = p._expand_tlcs_detailed(tenant_id=TENANT_A, seed_tlc="TLC-1", depth=2) + assert result.tlcs == ["TLC-1", "TLC-2", "TLC-3"] + assert result.depth_exhausted is True + assert result.max_depth == 2 + # The seed is reached by neither walk; TLC-2 and TLC-3 are downstream. + assert result.reached_via["TLC-1"] == {"depth": 0, "downstream": False, "upstream": False} + assert result.reached_via["TLC-3"] == {"depth": 2, "downstream": True, "upstream": False} + + def test_not_exhausted_when_graph_ends_early(self): + session = FakeSession() + session.add_rule(r"FROM\s+tlc_graph", _FakeResult(rows=[("TLC-1", 0, True, True), ("TLC-0", 1, False, True)])) + p = CTEPersistence(session=session) + result = p._expand_tlcs_detailed(tenant_id=TENANT_A, seed_tlc="TLC-1", depth=5) + assert result.depth_exhausted is False + assert result.reached_via["TLC-0"]["upstream"] is True + + def test_depth_limit_is_judged_per_direction_before_collapsing(self): + """Codex review on #3713: with S -> A -> X -> S and X -> Y, a + ``both`` trace from S at depth 2 reaches X at depth 2 forward (never + expanded, so Y is missing) but at depth 1 backward. Collapsing to + MIN(depth) first made every lot look shallower than the limit and + reported ``depth_exhausted=False``. The SQL now judges the limit per + (lot, direction) and returns it as a fifth column.""" + session = FakeSession() + session.add_rule( + r"FROM\s+tlc_graph", + _FakeResult( + rows=[ + ("S", 0, True, True, False), + ("A", 1, True, True, False), + ("X", 1, True, True, True), # forward min depth 2 == limit + ] + ), + ) + p = CTEPersistence(session=session) + result = p._expand_tlcs_detailed(tenant_id=TENANT_A, seed_tlc="S", depth=2) + assert result.tlcs == ["S", "A", "X"] + assert result.depth_exhausted is True + # Display depth is still the shallowest reach. + assert result.reached_via["X"]["depth"] == 1 + sql = self._cte_sql(session) + assert "GROUP BY tlc, dir" in sql + assert "BOOL_OR(depth >= :max_depth)" in sql + assert "MIN(depth)" in sql + + def test_explicit_limit_column_wins_over_collapsed_depth(self): + """When the fifth column says no lot sat at the limit in any + direction, a collapsed depth equal to the limit does not raise the + flag (the per-direction judgement is authoritative).""" + session = FakeSession() + session.add_rule( + r"FROM\s+tlc_graph", + _FakeResult(rows=[("S", 0, True, True, False), ("A", 1, True, False, False)]), + ) + p = CTEPersistence(session=session) + result = p._expand_tlcs_detailed(tenant_id=TENANT_A, seed_tlc="S", depth=1) + assert result.depth_exhausted is False + + def test_single_column_rows_still_work(self): + """Older fakes and the missing-table fallback hand back bare TLC + rows; the detailed form must not require the extra columns.""" + session = FakeSession() + session.add_rule(r"FROM\s+tlc_graph", _FakeResult(rows=[("TLC-1",), ("TLC-9",)])) + p = CTEPersistence(session=session) + result = p._expand_tlcs_detailed(tenant_id=TENANT_A, seed_tlc="TLC-1", depth=5) + assert result.tlcs == ["TLC-1", "TLC-9"] + assert result.depth_exhausted is False + + def test_missing_table_fallback_is_a_detailed_result_too(self): + session = FakeSession() + + def _raise_programming(sql, params): + raise ProgrammingError("stmt", {}, Exception('relation "fsma.transformation_links" does not exist')) + + session.add_rule(r"RECURSIVE tlc_graph", _raise_programming) + p = CTEPersistence(session=session) + result = p._expand_tlcs_detailed(tenant_id=TENANT_A, seed_tlc="TLC-X") + assert result.tlcs == ["TLC-X"] + assert result.depth_exhausted is False + + # --------------------------------------------------------------------------- # Shared-kernel #1336 — gaps in prior hardening test coverage # --------------------------------------------------------------------------- diff --git a/tests/shared/test_cte_traversal_live.py b/tests/shared/test_cte_traversal_live.py index 3fc76e821..56f4aec14 100644 --- a/tests/shared/test_cte_traversal_live.py +++ b/tests/shared/test_cte_traversal_live.py @@ -94,3 +94,99 @@ def test_tlc_traversal_executes_and_walks_both_directions() -> None: assert persistence._expand_tlcs_via_transformation_links_batch(tenant_id=tenant_id, seed_tlcs=[]) == [] session.rollback() + + +def test_tlc_traversal_is_directed_and_excludes_sibling_lots() -> None: + """S2 (gap analysis 2026-10-28): ``both`` is the union of a forward and + a backward walk, not an undirected walk. With A → B and A → S, a trace + from B must reach A (its input) but never S (a sibling output of A that + B never touched); a forward trace from B must not step back to A at all. + Also checks the depth-limit flag against real SQL.""" + from shared.cte_persistence.core import CTEPersistence + from sqlalchemy import create_engine + from sqlalchemy.orm import Session + + engine = create_engine(sync_postgres_url(DATABASE_URL)) + tenant_id = str(uuid.uuid4()) + + with Session(engine) as session: + # A → B → C (seed chain) + # A → S (sibling of B; shares the input A) + # S → T (downstream of the sibling) + _seed_link(session, tenant_id, "A", "B") + _seed_link(session, tenant_id, "B", "C") + _seed_link(session, tenant_id, "A", "S") + _seed_link(session, tenant_id, "S", "T") + + persistence = CTEPersistence(session) + + both = persistence._expand_tlcs_detailed(tenant_id=tenant_id, seed_tlc="B", direction="both") + assert set(both.tlcs) == {"A", "B", "C"} + assert both.reached_via["A"] == {"depth": 1, "downstream": False, "upstream": True} + assert both.reached_via["C"] == {"depth": 1, "downstream": True, "upstream": False} + assert both.depth_exhausted is False + + forward = persistence._expand_tlcs_detailed(tenant_id=tenant_id, seed_tlc="B", direction="forward") + assert set(forward.tlcs) == {"B", "C"} + + backward = persistence._expand_tlcs_detailed(tenant_id=tenant_id, seed_tlc="S", direction="backward") + assert set(backward.tlcs) == {"S", "A"} + + # Depth limit: from A forward with depth=1, B and S are reached at + # the limit and never expanded, so the flag must be raised and C, T + # must be absent. + capped = persistence._expand_tlcs_detailed(tenant_id=tenant_id, seed_tlc="A", depth=1, direction="forward") + assert set(capped.tlcs) == {"A", "B", "S"} + assert capped.depth_exhausted is True + + # Batch twin honours direction as well: forward from B and S never + # reaches A. + batch = persistence._expand_tlcs_via_transformation_links_batch( + tenant_id=tenant_id, seed_tlcs=["B", "S"], direction="forward" + ) + assert set(batch) == {"B", "C", "S", "T"} + + session.rollback() + + +def test_depth_limit_is_reported_per_direction_on_a_cycle() -> None: + """Codex review on #3713: S -> A -> X -> S plus X -> Y. A ``both`` trace + from S at depth 2 reaches X at depth 2 forward (so Y is never reached) + and at depth 1 backward. Collapsing the two walks to MIN(depth) before + judging the limit hid the exhausted forward frontier.""" + from shared.cte_persistence.core import CTEPersistence + from sqlalchemy import create_engine + from sqlalchemy.orm import Session + + engine = create_engine(sync_postgres_url(DATABASE_URL)) + tenant_id = str(uuid.uuid4()) + + with Session(engine) as session: + _seed_link(session, tenant_id, "S", "A") + _seed_link(session, tenant_id, "A", "X") + _seed_link(session, tenant_id, "X", "S") + _seed_link(session, tenant_id, "X", "Y") + + persistence = CTEPersistence(session) + + both = persistence._expand_tlcs_detailed(tenant_id=tenant_id, seed_tlc="S", depth=2, direction="both") + assert set(both.tlcs) == {"S", "A", "X"} + assert "Y" not in both.tlcs + assert both.reached_via["X"]["depth"] == 1 + assert both.depth_exhausted is True + + # One more hop and the forward walk reaches Y, but only at the new + # limit, so Y itself is unexpanded and the flag stays raised. + at_three = persistence._expand_tlcs_detailed(tenant_id=tenant_id, seed_tlc="S", depth=3, direction="both") + assert set(at_three.tlcs) == {"S", "A", "X", "Y"} + assert at_three.reached_via["Y"]["depth"] == 3 + assert at_three.depth_exhausted is True + + # At depth 4 every lot is expanded below the limit in the direction + # that found it (the cycle only re-finds S, A and X at depths above + # their per-direction minimum), so nothing is left unexplored. + at_four = persistence._expand_tlcs_detailed(tenant_id=tenant_id, seed_tlc="S", depth=4, direction="both") + assert set(at_four.tlcs) == {"S", "A", "X", "Y"} + assert at_four.depth_exhausted is False + + session.rollback()