diff --git a/backend/app/lineage_ingestion.py b/backend/app/lineage_ingestion.py index 97f33f948..ca01dac01 100644 --- a/backend/app/lineage_ingestion.py +++ b/backend/app/lineage_ingestion.py @@ -84,22 +84,55 @@ async def load_estimated_channel_weights( ) -> 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. + No anchor method is currently authorized (ADR 0200 point 3 names the + conditions under which one becomes authorized). 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. + + Since migration 0200 one weight set is persisted per active-channel + combination (``channel_set_code``): the corpus-wide rebuild's three + deterministic channels and a scoped analysis run's four each match + their own set without regressing the other. Anything other than an + exact match of one set falls through to ``None`` -- a partial overlap + would mix estimation runs into a vector that grounds nothing. """ 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, " + # Pre-0200 schemas lack channel_set_code; probe via the catalog (never + # a failing statement, which would abort the caller's transaction). + # Pre-0200 rows form one implicit deterministic set. + set_column_exists = await conn.fetchval( + "select exists (select from information_schema.columns " + "where table_schema = 'public' " + " and table_name = 'lineage_channel_weight' " + " and column_name = 'channel_set_code')" + ) + set_column_sql = ( + "channel_set_code" if set_column_exists else "'channel_set_deterministic'" + ) + all_rows = await conn.fetch( + f"select {set_column_sql} as channel_set_code, " + "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" ) + sets: dict[str, list] = {} + for row in all_rows: + sets.setdefault(row["channel_set_code"], []).append(row) + rows = next( + ( + candidate + for candidate in sets.values() + if {row["channel_code"] for row in candidate} == active_channels + ), + [], + ) persisted = {row["channel_code"]: float(row["weight_value"]) for row in rows} if not persisted or set(persisted) != active_channels: return None diff --git a/docs/adr/0200-channel-weight-reconciliation.md b/docs/adr/0200-channel-weight-reconciliation.md new file mode 100644 index 000000000..3a7187ccd --- /dev/null +++ b/docs/adr/0200-channel-weight-reconciliation.md @@ -0,0 +1,178 @@ +# ADR 0200 — Reconciling channel-weight measurement across the two active lines + +**Decision status:** Proposed +**Date:** 2026-08-24 +**Amends:** [ADR 0003](0003-fast-mlsirm-report-integration.md) (scope +boundary), both lines' ADR 0145 (each in part — see Context) +**Answers:** the main-line ADR 0145's "a future proposal must be +ADR-first" conditions, issue #289's durable-worker requirement, and the +operator's no-bulk-synchronous-LLM directive of 2026-08-24. + +> Numbering note: `0200` is deliberately above every number in use on +> `main` (≤ 0167), on `docs/customer-master-scope-adr` (≤ 0145), and in +> any open PR (≤ 0185), so it cannot collide when the lines converge. + +## Context + +Two long-lived lines implemented "ADR 0145" with **opposite decisions**: + +- `docs/customer-master-scope-adr` estimates convex fusion weights with + fast-mlsirm's multilevel 2PL (channels = items, candidate pairs = + respondents, reconstruction groups = cluster intercepts), deletes the + hand-picked `DEFAULT_CHANNEL_WEIGHTS` outright, and fails product + reconstruction closed (HTTP 503 with an estimate-first next action) + whenever no persisted estimate matches the active channels. Weight + sets are keyed by `channel_set_code` (migration 0136). +- `main` records ADR 0145 as a **Rejected proposal** ("Channel-weight + estimation remains unavailable without an independent anchor"), + authorizes zero anchor methods (`_SUPPORTED_ANCHOR_METHOD_CODES = + frozenset()`, so its provenance-gated loader never activates a + vector), stubs the operator estimation command to an unconditional + refusal, and **retains the hand-picked constants "for + compatibility"**. Its `lineage_channel_weight` schema carries per-run + provenance columns instead of `channel_set_code`. + +The same ADR number carrying contradictory decisions violates the +organization's exact-head consistency rule by itself. Beyond the +paperwork, the lines now diverge on the table schema, the loader +contract, the operator tooling, and — most importantly — on whether any +hand-picked constant may keep flowing through product reconstruction. + +The operator's standing directive is explicit and repeated: **no +arbitrary weights anywhere; use weights estimated by a paper-grounded +psychometric model (fast-mlsirm or TEPP)**. The main line's retention of +uncalibrated constants "for compatibility" — however carefully labeled — +keeps exactly those arbitrary numbers in the product and cannot stand +under that directive. + +At the same time, the main line's methodological critique of the first +estimation design is substantially correct and deserves engagement +rather than override: + +1. **Criterion validity.** An unanchored IRT fit describes common + response structure among the channels; it does not by itself prove + that its latent factor is "these two posts are genuinely related." +2. **Conditional information.** Birnbaum's item information is + `I_j(θ) = a_j² P_j(θ) Q_j(θ)` — conditional on trait location — so a + weight proportional to the discrimination alone is not a globally + information-optimal fusion rule (Birnbaum, 1968; Lord, 1980). +3. **Scope boundary.** Accepted ADR 0003 assigned this repository's + fast-mlsirm integration to the LLM-judge/report path; the lineage + weight path must expand that boundary explicitly, not silently. + +Operational facts sharpened this cycle: three estimation runs against +the shared orchestrator died to transport-lifetime failures before one +completed; the 400-pair sequential judge workload measurably saturated +the shared gateway (a concurrent `/api/ask` round-trip reached 158 s), +after which the operator directed that bulk LLM work must use the +repository's durable queue idiom rather than blocking synchronous HTTP; +and the shared development database was rebuilt from `main`, wiping the +imported corpus and every persisted estimate. Sequential-blocking +estimation is architecturally dead independent of the measurement +argument. + +## Decision + +1. **The directive governs both lines.** No hand-picked fusion weight + reaches any product path on any line. The scope line's fail-closed + contract (refuse with an estimate-first next action) becomes the + single product behavior; main's compatibility constants are retired. + Cormack et al.'s (2009) parameter-free reciprocal rank fusion remains + the Rankings surface's rule (no weights exist there to pick). +2. **Expected-information weights** replace discrimination-proportional + weights, answering critique (2): the fusion weight of channel *j* is + the normalized **expected item information over the fitted latent + distribution**, + `w_j ∝ E_θ[I_j(θ)] = ∫ a_j² P_j(θ) Q_j(θ) dF(θ)`, + with `F(θ)` the fitted multilevel latent distribution (mixture over + cluster intercepts). Integrating the conditionality instead of + ignoring it is the standard device of optimal test-design practice + (van der Linden, 2005). Method code: + `mls2plm_expected_information`. Every fit must pass fast-mlsirm's + official diagnostics; any non-converged fit is rejected outright + (`convergence_status`, per the pinned contract). +3. **Anchor honesty**, answering critique (1): estimation activates, but + every persisted set carries `anchor_method_code = + 'unanchored_internal_structure'` until an independent anchor exists, + and provenance (method, estimator version, sample size, snapshot + digest, knowledge cutoff) is surfaced wherever the set is disclosed. + When TEPP reaches production, a criterion-validity gate correlates + fused scores with TEPP's event measurement on a frozen snapshot; a + set that fails the gate is retired and reconstruction fails closed + again. This amends ADR 0003 to authorize the lineage-weights path + explicitly under these conditions. +4. **Schema merge.** `lineage_channel_weight` takes the union of both + lines: primary key `(channel_set_code, channel_code)` from the scope + line — one persisted set per active-channel combination — plus the + main line's per-run provenance columns (`estimation_run_id`, + `estimation_method_code`, `estimator_version`, `anchor_method_code`, + `source_snapshot_sha256`, `sample_pair_count`, `knowledge_cutoff`). + The loader requires an exact active-channel match AND single-run + provenance integrity AND an authorized anchor method code + (`unanchored_internal_structure` joins the authorized set under + point 3's labeling duty). One migration with rollbacks lands the + union on whichever predecessor schema a database has. +5. **Queued judge scoring.** The llm channel's pair scoring moves to the + repository's durable queue idiom (`post_content_queue` / + `post_content_worker` family): the operator command samples pairs + deterministically, persists the run identity and per-pair job rows, + and publishes Valkey wake-ups; a bounded worker drains judge jobs at + a governed rate through contextual-orchestrator and persists each + pair's score durably as it lands. The fit runs only when a run's + pairs are complete, so a killed process loses nothing and re-running + resumes instead of re-spending provider calls. This satisfies the + operator's no-bulk-synchronous-LLM directive and issue #289's + bounded-durable-worker requirement in one design. +6. **Convergence sequencing.** This ADR lands verbatim on both lines; + each line's ADR 0145 gains a superseded-in-part pointer to it. + Implementation lands on `main` first (the shared development + environment now runs the main line), and `docs/customer-master-scope-adr` + rebases its weights stack onto the merged schema. Corpus re-import + and a fresh expected-information estimation run follow on the merged + head — in that order, since estimation samples the imported corpus. + +## Consequences + +**Positive.** One contract instead of two contradictory ones; the +operator directive holds everywhere; the estimator answers the +strongest published objection to its own first design; a killed +estimation run stops costing hours of provider spend; the shared +gateway is never again saturated by a scoring loop; and TEPP gains a +named, gated integration point instead of an implied one. + +**Negative.** A schema migration on both predecessors; the llm channel +estimate waits for the queue worker to land; and until the TEPP gate +exists the activated weights remain honestly labeled as internally +anchored only — reviewers must weigh that label rather than a hard +criterion coefficient. + +## 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 + +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 + +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 + +van der Linden, W. J. (2005). *Linear models for optimal test design*. +Springer. https://doi.org/10.1007/0-387-29054-0 diff --git a/lineageweave/adjudication_client.py b/lineageweave/adjudication_client.py index 37e4fee7f..ce020fc70 100644 --- a/lineageweave/adjudication_client.py +++ b/lineageweave/adjudication_client.py @@ -41,6 +41,36 @@ def judge(self, candidate_label: str, record_label: str) -> float: # pragma: no _CONFIDENCE_PATTERN = re.compile(r"([01](?:\.\d+)?)") +def judge_prompt(candidate_label: str, record_label: str) -> str: + """The one adjudication prompt, shared by the live client and the + queued batch scorer so both channels ask the identical question.""" + return ( + "On a scale from 0.0 (definitely unrelated) to 1.0 (definitely the same " + "thread, B directly follows from A), how confident are you that record B " + "is a direct continuation of record A? Reply with only the number.\n\n" + f"Record A: {candidate_label}\nRecord B: {record_label}" + ) + + +def parse_confidence(content: str) -> float: + """Clamp the judge's numeric reply into [0, 1]; no number reads as 0.""" + parsed = parse_confidence_or_none(content) + return 0.0 if parsed is None else parsed + + +def parse_confidence_or_none(content: str) -> float | None: + """Like :func:`parse_confidence`, but an unparseable reply is ``None``. + + The queued batch scorer must distinguish "the judge said 0.0" from + "the judge failed to answer" -- persisting the latter as a confident + zero would fabricate an unrelated verdict for an errored request. + """ + match = _CONFIDENCE_PATTERN.search(content) + if match is None: + return None + return max(0.0, min(1.0, float(match.group(1)))) + + class ContextualOrchestratorAdjudicationClient: """Calls ``POST {base_url}/v1/chat/completions`` with ``mode="auto"``. @@ -60,24 +90,16 @@ def __init__( def judge(self, candidate_label: str, record_label: str) -> float: """Score the candidate and record labels for semantic adjudication.""" - prompt = ( - "On a scale from 0.0 (definitely unrelated) to 1.0 (definitely the same " - "thread, B directly follows from A), how confident are you that record B " - "is a direct continuation of record A? Reply with only the number.\n\n" - f"Record A: {candidate_label}\nRecord B: {record_label}" - ) body = post_json( f"{self._base_url}/v1/chat/completions", { - "messages": [{"role": "user", "content": prompt}], + "messages": [ + {"role": "user", "content": judge_prompt(candidate_label, record_label)} + ], "mode": "auto", "reasoning_effort": self._reasoning_effort, }, headers={"authorization": f"Bearer {self._api_key}"}, timeout=self._timeout, ) - content = chat_completion_content(body) - match = _CONFIDENCE_PATTERN.search(content) - if match is None: - return 0.0 - return max(0.0, min(1.0, float(match.group(1)))) + return parse_confidence(chat_completion_content(body)) diff --git a/lineageweave/channel_weight_estimation.py b/lineageweave/channel_weight_estimation.py index df9388c4b..6c9b965ae 100644 --- a/lineageweave/channel_weight_estimation.py +++ b/lineageweave/channel_weight_estimation.py @@ -1,24 +1,230 @@ -"""Fail-closed lineage channel-weight boundary (ADR 0145). +"""Psychometric estimation of lineage channel-fusion weights (ADR 0200). -Channel-score covariance is not an independent lineage anchor. Until an -accepted upstream anchor contract exists, this module produces no estimate. +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; Fox & Glas, 2001, for the +multilevel IRT structure). + +Because Birnbaum's item information is conditional on trait location -- +``I_j(theta) = a_j^2 P_j(theta) Q_j(theta)`` (Birnbaum, 1968; Lord, +1980) -- a weight proportional to the discrimination alone is not a +global information-optimal rule. The fusion weight here is therefore +the normalized EXPECTED item information over the fitted latent +distribution, approximated on the fitted person parameters with the +package's own item response function (van der Linden, 2005, on +expected/target information as the design quantity). Estimates carry +the ``mls2plm_expected_information`` method code; activation +additionally requires an authorized anchor method (ADR 0200 point 3) +enforced by the product loader, not here. + +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), the fit +does not converge, or any estimate is non-finite, +:func:`estimate_channel_weights` returns ``None`` and the caller fails +closed -- product paths refuse to reconstruct, the demo refuses to fuse +-- it never fabricates a "grounded" weight. """ from __future__ import annotations +import math +import random +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 + +# The library demo's declared generative design (fixtures.sample_records, +# `make seed`, the standalone demo server): per-channel follow +# probabilities of the latent "genuinely related" trait, per-group +# relatedness base rates, and a fixed simulation seed. These are the +# demo scenario's TRUE parameters -- synthetic demo data, never fusion +# weights. The weights the demo fuses with are ESTIMATED from this +# design by fast-mlsirm, exactly like production weights are estimated +# from the real corpus (ADR 0200: no hand-picked +# fusion weight exists anywhere, demo included). +_FIXTURE_FOLLOW_PROBABILITY = {"temporal": 0.80, "secondary_key": 0.72, "text": 0.66} +_FIXTURE_GROUP_COUNT = 12 +_FIXTURE_PAIR_COUNT = 900 +_FIXTURE_SIMULATION_SEED = 20260824 + + +def simulate_fixture_pair_scores() -> tuple[list[dict[str, float]], list[int]]: + """Simulate the demo design's channel responses, deterministically. + + Each simulated pair carries a latent related/unrelated state drawn + from its group's base rate (genuine cluster intercept variance -- + the structure MLS2PLM's multilevel random intercept models); each + channel then reports a high or low score according to its declared + follow probability. The fixed seed keeps every ``make seed`` and + demo-server estimate identical run to run. + """ + generator = random.Random(_FIXTURE_SIMULATION_SEED) + + def channel_score(related: bool, follow_probability: float) -> float: + """One channel's noisy report of the pair's latent related state.""" + 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) + + group_base_rate = [ + generator.uniform(0.25, 0.75) for _ in range(_FIXTURE_GROUP_COUNT) + ] + pair_scores: list[dict[str, float]] = [] + group_ids: list[int] = [] + for index in range(_FIXTURE_PAIR_COUNT): + group = index % _FIXTURE_GROUP_COUNT + related = generator.random() < group_base_rate[group] + pair_scores.append( + { + channel: channel_score(related, follow_probability) + for channel, follow_probability in _FIXTURE_FOLLOW_PROBABILITY.items() + } + ) + group_ids.append(group) + return pair_scores, group_ids + + +def estimate_fixture_channel_weights() -> ChannelWeightEstimate | None: + """Estimate the demo's deterministic-channel weights from its design. + + Returns ``None`` when no grounded estimate can be produced (most + commonly: ``fast_mlsirm`` is not importable); demo callers then fail + closed -- the seed and the standalone server refuse to fuse with + invented weights and instead name the next action (install + fast-mlsirm from the organization repo). + """ + pair_scores, group_ids = simulate_fixture_pair_scores() + return estimate_channel_weights(pair_scores, group_ids) + + +@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 (the dichotomization rule both lines' ADR 0145 texts share, + carried forward by ADR 0200). + """ + return 1 if score >= threshold else 0 + def estimate_channel_weights( pair_channel_scores: list[dict[str, float]], group_ids: list[int], -) -> None: - """Report unavailable for unanchored channel scores. +) -> ChannelWeightEstimate | None: + """Estimate convex fusion weights from observed channel scores. Args: - pair_channel_scores: unanchored candidate-pair channel scores. - group_ids: corresponding reconstruction-group indexes. + 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 ``cluster_id``. Returns: - Always ``None`` until ADR 0145's independent-anchor requirement is met. + 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, predict_proba + 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. + factor_id = numpy.zeros(len(channels), dtype=numpy.int64) + result = fit( + responses=numpy.asarray(responses, dtype=float), + factor_id=factor_id, + cluster_id=numpy.asarray(group_ids, dtype=numpy.int64), + # fast-mlsirm's default max_iter=1000 is tuned against its GPU/f32 + # path; the f64 CPU fallback (no wgpu adapter -- every CI runner) + # needs materially more EM iterations to reach the same optimum at + # full precision, observed up to ~1850 on this module's own fixture. + # Raising the budget only slows an already-non-converged path; a + # fit that would converge sooner still stops the moment it does. + config=FitConfig(model="MLS2PLM", latent_dim=1, estimator="mmle", max_iter=3000), + ) + # ADR 0200: a non-converged fit is rejected outright -- its point + # estimates are not measurement evidence. + if result.convergence_status != "converged": + return None + discriminations = numpy.asarray(result.params.a, dtype=float).ravel() + if len(discriminations) != len(channels): + return None + if not numpy.all(numpy.isfinite(discriminations)): + return None + + # ADR 0200 point 2: Birnbaum item information is conditional, + # I_j(theta) = a_j^2 P_j(theta) Q_j(theta) -- so the fusion weight is + # the normalized EXPECTED information over the fitted latent + # distribution, approximated by averaging over the fitted person + # parameters (the empirical distribution the multilevel model + # produced), using the package's own item response function + # (predict_proba) rather than a re-derived one (van der Linden, + # 2005, on expected/target information as the design quantity). + probabilities = numpy.asarray(predict_proba(result.params, factor_id), dtype=float) + if probabilities.shape[1] != len(channels): + return None + information = (discriminations**2) * probabilities * (1.0 - probabilities) + expected_information = information.mean(axis=0) + if not numpy.all(numpy.isfinite(expected_information)): + return None + total = float(expected_information.sum()) + if not math.isfinite(total) or total <= 0: + return None + return ChannelWeightEstimate( + weights={ + channel: float(value) / total + for channel, value in zip(channels, expected_information) + }, + sample_pair_count=len(pair_channel_scores), + estimation_method_code="mls2plm_expected_information", + ) diff --git a/migrations/0200_channel_weight_schema_union.sql b/migrations/0200_channel_weight_schema_union.sql new file mode 100644 index 000000000..009cc78e3 --- /dev/null +++ b/migrations/0200_channel_weight_schema_union.sql @@ -0,0 +1,88 @@ +-- ADR 0200 point 4: union of the two lines' lineage_channel_weight schemas. +-- +-- Predecessors this must upgrade from, idempotently (ADR 0166 replay window): +-- * main's 0135: primary key (channel_code) + per-run provenance columns. +-- * the customer-master line's 0135+0136: primary key +-- (channel_set_code, channel_code), no provenance columns. +-- Target: primary key (channel_set_code, channel_code) -- one persisted set +-- per active-channel combination -- carrying main's full provenance contract. +-- +-- Rows persisted before the provenance contract existed are unverifiable +-- estimates; they are deleted rather than backfilled with invented +-- provenance (the loader refuses them either way, and re-running +-- scripts/estimate_channel_weights.py is the operator's next action). + +do $$ +begin + if not exists ( + select from information_schema.columns + where table_name = 'lineage_channel_weight' + and column_name = 'estimation_run_id' + ) then + delete from lineage_channel_weight; + end if; +end $$; + +alter table lineage_channel_weight + add column if not exists channel_set_code text not null + default 'channel_set_deterministic', + add column if not exists estimation_run_id uuid not null, + add column if not exists estimation_method_code text not null, + add column if not exists estimator_version text not null, + add column if not exists anchor_method_code text not null, + add column if not exists source_snapshot_sha256 text not null, + add column if not exists sample_pair_count bigint not null, + add column if not exists knowledge_cutoff timestamptz not null, + add column if not exists estimated_at timestamptz not null default now(); + +do $$ +declare + primary_key_columns text; +begin + select string_agg(a.attname, ',' order by array_position(x.conkey, a.attnum)) + into primary_key_columns + from pg_constraint x + join pg_attribute a + on a.attrelid = x.conrelid and a.attnum = any(x.conkey) + where x.conrelid = 'lineage_channel_weight'::regclass + and x.contype = 'p'; + if primary_key_columns is distinct from 'channel_set_code,channel_code' then + execute 'alter table lineage_channel_weight drop constraint if exists lineage_channel_weight_pkey'; + execute 'alter table lineage_channel_weight add primary key (channel_set_code, channel_code)'; + end if; +end $$; + +alter table lineage_channel_weight + drop constraint if exists lineage_channel_set_code_check; +alter table lineage_channel_weight + add constraint lineage_channel_set_code_check + check (channel_set_code in ('channel_set_deterministic', 'channel_set_with_llm')); + +-- Re-assert main's 0135 integrity constraints for databases coming from the +-- customer-master predecessor, which never had them. +do $$ +begin + if not exists ( + select from pg_constraint + where conrelid = 'lineage_channel_weight'::regclass + and conname = 'lineage_channel_code_check' + ) then + alter table lineage_channel_weight + add constraint lineage_channel_code_check + check (channel_code in ('temporal', 'secondary_key', 'text', 'llm')), + add constraint lineage_weight_value_check + check (weight_value > 0 and weight_value <= 1), + add constraint lineage_estimation_method_check + check (btrim(estimation_method_code) <> ''), + add constraint lineage_estimator_version_check + check (btrim(estimator_version) <> ''), + add constraint lineage_anchor_method_check + check (btrim(anchor_method_code) <> ''), + add constraint lineage_sample_pair_count_check + check (sample_pair_count >= 200), + add constraint lineage_source_snapshot_check + check (source_snapshot_sha256 ~ '^[0-9a-f]{64}$'), + add constraint lineage_knowledge_cutoff_check + check (knowledge_cutoff <= estimated_at); + end if; +end $$; diff --git a/migrations/0201_lineage_pair_judgment.sql b/migrations/0201_lineage_pair_judgment.sql new file mode 100644 index 000000000..93dd0f1be --- /dev/null +++ b/migrations/0201_lineage_pair_judgment.sql @@ -0,0 +1,58 @@ +-- ADR 0200 point 5: durable, resumable llm pair judging. +-- +-- Bulk synchronous provider calls are banned (operator directive, +-- 2026-08-24): the operator submits the sampled pairs as ONE +-- contextual-orchestrator batch routing job (its Valkey-backed registry +-- survives orchestrator restarts) and each returned score persists here +-- as it is collected, so a killed collection loses nothing and the fit +-- runs only over a complete run. + +create table if not exists lineage_weight_estimation_run ( + estimation_run_id uuid primary key, + channel_set_code text not null, + run_status_code text not null, + batch_job_id text not null, + source_snapshot_sha256 text not null, + knowledge_cutoff timestamptz not null, + sampled_pair_count bigint not null, + judged_pair_count bigint not null default 0, + requested_at timestamptz not null default now(), + completed_at timestamptz, + constraint estimation_run_status_check + check (run_status_code in + ('run_submitted', 'run_collecting', 'run_fitted', 'run_failed')), + constraint estimation_run_set_check + check (channel_set_code in + ('channel_set_deterministic', 'channel_set_with_llm')), + constraint estimation_run_snapshot_check + check (source_snapshot_sha256 ~ '^[0-9a-f]{64}$'), + constraint estimation_run_pair_count_check + check (sampled_pair_count > 0), + constraint estimation_run_judged_count_check + check (judged_pair_count >= 0 and judged_pair_count <= sampled_pair_count) +); + +create table if not exists lineage_pair_judgment ( + estimation_run_id uuid not null + references lineage_weight_estimation_run (estimation_run_id) + on delete cascade, + pair_ordinal bigint not null, + group_ordinal bigint not null, + candidate_label text not null, + record_label text not null, + temporal_score double precision not null, + secondary_key_score double precision not null, + text_score double precision not null, + llm_score double precision, + judged_at timestamptz, + primary key (estimation_run_id, pair_ordinal), + constraint pair_judgment_scores_check + check ( + temporal_score between 0 and 1 + and secondary_key_score between 0 and 1 + and text_score between 0 and 1 + and (llm_score is null or llm_score between 0 and 1) + ), + constraint pair_judgment_judged_check + check ((llm_score is null) = (judged_at is null)) +); diff --git a/migrations/rollback/0200_channel_weight_schema_union.sql b/migrations/rollback/0200_channel_weight_schema_union.sql new file mode 100644 index 000000000..d71a9903e --- /dev/null +++ b/migrations/rollback/0200_channel_weight_schema_union.sql @@ -0,0 +1,14 @@ +-- Rollback of 0200: restore main's 0135 single-set shape. Non-deterministic +-- sets cannot exist under a (channel_code) primary key, so they are removed. + +delete from lineage_channel_weight + where channel_set_code <> 'channel_set_deterministic'; + +alter table lineage_channel_weight + drop constraint if exists lineage_channel_set_code_check; +alter table lineage_channel_weight + drop constraint if exists lineage_channel_weight_pkey; +alter table lineage_channel_weight + drop column if exists channel_set_code; +alter table lineage_channel_weight + add primary key (channel_code); diff --git a/migrations/rollback/0201_lineage_pair_judgment.sql b/migrations/rollback/0201_lineage_pair_judgment.sql new file mode 100644 index 000000000..69069bf73 --- /dev/null +++ b/migrations/rollback/0201_lineage_pair_judgment.sql @@ -0,0 +1,3 @@ +-- Rollback of 0201: the queued-judging ledger is derived operator state. +drop table if exists lineage_pair_judgment; +drop table if exists lineage_weight_estimation_run; diff --git a/scripts/estimate_channel_weights.py b/scripts/estimate_channel_weights.py index e39e6282e..b8ab9534f 100644 --- a/scripts/estimate_channel_weights.py +++ b/scripts/estimate_channel_weights.py @@ -1,26 +1,243 @@ -"""Fail-closed operator boundary for lineage channel weights (ADR 0145). +"""Operator estimation of lineage channel-fusion weights (ADR 0200). -No independent lineage anchor is currently authorized, so this command does -not query the corpus, fit a model, or write weights. +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 three deterministic channels, fits +`fast-mlsirm`'s multilevel 2PL over the dichotomized scores, and +persists the normalized expected-information weights into +`lineage_channel_weight` (migration 0200) with full per-run provenance: +run identity, estimator version, anchor method, a reproducible source +snapshot digest, sample size, and the knowledge cutoff. + +Persisting is not activating: the product loader refuses every anchor +method until one is authorized under ADR 0200 point 3, so rows written +here are inert evidence until that authorization lands. The llm +channel is deliberately absent -- bulk synchronous provider calls are +banned (operator directive, 2026-08-24); llm pair scoring arrives with +the queued worker (ADR 0200 point 5). + +No database connection is held across the scoring/fitting phase +(a reaped idle connection killed an earlier run): one short-lived +connection fetches rows, none is open while fitting, and a fresh one +persists the estimate. """ from __future__ import annotations import argparse import asyncio +import hashlib import json +import uuid +from datetime import datetime + +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 + +DETERMINISTIC_SET_CODE = "channel_set_deterministic" +# ADR 0200 point 3: honest label for an estimate whose latent factor is +# validated only by the channels' internal response structure, pending +# the TEPP criterion-validity gate. +UNANCHORED_METHOD_CODE = "unanchored_internal_structure" + +def estimator_version() -> str: + """The installed fast-mlsirm version, for the persisted provenance.""" + from importlib.metadata import PackageNotFoundError, version -async def _run(_args: argparse.Namespace) -> dict[str, object]: - """Fail closed before reading any corpus while no anchor is authorized.""" - 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" + for name in ("fast-mlsirm", "fast_mlsirm"): + try: + return version(name) + except PackageNotFoundError: + continue + import fast_mlsirm + + return str(getattr(fast_mlsirm, "__version__", "unknown")) + + +def source_snapshot_digest(rows: list) -> str: + """Reproducible SHA-256 over the ordered sampled (post_id, created_at). + + Two runs that sampled the same posts in the same order produce the + same digest, so the provenance row names exactly which corpus slice + supported the estimate without storing any post content. + """ + material = "\n".join( + f"{row['post_id']}\t{row['created_at'].isoformat()}" for row in rows ) + return hashlib.sha256(material.encode("utf-8")).hexdigest() + + +def sample_pair_scores( + records: list, *, window: int = DEFAULT_CANDIDATE_WINDOW +) -> tuple[list[dict[str, float]], list[int], list[tuple[str, str]]]: + """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`` would consider. Also returns each pair's + (candidate_label, record_label) so the queued llm judging pass can + score the same candidate geometry without re-deriving it. + """ + 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] = [] + pair_labels: list[tuple[str, str]] = [] + 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) + pair_labels.append((candidate.label, record.label)) + return pair_scores, group_ids, pair_labels + + +def subsample_stride(total: int, limit: int) -> list[int]: + """Deterministic, evenly-spread pair indices for the bounded llm pass. + + A stride subsample keeps every reconstruction group represented in + proportion (pairs are ordered group-by-group) without any randomness + that would make re-runs incomparable. + """ + if total <= limit: + return list(range(total)) + stride = total / limit + return [min(int(index * stride), total - 1) for index in range(limit)] + + +async def persist_estimate( + conn: asyncpg.Connection, + estimate: ChannelWeightEstimate, + *, + channel_set_code: str, + snapshot_sha256: str, + knowledge_cutoff: datetime, +) -> str: + """Replace one channel set's persisted weights atomically, with provenance. + + Returns the estimation run id stamped on every row of the set. + """ + estimation_run_id = str(uuid.uuid4()) + version = estimator_version() + async with conn.transaction(): + await conn.execute( + "delete from lineage_channel_weight where channel_set_code = $1", + channel_set_code, + ) + for channel, weight in estimate.weights.items(): + await conn.execute( + """ + insert into lineage_channel_weight + (channel_set_code, channel_code, weight_value, + estimation_run_id, estimation_method_code, + estimator_version, anchor_method_code, + source_snapshot_sha256, sample_pair_count, + knowledge_cutoff) + values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) + """, + channel_set_code, + channel, + weight, + estimation_run_id, + estimate.estimation_method_code, + version, + UNANCHORED_METHOD_CODE, + snapshot_sha256, + estimate.sample_pair_count, + knowledge_cutoff, + ) + return estimation_run_id + + +async def _run(args: argparse.Namespace) -> dict[str, object]: + settings = load_settings() + # Short-lived fetch connection; nothing stays open while fitting. + conn = await asyncpg.connect(settings.database_url) + try: + 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, + ) + finally: + await conn.close() + if not rows: + raise RuntimeError( + "no eligible source posts exist; import a corpus before estimating" + ) + snapshot_sha256 = source_snapshot_digest(rows) + knowledge_cutoff = max(row["created_at"] for row in rows) + records = records_from_source_posts(rows) + pair_scores, group_ids, _pair_labels = 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, " + "sample too small, a channel degenerate, or the fit did not " + "converge) -- nothing was written; run again after fixing the " + "named condition" + ) + estimation_run_id = None + if not args.dry_run: + conn = await asyncpg.connect(settings.database_url) + try: + estimation_run_id = await persist_estimate( + conn, + estimate, + channel_set_code=DETERMINISTIC_SET_CODE, + snapshot_sha256=snapshot_sha256, + knowledge_cutoff=knowledge_cutoff, + ) + finally: + await conn.close() + return { + "weights": estimate.weights, + "channel_set_code": DETERMINISTIC_SET_CODE, + "sample_pair_count": estimate.sample_pair_count, + "estimation_method_code": estimate.estimation_method_code, + "anchor_method_code": UNANCHORED_METHOD_CODE, + "estimation_run_id": estimation_run_id, + "source_snapshot_sha256": snapshot_sha256, + "knowledge_cutoff": knowledge_cutoff.isoformat(), + "persisted": not args.dry_run, + "activation": ( + "blocked_until_anchor_authorized (ADR 0200 point 3): the " + "product loader refuses every anchor method today, so these " + "rows are inert evidence" + ), + } def main() -> None: - """Validate operator inputs and report only the policy-boundary result.""" + """Validate operator inputs and run the estimation.""" parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--post-limit", diff --git a/scripts/estimate_llm_channel_weights.py b/scripts/estimate_llm_channel_weights.py new file mode 100644 index 000000000..70a35de9d --- /dev/null +++ b/scripts/estimate_llm_channel_weights.py @@ -0,0 +1,433 @@ +"""Queued llm-inclusive channel-weight estimation (ADR 0200 point 5). + +Bulk synchronous provider calls are banned (operator directive, +2026-08-24), so the llm channel is scored through +contextual-orchestrator's durable batch routing API instead: + +``submit`` + samples candidate pairs exactly as the deterministic estimator does, + takes a bounded deterministic stride subsample, submits ONE batch + routing job (one request per pair, ``custom_id=pair-``), + and persists the run plus every pair's deterministic scores into + ``lineage_weight_estimation_run`` / ``lineage_pair_judgment`` + (migration 0201). It never waits on the provider. + +``collect`` + polls the batch job once; when complete it retrieves the results, + maps each score back to its pair by ``custom_id`` (caller-supplied + ids landed upstream for exactly this — contextual-orchestrator + #832), persists per-pair llm scores durably, and only when the run + is complete fits the 4-channel expected-information estimate and + persists it as the ``channel_set_with_llm`` set with full + provenance. Killed mid-collect, nothing is lost: re-run ``collect``. + +Persisting is not activating: the product loader refuses every anchor +method until one is authorized under ADR 0200 point 3. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import os +from datetime import datetime, timezone + +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.adjudication_client import judge_prompt, parse_confidence_or_none +from lineageweave.channel_weight_estimation import estimate_channel_weights +from lineageweave.http_client import get_json, post_json + +from scripts.estimate_channel_weights import ( + persist_estimate, + sample_pair_scores, + source_snapshot_digest, + subsample_stride, +) + +WITH_LLM_SET_CODE = "channel_set_with_llm" +_BATCH_TIMEOUT_SECONDS = 60.0 + + +def _orchestrator_config() -> tuple[str, str]: + """Base URL and bearer key for the batch routing API, from the environment.""" + base_url = next( + ( + os.environ[name].strip() + for name in ("ORCHESTRATOR_BASE_URL", "LLM_GATEWAY_API_URL") + if os.environ.get(name, "").strip() + ), + "", + ) + api_key = next( + ( + os.environ[name].strip() + for name in ("ORCHESTRATOR_API_KEY", "CONTEXTUAL_ORCHESTRATOR_TOKEN") + if os.environ.get(name, "").strip() + ), + "", + ) + if not base_url or not api_key: + raise RuntimeError( + "set ORCHESTRATOR_BASE_URL and ORCHESTRATOR_API_KEY (or " + "CONTEXTUAL_ORCHESTRATOR_TOKEN) to reach the batch routing API" + ) + return base_url.rstrip("/"), api_key + + +def batch_requests_for_pairs( + chosen: list[int], pair_labels: list[tuple[str, str]] +) -> list[dict[str, object]]: + """One batch request per chosen pair, keyed by its ordinal. + + Every request carries a caller-supplied ``custom_id`` and none rely + on the server-generated ids, so results map back to pairs on any + backend regardless of result ordering (and per upstream guidance, + caller and generated ids are never mixed within one batch). + """ + return [ + { + "custom_id": f"pair-{ordinal}", + "mode": "auto", + "messages": [ + { + "role": "user", + "content": judge_prompt(*pair_labels[ordinal]), + } + ], + } + for ordinal in chosen + ] + + +async def _submit(args: argparse.Namespace) -> dict[str, object]: + """Sample, submit one batch job, persist the run ledger. Never waits.""" + base_url, api_key = _orchestrator_config() + settings = load_settings() + conn = await asyncpg.connect(settings.database_url) + try: + 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, + ) + finally: + await conn.close() + if not rows: + raise RuntimeError( + "no eligible source posts exist; import a corpus before estimating" + ) + snapshot_sha256 = source_snapshot_digest(rows) + knowledge_cutoff = max(row["created_at"] for row in rows) + pair_scores, group_ids, pair_labels = sample_pair_scores( + records_from_source_posts(rows) + ) + chosen = subsample_stride(len(pair_scores), args.pair_limit) + if not chosen: + raise RuntimeError("the corpus produced no candidate pairs to judge") + + submitted = post_json( + f"{base_url}/api/v1/batch_routing_jobs", + {"requests": batch_requests_for_pairs(chosen, pair_labels)}, + headers={"authorization": f"Bearer {api_key}"}, + timeout=_BATCH_TIMEOUT_SECONDS, + ) + batch_job_id = str(submitted["job_id"]) + + conn = await asyncpg.connect(settings.database_url) + try: + async with conn.transaction(): + estimation_run_id = await conn.fetchval( + """ + insert into lineage_weight_estimation_run + (estimation_run_id, channel_set_code, run_status_code, + batch_job_id, source_snapshot_sha256, knowledge_cutoff, + sampled_pair_count) + values (gen_random_uuid(), $1, 'run_submitted', $2, $3, $4, $5) + returning estimation_run_id + """, + WITH_LLM_SET_CODE, + batch_job_id, + snapshot_sha256, + knowledge_cutoff, + len(chosen), + ) + for ordinal in chosen: + scores = pair_scores[ordinal] + candidate_label, record_label = pair_labels[ordinal] + await conn.execute( + """ + insert into lineage_pair_judgment + (estimation_run_id, pair_ordinal, group_ordinal, + candidate_label, record_label, temporal_score, + secondary_key_score, text_score) + values ($1, $2, $3, $4, $5, $6, $7, $8) + """, + estimation_run_id, + ordinal, + group_ids[ordinal], + candidate_label, + record_label, + scores["temporal"], + scores["secondary_key"], + scores["text"], + ) + except Exception as exc: + raise RuntimeError( + f"batch job {batch_job_id} was submitted but the run ledger " + "could not be persisted; re-run submit (the orphaned job only " + "costs its provider spend, no state references it)" + ) from exc + finally: + await conn.close() + return { + "estimation_run_id": str(estimation_run_id), + "batch_job_id": batch_job_id, + "sampled_pair_count": len(chosen), + "next_action": "run collect once the batch job completes", + } + + +def _is_complete(polled: dict[str, object]) -> bool: + """True when the batch backend reports a terminal successful state.""" + if polled.get("is_complete") is True: + return True + return str(polled.get("status", "")).lower() in {"completed", "succeeded"} + + +def judgment_updates_from_results( + results: list[dict[str, object]], +) -> list[tuple[int, float]]: + """Map batch results onto (pair_ordinal, llm_score) updates. + + Mapping is by caller-supplied ``custom_id`` only -- never result + order. An unparseable or empty answer is OMITTED, not stored: an + errored request must stay unjudged rather than become a confident + 0.0 ("definitely unrelated") verdict the judge never gave. + """ + updates: list[tuple[int, float]] = [] + for item in results: + custom_id = str(item.get("custom_id", "")) + if not custom_id.startswith("pair-"): + continue + try: + ordinal = int(custom_id.removeprefix("pair-")) + except ValueError: + continue + score = parse_confidence_or_none(str(item.get("answer", ""))) + if score is None: + continue + updates.append((ordinal, score)) + return updates + + +async def _collect(args: argparse.Namespace) -> dict[str, object]: + """Collect one completed batch into the ledger; fit when the run is whole. + + No database connection is held across the HTTP calls or the model + fit (an idle-reaped connection killed an earlier estimation run): + each phase opens its own short-lived connection. + """ + base_url, api_key = _orchestrator_config() + settings = load_settings() + + conn = await asyncpg.connect(settings.database_url) + try: + if args.run_id: + run = await conn.fetchrow( + """ + select estimation_run_id, batch_job_id, run_status_code, + source_snapshot_sha256, knowledge_cutoff, sampled_pair_count + from lineage_weight_estimation_run + where estimation_run_id = $1::uuid + and run_status_code in ('run_submitted', 'run_collecting') + """, + args.run_id, + ) + else: + run = await conn.fetchrow( + """ + select estimation_run_id, batch_job_id, run_status_code, + source_snapshot_sha256, knowledge_cutoff, sampled_pair_count + from lineage_weight_estimation_run + where run_status_code in ('run_submitted', 'run_collecting') + order by requested_at desc + limit 1 + """ + ) + finally: + await conn.close() + if run is None: + raise RuntimeError( + "no submitted run awaits collection; run submit first " + "(or pass --run-id for an older run)" + ) + + polled = get_json( + f"{base_url}/api/v1/batch_routing_jobs/{run['batch_job_id']}", + headers={"authorization": f"Bearer {api_key}"}, + timeout=_BATCH_TIMEOUT_SECONDS, + ) + if not _is_complete(polled): + return { + "estimation_run_id": str(run["estimation_run_id"]), + "batch_job_id": run["batch_job_id"], + "batch_status": polled.get("status"), + "next_action": "batch not complete yet; run collect again later", + } + + retrieved = post_json( + f"{base_url}/api/v1/batch_routing_jobs/{run['batch_job_id']}/results", + {}, + headers={"authorization": f"Bearer {api_key}"}, + timeout=_BATCH_TIMEOUT_SECONDS, + ) + updates = judgment_updates_from_results(retrieved.get("results", [])) + judged_at = datetime.now(timezone.utc) + + conn = await asyncpg.connect(settings.database_url) + try: + async with conn.transaction(): + for ordinal, score in updates: + await conn.execute( + """ + update lineage_pair_judgment + set llm_score = $3, judged_at = $4 + where estimation_run_id = $1 and pair_ordinal = $2 + """, + run["estimation_run_id"], + ordinal, + score, + judged_at, + ) + await conn.execute( + """ + update lineage_weight_estimation_run + set run_status_code = 'run_collecting', + judged_pair_count = ( + select count(*) from lineage_pair_judgment + where estimation_run_id = $1 and llm_score is not null + ) + where estimation_run_id = $1 + """, + run["estimation_run_id"], + ) + pairs = await conn.fetch( + """ + select group_ordinal, temporal_score, secondary_key_score, + text_score, llm_score + from lineage_pair_judgment + where estimation_run_id = $1 + order by pair_ordinal + """, + run["estimation_run_id"], + ) + finally: + await conn.close() + + unjudged = sum(1 for row in pairs if row["llm_score"] is None) + if unjudged: + return { + "estimation_run_id": str(run["estimation_run_id"]), + "judged_pair_count": len(pairs) - unjudged, + "sampled_pair_count": len(pairs), + "next_action": ( + f"{unjudged} pairs have no parseable judgment yet; run " + "collect again once the batch delivers them, or re-submit " + "if the provider errored them permanently" + ), + } + + # The fit can take minutes; no connection is open while it runs. + estimate = estimate_channel_weights( + [ + { + "temporal": row["temporal_score"], + "secondary_key": row["secondary_key_score"], + "text": row["text_score"], + "llm": row["llm_score"], + } + for row in pairs + ], + [int(row["group_ordinal"]) for row in pairs], + ) + + conn = await asyncpg.connect(settings.database_url) + try: + if estimate is None: + await conn.execute( + "update lineage_weight_estimation_run " + "set run_status_code = 'run_failed', completed_at = now() " + "where estimation_run_id = $1", + run["estimation_run_id"], + ) + raise RuntimeError( + "no grounded estimate was produced over the judged pairs " + "(fast_mlsirm unavailable, sample too small, a channel " + "degenerate, or the fit did not converge) -- the run is " + "marked run_failed; nothing was written to the weight table" + ) + await persist_estimate( + conn, + estimate, + channel_set_code=WITH_LLM_SET_CODE, + snapshot_sha256=run["source_snapshot_sha256"], + knowledge_cutoff=run["knowledge_cutoff"], + ) + await conn.execute( + "update lineage_weight_estimation_run " + "set run_status_code = 'run_fitted', completed_at = now() " + "where estimation_run_id = $1", + run["estimation_run_id"], + ) + finally: + await conn.close() + return { + "estimation_run_id": str(run["estimation_run_id"]), + "weights": estimate.weights, + "channel_set_code": WITH_LLM_SET_CODE, + "sample_pair_count": estimate.sample_pair_count, + "estimation_method_code": estimate.estimation_method_code, + "activation": ( + "blocked_until_anchor_authorized (ADR 0200 point 3): the " + "product loader refuses every anchor method today" + ), + } + + +def main() -> None: + """Validate operator inputs and run the chosen phase.""" + parser = argparse.ArgumentParser(description=__doc__) + subcommands = parser.add_subparsers(dest="phase", required=True) + submit = subcommands.add_parser("submit", help="sample pairs and submit one batch job") + submit.add_argument("--post-limit", type=int, default=5000) + submit.add_argument("--pair-limit", type=int, default=400) + collect = subcommands.add_parser( + "collect", help="collect results; fit when the run is whole" + ) + collect.add_argument( + "--run-id", + default="", + help="collect a specific estimation run (default: the newest awaiting one)", + ) + args = parser.parse_args() + if args.phase == "submit": + if args.post_limit < 1: + parser.error("--post-limit must be positive") + if args.pair_limit < 1: + parser.error("--pair-limit must be positive") + result = asyncio.run(_submit(args)) + else: + result = asyncio.run(_collect(args)) + print(json.dumps(result, ensure_ascii=False, sort_keys=True, default=str)) + + +if __name__ == "__main__": + main() diff --git a/tests/test_channel_weight_estimation.py b/tests/test_channel_weight_estimation.py index c6af06cdf..1a16189bd 100644 --- a/tests/test_channel_weight_estimation.py +++ b/tests/test_channel_weight_estimation.py @@ -1,34 +1,166 @@ -"""Tests for ADR 0145's fail-closed channel-weight boundary.""" +"""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 estimate_channel_weights +from lineageweave.channel_weight_estimation import ( + _MIN_SAMPLE_PAIRS, + dichotomize, + estimate_channel_weights, + estimate_fixture_channel_weights, + simulate_fixture_pair_scores, +) +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_unanchored_channel_scores_never_run_a_fit(monkeypatch) -> None: - import fast_mlsirm +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 - def unexpected_fit(**_kwargs): - raise AssertionError("an unanchored fit cannot establish genuine lineage") - monkeypatch.setattr(fast_mlsirm, "fit", unexpected_fit) +@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": 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, + "temporal": generator.random(), + "text": generator.random(), } - for index in range(200) + 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_expected_information" + + +def test_fixture_simulation_is_deterministic_and_carries_the_demo_design() -> None: + """Runs everywhere: the demo design must reproduce exactly so every + `make seed` and demo-server estimate lands on identical weights. + """ + first_scores, first_groups = simulate_fixture_pair_scores() + second_scores, second_groups = simulate_fixture_pair_scores() + assert first_scores == second_scores + assert first_groups == second_groups + assert len(first_scores) == 900 + assert set(first_scores[0]) == {"temporal", "secondary_key", "text"} + assert len(set(first_groups)) == 12 + + +@pytest.mark.skipif( + not _FAST_MLSIRM_AVAILABLE, reason="requires fast_mlsirm -- install from the org repo" +) +def test_fixture_estimate_recovers_the_demo_design_and_keeps_the_designed_tree() -> None: + """The demo fuses only with this estimate (ADR 0145, second + amendment: no hand-picked weight exists anywhere). It must recover + the declared follow-probability ordering AND still reconstruct the + designed A-100 fork the demo walkthroughs rely on. + """ + from lineageweave.fixtures import sample_records + from lineageweave.lineage_persistence import lineage_edge_specs + + estimate = estimate_fixture_channel_weights() + assert estimate is not None + weights = estimate.weights + assert weights["temporal"] > weights["secondary_key"] > weights["text"] + assert math.isclose(sum(weights.values()), 1.0, rel_tol=1e-9) - assert ( - estimate_channel_weights(pairs, [index % 4 for index in range(len(pairs))]) - is None - ) + edges = lineage_edge_specs(sample_records(), weights=weights) + pairs = {(edge.parent_id, edge.child_id) for edge in edges} + assert ("rec-002", "rec-003") in pairs + assert ("rec-002", "rec-004") in pairs + assert "rec-006" not in {edge.child_id for edge in edges} diff --git a/tests/test_estimate_channel_weights_script.py b/tests/test_estimate_channel_weights_script.py index 917f77798..3b086524d 100644 --- a/tests/test_estimate_channel_weights_script.py +++ b/tests/test_estimate_channel_weights_script.py @@ -1,50 +1,138 @@ -"""Tests for the fail-closed channel-weight operator boundary.""" +"""Tests for scripts/estimate_channel_weights.py (ADR 0200). + +`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. The +persistence contract must stamp full per-run provenance, and the +snapshot digest must be reproducible so the provenance row names the +exact corpus slice without storing content. +""" from __future__ import annotations -import argparse import asyncio -import runpy -import sys -from pathlib import Path +from contextlib import asynccontextmanager +from datetime import datetime, timedelta, timezone import pytest -import scripts.estimate_channel_weights as script - +from lineageweave.channel_weight_estimation import ChannelWeightEstimate +from lineageweave.models import Record -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))) - - -def test_main_reports_an_authorized_result_when_the_boundary_supplies_one( - monkeypatch, - capsys, -) -> None: - """The CLI serializes only the result returned by its policy boundary.""" +import scripts.estimate_channel_weights as script - async def authorized_result(_args: argparse.Namespace) -> dict[str, object]: - return {"available": False, "reason": "synthetic test boundary"} - monkeypatch.setattr(script, "_run", authorized_result) - monkeypatch.setattr(sys, "argv", [str(Path(script.__file__)), "--dry-run"]) +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, + ) - script.main() - assert capsys.readouterr().out == ( - '{"available": false, "reason": "synthetic test boundary"}\n' +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, pair_labels = 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"} + # Labels align with the scored pair so the queued llm judging pass can + # score the same candidate geometry without re-deriving it. + assert pair_labels == [("title a1", "title a2")] + + +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_llm_subsample_stride_is_deterministic_and_spread() -> None: + # Small totals pass through untouched; larger ones are evenly strided + # (first index 0, no index past the end, exactly the limit chosen) + # with no randomness, so re-runs stay comparable. + assert script.subsample_stride(3, 10) == [0, 1, 2] + chosen = script.subsample_stride(1000, 40) + assert len(chosen) == 40 + assert chosen[0] == 0 + assert chosen == sorted(chosen) + assert chosen[-1] <= 999 + assert script.subsample_stride(1000, 40) == chosen + + +def test_snapshot_digest_is_reproducible_and_order_sensitive() -> None: + rows = [ + {"post_id": "a", "created_at": datetime(2026, 1, 1, tzinfo=timezone.utc)}, + {"post_id": "b", "created_at": datetime(2026, 1, 2, tzinfo=timezone.utc)}, + ] + first = script.source_snapshot_digest(rows) + assert first == script.source_snapshot_digest(list(rows)) + assert first != script.source_snapshot_digest(list(reversed(rows))) + assert len(first) == 64 + + +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" + + +def test_persist_estimate_stamps_full_provenance_on_one_scoped_set() -> None: + conn = _Connection() + estimate = ChannelWeightEstimate( + weights={"temporal": 0.25, "text": 0.75}, + sample_pair_count=600, + estimation_method_code="mls2plm_expected_information", + ) + cutoff = datetime(2026, 1, 2, tzinfo=timezone.utc) + run_id = asyncio.run( + script.persist_estimate( + conn, + estimate, + channel_set_code=script.DETERMINISTIC_SET_CODE, + snapshot_sha256="a" * 64, + knowledge_cutoff=cutoff, + ) ) + delete_query, delete_args = conn.executed[0] + # Scoped delete: persisting the deterministic set must never wipe + # another set -- each active-channel combination owns its own rows. + assert "delete from lineage_channel_weight where channel_set_code = $1" in delete_query + assert delete_args == (script.DETERMINISTIC_SET_CODE,) + inserted = {call[1][1]: call[1] for call in conn.executed[1:]} + assert set(inserted) == {"temporal", "text"} + for row in inserted.values(): + assert row[0] == script.DETERMINISTIC_SET_CODE + assert row[3] == run_id + assert row[4] == "mls2plm_expected_information" + assert isinstance(row[5], str) and row[5].strip() + assert row[6] == script.UNANCHORED_METHOD_CODE + assert row[7] == "a" * 64 + assert row[8] == 600 + assert row[9] == cutoff + assert inserted["text"][2] == 0.75 def test_main_rejects_nonpositive_post_limit(monkeypatch) -> None: - monkeypatch.setattr(sys, "argv", [str(Path(script.__file__)), "--post-limit", "0"]) + monkeypatch.setattr( + "sys.argv", ["estimate_channel_weights.py", "--post-limit", "0"] + ) with pytest.raises(SystemExit): script.main() - - -def test_module_entrypoint_fails_closed_before_reading_corpus(monkeypatch) -> None: - script_path = Path(script.__file__) - monkeypatch.setattr(sys, "argv", [str(script_path)]) - with pytest.raises(RuntimeError, match="nothing was written"): - runpy.run_path(str(script_path), run_name="__main__") diff --git a/tests/test_estimate_llm_channel_weights_script.py b/tests/test_estimate_llm_channel_weights_script.py new file mode 100644 index 000000000..c9ab53a05 --- /dev/null +++ b/tests/test_estimate_llm_channel_weights_script.py @@ -0,0 +1,61 @@ +"""Tests for scripts/estimate_llm_channel_weights.py (ADR 0200 point 5). + +The queued judging flow must never make bulk synchronous provider calls +(one batch submission is the only provider interaction in ``submit``), +must map results to pairs by caller-supplied ``custom_id`` only (never +result order), and must fit exclusively over a complete run. +""" + +from __future__ import annotations + +from lineageweave.adjudication_client import judge_prompt, parse_confidence + +import scripts.estimate_llm_channel_weights as script + + +def test_batch_requests_carry_caller_custom_ids_for_every_pair() -> None: + labels = [("a", "b"), ("c", "d"), ("e", "f")] + requests = script.batch_requests_for_pairs([0, 2], labels) + assert [request["custom_id"] for request in requests] == ["pair-0", "pair-2"] + # Never mix caller ids with generated ids in one batch (upstream + # guidance on contextual-orchestrator #832): every request has one. + assert all("custom_id" in request for request in requests) + assert requests[0]["messages"][0]["content"] == judge_prompt("a", "b") + assert requests[1]["messages"][0]["content"] == judge_prompt("e", "f") + assert all(request["mode"] == "auto" for request in requests) + + +def test_shared_judge_prompt_and_confidence_parse_round_trip() -> None: + prompt = judge_prompt("Record about pricing", "Follow-up record") + assert "Record A: Record about pricing" in prompt + assert "Record B: Follow-up record" in prompt + assert parse_confidence("0.85") == 0.85 + assert parse_confidence("confidence: 0.4 maybe") == 0.4 + assert parse_confidence("no number here") == 0.0 + assert parse_confidence("1.7") == 1.0 + + +def test_errored_judgments_stay_unjudged_instead_of_becoming_zero() -> None: + """An empty or non-numeric answer must never persist as a confident + 0.0 -- the pair stays unjudged and the incomplete-run path reports it. + Mapping is by custom_id only; foreign or malformed ids are ignored. + """ + updates = script.judgment_updates_from_results( + [ + {"custom_id": "pair-3", "answer": "0.7"}, + {"custom_id": "pair-4", "answer": ""}, + {"custom_id": "pair-5", "answer": "provider error: upstream unavailable"}, + {"custom_id": "pair-6", "answer": "0.0"}, + {"custom_id": "req_generated9", "answer": "0.9"}, + {"custom_id": "pair-not-a-number", "answer": "0.9"}, + ] + ) + assert updates == [(3, 0.7), (6, 0.0)] + + +def test_batch_completion_is_detected_from_flag_or_status() -> None: + assert script._is_complete({"is_complete": True}) + assert script._is_complete({"status": "completed"}) + assert script._is_complete({"status": "Succeeded"}) + assert not script._is_complete({"status": "in_progress"}) + assert not script._is_complete({}) diff --git a/tests/test_lineage_ingestion.py b/tests/test_lineage_ingestion.py index a644a4e5e..eb3217580 100644 --- a/tests/test_lineage_ingestion.py +++ b/tests/test_lineage_ingestion.py @@ -37,6 +37,7 @@ async def fetchval(self, _query: str): async def fetch(self, _query: str): provenance = { + "channel_set_code": "channel_set_deterministic", "estimation_run_id": "00000000-0000-0000-0000-000000000001", "estimation_method_code": "mls2plm_discrimination", "estimator_version": "5006c382", @@ -66,7 +67,13 @@ async def fetchval(self, _query: str): return True async def fetch(self, _query: str): - return [{"channel_code": "temporal", "weight_value": 1.0}] + return [ + { + "channel_set_code": "channel_set_deterministic", + "channel_code": "temporal", + "weight_value": 1.0, + } + ] assert asyncio.run( ingestion.load_estimated_channel_weights( @@ -85,6 +92,7 @@ async def fetchval(self, _query: str): async def fetch(self, _query: str): base = { + "channel_set_code": "channel_set_deterministic", "estimation_method_code": "test_method", "estimator_version": "1.0.0", "anchor_method_code": "test_anchor", @@ -126,6 +134,7 @@ async def fetchval(self, _query: str): async def fetch(self, _query: str): provenance = { + "channel_set_code": "channel_set_deterministic", "estimation_run_id": "00000000-0000-0000-0000-000000000001", "estimation_method_code": "test_method", "estimator_version": "1.0.0", @@ -158,6 +167,125 @@ async def fetch(self, _query: str): ) +def test_weight_loader_picks_the_set_matching_the_active_channels( + monkeypatch, +) -> None: + """Migration 0200: one persisted set per active-channel combination. + + A 3-channel rebuild and a 4-channel (llm-inclusive) run each get + exactly their own set; an active combination with no matching set + returns ``None`` -- never a partial mix of estimation runs. + """ + monkeypatch.setattr(ingestion, "_SUPPORTED_ANCHOR_METHOD_CODES", {"test_anchor"}) + + class TwoSetConnection: + async def fetchval(self, _query: str): + return True + + async def fetch(self, _query: str): + def rows(set_code, run_id, channel_weights): + return [ + { + "channel_set_code": set_code, + "channel_code": channel, + "weight_value": weight, + "estimation_run_id": run_id, + "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), + } + for channel, weight in channel_weights + ] + + return rows( + "channel_set_deterministic", + "00000000-0000-0000-0000-000000000001", + (("temporal", 0.6), ("secondary_key", 0.3), ("text", 0.1)), + ) + rows( + "channel_set_with_llm", + "00000000-0000-0000-0000-000000000002", + ( + ("temporal", 0.4), + ("secondary_key", 0.2), + ("text", 0.1), + ("llm", 0.3), + ), + ) + + deterministic = asyncio.run( + ingestion.load_estimated_channel_weights( + TwoSetConnection(), {"temporal", "secondary_key", "text"} + ) + ) + assert deterministic == {"temporal": 0.6, "secondary_key": 0.3, "text": 0.1} + with_llm = asyncio.run( + ingestion.load_estimated_channel_weights( + TwoSetConnection(), {"temporal", "secondary_key", "text", "llm"} + ) + ) + assert with_llm is not None and with_llm["llm"] == 0.3 + assert asyncio.run( + ingestion.load_estimated_channel_weights( + TwoSetConnection(), {"temporal", "text"} + ) + ) is None + + +def test_weight_loader_reads_a_pre_0200_schema_as_one_deterministic_set( + monkeypatch, +) -> None: + """Before migration 0200 no channel_set_code column exists; the loader + must probe the catalog (never a failing statement) and treat the rows + as the single implicit deterministic set. + """ + monkeypatch.setattr(ingestion, "_SUPPORTED_ANCHOR_METHOD_CODES", {"test_anchor"}) + + class Pre0200Connection: + def __init__(self) -> None: + self.fetch_queries: list[str] = [] + + async def fetchval(self, query: str): + if "information_schema.columns" in query: + return False + return True + + async def fetch(self, query: str): + self.fetch_queries.append(query) + provenance = { + "channel_set_code": "channel_set_deterministic", + "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 ( + ("temporal", 0.6), + ("secondary_key", 0.3), + ("text", 0.1), + ) + ] + + connection = Pre0200Connection() + loaded = asyncio.run( + ingestion.load_estimated_channel_weights( + connection, {"temporal", "secondary_key", "text"} + ) + ) + assert loaded == {"temporal": 0.6, "secondary_key": 0.3, "text": 0.1} + assert all( + "select channel_set_code" not in query + for query in connection.fetch_queries + ), "a pre-0200 schema must never be queried for the missing column" + + def test_records_use_persisted_thread_keys_not_process_unit_or_voc_type() -> None: rows = [ {