diff --git a/backend/app/lineage_ingestion.py b/backend/app/lineage_ingestion.py index f1e76d495..cffb96bd8 100644 --- a/backend/app/lineage_ingestion.py +++ b/backend/app/lineage_ingestion.py @@ -10,8 +10,11 @@ from __future__ import annotations +import math +import re +from collections.abc import Mapping from datetime import datetime -from typing import Any, Mapping +from typing import Any import asyncpg @@ -19,6 +22,10 @@ from lineageweave.lineage_persistence import lineage_edge_specs from lineageweave.models import Edge, Record +# ADR 0145 rejected the unanchored estimator. A future accepted ADR must add +# its independently validated method code here before persisted weights can run. +_SUPPORTED_ANCHOR_METHOD_CODES: frozenset[str] = frozenset() + def _occurred_at(value: datetime) -> datetime: """Reconstruct expects naive datetimes; asyncpg returns timestamptz.""" @@ -72,6 +79,80 @@ 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: + """Load only a complete vector from an independently anchored method. + + ADR 0145 currently authorizes no anchor method. A partial or invalid vector + returns ``None`` rather than being repaired. A database that has not applied + migration 0135 is likewise an unavailable state, detected without issuing a + statement that would abort the caller's outer PostgreSQL transaction. + """ + table_exists = await conn.fetchval( + "select to_regclass('public.lineage_channel_weight') is not null" + ) + if not table_exists: + return None + rows = await conn.fetch( + "select channel_code, weight_value, estimation_run_id, " + "estimation_method_code, estimator_version, anchor_method_code, " + "source_snapshot_sha256, sample_pair_count, knowledge_cutoff " + "from lineage_channel_weight" + ) + persisted = {row["channel_code"]: float(row["weight_value"]) for row in rows} + if not persisted or set(persisted) != active_channels: + return None + if any( + not math.isfinite(weight) or weight <= 0 or weight > 1 + for weight in persisted.values() + ): + return None + if not math.isclose(sum(persisted.values()), 1.0, rel_tol=0.0, abs_tol=1e-9): + return None + provenance = { + ( + row["estimation_run_id"], + row["estimation_method_code"], + row["estimator_version"], + row["anchor_method_code"], + row["source_snapshot_sha256"], + row["sample_pair_count"], + row["knowledge_cutoff"], + ) + for row in rows + } + if len(provenance) != 1: + return None + run = next(iter(provenance)) + ( + run_id, + estimation_method, + estimator_version, + anchor_method, + snapshot_digest, + sample_pair_count, + knowledge_cutoff, + ) = run + if ( + run_id is None + or not isinstance(estimation_method, str) + or not estimation_method.strip() + or not isinstance(estimator_version, str) + or not estimator_version.strip() + or not isinstance(anchor_method, str) + or anchor_method not in _SUPPORTED_ANCHOR_METHOD_CODES + or not isinstance(snapshot_digest, str) + or re.fullmatch(r"[0-9a-f]{64}", snapshot_digest) is None + or not isinstance(sample_pair_count, int) + or isinstance(sample_pair_count, bool) + or sample_pair_count < 200 + or not isinstance(knowledge_cutoff, datetime) + ): + 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 +160,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 438b4786a..562ab9e62 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -113,6 +113,11 @@ / "migrations" / "0102_project_bound_summary_event.sql" ) +_CHANNEL_WEIGHT_MIGRATION = ( + Path(__file__).resolve().parents[2] + / "migrations" + / "0135_lineage_channel_weight.sql" +) def _postgres_available() -> bool: @@ -226,6 +231,7 @@ def seeded_db(demo_analyst_token): cur.execute(_MAJOR_EVENT_ACTION_MIGRATION.read_text()) cur.execute(_PROJECT_BOUND_ACTION_MIGRATION.read_text()) cur.execute(_PROJECT_BOUND_EVENT_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 f329117d6..5d9e8dee7 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_*) ;; + 0060_*|0100_*|0101_*|0102_*|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..8bb45e40d --- /dev/null +++ b/docs/adr/0145-psychometric-channel-weight-estimation.md @@ -0,0 +1,87 @@ +# ADR 0145 — Channel-weight estimation remains unavailable without an independent anchor + +**Decision status:** Rejected proposal +**Date:** 2026-08-23 +**Reconciles with:** [ADR 0003](0003-fast-mlsirm-report-integration.md) + +## Context + +`lineageweave.reconstruct` currently uses hand-picked convex weights for its +temporal, secondary-key, text-similarity, and optional LLM channels. Those +constants are an explicitly ungrounded historical fallback; a citation does +not turn them into calibrated measurement. + +The rejected proposal treated channels as 2PL items, candidate pairs as +respondents, and normalized item discriminations as fusion weights. That does +not establish the product construct. An unanchored IRT fit can describe common +response structure, but it provides no independent evidence that its latent +factor is “these posts are genuinely related.” Birnbaum's 2PL item information +is also conditional on trait location and item difficulty, +`I_j(theta) = a_j^2 P_j(theta) (1 - P_j(theta))`; it is not a global constant +proportional only to `a_j`. Normalizing discriminations therefore is not an +information-optimal convex fusion rule. + +The official pinned `fast-mlsirm` contract makes two further boundaries +explicit: `factor_id` assigns items to latent dimensions, while `cluster_id` +represents respondent nesting; and a `FitResult` exposes +`convergence_status` plus package diagnostics that callers must inspect. Those +contracts can validate a fit's execution, but cannot supply the missing +criterion validity. + +Accepted ADR 0003 assigns temporal/event measurement to TEPP and limits this +repository's fast-mlsirm integration to the approved LLM-judge/report path. +This proposed lineage-weight path cannot silently expand that boundary. + +## Decision + +1. **No unanchored estimate.** LineageWeave does not run the proposed + candidate-pair IRT fit and does not persist or activate weights from it. + Estimation reports unavailable until an independent lineage anchor and its + upstream contract exist. +2. **No scientific claim for fallback constants.** Existing constants remain + unchanged for compatibility, but are neither calibrated nor + paper-grounded. This ADR does not promote them to measurement evidence. +3. **A future proposal must be ADR-first.** It must amend ADR 0003, identify an + independent outcome/anchor (for example, an accepted TEPP contract rather + than a local proxy), use official fast-mlsirm diagnostics, reject every + non-converged fit, and prove criterion validity before product activation. +4. **Future persisted vectors must be self-consistent and reproducible.** The + schema requires a known channel vocabulary, finite positive weights, an + exact sum of one at runtime, one estimation run identity, estimator and + anchor method versions, sample size, immutable source-snapshot digest, and + knowledge cutoff. Current code has no supported anchor method and therefore + loads no vector. +5. **Grouping repair is independent of measurement.** Source row identifiers + that were mapped as grouping values are normalized only in derived + reconstruction fields. Their caller-mapped raw values remain preserved in + separate source-provenance columns across backfill and re-import. + +## Consequences + +- `scripts/estimate_channel_weights.py` exits without writing because no + scientifically authorized anchor exists. +- Missing migration 0135 remains a normal rollout state and is detected via a + non-error PostgreSQL catalog probe, so an outer rebuild transaction is not + aborted. +- The persistence contract is fail-closed: malformed, mixed-provenance, or + unsupported-anchor rows are ignored rather than renormalized or repaired. +- Parameter-recovery tests cannot substitute for criterion validity; they may + return only with a future accepted anchored estimator. + +## References + +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. + +Fox, J.-P., & Glas, C. A. W. (2001). Bayesian estimation of a multilevel IRT +model using Gibbs sampling. *Psychometrika, 66*(2), 271–288. +https://doi.org/10.1007/BF02294839 + +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 + +ContextualWisdomLab. (2026). *fast-mlsirm*, pinned LineageWeave dependency +contract at commit `5006c38286a4fa1d81bcf57eeed5ce27ae743f50`. +https://github.com/ContextualWisdomLab/fast-mlsirm/tree/5006c38286a4fa1d81bcf57eeed5ce27ae743f50 diff --git a/lineageweave/channel_weight_estimation.py b/lineageweave/channel_weight_estimation.py new file mode 100644 index 000000000..45df25f59 --- /dev/null +++ b/lineageweave/channel_weight_estimation.py @@ -0,0 +1,25 @@ +"""Fail-closed lineage channel-weight boundary (ADR 0145). + +Channel-score covariance is not an independent lineage anchor. Until an +accepted upstream anchor contract exists, this module produces no estimate. +""" + +from __future__ import annotations + + +def estimate_channel_weights( + pair_channel_scores: list[dict[str, float]], + group_ids: list[int], +) -> None: + """Report unavailable for unanchored channel scores. + + Args: + pair_channel_scores: unanchored candidate-pair channel scores. + group_ids: corresponding reconstruction-group indexes. + + Returns: + Always ``None`` until ADR 0145's independent-anchor requirement is met. + """ + if len(pair_channel_scores) != len(group_ids): + raise ValueError("pair_channel_scores and group_ids must align") + return None 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/lineageweave/reconstruct.py b/lineageweave/reconstruct.py index e0a6f89d0..031300424 100644 --- a/lineageweave/reconstruct.py +++ b/lineageweave/reconstruct.py @@ -20,10 +20,9 @@ from .channels import secondary_key_match_score, temporal_score, text_similarity_score from .models import Edge, Record, Tree -# Channel weights when every channel is available. llm gets the most weight -# because it is the only channel that actually reasons about the content -# instead of approximating it; the rest renormalize when llm is unavailable -# (see active_weights()). +# Legacy compatibility weights when every channel is available. These constants +# are not calibrated measurement evidence (ADR 0145); unavailable channels are +# dropped and the remainder renormalized by active_weights(). DEFAULT_CHANNEL_WEIGHTS = {"temporal": 0.15, "secondary_key": 0.15, "text": 0.30, "llm": 0.40} # ponytail: only the most recent WINDOW prior records in a group are diff --git a/migrations/0135_lineage_channel_weight.sql b/migrations/0135_lineage_channel_weight.sql new file mode 100644 index 000000000..4a84450c3 --- /dev/null +++ b/migrations/0135_lineage_channel_weight.sql @@ -0,0 +1,41 @@ +-- ADR 0145: reserve an integrity- and provenance-bearing persistence contract. +-- No anchor method is currently authorized, so application code activates no +-- stored vector. + +alter table source_post + add column if not exists source_thread_group_key text, + add column if not exists source_secondary_grouping_key text; + +comment on column source_post.source_thread_group_key is + 'Raw caller-mapped source thread field; preserved separately from derived reconstruction grouping.'; +comment on column source_post.source_secondary_grouping_key is + 'Raw caller-mapped source secondary-group field; preserved separately from derived reconstruction evidence.'; + +create table if not exists lineage_channel_weight ( + channel_code text primary key, + weight_value double precision not null, + estimation_run_id uuid not null, + estimation_method_code text not null, + estimator_version text not null, + anchor_method_code text not null, + source_snapshot_sha256 text not null, + sample_pair_count bigint not null, + knowledge_cutoff timestamptz not null, + estimated_at timestamptz not null default now(), + constraint lineage_channel_code_check + check (channel_code in ('temporal', 'secondary_key', 'text', 'llm')), + constraint lineage_weight_value_check + check (weight_value > 0 and weight_value <= 1), + constraint lineage_estimation_method_check + check (btrim(estimation_method_code) <> ''), + constraint lineage_estimator_version_check + check (btrim(estimator_version) <> ''), + constraint lineage_anchor_method_check + check (btrim(anchor_method_code) <> ''), + constraint lineage_sample_pair_count_check + check (sample_pair_count >= 200), + constraint lineage_source_snapshot_check + check (source_snapshot_sha256 ~ '^[0-9a-f]{64}$'), + constraint lineage_knowledge_cutoff_check + check (knowledge_cutoff <= estimated_at) +); diff --git a/migrations/rollback/0135_lineage_channel_weight.sql b/migrations/rollback/0135_lineage_channel_weight.sql new file mode 100644 index 000000000..98bcc5812 --- /dev/null +++ b/migrations/rollback/0135_lineage_channel_weight.sql @@ -0,0 +1,8 @@ +-- Rollback for 0135: remove the unavailable channel-weight persistence +-- contract and the raw grouping provenance columns introduced with it. + +drop table if exists lineage_channel_weight; + +alter table source_post + drop column if exists source_secondary_grouping_key, + drop column if exists source_thread_group_key; diff --git a/scripts/backfill_thread_group_keys.py b/scripts/backfill_thread_group_keys.py new file mode 100644 index 000000000..919e4ec84 --- /dev/null +++ b/scripts/backfill_thread_group_keys.py @@ -0,0 +1,164 @@ +"""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. + +An authorized-source validation found that an import had inverted that design +by putting per-row identity values into both grouping columns. That produced +singleton candidate pools, while the secondary-key channel could not fire. +Only this aggregate failure mode is retained here; private source identifiers +and records remain outside repository artifacts. + +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 source_thread_group_key = coalesce( + source_thread_group_key, thread_group_key + ), + source_secondary_grouping_key = coalesce( + source_secondary_grouping_key, secondary_grouping_key + ), + 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..c9b2ead00 --- /dev/null +++ b/scripts/estimate_channel_weights.py @@ -0,0 +1,41 @@ +"""Fail-closed operator boundary for lineage channel weights (ADR 0145). + +No independent lineage anchor is currently authorized, so this command does +not query the corpus, fit a model, or write weights. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json + + +async def _run(_args: argparse.Namespace) -> dict[str, object]: + raise RuntimeError( + "lineage channel-weight estimation is unavailable until an independent anchor " + "and accepted ADR authorize it; no corpus data was read and nothing was written" + ) + + +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/scripts/import_postgresql_posts.py b/scripts/import_postgresql_posts.py index e6ccf5449..f086c1b26 100644 --- a/scripts/import_postgresql_posts.py +++ b/scripts/import_postgresql_posts.py @@ -220,6 +220,32 @@ def _source_code_matches( return normalized in {item.strip().casefold() for item in excluded_values} +def _lineage_grouping_values( + row: Any, + mapping: ColumnMapping, + *, + record_key: str, + default_group: str, +) -> tuple[str | None, str | None, str, str]: + """Return raw source grouping followed by reconstruction grouping. + + A source thread key equal to its own record key is row identity, not + grouping evidence. Preserve both raw fields for provenance while deriving + the documented empty-thread/project-key reconstruction inputs. + """ + source_thread = str(_value(row, mapping.thread_group) or "").strip() or None + source_secondary = str(_value(row, mapping.secondary_group) or "").strip() or None + if source_thread == record_key: + project = str(_value(row, mapping.project_code) or "").strip() + return source_thread, source_secondary, "", project + return ( + source_thread, + source_secondary, + source_thread or default_group, + source_secondary or "", + ) + + def _validate_source_mapping( sales_pool_column: str | None, process_unit_column: str | None, @@ -419,6 +445,17 @@ async def import_rows(args: argparse.Namespace) -> dict[str, int]: _value(row, mapping.voc_type, "voc"), mapped=mapping.voc_type is not None, ) + ( + source_thread_group_key, + source_secondary_grouping_key, + thread_group_key, + secondary_grouping_key, + ) = _lineage_grouping_values( + row, + mapping, + record_key=record_key, + default_group=args.process_unit_code, + ) await target.execute( """ insert into source_post @@ -432,8 +469,9 @@ async def import_rows(args: argparse.Namespace) -> dict[str, int]: source_customer_code, source_customer_name, source_project_code, source_project_name, source_system_code, source_record_key, + source_thread_group_key, source_secondary_grouping_key, thread_group_key, secondary_grouping_key, created_at, updated_at) - values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30) + values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32) on conflict (post_id) do update set author_account_id = excluded.author_account_id, corporate_entity_id = excluded.corporate_entity_id, @@ -460,6 +498,9 @@ async def import_rows(args: argparse.Namespace) -> dict[str, int]: source_project_name = excluded.source_project_name, source_system_code = excluded.source_system_code, source_record_key = excluded.source_record_key, + source_thread_group_key = excluded.source_thread_group_key, + source_secondary_grouping_key = + excluded.source_secondary_grouping_key, thread_group_key = excluded.thread_group_key, secondary_grouping_key = excluded.secondary_grouping_key, created_at = excluded.created_at, @@ -491,8 +532,10 @@ async def import_rows(args: argparse.Namespace) -> dict[str, int]: str(_value(row, mapping.project_name) or "").strip() or None, args.source_system_code, record_key, - str(_value(row, mapping.thread_group, args.process_unit_code) or args.process_unit_code), - str(_value(row, mapping.secondary_group, "") or ""), + source_thread_group_key, + source_secondary_grouping_key, + thread_group_key, + secondary_grouping_key, created_at, updated_at, ) diff --git a/tests/test_backfill_thread_group_keys.py b/tests/test_backfill_thread_group_keys.py new file mode 100644 index 000000000..031003b64 --- /dev/null +++ b/tests/test_backfill_thread_group_keys.py @@ -0,0 +1,157 @@ +"""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 "thread_group_key = ''" in update + assert "secondary_grouping_key = coalesce(nullif(btrim(source_project_code), ''), '')" in update + assert "source_thread_group_key = coalesce(" in update + assert "source_thread_group_key, thread_group_key" in update + assert "source_secondary_grouping_key = coalesce(" in update + assert "source_secondary_grouping_key, secondary_grouping_key" 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..c6af06cdf --- /dev/null +++ b/tests/test_channel_weight_estimation.py @@ -0,0 +1,34 @@ +"""Tests for ADR 0145's fail-closed channel-weight boundary.""" + +from __future__ import annotations + +import pytest + +from lineageweave.channel_weight_estimation import estimate_channel_weights + + +def test_misaligned_inputs_are_a_caller_bug_not_missing_data() -> None: + with pytest.raises(ValueError): + estimate_channel_weights([{"temporal": 0.5}], [0, 1]) + + +def test_unanchored_channel_scores_never_run_a_fit(monkeypatch) -> None: + import fast_mlsirm + + def unexpected_fit(**_kwargs): + raise AssertionError("an unanchored fit cannot establish genuine lineage") + + monkeypatch.setattr(fast_mlsirm, "fit", unexpected_fit) + pairs = [ + { + "temporal": 0.9 if index % 2 else 0.1, + "secondary_key": 0.9 if index % 3 else 0.1, + "text": 0.9 if index % 5 else 0.1, + } + for index in range(200) + ] + + assert ( + estimate_channel_weights(pairs, [index % 4 for index in range(len(pairs))]) + is None + ) diff --git a/tests/test_estimate_channel_weights_script.py b/tests/test_estimate_channel_weights_script.py new file mode 100644 index 000000000..f6a8d481e --- /dev/null +++ b/tests/test_estimate_channel_weights_script.py @@ -0,0 +1,15 @@ +"""Tests for the fail-closed channel-weight operator boundary.""" + +from __future__ import annotations + +import argparse +import asyncio + +import pytest + +import scripts.estimate_channel_weights as script + + +def test_operator_reports_the_missing_independent_anchor() -> None: + with pytest.raises(RuntimeError, match="independent anchor"): + asyncio.run(script._run(argparse.Namespace(post_limit=1, dry_run=True))) diff --git a/tests/test_import_postgresql_posts.py b/tests/test_import_postgresql_posts.py index c98e96531..939844383 100644 --- a/tests/test_import_postgresql_posts.py +++ b/tests/test_import_postgresql_posts.py @@ -4,6 +4,7 @@ import pytest +import scripts.import_postgresql_posts as importer from scripts.import_postgresql_posts import ( _parser, _normalize_voc_type, @@ -15,6 +16,38 @@ ) +def test_placeholder_grouping_is_derived_without_losing_raw_source_values() -> None: + assert hasattr(importer, "_lineage_grouping_values") + mapping = SimpleNamespace( + thread_group="thread", + secondary_group="document", + project_code="project", + ) + + assert importer._lineage_grouping_values( + {"thread": " record-1 ", "document": " document-1 ", "project": " project-1 "}, + mapping, + record_key="record-1", + default_group="pu-1", + ) == ("record-1", "document-1", "", "project-1") + + +def test_real_source_grouping_remains_the_derived_grouping() -> None: + assert hasattr(importer, "_lineage_grouping_values") + mapping = SimpleNamespace( + thread_group="thread", + secondary_group="secondary", + project_code="project", + ) + + assert importer._lineage_grouping_values( + {"thread": "thread-a", "secondary": "secondary-a", "project": "project-a"}, + mapping, + record_key="record-1", + default_group="pu-1", + ) == ("thread-a", "secondary-a", "thread-a", "secondary-a") + + @pytest.mark.parametrize( ("source_value", "expected"), [("VOC", "voc"), ("VOCC", "vocc"), ("VOCO", "voco"), ("VOM", "vom"), ("VOP", "vop")], diff --git a/tests/test_lineage_ingestion.py b/tests/test_lineage_ingestion.py index de8f289c6..0e5c6a32f 100644 --- a/tests/test_lineage_ingestion.py +++ b/tests/test_lineage_ingestion.py @@ -3,9 +3,12 @@ from __future__ import annotations import asyncio -from datetime import datetime, timezone +import math +from datetime import UTC, datetime, timezone +import backend.app.lineage_ingestion as ingestion from backend.app.lineage_ingestion import ( + load_estimated_channel_weights, reconstruct_group_key, records_from_source_posts, visible_lineage_graph, @@ -14,6 +17,94 @@ from lineageweave.lineage_persistence import lineage_edge_specs +def test_missing_weight_table_is_detected_without_an_aborting_query() -> None: + class MissingTableConnection: + async def fetchval(self, query: str): + assert "to_regclass('public.lineage_channel_weight')" in query + return False + + async def fetch(self, _query: str): + raise AssertionError( + "missing tables must not be queried inside the outer transaction" + ) + + assert asyncio.run( + load_estimated_channel_weights( + MissingTableConnection(), {"temporal", "secondary_key", "text"} + ) + ) is None + + +def test_unapproved_weight_provenance_is_never_activated() -> None: + class StoredWeightConnection: + async def fetchval(self, _query: str): + return True + + async def fetch(self, _query: str): + provenance = { + "estimation_run_id": "00000000-0000-0000-0000-000000000001", + "estimation_method_code": "mls2plm_discrimination", + "estimator_version": "5006c382", + "anchor_method_code": "unanchored_channel_covariance", + "source_snapshot_sha256": "a" * 64, + "sample_pair_count": 600, + "knowledge_cutoff": datetime(2026, 1, 1, tzinfo=UTC), + } + return [ + {**provenance, "channel_code": "temporal", "weight_value": 0.2}, + {**provenance, "channel_code": "secondary_key", "weight_value": 0.3}, + {**provenance, "channel_code": "text", "weight_value": 0.5}, + ] + + assert asyncio.run( + load_estimated_channel_weights( + StoredWeightConnection(), {"temporal", "secondary_key", "text"} + ) + ) is None + + +def test_supported_vectors_still_require_numeric_and_provenance_integrity( + monkeypatch, +) -> None: + monkeypatch.setattr(ingestion, "_SUPPORTED_ANCHOR_METHOD_CODES", {"test_anchor"}) + + class StoredWeightConnection: + def __init__(self, weights: tuple[float, float, float]) -> None: + self.weights = weights + + async def fetchval(self, _query: str): + return True + + async def fetch(self, _query: str): + provenance = { + "estimation_run_id": "00000000-0000-0000-0000-000000000001", + "estimation_method_code": "test_method", + "estimator_version": "1.0.0", + "anchor_method_code": "test_anchor", + "source_snapshot_sha256": "a" * 64, + "sample_pair_count": 600, + "knowledge_cutoff": datetime(2026, 1, 1, tzinfo=UTC), + } + return [ + {**provenance, "channel_code": channel, "weight_value": weight} + for channel, weight in zip( + ("temporal", "secondary_key", "text"), self.weights + ) + ] + + active = {"temporal", "secondary_key", "text"} + assert asyncio.run( + load_estimated_channel_weights(StoredWeightConnection((0.2, 0.3, 0.5)), active) + ) == {"temporal": 0.2, "secondary_key": 0.3, "text": 0.5} + for invalid in ((0.2, 0.3, 0.6), (0.2, 0.8, math.nan), (0.2, 0.8, 0.0)): + assert ( + asyncio.run( + load_estimated_channel_weights(StoredWeightConnection(invalid), active) + ) + is None + ) + + def test_records_use_persisted_thread_keys_not_process_unit_or_voc_type() -> None: rows = [ { diff --git a/tests/test_migration_replay.py b/tests/test_migration_replay.py index 29fe1c176..0608c2361 100644 --- a/tests/test_migration_replay.py +++ b/tests/test_migration_replay.py @@ -33,3 +33,33 @@ def test_migrate_sh_replays_leftover_pair_migration_on_existing_volumes() -> Non ).read_text(encoding="utf-8") assert "0012_*" in script +def test_channel_weight_migration_preserves_raw_source_grouping() -> None: + migration = ( + Path(__file__).resolve().parents[1] + / "migrations" + / "0135_lineage_channel_weight.sql" + ).read_text(encoding="utf-8") + + assert "source_thread_group_key" in migration + assert "source_secondary_grouping_key" in migration + + +def test_channel_weight_migration_enforces_integrity_and_provenance() -> None: + migration = ( + Path(__file__).resolve().parents[1] + / "migrations" + / "0135_lineage_channel_weight.sql" + ).read_text(encoding="utf-8").casefold() + + for field in ( + "estimation_run_id", + "estimator_version", + "anchor_method_code", + "source_snapshot_sha256", + "knowledge_cutoff", + ): + assert field in migration + assert "channel_code in ('temporal', 'secondary_key', 'text', 'llm')" in migration + assert "weight_value > 0 and weight_value <= 1" in migration + assert "sample_pair_count >= 200" in migration + assert "source_snapshot_sha256 ~ '^[0-9a-f]{64}$'" in migration