"""The retrieval gate for `okf consume` (capability loop, step 3).

One command, one exit code. For a frozen question set against a bundle it
asks: of N measurement units, how many does the payload carry the fasit for --
and for every miss, exactly ONE class with a reason, so that 0 of them are
unaccounted. It then asks the thing the payload does not do at all: say so
when it does not know.

WRITTEN RED, before any capability. Nothing in this module changes the
ranking, the fusion, the tokenisation or the cut; it only measures them. The
capability order is PM's to place after the rows here have been read.

WHY A ROW CAN BE RED WITHOUT A DEFECT IN THIS FILE. Rows 3, 4, 5, 8 and 9 are
red on the shipped code as it stands: the withheld label names the quota where
the truth is the rank (13 of 25 misses, measured 2026-09-17), the payload
carries no reading a consumer can act on when the bundle does not cover the
question (1 of 5 controls), no hold-out set has been registered, the real sets
live outside this repository, and the K2 gold set does not exist anywhere.

THE FASIT IS AN INPUT, NEVER A CONSTANT HERE -- `tools/okf_consume_measure.py`
states the rule and this module inherits it. This repository is PUBLIC: a gold
set names documents in a consumer's corpus, so a real set arrives as a path
plus an expected sha256 and is never committed. What IS committed is the
synthetic corpus below and the four synthetic sets beside it, whose subject
matter is invented for this gate and names no real document.

GRANULARITY, STATED BECAUSE TWO FORMS ARE IN CIRCULATION. A unit here is one
FASIT ENTRY: a (concept, citation) pair. A question carrying three fasit
entries is three units, and `k of N` over units is never summed with `k of N`
over questions -- both are reported, per set, and the difference is printed.
The rule is the strictest of the three real sets' own (the wiki set's
`hit_rule`, verbatim in its file: an excerpt whose source is the fasit's
document AND whose text carries the fasit's quote).

THE JUDGE OPENS THE BUNDLE (row 6). Every hit is confirmed against the
concept file on disk: the fasit's citation must be IN the bundle before a miss
can be blamed on the ranking, and a delivered excerpt whose text disagrees with
the concept file is not a hit. No number `okf consume` reports about itself is
trusted, and no file a capability session can edit decides a verdict.
"""

from __future__ import annotations

import argparse
import contextlib
import hashlib
import json
import sys
import tempfile
from collections.abc import Callable, Iterator, Mapping, Sequence
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

TOOLS = Path(__file__).resolve().parent
REPO = TOOLS.parent
if str(REPO / "src") not in sys.path:
    sys.path.insert(0, str(REPO / "src"))

from llm_ingestion_okf import consume  # noqa: E402

# The two title forms a fasit can be met in, imported rather than written a
# second time: `okf quality --fasit` decides a boundary with exactly these,
# and two implementations of one rule are two rules with one name.
from llm_ingestion_okf.quality import (  # noqa: E402
    _enclosing_directory,
    _split_numbering,
    normalise_title,
)

FIXTURES = REPO / "tests" / "fixtures" / "retrieval"

#: The hold-out registration. Absent today, which is what makes row 5 red; the
#: row reads a path so a test can drive both directions without editing code.
HOLDOUT_REGISTRATION = FIXTURES / "holdout-registration.json"

GREEN = "GREEN"
RED = "RED"
NOT_RUN = "NOT RUN"

#: Every miss lands in exactly one of these. (c) is decided by the quota-off
#: run and never by the label the payload prints, which is what row 3 measures.
CLASSES: tuple[tuple[str, str], ...] = (
    ("a", "the fasit is not in the bundle"),
    ("b", "ranked below k"),
    ("c", "inside k, cut by quota or budget"),
    ("d", "retrieved, the citation is not in the payload"),
    ("e", "other, with the rule named"),
)

#: Row 7's bar: the share of mechanical mutations of the ranking/cut path that
#: this gate must fell.
MUTANT_BAR = 0.90

#: K2's denominator, from `~/.claude/docs/okf-utfallsgrunnlag.md`. The bundles
#: are on this machine; the gold set is nowhere, by design.
K2_QUESTIONS = 6


class GateUsage(Exception):
    """Wrong input: exit 2, never a quiet row."""


# --- the synthetic corpus -----------------------------------------------------
#
# Generated rather than committed as 60 files: the bundle is then a function of
# this code, and a reader can see in one place why a gold concept ranks where
# it does. Invented subject matter (a mountain club's papers), so nothing here
# is corpus-near for any consumer.

_FRONTMATTER = (
    "---\ntype: reference\ntitle: {title}\nsource_file: {source_file}\n"
    "source_sha256: {digest}\ningested_at: 2026-09-01T00:00:00Z\n"
    "adjudication: proposed\nbundle_id: {bundle_id}\n"
    "{description}"
    "verified: [{{ by: process:okf-check, at: 2026-09-01T00:00:00Z }}]\n---\n\n"
)


@dataclass(frozen=True)
class ConceptSpec:
    slug: str
    title: str
    body: str
    #: Written into the frontmatter, never into the body. A fasit quoting THIS
    #: is in the bundle and cannot reach a payload: class (d).
    description: str = ""
    repeat: int = 1


@dataclass(frozen=True)
class DocumentSpec:
    directory: str
    source_file: str
    concepts: tuple[ConceptSpec, ...]


@dataclass(frozen=True)
class BundleSpec:
    bundle_id: str
    documents: tuple[DocumentSpec, ...]


def build_bundle(root: Path, spec: BundleSpec) -> Path:
    """Write `spec` as an OKF bundle under `root`. Deterministic bytes."""
    root.mkdir(parents=True, exist_ok=True)
    root_entries = [
        f"---\nokf_version: 0.2\nbundle_id: {spec.bundle_id}\n---\n\n",
    ]
    for document in spec.documents:
        directory = root / document.directory
        directory.mkdir(parents=True, exist_ok=True)
        root_entries.append(f"- [{document.directory} (index)]({document.directory}/index.md)\n")
        entries = []
        for concept in document.concepts:
            # The em dash is load-bearing: the profile's `entry_pattern`
            # reads facets only after ` \u2014 `, and a hyphen here makes the
            # whole line curated prose -- an index walk that reaches no
            # concept at all, which is how this fixture first came out.
            entries.append(
                f"- [{concept.title}]({concept.slug}.md) \u2014 adjudication: proposed\n"
            )
            digest = hashlib.sha256(document.source_file.encode("utf-8")).hexdigest()
            description = f"description: {concept.description}\n" if concept.description else ""
            (directory / f"{concept.slug}.md").write_text(
                _FRONTMATTER.format(
                    title=concept.title,
                    source_file=document.source_file,
                    digest=digest,
                    bundle_id=spec.bundle_id,
                    description=description,
                )
                + f"## {concept.title}\n\n"
                + "".join(f"{concept.body}\n" for _ in range(concept.repeat)),
                encoding="utf-8",
            )
        (directory / "index.md").write_text("".join(entries), encoding="utf-8")
    (root / "index.md").write_text("".join(root_entries), encoding="utf-8")
    return root


def _filler(directory: str, source_file: str, subject: str, count: int) -> tuple[ConceptSpec, ...]:
    """Decoys: they carry the corpus's common words and none of a fasit's."""
    return tuple(
        ConceptSpec(
            slug=f"{subject}-{number:02d}",
            title=f"Notat {number:02d} om {subject}",
            body=(
                f"Notatet gjelder {subject} i klubben og gjennomgaas av styret. "
                f"Styret foerer kontroll med arbeidet hvert aar."
            ),
        )
        for number in range(1, count + 1)
    )


POSITIVE = BundleSpec(
    "retrieval-positive",
    (
        DocumentSpec(
            "haandbok",
            "haandbok.md",
            (
                ConceptSpec(
                    "vinterberedskap",
                    "Vinterberedskap paa hytta",
                    "Vinterberedskapen kontrolleres innen 1. november hvert aar, "
                    "og avviket foeres i hytteboka.",
                ),
                ConceptSpec(
                    "noekkelkvittering",
                    "Noekkelkvittering",
                    "Hver noekkel kvitteres ut mot signatur i noekkelboka, og "
                    "noekkelen leveres tilbake ved endt sesong.",
                ),
                ConceptSpec(
                    "brannvarsling",
                    "Brannvarsling",
                    "Roekvarsleren testes ved hvert besoek, og batteriet byttes en gang i aaret.",
                ),
                *_filler("haandbok", "haandbok.md", "vedlikehold", 6),
            ),
        ),
        DocumentSpec(
            "referat",
            "referat.md",
            (
                ConceptSpec(
                    "aarsmote-kontingent",
                    "Vedtak om kontingent",
                    "Aarsmoetet vedtok at kontingenten settes til 480 kroner for voksne medlemmer.",
                ),
                ConceptSpec(
                    "aarsmote-valg",
                    "Valg av revisor",
                    "Revisoren velges for to aar av gangen, og gjenvalg er tillatt en gang.",
                ),
                *_filler("referat", "referat.md", "moeteplan", 5),
            ),
        ),
        DocumentSpec(
            "skjema",
            "skjema.md",
            (
                ConceptSpec(
                    "utstyrsliste",
                    "Utstyrsliste for vinteropphold",
                    "Utstyrslisten skal foelge soeknaden om vinteropphold og "
                    "signeres av turlederen.",
                ),
                ConceptSpec(
                    "avviksskjema",
                    "Avviksskjema",
                    "Avvik meldes paa eget skjema senest tre dager etter turen.",
                    description=("Avviksskjemaet arkiveres i fem aar hos sekretaeren."),
                ),
                *_filler("skjema", "skjema.md", "soeknader", 5),
            ),
        ),
    ),
)

#: The fasit is present, carries the citation and is reachable -- and the
#: question reaches it with ONE common token while fifteen decoys answer five.
#: Nothing is missing from this bundle; the ranking does not get there.
MISS = BundleSpec(
    "retrieval-miss",
    (
        DocumentSpec(
            "vedtekter",
            "vedtekter.md",
            (
                ConceptSpec(
                    "flertallskrav",
                    "Naar saken er avgjort",
                    "To tredjedeler av de fremmoette medlemmer maa si ja.",
                ),
                *tuple(
                    ConceptSpec(
                        slug=f"stemmegivning-{number:02d}",
                        title=f"Stemmegivning i sak {number:02d}",
                        body=(
                            "Saken avgjoeres ved votering blant medlemmer som har "
                            "betalt kontingent. Medlemmer kan stemme skriftlig for "
                            "eller mot en endring av vedtektene."
                        ),
                    )
                    for number in range(1, 16)
                ),
            ),
        ),
        DocumentSpec(
            "protokoll",
            "protokoll.md",
            _filler("protokoll", "protokoll.md", "protokollen", 6),
        ),
    ),
)

#: ONE source document, which is what makes the withheld label lie: every
#: concept past the first two shares a `source_file` with the ones delivered,
#: so the quota claims a concept the rank had already lost.
SINGLE_SOURCE = BundleSpec(
    "retrieval-single-source",
    (
        DocumentSpec(
            "aarsberetning",
            "aarsberetning.md",
            (
                ConceptSpec(
                    "loypekjoring",
                    "Naar loypene er klare",
                    "Kjoeringen starter naar snoedybden passerer tretti centimeter.",
                ),
                *tuple(
                    ConceptSpec(
                        slug=f"punkt-{number:02d}",
                        title=f"Punkt {number:02d} i beretningen",
                        body=(
                            "Punktet gjelder driften av klubben. Styret foerer "
                            "kontroll med snoedybden og dugnaden gjennom aaret."
                        ),
                    )
                    for number in range(1, 18)
                ),
            ),
        ),
    ),
)

#: Two concepts a question can only reach through ONE of the fusion's
#: partitions: the identifier lookup and the covered title. Everything else in
#: the bundle answers the question's other words three times over, so a
#: partition that stops firing takes its concept out of k.
LOOKUP = BundleSpec(
    "retrieval-lookup",
    (
        DocumentSpec(
            "rutiner",
            "rutiner.md",
            (
                ConceptSpec(
                    "vakthold-4-2",
                    "Vakthold 4.2",
                    "Vakten gaar fra fredag til soendag.",
                ),
                ConceptSpec(
                    "noekkelrutine",
                    "Noekkelrutine",
                    "Rutinen gjelder ved skifte av laas.",
                ),
                *tuple(
                    ConceptSpec(
                        slug=f"notat-{number:02d}",
                        title=f"Notat {number:02d} om ettersyn",
                        body=(
                            "Kontrollen av hytta foeres i skjema. Hvert punkt i "
                            "kontrollen kvitteres av den som gaar runden."
                        ),
                    )
                    for number in range(1, 13)
                ),
            ),
        ),
    ),
)

#: One document floods the question and another holds the answer: the defect
#: `--source-quota` was shipped for. Without the quota the dominant document
#: takes every delivered place and the fasit is not delivered at all.
QUOTA = BundleSpec(
    "retrieval-quota",
    (
        DocumentSpec(
            "oversikt",
            "oversikt.md",
            tuple(
                ConceptSpec(
                    slug=f"oversikt-{number:02d}",
                    title=f"Oversikt {number:02d} over dugnaden",
                    body=("Oversikten viser dugnaden og kontrollen av broennen gjennom sesongen."),
                )
                for number in range(1, 11)
            ),
        ),
        DocumentSpec(
            "svar",
            "svar.md",
            (
                ConceptSpec(
                    "broennproeve",
                    "Proeve av broennen",
                    "Broennen proevetas i juni, og analysen arkiveres av styret.",
                ),
            ),
        ),
    ),
)

#: One fasit concept far larger than the rest, so it can be ranked first and
#: still lose the pack: the shape class (c) is measured on.
BUDGET = BundleSpec(
    "retrieval-budget",
    (
        DocumentSpec(
            "tabell",
            "tabell.md",
            (
                ConceptSpec(
                    "dugnadstabell",
                    "Dugnadstabell",
                    "Dugnadstimene foeres i tabellen rad for rad av dugnadslederen.",
                    repeat=120,
                ),
            ),
        ),
        DocumentSpec(
            "notater",
            "notater.md",
            tuple(
                ConceptSpec(
                    slug=f"dugnad-{number:02d}",
                    title=f"Dugnadsnotat {number:02d}",
                    body="Notatet gjelder dugnadstimene for en enkelt helg.",
                )
                for number in range(1, 13)
            ),
        ),
    ),
)

SPECS: Mapping[str, BundleSpec] = {
    "positive": POSITIVE,
    "miss": MISS,
    "single-source": SINGLE_SOURCE,
    "budget": BUDGET,
    "lookup": LOOKUP,
    "quota": QUOTA,
}


def synthetic_bundles(root: Path) -> dict[str, Path]:
    """Every synthetic bundle, written once and reused by every row."""
    return {name: build_bundle(root / name, spec) for name, spec in SPECS.items()}


# --- the sets -----------------------------------------------------------------


#: How a fasit entry names the concept that answers it. Four, because the
#: three real sets name three different things and a gate that could read only
#: one of them would report two of the three as zero: the wiki set names a
#: source document plus a quote, `vegnormal` names a requirement number,
#: `R761-sk2` names an STS section title. The synthetic sets here name the
#: concept directly, which is the strictest form and the only one with no
#: resolution step between the set and the bundle.
MATCHERS = ("concept", "source_file", "req_number", "title")


@dataclass(frozen=True)
class Fasit:
    """One measurement unit: what a consumer must be handed, and how it is named.

    `quote` may be empty. A set whose fasit names no quote is measured at
    CONCEPT granularity -- delivery alone -- and every row that reports such a
    set says so, because "the right concept arrived" and "the cited sentence
    arrived" are different claims and must never be summed silently.
    """

    by: str
    value: str
    quote: str = ""

    def __post_init__(self) -> None:
        if self.by not in MATCHERS:
            raise GateUsage(f"unknown fasit matcher `{self.by}`; one of {MATCHERS}")


@dataclass(frozen=True)
class Question:
    id: str
    question: str
    fasit: tuple[Fasit, ...]
    expect_class: str | None = None
    k: int = consume.DEFAULT_K
    limit: int = consume.DEFAULT_LIMIT
    #: Set-level bundle override: one set can span several bundles (the
    #: `vegnormal` set names a road standard per question).
    bundle: str = ""


@dataclass(frozen=True)
class Control:
    id: str
    question: str
    kind: str
    #: True when the bundle DOES answer it. Such a control must come back
    #: unmarked, or the marking says nothing.
    covered: bool = False


@dataclass(frozen=True)
class QuestionSet:
    set_id: str
    bundle: str
    path: Path
    sha256: str
    questions: tuple[Question, ...]
    controls: tuple[Control, ...]

    @property
    def units(self) -> int:
        return sum(len(question.fasit) for question in self.questions)

    @property
    def quoted(self) -> bool:
        """Every unit names a citation, so the set is measured at citation
        granularity rather than at concept granularity."""
        return all(fasit.quote for question in self.questions for fasit in question.fasit)


def sha256_of(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def load_set(path: Path, expected_sha256: str) -> QuestionSet:
    """A frozen set, refused unless its bytes are the bytes that were pinned.

    A sha that does not match is exit 2 and never a quiet continuation: a
    measurement over a set someone edited is a different measurement wearing
    the old one's number.
    """
    try:
        raw = path.read_bytes()
    except OSError as error:
        raise GateUsage(f"cannot read the question set {path}: {error}") from error
    measured = hashlib.sha256(raw).hexdigest()
    if measured != expected_sha256:
        raise GateUsage(
            f"{path}: expected sha256 {expected_sha256}, measured {measured}; "
            "refusing to measure a set that is not the set that was pinned"
        )
    try:
        spec = json.loads(raw.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as error:
        raise GateUsage(f"{path}: not readable as JSON: {error}") from error
    try:
        return QuestionSet(
            set_id=str(spec["set_id"]),
            bundle=str(spec["bundle"]),
            path=path,
            sha256=measured,
            questions=tuple(_question(entry) for entry in spec.get("questions", [])),
            controls=tuple(
                Control(
                    id=str(entry["id"]),
                    question=str(entry["question"]),
                    kind=str(entry["kind"]),
                    covered=bool(entry.get("covered", False)),
                )
                for entry in spec.get("controls", [])
            ),
        )
    except (KeyError, TypeError, ValueError) as error:
        raise GateUsage(f"{path}: not a question set this gate can read: {error}") from error


def _question(entry: Mapping[str, Any]) -> Question:
    return Question(
        id=str(entry["id"]),
        question=str(entry["question"]),
        fasit=tuple(
            Fasit(by=str(item["by"]), value=str(item["value"]), quote=str(item.get("quote", "")))
            for item in entry["fasit"]
        ),
        expect_class=str(entry["expect_class"]) if entry.get("expect_class") else None,
        k=int(entry.get("k", consume.DEFAULT_K)),
        limit=int(entry.get("limit", consume.DEFAULT_LIMIT)),
        bundle=str(entry.get("bundle", "")),
    )


#: The synthetic sets, pinned by sha256. Editing one without moving its pin is
#: exit 2, which is the point: these bytes are the denominator every row below
#: counts against. A test asserts the pins match the files, so an edit fails in
#: the suite as well as in the gate.
SYNTHETIC_SETS: dict[str, str] = {
    "set-positive.json": "88b4d96185b6f905c9f3b29b3cdc9caf1cac8ec6d373fbb1f656829c8b9f4e5a",
    "set-miss.json": "3a75da8fe7177ebfc555ee3c8d1af94adcd927925b4f5dfccd0cb76e3fa5ae8c",
    "set-classes.json": "65c3eb272dcbb972abc2072c7a334490f016a8259b7b1a33d780aa497ec1393d",
    "set-signals.json": "17d83f305a10af8dd2a45b72943c1285704d23ed074ecb549c05e5fac3e137e9",
    "set-controls.json": "c2894656326e5a69ec7063fdc280e763910d20b4cbc2c0124f639283864ce506",
}


# --- reading the bundle -------------------------------------------------------


def _flat(text: str) -> str:
    """The comparison form: casefolded, whitespace collapsed.

    NO unicode normalisation, deliberately. `delivered_text` NFC-normalises
    what it hands over, so folding here too would hide a real way for a
    citation to be in the bundle and not in the payload.
    """
    return " ".join(text.split()).casefold()


def concept_path(bundle: Path, concept_id: str) -> Path:
    return bundle / f"{concept_id}{consume.CONCEPT_SUFFIX}"


@dataclass(frozen=True)
class ConceptView:
    """One concept as the JUDGE reads it: off the disk, never off a payload."""

    concept_id: str
    source_file: str
    req_number: str
    title: str
    body: str
    whole: str


class BundleIndex:
    """Every concept of a bundle, read once, keyed the four ways a fasit names one.

    The title keys are `quality.normalise_title` and the (directory, residual
    title) pair from `quality.DeclaredBoundary.pair_key` -- imported rather
    than re-implemented, so a number this gate produces and a number
    `okf quality --fasit` produces cannot drift apart.
    """

    def __init__(self, root: Path) -> None:
        self.root = root
        root_bundle_id = consume.root_bundle_id_of(root)
        self.concepts: dict[str, ConceptView] = {}
        for concept_id in consume.enumerate_concepts(root):
            path = concept_path(root, concept_id)
            concept = consume.read_concept(path, bundle_root=root, root_bundle_id=root_bundle_id)
            self.concepts[concept_id] = ConceptView(
                concept_id=concept_id,
                source_file=str(concept.frontmatter.get("source_file", "")),
                req_number=str(concept.frontmatter.get("req_number", "")),
                title=concept.title,
                body=consume.delivered_text(concept.body),
                whole=path.read_text(encoding="utf-8"),
            )

    def candidates(self, fasit: Fasit) -> tuple[str, ...]:
        """Every concept in this bundle the fasit entry could be met by."""
        if fasit.by == "concept":
            return tuple(
                concept_id
                for concept_id in self.concepts
                if concept_id == fasit.value or concept_id.startswith(f"{fasit.value}/")
            )
        if fasit.by == "source_file":
            return tuple(
                concept_id
                for concept_id, view in self.concepts.items()
                if view.source_file == fasit.value
            )
        if fasit.by == "req_number":
            return tuple(
                concept_id
                for concept_id, view in self.concepts.items()
                if view.req_number == fasit.value
            )
        norm = normalise_title(fasit.value)
        number, rest = _split_numbering(fasit.value)
        pair = (number.replace(".", "-"), normalise_title(rest))
        return tuple(
            concept_id
            for concept_id, view in self.concepts.items()
            if normalise_title(view.title) == norm
            or (_enclosing_directory(concept_id), normalise_title(view.title)) == pair
        )


_INDEXES: dict[Path, BundleIndex] = {}


def bundle_index(root: Path) -> BundleIndex:
    """Read once per bundle per process: the real bundles hold thousands of
    concepts and every row asks the same questions of them."""
    if root not in _INDEXES:
        _INDEXES[root] = BundleIndex(root)
    return _INDEXES[root]


# --- measuring one fasit entry ------------------------------------------------


@dataclass
class Unit:
    """One fasit entry, measured."""

    question_id: str
    question: str
    named: str
    quote: str
    hit: bool
    rank: int | None
    klass: str | None
    label_default: str | None
    truth: str | None
    confirmed: bool | None
    detail: str = ""
    #: The question declares the class it forces, so it is a fixture for row 2
    #: and never a row-1 unit: counting a miss built to miss as a miss would
    #: make row 1 unable to be green whatever the ranker does.
    forced: bool = False


def _withheld_rules(payload: Mapping[str, object]) -> dict[str, str]:
    entries = payload.get("withheld")
    assert isinstance(entries, list)
    return {
        str(entry["concept_id"]): str(entry["rule"]) for entry in entries if isinstance(entry, dict)
    }


def _delivered(payload: Mapping[str, object]) -> dict[str, Mapping[str, object]]:
    entries = payload.get("excerpts")
    assert isinstance(entries, list)
    return {str(entry["concept_id"]): entry for entry in entries if isinstance(entry, dict)}


def _carries(text: str, quote: str) -> bool:
    return not quote or _flat(quote) in _flat(text)


def measure_units(bundle: Path, question: Question) -> list[Unit]:
    """One question, measured at the shipped defaults, plus the quota-off run
    that says what the truth of a withheld concept is.

    The second run is not a second opinion about the ranking -- it is the same
    ranking cut without the quota, which is the only way to tell a concept the
    quota took from a concept that never reached k. Row 3 compares the label
    the first run printed against it.
    """
    index = bundle_index(bundle)
    default = consume.build_payload(
        bundle, question=question.question, k=question.k, limit=question.limit
    )
    truth_run = consume.build_payload(
        bundle,
        question=question.question,
        k=question.k,
        limit=question.limit,
        source_quota=None,
    )
    delivered = _delivered(default)
    withheld = _withheld_rules(default)
    truth_delivered = _delivered(truth_run)
    truth_withheld = _withheld_rules(truth_run)
    units: list[Unit] = []
    for fasit in question.fasit:
        candidates = index.candidates(fasit)
        # THE BUNDLE'S ANSWER, read off the disk before any payload is opened.
        # A miss is only the ranking's to answer for when the bundle holds the
        # fasit in the first place.
        holding = tuple(
            concept_id
            for concept_id in candidates
            if _carries(index.concepts[concept_id].whole, fasit.quote)
        )
        in_body = tuple(
            concept_id
            for concept_id in holding
            if _carries(index.concepts[concept_id].body, fasit.quote)
        )
        hit_ids = [
            concept_id
            for concept_id in holding
            if concept_id in delivered
            and _carries(str(delivered[concept_id].get("text", "")), fasit.quote)
        ]
        # The payload SAYS it delivered this; the bundle says what it is.
        confirmed: bool | None = None
        for concept_id in hit_ids:
            confirmed = _flat(str(delivered[concept_id].get("text", ""))) == _flat(
                index.concepts[concept_id].body
            )
            if confirmed:
                break
        # The concept this entry is really about: the one the bundle holds the
        # citation in, else the first candidate, in a stable order.
        target = (in_body or holding or tuple(sorted(candidates)) or ("",))[0]
        rank = None
        excerpt = delivered.get(target)
        if excerpt is not None and isinstance(excerpt.get("rank"), int):
            rank = int(excerpt["rank"])
        label = withheld.get(target) if target else None
        truth = (
            "delivered"
            if target in truth_delivered
            else (truth_withheld.get(target) if target else None)
        )
        hit = bool(hit_ids) and bool(confirmed)
        klass: str | None = None
        detail = ""
        if hit:
            pass
        elif not holding:
            klass = "a"
            detail = (
                f"{len(candidates)} candidate concept(s), none carrying the fasit"
                if candidates
                else "no concept in the bundle answers to this name"
            )
        elif hit_ids and not confirmed:
            klass, detail = "e", "the delivered text is not the bundle's bytes"
        elif target in delivered:
            klass = "d"
            detail = "delivered without the citation" + (
                "; the citation is in the bundle but not in the body" if not in_body else ""
            )
        elif truth == "delivered" or (truth or "").startswith("over_budget"):
            klass, detail = "c", f"the quota-off run says {truth}"
        elif truth == "below_k":
            klass, detail = "b", "below k with the quota off as well"
        elif truth is None:
            klass, detail = "e", "in neither the delivered nor the withheld list"
        else:
            klass, detail = "e", f"withheld as {truth}"
        units.append(
            Unit(
                question_id=question.id,
                question=question.question,
                named=f"{fasit.by}:{fasit.value}",
                quote=fasit.quote,
                hit=hit,
                rank=rank,
                klass=klass,
                label_default=None if hit or target in delivered else label,
                truth=truth,
                confirmed=confirmed,
                detail=detail,
                forced=bool(question.expect_class),
            )
        )
    return units


# --- rows ---------------------------------------------------------------------


@dataclass
class Row:
    number: int
    name: str
    k: int
    m: int
    status: str
    reason: str
    details: list[str] = field(default_factory=list)

    @property
    def fails(self) -> bool:
        """Every row fails the gate. A row that did not run fails too: it is
        the shape a green verdict hides behind."""
        return self.status != GREEN

    def to_json(self) -> dict[str, Any]:
        return {
            "row": self.number,
            "name": self.name,
            "k": self.k,
            "m": self.m,
            "status": self.status,
            "reason": self.reason,
            "details": self.details,
        }


def _row(number: int, name: str, k: int, m: int, reason: str, details: list[str]) -> Row:
    """`m == 0` is NEVER green: a row with nothing to count did not measure."""
    return Row(number, name, k, m, GREEN if m > 0 and k == m else RED, reason, details)


@dataclass(frozen=True)
class Case:
    """One set measured once, and read by several rows.

    A set may span bundles: `set-classes.json` forces one class per bundle,
    and the real `vegnormal` set names a road standard per question. The
    controls belong to the set's own bundle.
    """

    question_set: QuestionSet
    bundles: Mapping[str, Path]
    units: tuple[Unit, ...]

    @property
    def control_bundle(self) -> Path:
        return self.bundles[self.question_set.bundle]


def measure_case(question_set: QuestionSet, bundles: Mapping[str, Path]) -> Case:
    units: list[Unit] = []
    for question in question_set.questions:
        key = question.bundle or question_set.bundle
        if key not in bundles:
            raise GateUsage(f"{question_set.set_id}: no bundle named `{key}`")
        units.extend(measure_units(bundles[key], question))
    return Case(question_set, bundles, tuple(units))


def row_one(cases: Sequence[Case]) -> Row:
    """hit@payload per set, at UNIT granularity, with the question count beside
    it and never summed into it."""
    details: list[str] = []
    hits = units = 0
    for case in cases:
        by_question: dict[str, list[Unit]] = {}
        for unit in case.units:
            by_question.setdefault(unit.question_id, []).append(unit)
        counted = [unit for unit in case.units if not unit.forced]
        if not counted:
            details.append(
                f"{case.question_set.set_id}: every question declares the class it "
                "forces, so this set is row 2's fixture and not a row-1 unit"
            )
            continue
        case_hits = sum(1 for unit in counted if unit.hit)
        answered = sum(
            1
            for group in by_question.values()
            if any(u.hit for u in group) and not all(u.forced for u in group)
        )
        asked = len({unit.question_id for unit in counted})
        hits += case_hits
        units += len(counted)
        details.append(
            f"{case.question_set.set_id}: {case_hits} of {len(counted)} fasit entries "
            f"({'citation' if case.question_set.quoted else 'concept'} granularity) | "
            f"{answered} of {asked} questions | sha256 {case.question_set.sha256[:12]}"
        )
        for unit in counted:
            if not unit.hit:
                details.append(
                    f"    miss {unit.question_id} {unit.named}: "
                    f"class {unit.klass or '-'}, rank {unit.rank if unit.rank else '-'}"
                )
    return _row(
        1,
        "hit@payload, unit granularity (one fasit entry = one unit)",
        hits,
        units,
        "the fasit entry is delivered and its citation is in the excerpt",
        details,
    )


def row_two(cases: Sequence[Case]) -> Row:
    """Every miss has exactly ONE class -- and a miss with none takes the whole
    row to 0, because a classification with a hole is not a classification."""
    misses = [unit for case in cases for unit in case.units if not unit.hit]
    unplaced = [unit for unit in misses if unit.klass is None]
    wrong = [
        (unit, expected)
        for case in cases
        for question in case.question_set.questions
        if question.expect_class
        for unit in case.units
        if unit.question_id == question.id
        and not unit.hit
        and (expected := question.expect_class) != unit.klass
    ]
    counts = {letter: sum(1 for u in misses if u.klass == letter) for letter, _ in CLASSES}
    k = 0 if unplaced else len(misses) - len(wrong)
    details = [
        f"class {letter} ({description}): {counts[letter]}" for letter, description in CLASSES
    ]
    details += [f"    unplaced: {unit.question_id} {unit.named}" for unit in unplaced]
    details += [
        f"    forced class {expected}, measured {unit.klass}: {unit.question_id}"
        for unit, expected in wrong
    ]
    return _row(
        2,
        "every miss carries exactly one class",
        k,
        len(misses),
        "each class forced by its own fixture; an unplaced miss makes this 0 of N",
        details,
    )


def row_three(cases: Sequence[Case]) -> Row:
    """The `rule` the payload prints for a withheld fasit, against what the
    quota-off run says was true of it."""
    withheld = [unit for case in cases for unit in case.units if unit.label_default is not None]
    agreeing = []
    lying = []
    for unit in withheld:
        honest = (
            unit.label_default == "source_quota_exceeded"
            if unit.truth == "delivered"
            else unit.label_default == unit.truth
        )
        (agreeing if honest else lying).append(unit)
    details = [
        f"    {unit.question_id} {unit.named}: payload says "
        f"`{unit.label_default}`, the quota-off run says `{unit.truth}`"
        for unit in lying
    ]
    return _row(
        3,
        "the reason the payload prints for a withheld fasit is true",
        len(agreeing),
        len(withheld),
        "`rule` must say what the same cut without the quota says",
        details,
    )


def marked(payload: Mapping[str, object]) -> bool:
    """Can a consumer READ, from the payload alone, that the bundle does not
    cover the question?

    Today exactly one reading does that: nothing was delivered, and the
    withheld rules say why. Eight excerpts with no score, no confidence and no
    statement is not a marked answer -- it is the same payload an answered
    question gets. When the payload grows a field that says so, this function
    is where it is read, and the row moves with it.
    """
    counts = payload.get("denominators")
    assert isinstance(counts, dict)
    return int(counts["delivered"]) == 0


def row_four(cases: Sequence[Case]) -> Row:
    """It says when it does not know -- and stays quiet when it does."""
    details: list[str] = []
    correct = total = 0
    for case in cases:
        for control in case.question_set.controls:
            payload = consume.build_payload(case.control_bundle, question=control.question)
            counts = payload["denominators"]
            assert isinstance(counts, dict)
            is_marked = marked(payload)
            expected = not control.covered
            total += 1
            if is_marked == expected:
                correct += 1
            else:
                details.append(
                    f"    {control.id} ({control.kind}): "
                    + (
                        "covered by the bundle and MARKED anyway"
                        if control.covered
                        else f"not covered and NOT marked -- {counts['delivered']} "
                        "excerpts, no score, no statement"
                    )
                )
    return _row(
        4,
        "an uncovered question comes back marked, a covered one does not",
        correct,
        total,
        "marked = a reading the consumer can act on (today: nothing delivered)",
        details,
    )


def _display(path: Path) -> str:
    """Repo-relative where it is inside the repo, absolute otherwise: a test
    drives this row from `tmp_path`, and `relative_to` raises there."""
    try:
        return str(path.relative_to(REPO))
    except ValueError:
        return str(path)


def row_five(registration: Path = HOLDOUT_REGISTRATION) -> Row:
    """The hold-out set: written blind, frozen before the first capability
    line, its threshold written before anyone saw the number.

    Report-only without a written threshold is not a protection, so the
    absence of a threshold is red rather than absent.
    """
    if not registration.is_file():
        return Row(
            5,
            "hold-out registered, frozen and pre-registered",
            0,
            1,
            RED,
            f"no registration at {_display(registration)}: no hold-out set exists",
            [
                "    a set written by a session other than the one that changes the "
                "ranking, frozen with sha256 before the first capability line",
                "    its threshold written, with a date, before its number is read",
            ],
        )
    try:
        spec = json.loads(registration.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        return Row(5, "hold-out registered", 0, 1, RED, f"unreadable: {error}")
    checks: list[tuple[str, bool, str]] = []
    threshold = str(spec.get("threshold", ""))
    written_at = str(spec.get("threshold_written_at", ""))
    set_path = Path(str(spec.get("set", "")))
    pinned = str(spec.get("sha256", ""))
    readings = spec.get("readings", [])
    checks.append(("a set is named", bool(str(spec.get("set", ""))), str(set_path)))
    checks.append(("a sha256 is pinned", len(pinned) == 64, pinned[:12]))
    checks.append(("a threshold is written", bool(threshold), threshold))
    checks.append(("the threshold carries a date", bool(written_at), written_at))
    checks.append(
        (
            "written by another session than the ranking change",
            bool(str(spec.get("written_by", ""))),
            str(spec.get("written_by", "")),
        )
    )
    checks.append(
        (
            "the pinned bytes are the bytes on disk",
            set_path.is_file() and sha256_of(set_path) == pinned,
            "present" if set_path.is_file() else "absent",
        )
    )
    early = [
        reading
        for reading in readings
        if isinstance(reading, Mapping) and str(reading.get("at", "")) < written_at
    ]
    checks.append(
        (
            "no reading predates the threshold",
            not early,
            f"{len(early)} reading(s) before {written_at}" if early else "0 early readings",
        )
    )
    passed = sum(1 for _, ok, _ in checks if ok)
    return _row(
        5,
        "hold-out registered, frozen and pre-registered",
        passed,
        len(checks),
        "a number read before its threshold was written is report-only, not a gate",
        [f"    {name}: {'yes' if ok else 'NO'} ({value})" for name, ok, value in checks],
    )


def row_six(cases: Sequence[Case]) -> Row:
    """The judge opened the bundle: every claimed delivery is confirmed against
    the concept file's own bytes."""
    claimed = [unit for case in cases for unit in case.units if unit.confirmed is not None]
    confirmed = [unit for unit in claimed if unit.confirmed]
    details = [
        f"    {unit.question_id} {unit.named}: the delivered text is not the bundle's bytes"
        for unit in claimed
        if not unit.confirmed
    ]
    details.append(
        "    the fasit's presence is read off the concept file before any miss is "
        "blamed on the ranking (class a)"
    )
    return _row(
        6,
        "every delivery confirmed against the bundle, not against the payload",
        len(confirmed),
        len(claimed),
        "no number okf consume reports about itself decides a verdict here",
        details,
    )


# --- row 7: what fells this gate ----------------------------------------------


@dataclass(frozen=True)
class Mutant:
    """One mechanical weakening of the ranking or the cut, and the row that
    should see it. A survivor is printed with that row's number beside it: a
    mutation nothing here catches names a hole in the gate, not a property of
    the ranker."""

    label: str
    expected_row: int
    patch: Callable[[], contextlib.AbstractContextManager[None]]
    #: Why this one may be invisible to a gate that judges DELIVERY. Written
    #: for the two that survive, so a survivor is a statement about the
    #: mechanism and not a shrug.
    note: str = ""


@contextlib.contextmanager
def _patched(**attributes: object) -> Iterator[None]:
    original = {name: getattr(consume, name) for name in attributes}
    try:
        for name, value in attributes.items():
            setattr(consume, name, value)
        yield
    finally:
        for name, value in original.items():
            setattr(consume, name, value)


def _wrap_cut(**overrides: object) -> contextlib.AbstractContextManager[None]:
    original = consume.cut

    def mutant(ranked: Sequence[Any], **kwargs: Any) -> Any:
        return original(ranked, **{**kwargs, **overrides})

    return _patched(cut=mutant)


def _reverse_scores() -> contextlib.AbstractContextManager[None]:
    original = consume.concept_scores

    def mutant(*args: Any, **kwargs: Any) -> Any:
        return tuple(reversed(original(*args, **kwargs)))

    return _patched(concept_scores=mutant)


def _last_k() -> contextlib.AbstractContextManager[None]:
    original = consume.cut

    def mutant(ranked: Sequence[Any], **kwargs: Any) -> Any:
        return original(tuple(reversed(list(ranked))), **kwargs)

    return _patched(cut=mutant)


def _truncate_delivered() -> contextlib.AbstractContextManager[None]:
    original = consume.delivered_text

    def mutant(body: str) -> str:
        return original(body)[:40]

    return _patched(delivered_text=mutant)


def _drop_text_key() -> contextlib.AbstractContextManager[None]:
    original = consume.excerpt_for

    def mutant(concept: Any) -> Any:
        excerpt = original(concept)
        if excerpt is not None:
            excerpt["text"] = ""
        return excerpt

    return _patched(excerpt_for=mutant)


def _equality_only() -> contextlib.AbstractContextManager[None]:
    def mutant(left: str, right: str, *, stems: Any = None) -> bool:
        return left == right

    return _patched(tokens_match=mutant)


MUTANTS: tuple[Mutant, ...] = (
    Mutant("M01 the lookup partition is off", 1, lambda: _patched(lookup_hits=lambda *a: ())),
    Mutant(
        "M02 the title-covered partition is off",
        1,
        lambda: _patched(title_covered_hits=lambda *a: ()),
    ),
    Mutant("M03 k = 1", 1, lambda: _wrap_cut(k=1)),
    Mutant("M04 the ranking is reversed", 1, _reverse_scores),
    Mutant("M05 the source quota is removed", 3, lambda: _wrap_cut(source_quota=None)),
    Mutant("M06 the body signal is dead", 1, lambda: _patched(_overlap=lambda *a, **k: 0)),
    Mutant(
        "M07 the document prior is dead",
        1,
        lambda: _patched(document_scores=lambda *a, **k: {}),
        note=(
            "a question that NAMES its document reaches it through the title-and-id "
            "signal as well, since the prior reads the same id path; a question that "
            "does not can be moved by at most 1/(RRF_K+1)"
        ),
    ),
    Mutant(
        "M08 every token matches every token",
        4,
        lambda: _patched(tokens_match=lambda *a, **k: True),
    ),
    Mutant("M09 no stem, no prefix: equality only", 1, _equality_only),
    Mutant(
        "M10 the fusion is flattened (RRF_K = 10 000)",
        1,
        lambda: _patched(RRF_K=10_000),
        note=(
            "1/(K+r) is strictly decreasing in r for every K, so a larger K "
            "compresses the scores without reordering them on its own"
        ),
    ),
    Mutant("M11 the cut takes the LAST k", 1, _last_k),
    Mutant("M12 the delivered text is truncated to 40 characters", 6, _truncate_delivered),
    Mutant("M13 the excerpt carries no text", 6, _drop_text_key),
)


def _score(rows: Sequence[Row]) -> dict[int, int]:
    """`k - m` per row: 0 when green, and it falls whether the hits drop or the
    denominator grows.

    FELLED means a row got WORSE, never merely that something changed. A
    mutation that makes a red row green -- removing the quota makes row 3's
    label honest, because there is then no quota to name -- has not been
    caught by row 3, and saying so would credit this gate with a check it does
    not have.
    """
    return {row.number: row.k - row.m for row in rows}


def _unit_vector(
    cases: Sequence[Case],
) -> dict[tuple[str, str, str], tuple[bool, int | None, str | None]]:
    """Every unit's outcome, so a survivor can be reported with what it DID
    move rather than with a guess about why the gate missed it."""
    return {
        (case.question_set.set_id, unit.question_id, unit.named): (
            unit.hit,
            unit.rank,
            unit.klass,
        )
        for case in cases
        for unit in case.units
    }


def row_seven(
    cases: Sequence[Case],
    baseline: Sequence[Row],
    *,
    mutants: Sequence[Mutant] = MUTANTS,
) -> Row:
    """Mutate the ranking and the cut; a gate nothing can fell is not a gate.

    Every mutant is applied IN PROCESS to the module the payload is built
    from, and the judge's reading of the bundle is taken BEFORE the first
    mutation (`bundle_index` is warmed by the baseline), so a mutant cannot
    move the fasit it is being measured against.
    """
    before = _score(baseline)
    baseline_units = _unit_vector(cases)
    killed: list[str] = []
    survivors: list[str] = []
    for mutant in mutants:
        try:
            with mutant.patch():
                remeasured = [measure_case(case.question_set, case.bundles) for case in cases]
                rows = deterministic_rows(remeasured)
            after = _score(rows)
            moved = _unit_vector(remeasured)
            ranks_moved = sum(
                1
                for key, value in moved.items()
                if key in baseline_units and value[1] != baseline_units[key][1]
            )
            deliveries_moved = sum(
                1
                for key, value in moved.items()
                if key in baseline_units and value[0] != baseline_units[key][0]
            )
            worse = [
                number for number in before if after.get(number, before[number]) < before[number]
            ]
        except Exception as error:
            killed.append(f"{mutant.label} -> {type(error).__name__}")
            continue
        if worse:
            killed.append(f"{mutant.label} -> row {', '.join(str(n) for n in worse)}")
        else:
            survivors.append(
                f"{mutant.label} -- row {mutant.expected_row} should have taken it; "
                f"measured: it moved {ranks_moved} rank(s) and {deliveries_moved} "
                "delivery/deliveries on these fixtures"
                + (f"; {mutant.note}" if mutant.note else "")
            )
    total = len(mutants)
    bar = int(-(-total * MUTANT_BAR // 1))
    status = GREEN if total and len(killed) >= bar else RED
    return Row(
        7,
        f"mechanical mutants of the ranking and the cut, felled (bar {MUTANT_BAR:.0%})",
        len(killed),
        total,
        status,
        f"{len(killed)} of {total} felled; the bar is {bar} of {total}",
        [f"    killed: {item}" for item in killed]
        + [f"    SURVIVED: {item}" for item in survivors],
    )


# --- rows 8 and 9: the sets that are not in this repository --------------------

#: What PM measured 2026-09-17 with okf 0.10.0 at the shipped defaults. Carried
#: so row 8 is not blank when it has not run -- and labelled on every line,
#: because a figure this gate did not produce is not this gate's figure.
RECORDED = {
    "wiki-20": "6 of 20 questions (29 fasit entries)",
    "r761-sk2": "7 of 7 positives, all at rank 1 (8 entries incl. KP and KN)",
    "vegnormal-32": "32 of 43 citations = 21 of 32 questions",
    "total": "45 of 70 measurement units, 25 misses, 25 of 25 below_k",
}


def read_real_set(name: str, path: Path, expected_sha256: str) -> QuestionSet:
    """One of the three real sets, in ITS OWN shape, read never written.

    Each set names a fasit differently and each names it in its own file; the
    adapters below are the whole of this gate's knowledge of them, and none of
    the question text ever reaches a tracked file here.
    """
    raw = path.read_bytes()
    measured = hashlib.sha256(raw).hexdigest()
    if measured != expected_sha256:
        raise GateUsage(
            f"{path}: expected sha256 {expected_sha256}, measured {measured}; "
            "refusing to measure a set that is not the set that was pinned"
        )
    spec = json.loads(raw.decode("utf-8"))
    if name == "wiki":
        # The set's own `hit_rule`, verbatim: an excerpt whose `source_file` is
        # the fasit's document AND whose text carries the fasit's quote.
        questions = tuple(
            Question(
                id=str(entry["id"]),
                question=str(entry["question"]),
                fasit=tuple(
                    Fasit(by="source_file", value=f"{item['doc']}.md", quote=str(item["quote"]))
                    for item in entry["fasit"]
                ),
            )
            for entry in spec["questions"]
        )
        return QuestionSet("wiki-20", "wiki", path, measured, questions, ())
    if name == "r761":
        questions = []
        controls = []
        for entry in spec["sporsmal"]:
            if str(entry["id"]).startswith("KN"):
                controls.append(
                    Control(
                        str(entry["id"]), str(entry["sporsmal"]), "the set's own known-negative"
                    )
                )
                continue
            questions.append(
                Question(
                    id=str(entry["id"]),
                    question=str(entry["sporsmal"]),
                    # The fasit is a section TITLE, and the set carries no
                    # quote: this set is measured at concept granularity.
                    fasit=(Fasit(by="title", value=str(entry["fasit"])),),
                )
            )
        return QuestionSet("r761-sk2", "r761", path, measured, tuple(questions), tuple(controls))
    if name == "vegnormal":
        questions = []
        for entry in spec["sporsmal"]:
            by_normal: dict[str, list[Fasit]] = {}
            for item in entry["must_cite"]:
                by_normal.setdefault(str(item["normal"]), []).append(
                    Fasit(by="req_number", value=str(item["req_number"]))
                )
            # One question citing two standards is two Questions, one per
            # bundle, because a payload is built against one bundle. The unit
            # count is unchanged, which is what the denominator counts.
            for normal, fasit in sorted(by_normal.items()):
                suffix = f"/{normal}" if len(by_normal) > 1 else ""
                questions.append(
                    Question(
                        id=f"{entry['id']}{suffix}",
                        question=str(entry["sporsmal"]),
                        fasit=tuple(fasit),
                        bundle=normal,
                    )
                )
        return QuestionSet("vegnormal-32", "", path, measured, tuple(questions), ())
    raise GateUsage(f"unknown real set `{name}`; one of wiki, r761, vegnormal")


#: The three sets row 8 is the measurement of. All three, by name: a run that
#: hands over one of them has measured one of them, and the row says so. Left
#: to `len(real)` the row came back `6 of 6 GREEN` on a single set (PM's J2,
#: 2026-09-19) -- the realistic route being the one set that is at 7 of 7,
#: with the two that miss omitted.
REQUIRED_REAL_SETS: tuple[str, ...] = ("wiki-20", "r761-sk2", "vegnormal-32")


def row_eight(real: Sequence[tuple[QuestionSet, Mapping[str, Path]]]) -> Row:
    """The three real sets. RED when they have not run -- always, in this
    order -- and never green by leaving a set out.

    THE HEADLINE IS AT QUESTION GRANULARITY, and that is not a style choice:
    the three sets do not share a unit. `wiki-20` names a citation, `r761-sk2`
    and `vegnormal-32` name a concept and a requirement number, and adding a
    citation hit to a concept hit produces a number that is neither. A
    question is the one thing all three sets have, so the row counts questions
    -- answered meaning at least one of the question's fasit entries arrived,
    the same reading row 1 prints beside its own units -- and the two unit
    totals are printed below it, each with its own denominator, never summed.
    """
    name = "the real sets (wiki-20, r761-sk2, vegnormal-32), run from path + sha256"
    reason_tail = (
        "a question counts as answered when at least one of its fasit entries "
        "arrived; the two unit granularities are printed apart and never summed"
    )
    if not real:
        return Row(
            8,
            name,
            0,
            len(REQUIRED_REAL_SETS),
            NOT_RUN,
            "not run: no --real argument. The sets live in other repositories and "
            "are never committed here",
            [
                f"    recorded 2026-09-17 by PM, NOT measured by this gate: {key}: {value}"
                for key, value in RECORDED.items()
            ],
        )
    details: list[str] = []
    quoted_hits = quoted_units = concept_hits = concept_units = 0
    answered_total = asked_total = 0
    for question_set, bundles in real:
        cases = []
        for question in question_set.questions:
            key = question.bundle or question_set.bundle
            if key not in bundles:
                details.append(f"    {question_set.set_id}: no bundle given for `{key}`")
                continue
            cases.append((question, bundles[key]))
        units = [unit for question, bundle in cases for unit in measure_units(bundle, question)]
        hits = sum(1 for unit in units if unit.hit)
        answered = len({unit.question_id for unit in units if unit.hit})
        asked = len({unit.question_id for unit in units})
        answered_total += answered
        asked_total += asked
        if question_set.quoted:
            quoted_hits += hits
            quoted_units += len(units)
        else:
            concept_hits += hits
            concept_units += len(units)
        details.append(
            f"    {question_set.set_id}: {hits} of {len(units)} fasit entries "
            f"({'citation' if question_set.quoted else 'concept'} granularity) | "
            f"{answered} of {asked} questions | sha256 {question_set.sha256[:12]}"
        )
        for unit in units:
            if not unit.hit:
                details.append(
                    f"        miss {unit.question_id} {unit.named}: class "
                    f"{unit.klass or '-'} ({unit.detail})"
                )
    details.append(
        f"    NOT SUMMED INTO ONE NUMBER: {quoted_hits} of {quoted_units} at citation "
        f"granularity, {concept_hits} of {concept_units} at concept granularity"
    )
    measured = {question_set.set_id for question_set, _ in real}
    missing = [required for required in REQUIRED_REAL_SETS if required not in measured]
    if missing:
        details.append(
            "    the numbers above are what DID run; the row is not a measurement "
            "of the three sets until all three are given"
        )
        return Row(
            8,
            name,
            answered_total,
            asked_total,
            NOT_RUN,
            "not run: " + ", ".join(missing) + " was not given. " + reason_tail,
            details,
        )
    return _row(
        8,
        name,
        answered_total,
        asked_total,
        "local row: the sets are read from their own repositories, never committed "
        "here. " + reason_tail,
        details,
    )


def row_nine() -> Row:
    """K2: the bundles are on this machine and the gold set is nowhere."""
    return Row(
        9,
        "K2, the sixth set",
        0,
        K2_QUESTIONS,
        RED,
        f"not measured: 0 of {K2_QUESTIONS} questions have a gold set anywhere",
        [
            "    the bundles exist (~/corpora/okf-telling-20260829/K2-bundle-*), the "
            "answer key does not",
            "    a set that cannot be measured is a red number, never an absent row",
            "    who can write it: whoever holds the K2 corpus -- it names documents "
            "that may not be committed here, so it arrives as a path plus a sha256",
        ],
    )


# --- the run ------------------------------------------------------------------

#: What this gate cannot check, whatever the rows say. Printed on every run,
#: because a row's silence about something is not a finding of nothing.
LIMITS: tuple[str, ...] = (
    "It measures RETRIEVAL, never whether a model then answers correctly from what it was handed.",
    "The synthetic corpus is invented and small: it can show that a mechanism "
    "fires, never how often it fires on anyone's documents. That is row 8's job.",
    "A miss classed (b) says the ranking did not reach the fasit. It does not "
    "say which of the ranking's parts is responsible.",
    "Row 4's definition of a marked answer is today's: nothing delivered. A "
    "payload that grows a confidence field moves this row and not the other way.",
    "Row 6 proves the delivered bytes are the bundle's bytes. It cannot prove "
    "they answer the question.",
)

#: Exceptions to 100 %. The first one that arises is the OPERATOR's, so this
#: list is empty until they approve a named row -- and an empty list is printed
#: rather than omitted.
APPROVED_EXCEPTIONS: tuple[dict[str, str], ...] = ()


def deterministic_rows(cases: Sequence[Case]) -> list[Row]:
    """Rows 1, 2, 3, 4 and 6: the ones that run against synthetic fixtures in
    this repository, with no network, no private corpus and no clock."""
    return [
        row_one(cases),
        row_two(cases),
        row_three(cases),
        row_four(cases),
        row_six(cases),
    ]


def synthetic_cases(
    root: Path,
    fixtures: Path = FIXTURES,
    sets: Mapping[str, str] = SYNTHETIC_SETS,
) -> tuple[list[Case], dict[str, Path]]:
    bundles = synthetic_bundles(root)
    cases = []
    for name, pinned in sets.items():
        question_set = load_set(fixtures / name, pinned)
        cases.append(measure_case(question_set, bundles))
    return cases, bundles


def evaluate(
    root: Path,
    *,
    fixtures: Path = FIXTURES,
    registration: Path = HOLDOUT_REGISTRATION,
    real: Sequence[tuple[QuestionSet, Mapping[str, Path]]] = (),
    mutants: Sequence[Mutant] = MUTANTS,
    sets: Mapping[str, str] = SYNTHETIC_SETS,
) -> list[Row]:
    cases, _ = synthetic_cases(root, fixtures, sets)
    rows = deterministic_rows(cases)
    return [
        *rows,
        row_five(registration),
        row_seven(cases, rows, mutants=mutants),
        row_eight(real),
        row_nine(),
    ]


def render(rows: Sequence[Row]) -> str:
    ordered = sorted(rows, key=lambda row: row.number)
    lines = ["row | k of N | status | reason"]
    for row in ordered:
        lines.append(f"{row.number} {row.name} | {row.k} of {row.m} | {row.status} | {row.reason}")
        lines.extend(row.details)
    lines += ["", "exceptions to 100 % (APPROVED by the operator, by name):"]
    lines += [
        f"    - {item['row']}: {item['reason']}; carried instead: {item['carried_instead']}"
        for item in APPROVED_EXCEPTIONS
    ] or ["    - (none). The first exception to arise is the operator's to grant."]
    lines += ["", "what this gate cannot check, whatever the rows say:"]
    lines += [f"    - {limit}" for limit in LIMITS]
    failing = [str(row.number) for row in ordered if row.fails]
    lines.append("")
    verdict = "GATE " + (f"RED: rows {', '.join(failing)}" if failing else "GREEN")
    lines.append(verdict)
    return "\n".join(lines) + "\n"


def _bundle_map(value: str) -> dict[str, Path]:
    """`/a/bundle`, or `N100:2023=/a,N200:2024=/b` for a set spanning bundles."""
    if "=" not in value:
        return {"": Path(value).expanduser()}
    pairs = {}
    for item in value.split(","):
        key, _, path = item.partition("=")
        pairs[key.strip()] = Path(path.strip()).expanduser()
    return pairs


def _real_sets(
    arguments: Sequence[Sequence[str]],
) -> list[tuple[QuestionSet, Mapping[str, Path]]]:
    real = []
    for name, path, sha, bundle in arguments:
        question_set = read_real_set(name, Path(path).expanduser(), sha)
        bundles = _bundle_map(bundle)
        if list(bundles) == [""] and question_set.bundle:
            bundles = {question_set.bundle: bundles[""]}
        real.append((question_set, bundles))
    return real


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0])
    parser.add_argument("--json", action="store_true", help="emit the rows as JSON")
    parser.add_argument(
        "--real",
        nargs=4,
        action="append",
        metavar=("NAME", "SET", "SHA256", "BUNDLE"),
        default=[],
        help=(
            "run row 8 against one real set: NAME is wiki, r761 or vegnormal; "
            "BUNDLE is a path, or `key=path,key=path` for a set spanning bundles"
        ),
    )
    parser.add_argument(
        "--holdout",
        type=Path,
        default=HOLDOUT_REGISTRATION,
        help="the hold-out registration row 5 reads",
    )
    args = parser.parse_args(argv)
    try:
        real = _real_sets(args.real)
        with tempfile.TemporaryDirectory(prefix="okf-retrieval-gate-") as scratch:
            rows = evaluate(Path(scratch), registration=args.holdout, real=real)
    except GateUsage as error:
        print(f"okf-retrieval-gate: {error}", file=sys.stderr)
        return 2
    # Broad on purpose, and the house pattern: a gate that dies with a
    # traceback exits 1, the code it uses for RED, so a reader cannot tell a
    # finding from a crash.
    except Exception as error:
        print(
            f"okf-retrieval-gate: did not run: {type(error).__name__}: {error}",
            file=sys.stderr,
        )
        return 2
    if args.json:
        print(
            json.dumps(
                {
                    "rows": [row.to_json() for row in sorted(rows, key=lambda r: r.number)],
                    "classes": dict(CLASSES),
                    "exceptions": {"approved": list(APPROVED_EXCEPTIONS)},
                    "limits": list(LIMITS),
                    "gate": RED if any(row.fails for row in rows) else GREEN,
                },
                indent=2,
                ensure_ascii=False,
            )
        )
    else:
        print(render(rows), end="")
    return 1 if any(row.fails for row in rows) else 0


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