dds

#!/usr/bin/env python3
"""batch_judge.py — the judged leg as a Message Batches run (50% token
pricing), one request per nominee instead of one long agent session.

The monolithic agent leg's cost is quadratic in turns: every reply
re-reads the whole transcript. Here each nominee gets its own small
conversation (shared, cache-controlled system prefix + its ADR), and
the investigation loop is synchronized rounds: submit a batch, execute
the tool calls it returns locally (tsq/read), submit the next round.
The contract is unchanged and still externally enforced — the verdicts
file this writes goes through check_join's quote verification and
evidence replay exactly like an interactive agent's.

Scope: platforms with a MECHANICAL witness only. If the instrument
requests extraction (unprofiled platform), this driver refuses — batch
rounds are the wrong shape for open-ended extraction; use the
interactive agent leg there.

Usage:
    batch_judge.py --repo . --base origin/master \
        --out .dds-verdicts.json [--model claude-sonnet-5]
        [--samples 3] [--max-rounds 8] [--poll-seconds 10] [--deadline-min 30]

Stability (issue #45): a nominee whose FIRST sample lands anything but
conformant is borderline — the driver draws the remaining samples as
independent fresh conversations and the modal verdict stands (a tie
resolves to the most severe). The chosen sample's record carries a
`samples` list; the instrument surfaces splits as stability notes.
Seed-inherited verdicts skip sampling: replayed evidence is already
the determinism lever.

Env: ANTHROPIC_API_KEY. Deps: stdlib (tsq.py's tree-sitter deps must be
installed for the searches the model requests — same deps the witness
generation step already installed).
"""

import argparse
import json
import os
import re
import subprocess
import sys
import time
import urllib.error
import urllib.request
from functools import partial

print = partial(print, flush=True)  # live CI logs, not end-of-step bursts

API = "https://api.anthropic.com/v1/messages/batches"
HERE = os.path.dirname(os.path.abspath(__file__))

sys.path.insert(0, HERE)
from verdicts import VERDICTS, Verifier, rule_scope  # the shared verifier (W25)

SYSTEM = """You are the judged leg of /dds-check: one narrow per-ADR
conformance verdict, produced by investigation, enforced externally.

You are given ONE nominated ADR (its bytes, verbatim). Decide whether
the CURRENT tree satisfies its rule, then call submit_verdict exactly
once. Your record is verified by an instrument you do not control:
rule_quote is substring-checked against the ADR's ## Rule section
(quote the rule's own sentence, never the preamble), and every
evidence record is RE-EXECUTED against the tree — a claim that does not
reproduce rejects the whole verdict. So: quote, never paraphrase; copy
match records EXACTLY as run_searches returned them; cite only what a
tool showed you this run.

Verdict enum:
- conformant — the rule's satisfaction is POSITIVELY shown by evidence
  (an interim rule that accurately declares the witnessed state,
  including a declared absence, is conformant AS an interim).
- partial — a real, named gap: the mechanism exists but a path escapes
  it. MUST carry >=1 evidence record pointing at the gap.
- unwitnessed — the rule names witnessable things and the tree shows
  none of them. Never use this for off-topic rules.
- not_verifiable — the rule's claims name nothing witnessable by
  structural search (process rules, external systems). MUST name the
  searches you ran (evidence with expect "none" is how a shrug becomes
  a witnessed absence).

Your searches are RETRIEVAL NETS, not proofs: a query defines the
evaluated population (every call site, every matching structure); YOU
evaluate the retrieved source — that judgment is trusted; the net is
what gets re-verified. Prefer BROAD nets and filter visibly in your
basis ("matches 3 and 7 are test fixtures, irrelevant because …") —
an over-fetching query costs only reading; an under-fetching one
silently shrinks the population, the one failure replay cannot catch.
Read files freely to understand context; reading informs your basis,
but citations come only from search output.

Evidence record shape (each one is replayed):
  {"query": "call <name>"  |  "query_scm": "<the tree-sitter query>",
   "in": "<path prefix, optional>",
   "expect": "some" | "none",
   "matches": [{"at": "file:line", "text": "<span EXACTLY as returned>"}],
   "controls": [{"source": "<code fragment>", "matches": 1}],  // absence-via-scm only
   "note": "what this operationalizes"}
"expect": "some" lists matches that must reproduce. "expect": "none"
lists no matches; via `call` it additionally fails if plain-text hits
exist for the name; via a constructed query it REQUIRES positive
controls — code fragments your query provably matches (declared
count) — because a net with no proven teeth cannot witness absence.

Discipline (violations are REJECTED back to you with reasons — fix
and resubmit): matches are copied EXACTLY from run_searches output,
never from read_file and never from memory; citable queries are
`call <name>`, a profile query_ref, or the constructed query EXACTLY
as you ran it (query_scm); an absence via `call` needs textual_hits: 0,
an absence via query_scm needs controls.

Guards: judge WITNESSED state only — never your recollection of how
such apps usually work; conformant requires positive evidence, never
absence-of-findings; judge the rule's intent at the facts' grain. When
a rule RESISTS operationalization even with constructed queries — its
subjects are process, external systems, or prose rather than code
structure — the honest verdict is not_verifiable naming the searches
you ran; never force a partial on prose evidence, and never invent a
query form the instrument cannot replay.

Round shape (batch rounds are minutes apart — replies are precious):
typically reply 1 is ONE run_searches call carrying every net this ADR
needs (broad; add read_file calls in the same reply), and reply 2 is
submit_verdict. Investigate further only when the retrieved source
genuinely cannot decide the verdict — each extra round must earn its
delay. Do not narrate a report; the record is the deliverable."""

TOOLS = [
    {
        "name": "run_searches",
        "description": (
            "Run structural searches (tree-sitter call queries) over the "
            "repository. Every match's text is the exact source span — "
            "copy at/text EXACTLY into evidence matches and replay "
            "verification passes by construction. Batch all the searches "
            "you need into one call."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "searches": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "name": {
                                "type": "string",
                                "description": "call target, e.g. get_env or Repo.delete_all",
                            },
                            "scm": {
                                "type": "string",
                                "description": (
                                    "a constructed tree-sitter query (raw .scm) — "
                                    "the retrieval net; use instead of name for "
                                    "structure beyond call sites"
                                ),
                            },
                            "path_prefix": {
                                "type": "string",
                                "description": "limit to paths under this prefix, e.g. lib/",
                            },
                        },
                    },
                }
            },
            "required": ["searches"],
        },
    },
    {
        "name": "read_file",
        "description": (
            "Read a slice of a repository file (for config or module "
            "context a search pointed you at). Capped output."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "start_line": {"type": "integer"},
                "end_line": {"type": "integer"},
            },
            "required": ["path"],
        },
    },
    {
        "name": "submit_verdict",
        "description": "Submit the one verdict record for this ADR. Terminal.",
        "input_schema": {
            "type": "object",
            "properties": {
                "verdict": {"type": "string", "enum": VERDICTS},
                "rule_quote": {"type": "string"},
                "basis": {"type": "string"},
                "evidence": {"type": "array", "items": {"type": "object"}},
                "dormant_mechanism": {"type": "string"},
            },
            "required": ["verdict", "rule_quote", "basis"],
        },
    },
]


def http(method, url, body=None, key=None, tries=4):
    for attempt in range(tries):
        req = urllib.request.Request(url, method=method)
        req.add_header("x-api-key", key)
        req.add_header("anthropic-version", "2023-06-01")
        if body is not None:
            req.add_header("content-type", "application/json")
        try:
            with urllib.request.urlopen(
                req, json.dumps(body).encode() if body is not None else None, timeout=120
            ) as r:
                return r.read().decode()
        except urllib.error.HTTPError as e:
            if e.code in (429, 500, 502, 503, 529) and attempt < tries - 1:
                time.sleep(5 * (attempt + 1))
                continue
            raise SystemExit(f"batch API {method} {url}: HTTP {e.code}: {e.read().decode()[:300]}")
        except urllib.error.URLError as e:
            if attempt < tries - 1:
                time.sleep(5 * (attempt + 1))
                continue
            raise SystemExit(f"batch API unreachable: {e}")


def exec_scm_search(repo, scm, prefix):
    """A constructed retrieval net, run in-process. Returns (matches, err)."""
    eng = tsq_mod()
    lang = eng.LANGS["elixir"]
    try:
        eng.tree_sitter.Query(lang, scm)
    except Exception as e:
        return None, f"query does not compile: {str(e)[:200]}"
    parser = eng.tree_sitter.Parser(lang)
    matches = []
    for rel in eng.code_files(repo, [prefix] if prefix else []):
        try:
            src = open(os.path.join(repo, rel), "rb").read()
        except OSError:
            continue
        root = parser.parse(src).root_node
        for node in eng.run_scm(lang, src, root, scm):
            matches.append(
                {
                    "file": rel,
                    "line": node.start_point[0] + 1,
                    "text": src[node.start_byte : node.end_byte].decode(errors="replace"),
                }
            )
    return matches, None


def exec_run_searches(repo, inp):
    out = []
    searches = inp.get("searches") or []
    if isinstance(searches, (str, dict)):
        searches = [searches]
    for s in searches[:24]:
        if isinstance(s, str):
            s = {"name": s}
        if not isinstance(s, dict):
            out.append({"error": f"unrecognized search entry: {s!r}"})
            continue
        prefix = s.get("path_prefix")
        if s.get("scm"):
            matches, err = exec_scm_search(repo, s["scm"], prefix)
            if err:
                out.append({"query": "scm", "error": err})
                continue
            rec_out = {
                "query": "scm",
                "scm": s["scm"],
                "in": prefix,
                "count": len(matches),
                "matches": [
                    {"at": f"{m['file']}:{m['line']}", "text": m["text"][:1000]}
                    for m in matches[:30]
                ],
                "truncated": len(matches) > 30,
            }
            if not matches:
                rec_out["note"] = (
                    "0 matches — citing this as an ABSENCE requires positive "
                    "controls in the evidence record: {source: <a code fragment "
                    "the query DOES match>, matches: N}. A net with no proven "
                    "teeth cannot witness absence."
                )
            out.append(rec_out)
            continue
        name = s.get("name") or ""
        cmd = [sys.executable, os.path.join(HERE, "tsq.py"), "call", name, "--repo", repo, "--json"]
        if prefix:
            cmd += ["--in", prefix]
        r = subprocess.run(cmd, capture_output=True, text=True)
        if r.returncode != 0:
            out.append({"query": f"call {name}", "error": r.stderr.strip()[:200]})
            continue
        matches = json.loads(r.stdout or "[]")
        rec_out = {
            "query": f"call {name}",
            "in": prefix,
            "count": len(matches),
            "matches": [
                {"at": f"{m['file']}:{m['line']}", "text": m["text"][:1000]}
                for m in matches[:30]
            ],
            "truncated": len(matches) > 30,
        }
        if not matches:
            th = tsq_mod().textual_hits(repo, [prefix] if prefix else [], name)
            rec_out["textual_hits"] = th
            if th:
                rec_out["note"] = (
                    "0 structural matches but textual hits exist — this absence "
                    "is NOT citable. Try the MORE SPECIFIC form (full dotted "
                    "Module.fun) whose textual_hits is 0; a name that appears in "
                    "comments or docs anywhere in scope can never witness absence"
                )
        out.append(rec_out)
    return json.dumps(out, indent=1)


def exec_read_file(repo, inp):
    rel = os.path.normpath(inp.get("path") or "")
    if rel.startswith("..") or rel.startswith("/") or rel.startswith(".git"):
        return "refused: path escapes the repository"
    try:
        lines = open(os.path.join(repo, rel), encoding="utf-8", errors="replace").read().splitlines()
    except OSError as e:
        return f"unreadable: {e}"
    a = max(1, int(inp.get("start_line") or 1))
    b = min(len(lines), int(inp.get("end_line") or a + 120))
    body = "\n".join(f"{i}: {l}" for i, l in zip(range(a, b + 1), lines[a - 1 : b]))
    return body[:6000] or "(empty slice)"


_TSQ = None


def tsq_mod():
    global _TSQ
    if _TSQ is None:
        sys.path.insert(0, HERE)
        import tsq

        _TSQ = tsq
    return _TSQ


def validate_verdict(repo, adr_rel, rec, profile_queries):
    """The gate's replay checks, run at ROUND time so the model can
    correct — the batch equivalent of the interactive leg's re-run-the-
    instrument-until-clean loop. The checks ARE the gate's: the shared
    verdicts.Verifier, so round-time acceptance and gate acceptance can
    never drift (W25)."""
    try:
        adr = open(os.path.join(repo, adr_rel), encoding="utf-8", errors="replace").read()
    except OSError:
        adr = ""
    scope = rule_scope(adr) or adr
    problems, _notes = Verifier(repo, profile_queries, tsq_mod()).validate(
        scope, rec, replayable=True
    )
    return problems


def custom_id(stem):
    return re.sub(r"[^a-zA-Z0-9_-]", "-", stem)[:60]


SEV = ["conformant", "partial", "unwitnessed", "not_verifiable"]


def modal_verdict(recs):
    """The stabilized record from N sampled verdicts (issue #45): the
    modal verdict wins; a tie resolves to the MOST SEVERE — a coin toss
    never lands on the friendly answer. The chosen sample's record is
    returned whole (its evidence is what replays at the gate), carrying
    the full sample list so the walk's stability notes can surface a
    split."""
    counts = {}
    for r in recs:
        counts[r["verdict"]] = counts.get(r["verdict"], 0) + 1
    top = max(counts.values())
    winner = max((v for v, c in counts.items() if c == top), key=SEV.index)
    chosen = dict(next(r for r in recs if r["verdict"] == winner))
    if len(recs) > 1:
        chosen["samples"] = [r["verdict"] for r in recs]
    return chosen


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--repo", default=".")
    ap.add_argument("--base", required=True)
    ap.add_argument("--out", required=True)
    ap.add_argument("--model", default="claude-sonnet-5")
    ap.add_argument("--seed", help="prior verdicts to inherit where evidence replays clean")
    ap.add_argument("--samples", type=int, default=3,
                    help="samples for a borderline (non-conformant first) verdict; modal wins (1 disables)")
    ap.add_argument("--max-rounds", type=int, default=12)
    ap.add_argument("--poll-seconds", type=int, default=10)
    ap.add_argument("--deadline-min", type=int, default=50)
    args = ap.parse_args()

    key = os.environ.get("ANTHROPIC_API_KEY")
    if not key:
        raise SystemExit("ANTHROPIC_API_KEY not set")

    # pre-pass: what does the instrument nominate?
    pre = os.path.join(args.repo, ".dds-batch-pre.json")
    r = subprocess.run(
        [
            sys.executable,
            os.path.join(HERE, "check_join.py"),
            "--repo", args.repo, "--base", args.base, "--json", pre,
        ],
        capture_output=True, text=True,
    )
    if r.returncode not in (0, 1, 2):
        raise SystemExit(f"instrument pre-pass failed: {r.stderr[:400]}")
    doc = json.load(open(pre))
    os.unlink(pre)
    if doc.get("needs_extraction"):
        raise SystemExit(
            "instrument requests per-file extraction (unprofiled platform) — "
            "the batch driver serves mechanical-witness platforms only; run "
            "the interactive agent leg"
        )
    nominees = doc.get("nominees") or []
    print(f"{len(nominees)} nominee(s); model {args.model}; batched judged leg")
    if not nominees:
        json.dump([], open(args.out, "w"))
        return

    profile_queries = {}
    for p in doc.get("platforms") or []:
        profile_queries.update(p.get("queries") or {})

    seeds = {}
    if args.seed and os.path.exists(args.seed):
        try:
            for v in json.load(open(args.seed)) or []:
                if isinstance(v, dict) and v.get("adr") and v.get("verdict") in VERDICTS:
                    seeds[v["adr"]] = v
        except (OSError, json.JSONDecodeError):
            print("seed file unreadable — judging everything fresh")

    # keys: a nominee is a STEM; each judge conversation is a SAMPLE
    # `stem~k`. Sample 1 always runs; a non-conformant sample 1 is
    # borderline and draws the extra samples (issue #45) — the modal
    # verdict becomes the nominee's record.
    state, verdicts, nudged, failed, vretries = {}, {}, set(), {}, {}
    openings, samples_of, sample_results = {}, {}, {}
    for n in nominees:
        stem = os.path.basename(n["adr"])[:-3] if n["adr"].endswith(".md") else n["adr"]
        adr_bytes = open(os.path.join(args.repo, n["adr"]), encoding="utf-8", errors="replace").read()
        opening = (
            f"ADR under judgment: {n['adr']} (Status as parsed: {n.get('status')})\n"
            + (
                f"Profile query_refs you may also cite: {sorted(profile_queries)}\n"
                if profile_queries
                else ""
            )
            + f"\n----- ADR BYTES -----\n{adr_bytes}\n----- END ADR -----\n\n"
            "Investigate with run_searches (one batched call), then call "
            "submit_verdict."
        )
        openings[stem] = {"adr": n["adr"], "opening": opening}
        sv = seeds.get(n["adr"])
        if sv and not validate_verdict(args.repo, n["adr"], sv, profile_queries):
            # the prior judgment's evidence replays clean on THIS tree —
            # the translation is still grounded; inherit it, already
            # stabilized (no sampling: replay IS the determinism lever).
            # (The gate re-replays everything again regardless.)
            verdicts[stem] = sv

    usage = {"input_tokens": 0, "output_tokens": 0,
             "cache_creation_input_tokens": 0, "cache_read_input_tokens": 0}
    deadline = time.time() + args.deadline_min * 60
    open_batches, rounds = {}, {}

    def submit(stem):
        # `stem` here is a SAMPLE id (`<stem>~<k>`) — one conversation
        rounds[stem] = rounds.get(stem, 0) + 1
        if rounds[stem] > args.max_rounds:
            failed[stem] = "round budget exhausted"
            print(f"  {stem}: round budget exhausted")
            return
        req = {
            "custom_id": custom_id(stem),
            "params": {
                "model": args.model,
                "max_tokens": 4000,
                "system": [{"type": "text", "text": SYSTEM,
                            "cache_control": {"type": "ephemeral"}}],
                "tools": TOOLS,
                "messages": state[stem]["messages"],
            },
        }
        if rounds[stem] >= args.max_rounds:
            # the last budgeted reply MUST be the verdict — forced, not asked
            req["params"]["tool_choice"] = {"type": "tool", "name": "submit_verdict"}
        if rounds[stem] == args.max_rounds:
            warn = (
                "FINAL reply: you are out of budget after this — call "
                "submit_verdict now with your best-supported record. An honest "
                "not_verifiable naming clean searches (textual_hits: 0) beats "
                "delivering nothing."
            )
            last = state[stem]["messages"][-1]
            if isinstance(last["content"], list):
                last["content"] = list(last["content"]) + [{"type": "text", "text": warn}]
            else:
                last["content"] = f"{last['content']}\n\n{warn}"
        b = json.loads(http("POST", API, {"requests": [req]}, key))
        open_batches[stem] = b["id"]

    def launch(stem, k):
        sid = f"{stem}~{k}"
        state[sid] = {
            "adr": openings[stem]["adr"],
            "messages": [{"role": "user", "content": openings[stem]["opening"]}],
        }
        samples_of.setdefault(stem, []).append(sid)
        submit(sid)

    def maybe_finalize(stem, force=False):
        """When every sample of a nominee has landed (or, at the
        deadline, with whatever landed), the modal record stands."""
        if stem in verdicts:
            return
        sids = samples_of.get(stem) or []
        if not force and any(s not in sample_results and s not in failed for s in sids):
            return
        recs = [sample_results[s] for s in sids if s in sample_results]
        if not recs:
            return  # every sample failed — an honestly missing verdict
        rec = modal_verdict(recs)
        verdicts[stem] = rec
        if len(recs) > 1:
            split = " (split → modal)" if len(set(rec["samples"])) > 1 else ""
            print(f"  {stem}: samples {rec['samples']} → {rec['verdict']}{split}")

    def tool_result_for(t):
        try:
            if t["name"] == "run_searches":
                return exec_run_searches(args.repo, t["input"])
            if t["name"] == "read_file":
                return exec_read_file(args.repo, t["input"])
            return f"unknown tool {t['name']}"
        except Exception as e:  # a bad input is the MODEL's error
            return f"tool error: {e} — fix the input shape and retry"

    def handle_message(stem, msg):
        """Advance one nominee's conversation; True = it continues."""
        state[stem]["messages"].append({"role": "assistant", "content": msg["content"]})
        tool_uses = [c for c in msg["content"] if c.get("type") == "tool_use"]
        submitted = next((t for t in tool_uses if t["name"] == "submit_verdict"), None)
        if submitted:
            rec = dict(submitted["input"])
            rec["adr"] = state[stem]["adr"]
            if rec.get("verdict") not in VERDICTS:
                failed[stem] = f"bad verdict enum: {rec.get('verdict')}"
                return False
            problems = validate_verdict(args.repo, rec["adr"], rec, profile_queries)
            if not problems or rounds.get(stem, 0) >= args.max_rounds:
                if problems:
                    print(f"  {stem}: accepted with {len(problems)} unresolved "
                          "problem(s) — the gate decides")
                sample_results[stem] = rec
                print(f"  {stem}: {rec['verdict']}")
                base, k = stem.rsplit("~", 1)
                if k == "1" and rec["verdict"] != "conformant" and args.samples > 1:
                    # borderline first sample: one flaky judgment must not
                    # decide the record alone — draw independent samples
                    print(f"  {base}: borderline — drawing {args.samples - 1} more sample(s)")
                    for i in range(2, args.samples + 1):
                        launch(base, i)
                return False
            vretries[stem] = vretries.get(stem, 0) + 1
            rejection = (
                "REJECTED — the instrument will refuse this verdict:\n- "
                + "\n- ".join(problems)
                + "\nFix and call submit_verdict again. Matches are copied "
                "EXACTLY from run_searches output (expect \"some\" must LIST "
                "them); only `call <name>` queries (or profile query_refs) are "
                "citable; an absence is citable only with textual_hits: 0."
            )
            results = [
                {"type": "tool_result", "tool_use_id": t["id"],
                 "content": rejection if t["id"] == submitted["id"] else tool_result_for(t),
                 **({"is_error": True} if t["id"] == submitted["id"] else {})}
                for t in tool_uses
            ]
            state[stem]["messages"].append({"role": "user", "content": results})
            print(f"  {stem}: rejected at validation ({len(problems)} problem(s)), retrying")
            for prob in problems[:3]:
                print(f"      · {prob[:180]}")
            return True
        if tool_uses:
            results = [
                {"type": "tool_result", "tool_use_id": t["id"], "content": tool_result_for(t)}
                for t in tool_uses
            ]
            state[stem]["messages"].append({"role": "user", "content": results})
            return True
        if stem not in nudged:
            nudged.add(stem)
            state[stem]["messages"].append(
                {"role": "user",
                 "content": "Finish by calling submit_verdict — prose is not a deliverable."}
            )
            return True
        failed[stem] = "ended twice without submit_verdict"
        return False

    if verdicts:
        print(f"inherited {len(verdicts)} verdict(s) whose evidence replays clean on this tree")

    # per-nominee pipelines: each conversation advances the moment ITS
    # batch ends — no cohort barrier, a straggler delays only itself.
    for stem in openings:
        if stem not in verdicts:
            launch(stem, 1)
    print(f"{len(open_batches)} nominee pipeline(s) launched")

    while open_batches:
        if time.time() > deadline:
            print(f"deadline reached with {len(open_batches)} nominee(s) unjudged")
            break
        time.sleep(args.poll_seconds)
        done_this_pass = 0
        for stem in list(open_batches):
            bid = open_batches[stem]
            b = json.loads(http("GET", f"{API}/{bid}", key=key))
            if b["processing_status"] != "ended":
                continue
            del open_batches[stem]
            done_this_pass += 1
            lines = http("GET", b["results_url"], key=key).splitlines()
            res = json.loads(lines[0]) if lines else None
            if not res or res["result"]["type"] != "succeeded":
                failed[stem] = res["result"]["type"] if res else "no result"
                print(f"  {stem}: request {failed[stem]}")
                maybe_finalize(stem.rsplit("~", 1)[0])
                continue
            msg = res["result"]["message"]
            for k in usage:
                usage[k] += msg.get("usage", {}).get(k) or 0
            if handle_message(stem, msg):
                submit(stem)
            if stem not in open_batches:
                # this sample is over (verdict, failure, or budget) —
                # its nominee may now be decidable
                maybe_finalize(stem.rsplit("~", 1)[0])
        if done_this_pass:
            print(f"  … {len(verdicts)} judged, {len(open_batches)} in flight")

    # deadline stragglers: a nominee whose sample 1 landed but whose
    # extra samples did not finalizes on what arrived — a partial sample
    # set beats dropping the nominee to INCOMPLETE.
    for stem in samples_of:
        maybe_finalize(stem, force=True)

    out = [verdicts[s] for s in verdicts]
    json.dump(out, open(args.out, "w"), indent=1)
    # batch pricing = 50% of standard sonnet rates, per Mtok
    cost = (
        usage["input_tokens"] * 3.0
        + usage["output_tokens"] * 15.0
        + usage["cache_creation_input_tokens"] * 3.75
        + usage["cache_read_input_tokens"] * 0.30
    ) * 0.5 / 1_000_000
    line = (
        f"judged {len(out)}/{len(nominees)}"
        + (f" (failed: {failed})" if failed else "")
        + f" | tokens in={usage['input_tokens']} out={usage['output_tokens']}"
        f" cache_w={usage['cache_creation_input_tokens']}"
        f" cache_r={usage['cache_read_input_tokens']}"
        f" | est cost ${cost:.2f} (batch rates)"
    )
    print(line)
    summary = os.environ.get("GITHUB_STEP_SUMMARY")
    if summary:
        with open(summary, "a") as f:
            f.write(f"### Judged leg (batched)\n{line}\n")
    if len(out) < len(nominees):
        # the instrument's completeness gate turns this into INCOMPLETE
        print("NOTE: missing verdicts will fail the gate honestly")


if __name__ == "__main__":
    main()