diff --git a/CHANGELOG.d/2.24.0-lineage-text-channel-embeddings.md b/CHANGELOG.d/2.24.0-lineage-text-channel-embeddings.md new file mode 100644 index 000000000..f010672dc --- /dev/null +++ b/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`. diff --git a/CHANGELOG.md b/CHANGELOG.md index 21a01395b..efedae955 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/backend/app/analysis_run_start.py b/backend/app/analysis_run_start.py index 2387d940b..8b8a03420 100644 --- a/backend/app/analysis_run_start.py +++ b/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: diff --git a/backend/app/analysis_run_worker.py b/backend/app/analysis_run_worker.py index 592de6439..721fe9d18 100644 --- a/backend/app/analysis_run_worker.py +++ b/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, ) diff --git a/backend/app/lineage_ingestion.py b/backend/app/lineage_ingestion.py index 878839df4..9f11d89f2 100644 --- a/backend/app/lineage_ingestion.py +++ b/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") diff --git a/backend/app/main.py b/backend/app/main.py index 9f620be2a..b1b560b21 100644 --- a/backend/app/main.py +++ b/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: diff --git a/docs/adr/0190-lineage-text-channel-embedding-swap.md b/docs/adr/0190-lineage-text-channel-embedding-swap.md new file mode 100644 index 000000000..d5d62e3fc --- /dev/null +++ b/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 diff --git a/docs/lineage-bi-research-notes.md b/docs/lineage-bi-research-notes.md index 94f99d73d..6db61b3b3 100644 --- a/docs/lineage-bi-research-notes.md +++ b/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 diff --git a/lineageweave/lineage_persistence.py b/lineageweave/lineage_persistence.py index 8b9f4f16d..8fc774291 100644 --- a/lineageweave/lineage_persistence.py +++ b/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] diff --git a/lineageweave/reconstruct.py b/lineageweave/reconstruct.py index e0a6f89d0..d2a009461 100644 --- a/lineageweave/reconstruct.py +++ b/lineageweave/reconstruct.py @@ -18,6 +18,7 @@ from .adjudication_client import AdjudicationClient, NullAdjudicationClient from .channels import secondary_key_match_score, temporal_score, text_similarity_score +from .embedding_client import EmbeddingClient, NullEmbeddingClient, cosine_similarity from .models import Edge, Record, Tree # Channel weights when every channel is available. llm gets the most weight @@ -41,6 +42,12 @@ # candidates are plausible, which is wrong more often than it is right. DEFAULT_MIN_FUSED_SCORE = 0.3 +# ponytail: same batch-size philosophy as post_content_persistence's LLM +# batching -- a corpus-wide rebuild is tens of thousands of records, and +# provider embedding endpoints cap request size/latency per call. +_EMBEDDING_BATCH_MAX_RECORDS = 64 +_EMBEDDING_BATCH_MAX_CHARS = 24_000 + def active_weights( llm: AdjudicationClient, weights: dict[str, float] = DEFAULT_CHANNEL_WEIGHTS @@ -67,6 +74,7 @@ def _best_parent( llm: AdjudicationClient, weights: dict[str, float], min_score: float, + vectors: dict[str, list[float]], ) -> tuple[Record, float, dict[str, float]] | None: """Implement the _best_parent operation for this channel.""" if not candidates: @@ -76,11 +84,22 @@ def _best_parent( channel_results["llm"] = [] per_candidate_scores: dict[str, dict[str, float]] = defaultdict(dict) + record_vector = vectors.get(record.record_id) for candidate in candidates: + candidate_vector = vectors.get(candidate.record_id) + # A real embedding pair always wins over the difflib stand-in + # (channels.py's own docstring names this the intended swap); a + # candidate missing a vector (failed embedding batch) still gets a + # signal instead of silently losing the whole text channel. + text_score = ( + cosine_similarity(candidate_vector, record_vector) + if candidate_vector is not None and record_vector is not None + else text_similarity_score(candidate, record) + ) scores = { "temporal": temporal_score(candidate, record), "secondary_key": secondary_key_match_score(candidate, record), - "text": text_similarity_score(candidate, record), + "text": text_score, } if "llm" in weights: scores["llm"] = llm.judge(candidate.label, record.label) @@ -102,6 +121,7 @@ def _reconstruct_group( weights: dict[str, float], window: int, min_score: float, + vectors: dict[str, list[float]], ) -> tuple[list[tw.Container], list[Edge]]: """Implement the _reconstruct_group operation for this channel.""" ordered = sorted(records, key=lambda r: r.occurred_at) @@ -109,7 +129,7 @@ def _reconstruct_group( edges: list[Edge] = [] for index, record in enumerate(ordered): candidates = ordered[max(0, index - window) : index] - parent_choice = _best_parent(record, candidates, llm, weights, min_score) + parent_choice = _best_parent(record, candidates, llm, weights, min_score, vectors) references: list[str] = [] if parent_choice is not None: parent, score, channel_scores = parent_choice @@ -126,6 +146,48 @@ def _reconstruct_group( return tw.thread_messages(messages), edges +def _embed_labels(records: list[Record], embedding_client: EmbeddingClient) -> dict[str, list[float]]: + """Batch-embed every record's label once, up front. + + A failed batch yields no vectors for its records rather than raising -- + ``_best_parent`` falls back to ``text_similarity_score`` for any record + without a vector, so a provider hiccup degrades the text channel back to + its difflib stand-in instead of failing the whole reconstruction. + """ + vectors: dict[str, list[float]] = {} + embed_many = getattr(embedding_client, "embed_many", None) + batches: list[list[Record]] = [] + batch: list[Record] = [] + batch_chars = 0 + for record in records: + label_chars = len(record.label) + if batch and ( + len(batch) >= _EMBEDDING_BATCH_MAX_RECORDS + or batch_chars + label_chars > _EMBEDDING_BATCH_MAX_CHARS + ): + batches.append(batch) + batch = [] + batch_chars = 0 + batch.append(record) + batch_chars += label_chars + if batch: + batches.append(batch) + + for group in batches: + try: + embedded = ( + embed_many([record.label for record in group]) + if callable(embed_many) + else [embedding_client.embed(record.label) for record in group] + ) + except (OSError, RuntimeError, ValueError): + continue + for record, vector in zip(group, embedded, strict=True): + if isinstance(vector, list) and vector: + vectors[record.record_id] = vector + return vectors + + def _walk(roots: list[tw.Container]) -> tuple[list[str], dict[str, list[str]]]: """``thread_messages`` returns only the root Containers, each nesting its descendants under ``.children`` -- walk the whole forest to recover every @@ -147,6 +209,7 @@ def reconstruct( records: list[Record], *, llm: AdjudicationClient | None = None, + embedding: EmbeddingClient | None = None, weights: dict[str, float] = DEFAULT_CHANNEL_WEIGHTS, candidate_window: int = DEFAULT_CANDIDATE_WINDOW, min_fused_score: float = DEFAULT_MIN_FUSED_SCORE, @@ -158,6 +221,14 @@ def reconstruct( llm: adjudication channel client; defaults to :class:`~lineageweave.adjudication_client.NullAdjudicationClient` (the llm channel is then dropped, not faked). + embedding: embedding channel client for the ``text`` channel; + defaults to :class:`~lineageweave.embedding_client.NullEmbeddingClient` + (the ``text`` channel then falls back to + :func:`~lineageweave.channels.text_similarity_score`'s difflib + stand-in, not a fabricated cosine score). When available, every + record's label is embedded once up front (batched), not per + candidate pair -- unlike ``llm``, which is genuinely a per-pair + judgment. weights: per-channel fusion weights before llm-availability renormalization; see :data:`DEFAULT_CHANNEL_WEIGHTS`. candidate_window: how many recent prior records to consider as @@ -166,11 +237,13 @@ def reconstruct( instead of a forced weak match; see :data:`DEFAULT_MIN_FUSED_SCORE`. """ llm = llm or NullAdjudicationClient() + embedding = embedding or NullEmbeddingClient() active = active_weights(llm, weights) + vectors = _embed_labels(records, embedding) if embedding.available else {} trees: list[Tree] = [] for group_key, group_records in _group_by(records).items(): root_containers, edges = _reconstruct_group( - group_records, llm, active, candidate_window, min_fused_score + group_records, llm, active, candidate_window, min_fused_score, vectors ) record_by_id = {record.record_id: record for record in group_records} roots, children_of = _walk(root_containers) diff --git a/pyproject.toml b/pyproject.toml index cb4be2916..f4c04f1fd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "lineageweave" -version = "2.12.6" +version = "2.24.0" description = "Reconstructs git-branch-style lineage DAGs from scattered short records using multi-channel score fusion and LLM adjudication." readme = "README.md" license = { text = "MIT" } diff --git a/scripts/import_postgresql_posts.py b/scripts/import_postgresql_posts.py index fa2bedbdb..e606c4c1a 100644 --- a/scripts/import_postgresql_posts.py +++ b/scripts/import_postgresql_posts.py @@ -541,7 +541,9 @@ async def import_rows(args: argparse.Namespace) -> dict[str, int]: ) imported += 1 cleanup = await cleanup_synthetic_seed(target, apply=True) - edges = await rebuild_lineage(target, adjudication_client=adjudication_client) + edges = await rebuild_lineage( + target, adjudication_client=adjudication_client, embedding_client=embedding_client + ) return { "source_rows": len(rows), "imported_rows": imported, diff --git a/tests/test_analysis_run_worker.py b/tests/test_analysis_run_worker.py index e59783db1..88d140c6f 100644 --- a/tests/test_analysis_run_worker.py +++ b/tests/test_analysis_run_worker.py @@ -6,6 +6,7 @@ from backend.app import analysis_run_worker from lineageweave.adjudication_client import NullAdjudicationClient +from lineageweave.embedding_client import NullEmbeddingClient from lineageweave.tepp_client import TeppClient @@ -71,6 +72,7 @@ async def fake_deliver(conn, **kwargs): last_id="0-0", tepp_client=TeppClient(), adjudication_client=NullAdjudicationClient(), + embedding_client=NullEmbeddingClient(), ) assert last_id == "1-1" @@ -81,6 +83,7 @@ async def fake_deliver(conn, **kwargs): "affiliated_entity_ids": [], "tepp_client": calls[0]["tepp_client"], "adjudication_client": calls[0]["adjudication_client"], + "embedding_client": calls[0]["embedding_client"], "valkey_stream_entry_id": "1-0", } ] @@ -105,6 +108,7 @@ async def fake_deliver(conn, **kwargs): last_id="0-0", tepp_client=TeppClient(), adjudication_client=NullAdjudicationClient(), + embedding_client=NullEmbeddingClient(), ) assert last_id == "1-1" @@ -142,6 +146,7 @@ async def unexpected_delivery(*_args, **_kwargs): last_id="0-0", tepp_client=TeppClient(), adjudication_client=NullAdjudicationClient(), + embedding_client=NullEmbeddingClient(), ) assert last_id == "1-1" @@ -173,6 +178,7 @@ async def consume_once(*_args, last_id, **_kwargs): _Pool(), tepp_client=TeppClient(), adjudication_client=NullAdjudicationClient(), + embedding_client=NullEmbeddingClient(), ) assert observed == ["0-0"] diff --git a/tests/test_lineage_ingestion.py b/tests/test_lineage_ingestion.py index 8c822db86..53e233eed 100644 --- a/tests/test_lineage_ingestion.py +++ b/tests/test_lineage_ingestion.py @@ -275,7 +275,7 @@ def test_rebuild_lineage_passes_the_adjudication_client_through(monkeypatch: pyt """ captured: dict[str, object] = {} - def fake_lineage_edge_specs(records, *, llm=None): + def fake_lineage_edge_specs(records, *, llm=None, embedding=None): captured["llm"] = llm return [] @@ -288,6 +288,26 @@ def fake_lineage_edge_specs(records, *, llm=None): assert captured["llm"] is sentinel_client +def test_rebuild_lineage_passes_the_embedding_client_through(monkeypatch: pytest.MonkeyPatch) -> None: + """Same class of bug as the adjudication client above, for the ``text`` + channel's embedding source (ADR 0190): rebuild_lineage() must forward + its embedding_client all the way to lineage_edge_specs unchanged. + """ + captured: dict[str, object] = {} + + def fake_lineage_edge_specs(records, *, llm=None, embedding=None): + captured["embedding"] = embedding + return [] + + monkeypatch.setattr( + "backend.app.lineage_ingestion.lineage_edge_specs", fake_lineage_edge_specs + ) + + sentinel_client = object() + asyncio.run(rebuild_lineage(_RebuildConnection(), embedding_client=sentinel_client)) + assert captured["embedding"] is sentinel_client + + def test_rebuild_lineage_defaults_to_no_adjudication_client(monkeypatch: pytest.MonkeyPatch) -> None: """The default stays None (lineage_edge_specs/reconstruct's own documented fallback) -- this only guards that rebuild_lineage's new @@ -295,7 +315,7 @@ def test_rebuild_lineage_defaults_to_no_adjudication_client(monkeypatch: pytest. """ captured: dict[str, object] = {"llm": "unset"} - def fake_lineage_edge_specs(records, *, llm=None): + def fake_lineage_edge_specs(records, *, llm=None, embedding=None): captured["llm"] = llm return [] @@ -321,9 +341,9 @@ def test_rebuild_lineage_keeps_blocking_adjudication_outside_event_loop_and_tran provider_release = threading.Event() connection = _RebuildConnection() - def blocking_lineage_edge_specs(records, *, llm=None): + def blocking_lineage_edge_specs(records, *, llm=None, embedding=None): """Model a synchronous contextual-orchestrator request.""" - del records, llm + del records, llm, embedding connection.events.append("reconstruct") provider_entered.set() assert provider_release.wait(timeout=2), ( @@ -391,9 +411,9 @@ def test_rebuild_lineage_recomputes_a_source_snapshot_changed_during_adjudicatio ) reconstructed_titles: list[list[str]] = [] - def record_reconstruction(records, *, llm=None): + def record_reconstruction(records, *, llm=None, embedding=None): """Record which synthetic source version reached reconstruction.""" - del llm + del llm, embedding reconstructed_titles.append([record.label for record in records]) return [] diff --git a/tests/test_reconstruct.py b/tests/test_reconstruct.py index 6ae195941..03ec36c27 100644 --- a/tests/test_reconstruct.py +++ b/tests/test_reconstruct.py @@ -1,5 +1,6 @@ from __future__ import annotations +import difflib from datetime import datetime from lineageweave import Record, reconstruct @@ -76,3 +77,74 @@ def test_candidate_window_bounds_which_priors_are_considered() -> None: parent_index = int(edge.parent_id[1:]) child_index = int(edge.child_id[1:]) assert child_index - parent_index == 1 + + +class _StubEmbeddingClient: + """Maps each known label to a fixed vector so a test controls cosine + similarity directly, instead of depending on a real provider's opinion.""" + + available = True + + def __init__(self, vectors_by_label: dict[str, list[float]]) -> None: + self._vectors_by_label = vectors_by_label + self.calls = 0 + + def embed(self, text: str) -> list[float]: + self.calls += 1 + return self._vectors_by_label[text] + + +def test_embedding_channel_is_used_and_scored_when_a_client_is_supplied() -> None: + stub = _StubEmbeddingClient({record.label: [1.0, 0.0] for record in sample_records()}) + trees = reconstruct(sample_records(), embedding=stub) + tree_a = next(t for t in trees if t.group_key == "A-100") + + assert stub.calls > 0 + assert all("text" in edge.channel_scores for edge in tree_a.edges) + # Every label maps to the identical vector here, so cosine similarity is + # exactly 1.0 for every pair -- distinct from whatever difflib would + # have scored the same real (non-identical) titles. + assert all(edge.channel_scores["text"] == 1.0 for edge in tree_a.edges) + + +def test_embedding_channel_overrides_a_difflib_false_positive() -> None: + """Reproduces the reported production bug: two posts about different + topics ('budget' vs 'safety') share enough words and structure that + difflib.SequenceMatcher alone scores them a 0.86 character-overlap + ratio -- high enough, combined with temporal closeness, to clear + DEFAULT_MIN_FUSED_SCORE and force a spurious parent-child Event Lineage + edge between unrelated posts (see docs/product-technical-gap-baseline.md). + A real embedding channel judging the two topics as dissimilar (mapped + here to opposite unit vectors, cosine similarity 0.0) must be able to + keep them apart instead. + """ + budget_label = "Quarterly budget review for the northern region team" + safety_label = "Quarterly safety review for the northern region plant" + records = [ + Record("r1", "G", budget_label, datetime(2026, 1, 1, 0, 0), ""), + Record("r2", "G", safety_label, datetime(2026, 1, 1, 1, 0), ""), + ] + + assert difflib.SequenceMatcher(None, budget_label, safety_label).ratio() > 0.8 + + without_embedding = reconstruct(records) + assert without_embedding[0].edges, ( + "sanity check: difflib's character overlap alone should reproduce " + "the false-positive link this test's embedding channel must prevent" + ) + + stub = _StubEmbeddingClient({budget_label: [1.0, 0.0], safety_label: [-1.0, 0.0]}) + with_embedding = reconstruct(records, embedding=stub) + assert with_embedding[0].edges == [] + assert "r2" in with_embedding[0].roots + + +def test_embedding_channel_is_dropped_not_faked_when_unavailable() -> None: + trees = reconstruct(sample_records()) + tree_a = next(t for t in trees if t.group_key == "A-100") + + # No embedding client was supplied, so every "text" score must have come + # from text_similarity_score's difflib stand-in, never a fabricated + # cosine similarity -- there is nothing to assert this *against* beyond + # the channel simply working exactly as it did before this client existed. + assert all("text" in edge.channel_scores for edge in tree_a.edges)