feat: name Event Lineage isolation reasons (v2.16.0)

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(

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"]