diff --git a/backend/app/lineage_ingestion.py b/backend/app/lineage_ingestion.py index 21b8e88d7..ec069bee5 100644 --- a/backend/app/lineage_ingestion.py +++ b/backend/app/lineage_ingestion.py @@ -72,6 +72,31 @@ async def persist_lineage_edges(conn: asyncpg.Connection, edges: list[Edge]) -> ) +async def load_estimated_channel_weights( + conn: asyncpg.Connection, active_channels: set[str] +) -> dict[str, float] | None: + """Persisted psychometric weights, only on an exact channel-set match. + + ADR 0145: a partial overlap would mix estimated and hand-picked + weights into a vector that is neither grounded nor the documented + fallback -- so anything other than an exact match falls back + entirely (return ``None``). A database that has not applied + migration 0135 yet (rollout ordering, rollback) is the same "no + estimate persisted" state, not an error -- rebuilds keep working on + the fallback constants. + """ + try: + rows = await conn.fetch( + "select channel_code, weight_value from lineage_channel_weight" + ) + except asyncpg.UndefinedTableError: + return None + persisted = {row["channel_code"]: float(row["weight_value"]) for row in rows} + if not persisted or set(persisted) != active_channels: + return None + return persisted + + async def rebuild_lineage(conn: asyncpg.Connection) -> list[Edge]: """Reconstruct lineage for every ``source_post`` and persist the edges.""" rows = await conn.fetch( @@ -79,7 +104,13 @@ async def rebuild_lineage(conn: asyncpg.Connection) -> list[Edge]: "process_unit_id, thread_group_key, secondary_grouping_key " f"from source_post where {SOURCE_POST_ELIGIBILITY_SQL.format(alias='source_post')}" ) - edges = lineage_edge_specs(records_from_source_posts(rows)) + # 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). + weights = await load_estimated_channel_weights( + conn, {"temporal", "secondary_key", "text"} + ) + edges = lineage_edge_specs(records_from_source_posts(rows), weights=weights) await persist_lineage_edges(conn, edges) return edges diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index e7b94fe38..bb4603cda 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -169,6 +169,11 @@ / "migrations" / "0114_semantic_relationship_standard_predicates.sql" ) +_CHANNEL_WEIGHT_MIGRATION = ( + Path(__file__).resolve().parents[2] + / "migrations" + / "0135_lineage_channel_weight.sql" +) def _postgres_available() -> bool: @@ -294,6 +299,7 @@ def seeded_db(demo_analyst_token): cur.execute(_SOFTWARE_AGENT_MIGRATION.read_text()) cur.execute(_SEMANTIC_RELATIONSHIP_MIGRATION.read_text()) cur.execute(_SEMANTIC_RELATIONSHIP_PREDICATES_MIGRATION.read_text()) + cur.execute(_CHANNEL_WEIGHT_MIGRATION.read_text()) cur.execute( "insert into common_lookup_value (lookup_category, lookup_code, lookup_label) values " "('corporate_entity_level', 'group', 'Group'), " diff --git a/docker/postgres-init/migrate.sh b/docker/postgres-init/migrate.sh index 1edc94feb..7dcdec69b 100644 --- a/docker/postgres-init/migrate.sh +++ b/docker/postgres-init/migrate.sh @@ -18,7 +18,7 @@ for migration in /opt/lineageweave/migrations/*.sql; do migration_name=${migration##*/} case "$migration_name" in 0012_*|0013_*|0014_*|0015_*|0016_*|0017_*|0018_*|0019_*|0020_*|0021_*|0022_*|0023_*|0024_*|0025_*|0026_*|0027_*|0028_*|0029_*|0030_*|0031_*|0032_*|0033_*|0034_*|0035_*|0036_*|0037_*|0038_*|0039_*|0040_*|0041_*|0042_*|0043_*|0044_*|0045_*|0046_*|0047_*|0048_*|0049_*|0050_*) ;; - 0060_*|0100_*|0101_*|0102_*|0103_*|0104_*|0105_*|0106_*|0107_*|0108_*|0109_*|0110_*|0111_*|0112_*|0113_*|0114_*|0130_*|0131_*|0132_*|0133_*|0134_*) ;; + 0060_*|0100_*|0101_*|0102_*|0103_*|0104_*|0105_*|0106_*|0107_*|0108_*|0109_*|0110_*|0111_*|0112_*|0113_*|0114_*|0130_*|0131_*|0132_*|0133_*|0134_*|0135_*) ;; *) continue ;; esac printf 'Applying %s\n' "$migration_name" diff --git a/docs/adr/0145-psychometric-channel-weight-estimation.md b/docs/adr/0145-psychometric-channel-weight-estimation.md new file mode 100644 index 000000000..269e21c4f --- /dev/null +++ b/docs/adr/0145-psychometric-channel-weight-estimation.md @@ -0,0 +1,160 @@ +# ADR 0145 — Lineage channel-fusion weights come from psychometric estimation, not hand-picked constants + +**Decision status:** Proposed +**Date:** 2026-08-23 + +> Numbering note: parallel branches are assigning ADR numbers concurrently +> (0143 exists on an unmerged branch). `0145` was the next free number on +> `docs/customer-master-scope-adr` at time of writing and may need +> renumbering when branches converge. + +## Context + +`lineageweave.reconstruct` fuses four evidence channels (temporal, +secondary-key, text-similarity, llm adjudication) into one convex score +per candidate parent-child pair. The fusion weights, +`DEFAULT_CHANNEL_WEIGHTS = {"temporal": 0.15, "secondary_key": 0.15, +"text": 0.30, "llm": 0.40}`, were hand-picked: the module's own comment +justifies them with a qualitative argument ("llm ... is the only channel +that actually reasons about the content"), not with any estimate from +data. The product's standing requirement — repeated across many +sessions — is that scoring weights be grounded in published measurement +methodology through the organization's own psychometric libraries +(`fast-mlsirm`, TEPP), never asserted by fiat. + +The measurement literature gives an exact grounding. Treat each channel +as an *item* observing the latent trait "these two posts are genuinely +related", and each scored candidate pair as a *respondent*. Under the +two-parameter logistic model, the information-optimal scoring weight of +an item is proportional to its discrimination parameter (Birnbaum, 1968; +Lord, 1980) — an unweighted or arbitrarily-weighted composite discards +exactly that information (McNeish & Wolf, 2020). Pairs are nested inside +reconstruction groups (process unit / corporate entity / thread), so a +single-level fit would commit the ecological/atomistic inference error +the standing mandate calls out (Robinson, 1950); `fast-mlsirm`'s MLS2PLM +is a *multilevel* 2PL whose `factor_id` models exactly this nesting, and +its `MLSIRMParams.alpha` field is the per-item log-discrimination +(natural-scale discrimination `exp(alpha)` is positive by construction, +so normalizing to sum 1 always yields valid convex weights). + +`fast-mlsirm` is not yet published to PyPI, so LineageWeave cannot take +a hard install dependency today. This repository's established pattern +for every optional capability is fail-closed clients (Null client when +unconfigured, never a fabricated result); the same pattern applies here. + +## Decision + +1. **Estimation, not assertion.** A new module, + `lineageweave/channel_weight_estimation.py`, estimates channel + weights by fitting `fast-mlsirm`'s MLS2PLM over observed channel + scores: items = channels, respondents = candidate pairs sampled the + same way `reconstruct` forms them (same grouping, same candidate + window), `factor_id` = the pair's reconstruction group (multilevel + nesting per Robinson, 1950). Estimated weights are the normalized + natural-scale discriminations, `exp(alpha_j) / Σ exp(alpha_k)` + (Birnbaum, 1968). +2. **Dichotomization at the fusion floor.** MLS2PLM is dichotomous; + channel scores in [0, 1] are dichotomized at + `DEFAULT_MIN_FUSED_SCORE` (0.3) — the same threshold `reconstruct` + already treats as the boundary between "evidence of a link" and + "no plausible candidate", so the measurement model observes the same + binary event the fusion decision acts on. +3. **Fail closed, never fabricate.** When `fast-mlsirm` is not + importable, the sample is too small, or the fit degenerates (any + non-finite alpha), estimation returns nothing and callers keep the + documented fallback constants — now explicitly labeled as + *ungrounded fallback* in `reconstruct.py`'s docstring, not as a + justified default. +4. **Persisted, provenance-bearing weights.** An operator script + (`scripts/estimate_channel_weights.py`) runs the estimation against + the real corpus and upserts one row per channel into a new + `lineage_channel_weight` table (migration 0135) carrying the weight, + the estimation method code, the sample size, and the estimation + timestamp. `rebuild_lineage` loads these rows and passes them to + `reconstruct`; it uses them **only when the persisted channel set + exactly matches the active channel set** (no partial mixing of + estimated and hand-picked weights — a mixed vector is neither + grounded nor the documented fallback), otherwise it falls back + entirely. +5. **The llm channel is estimated only when adjudication is + configured.** Scoring sampled pairs through the adjudication client + costs provider calls; the script includes the llm channel when a + client is available and skips it otherwise (the exact-match rule in + (4) then keeps rebuilds on the fallback until a full estimate + exists). Accuracy over speed, per the standing mandate. + +## Consequences + +**Positive.** Fusion weights become an estimable, auditable quantity +with a citation trail instead of a code comment: the persisted row +records how many pairs supported the estimate and when. Re-running the +operator script after corpus growth re-calibrates the fusion without a +code change. The multilevel fit respects group nesting rather than +pooling pairs atomistically. + +**Negative.** A new optional dependency surface (fast-mlsirm via git +until it reaches PyPI) and a new operator step. Dichotomizing at 0.3 +discards within-interval score variation; a graded/continuous-response +model (fast-mlsirm ships `grm.py`/`crm.py`) is the natural upgrade once +this loop is validated end-to-end — deliberately out of scope for the +first landing. TEPP-side calibration (event-level theta as the latent +anchor) is a further integration this ADR does not attempt while TEPP +remains non-production (see the standing `tepp_readiness_watch`). + +## Rejected Alternatives + +- **Keep hand-picked constants.** The standing product requirement + explicitly forbids this; no citation supports the current 0.15/0.15/ + 0.30/0.40 split. +- **Reciprocal Rank Fusion for this surface.** RRF (Cormack et al., + 2009) is parameter-free and already grounds RankWeave's *rank* fusion + constant (η = 60), but reconstruct's decision is a thresholded + *score* over at most `candidate_window` candidates, not a deep + ranked-list merge; discarding score magnitude here would also discard + `DEFAULT_MIN_FUSED_SCORE`'s "no plausible parent" semantics. +- **Supervised weight learning (logistic regression on labeled pairs).** + There is no labeled corpus: the source system carries no ground-truth + thread links (their absence is why this library exists). IRT estimates + discriminations from response structure without per-pair labels. + +## Implementation Notes + +1. Migration `0135_lineage_channel_weight.sql`: + `lineage_channel_weight(channel_code text primary key, weight_value + double precision not null, estimation_method_code text not null, + sample_pair_count bigint not null, estimated_at timestamptz not null + default now())`, plus rollback. Two-word snake_case per ADR 0120. +2. `estimate_channel_weights()` returns `None` on: import failure, + fewer than `_MIN_SAMPLE_PAIRS` pairs, any channel with fewer than two + distinct dichotomized responses, or any non-finite estimated alpha. +3. `lineage_edge_specs` and `rebuild_lineage` gain an optional + `weights` pass-through; `None` keeps today's behavior exactly. +4. Tests: fail-closed paths run everywhere; a parameter-recovery test + (planted discriminations recovered within tolerance, the + organization's RMSE standard) runs when `fast_mlsirm` is importable + and skips honestly otherwise, same as this repo's live-service + skips. + +## References (APA 7th) + +Birnbaum, A. (1968). Some latent trait models and their use in +inferring an examinee's ability. In F. M. Lord & M. R. Novick, +*Statistical theories of mental test scores* (pp. 397–479). +Addison-Wesley. + +Cormack, G. V., Clarke, C. L. A., & Buettcher, S. (2009). Reciprocal +rank fusion outperforms Condorcet and individual rank learning methods. +*Proceedings of the 32nd International ACM SIGIR Conference on Research +and Development in Information Retrieval*, 758–759. +https://doi.org/10.1145/1571941.1572114 + +Lord, F. M. (1980). *Applications of item response theory to practical +testing problems*. Lawrence Erlbaum Associates. + +McNeish, D., & Wolf, M. G. (2020). Thinking twice about sum scores. +*Behavior Research Methods, 52*(6), 2287–2305. +https://doi.org/10.3758/s13428-020-01398-0 + +Robinson, W. S. (1950). Ecological correlations and the behavior of +individuals. *American Sociological Review, 15*(3), 351–357. +https://doi.org/10.2307/2087176 diff --git a/lineageweave/channel_weight_estimation.py b/lineageweave/channel_weight_estimation.py new file mode 100644 index 000000000..bbdf67623 --- /dev/null +++ b/lineageweave/channel_weight_estimation.py @@ -0,0 +1,134 @@ +"""Psychometric estimation of lineage channel-fusion weights (ADR 0145). + +The convex weights `reconstruct()` fuses its evidence channels with were +historically hand-picked constants. This module replaces assertion with +estimation: each channel is treated as an *item* observing the latent +trait "these two posts are genuinely related", each scored candidate +pair as a *respondent*, and the pair's reconstruction group as the +multilevel nesting factor (Robinson, 1950, on why pooling nested +observations atomistically misleads). Under the two-parameter logistic +model the information-optimal scoring weight of an item is proportional +to its discrimination (Birnbaum, 1968; Lord, 1980), so the estimated +weights are the normalized natural-scale discriminations from +`fast-mlsirm`'s multilevel 2PL (`MLS2PLM`), whose ``MLSIRMParams.alpha`` +holds per-item log-discriminations. + +Fail-closed like every optional capability in this codebase: when +`fast_mlsirm` is not importable, the sample is too small, any channel is +degenerate (fewer than two distinct dichotomized responses), or the fit +produces a non-finite estimate, :func:`estimate_channel_weights` returns +``None`` and the caller keeps the documented fallback constants -- it +never fabricates a "grounded" weight. +""" + +from __future__ import annotations + +import math +from dataclasses import dataclass + +from .reconstruct import DEFAULT_MIN_FUSED_SCORE + +# Below this many scored pairs a 2PL discrimination estimate is noise, +# not measurement -- refuse rather than persist an unstable weight. +_MIN_SAMPLE_PAIRS = 200 + + +@dataclass(frozen=True) +class ChannelWeightEstimate: + """One estimation run's convex weights plus its provenance.""" + + weights: dict[str, float] + sample_pair_count: int + estimation_method_code: str + + +def dichotomize(score: float, threshold: float = DEFAULT_MIN_FUSED_SCORE) -> int: + """Binary "evidence of a link" event at the fusion floor. + + `reconstruct` already treats ``DEFAULT_MIN_FUSED_SCORE`` as the + boundary between a plausible parent and no candidate at all, so the + measurement model observes the same event the fusion decision acts + on (ADR 0145 point 2). + """ + return 1 if score >= threshold else 0 + + +def estimate_channel_weights( + pair_channel_scores: list[dict[str, float]], + group_ids: list[int], +) -> ChannelWeightEstimate | None: + """Estimate convex fusion weights from observed channel scores. + + Args: + pair_channel_scores: one dict per candidate pair mapping every + active channel name to its score in [0, 1]. Every dict must + carry the same channel set -- a pair missing a channel is a + caller bug, not missing data to impute. + group_ids: the reconstruction-group index of each pair (same + length/order), used as MLS2PLM's multilevel ``factor_id``. + + Returns: + The estimate, or ``None`` whenever a grounded estimate cannot be + produced (fail closed -- see module docstring for the cases). + """ + if len(pair_channel_scores) != len(group_ids): + raise ValueError("pair_channel_scores and group_ids must align") + if len(pair_channel_scores) < _MIN_SAMPLE_PAIRS: + return None + channels = sorted(pair_channel_scores[0]) + if not channels: + return None + for scores in pair_channel_scores: + if sorted(scores) != channels: + raise ValueError("every pair must score the same channel set") + + responses = [ + [dichotomize(scores[channel]) for channel in channels] + for scores in pair_channel_scores + ] + for column, channel in enumerate(channels): + observed = {row[column] for row in responses} + if len(observed) < 2: + # A channel that always (or never) clears the floor carries no + # discriminating information; a 2PL slope for it is undefined + # in practice. Refuse rather than estimate around it. + return None + + try: + import numpy + from fast_mlsirm import FitConfig, fit + except ImportError: + return None + + # One latent "relatedness" trait loads every channel (factor_id maps + # items to latent dimensions); pairs are nested in reconstruction + # groups via cluster_id -- fast-mlsirm's multilevel random-intercept + # structure (Fox & Glas, 2001), which requires the marginal (mmle) + # estimator. + result = fit( + responses=numpy.asarray(responses, dtype=float), + factor_id=numpy.zeros(len(channels), dtype=numpy.int64), + cluster_id=numpy.asarray(group_ids, dtype=numpy.int64), + config=FitConfig(model="MLS2PLM", latent_dim=1, estimator="mmle"), + ) + log_discriminations = list(numpy.asarray(result.params.alpha, dtype=float).ravel()) + if len(log_discriminations) != len(channels): + return None + if any(not math.isfinite(alpha) for alpha in log_discriminations): + return None + + # exp(alpha) is the natural-scale discrimination -- positive by + # construction, so the normalization always yields valid convex + # weights (Birnbaum, 1968: optimal weight proportional to a_j). + discriminations = [math.exp(alpha) for alpha in log_discriminations] + total = sum(discriminations) + if not math.isfinite(total) or total <= 0: + return None + return ChannelWeightEstimate( + weights={ + channel: discrimination / total + for channel, discrimination in zip(channels, discriminations) + }, + sample_pair_count=len(pair_channel_scores), + estimation_method_code="mls2plm_discrimination", + ) diff --git a/lineageweave/lineage_persistence.py b/lineageweave/lineage_persistence.py index 8b9f4f16d..d103eab47 100644 --- a/lineageweave/lineage_persistence.py +++ b/lineageweave/lineage_persistence.py @@ -16,7 +16,12 @@ 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, + weights: dict[str, float] | 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 +34,13 @@ 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. + + ``weights`` defaults to ``None``, keeping ``reconstruct()``'s + documented fallback constants; callers with a persisted + psychometric estimate (ADR 0145) pass it here. """ - trees = reconstruct(list(records), llm=llm) + if weights is None: + trees = reconstruct(list(records), llm=llm) + else: + trees = reconstruct(list(records), llm=llm, weights=weights) return [edge for tree in trees for edge in tree.edges] diff --git a/migrations/0135_lineage_channel_weight.sql b/migrations/0135_lineage_channel_weight.sql new file mode 100644 index 000000000..089c0e8eb --- /dev/null +++ b/migrations/0135_lineage_channel_weight.sql @@ -0,0 +1,12 @@ +-- ADR 0145: lineage channel-fusion weights become estimated, persisted, +-- provenance-bearing quantities instead of hand-picked constants. +-- One row per channel; rebuild_lineage uses the set only when it exactly +-- matches the active channel set (no partial mixing). + +create table if not exists lineage_channel_weight ( + channel_code text primary key, + weight_value double precision not null, + estimation_method_code text not null, + sample_pair_count bigint not null, + estimated_at timestamptz not null default now() +); diff --git a/migrations/rollback/0135_lineage_channel_weight.sql b/migrations/rollback/0135_lineage_channel_weight.sql new file mode 100644 index 000000000..7e8038963 --- /dev/null +++ b/migrations/rollback/0135_lineage_channel_weight.sql @@ -0,0 +1,4 @@ +-- Rollback for 0135: drop the persisted channel-weight table. Rebuilds +-- fall back to the documented in-code constants (ADR 0145 fallback). + +drop table if exists lineage_channel_weight; diff --git a/scripts/backfill_thread_group_keys.py b/scripts/backfill_thread_group_keys.py new file mode 100644 index 000000000..60c117e96 --- /dev/null +++ b/scripts/backfill_thread_group_keys.py @@ -0,0 +1,163 @@ +"""Backfill for the dataset-wide Event Lineage grouping-key placeholders. + +The whole reason `lineageweave.reconstruct` exists is that the source +system carries no reliable thread key at all -- related posts arrive +unlinked, and lineage is *reconstructed* from evidence: the temporal, +secondary-key, text-similarity, and llm channels fused through RankWeave +(see reconstruct.py's own validation note: naive grouping-plus-recency +agreed with an independent signal only 2.6% of the time). `group_key` +is therefore only the candidate-pool bound; the channels decide the +actual links. + +The `zcrht811_export_rows` import inverted that design by stuffing +per-row-unique identifiers into BOTH grouping columns -- confirmed live: + +- `thread_group_key` = the row's own source record GUID on + 43,811/43,839 rows, so `_group_by` saw ~43k singleton groups and the + channels never scored a single real candidate pair; +- `secondary_grouping_key` = the row's own per-document number + (43,707 distinct / 43,814 imported), so the secondary-key channel -- + whose own docstring names "e.g. project code" as its intended signal + -- could never fire either. + +Event Lineage, the product's headline feature, was silently +non-functional for virtually the whole corpus. This backfill restores +the library's designed inputs, only on rows carrying the placeholder +signature (`thread_group_key` equal to the row's own +`source_record_key`) so seeded demo rows with real designed keys are +untouched: + +1. `thread_group_key` -> `''` (the column is NOT NULL; migration 0002 + made the empty string the "no persisted signal" value), so + `reconstruct_group_key`'s existing fallback to + `process_unit_id`/`corporate_entity_id` forms real candidate pools. +2. `secondary_grouping_key` -> `source_project_code` (empty where the + source had none), so posts naming the same project contribute + fused *evidence* toward a link without being walled off from + related posts that lack a project code -- a hard project partition + on `thread_group_key` would have blocked exactly the + high-relatedness cross-links this library exists to find. + +Idempotent and safe to re-run. Does not itself rebuild +`post_lineage_edge` -- run `POST /api/lineage/rebuild` (or the +equivalent `rebuild_lineage()` call) afterward. + +`thread_group_key` is not lineage-only: it is also the live scope key +for `analysis_scope_thread_group` analysis runs (TEPP measurement and +period reports included -- `analysis_run_ingestion.py` re-resolves +`p.thread_group_key = scope.scope_key` on every read for +ABAC-visibility, and `report_ingestion.py` groups thread-group reports +by it). A run's *member posts* are frozen in +`analysis_source_snapshot_member` at capture time, but that live scope +match is not. Rewriting keys would therefore orphan any existing +thread-group-scoped run: this script fails closed if one exists whose +`scope_key` still matches a current `thread_group_key`, rather than +silently detaching it. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json + +import asyncpg + +from backend.app.config import load_settings + + +async def backfill_thread_group_keys(conn: asyncpg.Connection, *, dry_run: bool) -> dict[str, int]: + """Clear placeholder grouping keys and route project codes to the + secondary-key channel. + + Isolated from pool/connection setup so the counting/rollback logic is + unit-testable without a real database. Fails closed (no write) when an + existing `analysis_scope_thread_group` run's scope_key would be + orphaned by the rewrite -- that run's ABAC-visibility scope match is + resolved live against `thread_group_key` on every read, not frozen in + its snapshot. + """ + async with conn.transaction(): + anchored_runs = await conn.fetch( + """ + select scope.analysis_run_id, scope.scope_key + from analysis_run_scope scope + where scope.scope_kind_code = 'analysis_scope_thread_group' + and exists ( + select 1 from source_post p + where btrim(p.thread_group_key) = scope.scope_key + and btrim(p.thread_group_key) = btrim(p.source_record_key) + ) + """ + ) + if anchored_runs: + run_ids = ", ".join(str(row["analysis_run_id"]) for row in anchored_runs) + raise RuntimeError( + "refusing to rewrite thread_group_key: existing " + f"analysis_scope_thread_group run(s) [{run_ids}] resolve their " + "scope against values this backfill would change. Retire or " + "re-scope those runs first." + ) + # Only rows carrying the placeholder signature -- a thread key equal + # to the row's own record key groups nothing and can only be import + # damage; a seeded or genuinely-mapped key never self-references. + rows = await conn.fetch( + """ + update source_post + set thread_group_key = '', + secondary_grouping_key = coalesce(nullif(btrim(source_project_code), ''), '') + where source_record_key is not null + and btrim(thread_group_key) = btrim(source_record_key) + returning (nullif(btrim(source_project_code), '') is not null) as had_project_code + """ + ) + project_evidence = sum(1 for row in rows if row["had_project_code"]) + cleared = len(rows) + if dry_run: + raise _RollbackDryRun(project_evidence, cleared) + return { + "cleared_placeholder_posts": cleared, + "project_secondary_evidence_posts": project_evidence, + } + + +class _RollbackDryRun(Exception): + """Raised inside the transaction to force a rollback for --dry-run.""" + + def __init__(self, project_evidence: int, cleared: int) -> None: + super().__init__("dry run -- rolled back") + self.project_evidence = project_evidence + self.cleared = cleared + + +async def _run(args: argparse.Namespace) -> dict[str, object]: + settings = load_settings() + pool = await asyncpg.create_pool(settings.database_url, min_size=1, max_size=1) + try: + async with pool.acquire() as conn: + try: + counts = await backfill_thread_group_keys(conn, dry_run=args.dry_run) + return {**counts, "dry_run": False} + except _RollbackDryRun as rolled_back: + return { + "cleared_placeholder_posts": rolled_back.cleared, + "project_secondary_evidence_posts": rolled_back.project_evidence, + "dry_run": True, + } + finally: + await pool.close() + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--dry-run", + action="store_true", + help="Report counts without writing (rolls back the transaction)", + ) + args = parser.parse_args() + print(json.dumps(asyncio.run(_run(args)), ensure_ascii=False, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/scripts/estimate_channel_weights.py b/scripts/estimate_channel_weights.py new file mode 100644 index 000000000..ac0020827 --- /dev/null +++ b/scripts/estimate_channel_weights.py @@ -0,0 +1,151 @@ +"""Operator estimation of lineage channel-fusion weights (ADR 0145). + +Samples candidate parent-child pairs from the real corpus exactly the +way `reconstruct` forms them (same grouping fallback, same candidate +window), scores each pair on the deterministic channels, fits +`fast-mlsirm`'s multilevel 2PL over the dichotomized scores, and +persists the normalized discriminations into `lineage_channel_weight` +(migration 0135). `rebuild_lineage` then uses them automatically on the +next rebuild -- but only when the persisted channel set exactly matches +the rebuild's active channels, so a partial estimate never silently +mixes with hand-picked constants. + +This first landing estimates over the three deterministic channels +only; the llm adjudication channel joins the estimate in a follow-up +(each sampled pair then costs a provider call -- ADR 0145 point 5). +Until then the exact-match rule keeps llm-active rebuilds on the +documented fallback. Fails closed: with `fast_mlsirm` not importable or +a degenerate sample, nothing is written and the exit is an explicit +error, never a fabricated weight. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json + +import asyncpg + +from backend.app.config import load_settings +from backend.app.lineage_ingestion import records_from_source_posts +from backend.app.post_eligibility import SOURCE_POST_ELIGIBILITY_SQL +from lineageweave.channel_weight_estimation import ( + ChannelWeightEstimate, + estimate_channel_weights, +) +from lineageweave.channels import ( + secondary_key_match_score, + temporal_score, + text_similarity_score, +) +from lineageweave.reconstruct import DEFAULT_CANDIDATE_WINDOW + + +def sample_pair_scores( + records: list, *, window: int = DEFAULT_CANDIDATE_WINDOW +) -> tuple[list[dict[str, float]], list[int]]: + """Score every in-window candidate pair, grouped as reconstruct groups. + + Pure so the sampling geometry itself is unit-testable: pairs come + only from within one group, only from the trailing ``window`` of + temporally prior records -- the exact candidate set + ``reconstruct._reconstruct_group`` would consider. + """ + groups: dict[str, list] = {} + for record in records: + groups.setdefault(record.group_key, []).append(record) + + pair_scores: list[dict[str, float]] = [] + group_ids: list[int] = [] + for group_index, group_records in enumerate(groups.values()): + ordered = sorted(group_records, key=lambda r: r.occurred_at) + for index, record in enumerate(ordered): + for candidate in ordered[max(0, index - window) : index]: + pair_scores.append( + { + "temporal": temporal_score(candidate, record), + "secondary_key": secondary_key_match_score(candidate, record), + "text": text_similarity_score(candidate, record), + } + ) + group_ids.append(group_index) + return pair_scores, group_ids + + +async def persist_estimate( + conn: asyncpg.Connection, estimate: ChannelWeightEstimate +) -> None: + """Replace the persisted weight set atomically with this estimate.""" + async with conn.transaction(): + await conn.execute("delete from lineage_channel_weight") + for channel, weight in estimate.weights.items(): + await conn.execute( + """ + insert into lineage_channel_weight + (channel_code, weight_value, estimation_method_code, sample_pair_count) + values ($1, $2, $3, $4) + """, + channel, + weight, + estimate.estimation_method_code, + estimate.sample_pair_count, + ) + + +async def _run(args: argparse.Namespace) -> dict[str, object]: + settings = load_settings() + pool = await asyncpg.create_pool(settings.database_url, min_size=1, max_size=1) + try: + async with pool.acquire() as conn: + 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')} " + "order by created_at, post_id limit $1::bigint", + args.post_limit, + ) + records = records_from_source_posts(rows) + pair_scores, group_ids = sample_pair_scores(records) + estimate = estimate_channel_weights(pair_scores, group_ids) + if estimate is None: + raise RuntimeError( + "no grounded estimate was produced (fast_mlsirm unavailable, " + f"sample of {len(pair_scores)} pairs too small, or a channel " + "was degenerate) -- nothing was written; rebuilds keep the " + "documented fallback weights" + ) + if not args.dry_run: + await persist_estimate(conn, estimate) + return { + "weights": estimate.weights, + "sample_pair_count": estimate.sample_pair_count, + "estimation_method_code": estimate.estimation_method_code, + "persisted": not args.dry_run, + } + finally: + await pool.close() + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--post-limit", + type=int, + default=5000, + help="Maximum eligible posts to sample pairs from (default: 5000)", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Estimate and report, but persist nothing", + ) + args = parser.parse_args() + if args.post_limit < 1: + parser.error("--post-limit must be positive") + print(json.dumps(asyncio.run(_run(args)), ensure_ascii=False, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/tests/test_backfill_thread_group_keys.py b/tests/test_backfill_thread_group_keys.py new file mode 100644 index 000000000..f29092782 --- /dev/null +++ b/tests/test_backfill_thread_group_keys.py @@ -0,0 +1,153 @@ +"""Tests for scripts/backfill_thread_group_keys.py's backfill logic. + +`backfill_thread_group_keys` is the pure per-connection operation, isolated +from pool/connection setup precisely so it can be exercised here without a +real database -- same reasoning as tests/test_customer_hint_ingestion.py's +fakes. +""" + +from __future__ import annotations + +import asyncio +from contextlib import asynccontextmanager + +import scripts.backfill_thread_group_keys as backfill + + +class _Connection: + """Simulates the guard SELECT and the UPDATE...RETURNING this script issues. + + ``rows`` models every ``source_post`` row carrying the placeholder + signature (``thread_group_key`` equal to the row's own record key): + each entry is ``had_project_code`` (True when the row's + ``source_project_code`` was non-empty, so it now feeds the + secondary-key evidence channel). A row NOT in ``rows`` models a + seeded/genuinely-mapped row the placeholder predicate never touches. + ``anchored_runs`` models existing analysis_scope_thread_group runs + whose live scope match the rewrite would orphan. + """ + + def __init__(self, rows: list[bool], anchored_runs: list[str] | None = None) -> None: + self._rows = rows + self._anchored_runs = anchored_runs or [] + self.executed: list[str] = [] + + @asynccontextmanager + async def transaction(self): + yield self + + async def fetch(self, query: str, *args: object): + self.executed.append(" ".join(query.split())) + if "analysis_run_scope" in query: + return [ + {"analysis_run_id": run_id, "scope_key": f"key-{run_id}"} + for run_id in self._anchored_runs + ] + return [{"had_project_code": had_project_code} for had_project_code in self._rows] + + +def test_backfill_clears_placeholders_and_routes_project_codes_to_secondary() -> None: + conn = _Connection([True, True, False, False, False]) + result = asyncio.run(backfill.backfill_thread_group_keys(conn, dry_run=False)) + assert result == { + "cleared_placeholder_posts": 5, + "project_secondary_evidence_posts": 2, + } + assert len(conn.executed) == 2 + assert "analysis_run_scope" in conn.executed[0] + update = conn.executed[1] + assert "update source_post" in update + # The placeholder signature is the only predicate -- seeded rows with + # real designed keys must never match. + assert "btrim(thread_group_key) = btrim(source_record_key)" in update + # Project code is routed to the secondary-key evidence channel, never + # to thread_group_key -- a hard project partition would wall off + # related posts that lack a project code, exactly the links the + # reconstruction library exists to find. + assert "set thread_group_key = ''" in update + assert "secondary_grouping_key = coalesce(nullif(btrim(source_project_code), ''), '')" in update + + +def test_backfill_fails_closed_when_a_thread_group_scoped_run_would_be_orphaned() -> None: + # analysis_scope_thread_group runs resolve `thread_group_key = + # scope_key` live on every read (ABAC visibility) -- their member + # posts are snapshot-frozen but the scope match is not. Rewriting + # the keys out from under such a run silently detaches it, so the + # backfill must refuse instead, before any UPDATE. + conn = _Connection([True, False], anchored_runs=["run-1", "run-2"]) + try: + asyncio.run(backfill.backfill_thread_group_keys(conn, dry_run=False)) + except RuntimeError as exc: + assert "run-1" in str(exc) + assert "run-2" in str(exc) + else: + raise AssertionError("expected RuntimeError") + assert all("update source_post" not in query for query in conn.executed) + + +def test_backfill_no_placeholder_rows_is_a_clean_no_op() -> None: + conn = _Connection([]) + result = asyncio.run(backfill.backfill_thread_group_keys(conn, dry_run=False)) + assert result == { + "cleared_placeholder_posts": 0, + "project_secondary_evidence_posts": 0, + } + + +def test_dry_run_reports_counts_but_raises_to_force_a_rollback() -> None: + conn = _Connection([True, False]) + try: + asyncio.run(backfill.backfill_thread_group_keys(conn, dry_run=True)) + except backfill._RollbackDryRun as rolled_back: + assert rolled_back.project_evidence == 1 + assert rolled_back.cleared == 2 + else: + raise AssertionError("expected _RollbackDryRun") + + +class _FakePool: + def __init__(self, conn: _Connection) -> None: + self._conn = conn + + @asynccontextmanager + async def acquire(self): + yield self._conn + + async def close(self) -> None: + return None + + +def _patch_pool(monkeypatch, conn: _Connection) -> None: + async def fake_create_pool(*_args, **_kwargs): + return _FakePool(conn) + + monkeypatch.setattr(backfill.asyncpg, "create_pool", fake_create_pool) + monkeypatch.setattr( + backfill, "load_settings", lambda: type("S", (), {"database_url": "postgresql://x"})() + ) + + +def test_run_reports_dry_run_counts_without_the_internal_exception_leaking(monkeypatch) -> None: + import argparse + + conn = _Connection([True, True, False]) + _patch_pool(monkeypatch, conn) + result = asyncio.run(backfill._run(argparse.Namespace(dry_run=True))) + assert result == { + "cleared_placeholder_posts": 3, + "project_secondary_evidence_posts": 2, + "dry_run": True, + } + + +def test_run_reports_write_counts_when_not_a_dry_run(monkeypatch) -> None: + import argparse + + conn = _Connection([True, False, False]) + _patch_pool(monkeypatch, conn) + result = asyncio.run(backfill._run(argparse.Namespace(dry_run=False))) + assert result == { + "cleared_placeholder_posts": 3, + "project_secondary_evidence_posts": 1, + "dry_run": False, + } diff --git a/tests/test_channel_weight_estimation.py b/tests/test_channel_weight_estimation.py new file mode 100644 index 000000000..17cb63843 --- /dev/null +++ b/tests/test_channel_weight_estimation.py @@ -0,0 +1,126 @@ +"""Tests for lineageweave.channel_weight_estimation (ADR 0145). + +The fail-closed paths run everywhere. The parameter-recovery test -- +the organization's standard for measurement code (planted true +parameters recovered by the estimate) -- runs when `fast_mlsirm` is +importable and skips honestly otherwise, same as this repo's +live-service skips. +""" + +from __future__ import annotations + +import importlib.util +import math +import random + +import pytest + +from lineageweave.channel_weight_estimation import ( + _MIN_SAMPLE_PAIRS, + dichotomize, + estimate_channel_weights, +) +from lineageweave.models import Record +from lineageweave.reconstruct import DEFAULT_MIN_FUSED_SCORE + +_FAST_MLSIRM_AVAILABLE = importlib.util.find_spec("fast_mlsirm") is not None + + +def test_dichotomize_uses_the_fusion_floor_as_the_link_event_boundary() -> None: + assert dichotomize(DEFAULT_MIN_FUSED_SCORE) == 1 + assert dichotomize(DEFAULT_MIN_FUSED_SCORE - 1e-9) == 0 + assert dichotomize(1.0) == 1 + assert dichotomize(0.0) == 0 + + +def test_too_small_a_sample_fails_closed() -> None: + pairs = [{"temporal": 0.9, "text": 0.1}] * (_MIN_SAMPLE_PAIRS - 1) + assert estimate_channel_weights(pairs, [0] * len(pairs)) is None + + +def test_misaligned_inputs_are_a_caller_bug_not_missing_data() -> None: + with pytest.raises(ValueError): + estimate_channel_weights([{"temporal": 0.5}], [0, 1]) + pairs = [{"temporal": 0.5}, {"text": 0.5}] * _MIN_SAMPLE_PAIRS + with pytest.raises(ValueError): + estimate_channel_weights(pairs, [0] * len(pairs)) + + +def test_a_degenerate_channel_fails_closed() -> None: + # `text` never clears the floor: its 2PL slope is undefined in + # practice, so the whole estimate is refused, never worked around. + pairs = [ + {"temporal": 0.9 if index % 2 else 0.1, "text": 0.0} + for index in range(_MIN_SAMPLE_PAIRS) + ] + assert estimate_channel_weights(pairs, [0] * len(pairs)) is None + + +@pytest.mark.skipif( + _FAST_MLSIRM_AVAILABLE, reason="exercises the import-failure fallback" +) +def test_without_fast_mlsirm_a_valid_sample_still_fails_closed() -> None: + generator = random.Random(20260823) + pairs = [ + { + "temporal": generator.random(), + "text": generator.random(), + } + for _ in range(_MIN_SAMPLE_PAIRS) + ] + assert estimate_channel_weights(pairs, [0] * len(pairs)) is None + + +@pytest.mark.skipif( + not _FAST_MLSIRM_AVAILABLE, reason="requires fast_mlsirm -- install from the org repo" +) +def test_recovery_a_more_discriminating_channel_earns_a_larger_weight() -> None: + """Parameter-recovery-shaped check per the org's measurement standard. + + Three channels, matching production's deterministic channel count (a + two-item 2PL leaves discriminations weakly identified). Plant a + latent per-pair relatedness; `strong` tracks it almost + deterministically, `mid` moderately, `weak` barely better than + chance. The estimated convex weights must recover the Birnbaum + (1968) ordering strong > weak and form a valid convex combination. + """ + generator = random.Random(20260823) + + def channel_score(related: bool, follow_probability: float) -> float: + follows = generator.random() < follow_probability + high = related if follows else not related + return (0.8 if high else 0.05) + generator.uniform(-0.04, 0.04) + + # Follow probabilities stay away from the quasi-separation regime: a + # near-deterministic item's slope is unstable under regularized + # estimation and can be shrunk below a moderate item's, which would + # test the estimator's penalty behavior rather than the Birnbaum + # ordering this check is about. Clusters carry genuine intercept + # variance (per-group relatedness base rates) -- the structure the + # multilevel random intercept exists to model; clusters that are a + # meaningless round-robin instead flatten the slope estimates. + group_base_rate = [generator.uniform(0.25, 0.75) for _ in range(12)] + pairs: list[dict[str, float]] = [] + group_ids: list[int] = [] + for index in range(900): + group = index % 12 + related = generator.random() < group_base_rate[group] + pairs.append( + { + "strong": channel_score(related, 0.85), + "mid": channel_score(related, 0.70), + "weak": channel_score(related, 0.55), + } + ) + group_ids.append(group) + + estimate = estimate_channel_weights(pairs, group_ids) + assert estimate is not None + weights = estimate.weights + assert set(weights) == {"strong", "mid", "weak"} + assert math.isclose(sum(weights.values()), 1.0, rel_tol=1e-9) + assert all(weight > 0 for weight in weights.values()) + assert weights["strong"] > weights["weak"] + assert weights["strong"] > weights["mid"] > weights["weak"] + assert estimate.sample_pair_count == 900 + assert estimate.estimation_method_code == "mls2plm_discrimination" diff --git a/tests/test_estimate_channel_weights_script.py b/tests/test_estimate_channel_weights_script.py new file mode 100644 index 000000000..b299df42d --- /dev/null +++ b/tests/test_estimate_channel_weights_script.py @@ -0,0 +1,77 @@ +"""Tests for scripts/estimate_channel_weights.py's sampling and persistence. + +`sample_pair_scores` must reproduce reconstruct's own candidate +geometry -- within-group only, trailing-window only -- because weights +estimated over a different pair population would ground nothing. +""" + +from __future__ import annotations + +import asyncio +from contextlib import asynccontextmanager +from datetime import datetime, timedelta + +from lineageweave.channel_weight_estimation import ChannelWeightEstimate +from lineageweave.models import Record + +import scripts.estimate_channel_weights as script + + +def _record(record_id: str, group: str, minute: int, secondary: str = "") -> Record: + return Record( + record_id, + group, + f"title {record_id}", + datetime(2026, 1, 1) + timedelta(minutes=minute), + secondary, + ) + + +def test_sampling_stays_within_groups_and_window() -> None: + records = [ + _record("a1", "g-a", 0), + _record("a2", "g-a", 1), + _record("b1", "g-b", 2), + ] + pair_scores, group_ids = script.sample_pair_scores(records, window=50) + # Only a1->a2 pairs up; b1 is alone in its group and never crosses. + assert len(pair_scores) == 1 + assert group_ids == [0] + assert set(pair_scores[0]) == {"temporal", "secondary_key", "text"} + + +def test_sampling_window_bounds_candidates_like_reconstruct() -> None: + records = [_record(f"r{index}", "g", index) for index in range(5)] + _, unbounded_ids = script.sample_pair_scores(records, window=50) + assert len(unbounded_ids) == 4 + 3 + 2 + 1 + pair_scores, _ = script.sample_pair_scores(records, window=2) + # Each record sees at most its two immediate predecessors. + assert len(pair_scores) == 1 + 2 + 2 + 2 + + +def test_persist_estimate_replaces_the_whole_weight_set() -> None: + class _Connection: + def __init__(self) -> None: + self.executed: list[tuple[str, tuple[object, ...]]] = [] + + @asynccontextmanager + async def transaction(self): + yield self + + async def execute(self, query: str, *args: object) -> str: + self.executed.append((" ".join(query.split()), args)) + return "OK" + + conn = _Connection() + estimate = ChannelWeightEstimate( + weights={"temporal": 0.25, "text": 0.75}, + sample_pair_count=600, + estimation_method_code="mls2plm_discrimination", + ) + asyncio.run(script.persist_estimate(conn, estimate)) + assert "delete from lineage_channel_weight" in conn.executed[0][0] + inserted = {call[1][0]: call[1] for call in conn.executed[1:]} + assert set(inserted) == {"temporal", "text"} + assert inserted["text"][1] == 0.75 + assert inserted["text"][2] == "mls2plm_discrimination" + assert inserted["text"][3] == 600