diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index c7f17f666..3a55015f5 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -66,7 +66,7 @@ flowchart LR | `tepp_client.py` | TEPP's published `AnalysisRunRequest` wire contract, pluggable transport | | `rankweave_client.py` | Fail-closed RankWeave ranking port (`weighted_reciprocal_rank_fuse` in-process; never invent a fused score or a theta) | | `reconstruct.py` | The pipeline: group → candidate window → score → fuse → thread | -| `lineage_persistence.py` | Flattens reconstruct trees into `post_lineage_edge` row specs (parent, child, fused_score) | +| `lineage_persistence.py` | Flattens reconstruct trees into `post_lineage_edge` rows plus `post_lineage_edge_signal` channel evidence | | `interval_relation.py` | Allen (1983) closed interval relations for those edges. Each post is a point interval on its observed UTC `created_at` day; mutable ticket dates are not Event Lineage evidence. | | `knowledge_graph.py` | Random-walk-with-restart relevance + per-node adaptive related-node cutoff (Tong et al., 2006) -- pure graph math, no Postgres | | `keyman_extraction.py` | Pluggable LLM extraction of two-sided (our-side/counterparty) person mentions + N:N org affiliations from a post | @@ -221,17 +221,20 @@ 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, 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. +(`{nodes, edges, truncated, reconstruction}`) from persisted `post_lineage_edge` +rows. Each edge includes additive `channel_evidence` from +`post_lineage_edge_signal`. Evidence for an endpoint the account cannot +see is omitted. 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` computed from the 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 +every `source_post` and atomically rewrites edges, channel signals, and +Allen interval relations. Reconstruct grouping is stored on the post as `thread_group_key` / `secondary_grouping_key` (not derived from process unit or voc type). diff --git a/CHANGELOG.d/2.14.0-event-lineage-channel-evidence.md b/CHANGELOG.d/2.14.0-event-lineage-channel-evidence.md new file mode 100644 index 000000000..dc8e9217c --- /dev/null +++ b/CHANGELOG.d/2.14.0-event-lineage-channel-evidence.md @@ -0,0 +1,20 @@ +# 2.14.0 — Event Lineage channel evidence + +Live Event Lineage now persists and explains the independent signals that +produced each reconstructed connection. + +- `post_lineage_edge_signal` stores per-channel score, the normalized + active weight actually used, and `weight * score` contribution beside + each `post_lineage_edge` row. The optional LLM channel is omitted when + it did not participate; it is never fabricated. +- `event_lineage_rebuild` records reconstruction version, generated-at + time, and the active weight profile so a later rebuild cannot silently + rewrite historic evidence. +- Administrator rebuilds release their read connection before CPU and + orchestrator work, then acquire one short transaction for the atomic + live-graph replacement. +- `GET /api/lineage` returns additive `channel_evidence` on each visible + edge. ABAC never reveals evidence for an invisible endpoint. +- The Event Lineage DAG discloses exact values with keyboard and screen-reader + access, labels the relation as inferred rather than causal, and keeps + the same values in print. diff --git a/CHANGELOG.md b/CHANGELOG.md index 8c2856dc5..8cf0eb0f8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,11 @@ All notable changes to this project are documented here. Format follows ### Added +- Event Lineage now persists each reconstructed connection's independent + channel scores, the normalized weights actually used, and their + contributions. The Event Lineage DAG discloses those exact values as inferred + evidence, not a causal claim, and omits the LLM channel when it did + not participate. - The repository-case public ontology namespace `https://contextualwisdomlab.github.io/LineageWeave/ontology#` is canonical (ADR 0207, superseding ADR 0157, resolving issue #372); the lowercase form @@ -104,6 +109,10 @@ All notable changes to this project are documented here. Format follows ### Fixed +- Full-corpus Event Lineage rebuilds now count candidate pairs before provider + work and omit the optional LLM channel above the 5,000-pair ADR budget, + preventing millions of synchronous orchestrator calls while retaining one + auditable weight profile for the entire rebuild. - Structure adjudication now rejects malformed or duplicate unit indexes before calling the orchestrator. - Event Lineage's DAG no longer leaves a linear (no-branch) reconstruct diff --git a/backend/app/analysis_run_worker.py b/backend/app/analysis_run_worker.py index 5a6109b16..dd63fe538 100644 --- a/backend/app/analysis_run_worker.py +++ b/backend/app/analysis_run_worker.py @@ -9,18 +9,17 @@ import asyncio import logging +from uuid import UUID import asyncpg import redis.asyncio as redis -from uuid import UUID - -from lineageweave.adjudication_client import AdjudicationClient -from lineageweave.observability import traced -from lineageweave.tepp_client import TeppClient from backend.app.analysis_run_ingestion import AnalysisRunCreateError from backend.app.analysis_run_outbox import OUTBOX_STREAM_KEY from backend.app.analysis_run_start import deliver_queued_analysis_run +from lineageweave.adjudication_client import AdjudicationClient +from lineageweave.observability import traced +from lineageweave.tepp_client import TeppClient _BROKER_RECOVERY_DELAY_SECONDS = 1.0 _worker_logger = logging.getLogger(__name__) @@ -77,17 +76,17 @@ async def consume_analysis_run_stream_once( # transaction rolls back, the durable outbox row stays # available, and an explicit HTTP start retries the run # once the operator resolves the named next action. - try: - async with pool.acquire() as conn: - owner = await conn.fetchrow( - """ - select requested_by_account_id - from analysis_run - where analysis_run_id = $1::uuid - """, - analysis_run_id, - ) - if owner is not None: + async with pool.acquire() as conn: + owner = await conn.fetchrow( + """ + select requested_by_account_id + from analysis_run + where analysis_run_id = $1::uuid + """, + analysis_run_id, + ) + if owner is not None: + try: await deliver_queued_analysis_run( pool, database_url=database_url, @@ -98,13 +97,19 @@ async def consume_analysis_run_stream_once( adjudication_client=adjudication_client, valkey_stream_entry_id=str(entry_id), ) - except AnalysisRunCreateError as exc: - _worker_logger.warning( - "analysis-run %s delivery refused (%s): %s", - analysis_run_id, - exc.status_code, - exc.detail, - ) + except AnalysisRunCreateError as exc: + _worker_logger.warning( + "analysis-run %s delivery refused (%s): %s", + analysis_run_id, + exc.status_code, + exc.detail, + ) + except Exception as exc: + _worker_logger.warning( + "analysis-run %s delivery failed (error_type=%s)", + analysis_run_id, + type(exc).__name__, + ) last_id = str(entry_id) return last_id diff --git a/backend/app/lineage_ingestion.py b/backend/app/lineage_ingestion.py index a176db247..ee240cd34 100644 --- a/backend/app/lineage_ingestion.py +++ b/backend/app/lineage_ingestion.py @@ -10,8 +10,10 @@ from __future__ import annotations +import asyncio import math import re +from collections import defaultdict from collections.abc import Mapping from datetime import datetime from typing import Any @@ -19,14 +21,23 @@ import asyncpg from backend.app.post_eligibility import SOURCE_POST_ELIGIBILITY_SQL +from lineageweave.adjudication_client import AdjudicationClient from lineageweave.interval_relation import ( INTERVAL_RELATION_LABELS, allen_interval_relation, interval_from_post, interval_relation_from_current, ) -from lineageweave.lineage_persistence import lineage_edge_specs +from lineageweave.lineage_persistence import ( + LOOKUP_CODE_TO_SIGNAL, + lineage_edge_specs, + lineage_rebuild_spec, + rank_channel_evidence, +) from lineageweave.models import Edge, Record +from lineageweave.reconstruct import DEFAULT_CANDIDATE_WINDOW + +MAXIMUM_LIVE_LLM_PAIR_EVALUATIONS = 5_000 ISOLATION_NO_COMPARISON_GROUP = "no_comparison_group" ISOLATION_COMPARISON_CANDIDATES_AVAILABLE = "comparison_candidates_available" @@ -37,6 +48,14 @@ ) +def estimated_weight_channels(llm: AdjudicationClient | None) -> set[str]: + """Return the channels that one live reconstruction can actually use.""" + channels = {"temporal", "secondary_key", "text"} + if getattr(llm, "available", False): + channels.add("llm") + return channels + + def _occurred_at(value: datetime) -> datetime: """Reconstruct expects naive datetimes; asyncpg returns timestamptz.""" return value.replace(tzinfo=None) if value.tzinfo is not None else value @@ -125,34 +144,77 @@ def interval_relation_code_for_edge( async def persist_lineage_edges( conn: asyncpg.Connection, edges: list[Edge], - points_by_post_id: Mapping[str, Mapping[str, Any]], + weights: dict[str, float], + points_by_post_id: Mapping[str, Mapping[str, Any]] | None = None, ) -> None: - """Replace ``post_lineage_edge`` with ``edges`` (reconstruct is source of truth).""" + """Replace live Event Lineage with edges, channel evidence, and intervals. + + Reconstruct is the source of truth. The delete is cascaded onto + ``post_lineage_edge_signal`` so a rebuild cannot leave orphan or + stale signal rows. Rebuild metadata is replaced in the same + connection so version, weights, and generated-at stay aligned with + the new graph. + """ + points = points_by_post_id or {} missing_point_ids = { post_id for edge in edges for post_id in (edge.parent_id, edge.child_id) - if post_id not in points_by_post_id + if post_id not in points } if missing_point_ids: raise ValueError( "missing observed interval point for post ids: " + ", ".join(sorted(missing_point_ids)) ) + spec = lineage_rebuild_spec(edges, weights=weights) await conn.execute("delete from post_lineage_edge") - for edge in edges: - relation_code = interval_relation_code_for_edge( - points_by_post_id[edge.parent_id], points_by_post_id[edge.child_id] - ) - await conn.execute( - "insert into post_lineage_edge " - "(parent_post_id, child_post_id, fused_score, interval_relation_code) " - "values ($1::uuid, $2::uuid, $3, $4)", - edge.parent_id, - edge.child_id, - edge.fused_score, - relation_code, - ) + await conn.execute("delete from event_lineage_rebuild") + await conn.execute( + "insert into event_lineage_rebuild " + "(rebuild_lock, reconstruction_version, generated_at, min_fused_score, candidate_window) " + "values (true, $1, now(), $2, $3)", + spec.reconstruction_version, + spec.min_fused_score, + spec.candidate_window, + ) + await conn.executemany( + "insert into event_lineage_rebuild_channel " + "(rebuild_lock, signal_code, signal_weight) values (true, $1, $2)", + spec.channel_weights, + ) + await conn.executemany( + "insert into post_lineage_edge " + "(parent_post_id, child_post_id, fused_score, interval_relation_code) " + "values ($1::uuid, $2::uuid, $3, $4)", + [ + ( + edge.parent_id, + edge.child_id, + edge.fused_score, + interval_relation_code_for_edge( + points[edge.parent_id], points[edge.child_id] + ), + ) + for edge in edges + ], + ) + await conn.executemany( + "insert into post_lineage_edge_signal " + "(parent_post_id, child_post_id, signal_code, signal_score, signal_weight, signal_contribution) " + "values ($1::uuid, $2::uuid, $3, $4, $5, $6)", + [ + ( + row["parent_post_id"], + row["child_post_id"], + row["signal_code"], + row["signal_score"], + row["signal_weight"], + row["signal_contribution"], + ) + for row in spec.signal_rows + ], + ) async def load_estimated_channel_weights( @@ -334,7 +396,7 @@ class ChannelWeightsNotEstimated(RuntimeError): only (ADR 0200 point 1): estimated by fast-mlsirm, provenance-gated, never hand-picked constants. No hand-picked default exists anywhere -- the library demo estimates its weights from its declared design, - and unit tests inject synthetic weights explicitly. + and tests use provenance-bearing estimates. """ def __init__(self, active_channels: set[str]) -> None: @@ -347,27 +409,87 @@ def __init__(self, active_channels: set[str]) -> None: self.active_channels = active_channels -async def rebuild_lineage(conn: asyncpg.Connection) -> list[Edge]: - """Reconstruct lineage for every ``source_post`` and persist the edges. - - Raises :class:`ChannelWeightsNotEstimated` when no activated - estimate matches this path's active channels -- run - ``scripts/estimate_channel_weights.py`` first (ADR 0200 point 1). - """ +async def _load_lineage_records(conn: asyncpg.Connection) -> list[Record]: + """Load the eligible source snapshot used by one reconstruction.""" rows = await conn.fetch( "select post_id, post_title, voc_type_code, created_at, corporate_entity_id, " "process_unit_id, thread_group_key, secondary_grouping_key " f"from source_post where {SOURCE_POST_ELIGIBILITY_SQL.format(alias='source_post')}" ) - # No adjudication client is wired on this path, so the active channel - # set is the three deterministic channels (reconstruct drops llm when - # unavailable rather than faking it). - active_channels = {"temporal", "secondary_key", "text"} + return records_from_source_posts(rows) + + +def _budgeted_llm( + records: list[Record], llm: AdjudicationClient | None +) -> AdjudicationClient | None: + """Drop adjudication before weight lookup when the pair budget is exceeded.""" + pair_count = 0 + records_per_group: defaultdict[str, int] = defaultdict(int) + for record in records: + pair_count += min(records_per_group[record.group_key], DEFAULT_CANDIDATE_WINDOW) + if pair_count > MAXIMUM_LIVE_LLM_PAIR_EVALUATIONS: + return None + records_per_group[record.group_key] += 1 + return llm + + +async def _reconstruct_lineage_records( + records: list[Record], + llm: AdjudicationClient | None, + weights: dict[str, float], +) -> list[Edge]: + """Run the CPU/provider reconstruction without blocking the event loop.""" + llm = _budgeted_llm(records, llm) + return await asyncio.to_thread(lineage_edge_specs, records, llm=llm, weights=weights) + + +async def rebuild_lineage( + conn: asyncpg.Connection, + *, + llm: AdjudicationClient | None = None, +) -> list[Edge]: + """Reconstruct lineage for every ``source_post`` and persist the edges. + + The adjudication channel is dropped before weight lookup when the exact + candidate-pair work exceeds the ADR 0172 budget. The resulting active + channel set must have a provenance-gated fast-mlsirm estimate; there is no + hand-picked fallback (ADR 0200). + """ + records = await _load_lineage_records(conn) + llm = _budgeted_llm(records, llm) + active_channels = estimated_weight_channels(llm) weights = await load_estimated_channel_weights(conn, active_channels) if weights is None: raise ChannelWeightsNotEstimated(active_channels) - edges = lineage_edge_specs(records_from_source_posts(rows), weights=weights) - await persist_lineage_edges(conn, edges, {str(row["post_id"]): row for row in rows}) + edges = await _reconstruct_lineage_records(records, llm, weights) + points = {record.record_id: {"created_at": record.occurred_at} for record in records} + async with conn.transaction(): + await persist_lineage_edges(conn, edges, weights, points) + return edges + + +async def rebuild_lineage_from_pool( + pool: asyncpg.Pool, + *, + llm: AdjudicationClient | None = None, +) -> list[Edge]: + """Reconstruct without holding a pooled connection during provider work. + + The source snapshot is read and released first. Only the replacement + writes run in a transaction, preserving ADR 0172 atomicity without an + idle-in-transaction connection during CPU or orchestrator calls. + """ + async with pool.acquire() as conn: + records = await _load_lineage_records(conn) + llm = _budgeted_llm(records, llm) + active_channels = estimated_weight_channels(llm) + weights = await load_estimated_channel_weights(conn, active_channels) + if weights is None: + raise ChannelWeightsNotEstimated(active_channels) + edges = await _reconstruct_lineage_records(records, llm, weights) + points = {record.record_id: {"created_at": record.occurred_at} for record in records} + async with pool.acquire() as conn, conn.transaction(): + await persist_lineage_edges(conn, edges, weights, points) return edges @@ -420,8 +542,7 @@ def _connected_visible_component( ) -> 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. + Invisible posts cannot bridge visible posts across the ABAC boundary. """ if focus_id not in allowed: return set() @@ -432,17 +553,41 @@ def _connected_visible_component( 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 + frontier.extend((neighbors.get(current_id, set()) & allowed) - component_ids) + return component_ids -def _lineage_graph_payload(visible, edge_rows, truncated: bool) -> dict[str, Any]: +async def _lineage_graph_payload( + conn: asyncpg.Connection, visible, edge_rows, truncated: bool +) -> dict[str, Any]: + """Build a channel-auditable graph from already authorized rows.""" visible_ids = {str(row["post_id"]) for row in visible} visible_edges = [ row for row in edge_rows if str(row["parent_post_id"]) in visible_ids and str(row["child_post_id"]) in visible_ids ] + visible_id_list = sorted(visible_ids) + signal_rows = await conn.fetch( + "select parent_post_id, child_post_id, signal_code, signal_score, " + "signal_weight, signal_contribution from post_lineage_edge_signal " + "where parent_post_id = any($1::uuid[]) " + "and child_post_id = any($2::uuid[])", + visible_id_list, + visible_id_list, + ) + rebuild_rows = await conn.fetch( + "select reconstruction_version, generated_at, min_fused_score, candidate_window " + "from event_lineage_rebuild" + ) + weight_rows = await conn.fetch( + "select channel.signal_code, channel.signal_weight " + "from event_lineage_rebuild_channel as channel " + "join common_lookup_value as lookup " + "on lookup.lookup_code = channel.signal_code " + "where channel.rebuild_lock = true " + "order by lookup.display_order, channel.signal_code" + ) children_of: dict[str, list[str]] = {} for row in visible_edges: children_of.setdefault(str(row["parent_post_id"]), []).append(str(row["child_post_id"])) @@ -460,16 +605,51 @@ def _lineage_graph_payload(visible, edge_rows, truncated: bool) -> dict[str, Any "is_branch_point": len(children_of.get(post_id, [])) >= 2, } ) + + signals_by_edge: dict[tuple[str, str], list[dict[str, object]]] = defaultdict(list) + for row in signal_rows: + parent_id = str(row["parent_post_id"]) + child_id = str(row["child_post_id"]) + if parent_id not in visible_ids or child_id not in visible_ids: + continue + payload = dict(row) + payload["channel_name"] = LOOKUP_CODE_TO_SIGNAL.get(str(row["signal_code"]), str(row["signal_code"])) + signals_by_edge[(parent_id, child_id)].append(payload) + edges = [ { "source": str(row["parent_post_id"]), "target": str(row["child_post_id"]), "fused_score": float(row["fused_score"]), + "channel_evidence": rank_channel_evidence( + signals_by_edge[(str(row["parent_post_id"]), str(row["child_post_id"]))] + ), **_interval_payload(row), } for row in visible_edges ] - return {"nodes": nodes, "edges": edges, "truncated": truncated} + reconstruction = None + if rebuild_rows: + rebuild = rebuild_rows[0] + reconstruction = { + "reconstruction_version": rebuild["reconstruction_version"], + "generated_at": rebuild["generated_at"].isoformat(), + "min_fused_score": float(rebuild["min_fused_score"]), + "candidate_window": int(rebuild["candidate_window"]), + "active_weights": [ + { + "signal_code": LOOKUP_CODE_TO_SIGNAL.get(str(row["signal_code"]), str(row["signal_code"])), + "signal_weight": float(row["signal_weight"]), + } + for row in weight_rows + ], + } + return { + "nodes": nodes, + "edges": edges, + "truncated": truncated, + "reconstruction": reconstruction, + } async def visible_lineage_graph( @@ -501,12 +681,12 @@ async def visible_lineage_graph( 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 + if include_isolated or len(component_ids) > 1 else [] ) truncated = False - payload = _lineage_graph_payload(visible, edge_rows, truncated) + payload = await _lineage_graph_payload(conn, visible, edge_rows, truncated) payload["isolation_reason"] = focused_isolation_reason( focus_post_id, visible_all, len(visible) ) @@ -560,7 +740,12 @@ async def lineage_graphs_for_posts( """ unique_ids = list(dict.fromkeys(str(post_id) for post_id in post_ids)) if not unique_ids: - return {"nodes": [], "edges": [], "truncated": False} + return { + "nodes": [], + "edges": [], + "truncated": False, + "reconstruction": None, + } 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} @@ -581,4 +766,4 @@ async def lineage_graphs_for_posts( 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) + return await _lineage_graph_payload(conn, visible, edge_rows, truncated) diff --git a/backend/app/main.py b/backend/app/main.py index 535dcff8d..1cd132363 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -34,68 +34,24 @@ from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel -from lineageweave.adjudication_client import ( - ContextualOrchestratorAdjudicationClient, - NullAdjudicationClient, -) -from lineageweave.commitment_extraction import ( - ContextualOrchestratorCommitmentExtractionClient, - NullCommitmentExtractionClient, -) -from lineageweave.entity_relationship_classification import ( - ContextualOrchestratorEntityRelationshipClient, - NullEntityRelationshipClient, -) -from lineageweave.image_content import orchestrator_vision_client -from lineageweave.embedding_client import orchestrator_embedding_client -from lineageweave.llm_context import build_post_llm_metadata, use_llm_metadata -from lineageweave.observability import ( - configure_telemetry, - record_server_failure, - shutdown_telemetry, - traced, -) -from lineageweave.corporate_hierarchy_inference import ( - ContextualOrchestratorHierarchyInferenceClient, - NullCorporateHierarchyInferenceClient, -) -from lineageweave.keyman_extraction import ( - COUNTERPARTY, - ContextualOrchestratorKeymanExtractionClient, - NullKeymanExtractionClient, -) -from lineageweave.customer_hint_resolution import ( - ContextualOrchestratorCustomerHintResolutionClient, - NullCustomerHintResolutionClient, -) -from lineageweave.organization_name_resolution import ( - ContextualOrchestratorOrganizationNameResolutionClient, - NullOrganizationNameResolutionClient, +from backend.app.activity_stream import ( + create_valkey_client, + get_valkey, + publish_activity_event, + read_activity_events, + ticket_created_summary, + ticket_status_changed_summary, ) -from lineageweave.post_chat import ( - ContextualOrchestratorPostChatClient, - NullPostChatClient, - cited_post_summaries, +from backend.app.affiliate_tree_ingestion import ( + fetch_affiliate_forest, + fetch_voc_evidence, ) -from lineageweave.post_content_normalization import normalize_post_body -from lineageweave.post_evaluation import ( - ContextualOrchestratorPostEvaluationClient, - NullPostEvaluationClient, - RUBRIC_VERSION, -) -from lineageweave.post_structure import ContextualOrchestratorPostStructureClient, NullPostStructureClient -from lineageweave.post_summary import ContextualOrchestratorPostSummaryClient, NullPostSummaryClient -from lineageweave.relation_verification import NullRelationVerificationClient, SearxngRelationVerificationClient -from lineageweave.semantic_hints import customer_hint_trust, format_semantic_hints from lineageweave.similar_voc import ContextualOrchestratorSimilarVocAnalysisClient -from lineageweave.ontology import LW -from lineageweave.rankweave_client import build_rankweave_client from lineageweave.naruon_calendar_workspace import ( build_workspace_naruon_client, default_calendar_window, load_observed_calendar_events, ) - from backend.app.analysis_run_ingestion import ( AnalysisRunCreateError, create_pending_analysis_run, @@ -110,49 +66,24 @@ enqueue_pending_analysis_run, ) from backend.app.analysis_run_worker import run_analysis_run_worker -from backend.app.post_content_queue import ( - ensure_post_content_job, - post_content_api_status, - post_content_is_complete, - publish_post_content_event, -) -from backend.app.global_ask_queue import ( - enqueue_global_ask_job, - run_global_ask_worker, -) -from backend.app.post_content_worker import run_post_content_worker -from backend.app.source_post_revision import fetch_known_at_revision, parse_as_of_clock -from backend.app.activity_stream import ( - create_valkey_client, - get_valkey, - publish_activity_event, - read_activity_events, - ticket_created_summary, - ticket_status_changed_summary, -) -from backend.app.affiliate_tree_ingestion import fetch_affiliate_forest, fetch_voc_evidence from backend.app.auth import CurrentAccount, get_current_account from backend.app.config import load_settings from backend.app.customer_hint_ingestion import resolve_customer_hint from backend.app.db import create_pool, get_pool +from backend.app.demo_scope import ( + fetch_demo_corporate_entity_ids, + has_real_source_context, +) from backend.app.entity_relationship_ingestion import ( fetch_post_counterparties, fetch_relationship_network, ingest_post_entity_relationships, ) from backend.app.five_w1h_ingestion import load_five_w1h_slots -from backend.app.post_evaluation_ingestion import fetch_post_evaluation, ingest_post_evaluation -from backend.app.ranking_ingestion import load_visible_ranking_posts -from backend.app.report_ingestion import ( - GROUPING_KINDS, - fetch_period_comparison, - fetch_period_reports, - iso_week_period, - list_period_report_summaries, - parse_period_code, - rebuild_period_reports, +from backend.app.global_ask_queue import ( + enqueue_global_ask_job, + run_global_ask_worker, ) -from backend.app.relation_verification_ingestion import verify_post_relations from backend.app.issue_ticket_ingestion import ( create_ticket, fetch_ticket_post_id, @@ -168,8 +99,8 @@ fetch_person_role_history, fetch_post_keymen, labels_for_codes, - person_exists, persist_edges_for_post, + person_exists, related_for_entity, related_for_person, related_for_team, @@ -181,7 +112,8 @@ from backend.app.lineage_ingestion import ( ChannelWeightsNotEstimated, interval_relations_for_post, - rebuild_lineage, + lineage_graphs_for_posts, + rebuild_lineage_from_pool, visible_lineage_graph, ) from backend.app.ontology_neighborhood_ingestion import ( @@ -191,15 +123,6 @@ parse_allowed_property_query, visible_ontology_neighborhood, ) -from lineageweave.ontology_neighborhood import ( - DEFAULT_MAXIMUM_DEPTH, - DEFAULT_MAXIMUM_EDGES, - DEFAULT_MAXIMUM_NODES, - HARD_MAXIMUM_DEPTH, - HARD_MAXIMUM_EDGES, - HARD_MAXIMUM_NODES, - OntologyNeighborhoodError, -) from backend.app.post_chat_ingestion import ( fetch_persisted_chat, fetch_persisted_chats, @@ -207,17 +130,113 @@ gather_chat_sources, persist_post_chat, ) +from backend.app.post_content_queue import ( + ensure_post_content_job, + post_content_api_status, + post_content_is_complete, + publish_post_content_event, +) +from backend.app.post_content_worker import run_post_content_worker +from backend.app.post_eligibility import SOURCE_POST_ELIGIBILITY_SQL +from backend.app.post_evaluation_ingestion import ( + fetch_post_evaluation, + ingest_post_evaluation, +) from backend.app.post_summary_ingestion import ( fetch_persisted_summary, persist_post_summary, require_summary_source_body, ) -from backend.app.post_eligibility import SOURCE_POST_ELIGIBILITY_SQL -from backend.app.demo_scope import ( - fetch_demo_corporate_entity_ids, - has_real_source_context, +from backend.app.ranking_ingestion import load_visible_ranking_posts +from backend.app.relation_verification_ingestion import verify_post_relations +from backend.app.report_ingestion import ( + GROUPING_KINDS, + fetch_period_comparison, + fetch_period_reports, + iso_week_period, + list_period_report_summaries, + parse_period_code, + rebuild_period_reports, +) +from backend.app.source_post_revision import fetch_known_at_revision, parse_as_of_clock +from lineageweave.adjudication_client import ( + ContextualOrchestratorAdjudicationClient, + NullAdjudicationClient, +) +from lineageweave.caldav_client import ( + CALDAV_UNAVAILABLE_NEXT_ACTION, + build_caldav_client, +) +from lineageweave.commitment_extraction import ( + ContextualOrchestratorCommitmentExtractionClient, + NullCommitmentExtractionClient, +) +from lineageweave.corporate_hierarchy_inference import ( + ContextualOrchestratorHierarchyInferenceClient, + NullCorporateHierarchyInferenceClient, +) +from lineageweave.customer_hint_resolution import ( + ContextualOrchestratorCustomerHintResolutionClient, + NullCustomerHintResolutionClient, +) +from lineageweave.embedding_client import orchestrator_embedding_client +from lineageweave.entity_relationship_classification import ( + ContextualOrchestratorEntityRelationshipClient, + NullEntityRelationshipClient, ) from lineageweave.http_client import HttpClientError +from lineageweave.image_content import orchestrator_vision_client +from lineageweave.keyman_extraction import ( + COUNTERPARTY, + ContextualOrchestratorKeymanExtractionClient, + NullKeymanExtractionClient, +) +from lineageweave.llm_context import build_post_llm_metadata, use_llm_metadata +from lineageweave.observability import ( + configure_telemetry, + record_server_failure, + shutdown_telemetry, + traced, +) +from lineageweave.ontology import LW +from lineageweave.ontology_neighborhood import ( + DEFAULT_MAXIMUM_DEPTH, + DEFAULT_MAXIMUM_EDGES, + DEFAULT_MAXIMUM_NODES, + HARD_MAXIMUM_DEPTH, + HARD_MAXIMUM_EDGES, + HARD_MAXIMUM_NODES, + OntologyNeighborhoodError, +) +from lineageweave.organization_name_resolution import ( + ContextualOrchestratorOrganizationNameResolutionClient, + NullOrganizationNameResolutionClient, +) +from lineageweave.post_chat import ( + ContextualOrchestratorPostChatClient, + NullPostChatClient, + cited_post_summaries, +) +from lineageweave.post_content_normalization import normalize_post_body +from lineageweave.post_evaluation import ( + RUBRIC_VERSION, + ContextualOrchestratorPostEvaluationClient, + NullPostEvaluationClient, +) +from lineageweave.post_structure import ( + ContextualOrchestratorPostStructureClient, + NullPostStructureClient, +) +from lineageweave.post_summary import ( + ContextualOrchestratorPostSummaryClient, + NullPostSummaryClient, +) +from lineageweave.rankweave_client import build_rankweave_client +from lineageweave.relation_verification import ( + NullRelationVerificationClient, + SearxngRelationVerificationClient, +) +from lineageweave.semantic_hints import customer_hint_trust, format_semantic_hints _POST_READ = "post_read" _POST_ADMIN = "post_admin" @@ -1252,16 +1271,24 @@ async def rebuild_lineage_graph( post_admin only: this is a corpus-wide write. Reads stay ABAC-gated. """ _require_post_admin(account) - async with pool.acquire() as conn: - async with conn.transaction(): - try: - edges = await rebuild_lineage(conn) - except ChannelWeightsNotEstimated: - raise HTTPException( - status.HTTP_503_SERVICE_UNAVAILABLE, - "Channel weights are not estimated yet. Run " - "scripts/estimate_channel_weights.py, then rebuild again.", - ) + try: + edges = await rebuild_lineage_from_pool(pool, llm=_adjudication_client()) + except ChannelWeightsNotEstimated as exc: + raise HTTPException( + status.HTTP_503_SERVICE_UNAVAILABLE, + "Channel weights are not estimated yet. Run " + "scripts/estimate_channel_weights.py, then rebuild again.", + ) from exc + except (HttpClientError, OSError) as exc: + # This can issue up to MAXIMUM_LIVE_LLM_PAIR_EVALUATIONS sequential + # adjudication calls across the whole corpus (lineage_ingestion.py); + # a transient orchestrator hiccup on any one of them must not + # discard the rest of the reconstruction as a raw 500 -- same + # discipline as this file's other orchestrator call sites. + raise HTTPException( + status.HTTP_503_SERVICE_UNAVAILABLE, + "Lineage rebuild is unavailable: the orchestrator did not respond", + ) from exc return {"edge_count": len(edges)} diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index bb9ee0df5..df1744dd2 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -196,6 +196,11 @@ / "migrations" / "0169_report_leftover_map_axis.sql" ) +_CHANNEL_EVIDENCE_MIGRATION = ( + Path(__file__).resolve().parents[2] + / "migrations" + / "0174_post_lineage_edge_signal.sql" +) _LEFTOVER_MAP_COVERAGE_MIGRATION = ( Path(__file__).resolve().parents[2] / "migrations" @@ -325,6 +330,8 @@ def seeded_db(demo_analyst_token): cur.execute(_PROJECT_BOUND_ACTION_MIGRATION.read_text()) cur.execute(_PROJECT_BOUND_EVENT_MIGRATION.read_text()) cur.execute(_TENANT_SETTINGS_MIGRATION.read_text()) + # Compose replays this gate on restart; the second apply is the contract. + cur.execute(_TENANT_SETTINGS_MIGRATION.read_text()) cur.execute(_TOPIC_LINEAGE_KIND_MIGRATION.read_text()) cur.execute(_TOPIC_LINEAGE_RESULT_MIGRATION.read_text()) cur.execute(_TOPIC_LINEAGE_VALIDATE_MIGRATION.read_text()) @@ -360,6 +367,7 @@ def seeded_db(demo_analyst_token): cur.execute(_GLOBAL_ASK_SCOPE_MIGRATION.read_text()) cur.execute(_EVENT_OCCURRED_AT_MIGRATION.read_text()) cur.execute(_LEFTOVER_MAP_AXIS_MIGRATION.read_text()) + cur.execute(_CHANNEL_EVIDENCE_MIGRATION.read_text()) cur.execute(_LEFTOVER_MAP_UNEXPLAINED_MIGRATION.read_text()) cur.execute(_LEFTOVER_MAP_CROSS_SHARE_MIGRATION.read_text()) cur.execute(_LEFTOVER_MAP_RECONSTRUCTION_MIGRATION.read_text()) @@ -4381,6 +4389,40 @@ def test_rebuild_lineage_requires_post_admin(client, demo_analyst_token) -> None assert response.status_code == 403 +def test_rebuild_lineage_reports_503_on_orchestrator_failure( + monkeypatch, client, demo_analyst_token, seeded_db +) -> None: + """A transient orchestrator failure mid-rebuild must degrade to a clean + 503, not discard the whole corpus reconstruction as a raw 500 (same + discipline as this file's other orchestrator call sites). + """ + admin_conn = psycopg2.connect(seeded_db["dsn"]) + admin_conn.autocommit = True + try: + with admin_conn.cursor() as cur: + cur.execute( + "insert into common_lookup_value (lookup_category, lookup_code, lookup_label) " + "values ('permission', 'post_admin', 'Administer posts') on conflict (lookup_code) do nothing" + ) + cur.execute("select access_role_id from account_role_assignment limit 1") + role_id = cur.fetchone()[0] + cur.execute( + "insert into role_permission (access_role_id, permission_code) values (%s, 'post_admin') " + "on conflict do nothing", + (role_id,), + ) + finally: + admin_conn.close() + + async def _raise(pool, *, llm): + raise HttpClientError("orchestrator hiccup") + + monkeypatch.setattr("backend.app.main.rebuild_lineage_from_pool", _raise) + + response = client.post("/api/lineage/rebuild", headers={"Authorization": f"Bearer {demo_analyst_token}"}) + assert response.status_code == 503 + + def test_rebuild_lineage_recovers_the_a100_fork(client, demo_analyst_token, seeded_db) -> None: """Rebuild on the same A-100+B-200 rows seed writes (grouping keys + occurred_at), not a hand-picked A-100-only insert that hides mapping bugs. diff --git a/docs/adr/0172-event-lineage-channel-evidence.md b/docs/adr/0172-event-lineage-channel-evidence.md new file mode 100644 index 000000000..717fe61ec --- /dev/null +++ b/docs/adr/0172-event-lineage-channel-evidence.md @@ -0,0 +1,103 @@ +# ADR 0172: Persist and explain Event Lineage channel evidence + +**Status:** Accepted +**Date:** 2026-08-21 +**Issue:** [#274](https://github.com/ContextualWisdomLab/LineageWeave/issues/274) + +## Context + +`reconstruct()` already computes the winning edge's per-channel scores +(`temporal`, `secondary_key`, `text`, optional `llm`) and RankWeave +fuses them with a weighted convex combination. Production persistence +collapsed each edge to `(parent_post_id, child_post_id, fused_score)`, +so `/api/lineage` and the Event Lineage DAG exposed only the fused score. + +A reader could see that two posts were linked but could not answer which +independent signals supported the edge, whether the optional LLM channel +participated, which signal dominated, or how to audit a later +reconstruction after model or weight changes. ADR 0064 already treats +every accepted edge as uncertainty-bearing evidence, not a proven +business fact; that contract was not visible in PostgreSQL or the UI. + +This is distinct from typed Knowledge Graph path repair. Event Lineage +explains reconstructed post-to-post links. PostgreSQL remains +authoritative; PROV-O/RDF export is a projection. + +## Decision + +1. Persist a child table `post_lineage_edge_signal` with a composite + foreign key to `post_lineage_edge` (`ON DELETE CASCADE`), one row per + edge and active signal, and exact `numeric(8,6)` score, weight, and + contribution. No JSONB for billable or auditable numeric facts. +2. Controlled lookup values are globally unique: + `lineage_signal_temporal`, `lineage_signal_secondary_key`, + `lineage_signal_text`, `lineage_signal_llm`. The LLM row is omitted + when the adjudication client is unavailable; it is never fabricated. +3. Weights are the normalized active weights actually used + (`reconstruct.active_weights`). Contribution is `weight * score` and + must reconcile with `fused_score` within + `CHANNEL_EVIDENCE_TOLERANCE` (`1e-6`). +4. Live Event Lineage is replaced atomically. A singleton + `event_lineage_rebuild` stores reconstruction version, generated-at + time, minimum fused score, and candidate window; + `event_lineage_rebuild_channel` stores the active weight profile. + Analysis-run reconstruction (`analysis_run_lineage_edge`) stays a + separate immutable run-scoped table. + The administrator-triggered live rebuild and PostgreSQL import pass the + configured contextual-orchestrator adjudication client through the same + reconstruction boundary only when the exact candidate-pair count is at + most 5,000. Larger snapshots drop the LLM channel before any provider call + and renormalize the remaining weights. This is an operational work bound, + not a model-quality or provider-ranking heuristic. One rebuild never mixes + LLM and non-LLM weight profiles across edges. +5. `GET /api/lineage` returns an additive `channel_evidence` collection + on each visible edge (`signal_code`, `signal_label`, `score`, + `weight`, `contribution`, `rank`) ordered by contribution, then + controlled signal order. The rebuild profile is returned in the same + controlled order using `common_lookup_value.display_order`. ABAC never + reveals evidence for an invisible endpoint. +6. The Event Lineage DAG provides an accessible edge-detail disclosure (not + hover-only), labels the relation as inferred rather than causal, and + states when no LLM channel participated only when at least one + recorded channel exists. Print/export uses the same values. + +## Consequences + +- Readers can inspect why a connection was selected and distinguish + inference from source evidence. +- A later rebuild rewrites live Event Lineage as a whole; historic + meaning is not silently mutated in place. +- Completeness is lower when the LLM channel is unavailable, matching + ADR 0064: missing channels are dropped and weights renormalize. + +## References + +Allen, J. F. (1983). Maintaining knowledge about temporal intervals. +*Communications of the ACM, 26*(11), 832–843. +https://doi.org/10.1145/182.358434 + +Cormack, G. V., Clarke, C. L. A., & Buettcher, S. (2009). Reciprocal +rank fusion outperforms Condorcet and individual rank learning methods. +In *Proceedings of the 32nd International ACM SIGIR Conference on +Research and Development in Information Retrieval* (pp. 758–759). ACM. +https://doi.org/10.1145/1571941.1572114 + +Hearst, M. A. (1997). TextTiling: Segmenting text into multi-paragraph +subtopic passages. *Computational Linguistics, 23*(1), 33–64. + +Jeon, J.-J., Kim, I., Vanli, N. D., & Choi, T. (2021). Logistic +structured interaction model for binary item response (arXiv:2007.08719). +https://arxiv.org/abs/2007.08719 + +Lebo, T., Sahoo, S., McGuinness, D., Belhajjame, K., Cheney, J., +Corsar, D., Garijo, D., Soiland-Reyes, S., Zednik, S., & Zhao, J. +(2013). *PROV-O: The PROV ontology* (W3C Recommendation). World Wide +Web Consortium. https://www.w3.org/TR/2013/REC-prov-o-20130430/ + +Tong, H., Faloutsos, C., & Pan, J.-Y. (2006). Fast random walk with +restart and its applications. In *Proceedings of the Sixth International +Conference on Data Mining* (pp. 613–622). IEEE. +https://doi.org/10.1109/ICDM.2006.70 + +ADR 0064 (uncertainty-bearing lineage evidence) +ADR 0024 (RankWeave fusion fail-closed) diff --git a/docs/adr/README.md b/docs/adr/README.md index 0892f8b93..a7704da7e 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -11,7 +11,7 @@ decision from them. |---|---| | [`product-requirements.md`](../product-requirements.md) | Product requirements projection across the ADR set; ADRs remain normative | | [`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), [0169](0169-ask-batched-lineage-graph.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), [0172](0172-event-lineage-channel-evidence.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) | [0207](0207-repository-case-ontology-namespace-canonical.md), [0157](0157-public-ontology-namespace-identity.md) | diff --git a/docs/storybook-inventory.md b/docs/storybook-inventory.md index 85db06359..3716656cd 100644 --- a/docs/storybook-inventory.md +++ b/docs/storybook-inventory.md @@ -11,6 +11,9 @@ operator-facing control you can click before changing product CSS. | `Evidence/OrganizationAliasChip` | Click a cataloged org; the parenthetical is the unique corroborated SKOS companion. | `--color-chip-border`, `--radius-chip`, `OrganizationAliasChip` | | `AnalysisRun/CutoffKnownBody` | Read the cutoff-known sentence, then compare it with the live body below. | `--color-accent-border`, `--space-panel-block`, `--radius-panel`, `CutoffKnownBody` | | `Analysis/LineageEntityPicker` | Choose which corp to reconstruct, then click Request a lineage reconstruction. | `--space-control-gap`, `--size-control-min`, `--radius-control`, `LineageEntityPicker` | +| `Admin/AdminPanel` | Change the tenant brand name, then verify the saved or failed state before leaving settings. | `--surface`, `--border`, `--space-panel-block`, `AdminPanel` | +| `Lineage/LineageDag` | Open a reconstructed connection to read its inferred channel scores and Allen interval relation, or open the current branch node; compare empty, single-branch, grouped/forked, mobile-scroll, ungrouped, and long-title states before changing graph CSS. On narrow viewports, swipe the named viewport or focus it and use arrow keys to inspect the full lineage. | `--color-accent-background`, `--radius-control`, `--surface`, `--border`, `--color-focus-border`, `--size-control-min`, `LineageDag` | +| `Chrome/PopupCloseButton` | Close the evidence panel or post popup. | `--space-close-inset`, `--font-size-close`, `PopupCloseButton` | | `Workspace/WorkspaceCalendar` | Read observed Naruon events, or open a commitment to land on that post. Fail-closed copy stays `이 범위의 일정을 아직 받을 수 없습니다`. | `--color-chip-border`, `WorkspaceCalendar`, `EvidenceStatusMark` | Repeated web objects must use `frontend/src/styles/tokens.css` and a module diff --git a/frontend/src/App.css b/frontend/src/App.css index 22d4ea06b..eeaa61299 100644 --- a/frontend/src/App.css +++ b/frontend/src/App.css @@ -546,6 +546,99 @@ stroke: var(--border); stroke-width: 1.5; fill: none; + cursor: pointer; +} + +.lineage-dag-edge:focus, +.lineage-dag-edge:focus-visible { + outline: 2px solid var(--color-primary, var(--text-h)); + outline-offset: 2px; +} + +.lineage-dag-edge-selected { + stroke: var(--text-h); + stroke-width: 2.5; +} + +.lineage-edge-evidence { + margin-top: 1rem; +} + +.lineage-edge-evidence h3 { + font-size: 1rem; + margin: 0 0 0.4rem; +} + +.lineage-edge-evidence p { + margin: 0 0 0.75rem; +} + +.lineage-rebuild-profile { + display: grid; + gap: 0.35rem; + margin: 0 0 0.75rem; +} + +.lineage-rebuild-profile div { + display: grid; + grid-template-columns: minmax(8rem, 14rem) 1fr; + gap: 0.5rem; +} + +.lineage-rebuild-profile dt { + font-weight: 600; +} + +.lineage-rebuild-profile dd { + margin: 0; +} + +.lineage-edge-evidence-item { + margin: 0 0 0.5rem; + padding: 0.4rem 0.6rem; + border: 1px solid var(--color-border, var(--border)); + border-radius: var(--radius-control, 8px); + background: var(--surface); +} + +.lineage-edge-evidence table { + width: 100%; + border-collapse: collapse; + margin-top: 0.5rem; +} + +.lineage-edge-evidence caption { + text-align: left; + font-weight: 600; + margin-bottom: 0.25rem; +} + +.lineage-edge-evidence th, +.lineage-edge-evidence td { + padding: 0.25rem 0.4rem; + border-bottom: 1px solid var(--color-border, var(--border)); +} + +.lineage-edge-evidence th:nth-child(1), +.lineage-edge-evidence td:nth-child(1), +.lineage-edge-evidence th:nth-child(n + 3), +.lineage-edge-evidence td:nth-child(n + 3) { + text-align: right; +} + +.lineage-edge-evidence th:nth-child(2), +.lineage-edge-evidence td:nth-child(2) { + text-align: left; +} + +@media print { + .lineage-edge-evidence details { + display: block; + } + + .lineage-edge-evidence details > * { + display: block !important; + } } .lineage-dag-node { diff --git a/frontend/src/App.test.tsx b/frontend/src/App.test.tsx index c48d63c1c..2dee4513d 100644 --- a/frontend/src/App.test.tsx +++ b/frontend/src/App.test.tsx @@ -1987,8 +1987,8 @@ describe("App, authenticated", () => { expect(await screen.findByLabelText("Reconstructed lineage")).toBeInTheDocument(); // Two distinct reconstruct threads (thread-alpha, thread-beta) must // render as two independent branch-tree figures, not merged into one. - expect(screen.getByRole("img", { name: "thread-alpha lineage" })).toBeInTheDocument(); - expect(screen.getByRole("img", { name: "thread-beta lineage" })).toBeInTheDocument(); + expect(screen.getByRole("group", { name: "thread-alpha lineage" })).toBeInTheDocument(); + expect(screen.getByRole("group", { name: "thread-beta lineage" })).toBeInTheDocument(); expect(screen.getByRole("button", { name: "Open post: Follow-up post" })).toBeInTheDocument(); expect(screen.getByRole("button", { name: "Open post: Unrelated thread post" })).toBeInTheDocument(); }); diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 6ba8409ed..76ff51dec 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -106,7 +106,11 @@ import { decodeHtmlEntities } from "./postBodyDisplay"; import { FiveW1H } from "./components/FiveW1H"; import { isFocusableVisible } from "./focusVisibility"; import { subgraphForPost } from "./lineageLayout"; -import { rememberOidcReturnUrl, returnUrlFromLocation, stripOidcCallbackParams } from "./oidcReturnUrl"; +import { + rememberOidcReturnUrl, + returnUrlFromLocation, + stripOidcCallbackParams, +} from "./oidcReturnUrl"; import { isSupportedLocale, LOCALE_LABELS, diff --git a/frontend/src/LineageDag.responsive.test.tsx b/frontend/src/LineageDag.responsive.test.tsx index 4ac272cb0..1b67d9b63 100644 --- a/frontend/src/LineageDag.responsive.test.tsx +++ b/frontend/src/LineageDag.responsive.test.tsx @@ -27,7 +27,7 @@ describe("LineageDag responsive viewport", () => { expect(viewport).toHaveAttribute("tabindex", "0"); expect(viewport).toHaveClass("lineage-dag-viewport"); - const svg = screen.getByRole("img", { name: "A-100 lineage" }); + const svg = screen.getByRole("group", { name: "A-100 lineage" }); expect(viewport).toContainElement(svg); expect(Number(svg.getAttribute("width"))).toBeGreaterThan(320); }); diff --git a/frontend/src/LineageDag.stories.tsx b/frontend/src/LineageDag.stories.tsx index eba3cd29b..1307a596a 100644 --- a/frontend/src/LineageDag.stories.tsx +++ b/frontend/src/LineageDag.stories.tsx @@ -48,9 +48,14 @@ const a100Graph: LineageGraph = { { source: "rec-002", target: "rec-003", - fused_score: 0.9, + fused_score: 0.797623792218737, interval_relation_code: "interval_contains", interval_relation_label: "Contains", + channel_evidence: [ + { signal_code: "temporal", signal_label: "Temporal proximity", score: 0.8, weight: 0.5306114573429468, contribution: 0.4244891658743575, rank: 1 }, + { signal_code: "secondary_key", signal_label: "Secondary key", score: 1, weight: 0.27688071002092646, contribution: 0.27688071002092646, rank: 2 }, + { signal_code: "text", signal_label: "Text similarity", score: 0.5, weight: 0.1925078326361269, contribution: 0.09625391631806345, rank: 3 }, + ], }, { source: "rec-002", @@ -60,6 +65,18 @@ const a100Graph: LineageGraph = { interval_relation_label: "Overlaps", }, ], + reconstruction: { + reconstruction_version: "lineageweave.reconstruct/2.14.0", + generated_at: "2026-08-21T12:00:00+00:00", + min_fused_score: 0.3, + candidate_window: 50, + active_weights: [ + // fast-mlsirm estimate_fixture_channel_weights(), not hand-picked UI weights. + { signal_code: "secondary_key", signal_weight: 0.27688071002092646 }, + { signal_code: "temporal", signal_weight: 0.5306114573429468 }, + { signal_code: "text", signal_weight: 0.1925078326361269 }, + ], + }, }; const meta = { @@ -76,6 +93,9 @@ export default meta; type Story = StoryObj; +export const ConnectionEvidence: Story = {}; + +export const NoLlmChannel: Story = {}; export const ContainsAndOverlaps: Story = {}; // Edge case: no reconstructed lineage yet -- must not render an empty SVG. diff --git a/frontend/src/LineageDag.test.tsx b/frontend/src/LineageDag.test.tsx index f18d3395a..cb71816c0 100644 --- a/frontend/src/LineageDag.test.tsx +++ b/frontend/src/LineageDag.test.tsx @@ -1,32 +1,24 @@ import { fireEvent, render, screen } from "@testing-library/react"; import userEvent from "@testing-library/user-event"; import { describe, expect, it, vi } from "vitest"; -import { LineageDag } from "./LineageDag"; import type { LineageGraph } from "./api"; +import { LineageDag } from "./LineageDag"; -const a100Graph: LineageGraph = { +const graph: LineageGraph = { nodes: [ { id: "rec-002", group: "A-100", - label: "Pricing renegotiation follow-up", - occurred_at: "2026-01-06T00:00:00", - is_root: false, + label: "Kickoff recap", + occurred_at: "2026-01-02T00:00:00", + is_root: true, is_branch_point: true, }, { id: "rec-003", group: "A-100", - label: "Pricing renegotiation: revised quote sent", - occurred_at: "2026-01-10T00:00:00", - is_root: false, - is_branch_point: false, - }, - { - id: "rec-004", - group: "A-100", - label: "Delivery schedule question raised", - occurred_at: "2026-01-07T00:00:00", + label: "Pricing follow-up", + occurred_at: "2026-01-03T00:00:00", is_root: false, is_branch_point: false, }, @@ -35,21 +27,155 @@ const a100Graph: LineageGraph = { { source: "rec-002", target: "rec-003", - fused_score: 0.9, - interval_relation_code: "interval_contains", - interval_relation_label: "Contains", - }, - { - source: "rec-002", - target: "rec-004", - fused_score: 0.85, - interval_relation_code: "interval_overlaps", - interval_relation_label: "Overlaps", + fused_score: 0.7, + channel_evidence: [ + { + signal_code: "text", + signal_label: "Text similarity", + score: 0.5, + weight: 0.5, + contribution: 0.25, + rank: 1, + }, + { + signal_code: "secondary_key", + signal_label: "Secondary key match", + score: 1.0, + weight: 0.25, + contribution: 0.25, + rank: 2, + }, + { + signal_code: "temporal", + signal_label: "Temporal proximity", + score: 0.8, + weight: 0.25, + contribution: 0.2, + rank: 3, + }, + ], }, ], + reconstruction: { + reconstruction_version: "lineageweave.reconstruct/2.14.0", + generated_at: "2026-08-21T12:00:00+00:00", + min_fused_score: 0.3, + candidate_window: 50, + active_weights: [ + { signal_code: "temporal", signal_weight: 0.25 }, + { signal_code: "secondary_key", signal_weight: 0.25 }, + { signal_code: "text", signal_weight: 0.5 }, + ], + }, }; -const graph = { +describe("LineageDag channel evidence", () => { + it("groups connection evidence by thread in a merged graph", () => { + const secondGroup: LineageGraph = { + nodes: graph.nodes.map((node) => ({ + ...node, + id: `second-${node.id}`, + group: "B-200", + label: `Second ${node.label}`, + })), + edges: graph.edges.map((edge) => ({ + ...edge, + source: `second-${edge.source}`, + target: `second-${edge.target}`, + })), + }; + + render( + , + ); + + const firstEvidence = screen.getByRole("region", { name: "A-100 lineage viewport" }); + const secondEvidence = screen.getByRole("region", { name: "B-200 lineage viewport" }); + expect(firstEvidence).toHaveTextContent("Pricing follow-up follows Kickoff recap"); + expect(firstEvidence).not.toHaveTextContent("Second Kickoff recap"); + expect(secondEvidence).toHaveTextContent("Second Pricing follow-up follows Second Kickoff recap"); + }); + + it("discloses exact inferred values without hover-only interaction", async () => { + render(); + const disclosure = screen.getByText(/fused score 0.700000/).closest("details"); + expect(disclosure).toHaveTextContent("Pricing follow-up follows Kickoff recap"); + const edgeButton = screen.getByRole("button", { + name: "Open connection evidence: Kickoff recap to Pricing follow-up", + }); + expect(screen.getByRole("group", { name: "A-100 lineage" })).toBeInTheDocument(); + expect(disclosure).not.toHaveAttribute("open"); + await userEvent.click(edgeButton); + expect(disclosure).toHaveAttribute("open"); + expect( + screen.getByText("Each connection is inferred from independent signals. It is not a causal claim."), + ).toBeInTheDocument(); + expect(screen.getByText("No LLM adjudication participated in this connection.")).toBeInTheDocument(); + expect(screen.getByText("lineageweave.reconstruct/2.14.0")).toBeInTheDocument(); + expect(screen.getAllByText("0.250000").length).toBeGreaterThan(0); + expect(screen.getByText("0.200000")).toBeInTheDocument(); + expect(screen.getByText(/fused score 0.700000/)).toBeInTheDocument(); + expect(screen.queryByText(/causal relationship/i)).not.toBeInTheDocument(); + expect(edgeButton).toHaveAttribute("tabindex", "0"); + expect(edgeButton).toHaveAttribute("aria-pressed", "true"); + await userEvent.click(screen.getByText(/fused score 0.700000/)); + expect(disclosure).not.toHaveAttribute("open"); + expect(edgeButton).toHaveAttribute("aria-pressed", "false"); + }); + + it("does not claim a missing LLM channel when no evidence was recorded", () => { + render( + , + ); + expect(screen.queryByText("No LLM adjudication participated in this connection.")).not.toBeInTheDocument(); + }); + + it("keeps the LLM channel visible when it participated", async () => { + render( + , + ); + await userEvent.click(screen.getByText(/fused score 0.780000/)); + expect(screen.getByText("LLM adjudication")).toBeInTheDocument(); + expect(screen.queryByText("No LLM adjudication participated in this connection.")).not.toBeInTheDocument(); + }); +}); + +const nodeHitTargetGraph = { nodes: [ { id: "rec-001", @@ -71,6 +197,33 @@ const graph = { edges: [{ source: "rec-001", target: "rec-002", fused_score: 0.8 }], }; +const a100Graph: LineageGraph = { + nodes: [ + { + id: "rec-002", group: "A-100", label: "Pricing renegotiation follow-up", + occurred_at: "2026-01-06T00:00:00", is_root: false, is_branch_point: true, + }, + { + id: "rec-003", group: "A-100", label: "Pricing renegotiation: revised quote sent", + occurred_at: "2026-01-10T00:00:00", is_root: false, is_branch_point: false, + }, + { + id: "rec-004", group: "A-100", label: "Delivery schedule question raised", + occurred_at: "2026-01-07T00:00:00", is_root: false, is_branch_point: false, + }, + ], + edges: [ + { + source: "rec-002", target: "rec-003", fused_score: 0.9, + interval_relation_code: "interval_contains", interval_relation_label: "Contains", + }, + { + source: "rec-002", target: "rec-004", fused_score: 0.85, + interval_relation_code: "interval_overlaps", interval_relation_label: "Overlaps", + }, + ], +}; + describe("LineageDag", () => { it("shows Contains and Overlaps as visible text, not hover-only", () => { const { container } = render( @@ -107,7 +260,7 @@ describe("LineageDag", () => { }); it("gives every node mark a 24x24px-minimum transparent hit target ahead of the visible mark", () => { - render( undefined} />); + render( undefined} />); const button = screen.getByRole("button", { name: "Open post: Initial site visit and project scope discussion" }); const circles = button.querySelectorAll("circle"); expect(circles).toHaveLength(2); @@ -124,7 +277,7 @@ describe("LineageDag", () => { it("still opens the post when the enlarged hit target is clicked", async () => { const onSelectPost = vi.fn(); - render(); + render(); await userEvent.click(screen.getByRole("button", { name: "Open post: Pricing renegotiation follow-up" })); expect(onSelectPost).toHaveBeenCalledWith("rec-002"); }); @@ -132,7 +285,7 @@ describe("LineageDag", () => { it("shows an empty-state message instead of an empty graph", () => { render(); expect(screen.getByText("No reconstructed lineage yet. Rebuild after seeding posts.")).toBeInTheDocument(); - expect(screen.queryByRole("img")).not.toBeInTheDocument(); + expect(document.querySelector("svg")).not.toBeInTheDocument(); }); it("renders one branch figure per lineage group, git-branch style", () => { @@ -148,7 +301,7 @@ describe("LineageDag", () => { expect(screen.getByText("Project Alpha (2 records, 1 lineage edges)")).toBeInTheDocument(); expect(screen.getByText("Project Beta (1 records, 0 lineage edges)")).toBeInTheDocument(); - expect(screen.getAllByRole("img")).toHaveLength(2); + expect(container.querySelectorAll('svg[role="group"]')).toHaveLength(2); expect(container.querySelectorAll(".lineage-dag-edge")).toHaveLength(1); expect(container.querySelector(".lineage-dag-branch")).toBeInTheDocument(); expect(container.querySelector(".lineage-dag-root")).toBeInTheDocument(); @@ -163,9 +316,11 @@ describe("LineageDag", () => { ], edges: [], }; - render(); + const { container } = render(); - const headings = screen.getAllByRole("img").map((img) => img.getAttribute("aria-label")); + const headings = [...container.querySelectorAll('svg[role="group"]')].map((svg) => + svg.getAttribute("aria-label"), + ); expect(headings).toEqual(["Zeta Corp lineage", "Ungrouped lineage"]); }); diff --git a/frontend/src/LineageDag.tsx b/frontend/src/LineageDag.tsx index eb5752c57..3a0f3f92f 100644 --- a/frontend/src/LineageDag.tsx +++ b/frontend/src/LineageDag.tsx @@ -1,4 +1,5 @@ -import type { LineageGraph, LineageGraphEdge } from "./api"; +import { useMemo, useState } from "react"; +import type { LineageChannelEvidence, LineageGraph, LineageGraphEdge } from "./api"; import { t, tf } from "./i18n"; import { layoutLineageDag } from "./lineageLayout"; import "./LineageDag.css"; @@ -7,6 +8,18 @@ function truncateLabel(label: string): string { return label.length > 34 ? `${label.slice(0, 33)}…` : label; } +function edgeKey(edge: LineageGraphEdge): string { + return `${edge.source}:${edge.target}`; +} + +function formatExact(value: number): string { + return value.toFixed(6); +} + +function llmParticipated(evidence: LineageChannelEvidence[]): boolean { + return evidence.some((item) => item.signal_code === "llm"); +} + function intervalLabel(edge: LineageGraphEdge): string | undefined { const label = edge.interval_relation_label?.trim(); return label || undefined; @@ -34,6 +47,12 @@ export function LineageDag({ currentPostId?: string; }) { const groups = layoutLineageDag(graph); + const labelById = useMemo( + () => Object.fromEntries(graph.nodes.map((node) => [node.id, node.label])), + [graph.nodes], + ); + const [selectedEdge, setSelectedEdge] = useState(null); + if (graph.nodes.length === 0) { return

{t("No reconstructed lineage yet. Rebuild after seeding posts.")}

; } @@ -73,20 +92,35 @@ export function LineageDag({ viewBox={`0 0 ${group.width} ${group.height}`} width={group.width} height={Math.max(120, group.height)} - role="img" + role="group" aria-label={tf("{group} lineage", { group: group.heading })} > {group.edges.map((edge) => { const from = byId[edge.source]; const to = byId[edge.target]; const midX = (from.x + to.x) / 2; - const midY = (from.y + to.y) / 2; + const key = edgeKey(edge); + const selected = selectedEdge === key; const relation = intervalLabel(edge); return ( - + setSelectedEdge(key)} + onKeyDown={(event) => { + if (event.key === "Enter" || event.key === " ") { + event.preventDefault(); + setSelectedEdge(key); + } + }} > {relation @@ -103,16 +137,6 @@ export function LineageDag({ })} - {relation ? ( - - {t(relation)} - - ) : null} ); })} @@ -190,6 +214,109 @@ export function LineageDag({ ); })} +
+

{t("Connection evidence")}

+

+ {t("Each connection is inferred from independent signals. It is not a causal claim.")} +

+ {graph.reconstruction ? ( +
+
+
{t("Reconstruction version")}
+
{graph.reconstruction.reconstruction_version}
+
+
+
{t("Generated at")}
+
{graph.reconstruction.generated_at}
+
+
+
{t("Active weight profile")}
+
+ {graph.reconstruction.active_weights + .map((item) => `${t(signalLabel(item.signal_code))}: ${formatExact(item.signal_weight)}`) + .join(", ")} +
+
+
+ ) : null} + {groups.map((group) => ( +
+

{group.heading}

+ {group.edges.map((edge) => { + const key = edgeKey(edge); + const evidence = edge.channel_evidence ?? []; + const fromLabel = labelById[edge.source] ?? edge.source; + const toLabel = labelById[edge.target] ?? edge.target; + return ( +
{ + const details = event.currentTarget; + if (details.open) { + setSelectedEdge(key); + } else if (selectedEdge === key) { + setSelectedEdge(null); + } + }} + > + + {tf("{from} follows {to}, fused score {score}", { + from: toLabel, + to: fromLabel, + score: formatExact(edge.fused_score), + })} + + {evidence.length > 0 && !llmParticipated(evidence) ? ( +

{t("No LLM adjudication participated in this connection.")}

+ ) : null} + {evidence.length > 0 ? ( + + + + + + + + + + + + + {evidence.map((item) => ( + + + + + + + + ))} + +
{t("Connection evidence")}
{t("Rank")}{t("Signal")}{t("Score")}{t("Weight")}{t("Contribution")}
{item.rank}{t(item.signal_label)}{formatExact(item.score)}{formatExact(item.weight)}{formatExact(item.contribution)}
+ ) : null} +
+ ); + })} +
+ ))} +
); } + +function signalLabel(signalCode: string): string { + switch (signalCode) { + case "temporal": + return "Temporal proximity"; + case "secondary_key": + return "Secondary key match"; + case "text": + return "Text similarity"; + case "llm": + return "LLM adjudication"; + default: + return signalCode; + } +} diff --git a/frontend/src/PostBody.test.tsx b/frontend/src/PostBody.test.tsx index b0b768195..0b6844e38 100644 --- a/frontend/src/PostBody.test.tsx +++ b/frontend/src/PostBody.test.tsx @@ -391,6 +391,71 @@ describe("PostBody", () => { expect(screen.getByText("Panel")).toBeInTheDocument(); }); + it("keeps separator-free OCR rows in the existing image table path", () => { + render( + '} + imageContent={[ + { + unit_index: 0, + mime_type: "image/png", + status_code: "described", + extracted_text: "No. | Item\n1 | Panel", + caption: "A table image", + tags: [], + }, + ]} + />, + ); + + expect(screen.getByRole("table")).toHaveClass("post-image-text-table"); + expect(screen.getAllByRole("row")).toHaveLength(2); + }); + + it("renders a Markdown table in the source body and keeps empty cells", () => { + render( + , + ); + + expect(screen.getByRole("table")).toHaveClass("post-markdown-table"); + expect(screen.getAllByRole("row")).toHaveLength(2); + expect(screen.getByText("Owner")).toBeInTheDocument(); + expect(screen.getByText("Before")).toBeInTheDocument(); + expect(screen.getByText("After")).toBeInTheDocument(); + }); + + it("does not turn pipe-delimited prose into a table", () => { + render(); + + expect(screen.queryByRole("table")).not.toBeInTheDocument(); + expect(screen.getByText((text) => text.includes("Alice | manager"))).toBeInTheDocument(); + }); + + it("renders Markdown tables when persisted text units are present", () => { + const table = "| Field | Value |\n| --- | --- |\n| Owner | Buyer |"; + render( + , + ); + + expect(screen.getByRole("table")).toHaveClass("post-markdown-table"); + expect(screen.getByText("Owner")).toBeInTheDocument(); + }); + it("shows persisted image-region bounding ranges beside captions", () => { render( { const cells = row.split("|").map((cell) => cell.trim()); if (cells[0] === "") cells.shift(); if (cells[cells.length - 1] === "") cells.pop(); return cells; - }) - .filter((row) => !row.every((cell) => /^:?-{3,}:?$/.test(cell))) + }); + const separatorIndex = rawRows.findIndex( + (row) => row.length > 1 && row.every((cell) => /^:?-{3,}:?$/.test(cell)), + ); + if (requireSeparator && separatorIndex !== 1) return null; + const rows = rawRows + .filter((_row, rowIndex) => rowIndex !== separatorIndex) .filter((row) => row.length > 1 && row.some(Boolean)); if (rows.length < 2 || rows.some((row) => row.length !== rows[0].length)) return null; if (rows[0].length < 2) return null; return rows; } -function renderImageText(text: string) { - const rows = parsePipeDelimitedTable(text); - if (!rows) return

{renderStyledText(text)}

; +function renderPipeTable( + text: string, + className: string, + keyPrefix: string, + requireSeparator = true, +): ReactNode | null { + const rows = parsePipeDelimitedTable(text, requireSeparator); + if (!rows) return null; return ( - +
{rows.map((row, rowIndex) => ( - + {row.map((cell, cellIndex) => ( - + ))} ))} @@ -55,6 +69,14 @@ function renderImageText(text: string) { ); } +function renderImageText(text: string) { + return ( + renderPipeTable(text, "post-body-table post-image-text-table", "post-image-text", false) ?? ( +

{renderStyledText(text)}

+ ) + ); +} + const SAFE_EMBEDDED_IMAGE_SOURCE = /^data:image\/(?:png|jpe?g|gif|webp|avif|bmp|x-icon|vnd\.microsoft\.icon);base64,[A-Za-z0-9+/]+={0,2}$/i; @@ -161,6 +183,13 @@ function renderSegment(segment: PostBodySegment, index: number, imageContent?: P } } +function renderTextSegment(segment: Extract, index: number) { + return ( + renderPipeTable(segment.text, "post-body-table post-markdown-table", `post-markdown-${index}`) ?? + renderSegment(segment, index) + ); +} + function isStructuredTableRow(unit: PostContentUnit): boolean { return ( unit.unit_label === "tr" || @@ -289,7 +318,7 @@ function renderStructuredUnits( ? unit.indent_level : undefined; rendered.push( - renderSegment( + renderTextSegment( { kind: "text", text: displayUnitText(unit.unit_text), @@ -327,7 +356,7 @@ export function PostBody({ {splitPostBody(body).map((segment, index) => { const content = segment.kind === "image" ? imageContent[imageOrdinal++] : undefined; if (segment.kind !== "text") return renderSegment(segment, index, content); - return renderSegment(segment, index, content); + return renderTextSegment(segment, index); })} ); diff --git a/frontend/src/api.ts b/frontend/src/api.ts index a0826e242..2c812ccaa 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -507,10 +507,28 @@ export interface LineageGraphNode { is_branch_point: boolean; } +export interface LineageChannelEvidence { + signal_code: string; + signal_label: string; + score: number; + weight: number; + contribution: number; + rank: number; +} + +export interface LineageRebuildProfile { + reconstruction_version: string; + generated_at: string; + min_fused_score: number; + candidate_window: number; + active_weights: { signal_code: string; signal_weight: number }[]; +} + export interface LineageGraphEdge { source: string; target: string; fused_score: number; + channel_evidence?: LineageChannelEvidence[]; interval_relation_code?: string; interval_relation_label?: string; } @@ -519,6 +537,7 @@ export interface LineageGraph { nodes: LineageGraphNode[]; edges: LineageGraphEdge[]; truncated?: boolean; + reconstruction?: LineageRebuildProfile | null; isolation_reason?: "comparison_candidates_available" | "no_comparison_group" | null; } diff --git a/frontend/src/i18n.test.ts b/frontend/src/i18n.test.ts index 00650624c..18485aba9 100644 --- a/frontend/src/i18n.test.ts +++ b/frontend/src/i18n.test.ts @@ -61,6 +61,10 @@ describe("i18n", () => { "Read leftover map rank {rank}, observed Y {observed}, and expected E {expected} after IRT main effects, then open this post.", "Leftover map rank 0 means no leftover structure after IRT main effects. Read observed Y {observed} and expected E {expected}, then open this post.", "Showing the first {shown} of {total} posts known at this cutoff.", + "Connection evidence", + "Each connection is inferred from independent signals. It is not a causal claim.", + "No LLM adjudication participated in this connection.", + "Temporal proximity", "Contains", "Overlaps", "Interval relations", diff --git a/frontend/src/i18n.ts b/frontend/src/i18n.ts index 711d3e60b..2d19b1cb8 100644 --- a/frontend/src/i18n.ts +++ b/frontend/src/i18n.ts @@ -335,6 +335,26 @@ const TRANSLATIONS: Partial>> = { "Open evidence: {title}": "근거 열기: {title}", "Open post: {label}": "글 열기: {label}", "{from} follows {to} ({score})": "{from}이(가) {to}을(를) 따름 ({score})", + "Connection evidence": "연결 근거", + "Each connection is inferred from independent signals. It is not a causal claim.": + "각 연결은 독립된 신호로부터 추론된 것이며, 인과 관계가 아닙니다.", + "No LLM adjudication participated in this connection.": + "이 연결에는 LLM 판정이 참여하지 않았습니다.", + "Open connection evidence: {from} to {to}": "연결 근거 열기: {from} → {to}", + "{from} follows {to}, fused score {score}": + "{from}이(가) {to}을(를) 따름, 융합 점수 {score}", + Signal: "신호", + Score: "점수", + Weight: "가중치", + Contribution: "기여", + Rank: "순위", + "Reconstruction version": "재구성 버전", + "Generated at": "생성 시각", + "Active weight profile": "사용한 가중치 프로필", + "Temporal proximity": "시간 근접성", + "Secondary key match": "보조 키 일치", + "Text similarity": "텍스트 유사도", + "LLM adjudication": "LLM 판정", "{from} follows {to} ({score}) — {relation}": "{from}이(가) {to}을(를) 따름 ({score}) — {relation}", "Interval relations": "시간 구간 관계", "{from} relates to {to} as {relation}; open {label}": @@ -815,6 +835,24 @@ const TRANSLATIONS: Partial>> = { "Open evidence: {title}": "打开证据:{title}", "Open post: {label}": "打开文章:{label}", "{from} follows {to} ({score})": "{from} 接续 {to}({score})", + "Connection evidence": "连接证据", + "Each connection is inferred from independent signals. It is not a causal claim.": + "每条连接均由独立信号推断得出,并非因果关系。", + "No LLM adjudication participated in this connection.": "此连接未使用 LLM 裁定。", + "Open connection evidence: {from} to {to}": "打开连接证据:{from} 至 {to}", + "{from} follows {to}, fused score {score}": "{from} 接续 {to},融合分数 {score}", + Signal: "信号", + Score: "分数", + Weight: "权重", + Contribution: "贡献", + Rank: "排名", + "Reconstruction version": "重建版本", + "Generated at": "生成时间", + "Active weight profile": "所用权重配置", + "Temporal proximity": "时间接近", + "Secondary key match": "次级键匹配", + "Text similarity": "文本相似度", + "LLM adjudication": "LLM 裁定", "{from} follows {to} ({score}) — {relation}": "{from} 接续 {to}({score})— {relation}", "Interval relations": "时间区间关系", "{from} relates to {to} as {relation}; open {label}": @@ -1294,6 +1332,26 @@ const TRANSLATIONS: Partial>> = { "{group} lineage": "{group}の系譜", "Open post: {label}": "投稿を開く: {label}", "{from} follows {to} ({score})": "{from}は{to}に続く({score})", + "Connection evidence": "接続の根拠", + "Each connection is inferred from independent signals. It is not a causal claim.": + "各接続は独立した信号から推論されたものであり、因果関係ではありません。", + "No LLM adjudication participated in this connection.": + "この接続に LLM 判定は関与していません。", + "Open connection evidence: {from} to {to}": "接続の根拠を開く: {from} → {to}", + "{from} follows {to}, fused score {score}": + "{from}は{to}に続く、融合スコア {score}", + Signal: "信号", + Score: "スコア", + Weight: "重み", + Contribution: "寄与", + Rank: "順位", + "Reconstruction version": "再構成バージョン", + "Generated at": "生成日時", + "Active weight profile": "使用した重みプロファイル", + "Temporal proximity": "時間的近接", + "Secondary key match": "副次キー一致", + "Text similarity": "テキスト類似度", + "LLM adjudication": "LLM 判定", "{from} follows {to} ({score}) — {relation}": "{from}は{to}に続く({score})— {relation}", "Interval relations": "時間区間の関係", "{from} relates to {to} as {relation}; open {label}": @@ -1774,6 +1832,26 @@ const TRANSLATIONS: Partial>> = { "{group} lineage": "Dòng sự kiện {group}", "Open post: {label}": "Mở bài viết: {label}", "{from} follows {to} ({score})": "{from} tiếp nối {to} ({score})", + "Connection evidence": "Bằng chứng liên kết", + "Each connection is inferred from independent signals. It is not a causal claim.": + "Mỗi liên kết được suy ra từ các tín hiệu độc lập. Đây không phải là quan hệ nhân quả.", + "No LLM adjudication participated in this connection.": + "Kết nối này không có sự tham gia của phán định LLM.", + "Open connection evidence: {from} to {to}": "Mở bằng chứng liên kết: {from} đến {to}", + "{from} follows {to}, fused score {score}": + "{from} tiếp nối {to}, điểm hợp nhất {score}", + Signal: "Tín hiệu", + Score: "Điểm", + Weight: "Trọng số", + Contribution: "Đóng góp", + Rank: "Hạng", + "Reconstruction version": "Phiên bản tái dựng", + "Generated at": "Thời điểm tạo", + "Active weight profile": "Hồ sơ trọng số đã dùng", + "Temporal proximity": "Gần về thời gian", + "Secondary key match": "Khớp khóa phụ", + "Text similarity": "Độ tương đồng văn bản", + "LLM adjudication": "Phán định LLM", "{from} follows {to} ({score}) — {relation}": "{from} tiếp nối {to} ({score}) — {relation}", "Interval relations": "Quan hệ khoảng thời gian", "{from} relates to {to} as {relation}; open {label}": diff --git a/frontend/src/lineageLayout.ts b/frontend/src/lineageLayout.ts index 17f7606d5..c5b54c9f6 100644 --- a/frontend/src/lineageLayout.ts +++ b/frontend/src/lineageLayout.ts @@ -99,6 +99,8 @@ export function subgraphForPost(graph: LineageGraph, postId: string): LineageGra return { nodes, edges: graph.edges.filter((edge) => ids.has(edge.source) && ids.has(edge.target)), + truncated: graph.truncated, + reconstruction: graph.reconstruction, }; } diff --git a/frontend/src/postBodyDisplay.test.ts b/frontend/src/postBodyDisplay.test.ts index 3a5a5e3af..ac3afb35d 100644 --- a/frontend/src/postBodyDisplay.test.ts +++ b/frontend/src/postBodyDisplay.test.ts @@ -65,6 +65,109 @@ describe("splitPostBody", () => { ]); }); + it("preserves nested HTML list depth as semantic indentation", () => { + expect( + splitPostBody("
  1. Parent
    1. Child
  2. Sibling
"), + ).toEqual([ + { kind: "text", text: "Parent" }, + { kind: "text", text: "Child", indentLevel: 1 }, + { kind: "text", text: "Sibling" }, + ]); + }); + + it("preserves nested list depth when item text is wrapped in a block child", () => { + expect( + splitPostBody( + "
  1. Parent

    1. Child

  2. Sibling

", + ), + ).toEqual([ + { kind: "text", text: "Parent" }, + { kind: "text", text: "Child", indentLevel: 1 }, + { kind: "text", text: "Sibling" }, + ]); + }); + + it("labels HTML, Word, and OOXML footnotes in the fallback renderer", () => { + expect( + splitPostBody( + '

Body text

' + + '
  1. HTML footnote body

' + + '

1 Word footnote body

' + + "OOXML footnote body", + ), + ).toEqual([ + { kind: "text", text: "Body text" }, + { kind: "text", text: "HTML footnote body", role: "footnote" }, + { kind: "text", text: "¹ Word footnote body", role: "footnote" }, + { kind: "text", text: "OOXML footnote body", role: "footnote" }, + ]); + }); + + it("labels unquoted HTML footnote attributes", () => { + expect(splitPostBody("
  1. Unquoted footnote
")).toEqual([ + { kind: "text", text: "Unquoted footnote", role: "footnote" }, + ]); + }); + + it("stops labeling ordinary content after an HTML footnote list", () => { + expect( + splitPostBody( + '
  1. HTML footnote body

Ordinary body after footnotes

', + ), + ).toEqual([ + { kind: "text", text: "HTML footnote body", role: "footnote" }, + { kind: "text", text: "Ordinary body after footnotes" }, + ]); + }); + + it("labels footnotes inside a labeled wrapper around an HTML list", () => { + expect( + splitPostBody( + '

Body text

' + + '
  1. Wrapped footnote body

' + + "

Ordinary body after footnotes

", + ), + ).toEqual([ + { kind: "text", text: "Body text" }, + { kind: "text", text: "Wrapped footnote body", role: "footnote" }, + { kind: "text", text: "Ordinary body after footnotes" }, + ]); + }); + + it("does not expose control markers for an empty footnote container", () => { + expect(splitPostBody('
    ')).toEqual([{ kind: "text", text: "" }]); + }); + + it("does not infer footnotes from unrelated attribute values", () => { + expect( + splitPostBody( + '
    1. Ordinary list
    ' + + '

    Ordinary paragraph

    ', + ), + ).toEqual([ + { kind: "text", text: "Ordinary list" }, + { kind: "text", text: "Ordinary paragraph" }, + ]); + }); + + it("keeps text boundaries for tags whose names start with a", () => { + expect(splitPostBody('

    AlphaBetaGamma

    ')).toEqual([ + { kind: "text", text: "Alpha Beta Gamma" }, + ]); + }); + + it("keeps a stray pipe line inside its surrounding paragraph", () => { + expect(splitPostBody("

    Before
    ratio A | B
    After

    ")).toEqual([ + { kind: "text", text: "Before ratio A | B After" }, + ]); + }); + + it("space-joins consecutive pipe prose when no Markdown separator exists", () => { + expect(splitPostBody("Alice | manager\nBob | engineer")).toEqual([ + { kind: "text", text: "Alice | manager Bob | engineer" }, + ]); + }); + it("leaves a plain-text post unchanged so existing popups keep their wording", () => { expect(splitPostBody("The full body text.")).toEqual([ { kind: "text", text: "The full body text." }, diff --git a/frontend/src/postBodyDisplay.ts b/frontend/src/postBodyDisplay.ts index 6a4f17540..240b63d22 100644 --- a/frontend/src/postBodyDisplay.ts +++ b/frontend/src/postBodyDisplay.ts @@ -23,10 +23,65 @@ const LIST_ITEM_START = /^\s*(?:[-*•·]\s+|[*†‡](?=\S)|(?:\d{1,3}|[A-Za-z const INDENT_MARKER = "\u0001lw-indent:"; const INDENT_MARKER_END = "\u0002"; const INDENT_MARKER_PATTERN = /lw-indent:(\d+)/g; +const FOOTNOTE_MARKER = "\u0001lw-footnote\u0002"; +const FOOTNOTE_MARKER_PATTERN = new RegExp(FOOTNOTE_MARKER, "g"); + +function markFootnoteTags(markup: string): string { + let footnoteDepth = 0; + const openTags: Array<{ name: string; isFootnote: boolean }> = []; + const voidTags = new Set(["area", "base", "br", "col", "embed", "hr", "img", "input", "link", "meta", "param", "source", "track", "w:br"]); + return markup.replace(HTML_TAG, (tag) => { + const match = tag.match(/^<\s*(\/?)\s*([a-z][a-z0-9:-]*)\b/i); + if (!match) return tag; + const closing = Boolean(match[1]); + const name = match[2].toLowerCase(); + const hasFootnoteLabel = [...tag.matchAll( + /\b(?:class|role)\s*=\s*(?:"([^"]*)"|'([^']*)'|([^\s>"']+))/gi, + )].some((attribute) => + /\b(?:footnotes?|endnotes?|msofootnotetext|msoendnotetext)\b/i.test( + attribute[1] ?? attribute[2] ?? attribute[3] ?? "", + ), + ); + const isContainer = + hasFootnoteLabel && (name === "div" || name === "ol" || name === "ul"); + const isWordParagraph = + name === "p" && hasFootnoteLabel; + const isOoxmlContainer = name === "w:footnote" || name === "w:endnote"; + + if (closing) { + const matchingIndex = openTags.map((entry) => entry.name).lastIndexOf(name); + if (matchingIndex >= 0) { + const closedTags = openTags.splice(matchingIndex); + footnoteDepth = Math.max( + 0, + footnoteDepth - closedTags.filter((entry) => entry.isFootnote).length, + ); + } + return tag; + } + const selfClosing = /\/\s*>$/.test(tag) || voidTags.has(name); + const opensFootnote = isOoxmlContainer || isContainer; + if (!selfClosing) { + openTags.push({ name, isFootnote: opensFootnote }); + } + if (opensFootnote) { + if (!selfClosing) footnoteDepth += 1; + return `${tag}${FOOTNOTE_MARKER}`; + } + if ( + isWordParagraph || + (footnoteDepth > 0 && (name === "li" || name === "p" || name === "w:p")) + ) { + return `${tag}${FOOTNOTE_MARKER}`; + } + return tag; + }); +} function stripIndentMarkers(value: string): string { return value .replace(INDENT_MARKER_PATTERN, "") + .replace(FOOTNOTE_MARKER_PATTERN, "") .split(String.fromCharCode(1)) .join("") .split(String.fromCharCode(2)) @@ -262,16 +317,25 @@ export function splitScriptRuns(text: string): ScriptRun[] { } function stripHtmlTags(text: string): string { - const withScripts = normalizeScriptText(text); + const withScripts = normalizeScriptText(markFootnoteTags(text)); + let listDepth = 0; const withBoundaries = withScripts .replace(BREAK_TAG, "\n") .replace(BLOCK_TAG, (tag) => { - if (/^<\//.test(tag)) return "\n\n"; - return `\n\n${indentMarker(declaredIndentWidth(tag))}`; + const name = tag.match(/^<\/?\s*([a-z0-9:]+)/i)?.[1]?.toLowerCase() ?? ""; + const closing = /^<\//.test(tag); + if (name === "ul" || name === "ol") { + if (closing) listDepth = Math.max(0, listDepth - 1); + else listDepth += 1; + return "\n\n"; + } + if (closing) return "\n\n"; + const nestedListIndent = !closing && listDepth > 0 ? Math.max(0, listDepth - 1) * 4 : 0; + return `\n\n${indentMarker(declaredIndentWidth(tag) + nestedListIndent)}`; }) .replace(WORD_INDENT_TAG, (tag) => indentMarker(declaredIndentWidth(tag))); const withoutTags = withBoundaries.replace(HTML_TAG, (tag) => - /^<\/?w:/i.test(tag) ? "" : " ", + /^<\/?(?:a\b|w:)/i.test(tag) ? "" : " ", ); return decodeHtmlEntities(withoutTags) .split("\n") @@ -290,13 +354,36 @@ function stripHtmlTags(text: string): string { function splitSemanticParagraphs(text: string): string[] { const paragraphs: string[] = []; let lines: string[] = []; + let pipeTableRows: string[] = []; const flush = () => { const paragraph = lines.join(" ").trimEnd(); if (paragraph.trim()) paragraphs.push(paragraph); lines = []; }; + const flushPipeTableRows = () => { + const hasSeparator = pipeTableRows.some((row) => { + const cells = row.trim().replace(/^\|/, "").replace(/\|$/, "").split("|"); + return cells.length >= 2 && cells.every((cell) => /^\s*:?-{3,}:?\s*$/.test(cell)); + }); + if (pipeTableRows.length >= 2 && hasSeparator) { + flush(); + paragraphs.push(pipeTableRows.map((row) => row.trim()).join("\n")); + } else { + lines.push(...pipeTableRows); + } + pipeTableRows = []; + }; for (const line of text.split("\n")) { + const trimmed = line.trim(); + if (trimmed.includes("|")) { + const cells = trimmed.replace(/^\|/, "").replace(/\|$/, "").split("|"); + if (cells.length >= 2 && cells.some((cell) => cell.trim())) { + pipeTableRows.push(line); + continue; + } + } + if (pipeTableRows.length > 0) flushPipeTableRows(); if (!line.trim()) { flush(); continue; @@ -304,6 +391,7 @@ function splitSemanticParagraphs(text: string): string[] { if (lines.length > 0 && LIST_ITEM_START.test(line)) flush(); lines.push(lines.length === 0 ? line.replace(/[ \t]+$/g, "") : line.trim()); } + if (pipeTableRows.length > 0) flushPipeTableRows(); flush(); return paragraphs; } @@ -371,6 +459,7 @@ function isDecodableBase64(raw: string): boolean { function pushText(segments: PostBodySegment[], raw: string, indentUnit: number): void { const text = stripHtmlTags(raw); for (const paragraph of splitSemanticParagraphs(text)) { + const isMarkedFootnote = paragraph.includes(FOOTNOTE_MARKER); const indentLevel = indentationLevel(paragraph, indentUnit); const normalized = stripIndentMarkers(paragraph) .replace(/^[ \t]+/, "") @@ -380,6 +469,7 @@ function pushText(segments: PostBodySegment[], raw: string, indentUnit: number): kind: "text", text: normalized, ...(indentLevel > 0 ? { indentLevel } : {}), + ...(isMarkedFootnote ? { role: "footnote" as const } : {}), }); } } @@ -413,7 +503,7 @@ export function splitPostBody(body: string): PostBodySegment[] { } pushText(segments, body.slice(lastIndex), indentUnit); if (segments.length === 0) { - return [{ kind: "text", text: stripHtmlTags(body) }]; + return [{ kind: "text", text: stripIndentMarkers(stripHtmlTags(body)) }]; } return segments; } diff --git a/lineageweave/__init__.py b/lineageweave/__init__.py index 238421556..c86da1e05 100644 --- a/lineageweave/__init__.py +++ b/lineageweave/__init__.py @@ -10,7 +10,12 @@ from .corporate_hierarchy_resolution import resolve_corporate_entity from .entity_relationship_classification import OrganizationRelationship from .knowledge_graph import random_walk_with_restart, select_related_nodes -from .lineage_persistence import lineage_edge_specs +from .lineage_persistence import ( + CHANNEL_EVIDENCE_TOLERANCE, + lineage_edge_specs, + rank_channel_evidence, + reconstruction_version, +) from .models import Edge, Record, Tree from .naruon_calendar_projection import ( NARUON_CALENDAR_MEDIA_TYPE, @@ -36,8 +41,8 @@ PROV, PROV_CLASSES, PROV_QUALIFICATIONS, - PROV_RELATIONS, PROV_RECOMMENDED_INVERSES, + PROV_RELATIONS, ProvAssertion, ProvGraph, ProvLiteral, @@ -47,8 +52,7 @@ from .voc_evidence import sentence_excerpts __all__ = [ - "ChatAnswer", - "Edge", + "CHANNEL_EVIDENCE_TOLERANCE", "NARUON_CALENDAR_MEDIA_TYPE", "NARUON_CALENDAR_SCHEMA_VERSION", "NARUON_CALENDAR_UNAVAILABLE_NEXT_ACTION", @@ -58,12 +62,14 @@ "NaruonCalendarProjectionClient", "NaruonCalendarWorkspaceEvent", "NaruonCalendarWorkspaceResult", - "OrganizationRelationship", "PROV", "PROV_CLASSES", "PROV_QUALIFICATIONS", - "PROV_RELATIONS", "PROV_RECOMMENDED_INVERSES", + "PROV_RELATIONS", + "ChatAnswer", + "Edge", + "OrganizationRelationship", "PostSummary", "ProvAssertion", "ProvGraph", @@ -80,7 +86,9 @@ "occurrence_to_workspace_event", "parse_naruon_calendar_page", "random_walk_with_restart", + "rank_channel_evidence", "reconstruct", + "reconstruction_version", "resolve_corporate_entity", "select_related_nodes", "sentence_excerpts", diff --git a/lineageweave/adjudication_client.py b/lineageweave/adjudication_client.py index ce020fc70..60cfd61ce 100644 --- a/lineageweave/adjudication_client.py +++ b/lineageweave/adjudication_client.py @@ -15,7 +15,7 @@ import re from typing import Protocol -from .http_client import chat_completion_content, post_json +from .http_client import HttpClientError, chat_completion_content, post_json class AdjudicationClient(Protocol): @@ -53,9 +53,11 @@ def judge_prompt(candidate_label: str, record_label: str) -> str: def parse_confidence(content: str) -> float: - """Clamp the judge's numeric reply into [0, 1]; no number reads as 0.""" + """Clamp a numeric reply into ``[0, 1]`` or fail without inventing zero.""" parsed = parse_confidence_or_none(content) - return 0.0 if parsed is None else parsed + if parsed is None: + raise HttpClientError("adjudication response had no confidence score") + return parsed def parse_confidence_or_none(content: str) -> float | None: @@ -102,4 +104,8 @@ def judge(self, candidate_label: str, record_label: str) -> float: headers={"authorization": f"Bearer {self._api_key}"}, timeout=self._timeout, ) - return parse_confidence(chat_completion_content(body)) + try: + content = chat_completion_content(body) + except (TypeError, ValueError) as exc: + raise HttpClientError("adjudication response did not contain text") from exc + return parse_confidence(content) diff --git a/lineageweave/channel_weight_estimation.py b/lineageweave/channel_weight_estimation.py index c46b6ce5e..38c2b8191 100644 --- a/lineageweave/channel_weight_estimation.py +++ b/lineageweave/channel_weight_estimation.py @@ -183,6 +183,13 @@ def estimate_channel_weights( [dichotomize(scores[channel]) for channel in channels] for scores in pair_channel_scores ] + distinct_columns = { + tuple(row[column] for row in responses) for column in range(len(channels)) + } + if len(distinct_columns) != len(channels): + # Identical channels are one signal copied twice, not independent + # measurement evidence. Refuse instead of double-counting it. + return None for column, channel in enumerate(channels): observed = {row[column] for row in responses} if len(observed) < 2: diff --git a/lineageweave/lineage_persistence.py b/lineageweave/lineage_persistence.py index 4d5ef0dd2..97cfdfce3 100644 --- a/lineageweave/lineage_persistence.py +++ b/lineageweave/lineage_persistence.py @@ -1,19 +1,52 @@ -"""Flatten ``reconstruct()`` trees into the rows ``post_lineage_edge`` stores. +"""Flatten ``reconstruct()`` trees into the rows Event Lineage persists. The reconstruction algorithm stays in ``reconstruct.py``. This module is -only the persistence contract: one ``Edge`` becomes one -``(parent_post_id, child_post_id, fused_score)`` row. Seed scripts and a -future rebuild endpoint share this so they cannot drift from what the -Event Lineage panel reads. +the persistence contract shared by seed scripts and the live rebuild +writer so they cannot drift from what the Event Lineage panel reads. + +Each parent→child edge is still one ``post_lineage_edge`` row. The +winning edge's active channel scores are persisted beside it as +``post_lineage_edge_signal`` rows (ADR 0172). A missing LLM channel is +dropped, never fabricated. Contribution is ``weight * score`` and must +reconcile with ``fused_score`` within the base tolerance plus the bounded +persistence quantization budget. """ from __future__ import annotations -from collections.abc import Sequence +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from decimal import ROUND_HALF_EVEN, Decimal from .adjudication_client import AdjudicationClient from .models import Edge, Record -from .reconstruct import reconstruct +from .reconstruct import ( + DEFAULT_CANDIDATE_WINDOW, + DEFAULT_MIN_FUSED_SCORE, + reconstruct, +) + +RECONSTRUCTION_VERSION_PREFIX = "lineageweave.reconstruct" +CHANNEL_EVIDENCE_TOLERANCE = 1e-6 +SIGNAL_QUANTUM = Decimal("0.000001") + +LINEAGE_SIGNAL_ORDER = ("temporal", "secondary_key", "text", "llm") + +LINEAGE_SIGNAL_LOOKUP_CODES = { + "temporal": "lineage_signal_temporal", + "secondary_key": "lineage_signal_secondary_key", + "text": "lineage_signal_text", + "llm": "lineage_signal_llm", +} + +LINEAGE_SIGNAL_LABELS = { + "temporal": "Temporal proximity", + "secondary_key": "Secondary key match", + "text": "Text similarity", + "llm": "LLM adjudication", +} + +LOOKUP_CODE_TO_SIGNAL = {code: name for name, code in LINEAGE_SIGNAL_LOOKUP_CODES.items()} def lineage_edge_specs( @@ -24,10 +57,11 @@ def lineage_edge_specs( ) -> list[Edge]: """Run reconstruct and return every resulting parent→child edge. - Callers persist these as ``post_lineage_edge`` rows. Record ids must - already be the ids the database will store (UUIDs for product posts, - fixture ids for the library-only demo) -- this function does not - invent or rewrite identifiers. + Callers persist these as ``post_lineage_edge`` rows plus matching + ``post_lineage_edge_signal`` rows. Record ids must already be the ids + the database will store (UUIDs for product posts, fixture ids for the + library-only demo) -- this function does not invent or rewrite + identifiers. ``llm`` defaults to ``None``, which ``reconstruct()`` treats as the unavailable :class:`~lineageweave.adjudication_client.NullAdjudicationClient` @@ -43,3 +77,161 @@ def lineage_edge_specs( """ trees = reconstruct(list(records), llm=llm, weights=weights) return [edge for tree in trees for edge in tree.edges] + + +def reconstruction_version(package_version: str | None = None) -> str: + """Return the reconstruction identity stored on ``event_lineage_rebuild``. + + PostgreSQL remains the authority for live Event Lineage. This string + names the reconstruct implementation that produced the current graph + so a later rebuild cannot silently rewrite historic evidence. + """ + if package_version is None: + from lineageweave import __version__ as package_version + return f"{RECONSTRUCTION_VERSION_PREFIX}/{package_version}" + + +def quantize_signal_value(value: float) -> float: + """Quantize a score, weight, or contribution onto the persisted numeric scale.""" + quantized = Decimal(str(value)).quantize(SIGNAL_QUANTUM, rounding=ROUND_HALF_EVEN) + return float(quantized) + + +@dataclass(frozen=True) +class LineageRebuildSpec: + """Rows one atomic Event Lineage rebuild writes besides the edge list.""" + + reconstruction_version: str + min_fused_score: float + candidate_window: int + channel_weights: tuple[tuple[str, float], ...] + signal_rows: tuple[dict[str, object], ...] + + +def channel_signal_rows( + edge: Edge, + weights: Mapping[str, float], +) -> list[dict[str, object]]: + """Build persistable signal rows for one reconstructed edge. + + One row per active channel. The LLM channel is omitted when it did + not participate. ``signal_weight`` is the normalized active weight + actually used. ``signal_contribution`` is ``weight * score``. + + Raises: + ValueError: if recorded contributions do not reconcile with + ``edge.fused_score`` after accounting for one half quantum per + persisted channel and a small floating-point guard. + """ + active_weights = dict(weights) + rows: list[dict[str, object]] = [] + contribution_sum = 0.0 + for channel in LINEAGE_SIGNAL_ORDER: + if channel not in edge.channel_scores or channel not in active_weights: + continue + score = quantize_signal_value(float(edge.channel_scores[channel])) + weight = quantize_signal_value(float(active_weights[channel])) + contribution = quantize_signal_value(float(active_weights[channel]) * float(edge.channel_scores[channel])) + contribution_sum += contribution + rows.append( + { + "parent_post_id": edge.parent_id, + "child_post_id": edge.child_id, + "signal_code": LINEAGE_SIGNAL_LOOKUP_CODES[channel], + "channel_name": channel, + "signal_score": score, + "signal_weight": weight, + "signal_contribution": contribution, + } + ) + # Each numeric(8,6) contribution can differ from its exact product by + # half a quantum. A fixed tolerance fails valid 3/4-channel edges when + # those independent rounding errors accumulate. + rounding_budget = len(rows) * float(SIGNAL_QUANTUM) / 2 + float(SIGNAL_QUANTUM) + reconciliation_tolerance = max(CHANNEL_EVIDENCE_TOLERANCE, rounding_budget) + residual = abs(contribution_sum - float(edge.fused_score)) + if rows and residual > reconciliation_tolerance: + raise ValueError( + f"channel contributions {contribution_sum} do not reconcile with " + f"fused_score {edge.fused_score} (tolerance {reconciliation_tolerance})" + ) + return rows + + +def rank_channel_evidence(rows: Sequence[Mapping[str, object]]) -> list[dict[str, object]]: + """Project persisted signal rows onto the additive API collection. + + Ordering is contribution descending, then the controlled signal order + ``temporal``, ``secondary_key``, ``text``, ``llm``. Rank is 1-based. + The payload never includes prompts, responses, credentials, or source + text. + """ + + def sort_key(row: Mapping[str, object]) -> tuple[float, int]: + """Contribution descending, ties broken by the canonical channel order.""" + channel = _channel_name(row) + order = LINEAGE_SIGNAL_ORDER.index(channel) if channel in LINEAGE_SIGNAL_ORDER else len(LINEAGE_SIGNAL_ORDER) + return (-float(row["signal_contribution"]), order) + + evidence: list[dict[str, object]] = [] + for rank, row in enumerate(sorted(rows, key=sort_key), start=1): + channel = _channel_name(row) + label = str(row["signal_label"]) if row.get("signal_label") else LINEAGE_SIGNAL_LABELS.get(channel, channel) + evidence.append( + { + "signal_code": channel, + "signal_label": label, + "score": float(row["signal_score"]), + "weight": float(row["signal_weight"]), + "contribution": float(row["signal_contribution"]), + "rank": rank, + } + ) + return evidence + + +def llm_participated(evidence: Sequence[Mapping[str, object]]) -> bool: + """Return whether the optional LLM channel is present in ``evidence``.""" + return any(item.get("signal_code") == "llm" for item in evidence) + + +def lineage_rebuild_spec( + edges: Sequence[Edge], + *, + weights: Mapping[str, float], + min_fused_score: float = DEFAULT_MIN_FUSED_SCORE, + candidate_window: int = DEFAULT_CANDIDATE_WINDOW, + package_version: str | None = None, +) -> LineageRebuildSpec: + """Assemble the rebuild metadata and signal rows for ``edges``. + + ``weights`` is the provenance-bearing psychometric estimate that fused + the graph. Every signal row uses that same profile so a later audit can + reproduce the active measurement contract. + """ + active_weights = dict(weights) + + signal_rows: list[dict[str, object]] = [] + for edge in edges: + signal_rows.extend(channel_signal_rows(edge, active_weights)) + + ordered_weights = tuple( + (LINEAGE_SIGNAL_LOOKUP_CODES[name], quantize_signal_value(active_weights[name])) + for name in LINEAGE_SIGNAL_ORDER + if name in active_weights + ) + return LineageRebuildSpec( + reconstruction_version=reconstruction_version(package_version), + min_fused_score=min_fused_score, + candidate_window=candidate_window, + channel_weights=ordered_weights, + signal_rows=tuple(signal_rows), + ) + + +def _channel_name(row: Mapping[str, object]) -> str: + stored = row.get("channel_name") + if isinstance(stored, str) and stored: + return stored + lookup = str(row.get("signal_code") or "") + return LOOKUP_CODE_TO_SIGNAL.get(lookup, lookup) diff --git a/migrations/0103_tenant_settings.sql b/migrations/0103_tenant_settings.sql index 2b78b6b0e..46d52ced2 100644 --- a/migrations/0103_tenant_settings.sql +++ b/migrations/0103_tenant_settings.sql @@ -1,3 +1,4 @@ +-- migrate.sh replays gated migrations on every Compose start. CREATE TABLE IF NOT EXISTS tenant_settings ( id int PRIMARY KEY CHECK (id = 1), brand_name text NOT NULL DEFAULT 'LineageWeave', diff --git a/migrations/0174_post_lineage_edge_signal.sql b/migrations/0174_post_lineage_edge_signal.sql new file mode 100644 index 000000000..fe237ab07 --- /dev/null +++ b/migrations/0174_post_lineage_edge_signal.sql @@ -0,0 +1,90 @@ +-- ADR 0172: persist Event Lineage channel evidence beside each fused edge. +-- lookup_code is globally unique, so signal codes are prefixed. +-- Migration 0174 uses CREATE IF NOT EXISTS / ON CONFLICT for idempotent replay. + +insert into common_lookup_value (lookup_category, lookup_code, lookup_label, display_order) values + ('lineage_signal', 'lineage_signal_temporal', 'Temporal proximity', 0), + ('lineage_signal', 'lineage_signal_secondary_key', 'Secondary key match', 1), + ('lineage_signal', 'lineage_signal_text', 'Text similarity', 2), + ('lineage_signal', 'lineage_signal_llm', 'LLM adjudication', 3) +on conflict (lookup_code) do nothing; + +create table if not exists event_lineage_rebuild ( + rebuild_lock boolean primary key default true check (rebuild_lock), + reconstruction_version text not null, + generated_at timestamptz not null, + min_fused_score numeric(8,6) not null, + candidate_window integer not null, + check (min_fused_score >= 0 and min_fused_score <= 1), + check (candidate_window >= 1) +); + +comment on table event_lineage_rebuild is + 'Singleton identity of the live Event Lineage rebuild; replaced atomically.'; + +create table if not exists event_lineage_rebuild_channel ( + rebuild_lock boolean not null default true + references event_lineage_rebuild (rebuild_lock) on delete cascade, + signal_code text not null references common_lookup_value (lookup_code), + signal_weight numeric(8,6) not null, + primary key (rebuild_lock, signal_code), + check (signal_weight > 0 and signal_weight <= 1) +); + +comment on table event_lineage_rebuild_channel is + 'Normalized active channel weights used by the live Event Lineage rebuild.'; + +create table if not exists post_lineage_edge_signal ( + parent_post_id uuid not null, + child_post_id uuid not null, + signal_code text not null references common_lookup_value (lookup_code), + signal_score numeric(8,6) not null, + signal_weight numeric(8,6) not null, + signal_contribution numeric(8,6) not null, + primary key (parent_post_id, child_post_id, signal_code), + foreign key (parent_post_id, child_post_id) + references post_lineage_edge (parent_post_id, child_post_id) + on delete cascade, + check (signal_score >= 0 and signal_score <= 1), + check (signal_weight > 0 and signal_weight <= 1), + check (signal_contribution >= 0 and signal_contribution <= 1) +); + +comment on table post_lineage_edge_signal is + 'Per-channel score, active weight, and contribution for one reconstructed lineage edge.'; + +do $migration$ +begin + if not exists ( + select 1 from pg_constraint + where conname = 'event_lineage_rebuild_channel_signal_code_check' + and conrelid = 'event_lineage_rebuild_channel'::regclass + ) then + alter table event_lineage_rebuild_channel + add constraint event_lineage_rebuild_channel_signal_code_check + check (signal_code in ( + 'lineage_signal_temporal', + 'lineage_signal_secondary_key', + 'lineage_signal_text', + 'lineage_signal_llm' + )); + end if; + if not exists ( + select 1 from pg_constraint + where conname = 'post_lineage_edge_signal_code_check' + and conrelid = 'post_lineage_edge_signal'::regclass + ) then + alter table post_lineage_edge_signal + add constraint post_lineage_edge_signal_code_check + check (signal_code in ( + 'lineage_signal_temporal', + 'lineage_signal_secondary_key', + 'lineage_signal_text', + 'lineage_signal_llm' + )); + end if; +end; +$migration$; + +create index if not exists post_lineage_edge_signal_child_idx + on post_lineage_edge_signal (child_post_id, parent_post_id); diff --git a/migrations/rollback/0174_post_lineage_edge_signal.sql b/migrations/rollback/0174_post_lineage_edge_signal.sql new file mode 100644 index 000000000..6e484775b --- /dev/null +++ b/migrations/rollback/0174_post_lineage_edge_signal.sql @@ -0,0 +1,10 @@ +drop table if exists post_lineage_edge_signal; +drop table if exists event_lineage_rebuild_channel; +drop table if exists event_lineage_rebuild; +delete from common_lookup_value + where lookup_code in ( + 'lineage_signal_temporal', + 'lineage_signal_secondary_key', + 'lineage_signal_text', + 'lineage_signal_llm' + ); diff --git a/scripts/import_postgresql_posts.py b/scripts/import_postgresql_posts.py index 0484a2800..26de5f856 100644 --- a/scripts/import_postgresql_posts.py +++ b/scripts/import_postgresql_posts.py @@ -26,17 +26,20 @@ if str(REPOSITORY_ROOT) not in sys.path: sys.path.insert(0, str(REPOSITORY_ROOT)) -from backend.app.lineage_ingestion import ( - ChannelWeightsNotEstimated, - rebuild_lineage, +from backend.app.lineage_ingestion import ChannelWeightsNotEstimated, rebuild_lineage +from lineageweave.adjudication_client import ( + ContextualOrchestratorAdjudicationClient, + NullAdjudicationClient, ) -from lineageweave.synthetic_seed_cleanup import cleanup_synthetic_seed from lineageweave.embedding_client import orchestrator_embedding_client from lineageweave.image_content import orchestrator_vision_client from lineageweave.llm_context import build_post_llm_metadata, use_llm_metadata from lineageweave.post_content_persistence import persist_post_content -from lineageweave.post_structure import ContextualOrchestratorPostStructureClient, NullPostStructureClient - +from lineageweave.post_structure import ( + ContextualOrchestratorPostStructureClient, + NullPostStructureClient, +) +from lineageweave.synthetic_seed_cleanup import cleanup_synthetic_seed SOURCE_NAMESPACE = uuid.UUID("b6e4b1d6-5fd0-4ca1-92b0-8f7a4e2df83e") @@ -471,6 +474,14 @@ async def import_rows(args: argparse.Namespace) -> dict[str, object]: if orchestrator_base_url and orchestrator_api_key else NullPostStructureClient() ) + adjudication_client = ( + ContextualOrchestratorAdjudicationClient( + orchestrator_base_url, + orchestrator_api_key, + ) + if orchestrator_base_url and orchestrator_api_key + else NullAdjudicationClient() + ) for row in rows: if _source_code_matches(row, mapping.draft, args.exclude_draft_value) or _source_code_matches( row, mapping.deleted, args.exclude_deleted_value @@ -638,7 +649,7 @@ async def import_rows(args: argparse.Namespace) -> dict[str, object]: # (ADR 0200 point 1). Skip the rebuild with a next-action note # instead of failing the whole import. try: - edges = await rebuild_lineage(target) + edges = await rebuild_lineage(target, llm=adjudication_client) lineage_summary: dict[str, object] = {"lineage_edges": len(edges)} except ChannelWeightsNotEstimated as exc: lineage_summary = { diff --git a/scripts/seed_demo_data.py b/scripts/seed_demo_data.py index 490c326cb..092d28fa5 100644 --- a/scripts/seed_demo_data.py +++ b/scripts/seed_demo_data.py @@ -637,7 +637,7 @@ def _seed_reconstructed_lineage(cur, author_account_id, corporate_entity_id, pro reconstruct() already knows the A-100 fork. """ from lineageweave.fixtures import sample_records - from lineageweave.lineage_persistence import lineage_edge_specs + from lineageweave.lineage_persistence import lineage_edge_specs, lineage_rebuild_spec # Idempotency first: a re-seed of an already-seeded database (whose # estimate was persisted on the first pass) must not abort just @@ -653,13 +653,42 @@ def _seed_reconstructed_lineage(cur, author_account_id, corporate_entity_id, pro persisted = insert_fixture_source_posts( cur, author_account_id, corporate_entity_id, process_unit_id ) - for edge in lineage_edge_specs(persisted, weights=estimate.weights): + edges = lineage_edge_specs(persisted, weights=estimate.weights) + spec = lineage_rebuild_spec(edges, weights=estimate.weights) + cur.execute("delete from event_lineage_rebuild") + cur.execute( + "insert into event_lineage_rebuild " + "(rebuild_lock, reconstruction_version, generated_at, min_fused_score, candidate_window) " + "values (true, %s, now(), %s, %s)", + (spec.reconstruction_version, spec.min_fused_score, spec.candidate_window), + ) + for signal_code, signal_weight in spec.channel_weights: + cur.execute( + "insert into event_lineage_rebuild_channel " + "(rebuild_lock, signal_code, signal_weight) values (true, %s, %s)", + (signal_code, signal_weight), + ) + for edge in edges: cur.execute( "insert into post_lineage_edge " "(parent_post_id, child_post_id, fused_score, interval_relation_code) " "values (%s, %s, %s, 'interval_before') on conflict do nothing", (edge.parent_id, edge.child_id, edge.fused_score), ) + for row in spec.signal_rows: + cur.execute( + "insert into post_lineage_edge_signal " + "(parent_post_id, child_post_id, signal_code, signal_score, signal_weight, signal_contribution) " + "values (%s, %s, %s, %s, %s, %s) on conflict do nothing", + ( + row["parent_post_id"], + row["child_post_id"], + row["signal_code"], + row["signal_score"], + row["signal_weight"], + row["signal_contribution"], + ), + ) def _write_post_summary(cur, post_id, summary) -> None: diff --git a/tests/test_adjudication_client.py b/tests/test_adjudication_client.py index 576f5e9c4..0ea9f6e6c 100644 --- a/tests/test_adjudication_client.py +++ b/tests/test_adjudication_client.py @@ -1,6 +1,9 @@ from __future__ import annotations +import pytest + from lineageweave.adjudication_client import ContextualOrchestratorAdjudicationClient +from lineageweave.http_client import HttpClientError def test_adjudication_uses_supported_auto_mode_and_long_local_timeout(monkeypatch) -> None: @@ -21,3 +24,24 @@ def fake_post_json(url, payload, *, headers, timeout): assert captured["payload"]["mode"] == "auto" assert captured["payload"]["reasoning_effort"] == "auto" assert captured["timeout"] == 180.0 + + +@pytest.mark.parametrize( + "body", + [ + {"choices": [{"message": {"content": "not a score"}}]}, + {"choices": []}, + {"choices": [{"message": {"content": None}}]}, + ], +) +def test_adjudication_fails_closed_for_unscoreable_responses(monkeypatch, body) -> None: + """A provider failure must not become a genuine unrelated score of zero.""" + monkeypatch.setattr( + "lineageweave.adjudication_client.post_json", lambda *args, **kwargs: body + ) + client = ContextualOrchestratorAdjudicationClient( + base_url="http://orchestrator:8000", api_key="synthetic-token" + ) + + with pytest.raises(HttpClientError): + client.judge("workshop", "follow-up bid") diff --git a/tests/test_analysis_run_worker.py b/tests/test_analysis_run_worker.py index 8dcbb500a..26466430a 100644 --- a/tests/test_analysis_run_worker.py +++ b/tests/test_analysis_run_worker.py @@ -317,3 +317,29 @@ async def fake_deliver(_pool, **kwargs): assert last_id == "2-1" assert delivered == ["00000000-0000-0000-0000-000000000002"] + + +@pytest.mark.anyio +async def test_one_unexpected_delivery_failure_does_not_end_the_worker(monkeypatch): + """A malformed provider reply must not stop later durable deliveries.""" + delivered = [] + + async def fake_deliver(conn, **kwargs): + del conn + if kwargs["analysis_run_id"].endswith("1"): + raise RuntimeError("malformed provider reply") + delivered.append(kwargs["analysis_run_id"]) + + monkeypatch.setattr(analysis_run_worker, "deliver_queued_analysis_run", fake_deliver) + + last_id = await analysis_run_worker.consume_analysis_run_stream_once( + _TwoRunsValkey(), + _Pool(), + last_id="0-0", + database_url="postgresql://synthetic", + tepp_client=TeppClient(), + adjudication_client=NullAdjudicationClient(), + ) + + assert last_id == "2-1" + assert delivered == ["00000000-0000-0000-0000-000000000002"] diff --git a/tests/test_estimate_llm_channel_weights_script.py b/tests/test_estimate_llm_channel_weights_script.py index c9ab53a05..210c95d84 100644 --- a/tests/test_estimate_llm_channel_weights_script.py +++ b/tests/test_estimate_llm_channel_weights_script.py @@ -8,7 +8,10 @@ from __future__ import annotations +import pytest + from lineageweave.adjudication_client import judge_prompt, parse_confidence +from lineageweave.http_client import HttpClientError import scripts.estimate_llm_channel_weights as script @@ -31,7 +34,8 @@ def test_shared_judge_prompt_and_confidence_parse_round_trip() -> None: assert "Record B: Follow-up record" in prompt assert parse_confidence("0.85") == 0.85 assert parse_confidence("confidence: 0.4 maybe") == 0.4 - assert parse_confidence("no number here") == 0.0 + with pytest.raises(HttpClientError): + parse_confidence("no number here") assert parse_confidence("1.7") == 1.0 diff --git a/tests/test_import_postgresql_posts.py b/tests/test_import_postgresql_posts.py index 12bbe86ff..cd35d4f57 100644 --- a/tests/test_import_postgresql_posts.py +++ b/tests/test_import_postgresql_posts.py @@ -101,7 +101,7 @@ async def no_content(*_args, **_kwargs) -> None: async def no_cleanup(*_args, **_kwargs) -> dict[str, int]: return {"synthetic_rows_removed": 0} - async def no_edges(_conn) -> list[object]: + async def no_edges(_conn, *, llm=None) -> list[object]: return [] monkeypatch.setattr("scripts.import_postgresql_posts.asyncpg.connect", fake_connect) diff --git a/tests/test_lineage_channel_evidence.py b/tests/test_lineage_channel_evidence.py new file mode 100644 index 000000000..47d42f8da --- /dev/null +++ b/tests/test_lineage_channel_evidence.py @@ -0,0 +1,190 @@ +"""Event Lineage channel evidence must round-trip without fabricating LLM scores.""" + +from __future__ import annotations + +from functools import lru_cache +from pathlib import Path + +import pytest + +from lineageweave.channel_weight_estimation import ( + estimate_channel_weights, + estimate_fixture_channel_weights, + simulate_fixture_pair_scores, +) +from lineageweave.fixtures import sample_records +from lineageweave.lineage_persistence import ( + CHANNEL_EVIDENCE_TOLERANCE, + LINEAGE_SIGNAL_LOOKUP_CODES, + channel_signal_rows, + lineage_edge_specs, + lineage_rebuild_spec, + llm_participated, + rank_channel_evidence, + quantize_signal_value, + reconstruction_version, +) +from lineageweave.models import Edge + +_ROOT = Path(__file__).resolve().parents[1] + + +@lru_cache(maxsize=1) +def _estimated_weights() -> dict[str, float]: + """Return fast-mlsirm estimates for the declared synthetic design.""" + estimate = estimate_fixture_channel_weights() + assert estimate is not None + return estimate.weights + + +def _no_llm_edge() -> Edge: + scores = {"temporal": 0.8, "secondary_key": 1.0, "text": 0.5} + weights = _estimated_weights() + fused = sum(weights[name] * scores[name] for name in scores) + return Edge("parent-a", "child-b", fused, scores) + + +def test_grounded_edge_round_trips_scores_and_normalized_weights() -> None: + edge = _no_llm_edge() + weights = _estimated_weights() + rows = channel_signal_rows(edge, weights) + codes = [row["channel_name"] for row in rows] + assert codes == ["temporal", "secondary_key", "text"] + by_name = {row["channel_name"]: row for row in rows} + assert by_name["temporal"]["signal_weight"] == quantize_signal_value(weights["temporal"]) + assert by_name["text"]["signal_weight"] == quantize_signal_value(weights["text"]) + evidence = rank_channel_evidence(rows) + assert [item["rank"] for item in evidence] == [1, 2, 3] + assert sum(item["contribution"] for item in evidence) == pytest.approx(edge.fused_score) + + +def test_no_llm_reconstruction_persists_exactly_three_channels() -> None: + edge = _no_llm_edge() + weights = _estimated_weights() + rows = channel_signal_rows(edge, weights) + assert [row["channel_name"] for row in rows] == ["temporal", "secondary_key", "text"] + assert "llm" not in {row["channel_name"] for row in rows} + assert "lineage_signal_llm" not in {row["signal_code"] for row in rows} + evidence = rank_channel_evidence(rows) + assert llm_participated(evidence) is False + assert {row["channel_name"]: row["signal_weight"] for row in rows} == { + channel: quantize_signal_value(weight) for channel, weight in weights.items() + } + + +def test_contributions_reconcile_to_fused_score_within_tolerance() -> None: + edge = _no_llm_edge() + rows = channel_signal_rows(edge, _estimated_weights()) + residual = abs(sum(float(row["signal_contribution"]) for row in rows) - edge.fused_score) + assert residual <= CHANNEL_EVIDENCE_TOLERANCE + + +def test_rebuild_accepts_expected_multi_channel_quantization_error() -> None: + scores = { + "temporal": 0.1234567, + "secondary_key": 0.2345678, + "text": 0.3456789, + } + weights = _estimated_weights() + edge = Edge( + "parent-a", + "child-b", + sum(weights[name] * scores[name] for name in scores), + scores, + ) + + rows = channel_signal_rows(edge, weights) + assert len(rows) == 3 + assert lineage_rebuild_spec([edge], weights=weights).signal_rows + + +def test_duplicated_text_proxy_cannot_invent_an_llm_weight() -> None: + """A copied text score is not an independent LLM validity anchor.""" + pair_scores, group_ids = simulate_fixture_pair_scores() + assert ( + estimate_channel_weights( + [{**scores, "llm": scores["text"]} for scores in pair_scores], + group_ids, + ) + is None + ) + + +def test_mismatched_fused_score_is_rejected() -> None: + edge = Edge("parent-a", "child-b", 0.99, {"temporal": 0.1, "secondary_key": 0.1, "text": 0.1}) + with pytest.raises(ValueError, match="do not reconcile"): + channel_signal_rows(edge, _estimated_weights()) + + +def test_fixture_reconstruction_never_fabricates_llm() -> None: + weights = _estimated_weights() + edges = lineage_edge_specs(sample_records(), weights=weights) + assert edges + spec = lineage_rebuild_spec(edges, weights=weights) + assert all(row["channel_name"] != "llm" for row in spec.signal_rows) + assert "lineage_signal_llm" not in {code for code, _weight in spec.channel_weights} + for edge in edges: + rows = channel_signal_rows(edge, weights) + residual = abs(sum(float(row["signal_contribution"]) for row in rows) - edge.fused_score) + assert residual <= CHANNEL_EVIDENCE_TOLERANCE + + +def test_rebuild_spec_is_idempotent_for_the_same_edges() -> None: + weights = _estimated_weights() + edges = lineage_edge_specs(sample_records(), weights=weights) + first = lineage_rebuild_spec(edges, weights=weights, package_version="2.14.0") + second = lineage_rebuild_spec(edges, weights=weights, package_version="2.14.0") + assert first == second + assert first.reconstruction_version == reconstruction_version("2.14.0") + assert first.reconstruction_version == "lineageweave.reconstruct/2.14.0" + + +def test_rank_is_contribution_then_controlled_signal_order() -> None: + weights = _estimated_weights() + tied_contribution = min(weights.values()) + rows = [ + { + "channel_name": "text", + "signal_code": LINEAGE_SIGNAL_LOOKUP_CODES["text"], + "signal_score": tied_contribution / weights["text"], + "signal_weight": weights["text"], + "signal_contribution": tied_contribution, + }, + { + "channel_name": "temporal", + "signal_code": LINEAGE_SIGNAL_LOOKUP_CODES["temporal"], + "signal_score": tied_contribution / weights["temporal"], + "signal_weight": weights["temporal"], + "signal_contribution": tied_contribution, + }, + { + "channel_name": "secondary_key", + "signal_code": LINEAGE_SIGNAL_LOOKUP_CODES["secondary_key"], + "signal_score": tied_contribution / weights["secondary_key"], + "signal_weight": weights["secondary_key"], + "signal_contribution": tied_contribution, + }, + ] + evidence = rank_channel_evidence(rows) + assert [item["signal_code"] for item in evidence] == ["temporal", "secondary_key", "text"] + assert [item["rank"] for item in evidence] == [1, 2, 3] + + +def test_migrate_sh_replays_channel_evidence_and_tenant_settings() -> None: + """ADR 0166's portable filename boundary must still cover 0103 and 0174 + on existing volumes -- no individual allowlist entry is needed for + either, since the general four-digit pattern already replays both. + """ + migrate = (_ROOT / "docker/postgres-init/migrate.sh").read_text() + assert "[0-9][0-9][0-9][0-9]_*" in migrate + assert "000[0-9]_*|001[01]_*" in migrate + assert "0103_*" not in migrate + assert "0174_*" not in migrate + + +def test_channel_evidence_migration_has_no_jsonb() -> None: + migration = (_ROOT / "migrations" / "0174_post_lineage_edge_signal.sql").read_text() + assert "jsonb" not in migration.casefold() + assert "post_lineage_edge_signal" in migration + assert "event_lineage_rebuild" in migration + assert "on delete cascade" in migration.casefold() diff --git a/tests/test_lineage_ingestion.py b/tests/test_lineage_ingestion.py index 6e708d87a..38cf7d282 100644 --- a/tests/test_lineage_ingestion.py +++ b/tests/test_lineage_ingestion.py @@ -5,21 +5,34 @@ import asyncio import math from datetime import UTC, datetime, timezone +from functools import lru_cache import pytest import backend.app.lineage_ingestion as ingestion from backend.app.lineage_ingestion import ( + _budgeted_llm, interval_relations_for_post, lineage_graphs_for_posts, persist_lineage_edges, + rebuild_lineage, + rebuild_lineage_from_pool, + reconstruct_group_key, + records_from_source_posts, visible_lineage_graph, ) +from lineageweave.channel_weight_estimation import estimate_fixture_channel_weights from lineageweave.fixtures import sample_records -from lineageweave.lineage_persistence import lineage_edge_specs -from lineageweave.models import Edge +from lineageweave.lineage_persistence import lineage_edge_specs, quantize_signal_value +from lineageweave.models import Edge, Record +@lru_cache(maxsize=1) +def _fixture_weights() -> dict[str, float]: + """Return the fast-mlsirm estimate for the declared synthetic design.""" + estimate = estimate_fixture_channel_weights() + assert estimate is not None + return estimate.weights def test_missing_weight_table_is_detected_without_an_aborting_query() -> None: class MissingTableConnection: async def fetchval(self, query: str): @@ -55,9 +68,8 @@ async def fetch(self, _query: str): "knowledge_cutoff": datetime(2026, 1, 1, tzinfo=UTC), } return [ - {**provenance, "channel_code": "temporal", "weight_value": 0.2}, - {**provenance, "channel_code": "secondary_key", "weight_value": 0.3}, - {**provenance, "channel_code": "text", "weight_value": 0.5}, + {**provenance, "channel_code": channel, "weight_value": weight} + for channel, weight in _fixture_weights().items() ] assert asyncio.run( @@ -96,16 +108,15 @@ async def fetch(self, _query: str): "tepp_knowledge_cutoff": datetime(2026, 1, 1, tzinfo=UTC), } return [ - {**provenance, "channel_code": "temporal", "weight_value": 0.5}, - {**provenance, "channel_code": "secondary_key", "weight_value": 0.3}, - {**provenance, "channel_code": "text", "weight_value": 0.2}, + {**provenance, "channel_code": channel, "weight_value": weight} + for channel, weight in _fixture_weights().items() ] assert asyncio.run( ingestion.load_estimated_channel_weights( StoredWeightConnection(), {"temporal", "secondary_key", "text"} ) - ) == {"temporal": 0.5, "secondary_key": 0.3, "text": 0.2} + ) == _fixture_weights() def test_duplicate_exact_channel_sets_fail_closed() -> None: @@ -448,12 +459,229 @@ def test_records_use_persisted_thread_keys_not_process_unit_or_voc_type() -> Non "created_at": datetime(2026, 1, 6, tzinfo=timezone.utc), } ] - records = ingestion.records_from_source_posts(rows) + records = records_from_source_posts(rows) assert records[0].group_key == "A-100" assert records[0].secondary_key == "proj-alpha" assert records[0].occurred_at.tzinfo is None +def test_rebuild_passes_the_configured_adjudication_client(monkeypatch) -> None: + events: list[str] = [] + + class FakeTransaction: + async def __aenter__(self): + events.append("transaction_enter") + + async def __aexit__(self, *_args): + events.append("transaction_exit") + + class FakeConnection: + async def fetch(self, _query: str, *_args): + events.append("fetch") + return [] + + async def fetchval(self, _query: str): + return False + + def transaction(self): + return FakeTransaction() + + client = object() + captured: dict[str, object] = {} + + def fake_lineage_edge_specs(_records, *, llm=None, weights=None): + captured["llm"] = llm + captured["weights"] = weights + return [] + + async def fake_load_weights(_conn, _channels): + return _fixture_weights() + + async def fake_persist_lineage_edges(_conn, _edges, _weights=None, _points=None): + events.append("persist") + return None + + async def fake_to_thread(function, *args, **kwargs): + events.append("reconstruct") + captured["offloaded_function"] = function + return function(*args, **kwargs) + + monkeypatch.setattr( + "backend.app.lineage_ingestion.lineage_edge_specs", fake_lineage_edge_specs + ) + monkeypatch.setattr( + "backend.app.lineage_ingestion.persist_lineage_edges", fake_persist_lineage_edges + ) + monkeypatch.setattr( + "backend.app.lineage_ingestion.load_estimated_channel_weights", fake_load_weights + ) + monkeypatch.setattr(asyncio, "to_thread", fake_to_thread) + asyncio.run(rebuild_lineage(FakeConnection(), llm=client)) + + assert captured["llm"] is client + assert captured["weights"] == _fixture_weights() + assert captured["offloaded_function"] is fake_lineage_edge_specs + assert events == ["fetch", "reconstruct", "transaction_enter", "persist", "transaction_exit"] + + +def test_estimated_weight_channels_include_only_an_available_llm() -> None: + """Weight lookup must match the exact channel set reconstruction can use.""" + assert ingestion.estimated_weight_channels(None) == { + "temporal", + "secondary_key", + "text", + } + assert ingestion.estimated_weight_channels(type("LLM", (), {"available": True})()) == { + "temporal", + "secondary_key", + "text", + "llm", + } + + +def test_rebuild_drops_llm_before_candidate_pair_budget_is_exceeded(monkeypatch) -> None: + """Keep a large live rebuild from issuing unbounded provider calls.""" + + records = [ + Record(f"record-{index}", "shared-group", f"Record {index}", datetime(2026, 1, index + 1)) + for index in range(3) + ] + monkeypatch.setattr( + "backend.app.lineage_ingestion.MAXIMUM_LIVE_LLM_PAIR_EVALUATIONS", 1 + ) + + assert _budgeted_llm(records, object()) is None + + +def test_rebuild_drops_llm_before_estimated_weight_lookup(monkeypatch) -> None: + """Never load a four-channel estimate for a three-channel reconstruction.""" + + records = [ + Record(f"record-{index}", "shared-group", f"Record {index}", datetime(2026, 1, index + 1)) + for index in range(3) + ] + captured: dict[str, object] = {} + + class FakeTransaction: + async def __aenter__(self): + return None + + async def __aexit__(self, *_args): + return None + + class FakeConnection: + def transaction(self): + return FakeTransaction() + + async def fake_load_records(_conn): + return records + + async def fake_load_weights(_conn, channels): + captured["channels"] = channels + return _fixture_weights() + + async def fake_reconstruct(_records, llm, _weights): + captured["llm"] = llm + return [] + + async def fake_persist(_conn, _edges, _weights, _points): + return None + + monkeypatch.setattr(ingestion, "MAXIMUM_LIVE_LLM_PAIR_EVALUATIONS", 1) + monkeypatch.setattr(ingestion, "_load_lineage_records", fake_load_records) + monkeypatch.setattr(ingestion, "load_estimated_channel_weights", fake_load_weights) + monkeypatch.setattr(ingestion, "_reconstruct_lineage_records", fake_reconstruct) + monkeypatch.setattr(ingestion, "persist_lineage_edges", fake_persist) + + asyncio.run(rebuild_lineage(FakeConnection(), llm=type("LLM", (), {"available": True})())) + + assert captured == { + "channels": {"temporal", "secondary_key", "text"}, + "llm": None, + } + + +def test_pooled_rebuild_releases_the_connection_during_reconstruction(monkeypatch) -> None: + """Keep provider work outside the pool and transaction, then replace atomically.""" + events: list[str] = [] + + class FakeTransaction: + async def __aenter__(self): + events.append("transaction-enter") + + async def __aexit__(self, *_args): + events.append("transaction-exit") + + class FakeConnection: + async def fetch(self, _query: str, *_args): + events.append("fetch") + return [] + + async def fetchval(self, _query: str): + return False + + def transaction(self): + return FakeTransaction() + + class FakeAcquire: + def __init__(self, pool): + self.pool = pool + + async def __aenter__(self): + self.pool.active += 1 + events.append("acquire") + return self.pool.connection + + async def __aexit__(self, *_args): + self.pool.active -= 1 + events.append("release") + + class FakePool: + def __init__(self): + self.active = 0 + self.connection = FakeConnection() + + def acquire(self): + return FakeAcquire(self) + + pool = FakePool() + + async def fake_to_thread(function, *args, **kwargs): + assert pool.active == 0 + events.append("reconstruct") + return function(*args, **kwargs) + + async def fake_persist(_conn, _edges, _weights=None, _points=None): + assert pool.active == 1 + assert events[-1] == "transaction-enter" + events.append("persist") + + async def fake_load_weights(_conn, _channels): + return _fixture_weights() + + monkeypatch.setattr(asyncio, "to_thread", fake_to_thread) + monkeypatch.setattr( + "backend.app.lineage_ingestion.persist_lineage_edges", fake_persist + ) + monkeypatch.setattr( + "backend.app.lineage_ingestion.load_estimated_channel_weights", fake_load_weights + ) + + asyncio.run(rebuild_lineage_from_pool(pool)) + + assert events == [ + "acquire", + "fetch", + "release", + "reconstruct", + "acquire", + "transaction-enter", + "persist", + "transaction-exit", + "release", + ] + + def test_records_fall_back_to_corporate_entity_when_thread_keys_are_empty() -> None: rows = [ { @@ -467,7 +695,7 @@ def test_records_fall_back_to_corporate_entity_when_thread_keys_are_empty() -> N "created_at": datetime(2026, 2, 1), } ] - records = ingestion.records_from_source_posts(rows) + records = records_from_source_posts(rows) assert records[0].group_key == "cccccccc-cccc-cccc-cccc-cccccccccccc" assert records[0].secondary_key == "" assert records[0].label == "Corp-only post" @@ -485,8 +713,8 @@ def test_display_group_matches_reconstruct_group_key() -> None: "corporate_entity_id": "shared-corp", "thread_group_key": "", } - assert ingestion.reconstruct_group_key(a100) == "A-100" - assert ingestion.reconstruct_group_key(ungrouped) == "shared-pu" + assert reconstruct_group_key(a100) == "A-100" + assert reconstruct_group_key(ungrouped) == "shared-pu" def test_seed_shaped_rows_rebuild_to_the_designed_a100_fork() -> None: @@ -510,9 +738,8 @@ def test_seed_shaped_rows_rebuild_to_the_designed_a100_fork() -> None: } ) edges = lineage_edge_specs( - ingestion.records_from_source_posts(rows), - # Synthetic unit-test weights (ADR 0200 point 1: no library default). - weights={"temporal": 0.5, "secondary_key": 0.34, "text": 0.16}, + records_from_source_posts(rows), + weights=_fixture_weights(), ) pairs = {(edge.parent_id, edge.child_id) for edge in edges} assert ("rec-002", "rec-003") in pairs @@ -532,10 +759,11 @@ async def execute(self, query: str, *_args): with pytest.raises(ValueError, match="child"): asyncio.run( - persist_lineage_edges( - connection, - [edge], - {"parent": {"created_at": datetime(2026, 1, 1)}}, + persist_lineage_edges( + connection, + [edge], + {}, + points_by_post_id={"parent": {"created_at": datetime(2026, 1, 1)}}, ) ) assert connection.calls == [] @@ -561,11 +789,18 @@ def test_rebuild_fails_closed_without_an_activated_weight_estimate() -> None: for rec in sample_records() ] + class FakeTransaction: + async def __aenter__(self): + return None + + async def __aexit__(self, *_args): + return None + class FakeConnection: def __init__(self) -> None: self.executions: list[tuple[str, tuple[object, ...]]] = [] - async def fetch(self, query: str): + async def fetch(self, query: str, *_args): assert "from source_post" in query return rows @@ -573,8 +808,12 @@ async def fetchval(self, query: str): assert "to_regclass('public.lineage_channel_weight')" in query return False - async def execute(self, query: str, *args: object) -> None: - self.executions.append((query, args)) + async def executemany(self, query: str, args) -> None: + for row in args: + self.executions.append((query, row)) + + def transaction(self): + return FakeTransaction() connection = FakeConnection() with pytest.raises(ingestion.ChannelWeightsNotEstimated) as raised: @@ -621,13 +860,16 @@ def test_rebuild_reconstructs_with_an_activated_estimate() -> None: "tepp_snapshot_sha256": "a" * 64, "tepp_knowledge_cutoff": datetime(2026, 1, 1, tzinfo=UTC), } - for channel, weight in ( - ("temporal", 0.5), - ("secondary_key", 0.34), - ("text", 0.16), - ) + for channel, weight in _fixture_weights().items() ] + class FakeTransaction: + async def __aenter__(self): + return None + + async def __aexit__(self, *_args): + return None + class FakeConnection: def __init__(self) -> None: self.executions: list[tuple[str, tuple[object, ...]]] = [] @@ -644,13 +886,25 @@ async def fetchval(self, _query: str): async def execute(self, query: str, *args: object) -> None: self.executions.append((query, args)) + async def executemany(self, query: str, args) -> None: + for row in args: + self.executions.append((query, row)) + + def transaction(self): + return FakeTransaction() + connection = FakeConnection() edges = asyncio.run(ingestion.rebuild_lineage(connection)) pairs = {(edge.parent_id, edge.child_id) for edge in edges} assert {("rec-002", "rec-003"), ("rec-002", "rec-004")} <= pairs assert connection.executions[0] == ("delete from post_lineage_edge", ()) - assert len(connection.executions) == len(edges) + 1 + inserted_edges = [ + row + for query, row in connection.executions + if query.startswith("insert into post_lineage_edge (") + ] + assert len(inserted_edges) == len(edges) def test_focused_lineage_graph_includes_a_post_outside_landing_limit() -> None: @@ -692,43 +946,317 @@ class FakeConnection: ] async def fetch(self, query: str, *_args): + if "post_lineage_edge_signal" in query: + return getattr(self, "signals", []) + if "event_lineage_rebuild_channel" in query: + return getattr(self, "rebuild_channels", []) + if "event_lineage_rebuild" in query: + return getattr(self, "rebuilds", []) return self.edges if "post_lineage_edge" in query else self.posts connection = FakeConnection() - landing = asyncio.run( - ingestion.visible_lineage_graph(connection, lambda row: True, limit=1) - ) + landing = asyncio.run(visible_lineage_graph(connection, lambda row: True, limit=1)) focused = asyncio.run( - ingestion.visible_lineage_graph( - connection, lambda row: True, limit=1, focus_post_id="post-a" - ) + visible_lineage_graph(connection, lambda row: True, limit=1, focus_post_id="post-a") ) isolated = asyncio.run( - ingestion.visible_lineage_graph( - connection, lambda row: True, limit=1, focus_post_id="post-c" - ) + visible_lineage_graph(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, - "isolation_reason": "no_comparison_group", + assert isolated["nodes"] == [] + assert isolated["edges"] == [] + assert isolated["truncated"] is False + assert isolated["reconstruction"] is None + assert isolated["isolation_reason"] == "no_comparison_group" + assert focused["edges"][0]["channel_evidence"] == [] + + +class _RecordingConnection: + def __init__(self) -> None: + self.statements: list[tuple[str, tuple]] = [] + self.batches: list[tuple[str, list[tuple]]] = [] + + async def execute(self, query: str, *args): + self.statements.append((query, args)) + + async def executemany(self, query: str, args): + rows = list(args) + self.batches.append((query, rows)) + self.statements.extend((query, row) for row in rows) + + async def fetch(self, query: str, *_args): + return [] + + +def test_visible_graph_attaches_ranked_channel_evidence() -> None: + class FakeConnection: + def __init__(self) -> None: + self.queries: list[str] = [] + + posts = [ + { + "post_id": "post-a", + "post_title": "A", + "voc_type_code": "voc", + "visibility_code": "public", + "corporate_entity_id": "corp", + "process_unit_id": "pu", + "thread_group_key": "thread-a", + "created_at": datetime(2026, 1, 1), + }, + { + "post_id": "post-b", + "post_title": "B", + "voc_type_code": "voc", + "visibility_code": "public", + "corporate_entity_id": "corp", + "process_unit_id": "pu", + "thread_group_key": "thread-a", + "created_at": datetime(2026, 1, 2), + }, + ] + edges = [{"parent_post_id": "post-a", "child_post_id": "post-b", "fused_score": 0.7}] + signals = [ + { + "parent_post_id": "post-a", + "child_post_id": "post-b", + "signal_code": "lineage_signal_text", + "signal_score": 0.5, + "signal_weight": _fixture_weights()["text"], + "signal_contribution": _fixture_weights()["text"] * 0.5, + }, + { + "parent_post_id": "post-a", + "child_post_id": "post-b", + "signal_code": "lineage_signal_temporal", + "signal_score": 0.8, + "signal_weight": _fixture_weights()["temporal"], + "signal_contribution": _fixture_weights()["temporal"] * 0.8, + }, + { + "parent_post_id": "post-a", + "child_post_id": "post-b", + "signal_code": "lineage_signal_secondary_key", + "signal_score": 1.0, + "signal_weight": _fixture_weights()["secondary_key"], + "signal_contribution": _fixture_weights()["secondary_key"], + }, + ] + rebuilds = [ + { + "reconstruction_version": "lineageweave.reconstruct/2.14.0", + "generated_at": datetime(2026, 8, 21, 12, 0, 0), + "min_fused_score": 0.3, + "candidate_window": 50, + } + ] + rebuild_channels = [ + { + "signal_code": f"lineage_signal_{channel}", + "signal_weight": weight, + } + for channel in ("temporal", "secondary_key", "text") + for weight in (_fixture_weights()[channel],) + ] + + async def fetch(self, query: str, *_args): + self.queries.append(query) + if "post_lineage_edge_signal" in query: + return self.signals + if "event_lineage_rebuild_channel" in query: + return self.rebuild_channels + if "event_lineage_rebuild" in query: + return self.rebuilds + return self.edges if "post_lineage_edge" in query else self.posts + + connection = FakeConnection() + graph = asyncio.run(visible_lineage_graph(connection, lambda row: True)) + evidence = graph["edges"][0]["channel_evidence"] + assert {item["signal_code"] for item in evidence} == {"secondary_key", "text", "temporal"} + assert [item["rank"] for item in evidence] == [1, 2, 3] + assert "llm" not in {item["signal_code"] for item in evidence} + assert graph["reconstruction"]["reconstruction_version"] == "lineageweave.reconstruct/2.14.0" + assert graph["reconstruction"]["active_weights"][0]["signal_code"] == "temporal" + signal_query = next(query for query in connection.queries if "post_lineage_edge_signal" in query) + assert "parent_post_id = any($1::uuid[])" in signal_query + assert "child_post_id = any($2::uuid[])" in signal_query + weight_query = next(query for query in connection.queries if "event_lineage_rebuild_channel" in query) + assert "join common_lookup_value as lookup" in weight_query + assert "order by lookup.display_order, channel.signal_code" in weight_query + + +def test_abac_never_reveals_channel_evidence_for_an_invisible_endpoint() -> None: + class FakeConnection: + posts = [ + { + "post_id": "post-public", + "post_title": "Public", + "voc_type_code": "voc", + "visibility_code": "public", + "corporate_entity_id": "corp", + "process_unit_id": "pu", + "thread_group_key": "thread-a", + "created_at": datetime(2026, 1, 1), + }, + { + "post_id": "post-secret", + "post_title": "Secret", + "voc_type_code": "voc", + "visibility_code": "restricted", + "corporate_entity_id": "corp", + "process_unit_id": "pu", + "thread_group_key": "thread-a", + "created_at": datetime(2026, 1, 2), + }, + ] + edges = [ + {"parent_post_id": "post-public", "child_post_id": "post-secret", "fused_score": 0.8} + ] + signals = [ + { + "parent_post_id": "post-public", + "child_post_id": "post-secret", + "signal_code": "lineage_signal_text", + "signal_score": 0.9, + "signal_weight": _fixture_weights()["text"], + "signal_contribution": _fixture_weights()["text"] * 0.9, + } + ] + rebuilds = [] + rebuild_channels = [] + + async def fetch(self, query: str, *_args): + if "post_lineage_edge_signal" in query: + return self.signals + if "event_lineage_rebuild_channel" in query: + return self.rebuild_channels + if "event_lineage_rebuild" in query: + return self.rebuilds + return self.edges if "post_lineage_edge" in query else self.posts + + graph = asyncio.run( + visible_lineage_graph(FakeConnection(), lambda row: row["post_id"] == "post-public") + ) + assert [node["id"] for node in graph["nodes"]] == ["post-public"] + assert graph["edges"] == [] + serialized = str(graph) + assert "post-secret" not in serialized + assert "0.45" not in serialized + assert "lineage_signal_text" not in serialized + + +def test_focused_graph_cannot_bridge_through_an_invisible_post() -> None: + class FakeConnection: + posts = tuple( + { + "post_id": post_id, + "post_title": post_id, + "voc_type_code": "voc", + "visibility_code": visibility, + "corporate_entity_id": "corp", + "process_unit_id": "pu", + "thread_group_key": "thread-a", + "created_at": datetime(2026, 1, day, tzinfo=timezone.utc), + } + for day, post_id, visibility in ( + (1, "post-a", "public"), + (2, "post-hidden", "restricted"), + (3, "post-b", "public"), + ) + ) + edges = ( + {"parent_post_id": "post-a", "child_post_id": "post-hidden", "fused_score": 0.8}, + {"parent_post_id": "post-hidden", "child_post_id": "post-b", "fused_score": 0.8}, + ) + + async def fetch(self, query: str, *_args): + if "post_lineage_edge_signal" in query: + return [] + if "event_lineage_rebuild_channel" in query: + return [] + if "event_lineage_rebuild" in query: + return [] + return self.edges if "post_lineage_edge" in query else self.posts + + graph = asyncio.run( + visible_lineage_graph( + FakeConnection(), + lambda row: row["visibility_code"] == "public", + focus_post_id="post-a", + ) + ) + + assert graph["nodes"] == [] + assert graph["edges"] == [] + + +def test_persist_lineage_edges_replaces_signals_atomically_without_llm() -> None: + from lineageweave.lineage_persistence import lineage_rebuild_spec + from lineageweave.models import Edge + + scores = {"temporal": 0.8, "secondary_key": 1.0, "text": 0.5} + weights = _fixture_weights() + fused = sum(weights[name] * scores[name] for name in scores) + edge = Edge( + "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", + "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb", + fused, + scores, + ) + connection = _RecordingConnection() + points = { + edge.parent_id: {"created_at": datetime(2026, 1, 1)}, + edge.child_id: {"created_at": datetime(2026, 1, 2)}, } - assert [node["id"] for node in hidden_neighbor["nodes"]] == ["post-a"] - assert hidden_neighbor["edges"] == [] + asyncio.run(persist_lineage_edges(connection, [edge], weights, points)) + statements = [sql.casefold() for sql, _args in connection.statements] + assert statements[0].startswith("delete from post_lineage_edge") + assert any("delete from event_lineage_rebuild" in sql for sql in statements) + assert any("insert into post_lineage_edge_signal" in sql for sql in statements) + inserted_codes = [ + args[2] + for sql, args in connection.statements + if "insert into post_lineage_edge_signal" in sql.casefold() + ] + assert inserted_codes == [ + "lineage_signal_temporal", + "lineage_signal_secondary_key", + "lineage_signal_text", + ] + assert [len(rows) for _query, rows in connection.batches] == [3, 1, 3] + assert connection.batches[0][1] == [ + (f"lineage_signal_{channel}", quantize_signal_value(weights[channel])) + for channel in ("temporal", "secondary_key", "text") + ] + spec = lineage_rebuild_spec([edge], weights=weights, package_version="2.14.0") + assert spec.reconstruction_version == "lineageweave.reconstruct/2.14.0" + + +def test_duplicate_rebuild_replays_the_same_delete_insert_sequence() -> None: + from lineageweave.models import Edge + + scores = {"temporal": 0.8, "secondary_key": 1.0, "text": 0.5} + weights = _fixture_weights() + fused = sum(weights[name] * scores[name] for name in scores) + edge = Edge( + "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", + "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb", + fused, + scores, + ) + first = _RecordingConnection() + second = _RecordingConnection() + points = { + edge.parent_id: {"created_at": datetime(2026, 1, 1)}, + edge.child_id: {"created_at": datetime(2026, 1, 2)}, + } + asyncio.run(persist_lineage_edges(first, [edge], weights, points)) + asyncio.run(persist_lineage_edges(second, [edge], weights, points)) + assert [sql for sql, _args in first.statements] == [sql for sql, _args in second.statements] + assert [args for _sql, args in first.statements] == [args for _sql, args in second.statements] def test_visible_lineage_graph_attaches_allen_labels() -> None: @@ -765,6 +1293,10 @@ class FakeConnection: ] async def fetch(self, query: str, *_args): + if "post_lineage_edge_signal" in query: + return [] + if "event_lineage_rebuild" in query: + return [] return self.edges if "post_lineage_edge" in query else self.posts graph = asyncio.run( @@ -860,7 +1392,20 @@ class FakeConnection: {"parent_post_id": "post-c", "child_post_id": "post-d", "fused_score": 0.6}, ] - async def fetch(self, query: str): + async def fetch(self, query: str, *_args): + if "post_lineage_edge_signal" in query: + return [] + if "event_lineage_rebuild_channel" in query: + return [] + if "event_lineage_rebuild" in query: + return [ + { + "reconstruction_version": "lineageweave.reconstruct/2.14.0", + "generated_at": datetime(2026, 1, 5), + "min_fused_score": 0.5, + "candidate_window": 6, + } + ] return self.edges if "post_lineage_edge" in query else self.posts connection = FakeConnection() @@ -888,17 +1433,23 @@ async def fetch(self, query: str): } assert len(merged["edges"]) == 2 assert merged["truncated"] is False + assert merged["reconstruction"]["reconstruction_version"] == ( + "lineageweave.reconstruct/2.14.0" + ) def test_lineage_graphs_for_posts_with_no_citations_is_empty() -> None: class FakeConnection: - async def fetch(self, query: str): + async def fetch(self, query: str, *_args): return [] merged = asyncio.run(lineage_graphs_for_posts(FakeConnection(), lambda row: True, [])) - assert merged == {"nodes": [], "edges": [], "truncated": False} - - + assert merged == { + "nodes": [], + "edges": [], + "truncated": False, + "reconstruction": None, + } 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.""" @@ -925,8 +1476,14 @@ class FakeConnection: ] queries: list[str] = [] - async def fetch(self, query: str): + async def fetch(self, query: str, *_args): self.queries.append(query) + if "post_lineage_edge_signal" in query: + return [] + if "event_lineage_rebuild_channel" in query: + return [] + if "event_lineage_rebuild" in query: + return [] return self.edges if "post_lineage_edge" in query else self.posts connection = FakeConnection() @@ -939,7 +1496,7 @@ async def fetch(self, query: str): ) 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] + edge_queries = [query for query in connection.queries if "from post_lineage_edge" in query and "post_lineage_edge_signal" not in query] assert len(post_queries) == 1 assert len(edge_queries) == 1 assert {node["id"] for node in merged["nodes"]} == { @@ -975,7 +1532,13 @@ def test_lineage_graphs_for_posts_names_truncation_and_keeps_cited_posts() -> No ] class FakeConnection: - async def fetch(self, query: str): + async def fetch(self, query: str, *_args): + if "post_lineage_edge_signal" in query: + return [] + if "event_lineage_rebuild_channel" in query: + return [] + if "event_lineage_rebuild" in query: + return [] return edges if "post_lineage_edge" in query else posts merged = asyncio.run( @@ -1021,7 +1584,13 @@ def __init__(self, posts: list[dict[str, object]], edges: list[dict[str, object] self.edges = edges self.executions: list[tuple[object, ...]] = [] - async def fetch(self, query: str): + async def fetch(self, query: str, *_args): + if "post_lineage_edge_signal" in query: + return [] + if "event_lineage_rebuild_channel" in query: + return [] + if "event_lineage_rebuild" in query: + return [] return self.edges if "post_lineage_edge" in query else self.posts async def fetchval(self, query: str): @@ -1047,6 +1616,7 @@ def test_two_visible_group_members_report_comparison_candidates_available() -> N "nodes": [], "edges": [], "truncated": False, + "reconstruction": None, "isolation_reason": "comparison_candidates_available", } @@ -1087,6 +1657,7 @@ def test_inaccessible_focus_does_not_report_an_isolation_reason() -> None: "nodes": [], "edges": [], "truncated": False, + "reconstruction": None, "isolation_reason": None, } diff --git a/tests/test_schema.py b/tests/test_schema.py index a1ab41869..d88d07bd2 100644 --- a/tests/test_schema.py +++ b/tests/test_schema.py @@ -73,6 +73,9 @@ / "migrations" / "0169_report_leftover_map_axis.sql" ) +_CHANNEL_EVIDENCE_MIGRATION = ( + Path(__file__).resolve().parents[1] / "migrations" / "0174_post_lineage_edge_signal.sql" +) _LEFTOVER_MAP_COVERAGE_MIGRATION = ( Path(__file__).resolve().parents[1] / "migrations" @@ -124,6 +127,7 @@ def schema_db(): cur.execute(_LEFTOVER_MAP_RANK_MIGRATION.read_text()) cur.execute(_LEFTOVER_MAP_COVERAGE_MIGRATION.read_text()) cur.execute(_LEFTOVER_MAP_AXIS_MIGRATION.read_text()) + cur.execute(_CHANNEL_EVIDENCE_MIGRATION.read_text()) cur.execute(_LEFTOVER_MAP_UNEXPLAINED_MIGRATION.read_text()) cur.execute(_LEFTOVER_MAP_CROSS_SHARE_MIGRATION.read_text()) cur.execute(_LEFTOVER_MAP_RECONSTRUCTION_MIGRATION.read_text()) @@ -164,6 +168,9 @@ def test_migration_applies_cleanly(schema_db) -> None: "knowledge_graph_edge_evidence", "issue_ticket", "post_lineage_edge", + "post_lineage_edge_signal", + "event_lineage_rebuild", + "event_lineage_rebuild_channel", "post_evaluation_response", "report_period_score", "report_member_score", @@ -243,6 +250,108 @@ def test_leftover_pair_references_member_and_item_rows(schema_db) -> None: assert "report_period_score" in targets +def test_lineage_channel_evidence_is_cascaded_and_lookup_controlled(schema_db) -> None: + """Signal rows cannot outlive their edge or name an unknown channel.""" + with schema_db.cursor() as cur: + cur.execute( + """ + select pg_get_constraintdef(oid) + from pg_constraint + where conrelid = 'post_lineage_edge_signal'::regclass + order by conname + """ + ) + definitions = " ".join(row[0].casefold() for row in cur.fetchall()) + assert "references post_lineage_edge" in definitions + assert "on delete cascade" in definitions + assert "references common_lookup_value" in definitions + cur.execute( + "select lookup_code from common_lookup_value " + "where lookup_category = 'lineage_signal' order by display_order" + ) + assert [row[0] for row in cur.fetchall()] == [ + "lineage_signal_temporal", + "lineage_signal_secondary_key", + "lineage_signal_text", + "lineage_signal_llm", + ] + cur.execute( + """ + select data_type, numeric_precision, numeric_scale + from information_schema.columns + where table_name = 'post_lineage_edge_signal' + and column_name in ('signal_score', 'signal_weight', 'signal_contribution') + """ + ) + for data_type, precision, scale in cur.fetchall(): + assert data_type == "numeric" + assert precision == 8 + assert scale == 6 + cur.execute( + """ + select column_name from information_schema.columns + where table_name in ( + 'post_lineage_edge_signal', + 'event_lineage_rebuild', + 'event_lineage_rebuild_channel' + ) and data_type = 'jsonb' + """ + ) + assert cur.fetchall() == [] + + +def test_lineage_signal_tables_reject_other_lookup_categories(schema_db) -> None: + with schema_db.cursor() as cur: + cur.execute( + "insert into common_lookup_value (lookup_category, lookup_code, lookup_label) " + "values ('test_category', 'not_lineage_signal', 'Not lineage')" + ) + cur.execute( + "insert into event_lineage_rebuild " + "(rebuild_lock, reconstruction_version, generated_at, min_fused_score, candidate_window) " + "values (true, 'test', now(), 0.3, 50)" + ) + statements = ( + "insert into event_lineage_rebuild_channel " + "(rebuild_lock, signal_code, signal_weight) " + "values (true, 'not_lineage_signal', 0.5)", + "insert into post_lineage_edge_signal " + "(parent_post_id, child_post_id, signal_code, signal_score, signal_weight, signal_contribution) " + "values ('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', " + "'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb', 'not_lineage_signal', 0.5, 0.5, 0.25)", + ) + for index, statement in enumerate(statements): + savepoint = f"wrong_signal_{index}" + cur.execute(f"savepoint {savepoint}") + with pytest.raises(psycopg2.errors.CheckViolation): + cur.execute(statement) + cur.execute(f"rollback to savepoint {savepoint}") + schema_db.rollback() + + +def test_lineage_channel_evidence_migration_upgrades_existing_tables(schema_db) -> None: + with schema_db.cursor() as cur: + cur.execute( + "alter table event_lineage_rebuild_channel " + "drop constraint event_lineage_rebuild_channel_signal_code_check" + ) + cur.execute( + "alter table post_lineage_edge_signal " + "drop constraint post_lineage_edge_signal_code_check" + ) + cur.execute(_CHANNEL_EVIDENCE_MIGRATION.read_text()) + cur.execute( + "select conname from pg_constraint where conname in (" + "'event_lineage_rebuild_channel_signal_code_check', " + "'post_lineage_edge_signal_code_check') order by conname" + ) + assert [row[0] for row in cur.fetchall()] == [ + "event_lineage_rebuild_channel_signal_code_check", + "post_lineage_edge_signal_code_check", + ] + schema_db.commit() + + def test_leftover_pair_names_nullable_observed_and_expected_columns(schema_db) -> None: """Every install path preserves legacy pairs while naming new Y and E.""" with schema_db.cursor() as cur:
    {renderStyledText(cell)}{renderStyledText(cell)}