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/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/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 483108952..b8ab9534f 100644 --- a/scripts/estimate_channel_weights.py +++ b/scripts/estimate_channel_weights.py @@ -83,13 +83,15 @@ def source_snapshot_digest(rows: list) -> str: def sample_pair_scores( records: list, *, window: int = DEFAULT_CANDIDATE_WINDOW -) -> tuple[list[dict[str, float]], list[int]]: +) -> 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. + ``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: @@ -97,6 +99,7 @@ def sample_pair_scores( 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): @@ -109,7 +112,21 @@ def sample_pair_scores( } ) group_ids.append(group_index) - return pair_scores, group_ids + 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( @@ -178,7 +195,7 @@ async def _run(args: argparse.Namespace) -> dict[str, object]: 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 = sample_pair_scores(records) + pair_scores, group_ids, _pair_labels = sample_pair_scores(records) estimate = estimate_channel_weights(pair_scores, group_ids) if estimate is None: 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_estimate_channel_weights_script.py b/tests/test_estimate_channel_weights_script.py index 5c756971c..3b086524d 100644 --- a/tests/test_estimate_channel_weights_script.py +++ b/tests/test_estimate_channel_weights_script.py @@ -38,22 +38,38 @@ def test_sampling_stays_within_groups_and_window() -> None: _record("a2", "g-a", 1), _record("b1", "g-b", 2), ] - pair_scores, group_ids = script.sample_pair_scores(records, window=50) + 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) + _, 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) + 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)}, 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({})