chore(stack): propagate merged Phase C (queued judging, #586) up to the ADR 0200 branch

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))

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))
+);

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;

scripts/estimate_channel_weights.py

@@ -83,20 +83,23 @@ 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:
         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):
@@ -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: