feat: persist and explain Event Lineage channel evidence

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).
 

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.

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

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