#!/usr/bin/env python3
"""The §7 bench — MECHANICAL ONLY. No quality verdict is claimed or claimable.

    python3 bench/run.py --concurrency 1 --arm cold --seed "<what the unit ran>"
    python3 bench/run.py --concurrency 6 --arm warm --seed "no --seed flag" --out report.json

WHAT THIS MEASURES, and it is worth being blunt about the boundary: whether the
seat's answers are GROUNDED IN THE ARTICLE by the only tests a machine can run —
every figure verbatim, every cite real, no redaction hit, and the abstain when
the page does not say. It measures nothing about whether an answer is GOOD. The
judge seat is empty by the 2026-09-08 bench's own verdict, and un-judged model
text cannot grade a model, so the page states that groundedness is not measured
until a judge passes its own bench.

WHY IT CALLS THE SEAT AND NOT THE DOOR. The fences are deliberately tight — three
free-text calls per peer per minute, sixty an hour for everyone — so a
sixty-question run driven through `POST /ask` would spend four hours being
refused, and a run with the fences widened would not be measuring the deployed
posture anyway. The FENCES are the suite's subject (`tests/test_server.py`); the
SEAT is this one's. Everything between them is shared code: the same corpus
loader, the same prompt, the same five gates, in the same order.

WHAT IT RECORDS THAT IS NOT A NUMBER. The seat's SEED, as `--seed`, verbatim and
unvalidated. A fixed seed makes every visitor's same question byte-identical —
right for a bench, wrong for a reading room — and the seat's serving unit drops `--seed`
in the same change that adds the bearer. A run that cannot say which side of that
change it was taken on is a run whose numbers cannot be compared to the next one,
so the flag is REQUIRED and the report carries it in full.

NOTHING HERE IS RUN BY A LANE. R54 must be live and the bank frozen; the
orchestrator runs it from the serving box after the deploy.

TWO THINGS WERE RE-SPECIFIED ON 2026-09-09, both off the FIRST run's own rows,
and both are recorded here rather than in a plan because this file is what the
next run executes.

  1. **EACH ARM IS SCORED ON ITS OWN DENOMINATOR, not on 54 of 60.** SPEC §7's
     number conflated the two: 60 is every question in the bank and only the
     on-page ones can be answered, so a run could reach "54 answered" only by
     counting off-page abstains as answers — the exact thing the off-page gate
     exists to require. The on-page floor was 35 of 39 when it was ruled; the
     two rooms poured after the bank was frozen got their questions on
     2026-09-09 (§6.5), so the arm is 41 and the floor is **37 of 41** — the
     same allowance of four failures the ruling set, over a longer arm. The
     off-page floor is unchanged at 21 of 21.
  2. **GATE (d) COUNTS THE ABSTAIN THAT REACHED THE READER, not the string the
     seat happened to emit.** The first run measured 18 of 21 off-page questions
     as failures on rows where the seat had refused CORRECTLY — *"the article
     does not answer the question about …"* — in its own words. The service now
     recognises that shape and replaces it with the authored abstain
     (`docent/normalise.py`, D-20260909-22); this harness applies the same
     substitution, in the same order, and scores `abstained`. What the gate asks
     is unchanged: did a reader get the workshop's own refusal, character for
     character. What changed is that the service now makes that true when the
     seat is right, instead of failing the release for it.

Every row also carries `gate_detail` — the gates' own reasons, the tokens gate
(d) refused and the cites the cite gate dropped — so the next report can show
the token rather than a boolean. The first one could not, and four replaced
answers could not be triaged at all.

AN INSTRUMENT FIX, 2026-09-09 (off the THIRD run's rows). A CITE IS AN ADDRESS
IN THIS FILE TOO. v0.1.2 taught the SERVICE's four prose gates to read
`gates.prose_of(answer)` — the answer with every `§ {#id}` token cut — because a
section id is a hyphenated slug of the heading's own words and no article
contains it. This harness's own `figures` column was left reading the RAW answer,
so `non_verbatim_figures` went on counting the digits inside those ids: run #3's
only failing gate was rows whose "figures" were `2-2826` and `38-49`, both lifted
out of the id `2-2826-over-56-metered-calls-against-38-49-of-caps`, and `96`, out
of `the-2026-08-field-six-arms-on-the-96g-workstation-blind`. The service passed
those answers and the bench failed them, which makes the disagreement the
instrument's and not the seat's.

`figures_of` below is the one extractor, and it is the SERVICE's cutter imported
rather than a second copy of it: `gates.prose_of` is the definition, here and at
the door, and the day it changes there is no second place to remember. Pinned on
the run's own committed answers in `tests/test_bench_figures.py`; everything else
in run #3's cold arms passed (on-page 39/41, off-page 21/21, redaction 0,
p50 0.97 s, p95 7.2 s).

A SECOND INSTRUMENT CORRECTION, 2026-09-09 — THE WARM ARM WAS MEASURING
EVICTION. The arm warmed ALL 41 pages in one pre-pass and then walked 62
questions in bank order. The corpus is 571,779 tokens against a KV pool of
359,764 (SPEC §2, 1.59×), so a page warmed at the top of the pre-pass is
EVICTED long before its question is asked: every row in the arm read
`cached_tokens: 32` — the shared preamble and nothing else — and the longest
page cold-prefilled at 10–13 s, which failed the p95 ceiling. The ceiling was
right and the arm was wrong: it was measuring what a reader meets on a page
somebody read forty pages ago, which is the COLD case wearing the warm arm's
name.

MEASURED, not assumed, before the arm was changed (from a workshop machine, through
this repo's own `prompt` module, against the deployed seat): warm 32 cached →
first ask 7,040 → second ask 7,264, on a 7.3k-token page. The prefix cache
works; the arm's ordering was defeating it.

So the warm arm now walks BY PAGE — one warm call immediately before that
page's own questions, then those questions in order, then the next page — which
is the arm's own docstring executed. The cold arm is unchanged: it stays the
corpus walk in bank order, and that IS the first-read case. Both gates are
unchanged (p50 < 3 s, p95 < 8 s, warm arm only). The report prints the warm
arm's CACHE-HIT SHARE beside the latencies — the rows whose `cached_tokens`
reached half their `tokens_in` — because a warm arm that cannot say whether the
cache was hit is an arm that can fail its ceiling for a reason nobody can name,
which is exactly what happened. `tests/test_bench_warm_arm.py` drives the order
against a fake seat.

AND THE PRICE OF THAT ORDERING, STATED RATHER THAN DISCOVERED: **a page is now
the unit of parallelism, so the warm arm's real concurrency is bounded by how
many questions one page carries.** Measured on the frozen bank: 62 questions
over 41 pages, 21 pages with two questions and 20 with one — so `--concurrency
6` on the WARM arm puts at most **2** calls in flight, while the cold arm at 6
still puts 6. The two c6 arms therefore no longer measure the same thing, and a
warm c6 number is not a six-readers-at-once reading any more.

That is a trade, not an oversight: an arm that overlaps pages is an arm whose
warms are evicted again, which is the defect above. It is disclosed rather than
hidden — `effective_concurrency` is computed, carried in the report and printed
on the arm's own line, so a c6 warm number cannot be read as something it is
not. The §7 c6 numbers are REPORTED and not gated (D-20260909-22), which is why
this is a disclosure and not a re-design; restoring six-in-flight on a warmed
prefix needs either more questions per page in the bank or a pipelined arm, and
both are changes to the instrument that want their own pre-registration.
"""

from __future__ import annotations

import argparse
import json
import os
import statistics
import sys
import time
from concurrent.futures import ThreadPoolExecutor

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from docent import config, corpus, fences, gates, normalise, prompt  # noqa: E402
from docent import overlap, register, resolvers, seat  # noqa: E402
from docent.answer import Answer, paragraph  # noqa: E402
from docent import packnorm as pn  # noqa: E402

HERE = os.path.dirname(os.path.abspath(__file__))
PKG = os.path.join(os.path.dirname(HERE), "docent")


def figures_of(answer: str) -> list[str]:
    """The figures in an answer, from the SAME text the service's figure gate
    reads — the answer with its cite addresses cut out.

    One function, called by the row builder and by its tests, so the bench and
    the door cannot disagree about what a figure is. `gates.prose_of` is the
    cutter and it is imported, never copied.
    """
    return pn.figures(gates.prose_of(answer))


def figures_verbatim(answer: str, twin_body: str) -> bool:
    """Whether every figure in that same cite-cut prose is verbatim in the twin.

    THE SERVICE'S GATE ITSELF, not its body copied out. The first cut of this
    function re-implemented `gate_figures` as `pn.figures_all_present(prose,
    twin)` — which is what it does today, and the day it stops being what it
    does the bench diverges from the door again. That is the defect this whole
    correction exists to close, one layer up, so the brief's words apply to it
    too: one function, imported, not a copy.
    """
    return gates.gate_figures(gates.prose_of(answer), twin_body)[0]


def by_page(questions: list[dict]) -> list[tuple[str, list[dict]]]:
    """The bank's questions grouped by slug, in the order the slugs first appear.

    First-appearance rather than sorted, so a run's page order is the bank's own
    and two arms of the same run walk the corpus the same way.
    """
    groups: dict[str, list[dict]] = {}
    for item in questions:
        groups.setdefault(item["slug"], []).append(item)
    return list(groups.items())


def warm_page(settings, article, complete=None) -> bool:
    """One prefix warm for one page — the same call `/warm/<slug>` makes.

    Returns whether it landed. A failed warm is reported and never fatal: the
    arm is still a run, it is just a run whose cache-hit share will say so.
    """
    try:
        (complete or seat.complete)(
            settings, prompt.warm_messages(article.prompt_body), max_tokens=1)
        return True
    except seat.SeatError as exc:
        print("bench: warm failed on %s (%s)"
              % (article.slug, exc.__class__.__name__), file=sys.stderr)
        return False


def run_rows(questions: list[dict], run_one, concurrency: int) -> list[dict]:
    with ThreadPoolExecutor(max_workers=concurrency) as pool:
        return list(pool.map(run_one, questions))


def walk(arm: str, questions: list[dict], *, concurrency: int, run_one,
         warm) -> list[dict]:
    """The order the questions are asked in, and it is the arm's whole meaning.

    COLD is the bank's order, straight through: the first read of each page.

    WARM is page by page — `warm(slug)`, then that page's questions, then the
    next page — because a warm forty pages before its question is not a warm on
    a corpus 1.59× the pool. Concurrency lives INSIDE a page's group so the
    guarantee holds at 6 as well as at 1: no other page's call sits between a
    page's warm and its questions.
    """
    if arm != "warm":
        return run_rows(questions, run_one, concurrency)
    rows: list[dict] = []
    for slug, group in by_page(questions):
        warm(slug)
        rows.extend(run_rows(group, run_one, concurrency))
    return rows


def effective_concurrency(arm: str, questions: list[dict], concurrency: int) -> int:
    """How many calls this arm can actually have in flight at once.

    The cold arm maps every question over one pool, so it is `--concurrency`.
    The warm arm's pool is per PAGE, so it is bounded by the largest page's
    question count — 2 on the frozen bank, whatever `--concurrency` says. It is
    COMPUTED rather than assumed, because a row labelled c6 that is really c2 is
    a figure with no path back to what it counted.
    """
    if arm != "warm" or not questions:
        return concurrency
    return min(concurrency, max(len(group) for _, group in by_page(questions)))


def cache_hit_share(rows: list[dict]) -> tuple[int, int]:
    """(rows whose prefix was at least half served from cache, rows that can say).

    HALF IS THE LINE AND IT IS A THRESHOLD, so it is named here rather than
    buried: a row counts as a hit when `cached_tokens >= tokens_in / 2`. The
    denominator is the rows that carry both numbers — `cached_tokens` is null
    until the seat's unit runs `--enable-prompt-tokens-details` (SPEC §2), and a
    share over rows that cannot answer would be a number with nothing under it.
    """
    countable = [r for r in rows
                 if r.get("cached_tokens") is not None and r.get("tokens_in")]
    hits = sum(1 for r in countable
               if r["cached_tokens"] >= r["tokens_in"] / 2)
    return hits, len(countable)


def load_bank(path: str) -> dict:
    with open(path, encoding="utf-8") as fh:
        return json.load(fh)


def ask(settings, article, question: str, vocabs, copy: dict,
        abstain_vocab, world=None, complete=None) -> dict:
    """One question, through the same code the door runs, timed end to end.

    `complete` IS INJECTED THE WAY THE DOOR INJECTS IT (`Room.complete`), and
    for the same reason: without it the row-building path — which decides what
    the §7 gates count — could only be exercised by spending GPU seconds, so it
    was not exercised at all. A mutation round proved it: blanking the section
    the refusal named left the whole suite green. The default is the real seat,
    so the deployed posture is never one a test chose.

    THE DOOR'S SUBSTITUTION IS APPLIED HERE TOO, and it has to be, or gate (d)
    would measure the wrong thing. A gate that fires replaces the whole answer
    with the abstain before a reader sees a word of it, so `served` — not the
    seat's raw text — is what a reader would have got, and it is what the
    off-page arm is scored against.
    """
    redaction, blocked, pack_extra = vocabs
    started = time.monotonic()
    row = {"slug": article.slug, "q": question}
    try:
        done = (complete or seat.complete)(
            settings, prompt.messages(article.prompt_body, question))
    except seat.SeatError as exc:
        row.update(ok=False, error=exc.__class__.__name__,
                   seconds=time.monotonic() - started)
        return row
    verdict = gates.run(done.text, twin_body=article.prompt_body,
                        anchor_ids=article.anchor_ids, redaction=redaction,
                        blocked=blocked, pack_extra=pack_extra,
                        cite_index=article.cite_index)
    # THE DOOR'S PER-SENTENCE VIEW, run here for the same reason the door's
    # substitution is: what reaches a reader is what this bench scores. A
    # partial release is an answer, and a run that counted it as an abstention
    # would report a floor for a behaviour the service no longer has.
    marks = gates.mark(done.text, verdict, twin_body=article.prompt_body,
                       redaction=redaction, blocked=blocked,
                       pack_extra=pack_extra, cite_index=article.cite_index)
    kept_count, written_count = gates.kept_counts(marks)

    # THE DOOR'S OWN ORDER: the five gates, then the abstain normaliser. A gate
    # that fires replaces the whole answer; past them, a seat that refused in
    # its own words is replaced by the authored refusal. Both are substitutions
    # a reader would have met, so both happen before anything is scored.
    abstain_source, abstain_section = None, None
    partial = False
    if not verdict.ok:
        part = gates.partial_release(
            done.text, marks, twin_body=article.prompt_body,
            anchor_ids=article.anchor_ids, redaction=redaction,
            blocked=blocked, pack_extra=pack_extra,
            cite_index=article.cite_index)
        if part is not None:
            kept_text, _kept_verdict = part
            served = register.sentence_case(kept_text)
            partial = True
        else:
            refusal = resolvers.abstain(article, question, copy, world=world)
            served = refusal.blocks[0].text
            abstain_source = "gate"
            abstain_section = refusal.cites[0] if refusal.cites else None
    else:
        answer = Answer(blocks=[paragraph(register.sentence_case(done.text))],
                        route="docent", cites=verdict.cites)
        answer, matched = normalise.normalise(
            answer, article=article, question=question, copy=copy,
            vocab=abstain_vocab, text=done.text, world=world)
        served = answer.blocks[0].text if answer.blocks else done.text
        if matched:
            abstain_source = "model"
            # THE SECTION THE REFUSAL ACTUALLY NAMED, recorded rather than
            # re-derived. The normaliser uses the model's own first valid cite;
            # `is_the_abstain` used to rebuild the sentence from a fresh token
            # overlap, which names a DIFFERENT section on any question where the
            # two disagree — and then scored a correct refusal as a failure
            # because two authored sentences differed in the middle.
            abstain_section = answer.cites[0] if answer.cites else None
    row.update(
        ok=verdict.ok, answer=done.text, served=served,
        # THE PARTIAL RELEASE, PER ROW (v0.17.0, D-20260912-07). `partial` is
        # what makes a row answered below; the two counts are what let the NEXT
        # run read the partial rate off the file rather than off a summary.
        partial=partial,
        sentences_kept=kept_count, sentences_written=written_count,
        abstained=abstain_source is not None, abstain_source=abstain_source,
        abstain_section=abstain_section,
        seconds=time.monotonic() - started,
        ttft_ms=done.ttft_ms, tokens_in=done.tokens_in, tokens_out=done.tokens_out,
        cached_tokens=done.cached_tokens, cites=verdict.cites,
        gates=verdict.as_trailer(),
        # THE TOKEN, NOT A BOOLEAN. The first run's rows said `new_noun: true`
        # and stopped there, and four replaced answers could not be triaged.
        gate_detail={"reasons": list(verdict.detail),
                     "new_noun_tokens": list(verdict.new_noun_tokens),
                     "new_noun_tokens_withheld": verdict.new_noun_tokens_withheld,
                     "cites_dropped": list(verdict.cites_dropped)},
        # (a) every figure in the answer verbatim in the twin, via packnorm —
        # off the CITE-CUT prose, which is what the service's figure gate reads.
        figures=figures_of(done.text),
        figures_verbatim=figures_verbatim(done.text, article.prompt_body),
        # (b) every cite in that slug's own anchors.
        cites_valid=bool(verdict.cites) and all(
            c in article.anchor_ids for c in verdict.cites),
        # (c) zero redaction hits.
        redaction_hits=len(redaction.hits(done.text)))
    # §6 — CONTAINMENT, MEASURED AND NOT GATED. Per (sentence, cite) pair, how
    # much of the sentence's own vocabulary is in the section it points at; the
    # ROW carries its own median so the bank's distribution is a median of
    # medians rather than a pile of sentences weighted by how many each answer
    # happened to write. Read off the seat's raw text, which is the text the
    # cites belong to: a gated answer's prose was destroyed, and measuring the
    # abstain would be measuring this workshop's own sentence.
    readings = overlap.rows(done.text, article)
    row["cite_overlap"] = readings
    row["cite_overlap_p50"] = overlap.median(r["overlap"] for r in readings)
    return row


def is_the_abstain(served: str, article, question: str, copy: dict,
                   world=None, section_id=None) -> bool:
    """Whether what a reader would have got IS the abstain — character for
    character.

    Not "looks like" and not "contains": §7's gate (d) is the exact text, because
    an answer that merely resembles a refusal is an answer that has started
    improvising about a page that does not say.

    IT REBUILDS WITH THE SECTION THE REFUSAL NAMED, and that is the fix the
    second bench run paid for. The first cut re-derived the nearest section from
    the QUESTION; the normaliser names the model's own first valid cite, so on
    every question where the two disagree the two authored sentences differed in
    the middle and a correct refusal was scored as a failure. Measured on
    r2-cold-c1: all 21 off-page rows carried `abstained: true` and the gate
    printed **9 of 21**.

    NOTE WHAT THIS CAN AND CANNOT SEE, because it bounds the gate. An off-page
    question reaches the abstain when a gate fires — most often the cite gate,
    since a model with nothing to cite tends to cite nothing. An off-page answer
    that passes every gate is one where the model produced a CITED sentence about
    a question the page does not answer; that is the interesting failure, it is
    counted separately as `off_page_cited`, and no machine here can tell a good
    refusal from a bad answer. That is the judge's job and the judge seat is
    empty.
    """
    expected = resolvers.abstain(article, question, copy, world=world,
                                 section_id=section_id)
    return served.strip() == expected.blocks[0].text.strip()


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--bank", default=os.path.join(HERE, "bank.json"))
    ap.add_argument("--concurrency", type=int, default=1, choices=(1, 6),
                    help="§7 runs both: 1 and 6")
    ap.add_argument("--arm", choices=("cold", "warm"), required=True,
                    help="cold is the first read of a page; warm follows a "
                         "prefix warm of the same page")
    ap.add_argument("--seed", required=True,
                    help="what the seat's unit was running, verbatim — recorded, "
                         "never validated. A run that cannot say is a run that "
                         "cannot be compared.")
    ap.add_argument("--out", default=None, help="write the full report here")
    ap.add_argument("--limit", type=int, default=None,
                    help="stop after N questions (a smoke run, never a verdict)")
    args = ap.parse_args(argv)

    settings = config.from_env()
    settings.assert_docroot()
    world = corpus.load(settings.docroot)
    copy = resolvers.load_copy(os.path.join(PKG, "copy.json"))
    vocabs = (fences.load_redaction(), fences.load_blocked(),
              __import__("docent.server", fromlist=["_pack_extra_pattern"])
              ._pack_extra_pattern())
    abstain_vocab = normalise.load()

    bank = load_bank(args.bank)
    questions = bank["questions"][:args.limit] if args.limit else bank["questions"]

    missing = sorted({q["slug"] for q in questions} - set(world.articles))
    if missing:
        print("bench: the bank names %d slug(s) this docroot does not serve: %s"
              % (len(missing), ", ".join(missing)), file=sys.stderr)
        return 2

    def run_one(item: dict) -> dict:
        return dict(ask(settings, world.get(item["slug"]), item["q"],
                        vocabs, copy, abstain_vocab, world),
                    id=item["id"], kind=item["kind"])

    warmed: list[str] = []

    def warm(slug: str) -> None:
        if warm_page(settings, world.get(slug)):
            warmed.append(slug)

    started = time.time()
    rows = walk(args.arm, questions, concurrency=args.concurrency,
                run_one=run_one, warm=warm)

    on_page = [r for r in rows if r["kind"] == "on-page"]
    off_page = [r for r in rows if r["kind"] == "off-page"]
    # GATE (a): ANSWERED — and since v0.17.0 a PARTIAL release is one of them.
    # `partial` is set by `ask` only when `gates.partial_release` allowed it,
    # which requires at least one kept sentence carrying a cite this page
    # admits AND the kept text passing every gate on its own. So the clause the
    # spec names — "≥ 1 kept cited sentence" — is already proven by the flag,
    # and re-deriving it here would be a second opinion about one decision.
    #
    # `figures_verbatim` and `cites_valid` still read the DRAFT, deliberately:
    # they are facts about what the model wrote, and a partial row's draft did
    # carry a figure the article does not. The row that says a reader was
    # answered and the row that says the model invented a figure are both true
    # and are both worth keeping.
    answered = [r for r in on_page
                if (r.get("ok") and r.get("figures_verbatim")
                    and r.get("cites_valid")) or r.get("partial")]
    # GATE (d): the authored abstain REACHED THE READER. `abstained` is set by
    # `ask` when a gate fired or when the normaliser recognised the seat
    # refusing in its own words; `is_the_abstain` re-checks the served text
    # character for character, so the flag cannot drift from the string.
    # GATE (d) COUNTS WHAT REACHED THE READER. `abstained` is set by `ask` at
    # the moment it substituted the authored refusal; `is_the_abstain` re-checks
    # the served text against the abstain FOR THE SECTION THAT REFUSAL NAMED, so
    # the flag and the string cannot drift and the check cannot fail for naming
    # a different section than the room did.
    abstained = [r for r in off_page if r.get("abstained")
                 and is_the_abstain(r.get("served", ""), world.get(r["slug"]),
                                    r["q"], copy, world, r.get("abstain_section"))]
    disagreed = [r["id"] for r in off_page
                 if r.get("abstained") and r not in abstained]
    seconds = sorted(r["seconds"] for r in rows if "seconds" in r)
    # One reading per ROW (each row's own median), so a chatty answer does not
    # outvote a short one in the bank's distribution.
    overlaps = sorted(r["cite_overlap_p50"] for r in rows
                      if r.get("cite_overlap_p50") is not None)
    cache_hits, cache_countable = cache_hit_share(rows)

    def pct(p: float) -> float:
        if not seconds:
            return float("nan")
        return statistics.quantiles(seconds, n=100)[int(p) - 1] if len(seconds) > 1 \
            else seconds[0]

    report = {
        # 1.1 — THE MEASUREMENTS MOVED, SO THE SCHEMA MOVES WITH THEM. `figures`,
        # `figures_verbatim` and `non_verbatim_figures` are computed over
        # cite-cut prose from v0.1.3 on, and the warm arm's latencies are taken
        # under a different walk. A stored run #3 report and a run #4 report
        # both claiming "1.0" would be two incomparable files with one label, on
        # the very gate that failed.
        "schema_version": "1.1",
        # WHICH INSTRUMENT TOOK THESE NUMBERS, in the file that carries them.
        "instrument": {
            "figures_read_from": "cite-cut prose (gates.prose_of) since 2026-09-09",
            "warm_arm_walk": "per page, one warm immediately before that page's "
                             "own questions, since 2026-09-09",
            "not_comparable_with": "reports whose schema_version is 1.0",
        },
        "run_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(started)),
        "arm": args.arm, "concurrency": args.concurrency,
        # WHAT `--concurrency` ACTUALLY BOUGHT ON THIS ARM. The warm arm's pool
        # is per page, so it is capped by the largest page's question count; the
        # cold arm's is the flag. Recorded so a c6 warm row cannot be read as a
        # six-readers-at-once measurement when the bank cannot supply six.
        "effective_concurrency": effective_concurrency(
            args.arm, questions, args.concurrency),
        "seat": {"url_host_withheld": True, "model": settings.seat_model,
                 "class": settings.seat_class, "seed": args.seed},
        # RECORDED, NOT GOVERNING — and the difference is worth stating so a
        # later reader does not re-derive it. §3.0's v2/v3 ruling decides what
        # counts as the terminal FIGURE of a composed bullet, which is the pack
        # writer's problem. This bench asks a different question —
        # `figures_all_present`, "is every figure in this answer verbatim in the
        # article" — and that is definition-independent. The setting is stamped
        # because a report that cannot say how its packnorm was configured is a
        # report that cannot be replayed.
        "packnorm_definition_recorded_not_governing": pn.DEFINITION,
        "bank": {"path": os.path.abspath(args.bank), "counts": bank["counts"]},
        "results": {
            "on_page_total": len(on_page),
            "on_page_answered": len(answered),
            # HOW MANY OF THOSE WERE PARTIAL, and how much of the draft the
            # reader got across the whole arm. A run whose answered count rose
            # and whose partial count rose with it moved for a different reason
            # than one whose clean count did.
            "on_page_partial": sum(1 for r in on_page if r.get("partial")),
            "off_page_partial": sum(1 for r in off_page if r.get("partial")),
            "sentences_kept": sum(r.get("sentences_kept") or 0 for r in rows),
            "sentences_written": sum(r.get("sentences_written") or 0
                                     for r in rows),
            "non_verbatim_figures": sum(
                1 for r in on_page if r.get("figures_verbatim") is False),
            "uncited": sum(1 for r in on_page if not r.get("cites_valid")),
            "off_page_total": len(off_page),
            "off_page_abstained": len(abstained),
            # WHICH ROAD each abstain took, because they are different facts: a
            # gate firing means the model produced something the page could not
            # support, and the normaliser firing means the model refused
            # correctly and this room said so in its own words.
            "off_page_abstained_by_gate": sum(
                1 for r in abstained if r.get("abstain_source") == "gate"),
            "off_page_abstained_by_model": sum(
                1 for r in abstained if r.get("abstain_source") == "model"),
            # A ROW THAT SAYS IT ABSTAINED AND WHOSE TEXT DISAGREES is the
            # instrument breaking, not the seat: it is counted separately and
            # named, because a silent difference between the flag and the string
            # is what made the first run print 9 where the rows said 21.
            "off_page_flagged_but_not_verbatim": disagreed,
            "on_page_abstained": sum(1 for r in on_page if r.get("abstained")),
            # The off-page answers that passed every gate: the model wrote a
            # CITED sentence about a question its page does not answer. Counted,
            # never judged — see `is_the_abstain`.
            "off_page_cited": sum(1 for r in off_page if r.get("ok")),
            "redaction_hits": sum(r.get("redaction_hits", 0) for r in rows),
            # §6'S READING, AND THE FLOOR IS NOT PRE-REGISTERED FROM IT YET.
            # Printed so a floor can be chosen from a measurement — never the
            # other way round, and never a threshold table for a rung nothing
            # has measured. `cite_overlap_rows` is how many rows carried a
            # reading at all, because a median over three sentences and a median
            # over sixty are different claims.
            "cite_overlap_p50": statistics.median(overlaps) if overlaps else None,
            "cite_overlap_min": min(overlaps) if overlaps else None,
            "cite_overlap_max": max(overlaps) if overlaps else None,
            "cite_overlap_rows": len(overlaps),
            "seat_errors": sum(1 for r in rows if r.get("error")),
            "p50_seconds": round(statistics.median(seconds), 3) if seconds else None,
            "p95_seconds": round(pct(95), 3) if seconds else None,
            # THE WARM ARM'S OWN RECEIPT. A latency ceiling with no cache
            # reading beside it cannot tell a slow seat from an evicted prefix,
            # and telling those two apart is the whole of what the third run
            # cost. Null on the cold arm, and null-denominator whenever the seat
            # does not report `cached_tokens` at all.
            "pages_warmed": len(warmed) if args.arm == "warm" else None,
            "cache_hits": cache_hits if args.arm == "warm" else None,
            "cache_hit_of": cache_countable if args.arm == "warm" else None,
        },
        "rows": rows,
    }

    gate = bank["gates"]
    # NO VERDICT ON `cite_overlap` — §6 is a measurement this release, and a
    # verdict here would be a floor nobody registered.
    res = report["results"]
    verdicts = {
        "on_page_answered": res["on_page_answered"] >= gate["on_page_answered"]["floor"],
        "non_verbatim_figures": res["non_verbatim_figures"] == 0,
        "off_page_abstained": res["off_page_abstained"] >= gate["off_page_abstained"]["floor"],
        "redaction_hits": res["redaction_hits"] == 0,
    }
    if args.arm == "warm":
        verdicts["p50_warm_seconds"] = (res["p50_seconds"] or 99) < gate["p50_warm_seconds"]["ceiling"]
        verdicts["p95_warm_seconds"] = (res["p95_seconds"] or 99) < gate["p95_warm_seconds"]["ceiling"]
    report["verdicts"] = verdicts
    report["passed"] = all(verdicts.values())

    if args.out:
        with open(args.out, "w", encoding="utf-8") as fh:
            json.dump(report, fh, indent=1, ensure_ascii=False)

    effective = report["effective_concurrency"]
    print("bench · %s arm · concurrency %d%s · seed %r · packnorm %s"
          % (args.arm, args.concurrency,
             "" if effective == args.concurrency
             else " (effective %d — a page is the unit of parallelism on this "
                  "arm and no page carries more questions)" % effective,
             args.seed, pn.DEFINITION))
    print("  on-page answered   : %d / %d   (floor %d)"
          % (res["on_page_answered"], res["on_page_total"], gate["on_page_answered"]["floor"]))
    print("  non-verbatim figures: %d          (ceiling 0)" % res["non_verbatim_figures"])
    print("  off-page abstained : %d / %d   (floor %d)"
          % (res["off_page_abstained"], res["off_page_total"],
             gate["off_page_abstained"]["floor"]))
    print("       of which: %d by a gate, %d by the seat's own words, %d neither"
          % (res["off_page_abstained_by_gate"], res["off_page_abstained_by_model"],
             res["off_page_total"] - res["off_page_abstained"]))
    if res["off_page_flagged_but_not_verbatim"]:
        print("       FLAGGED BUT NOT VERBATIM: %s — the flag and the served "
              "text disagree, which is the instrument and not the seat"
              % ", ".join(res["off_page_flagged_but_not_verbatim"]))
    print("  off-page cited     : %d          (answers that passed every gate on a"
          " page that does not say — counted, not judged)" % res["off_page_cited"])
    print("  redaction hits     : %d          (ceiling 0)" % res["redaction_hits"])
    print("  seat errors        : %d" % res["seat_errors"])
    print("  p50 / p95 seconds  : %s / %s" % (res["p50_seconds"], res["p95_seconds"]))
    if args.arm == "warm":
        print("  cache-hit share    : %d / %d rows read >= half their prefix from "
              "cache   (%d page(s) warmed, one immediately before its own "
              "questions)"
              % (res["cache_hits"], res["cache_hit_of"], res["pages_warmed"]))
        if not res["cache_hit_of"]:
            print("       the seat reported no cached_tokens on any row — the unit is")
            print("       running without --enable-prompt-tokens-details (SPEC §2), so")
            print("       this arm cannot say whether its prefixes were hit at all.")
    print("  VERDICT            : %s" % ("PASS" if report["passed"] else "FAIL"))
    if not report["passed"]:
        print("  failed gates       : %s"
              % ", ".join(k for k, v in verdicts.items() if not v))
        print("  A FAILED GATE SHIPS FREE TEXT DARK: set FREE_TEXT_ENABLED=false in the")
        print("  env file and restart. Chips and sections keep working; no room offers")
        print("  the box. A rung rejected by its own gate gets no threshold table.")
    return 0 if report["passed"] else 1


if __name__ == "__main__":
    raise SystemExit(main())