combine.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# bench/argv_policy/combine.py
 
"""
Compose two committed result files into one: the first decides, the second is
consulted only where the first is unsure.
 
Purpose
-------
Every leg of this bench so far has measured one model alone. The escalation
column says what a model would hand to a human below a confidence threshold,
and plan 293 showed that Jev, asked to flag below 0.9, catches every one of its
own permission-granting errors. The fine-tuned encoders beat Jev on agreement
at half the latency and no per-call cost, and at 0.9 they hand 42 to 51 of 300
hard-set decisions upwards. Nobody has measured what happens when those
escalated decisions go to Jev instead of to a human.
 
Every piece of that measurement is already on disk, so this is a join rather
than a run: the encoder result files and the Jev result files carry the same
commands in the same order, and a combination is a pure function of two rows
and a threshold. Nothing here opens a socket, loads a checkpoint or touches a
GPU; the two inputs are read-only and the output is a result file in the bench
schema that :mod:`score` reads unchanged.
 
A row is UNSURE when the first model's top probability is strictly below the
threshold or the row has no usable ``probs``. At or above the threshold every
rule keeps the first model's verdict and ``probs`` untouched, so the four rules
differ only in what they do with the unsure rows:
 
``escalate``
    The unsure row becomes ``confirm`` with no ``probs`` and nothing is
    consulted. This is the CONTROL: the encoder alone with a human behind it,
    which is the operating point plans 293 and 300 already recommend.
``replace``
    The unsure row takes the second model's verdict and ``probs``.
``stricter``
    The unsure row takes whichever of the two verdicts grants LESS permission,
    with the ``probs`` of the model whose verdict was taken.
``disagree``
    The unsure row keeps the first's verdict and ``probs`` where the two agree,
    and becomes ``confirm`` with no ``probs`` where they differ.
 
``--sweep`` writes ``results/combined_sweep.md``: the same compositions at six
thresholds, and the movement across the repeated Jev runs, scored in memory so
that a threshold curve costs no committed files. It carries no generation date
precisely so that a second run reproduces it byte for byte.
 
Run it under the project venv from the subproject root::
 
    python bench/argv_policy/combine.py --first results/encoder_small-7-c299_hard.json \\
        --second results/jev_direct_hard.json --rule replace --threshold 0.9 \\
        --out results/combined_encoder_small-7-c299_replace_t90_hard.json
 
    python bench/argv_policy/combine.py --sweep
 
Responsibilities
----------------
- :func:`rule_escalate` / :func:`rule_replace` / :func:`rule_stricter` /
  :func:`rule_disagree` -- the four rules, each a pure function of one first
  row, one second row and the threshold.
- :func:`join_rows` -- the command-keyed join, and the refusal when the two
  files do not carry the same commands.
- :func:`build_payload` -- two payloads to one result payload with provenance.
- :func:`render_sweep` -- the generated markdown of the threshold sweep and the
  repeated-run movement.
- :func:`main` -- the CLI.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - combination lifecycle (files read, file written, sweep rendered)
    L3 - per-combination counts (consulted, disagreeing)
    L4 - module load
 
Contracts
---------
- Read-only in both inputs: a combination never rewrites a result file, and the
  only paths written are the ``--out`` target and the sweep document.
- No network, no torch, no key: the Jev files ARE the Jev runs.
- The permission ordering is :data:`score.PERMISSIVENESS` itself, imported
  rather than restated, so ``stricter`` can never disagree with the direction
  columns the scorer reports.
- A threshold comparison is STRICT: a top probability exactly equal to the
  threshold is sure, which is the same reading ``score.py`` uses for its
  escalation count.
- Provenance stores each input path RELATIVE to the bench directory when it
  lies inside it, so a committed combination is reproducible from any checkout
  rather than naming the worktree it was made in.
- Console output is ASCII only; commands are printed through :func:`ascii`.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- score (the bench scorer, for PERMISSIVENESS and for the sweep's arithmetic)
"""
from __future__ import annotations
 
import argparse
import datetime
import hashlib
import json
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable
 
from oc_diagnostics import _dbg
 
from score import PERMISSIVENESS, Score, score_payload
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
#: One row of a result file, and one whole result file. Both are JSON read off
#: disk, so their value types are genuinely unknown to this module: the schema
#: lives in the runners that wrote them and in ``score.py`` that reads them.
Row = dict[str, Any]
Payload = dict[str, Any]
 
#: The four combination rules, in the order the tables report them.
RULES: tuple[str, ...] = ("escalate", "replace", "stricter", "disagree")
 
#: The rules that read the second model at all. ``escalate`` is the control and
#: consults nothing, so a second file is optional for it.
CONSULTING_RULES: tuple[str, ...] = ("replace", "stricter", "disagree")
 
#: The verdict an unsure row is turned into when a rule sends it to a human.
ESCALATION_VERDICT = "confirm"
 
#: The rank given to a verdict outside the three labels. It is MORE permissive
#: than ``allow`` so that ``stricter`` never prefers an unrecognised verdict,
#: which is the same direction ``score.error_direction`` rounds in.
UNKNOWN_RANK: int = max(PERMISSIVENESS.values()) + 1
 
#: Default confidence below which the first model is treated as unsure. The
#: same figure as ``score.DEFAULT_THRESHOLD``, and the plan's main tables.
DEFAULT_THRESHOLD = 0.9
 
#: The three firsts the sweep reads, by result-file stem, hard set then hand.
SWEEP_FIRSTS: tuple[str, ...] = (
    "encoder_small-7-c297",
    "encoder_small-7-c299",
    "encoder_xsmall-7-c297-sc4-dw1",
)
 
#: The repeated Jev runs on the hard set, first run first.
SWEEP_SECONDS_HARD: tuple[str, ...] = (
    "jev_direct_hard",
    "jev_direct_hard_rep1",
    "jev_direct_hard_rep2",
)
 
#: The repeated Jev runs on the hand set, first run first.
SWEEP_SECONDS_HAND: tuple[str, ...] = ("jev_direct_hand", "jev_direct_hand_ref")
 
#: The thresholds the sweep walks.
SWEEP_THRESHOLDS: tuple[float, ...] = (0.5, 0.7, 0.8, 0.9, 0.95, 0.99)
 
 
class CombineError(Exception):
    """A combination that cannot be made: a bad join, or a missing input."""
 
 
@dataclass(frozen=True)
class Decision:
    """What one rule did with one pair of rows."""
 
    pred: str
    probs: dict[str, float] | None
    consulted: bool
 
 
def top_probability(row: Row) -> float | None:
    """The row's top forecast probability, or ``None`` when it has none.
 
    The same reading ``score.py`` takes of the same field, and the tests assert
    that equality on a battery of rows rather than trusting this comment: a row
    whose ``probs`` is missing, empty or not a mapping has no forecast at all,
    and is not a zero-confidence one.
    """
 
    probs = row.get("probs")
    if not isinstance(probs, dict) or not probs:
        return None
    values = [float(value) for value in probs.values()]
    return max(values) if values else None
 
 
def is_unsure(row: Row, threshold: float) -> bool:
    """Whether the first model is unsure enough on this row to be combined.
 
    Strictly below the threshold, so a top probability exactly equal to it is
    SURE and the row passes through untouched. A row with no usable forecast is
    unsure: a model that produced no distribution has expressed no confidence,
    and reading that as certainty is the one direction this must not round in.
    """
 
    top = top_probability(row)
    return top is None or top < threshold
 
 
def _usable_probs(row: Row) -> dict[str, float] | None:
    """A copy of the row's ``probs``, or ``None`` when it has none worth carrying."""
 
    probs = row.get("probs")
    if not isinstance(probs, dict) or not probs:
        return None
    return dict(probs)
 
 
def _rank(prediction: str) -> int:
    """How much permission one verdict grants, under the scorer's own ordering."""
 
    return PERMISSIVENESS.get(prediction, UNKNOWN_RANK)
 
 
def _keep_first(first: Row) -> Decision:
    """The sure case, which every rule shares: the first model, untouched."""
 
    return Decision(str(first["pred"]), _usable_probs(first), False)
 
 
def rule_escalate(first: Row, second: Row | None,
                  threshold: float) -> Decision:
    """The control: an unsure row goes to a human, and nothing is consulted.
 
    ``second`` is accepted and deliberately unread. Every rule shares one
    signature so that the sweep can run all four over the same joined rows, and
    the control is the rule that has a second opinion available and declines to
    ask for it -- which is exactly the operating point it controls for.
    """
 
    if not is_unsure(first, threshold):
        return _keep_first(first)
    return Decision(ESCALATION_VERDICT, None, False)
 
 
def rule_replace(first: Row, second: Row | None,
                 threshold: float) -> Decision:
    """An unsure row takes the second model's verdict and its ``probs``."""
 
    if not is_unsure(first, threshold):
        return _keep_first(first)
    if second is None:
        raise CombineError("rule 'replace' needs a second result file")
    return Decision(str(second["pred"]), _usable_probs(second), True)
 
 
def rule_stricter(first: Row, second: Row | None,
                  threshold: float) -> Decision:
    """An unsure row takes whichever verdict grants less permission.
 
    The ``probs`` travel with the verdict that was taken, so the forecast a row
    carries is always the forecast of the model that decided it. Where the two
    agree -- including where they agree on a verdict neither label set knows --
    the first model's forecast is kept, because it is the model that would have
    answered anyway.
    """
 
    if not is_unsure(first, threshold):
        return _keep_first(first)
    if second is None:
        raise CombineError("rule 'stricter' needs a second result file")
    first_pred, second_pred = str(first["pred"]), str(second["pred"])
    if _rank(second_pred) < _rank(first_pred):
        return Decision(second_pred, _usable_probs(second), True)
    return Decision(first_pred, _usable_probs(first), True)
 
 
def rule_disagree(first: Row, second: Row | None,
                  threshold: float) -> Decision:
    """An unsure row survives agreement and goes to a human on disagreement."""
 
    if not is_unsure(first, threshold):
        return _keep_first(first)
    if second is None:
        raise CombineError("rule 'disagree' needs a second result file")
    if str(second["pred"]) == str(first["pred"]):
        return Decision(str(first["pred"]), _usable_probs(first), True)
    return Decision(ESCALATION_VERDICT, None, True)
 
 
#: The rules by name. Each is a pure function of (first row, second row,
#: threshold) and holds no state between rows, which is what lets the sweep run
#: the same four rules over six thresholds without re-reading a file.
RULE_FUNCTIONS: dict[str, Callable[..., Decision]] = {
    "escalate": rule_escalate,
    "replace": rule_replace,
    "stricter": rule_stricter,
    "disagree": rule_disagree,
}
 
 
def _index_by_command(rows: list[Row], label: str) -> dict[str, Row]:
    """Index one file's rows by command, refusing a duplicate.
 
    A repeated command would make the join ambiguous -- two truths, two
    forecasts, and no rule for which one wins -- so it is refused here rather
    than resolved silently. No case file on this bench carries one.
    """
 
    index: dict[str, Row] = {}
    for row in rows:
        command = str(row["command"])
        if command in index:
            raise CombineError(f"{label}: command appears twice: {ascii(command)}")
        index[command] = row
    return index
 
 
def join_rows(first_rows: list[Row],
              second_rows: list[Row] | None,
              first_label: str, second_label: str) -> list[Row | None]:
    """Line the second file's rows up with the first file's, in the first's order.
 
    Returns one entry per first row: the matching second row, or ``None`` for
    every row when there is no second file. Refuses when the two files do not
    carry exactly the same commands, or when they disagree about the truth of
    one -- both of those mean the two runs are not the same measurement, and
    every figure downstream would be a join artefact rather than a reading.
    """
 
    if second_rows is None:
        return [None] * len(first_rows)
    first_index = _index_by_command(first_rows, first_label)
    second_index = _index_by_command(second_rows, second_label)
    missing = sorted(set(first_index) - set(second_index))
    extra = sorted(set(second_index) - set(first_index))
    if missing or extra:
        raise CombineError(
            f"{first_label} and {second_label} do not carry the same commands: "
            f"{len(missing)} only in the first, {len(extra)} only in the second"
            + (f"; first missing: {ascii(missing[0])}" if missing else "")
            + (f"; first extra: {ascii(extra[0])}" if extra else "")
        )
    for command, row in first_index.items():
        other = second_index[command]
        if str(row.get("truth")) != str(other.get("truth")):
            raise CombineError(
                f"{first_label} and {second_label} disagree on the truth of "
                f"{ascii(command)}: {row.get('truth')!r} against {other.get('truth')!r}"
            )
    return [second_index[str(row["command"])] for row in first_rows]
 
 
def _effective_ms(first: Row, second: Row | None,
                  consulted: bool) -> float | None:
    """The latency of the combined decision: the first's, plus the second's when asked."""
 
    values = [float(first["ms"])] if first.get("ms") is not None else []
    if consulted and second is not None and second.get("ms") is not None:
        values.append(float(second["ms"]))
    return sum(values) if values else None
 
 
def _threshold_tag(threshold: float) -> str:
    """The threshold as the two digits the file names and model names carry."""
 
    return f"t{round(threshold * 100):02d}"
 
 
def _digest(path: Path) -> str:
    """The SHA-256 of one input file, which is what pins a combination to its inputs."""
 
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()
 
 
def _relative_path(path: Path, bench_dir: Path) -> str:
    """One input path, relative to the bench directory when it lies inside it.
 
    A committed combination names ``results/jev_direct_hard.json`` rather than
    the worktree it happened to be made in, so anyone can rerun it from their
    own checkout and compare digests instead of paths.
    """
 
    resolved = Path(path).resolve()
    try:
        return resolved.relative_to(Path(bench_dir).resolve()).as_posix()
    except ValueError:
        return resolved.as_posix()
 
 
def combine_rows(first_rows: list[Row],
                 seconds: list[Row | None],
                 rule: str, threshold: float) -> list[Row]:
    """Apply one rule down the two joined row lists, in the first file's order."""
 
    apply_rule = RULE_FUNCTIONS[rule]
    rows: list[Row] = []
    for first, second in zip(first_rows, seconds):
        decision = apply_rule(first, second, threshold)
        row: Row = {
            "command": str(first["command"]),
            "truth": str(first["truth"]),
            "slice": str(first.get("slice", "")),
            "pred": decision.pred,
        }
        if decision.probs is not None:
            row["probs"] = decision.probs
        row["ms"] = _effective_ms(first, second, decision.consulted)
        row["consulted"] = decision.consulted
        row["first_pred"] = str(first["pred"])
        row["second_pred"] = None if second is None else str(second["pred"])
        rows.append(row)
    return rows
 
 
def build_payload(first_payload: Payload,
                  second_payload: Payload | None,
                  *, rule: str, threshold: float,
                  first_path: Path, second_path: Path | None,
                  bench_dir: Path, date: str) -> Payload:
    """Two loaded result payloads to one combined result payload.
 
    The output is a bench result file and nothing more exotic: ``score.py``
    reads it with no change, and the three fields it does not read --
    ``consulted``, ``first_pred``, ``second_pred`` -- are there so a reader can
    take any row apart without rerunning the combination.
    """
 
    func = "build_payload"
    first_rows = list(first_payload.get("results", []))
    second_rows = list(second_payload.get("results", [])) if second_payload else None
    first_model = str(first_payload.get("model", "?"))
    second_model = str(second_payload.get("model", "?")) if second_payload else None
    seconds = join_rows(first_rows, second_rows, first_model,
                        second_model or "(no second file)")
    rows = combine_rows(first_rows, seconds, rule, threshold)
 
    consulted = sum(1 for row in rows if row["consulted"])
    disagreeing = sum(1 for row in rows
                      if row["consulted"] and row["first_pred"] != row["second_pred"])
    cost_usd: float | None = None
    if second_payload is not None and second_payload.get("cost_usd") is not None and second_rows:
        cost_usd = consulted * float(second_payload["cost_usd"]) / len(second_rows)
    payload: Payload = {
        "model": f"combined_{first_model}_{rule}_{_threshold_tag(threshold)}",
        "condition": second_model or "none",
        "results": rows,
        "cost_usd": cost_usd,
        "provenance": {
            "plan": "302",
            "date": date,
            "rule": rule,
            "threshold": threshold,
            "first_file": _relative_path(first_path, bench_dir),
            "first_model": first_model,
            "first_condition": str(first_payload.get("condition", "-")),
            "first_sha256": _digest(first_path),
            "second_file": None if second_path is None else _relative_path(second_path, bench_dir),
            "second_model": second_model,
            "second_condition": (str(second_payload.get("condition", "-"))
                                 if second_payload else None),
            "second_sha256": None if second_path is None else _digest(second_path),
            "consulted": consulted,
            "disagreeing": disagreeing,
            "confirm_issued": sum(1 for row in rows if row["pred"] == ESCALATION_VERDICT),
            "confirm_required": sum(1 for row in rows if row["truth"] == ESCALATION_VERDICT),
        },
    }
    _dbg("OC_CHAT.MAIN", func,
         f"{payload['model']}: consulted {consulted}, disagreeing {disagreeing}", 3)
    return payload
 
 
def load_payload(path: Path) -> Payload:
    """Read one result file, refusing a path that is not there."""
 
    target = Path(path)
    if not target.is_file():
        raise CombineError(f"result file not found: {target}")
    return json.loads(target.read_text(encoding="utf-8"))
 
 
# --------------------------------------------------------------------------
# The sweep document
# --------------------------------------------------------------------------
 
 
@dataclass(frozen=True)
class SweepRow:
    """One scored combination, as the sweep tables print it."""
 
    first: str
    rule: str
    threshold: float
    second: str
    score: Score
    consulted: int
 
    @property
    def cost_usd(self) -> float | None:
        """What the consulted decisions would have cost at the second's own rate."""
 
        return self.score.cost_usd
 
 
class _PayloadCache:
    """Result files read once and reused across the sweep's 130-odd combinations."""
 
    def __init__(self, results_dir: Path) -> None:
        self._dir = Path(results_dir)
        self._payloads: dict[str, Payload] = {}
        self._digests: dict[str, str] = {}
 
    def path(self, stem: str) -> Path:
        """The result file for one stem."""
 
        return self._dir / f"{stem}.json"
 
    def payload(self, stem: str) -> Payload:
        """One result payload, loaded at most once."""
 
        if stem not in self._payloads:
            self._payloads[stem] = load_payload(self.path(stem))
            self._digests[stem] = _digest(self.path(stem))
        return self._payloads[stem]
 
    def digest(self, stem: str) -> str:
        """The SHA-256 of one result file, computed at most once."""
 
        self.payload(stem)
        return self._digests[stem]
 
 
def _sweep_row(cache: _PayloadCache, first: str, second: str, rule: str,
               threshold: float, bench_dir: Path) -> SweepRow:
    """Compose and score one combination in memory, writing nothing."""
 
    payload = build_payload(
        cache.payload(first), cache.payload(second), rule=rule, threshold=threshold,
        first_path=cache.path(first), second_path=cache.path(second),
        bench_dir=bench_dir, date="",
    )
    provenance = payload["provenance"]
    assert isinstance(provenance, dict)
    return SweepRow(first=first, rule=rule, threshold=threshold, second=second,
                    score=score_payload(payload, f"{first}|{rule}", threshold),
                    consulted=int(provenance["consulted"]))
 
 
def _ms_cell(value: float | None) -> str:
    """A latency in milliseconds, to one decimal, or a dash."""
 
    return "-" if value is None else f"{value:.1f}"
 
 
def _cost_cell(value: float | None) -> str:
    """A cost in US dollars, to four decimals, or a dash."""
 
    return "-" if value is None else f"{value:.4f}"
 
 
def _threshold_table(rows: list[SweepRow]) -> str:
    """The threshold sweep: one line per first, rule and threshold."""
 
    lines = ["| first | rule | threshold | agreement | granted more | granted less "
             "| consulted | confirm issued | median ms | cost USD |",
             "|---|---|---:|---:|---:|---:|---:|---:|---:|---:|"]
    for row in rows:
        score = row.score
        lines.append(
            f"| `{row.first}` | `{row.rule}` | {row.threshold:.2f} | {score.agreement} "
            f"| {score.granted_more} | {score.granted_less} | {row.consulted} "
            f"| {_confirm_issued(score)} | {_ms_cell(score.median_ms)} "
            f"| {_cost_cell(row.cost_usd)} |"
        )
    return "\n".join(lines) + "\n"
 
 
def _confirm_issued(score: Score) -> int:
    """How many ``confirm`` verdicts one scored combination issued."""
 
    return int(score.provenance["confirm_issued"]) if score.provenance else 0
 
 
def _grouped(rows: list[SweepRow],
             key: Callable[[SweepRow], tuple[object, ...]]
             ) -> list[tuple[tuple[object, ...], list[SweepRow]]]:
    """Group scored rows by a key, in the order the keys first appear.
 
    Grouping on the row's OWN fields rather than on the module constants is
    what keeps a range table honest: a key spelled by hand can miss every row
    and print an empty table, which is exactly what a reader would read as "no
    movement between runs".
    """
 
    groups: dict[tuple[object, ...], list[SweepRow]] = {}
    for row in rows:
        groups.setdefault(key(row), []).append(row)
    return list(groups.items())
 
 
def _repeat_rows_table(rows: list[SweepRow]) -> str:
    """Every (first, rule, run) reading at one threshold, one line each."""
 
    lines = ["| first | rule | second | agreement | granted more | granted less "
             "| consulted | confirm issued |", "|---|---|---|---:|---:|---:|---:|---:|"]
    for row in rows:
        score = row.score
        lines.append(
            f"| `{row.first}` | `{row.rule}` | `{row.second}` | {score.agreement} "
            f"| {score.granted_more} | {score.granted_less} | {row.consulted} "
            f"| {_confirm_issued(score)} |"
        )
    return "\n".join(lines)
 
 
def _repeat_range_table(rows: list[SweepRow]) -> str:
    """The range each (first, rule) covers across the repeated runs at one threshold."""
 
    lines = ["| first | rule | agreement | granted more | granted less | consulted |",
             "|---|---|---|---|---|---:|"]
    for (first, rule), group in _grouped(rows, lambda row: (row.first, row.rule)):
        lines.append(
            f"| `{first}` | `{rule}` "
            f"| {_range_cell([item.score.correct for item in group])}"
            f" of {group[0].score.total} "
            f"| {_range_cell([item.score.granted_more for item in group])} "
            f"| {_range_cell([item.score.granted_less for item in group])} "
            f"| {group[0].consulted} |"
        )
    return "\n".join(lines)
 
 
def _threshold_range_table(rows: list[SweepRow]) -> str:
    """The same range, at every swept threshold, for reading an operating point.
 
    A recommendation made at one threshold needs its noise band at THAT
    threshold. The pinned table above is at 0.90 because that is where the
    plan's main tables sit; this one is what says whether a point chosen
    elsewhere on the curve survives a second run of the second model.
    """
 
    lines = ["| first | rule | threshold | agreement | granted more | granted less "
             "| consulted |", "|---|---|---:|---|---|---|---:|"]
    for (first, rule, threshold), group in _grouped(
            rows, lambda row: (row.first, row.rule, row.threshold)):
        assert isinstance(threshold, float)
        lines.append(
            f"| `{first}` | `{rule}` | {threshold:.2f} "
            f"| {_range_cell([item.score.correct for item in group])}"
            f" of {group[0].score.total} "
            f"| {_range_cell([item.score.granted_more for item in group])} "
            f"| {_range_cell([item.score.granted_less for item in group])} "
            f"| {group[0].consulted} |"
        )
    return "\n".join(lines)
 
 
def _range_cell(values: list[int]) -> str:
    """``n`` when every run agrees, ``lo-hi`` when they do not."""
 
    low, high = min(values), max(values)
    return str(low) if low == high else f"{low}-{high}"
 
 
def render_sweep(results_dir: Path, bench_dir: Path) -> str:
    """The whole generated sweep document.
 
    Deliberately DATELESS. The document is a pure function of the committed
    result files it names, and it carries their digests instead of a timestamp,
    so a second run reproduces it byte for byte and a diff means an input
    moved.
    """
 
    func = "render_sweep"
    cache = _PayloadCache(results_dir)
    hard_first = SWEEP_SECONDS_HARD[0]
 
    hand_first = SWEEP_SECONDS_HAND[0]
    sweep_rows = [_sweep_row(cache, f"{first}_hard", hard_first, rule, threshold, bench_dir)
                  for first in SWEEP_FIRSTS for rule in RULES
                  for threshold in SWEEP_THRESHOLDS]
    hand_sweep_rows = [_sweep_row(cache, f"{first}_hand", hand_first, rule, threshold,
                                  bench_dir)
                       for first in SWEEP_FIRSTS for rule in RULES
                       for threshold in SWEEP_THRESHOLDS]
    hard_rows = [_sweep_row(cache, f"{first}_hard", second, rule, DEFAULT_THRESHOLD, bench_dir)
                 for first in SWEEP_FIRSTS for rule in RULES for second in SWEEP_SECONDS_HARD]
    hand_rows = [_sweep_row(cache, f"{first}_hand", second, rule, DEFAULT_THRESHOLD, bench_dir)
                 for first in SWEEP_FIRSTS for rule in RULES for second in SWEEP_SECONDS_HAND]
    hard_curve = [_sweep_row(cache, f"{first}_hard", second, rule, threshold, bench_dir)
                  for first in SWEEP_FIRSTS for rule in RULES
                  for threshold in SWEEP_THRESHOLDS for second in SWEEP_SECONDS_HARD]
 
    inputs = sorted({row.first for row in sweep_rows + hard_rows + hand_rows}
                    | set(SWEEP_SECONDS_HARD) | set(SWEEP_SECONDS_HAND))
    parts = [
        "# Combining two opinions: the threshold sweep and the repeated runs",
        "",
        "Generated by `python bench/argv_policy/combine.py --sweep` (plan 302). Do not edit"
        " by hand: every figure below is composed in memory from the committed result files"
        " named at the foot of this document and scored by `score.py`, and the file carries"
        " no generation date so that a second run reproduces it byte for byte.",
        "",
        "The first model decides. A row is unsure when its top probability is strictly below"
        " the threshold, and only then is the second model consulted. `escalate` consults"
        " nothing and is the control: it is the encoder alone with a human behind it.",
        "",
        "## The threshold sweep, hard set, first Jev run",
        "",
        f"Second: `{hard_first}`. Thresholds: "
        + ", ".join(f"{value:.2f}" for value in SWEEP_THRESHOLDS) + ".",
        "",
        _threshold_table(sweep_rows).rstrip("\n"),
        "",
        "## The threshold sweep, hand set, first Jev run",
        "",
        "The 104 hand-written commands are too few for a threshold curve to be read on its"
        " own -- one command is almost a percentage point -- so this table is here to say"
        " whether an operating point chosen on the hard set above survives on the other case"
        f" set, not to choose one. Second: `{hand_first}`.",
        "",
        _threshold_table(hand_sweep_rows).rstrip("\n"),
        "",
        "## The movement between repeated runs of the second model",
        "",
        "Plan 293 measured two identical Jev requests moving probabilities on 60 of 104 rows"
        " by up to 0.13, so a combination that consults Jev inherits that slack. These are"
        " the same compositions at "
        f"{DEFAULT_THRESHOLD:.2f} against each repeated run. `escalate` consults nothing, so"
        " its range is zero by construction and is the control for the reading: where an"
        " `escalate` range is not zero the fault is in this document, not in the model.",
        "",
        f"### Hard set, 300 commands, every run at {DEFAULT_THRESHOLD:.2f}",
        "",
        _repeat_rows_table(hard_rows),
        "",
        f"### Hard set, 300 commands, the range across "
        f"{len(SWEEP_SECONDS_HARD)} runs at {DEFAULT_THRESHOLD:.2f}",
        "",
        _repeat_range_table(hard_rows),
        "",
        f"### Hand set, 104 commands, every run at {DEFAULT_THRESHOLD:.2f}",
        "",
        _repeat_rows_table(hand_rows),
        "",
        f"### Hand set, 104 commands, the range across "
        f"{len(SWEEP_SECONDS_HAND)} runs at {DEFAULT_THRESHOLD:.2f}",
        "",
        _repeat_range_table(hand_rows),
        "",
        "## The same range at every swept threshold, hard set",
        "",
        "An operating point read off the sweep above needs its noise band at the threshold"
        " it was read at, not only at "
        f"{DEFAULT_THRESHOLD:.2f}. This is the range each combination covers across the"
        f" {len(SWEEP_SECONDS_HARD)} hard-set Jev runs, at each of the six thresholds.",
        "",
        _threshold_range_table(hard_curve),
        "",
        "## The inputs",
        "",
        "| result file | SHA-256 |",
        "|---|---|",
    ]
    parts += [f"| `results/{stem}.json` | `{cache.digest(stem)}` |" for stem in inputs]
    parts.append("")
    _dbg("OC_CHAT.MAIN", func,
         f"rendered {len(sweep_rows)} sweep rows and "
         f"{len(hard_rows) + len(hand_rows) + len(hard_curve)} repeat rows", 2)
    return "\n".join(parts)
 
 
def _build_parser() -> argparse.ArgumentParser:
    """Construct the CLI argument parser."""
 
    parser = argparse.ArgumentParser(
        prog="python bench/argv_policy/combine.py",
        description="Combine two argv-policy result files: the first decides, the "
                    "second is consulted where the first is unsure.",
    )
    parser.add_argument("--first", default=None,
                        help="Result file of the model that DECIDES.")
    parser.add_argument("--second", default=None,
                        help="Result file of the model CONSULTED where the first is "
                             "unsure. Optional for --rule escalate, which consults "
                             "nothing and only records which pairing it controls for.")
    parser.add_argument("--rule", choices=RULES, default=None,
                        help="How an unsure row is decided.")
    parser.add_argument("--threshold", type=float, default=DEFAULT_THRESHOLD,
                        help=f"Confidence below which the first model is unsure "
                             f"(default: {DEFAULT_THRESHOLD}). The comparison is strict.")
    parser.add_argument("--out", default=None,
                        help="Where to write the combined result file, or the sweep "
                             "document under --sweep.")
    parser.add_argument("--date", default=None,
                        help="The date recorded in provenance (default: today). Pin it "
                             "to reproduce a committed file byte for byte.")
    parser.add_argument("--sweep", action="store_true",
                        help="Render the threshold sweep and the repeated-run movement "
                             "to results/combined_sweep.md instead of combining once.")
    parser.add_argument("--results-dir", default=None,
                        help="Directory the sweep reads (default: results/ beside "
                             "this script).")
    return parser
 
 
def _run_sweep(args: argparse.Namespace, bench_dir: Path) -> int:
    """The ``--sweep`` path: render the document and write it."""
 
    func = "_run_sweep"
    results_dir = Path(args.results_dir) if args.results_dir else bench_dir / "results"
    if not results_dir.is_dir():
        print(f"combine: results directory not found: {results_dir}", file=sys.stderr)
        return 2
    out = Path(args.out) if args.out else results_dir / "combined_sweep.md"
    out.write_text(render_sweep(results_dir, bench_dir), encoding="utf-8")
    print(f"combine: wrote {out}")
    _dbg("OC_CHAT.MAIN", func, f"wrote {out}", 2)
    return 0
 
 
def _run_combination(args: argparse.Namespace, bench_dir: Path) -> int:
    """The single-combination path: compose two files and write the result."""
 
    func = "_run_combination"
    if not args.first or not args.rule or not args.out:
        print("combine: --first, --rule and --out are required without --sweep",
              file=sys.stderr)
        return 2
    if args.second is None and args.rule in CONSULTING_RULES:
        print(f"combine: --rule {args.rule} consults a second model, so --second "
              "is required", file=sys.stderr)
        return 2
    first_path = Path(args.first)
    second_path = Path(args.second) if args.second else None
    first_payload = load_payload(first_path)
    second_payload = load_payload(second_path) if second_path else None
    date = args.date or datetime.date.today().isoformat()
    payload = build_payload(first_payload, second_payload, rule=args.rule,
                            threshold=args.threshold, first_path=first_path,
                            second_path=second_path, bench_dir=bench_dir, date=date)
    Path(args.out).write_text(json.dumps(payload, indent=1, sort_keys=True),
                              encoding="utf-8")
    provenance = payload["provenance"]
    assert isinstance(provenance, dict)
    print(f"combine: wrote {args.out}")
    print(f"combine: {payload['model']} against {payload['condition']}, "
          f"{provenance['consulted']} of {len(payload['results'])} consulted, "
          f"{provenance['disagreeing']} disagreeing")
    _dbg("OC_CHAT.MAIN", func, f"wrote {args.out}", 2)
    return 0
 
 
def main(argv: list[str] | None = None) -> int:
    """CLI entry point: one combination, or the whole sweep document."""
 
    args = _build_parser().parse_args(argv)
    bench_dir = Path(__file__).resolve().parent
    try:
        if args.sweep:
            return _run_sweep(args, bench_dir)
        return _run_combination(args, bench_dir)
    except CombineError as error:
        print(f"combine: {error}", file=sys.stderr)
        return 2
 
 
if __name__ == "__main__":
    sys.exit(main())