diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 906990241..a1606c6e0 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -223,11 +223,14 @@ lives in `lineageweave/keyman_extraction.py` and talks to contextual-orchestrator; persist is `backend/app/keyman_ingestion.py`. `GET /api/lineage` returns the ABAC-filtered reconstruct graph -(`{nodes, edges}`) from persisted `post_lineage_edge` rows. Each node +(`{nodes, edges, truncated}`) from persisted `post_lineage_edge` rows. Each node includes `group` from the same `reconstruct_group_key()` rebuild uses (persisted `thread_group_key`, else process unit, else corp). Each direct edge includes `interval_relation_code` / `interval_relation_label` (Allen, 1983; ADR 0161) computed from the two posts' observed windows. +Global Ask merges cited threads from one post/edge fetch pair and +caps the payload at the landing node bound, keeping cited posts first +(ADR 0169). Open a cited post to read the focused thread. `POST /api/lineage/rebuild` (`post_admin`) re-runs `reconstruct()` over every `source_post` and rewrites those edges, then names the interval relation in the same transaction. Reconstruct grouping is diff --git a/CHANGELOG.d/2.14.0-ask-batched-lineage-graph.md b/CHANGELOG.d/2.14.0-ask-batched-lineage-graph.md new file mode 100644 index 000000000..e292715f2 --- /dev/null +++ b/CHANGELOG.d/2.14.0-ask-batched-lineage-graph.md @@ -0,0 +1,2 @@ +Global Ask merges cited Event Lineage from one post/edge fetch pair +and names truncation at the landing node bound (ADR 0169 / #568). diff --git a/CHANGELOG.md b/CHANGELOG.md index a8fec0f90..98f3c45c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -162,6 +162,16 @@ All notable changes to this project are documented here. Format follows is unwired, with `이 범위의 일정을 아직 받을 수 없습니다`. Weekly VOC and newspaper stay on the board. +## [2.14.0] - 2026-08-24 + +### Changed + +- Global Ask now builds the merged cited Event Lineage graph from one + `source_post` scan and one `post_lineage_edge` read, keeps cited posts + first when the graph exceeds the landing node bound, and names + **truncated** so a click still opens that cited post (ADR 0169 / #568). + Do not invent a theta. + ## [2.13.2] - 2026-08-24 ### Added diff --git a/backend/app/lineage_ingestion.py b/backend/app/lineage_ingestion.py index 7d3107db1..234c0d2e7 100644 --- a/backend/app/lineage_ingestion.py +++ b/backend/app/lineage_ingestion.py @@ -272,6 +272,12 @@ async def rebuild_lineage(conn: asyncpg.Connection) -> list[Edge]: return edges +# Browser landing viewport and Global Ask merged-graph payload share this +# bound. Ask keeps cited posts first, then newest remaining nodes, and +# names truncation instead of shipping an unbounded component (ADR 0169). +_LINEAGE_GRAPH_NODE_LIMIT = 500 + + def _interval_payload(row: Mapping[str, Any]) -> dict[str, Any]: code = row.get("interval_relation_code") if not code: @@ -283,19 +289,8 @@ def _interval_payload(row: Mapping[str, Any]) -> dict[str, Any]: return payload -async def visible_lineage_graph( - conn: asyncpg.Connection, - can_see_post, - limit: int = 500, - focus_post_id: str | None = None, - include_isolated: bool = False, -) -> dict[str, Any]: - """ABAC-filtered graph bounded for the browser's initial viewport. - - The persisted graph can contain tens of thousands of posts. The UI opens - individual posts for complete lineage, while this landing projection keeps - only the newest ``limit`` visible nodes and edges between them. - """ +async def _fetch_visible_lineage_rows(conn: asyncpg.Connection, can_see_post): + """One ABAC-filtered ``source_post`` scan plus one edge-table read.""" posts = await conn.fetch( "select post_id, post_title, voc_type_code, visibility_code, " "corporate_entity_id, process_unit_id, thread_group_key, created_at " @@ -306,40 +301,43 @@ async def visible_lineage_graph( "select parent_post_id, child_post_id, fused_score, " "interval_relation_code from post_lineage_edge" ) + return visible_all, edge_rows - if focus_post_id is None: - visible = sorted( - visible_all, - key=lambda row: (row["created_at"], str(row["post_id"])), - reverse=True, - )[:limit] - truncated = len(visible_all) > len(visible) - else: - focus_id = str(focus_post_id) - focus_visible = any(str(row["post_id"]) == focus_id for row in visible_all) - neighbors: dict[str, set[str]] = {} - for edge in edge_rows: - parent_id = str(edge["parent_post_id"]) - child_id = str(edge["child_post_id"]) - neighbors.setdefault(parent_id, set()).add(child_id) - neighbors.setdefault(child_id, set()).add(parent_id) - - component_ids: set[str] = set() - frontier = [focus_id] if focus_visible else [] - while frontier: - current_id = frontier.pop() - if current_id in component_ids: - continue - component_ids.add(current_id) - frontier.extend(neighbors.get(current_id, set()) - component_ids) - visible = ( - [row for row in visible_all if str(row["post_id"]) in component_ids] - if include_isolated or len(component_ids) > 1 - else [] - ) - truncated = False +def _undirected_neighbors(edge_rows) -> dict[str, set[str]]: + neighbors: dict[str, set[str]] = {} + for edge in edge_rows: + parent_id = str(edge["parent_post_id"]) + child_id = str(edge["child_post_id"]) + neighbors.setdefault(parent_id, set()).add(child_id) + neighbors.setdefault(child_id, set()).add(parent_id) + return neighbors + + +def _connected_visible_component( + focus_id: str, + neighbors: dict[str, set[str]], + allowed: set[str], +) -> set[str]: + """Walk the undirected reconstruct graph; return visible posts only. + + Invisible posts may still bridge two visible posts so the component + matches ``visible_lineage_graph`` focus mode. + """ + if focus_id not in allowed: + return set() + component_ids: set[str] = set() + frontier = [focus_id] + while frontier: + current_id = frontier.pop() + if current_id in component_ids: + continue + component_ids.add(current_id) + frontier.extend(neighbors.get(current_id, set()) - component_ids) + return component_ids & allowed + +def _lineage_graph_payload(visible, edge_rows, truncated: bool) -> dict[str, Any]: visible_ids = {str(row["post_id"]) for row in visible} visible_edges = [ row @@ -375,6 +373,43 @@ async def visible_lineage_graph( return {"nodes": nodes, "edges": edges, "truncated": truncated} +async def visible_lineage_graph( + conn: asyncpg.Connection, + can_see_post, + limit: int = _LINEAGE_GRAPH_NODE_LIMIT, + focus_post_id: str | None = None, + include_isolated: bool = False, +) -> dict[str, Any]: + """ABAC-filtered graph bounded for the browser's initial viewport. + + The persisted graph can contain tens of thousands of posts. The UI opens + individual posts for complete lineage, while this landing projection keeps + only the newest ``limit`` visible nodes and edges between them. + """ + visible_all, edge_rows = await _fetch_visible_lineage_rows(conn, can_see_post) + + if focus_post_id is None: + visible = sorted( + visible_all, + key=lambda row: (row["created_at"], str(row["post_id"])), + reverse=True, + )[:limit] + truncated = len(visible_all) > len(visible) + else: + focus_id = str(focus_post_id) + neighbors = _undirected_neighbors(edge_rows) + allowed = {str(row["post_id"]) for row in visible_all} + component_ids = _connected_visible_component(focus_id, neighbors, allowed) + visible = ( + [row for row in visible_all if str(row["post_id"]) in component_ids] + if include_isolated or len(component_ids) > 1 or focus_id in neighbors + else [] + ) + truncated = False + + return _lineage_graph_payload(visible, edge_rows, truncated) + + async def interval_relations_for_post( conn: asyncpg.Connection, post_id: str ) -> dict[str, dict[str, Any]]: @@ -411,36 +446,36 @@ async def lineage_graphs_for_posts( conn: asyncpg.Connection, can_see_post, post_ids: list[str], + node_limit: int = _LINEAGE_GRAPH_NODE_LIMIT, ) -> dict[str, Any]: - """Merge each post's full reconstructed thread into one ``LineageGraph``. - - An Ask Agent answer can cite several posts from unrelated reconstruct - threads -- e.g. two separate customer complaints that happen to share a - keyword. The frontend's ``LineageDag`` already renders one ``LineageGraph`` - as several independent git-branch-style figures, one per - ``reconstruct_group_key`` (see ``lineageLayout.ts``'s ``layoutLineageDag``); - merging every cited post's thread into a single graph is enough to get - that multi-graph rendering for free, no new frontend layout needed. - - ponytail: one ``visible_lineage_graph`` call per post (each a bounded - ``source_post`` + full ``post_lineage_edge`` scan) -- fine for the - existing citation cap (``_POST_CHAT_SOURCE_LIMIT`` = 8), revisit with a - single batched query if that cap grows materially. + """Merge each cited post's reconstructed thread into one ``LineageGraph``. + + One ``source_post`` scan and one ``post_lineage_edge`` read cover every + citation. Cited posts stay first when the merged graph is larger than + ``node_limit``; ``truncated`` is then true so Ask can name the bound + (ADR 0151 / 0169). Isolated cited posts still appear. """ - nodes_by_id: dict[str, dict[str, Any]] = {} - edges_by_key: dict[tuple[str, str], dict[str, Any]] = {} - truncated = False - for post_id in dict.fromkeys(post_ids): - graph = await visible_lineage_graph( - conn, can_see_post, focus_post_id=post_id, include_isolated=True - ) - truncated = truncated or graph["truncated"] - for node in graph["nodes"]: - nodes_by_id[node["id"]] = node - for edge in graph["edges"]: - edges_by_key[(edge["source"], edge["target"])] = edge - return { - "nodes": list(nodes_by_id.values()), - "edges": list(edges_by_key.values()), - "truncated": truncated, - } + unique_ids = list(dict.fromkeys(str(post_id) for post_id in post_ids)) + if not unique_ids: + return {"nodes": [], "edges": [], "truncated": False} + visible_all, edge_rows = await _fetch_visible_lineage_rows(conn, can_see_post) + neighbors = _undirected_neighbors(edge_rows) + allowed = {str(row["post_id"]) for row in visible_all} + keep_ids: set[str] = set() + cited_visible: list[str] = [] + for post_id in unique_ids: + component = _connected_visible_component(post_id, neighbors, allowed) + if not component: + continue + keep_ids.update(component) + cited_visible.append(post_id) + rows_by_id = {str(row["post_id"]): row for row in visible_all} + newest_ids = sorted( + keep_ids, + key=lambda pid: (rows_by_id[pid]["created_at"], pid), + reverse=True, + ) + ordered_ids = list(dict.fromkeys([*cited_visible, *newest_ids])) + truncated = len(ordered_ids) > node_limit + visible = [rows_by_id[post_id] for post_id in ordered_ids[:node_limit]] + return _lineage_graph_payload(visible, edge_rows, truncated) diff --git a/docs/adr/0151-ask-multi-lineage-graph.md b/docs/adr/0151-ask-multi-lineage-graph.md index 22c9a14f4..1c10840ef 100644 --- a/docs/adr/0151-ask-multi-lineage-graph.md +++ b/docs/adr/0151-ask-multi-lineage-graph.md @@ -2,7 +2,7 @@ - Status: Accepted - Date: 2026-08-22 -- Related: [0090](0090-global-ask-lineage-timeline-expansion.md), [0064](0064-lineage-evidence-and-tree-assembly.md) +- Related: [0090](0090-global-ask-lineage-timeline-expansion.md), [0064](0064-lineage-evidence-and-tree-assembly.md), [0169](0169-ask-batched-lineage-graph.md) ## Context @@ -51,11 +51,9 @@ independent branch-tree figures with no new frontend layout code. - An Ask answer's lineage evidence is now visually traceable per cited thread, not summarized as prose for one anchor post only. -- `lineage_graphs_for_posts` issues one `visible_lineage_graph` call per - cited post (each a bounded `source_post` scan plus a full - `post_lineage_edge` table read); acceptable at the current citation cap - (`_POST_CHAT_SOURCE_LIMIT` = 8) and the existing `visible_lineage_graph` - precedent for the post-detail popup, revisit with a single batched query - if the citation cap grows materially. +- `lineage_graphs_for_posts` loads visible posts and lineage edges + once, then slices per-citation subgraphs in memory (ADR 0169). The + merged payload is bounded; `truncated` names that bound. Cited + posts stay first so every cited thread still appears. - The response payload grows by one field (`lineage_graph`); existing consumers that ignore unknown fields are unaffected. diff --git a/docs/adr/0169-ask-batched-lineage-graph.md b/docs/adr/0169-ask-batched-lineage-graph.md new file mode 100644 index 000000000..7a4e2f0e4 --- /dev/null +++ b/docs/adr/0169-ask-batched-lineage-graph.md @@ -0,0 +1,43 @@ +# ADR 0169 — Global Ask merges cited lineage from one fetch pair + +**Decision status:** Accepted +**Date:** 2026-08-24 +**Related:** [0151](0151-ask-multi-lineage-graph.md), [0090](0090-global-ask-lineage-timeline-expansion.md), issue [#568](https://github.com/ContextualWisdomLab/LineageWeave/issues/568) + +## Context + +ADR 0151 merges every cited post's reconstructed thread into one +`LineageGraph` so Ask can reuse `LineageDag`. The first implementation +called `visible_lineage_graph` once per citation. Each call re-read +every eligible `source_post` row and the full `post_lineage_edge` +table. Focus mode also returned the entire connected component with +`truncated: false`, so a cited post inside a large component could +inflate the Ask payload without naming a bound. + +## Decision + +1. `lineage_graphs_for_posts` loads visible posts and lineage edges + once, partitions connected components in memory, and slices the + per-citation subgraphs from that pair of reads. +2. The merged payload uses the same node bound as the landing + viewport (`_LINEAGE_GRAPH_NODE_LIMIT` = 500). Cited posts stay + first. Remaining component nodes follow newest-first. Nodes beyond + the bound are omitted and `truncated` is true. +3. Isolated cited posts still appear. Analysis-run knowledge cutoff + and leftover pairs are unchanged. Do not invent a theta. + +## Considered alternatives + +- Keep the per-citation refetch because the citation cap is 8: + rejected. The cost is the table scan, not the citation count, and + the uncapped component is independent of that cap. +- Depth-from-citation hop limit instead of a node cap: deferred. A + hop limit can hide a cited thread's own branch point; the node cap + keeps cited posts visible and names the bound. + +## Consequences + +- N citations issue one `source_post` query and one + `post_lineage_edge` query. +- A large connected component no longer ships unbounded into Ask. + Open a cited post to read the full focused thread. diff --git a/docs/adr/README.md b/docs/adr/README.md index 0af8db2cd..8300295f4 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -10,7 +10,7 @@ decision from them. | Supporting document | Normative ADR | |---|---| | [`product-technical-gap-baseline.md`](../product-technical-gap-baseline.md) | Product/technical traceability projection across the ADR set; ADRs remain normative | -| [`lineage-bi-research-notes.md`](../lineage-bi-research-notes.md) | [0084](0084-lineage-research-grounding.md), [0062](0062-semantic-unit-embedding.md), [0064](0064-lineage-evidence-and-tree-assembly.md), [0024](0024-rankweave-fusion-fail-closed.md), [0165](0165-quantity-script-display.md), [0167](0167-rankweave-ranking-channel-evidence.md), [0202](0202-ask-event-time-filter.md) | +| [`lineage-bi-research-notes.md`](../lineage-bi-research-notes.md) | [0084](0084-lineage-research-grounding.md), [0062](0062-semantic-unit-embedding.md), [0064](0064-lineage-evidence-and-tree-assembly.md), [0024](0024-rankweave-fusion-fail-closed.md), [0165](0165-quantity-script-display.md), [0167](0167-rankweave-ranking-channel-evidence.md), [0169](0169-ask-batched-lineage-graph.md), [0202](0202-ask-event-time-filter.md) | | [`PROV_O_IMPLEMENTATION.md`](../PROV_O_IMPLEMENTATION.md) | [0065](0065-prov-o-provenance-boundary.md) | | [`PROV_O_IMPLEMENTATION_MATRIX.md`](../PROV_O_IMPLEMENTATION_MATRIX.md) | [0065](0065-prov-o-provenance-boundary.md) | | [`ONTOLOGY_NAMESPACE_INVENTORY.md`](../doctoring/ONTOLOGY_NAMESPACE_INVENTORY.md) | [0157](0157-public-ontology-namespace-identity.md) | diff --git a/tests/test_lineage_ingestion.py b/tests/test_lineage_ingestion.py index 488b6066b..f2b6be6d8 100644 --- a/tests/test_lineage_ingestion.py +++ b/tests/test_lineage_ingestion.py @@ -593,12 +593,22 @@ async def fetch(self, query: str, *_args): connection, lambda row: True, limit=1, focus_post_id="post-c" ) ) + hidden_neighbor = asyncio.run( + ingestion.visible_lineage_graph( + connection, + lambda row: row["post_id"] != "post-b", + limit=1, + focus_post_id="post-a", + ) + ) assert [node["id"] for node in landing["nodes"]] == ["post-c"] 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 [node["id"] for node in hidden_neighbor["nodes"]] == ["post-a"] + assert hidden_neighbor["edges"] == [] def test_visible_lineage_graph_attaches_allen_labels() -> None: @@ -767,3 +777,101 @@ async def fetch(self, query: str): merged = asyncio.run(lineage_graphs_for_posts(FakeConnection(), lambda row: True, [])) assert merged == {"nodes": [], "edges": [], "truncated": False} + + +def test_lineage_graphs_for_posts_fetches_posts_and_edges_once_for_n_citations() -> None: + """N citations must not refetch source_post / post_lineage_edge per cite.""" + + class FakeConnection: + posts = [ + { + "post_id": f"post-{index}", + "post_title": f"P{index}", + "voc_type_code": "voc", + "visibility_code": "public", + "corporate_entity_id": "corp", + "process_unit_id": "pu", + "thread_group_key": f"thread-{index}", + "created_at": datetime(2026, 1, index), + } + for index in range(1, 5) + ] + edges = [ + { + "parent_post_id": "post-1", + "child_post_id": "post-2", + "fused_score": 0.5, + } + ] + queries: list[str] = [] + + async def fetch(self, query: str): + self.queries.append(query) + return self.edges if "post_lineage_edge" in query else self.posts + + connection = FakeConnection() + merged = asyncio.run( + lineage_graphs_for_posts( + connection, + lambda row: True, + ["post-1", "post-3", "post-4", "post-1"], + ) + ) + + post_queries = [query for query in connection.queries if "from source_post" in query] + edge_queries = [query for query in connection.queries if "post_lineage_edge" in query] + assert len(post_queries) == 1 + assert len(edge_queries) == 1 + assert {node["id"] for node in merged["nodes"]} == { + "post-1", + "post-2", + "post-3", + "post-4", + } + assert merged["truncated"] is False + + +def test_lineage_graphs_for_posts_names_truncation_and_keeps_cited_posts() -> None: + posts = [ + { + "post_id": f"post-{index}", + "post_title": f"P{index}", + "voc_type_code": "voc", + "visibility_code": "public", + "corporate_entity_id": "corp", + "process_unit_id": "pu", + "thread_group_key": "thread-chain", + "created_at": datetime(2026, 1, index), + } + for index in range(1, 7) + ] + edges = [ + { + "parent_post_id": f"post-{index}", + "child_post_id": f"post-{index + 1}", + "fused_score": 0.4, + } + for index in range(1, 6) + ] + + class FakeConnection: + async def fetch(self, query: str): + return edges if "post_lineage_edge" in query else posts + + merged = asyncio.run( + lineage_graphs_for_posts( + FakeConnection(), + lambda row: True, + ["post-1", "post-6"], + node_limit=3, + ) + ) + + node_ids = [node["id"] for node in merged["nodes"]] + assert node_ids[:2] == ["post-1", "post-6"] + assert len(node_ids) == 3 + assert merged["truncated"] is True + assert all( + edge["source"] in node_ids and edge["target"] in node_ids + for edge in merged["edges"] + )