feat(lineage): wire real embeddings into the Event Lineage text channel (v2.24.0)

CHANGELOG.d/2.24.0-lineage-text-channel-embeddings.md

@@ -0,0 +1,11 @@
+## [2.24.0] - 2026-08-24
+
+### Fixed
+
+- Event Lineage no longer relies solely on character-overlap text matching
+  when linking posts within a coarsely-grouped bucket: `reconstruct()`'s
+  `text` channel now uses real embedding cosine similarity when an
+  embedding provider is configured, falling back to the prior
+  `difflib.SequenceMatcher` behavior otherwise (ADR 0190). Threaded through
+  `rebuild_lineage`, the analysis-run worker, and
+  `scripts/import_postgresql_posts.py`.

CHANGELOG.md

@@ -8,6 +8,12 @@ All notable changes to this project are documented here. Format follows
 
 ### Fixed
 
+- Event Lineage's `text` scoring channel now uses real embedding cosine
+  similarity when an embedding provider is configured, instead of always
+  falling back to character-overlap matching -- reduces the case where two
+  posts about unrelated topics that happen to share common words or
+  sentence structure get linked as parent/child within a coarsely-grouped
+  bucket (ADR 0190).
 - Corpus-wide lineage rebuilds now run synchronous orchestrator adjudication
   off the API event-loop thread and before the short atomic projection-write
   transaction. A temporary adjudication failure leaves the existing graph

backend/app/analysis_run_start.py

@@ -29,6 +29,7 @@
 )
 from backend.app.lineage_ingestion import records_from_source_posts
 from lineageweave.adjudication_client import AdjudicationClient
+from lineageweave.embedding_client import EmbeddingClient
 from lineageweave.http_client import HttpClientError, post_json
 from lineageweave.lineage_persistence import lineage_edge_specs
 from lineageweave.models import Edge
@@ -559,6 +560,7 @@ async def deliver_queued_analysis_run(
     affiliated_entity_ids: list[str],
     tepp_client: TeppClient | None = None,
     adjudication_client: AdjudicationClient | None = None,
+    embedding_client: EmbeddingClient | None = None,
     valkey_stream_entry_id: str | None = None,
 ) -> dict[str, Any]:
     """Claim the outbox row and finish ThreadWeave or TEPP.
@@ -635,6 +637,7 @@ async def deliver_queued_analysis_run(
                 locked=outbox,
                 affiliated_entity_ids=affiliated_entity_ids,
                 adjudication_client=adjudication_client,
+                embedding_client=embedding_client,
             )
         finished = datetime.now(timezone.utc)
         if finished < now:
@@ -662,6 +665,7 @@ async def start_pending_analysis_run(
     affiliated_entity_ids: list[str],
     tepp_client: TeppClient | None = None,
     adjudication_client: AdjudicationClient | None = None,
+    embedding_client: EmbeddingClient | None = None,
     valkey_stream_entry_id: str | None = None,
 ) -> dict[str, Any]:
     """Enqueue then deliver on one connection.
@@ -686,6 +690,7 @@ async def start_pending_analysis_run(
         affiliated_entity_ids=affiliated_entity_ids,
         tepp_client=tepp_client,
         adjudication_client=adjudication_client,
+        embedding_client=embedding_client,
         valkey_stream_entry_id=valkey_stream_entry_id,
     )
 
@@ -697,6 +702,7 @@ async def _deliver_lineage_reconstruction(
     locked: asyncpg.Record,
     affiliated_entity_ids: list[str],
     adjudication_client: AdjudicationClient | None = None,
+    embedding_client: EmbeddingClient | None = None,
 ) -> None:
     """Persist ThreadWeave parent choices for the frozen bag."""
     now = datetime.now(timezone.utc)
@@ -713,7 +719,9 @@ async def _deliver_lineage_reconstruction(
             knowledge_cutoff=locked["knowledge_cutoff"],
             affiliated_entity_ids=affiliated_entity_ids,
         )
-    edges = lineage_edge_specs(records_from_source_posts(rows), llm=adjudication_client)
+    edges = lineage_edge_specs(
+        records_from_source_posts(rows), llm=adjudication_client, embedding=embedding_client
+    )
     digest = reconstruction_result_digest(edges)
     finished = datetime.now(timezone.utc)
     if finished < now:

backend/app/analysis_run_worker.py

@@ -14,6 +14,7 @@
 from uuid import UUID
 
 from lineageweave.adjudication_client import AdjudicationClient
+from lineageweave.embedding_client import EmbeddingClient
 from lineageweave.tepp_client import TeppClient
 
 from backend.app.analysis_run_outbox import OUTBOX_STREAM_KEY
@@ -29,6 +30,7 @@ async def consume_analysis_run_stream_once(
     last_id: str,
     tepp_client: TeppClient,
     adjudication_client: AdjudicationClient,
+    embedding_client: EmbeddingClient,
 ) -> str:
     """Consume one batch and return the last inspected Valkey entry id.
 
@@ -63,6 +65,7 @@ async def consume_analysis_run_stream_once(
                                     affiliated_entity_ids=[],
                                     tepp_client=tepp_client,
                                     adjudication_client=adjudication_client,
+                                    embedding_client=embedding_client,
                                     valkey_stream_entry_id=str(entry_id),
                                 )
                 except Exception:  # noqa: BLE001 - one bad delivery must not kill the worker task; the run stays retryable.
@@ -79,6 +82,7 @@ async def run_analysis_run_worker(
     *,
     tepp_client: TeppClient,
     adjudication_client: AdjudicationClient,
+    embedding_client: EmbeddingClient,
 ) -> None:
     """Run the single-process wake-up consumer until task cancellation."""
     last_id = "0-0"
@@ -89,4 +93,5 @@ async def run_analysis_run_worker(
             last_id=last_id,
             tepp_client=tepp_client,
             adjudication_client=adjudication_client,
+            embedding_client=embedding_client,
         )

backend/app/lineage_ingestion.py

@@ -19,6 +19,7 @@
 
 from backend.app.post_eligibility import SOURCE_POST_ELIGIBILITY_SQL
 from lineageweave.adjudication_client import AdjudicationClient
+from lineageweave.embedding_client import EmbeddingClient
 from lineageweave.lineage_persistence import lineage_edge_specs
 from lineageweave.models import Edge, Record
 
@@ -92,6 +93,7 @@ async def rebuild_lineage(
     conn: asyncpg.Connection,
     *,
     adjudication_client: AdjudicationClient | None = None,
+    embedding_client: EmbeddingClient | None = None,
 ) -> list[Edge]:
     """Reconstruct lineage for every ``source_post`` and persist the edges.
 
@@ -105,6 +107,11 @@ async def rebuild_lineage(
     (``DEFAULT_CHANNEL_WEIGHTS``, the only channel that reasons about
     content instead of approximating it, ADR 0064).
 
+    ``embedding_client`` defaults to ``None`` the same way -- without it the
+    ``text`` channel falls back to difflib character overlap, which cannot
+    tell two topically unrelated posts apart from two paraphrases of the
+    same one (ADR 0190).
+
     Reconstruction runs in a worker thread because the published adjudication
     adapter is synchronous. A session advisory lock serializes rebuilds without
     holding a transaction across provider latency. Before publication, a short
@@ -119,7 +126,7 @@ async def rebuild_lineage(
             rows = await conn.fetch(_SOURCE_RECORDS_SQL)
             records = records_from_source_posts(rows)
             edges = await asyncio.to_thread(
-                lineage_edge_specs, records, llm=adjudication_client
+                lineage_edge_specs, records, llm=adjudication_client, embedding=embedding_client
             )
             async with conn.transaction():
                 await conn.execute("lock table source_post in share mode")

backend/app/main.py

@@ -211,6 +211,7 @@ async def lifespan(app: FastAPI):
                 settings.tepp_api_key,
             ),
             adjudication_client=_adjudication_client(),
+            embedding_client=_embedding_client(),
         )
     )
     app.state.post_content_worker = asyncio.create_task(
@@ -1122,7 +1123,9 @@ async def rebuild_lineage_graph(
     try:
         async with pool.acquire() as conn:
             edges = await rebuild_lineage(
-                conn, adjudication_client=_adjudication_client()
+                conn,
+                adjudication_client=_adjudication_client(),
+                embedding_client=_embedding_client(),
             )
     except LineageSourceChangedError as exc:
         raise HTTPException(
@@ -3060,6 +3063,7 @@ async def start_analysis_run(
                         settings.tepp_api_key,
                     ),
                     adjudication_client=_adjudication_client(),
+                    embedding_client=_embedding_client(),
                     valkey_stream_entry_id=stream_id,
                 )
             except AnalysisRunStartError as exc:

docs/adr/0190-lineage-text-channel-embedding-swap.md

@@ -0,0 +1,113 @@
+# ADR 0190: Swap the lineage `text` channel to real embeddings when available
+
+- Status: Accepted
+- Date: 2026-08-24
+
+## Context
+
+A reader reported Event Lineage entangling topically and causally unrelated
+posts. Root-caused against a live corpus (not a hypothetical): the reported
+post's `thread_group_key` and `secondary_grouping_key` were both empty, so
+`reconstruct_group_key` (`backend/app/lineage_ingestion.py`) fell back to
+`process_unit_id` -- an entire team's inbox, over 21,000 posts spanning
+three-plus years, not a thread. That coarse-group fallback is intentional,
+documented design (it must match `GET /api/lineage`'s own display grouping;
+see ADR 0064, ADR 0143) and is out of scope here.
+
+Within that group, `reconstruct()`'s `DEFAULT_CANDIDATE_WINDOW` (50
+temporally-preceding records) drew candidates from a topically unbounded
+pool, and the `text` channel -- `lineageweave/channels.py`'s
+`text_similarity_score` -- is `difflib.SequenceMatcher` character overlap on
+raw titles, a channel the module's own docstring already named as a
+temporary stand-in ("swap in `EmbeddingClient` + cosine similarity ... once
+an embedding provider is configured"). With no `AdjudicationClient`
+configured for the corpus-wide rebuild path (a separate, independently
+tracked gap), the `llm` channel drops out and `temporal`/`secondary_key`/
+`text` renormalize to 0.25/0.25/0.50 (`active_weights`,
+`DEFAULT_CHANNEL_WEIGHTS`). Two short titles that happen to share common
+words or sentence structure -- e.g. two differently-topiced status updates
+with the same subject/verb shape -- can clear a 0.8+ difflib ratio despite
+being about unrelated topics; combined with temporal proximity, that alone
+clears `DEFAULT_MIN_FUSED_SCORE` (0.3) and produces a spurious parent-child
+edge. `tests/test_reconstruct.py::test_embedding_channel_overrides_a_difflib_false_positive`
+reproduces this synthetically and asserts the fix.
+
+This environment already has `LLM_GATEWAY_EMBEDDING_MODEL` configured and
+`lineageweave.embedding_client.ContextualOrchestratorEmbeddingClient` already
+in production use for post-content search embeddings
+(`lineageweave/post_content_persistence.py`) -- the embedding channel this
+ADR wires was staged capability, not new infrastructure.
+
+## Decision
+
+- `reconstruct()` gains an `embedding: EmbeddingClient | None = None`
+  parameter, mirroring `llm`'s existing shape: `None` maps to
+  `NullEmbeddingClient` (channel absent, never faked), matching ADR 0064's
+  "a missing channel is dropped ... it is never replaced with a fabricated
+  score" rule.
+- Unlike `llm` (a genuine per-candidate-pair judgment), embeddings are
+  precomputed **once per reconstruction**, batched, before scoring begins --
+  `_embed_labels` embeds every input record's label up front (bounded
+  batches, same 64-record/24,000-character philosophy as
+  `post_content_persistence.py`'s LLM batching) into a `record_id -> vector`
+  map. `_best_parent` then scores the `text` channel as
+  `cosine_similarity(candidate_vector, record_vector)` when both records
+  have a vector, falling back to `text_similarity_score`'s difflib ratio
+  otherwise (a missing vector -- e.g. one failed batch -- degrades that pair
+  back to the pre-embedding behavior, not to a fabricated score or a hard
+  failure).
+- The `text` channel key, its 0.30 weight, and the fusion code in
+  `reconstruct.py` are unchanged -- only the score's *source* changes,
+  exactly as the pre-existing docstring specified. No new channel, no weight
+  rebalancing.
+- `embedding` is threaded through the same call chain `llm` already uses:
+  `lineage_edge_specs` (`lineageweave/lineage_persistence.py`) ->
+  `rebuild_lineage` (`backend/app/lineage_ingestion.py`) ->
+  `deliver_queued_analysis_run`/`_deliver_lineage_reconstruction`
+  (`backend/app/analysis_run_start.py`) ->
+  `consume_analysis_run_stream_once`/`run_analysis_run_worker`
+  (`backend/app/analysis_run_worker.py`) -> `backend/app/main.py`'s
+  `lifespan()` worker task, `POST /api/lineage/rebuild`, and the
+  analysis-run start endpoint, all now passing the existing
+  `_embedding_client()` factory alongside `_adjudication_client()`.
+  `scripts/import_postgresql_posts.py` passes the `embedding_client` it
+  already builds for post-content embedding into `rebuild_lineage` too.
+
+## Consequences
+
+- When an embedding provider is configured (already true in this
+  environment), Event Lineage edges within a large, coarsely-grouped bucket
+  are judged on semantic similarity instead of character overlap, directly
+  reducing the reported entanglement failure mode.
+- No behavior change when no embedding provider is configured -- the
+  `text` channel is exactly what it was before this ADR (difflib), so this
+  is a pure addition, not a breaking change to any environment without
+  `LLM_GATEWAY_EMBEDDING_MODEL` set.
+- A corpus-wide rebuild now makes one batched embedding call per
+  `_EMBEDDING_BATCH_MAX_RECORDS` (64) records in addition to existing work,
+  bounded and logged-through the same provider-failure handling pattern
+  already established for post-content embedding (a failed batch is an
+  absent signal, not an aborted rebuild).
+- Does not address the separate, larger architectural question of whether
+  `process_unit_id` should ever be the grouping fallback for a thread-shaped
+  UI -- that decision needs its own ADR and is out of scope here (see
+  `docs/product-technical-gap-baseline.md`). This ADR narrows how often a
+  *wrong* group produces a *visibly wrong* edge; it does not shrink the
+  group itself.
+- Independent of, and does not duplicate, the separate open PR wiring
+  `AdjudicationClient` into `rebuild_lineage()` (restores the `llm`
+  channel's 0.40 weight for the corpus-wide rebuild path specifically) --
+  that PR and this ADR fix two different channels of the same fusion and
+  compose without conflict.
+
+## References — APA 7th
+
+Reimers, N., & Gurevych, I. (2019). Sentence-BERT: Sentence embeddings using
+Siamese BERT-networks. In *Proceedings of the 2019 Conference on Empirical
+Methods in Natural Language Processing and the 9th International Joint
+Conference on Natural Language Processing (EMNLP-IJCNLP)* (pp. 3982-3992).
+Association for Computational Linguistics. https://doi.org/10.18653/v1/D19-1410
+
+Christen, P. (2012). *Data matching: Concepts and techniques for record
+linkage, entity resolution, and duplicate detection*. Springer.
+https://doi.org/10.1007/978-3-642-31164-2

docs/lineage-bi-research-notes.md

@@ -159,7 +159,7 @@ nobody can tell which paragraph it illustrated.
 |---|---|---|
 | `temporal` | Prefers a candidate parent closer in time | Standard recency prior in event/story linking (Allan, 2002) |
 | `secondary_key` | Rewards a shared finer-grained key (e.g. project code) | Coarse blocking key, the standard first step in record-linkage pipelines (Fellegi & Sunter, 1969; Christen, 2012) |
-| `text` | String-similarity between record labels | A dependency-free stand-in for an embedding-cosine channel (`lineageweave.embedding_client`); same role, cheaper, swappable without touching the fusion code |
+| `text` | Cosine similarity between record-label embeddings (`lineageweave.embedding_client`) when a provider is configured; falls back to `difflib.SequenceMatcher` character overlap otherwise | Embedding-based semantic similarity is the standard replacement for lexical overlap in modern record linkage/entity resolution (Reimers & Gurevych, 2019). The difflib fallback (ADR 0064's original stand-in) has no notion of meaning: two records about different topics that happen to share common words or structure -- e.g. two status-update titles differing only in subject -- can score a high character-overlap ratio despite being unrelated, which is exactly the failure mode ADR 0190 root-caused against a live corpus |
 | `llm` | An LLM's judged confidence that one record follows from another | Optional, highest-weighted when available -- see "LLM adjudication" below |
 
 Channel scores are fused with a **weighted convex combination**
@@ -233,6 +233,8 @@ Radford, A., Kim, J. W., Hallacy, C., Ramesh, A., Goh, G., Agarwal, S., Sastry,
 
 Raudenbush, S. W., & Bryk, A. S. (2002). *Hierarchical linear models: Applications and data analysis methods* (2nd ed.). Sage Publications.
 
+Reimers, N., & Gurevych, I. (2019). Sentence-BERT: Sentence embeddings using Siamese BERT-networks. In *Proceedings of the 2019 Conference on Empirical Methods in Natural Language Processing and the 9th International Joint Conference on Natural Language Processing (EMNLP-IJCNLP)* (pp. 3982-3992). Association for Computational Linguistics. https://doi.org/10.18653/v1/D19-1410
+
 Resnick, P. (2008). *Internet Message Format* (RFC 5322). IETF. https://doi.org/10.17487/RFC5322
 
 See, A., Liu, P. J., & Manning, C. D. (2017). Get to the point: Summarization with pointer-generator networks. In *Proceedings of the 55th Annual Meeting of the Association for Computational Linguistics* (Vol. 1, pp. 1073-1083). Association for Computational Linguistics. https://doi.org/10.18653/v1/P17-1099

lineageweave/lineage_persistence.py

@@ -12,11 +12,17 @@
 from collections.abc import Sequence
 
 from .adjudication_client import AdjudicationClient
+from .embedding_client import EmbeddingClient
 from .models import Edge, Record
 from .reconstruct import reconstruct
 
 
-def lineage_edge_specs(records: Sequence[Record], *, llm: AdjudicationClient | None = None) -> list[Edge]:
+def lineage_edge_specs(
+    records: Sequence[Record],
+    *,
+    llm: AdjudicationClient | None = None,
+    embedding: EmbeddingClient | None = None,
+) -> list[Edge]:
     """Run reconstruct and return every resulting parent→child edge.
 
     Callers persist these as ``post_lineage_edge`` rows. Record ids must
@@ -29,6 +35,12 @@ def lineage_edge_specs(records: Sequence[Record], *, llm: AdjudicationClient | N
     (the llm channel is then dropped and the rest renormalized, not
     faked) -- callers that want the highest-weighted reasoning channel
     actually contributing to real reconstructions must pass a real one.
+
+    ``embedding`` defaults to ``None``, which ``reconstruct()`` treats as
+    the unavailable :class:`~lineageweave.embedding_client.NullEmbeddingClient`
+    (the ``text`` channel then falls back to its difflib stand-in) --
+    callers that want real semantic text matching instead of character
+    overlap must pass a real one.
     """
-    trees = reconstruct(list(records), llm=llm)
+    trees = reconstruct(list(records), llm=llm, embedding=embedding)
     return [edge for tree in trees for edge in tree.edges]