diff --git a/backend/app/lineage_ingestion.py b/backend/app/lineage_ingestion.py index 234c0d2e7..64fb05c04 100644 --- a/backend/app/lineage_ingestion.py +++ b/backend/app/lineage_ingestion.py @@ -28,6 +28,9 @@ from lineageweave.lineage_persistence import lineage_edge_specs from lineageweave.models import Edge, Record +ISOLATION_NO_COMPARISON_GROUP = "no_comparison_group" +ISOLATION_COMPARISON_CANDIDATES_AVAILABLE = "comparison_candidates_available" + # Accepted ADR 0200 (points 2-3) authorizes exactly one anchor method: # expected-information estimates honestly labeled as validated by the # channels' internal response structure only, pending the TEPP @@ -54,6 +57,42 @@ def reconstruct_group_key(row: Mapping[str, Any]) -> str: return stored_group or str(row["process_unit_id"] or row["corporate_entity_id"]) +def focused_isolation_reason( + focus_post_id: str | None, + visible_posts: list[Mapping[str, Any]], + node_count: int, +) -> str | None: + """Why a focused Event Lineage DAG is empty (ADR 0143). + + Count only ABAC-visible group members. A hidden sibling must not + flip ``no_comparison_group`` into candidate availability. Import + backfills ``thread_group_key`` from process-unit code, so key + presence is not evidence a real comparison group existed. + """ + if focus_post_id is None or node_count > 0: + return None + focus_id = str(focus_post_id) + focus_row = next( + (row for row in visible_posts if str(row["post_id"]) == focus_id), + None, + ) + if focus_row is None: + return None + group_key = reconstruct_group_key(focus_row) + visible_group_size = sum( + 1 for row in visible_posts if reconstruct_group_key(row) == group_key + ) + if visible_group_size <= 1: + return ISOLATION_NO_COMPARISON_GROUP + # Multiple current members do not prove that the published projection + # was rebuilt after they arrived. Report only candidate availability, + # never a completed-comparison or no-relation conclusion. + return ISOLATION_COMPARISON_CANDIDATES_AVAILABLE + + + + + def records_from_source_posts(rows: list[Mapping[str, Any]]) -> list[Record]: """Map ``source_post`` rows onto reconstruct ``Record``s. @@ -407,7 +446,11 @@ async def visible_lineage_graph( ) truncated = False - return _lineage_graph_payload(visible, edge_rows, truncated) + payload = _lineage_graph_payload(visible, edge_rows, truncated) + payload["isolation_reason"] = focused_isolation_reason( + focus_post_id, visible_all, len(visible) + ) + return payload async def interval_relations_for_post( diff --git a/tests/test_lineage_ingestion.py b/tests/test_lineage_ingestion.py index f2b6be6d8..3e4c20394 100644 --- a/tests/test_lineage_ingestion.py +++ b/tests/test_lineage_ingestion.py @@ -606,7 +606,12 @@ async def fetch(self, query: str, *_args): assert {node["id"] for node in focused["nodes"]} == {"post-a", "post-b"} assert len(focused["edges"]) == 1 assert focused["truncated"] is False - assert isolated == {"nodes": [], "edges": [], "truncated": False} + assert isolated == { + "nodes": [], + "edges": [], + "truncated": False, + "isolation_reason": "no_comparison_group", + } assert [node["id"] for node in hidden_neighbor["nodes"]] == ["post-a"] assert hidden_neighbor["edges"] == [] @@ -875,3 +880,107 @@ async def fetch(self, query: str): edge["source"] in node_ids and edge["target"] in node_ids for edge in merged["edges"] ) + + +def _post( + post_id: str, + title: str, + thread_group_key: str, + created_at: datetime, +) -> dict[str, object]: + return { + "post_id": post_id, + "post_title": title, + "voc_type_code": "voc", + "visibility_code": "public", + "corporate_entity_id": "corp", + "process_unit_id": "pu", + "thread_group_key": thread_group_key, + "created_at": created_at, + } + + +class FakeConnection: + def __init__(self, posts: list[dict[str, object]], edges: list[dict[str, object]]) -> None: + self.posts = posts + self.edges = edges + self.executions: list[tuple[object, ...]] = [] + + async def fetch(self, query: str): + return self.edges if "post_lineage_edge" in query else self.posts + + async def fetchval(self, query: str): + assert "to_regclass('public.lineage_channel_weight')" in query + return False + + async def execute(self, *args: object) -> None: + self.executions.append(args) + + +def test_two_visible_group_members_report_comparison_candidates_available() -> None: + connection = FakeConnection( + [ + _post("post-a", "A", "thread-a", datetime(2026, 1, 1)), + _post("post-b", "B", "thread-a", datetime(2026, 1, 2)), + ], + [], + ) + focused = asyncio.run( + visible_lineage_graph(connection, lambda row: True, focus_post_id="post-a") + ) + assert focused == { + "nodes": [], + "edges": [], + "truncated": False, + "isolation_reason": "comparison_candidates_available", + } + + +def test_hidden_sibling_does_not_flip_isolation_to_candidates_available() -> None: + """ABAC-hidden siblings must not leak through candidate availability.""" + connection = FakeConnection( + [ + _post("post-c", "C", "thread-c", datetime(2026, 1, 3)), + _post("post-d", "D", "thread-c", datetime(2026, 1, 4)), + ], + [], + ) + isolated = asyncio.run( + visible_lineage_graph( + connection, + lambda row: str(row["post_id"]) != "post-d", + focus_post_id="post-c", + ) + ) + assert isolated["isolation_reason"] == "no_comparison_group" + assert isolated["nodes"] == [] + + +def test_inaccessible_focus_does_not_report_an_isolation_reason() -> None: + connection = FakeConnection( + [_post("post-a", "A", "thread-a", datetime(2026, 1, 1))], + [], + ) + hidden = asyncio.run( + visible_lineage_graph( + connection, + lambda row: False, + focus_post_id="post-a", + ) + ) + assert hidden == { + "nodes": [], + "edges": [], + "truncated": False, + "isolation_reason": None, + } + + +def test_landing_graph_never_reports_an_isolation_reason() -> None: + connection = FakeConnection( + [_post("post-c", "C", "thread-c", datetime(2026, 1, 3))], + [], + ) + landing = asyncio.run(visible_lineage_graph(connection, lambda row: True)) + assert landing["isolation_reason"] is None + assert [node["id"] for node in landing["nodes"]] == ["post-c"]