open/llm-ingestion-okf
1
0
Fork 0
Code Issues Releases 4 Activity
llm-ingestion-okf/tools/okf_retrieval_gate.py
Kjell Tore Guttormsen cf21449ddb docs: remove what this repository published about a consumer's corpus 
Operator decision 2026-09-21: nothing from that consumer's collection goes out
on the public remote. The NAME stays where it is already published -- it is a
consumer of this library, named as such, and removing it would mean rewriting
published history, which this repository does not do. What goes is everything
that describes their CONTENT.

Removed across README, CLAUDE.md, CHANGELOG, four dated reports, the
consumption contract, three source modules and three test modules: their
corpus's document and page counts, the concept count of a bundle built from
it, the byte figures of a payload built from it, the question and fasit counts
and recorded score of their evaluation set, a bundle id with two content refs,
an order id naming them, and a path into their repository.

Kept, because the argument survives without the corpus: RATIOS and
percentages. A ratio is the finding -- a withheld list that is 65.5 % of a
payload is a defect at any corpus size -- and it discloses nothing about how
large anyone's collection is. Where a claim lost its denominator it now SAYS
so rather than quietly reading as unmeasured: the gate-refusal limitation in
the README states that the corpus and its counts are deliberately withheld and
points the reader at their own build, which is the number that binds them
anyway.

One integrity pin is kept and named here rather than left to be found: the
retrieval gate still pins that set by sha256, because the pin is what refuses
a self-written file in the right shape, and a checksum discloses nothing about
what it checksums. Its recorded SCORE is gone -- that was their figure about
their own corpus, and the row now says so instead of restating it.

The known-positive constants move with the contract document, as they must.
Suite green, 2372 passed.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-21 04:28:15 +02:00

2459 lines
95 KiB
Python
Raw Blame History

"""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 subprocess
import sys
import tempfile
from collections.abc import Callable, Iterator, Mapping, Sequence
from dataclasses import dataclass, field, replace
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 an internal measurement note. The bundles exist on
#: the machine this row was written against; 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 specs_digest(specs: Mapping[str, BundleSpec] = SPECS) -> str:
"""The synthetic corpus as one sha256 over its own fields.
The SETS are pinned and the CORPUS was not, which is a hole one size
smaller than the one it guards: PM tuned the corpus 2026-09-19 and took
row 3 green, and what caught it was row 2's forced classes rather than a
pin. A more careful tuning that preserved those classes was left standing.
A digest over the specs is not a digest over the bundle's bytes -- that is
`build_bundle`'s job and it is deterministic -- but it is the same
guarantee the sets have: these bytes, or exit 2.
"""
canonical = json.dumps(
{
name: [
[
document.directory,
document.source_file,
[
[
concept.slug,
concept.title,
concept.body,
concept.description,
concept.repeat,
]
for concept in document.concepts
],
]
for document in spec.documents
]
for name, spec in sorted(specs.items())
},
ensure_ascii=False,
sort_keys=True,
)
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
#: The synthetic corpus, pinned the way the sets are.
SPECS_SHA256 = "8d999838f72a4c151e12ff6ac511b253c3437dba6290d2a7ea97dc546747242d"
def synthetic_bundles(root: Path, specs: Mapping[str, BundleSpec] = SPECS) -> dict[str, Path]:
"""Every synthetic bundle, written once and reused by every row."""
measured = specs_digest(specs)
if specs is SPECS and measured != SPECS_SHA256:
raise GateUsage(
f"the synthetic corpus is not the corpus that was pinned: expected "
f"{SPECS_SHA256}, measured {measured}. Every row below counts against "
"these documents; move the pin in the same commit that moves them"
)
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, ...]
#: What the set says must be TRUE OF THE RUN for its rows to mean
#: anything. Today one name: `source_quota_in_force`, declared by the set
#: whose fixtures exist to catch a false `source_quota_exceeded` label.
requires: tuple[str, ...] = ()
@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", [])
),
requires=tuple(str(item) for item in spec.get("requires", [])),
)
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-quota.json": "61fda652719d7403ddf9701d50e914c46572a4fda77dc0336838695e18c498cf",
"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
#: The class the set DECLARES for this unit, carried so rows 2 and 3 can
#: take their denominator from the pinned bytes instead of from the run.
expect_class: str | None = None
#: Did the source quota move this question at all -- did the default cut
#: and the quota-off cut deliver different concepts? Row 3's known-positive:
#: with no quota in force, the label row 3 judges is never printed.
quota_moved: bool = False
def _withheld_rules(payload: Mapping[str, object]) -> dict[str, str]:
"""The rule for EVERY withheld concept, which is why the runs below ask
for the whole list.
Since `okf-consumption/2` a payload names only the nearest N drops by
default -- the right shape for a reader and the wrong one for an
instrument that classifies every miss by the rule it fell under. The block
states `complete`, so the demand is checked rather than assumed: a
truncated block here would silently classify most misses as unfound.
"""
block = payload.get("withheld")
assert isinstance(block, Mapping)
assert block.get("complete") is True, (
"the payload names a sample of the withheld set, so a rule map built "
"from it would be missing the concepts it was asked about"
)
entries = block.get("nearest")
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,
withheld_full=True,
)
truth_run = consume.build_payload(
bundle,
question=question.question,
k=question.k,
limit=question.limit,
source_quota=None,
withheld_full=True,
)
delivered = _delivered(default)
withheld = _withheld_rules(default)
truth_delivered = _delivered(truth_run)
truth_withheld = _withheld_rules(truth_run)
quota_moved = set(delivered) != set(truth_delivered)
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:
declared = excerpt.get("rank")
rank = declared if isinstance(declared, int) else None
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),
expect_class=question.expect_class,
quota_moved=quota_moved,
)
)
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]
# A fixture that DECLARES the class it forces and then comes back a hit has
# not been classified: its premise broke. Counting it out of the
# denominator is what let `k = 32` take this row from 7 of 7 to 4 of 4 and
# call it green (PM's J10, 2026-09-19). The denominator is therefore the
# pinned set's own, not the run's.
broken = [unit for case in cases for unit in case.units if unit.hit and unit.expect_class]
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
]
details += [
f" premise broken: {unit.question_id} {unit.named} declares class "
f"{unit.expect_class} and came back a hit; the fixture must be re-measured"
for unit in broken
]
return _row(
2,
"every miss carries exactly one class",
k,
len(misses) + len(broken),
"each class forced by its own fixture; an unplaced miss makes this 0 of N, "
"and a fixture that stops producing its declared class counts against it",
details,
)
#: The classes a set DECLARES that come back withheld with a reason printed.
#: (a) is not in the bundle at all and (d) was delivered, so neither can carry
#: one -- the other three must.
WITHHELD_CLASSES = ("b", "c", "e")
#: A set may say what must be true OF THE RUN before its rows mean anything.
REQUIRE_QUOTA = "source_quota_in_force"
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.
THE DENOMINATOR IS THE PINNED SET'S, NOT THE RUN'S. Counting only what
this run happened to withhold is how `k = 32` took the row from 2 of 5 to
2 of 2 and called it green (PM's J10): the three fixtures that declare
class b were delivered, and a unit that leaves the denominator answers
nothing. Every unit whose set declares a withheld class is judged whether
or not this run withheld it, and a declared unit with no printed reason is
not an honest one.
AND THE ROW CARRIES ITS OWN KNOWN-POSITIVE. With `--source-quota` off
every printed reason is true -- there is no quota left to name falsely --
and the row read `6 of 6 GREEN` while row 1 fell to 8 of 9 (PM's J8). That
reading is not a lie, it is an empty measurement, so a set may declare
`source_quota_in_force` and the row is NOT RUN for it when the default cut
and the quota-off cut deliver the same concepts everywhere.
"""
judged: list[Unit] = []
for case in cases:
for unit in case.units:
declared = unit.expect_class in WITHHELD_CLASSES
if declared or unit.label_default is not None:
judged.append(unit)
agreeing: list[Unit] = []
lying: list[Unit] = []
for unit in judged:
if unit.label_default is None:
lying.append(unit)
continue
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}: the set declares class "
f"{unit.expect_class} and the payload printed no reason "
f"({'delivered' if unit.hit else unit.truth})"
)
if unit.label_default is None
else (
f" {unit.question_id} {unit.named}: payload says "
f"`{unit.label_default}`, the quota-off run says `{unit.truth}`"
)
for unit in lying
]
silent = [
case
for case in cases
if REQUIRE_QUOTA in case.question_set.requires
and not any(unit.quota_moved for unit in case.units)
]
if silent:
return Row(
3,
"the reason the payload prints for a withheld fasit is true",
len(agreeing),
len(judged),
NOT_RUN,
"not run: no quota is in force for "
+ ", ".join(case.question_set.set_id for case in silent)
+ " -- the default cut and the quota-off cut deliver the same concepts, "
"so the label this row judges was never printed",
details,
)
return _row(
3,
"the reason the payload prints for a withheld fasit is true",
len(agreeing),
len(judged),
"`rule` must say what the same cut without the quota says; the denominator "
"is the set's declared withheld units, never the ones this run withheld",
details,
)
#: Row 4's bar, and the ONE threshold this gate applies to a payload.
#:
#: SWEPT over 81 questions on 2026-09-20 -- the 16 of the synthetic sets and
#: the 65 of the three real sets -- against `coverage.unanswered_in_bundle` as
#: a share of the question's own terms:
#:
#: - at **0.50** row 4 is 6 of 6 and ELEVEN real questions whose fasit is in
#: their bundle come back marked;
#: - at **2/3** row 4 is 6 of 6 and **0 of 65** real questions are marked; the
#: highest a real covered question reaches is 0.6087 (two of the vegnormal
#: set's, 14 of 23 terms), and the lowest an uncovered control reaches is
#: 0.6667 (N3, 4 of 6);
#: - at **0.70** N3 falls under the bar and row 4 is 5 of 6.
#:
#: It collapses at both ends, which is what makes 2/3 a reading of the
#: distribution rather than a number picked to pass. The margin is THIN --
#: 0.6087 against 0.6667 -- and it is published that way.
#:
#: WHAT IT DOES NOT CATCH, measured on the same data: `r761-sk2`'s own
#: known-negative, a question naming a section number the document does not
#: have, sits at 0.2857 -- every other word of it is in a 2 756-concept road
#: standard. One term absent among seven is not a share, and no bar over this
#: list reaches it. Row 4's denominator is the synthetic controls, so this
#: costs the row nothing and is stated rather than implied.
UNANSWERED_BAR = 2 / 3
def marked(payload: Mapping[str, object]) -> bool:
"""Can a consumer READ, from the payload alone, that the bundle does not
cover the question?
Two readings, and both are facts the payload carries:
- nothing was delivered, and the withheld rules say why; or
- the bundle answers none of at least `UNANSWERED_BAR` of the question's
own terms (`coverage.unanswered_in_bundle`, added to the payload
2026-09-20 for exactly this reading).
Eight excerpts with no score, no confidence and no statement is not a
marked answer -- it is the same payload an answered question gets, which
is what took this row to 3 of 6 on N3, N4 and N5.
THE BAR IS THIS GATE'S AND NOT THE LIBRARY'S. `consume.unanswered_terms`
records the two readings that were built and falsified against real
corpora; a library asserting "this bundle does not cover your question"
would be asserting across corpora what was measured on one. A gate is
where a declared bar belongs, and this one carries its sweep above.
"""
counts = payload.get("denominators")
assert isinstance(counts, dict)
delivered = counts["delivered"]
assert isinstance(delivered, int)
if delivered == 0:
return True
coverage = payload.get("coverage")
assert isinstance(coverage, dict), "the payload carries no coverage block to read"
terms = coverage["question_terms"]
unanswered = coverage["unanswered_in_bundle"]
assert isinstance(terms, list) and isinstance(unanswered, list)
if not terms:
return True
return len(unanswered) / len(terms) >= UNANSWERED_BAR
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, and the bundle answers enough of the question's "
"terms that nothing in the payload says so"
)
)
return _row(
4,
"an uncovered question comes back marked, a covered one does not",
correct,
total,
"marked = a reading the consumer can act on: nothing delivered, or the "
f"bundle answers none of >= {UNANSWERED_BAR:.0%} of the question's own terms",
details,
)
#: The path a capability session changes. Row 5 reads git for the ONE thing a
#: registration cannot assert about itself: that it was already committed when
#: that path moved.
RANKING_PATH = "src/llm_ingestion_okf/consume.py"
@dataclass(frozen=True)
class Provenance:
"""What GIT says about a file. Every field here is a fact about the
repository's history, which is the one thing the file cannot also write.
Unknown is NOT unknown-and-therefore-fine: outside a git tree, or with no
git on PATH, the fields come back in their refusing form and the note says
why.
"""
tracked: bool
unmodified: bool
commit: str
commit_touches_ranking: bool
ranking_commits_after: int
note: str = ""
def _git(repo: Path, *arguments: str) -> tuple[int, str]:
try:
finished = subprocess.run(
["git", "-C", str(repo), *arguments],
capture_output=True,
text=True,
check=False,
)
except OSError as error:
return 127, str(error)
return finished.returncode, finished.stdout
def git_provenance(path: Path, *, repo: Path = REPO, ranking: str = RANKING_PATH) -> Provenance:
"""`path`'s history, read from git and never from `path`."""
refusing = Provenance(False, False, "", True, 0)
code, _ = _git(repo, "rev-parse", "--git-dir")
if code != 0:
return replace(refusing, note="not a git tree, or git is not on PATH")
tracked = _git(repo, "ls-files", "--error-unmatch", "--", str(path))[0] == 0
if not tracked:
return replace(refusing, note=f"{_display(path)} is not tracked in this repository")
unmodified = _git(repo, "diff", "--quiet", "HEAD", "--", str(path))[0] == 0
_, log = _git(repo, "log", "--diff-filter=A", "--format=%H", "--", str(path))
commits = [line.strip() for line in log.splitlines() if line.strip()]
if not commits:
return replace(
refusing,
tracked=True,
unmodified=unmodified,
note="no commit adds this file; it is staged and not committed",
)
commit = commits[-1]
_, touched = _git(repo, "show", "--pretty=", "--name-only", commit)
_, after = _git(repo, "log", "--format=%H", f"{commit}..HEAD", "--", ranking)
return Provenance(
tracked=True,
unmodified=unmodified,
commit=commit,
commit_touches_ranking=ranking in touched.split(),
ranking_commits_after=len([line for line in after.splitlines() if line.strip()]),
)
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 _as_share(value: object) -> float | None:
"""A threshold as a number in [0, 1], or None.
A SHARE and not any float: every metric this gate carries is `k of N`, so
a threshold of `80` is either 80 % written wrongly or a bar no run can
clear, and both are refusals rather than guesses.
"""
try:
number = float(str(value).strip())
except (TypeError, ValueError):
return None
if not 0.0 <= number <= 1.0:
return None
return number
def _hold_out_verdict(
spec: Mapping[str, object],
set_path: Path,
pinned: str,
threshold: float | None,
) -> tuple[bool, str]:
"""Run the registered hold-out set against the registered bundle and put
its share beside the threshold.
Every refusal is a NO with its reason and never an exception: a
registration naming an absent set is a finding about the registration.
"""
if threshold is None:
return False, "no numeric threshold to compare with"
bundle = str(spec.get("bundle", ""))
if not bundle:
return False, "the registration names no bundle to measure against"
if not set_path.is_file():
return False, f"the set is absent at {_display(set_path)}"
try:
question_set = load_set(set_path, pinned)
bundles = _bundle_map(bundle)
if list(bundles) == [""] and question_set.bundle:
bundles = {question_set.bundle: bundles[""]}
units = [
unit
for question in question_set.questions
for unit in measure_units(bundles[question.bundle or question_set.bundle], question)
]
except Exception as error: # a registration that cannot be run is a NO
return False, f"the hold-out did not run: {type(error).__name__}: {error}"
asked = len({unit.question_id for unit in units})
if not asked:
return False, "the hold-out set carries no question; a share over 0 is not a number"
answered = len({unit.question_id for unit in units if unit.hit})
share = answered / asked
return share >= threshold, f"{answered} of {asked} = {share:.4f} against {threshold:.4f}"
def row_five(
registration: Path = HOLDOUT_REGISTRATION,
*,
provenance: Callable[[Path], Provenance] = git_provenance,
) -> 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.
AND A THRESHOLD IS A NUMBER THAT IS COMPARED WITH ONE. Until 2026-09-20
the check was `bool(threshold)`, so the threshold `report-only; any number
is acceptable for v1` read as `a threshold is written: yes`. Two checks
now: it parses as a share in [0, 1], and the registered set is RUN against
the registered bundle so its own `answered of asked` stands beside it. A
registration naming an absent set, an unreadable bundle or an empty set is
a NO with its reason -- never an exception, and never a silent pass.
AND NEITHER IS A PROTECTION THE FILE WRITES ABOUT ITSELF. Until
2026-09-19 every check here read a field the registration owned, and two
files PM wrote in the moment came back `7 of 7 GREEN`. Three checks now
read GIT instead: the registration is committed and unmodified, the commit
that ADDED it is not itself a change to the ranking, and a change to the
ranking landed AFTER it. The third is the one that cannot be self-attested
-- it is green only in the order a pre-registration actually happens, the
registration first and the ranking change second, and it is red today
because neither has happened.
WHAT GIT CANNOT PROVE, stated rather than implied: that nobody read the
number before writing the threshold. A number can be read from an
uncommitted working tree, and no history shows that. What history does
show is ORDER, and order is what these three checks are.
"""
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 as a NUMBER in [0, 1], with a date, "
"before its number is read -- `report-only` passed the old check "
"and could fell nothing",
" and a bundle named, so the set can be RUN and its share put "
"beside the threshold. A number never compared with a measurement "
"is a note",
" and COMMITTED before the ranking moves: git must show the "
f"registration in a commit of its own, with a later commit to "
f"{RANKING_PATH}. That is the half a session cannot write about itself",
],
)
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]))
number = _as_share(threshold)
checks.append(
(
"the threshold is a number",
number is not None,
threshold if number is None else f"{number:.4f}",
)
)
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",
)
)
# AND IT IS COMPARED WITH SOMETHING. A number written down and never put
# beside a measurement is a note, not a gate: `bool(threshold)` was the
# whole check until 2026-09-20, and `report-only; any number is acceptable
# for v1` passed it. The hold-out is run HERE, against the bundle the
# registration names, and the row says what it measured.
cleared, note = _hold_out_verdict(spec, set_path, pinned, number)
checks.append(("the measured hold-out clears the threshold", cleared, note))
history = provenance(registration)
checks.append(
(
"git: the registration is committed, unmodified",
history.tracked and history.unmodified and bool(history.commit),
history.note or (f"{history.commit[:12]} clean" if history.unmodified else "modified"),
)
)
checks.append(
(
"git: its commit is not itself a ranking change",
bool(history.commit) and not history.commit_touches_ranking,
f"{RANKING_PATH} in the same commit"
if history.commit_touches_ranking
else "separate commit",
)
)
checks.append(
(
"git: a ranking change landed after it",
history.ranking_commits_after > 0,
f"{history.ranking_commits_after} commit(s) touching {RANKING_PATH} since",
)
)
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 _extend_delivered() -> contextlib.AbstractContextManager[None]:
"""The delivered text still CARRIES the citation and is no longer the
concept file's bytes: the one shape that reaches `confirmed` at all, since
every other mutation of the text empties `hit_ids` one step earlier."""
original = consume.delivered_text
def mutant(body: str) -> str:
return original(body) + "\n\nEn setning som ikke staar i konseptfila."
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),
Mutant(
"M14 the delivered text carries a sentence the concept file does not", 6, _extend_delivered
),
)
#: THE ROSTER, PINNED APART FROM THE LIST IT NAMES. `MUTANTS` lives in the
#: file a capability session edits, and the bar used to be 90 % of
#: `len(MUTANTS)`: seven duplicate `M03` entries appended took row 7 from
#: 11 of 13 RED to 18 of 20 GREEN with the same two survivors (PM's J3,
#: 2026-09-19). Lengthening the list now has to lengthen this tuple and move
#: the count below it -- three edits in three places, each of them readable as
#: what it is -- and a duplicate label is refused outright, because two copies
#: of one mutation are one mutation whatever the roster says.
MUTANT_ROSTER: tuple[str, ...] = (
"M01 the lookup partition is off",
"M02 the title-covered partition is off",
"M03 k = 1",
"M04 the ranking is reversed",
"M05 the source quota is removed",
"M06 the body signal is dead",
"M07 the document prior is dead",
"M08 every token matches every token",
"M09 no stem, no prefix: equality only",
"M10 the fusion is flattened (RRF_K = 10 000)",
"M11 the cut takes the LAST k",
"M12 the delivered text is truncated to 40 characters",
"M13 the excerpt carries no text",
"M14 the delivered text carries a sentence the concept file does not",
)
#: The roster's length, written as a number so appending is not one edit.
MUTANT_COUNT = 14
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,
roster: Sequence[str] = MUTANT_ROSTER,
) -> Row:
"""Mutate the ranking and the cut; a gate nothing can fell is not a gate.
THE LIST IS THE ROSTER OR THE ROW DID NOT RUN. The bar is a share, so a
longer list is a lower bar per survivor; padding `MUTANTS` with seven
copies of one easy mutation made the row green without felling anything
new. The labels must be the pinned roster exactly, in order, with no
duplicate, and the bar is taken from the ROSTER's length.
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.
"""
labels = [mutant.label for mutant in mutants]
duplicates = sorted({label for label in labels if labels.count(label) > 1})
name = f"mechanical mutants of the ranking and the cut, felled (bar {MUTANT_BAR:.0%})"
if duplicates or tuple(labels) != tuple(roster):
return Row(
7,
name,
0,
len(roster),
NOT_RUN,
"not run: the mutant list is not the pinned roster "
+ (
f"({len(duplicates)} duplicate label(s): {', '.join(duplicates)})"
if duplicates
else f"({len(labels)} given, {len(roster)} pinned)"
),
[
" a bar that is a share of the list makes a longer list an "
"easier bar; the roster is pinned apart from the list it names"
],
)
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(roster)
bar = int(-(-total * MUTANT_BAR // 1))
status = GREEN if total and len(killed) >= bar else RED
return Row(
7,
name,
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.
#: One entry is a CONSUMER's set, and its recorded score is that consumer's
#: figure about their own corpus. It is not restated here -- this repository
#: publishes the shape of a measurement, never a consumer's content or its
#: counts -- so the row says the set was measured elsewhere and leaves the
#: number to them. The pin below still refuses a self-written file, because an
#: integrity check is not a disclosure.
RECORDED = {
"wiki-20": "measured by its owner; figure not restated here",
"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": "not summed: one set's figure is not restated here",
}
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":
r761_questions: list[Question] = []
controls: list[Control] = []
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
r761_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(r761_questions), tuple(controls)
)
if name == "vegnormal":
vegnormal_questions: list[Question] = []
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 ""
vegnormal_questions.append(
Question(
id=f"{entry['id']}{suffix}",
question=str(entry["sporsmal"]),
fasit=tuple(fasit),
bundle=normal,
)
)
return QuestionSet("vegnormal-32", "", path, measured, tuple(vegnormal_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")
@dataclass(frozen=True)
class RealSetPin:
"""What a real set IS, stated here rather than taken from the command line.
Until 2026-09-20 BOTH the file and its expected sha256 came from the
caller, and `set_id` was decided by the adapter rather than by the file:
three one-question files written in the three shapes, against a
self-written bundle, read `wiki-20: 1 of 1 ... | 3 of 3 | GREEN`. Nothing
said how big `wiki-20` is. Row 9 has had the mechanism since 2026-09-19
(`K2_QUESTIONS` refuses a set of another size); this is that mechanism for
the three sets row 8 requires.
THREE COUNTS, NOT ONE. The sha256 is the strongest and the least
informative: it says the bytes are the pinned bytes and nothing about what
they contain. The two counts are what a reader can check against the
source, and they are what a re-freeze of a set would move. All three are
facts about files this repository never holds -- a digest and two integers
name no document.
THE LIMIT, STATED: this table is in the file a capability session edits,
exactly as `SYNTHETIC_SETS` and `SPECS_SHA256` are. It raises the cost of
the attack (the set, the bundle AND this table) and does not remove it;
the suite is the rest of the gate, and says so.
"""
questions: int
fasit_entries: int
controls: int
sha256: str
#: Measured 2026-09-20 against the three sources, each read through its own
#: adapter. `questions` is the number of `Question` objects the adapter
#: produces, which is why `vegnormal-32` is 37: five of its 32 questions cite
#: two standards, and a payload is built against one bundle.
REAL_SET_PINS: Mapping[str, RealSetPin] = {
"wiki-20": RealSetPin(
questions=20,
fasit_entries=29,
controls=0,
sha256="972d0f5715d1377b3d89b8ddf391612709b96cd0fe8b96dfe517fe1931a9e333",
),
"r761-sk2": RealSetPin(
questions=7,
fasit_entries=7,
controls=1,
sha256="c834a478e4888300845de9e166808a3942085cb73c6e9e5fd2a3e1a6e9c5e6fd",
),
"vegnormal-32": RealSetPin(
questions=37,
fasit_entries=43,
controls=0,
sha256="c3932fc9abd144989bdbc50c4e4627ac5cc59937c4204f92422b7fe10af87faa",
),
}
def check_real_pin(question_set: QuestionSet) -> None:
"""The set the adapter produced, against what this gate says that set is.
Raises rather than warning: a set that is not the pinned set measured
something else, and row 8's number is then attached to a name it has not
earned.
"""
pin = REAL_SET_PINS.get(question_set.set_id)
if pin is None:
raise GateUsage(
f"{question_set.set_id}: no pin for this set; row 8 measures the "
f"pinned sets {', '.join(REQUIRED_REAL_SETS)} and no others"
)
measured = (
len(question_set.questions),
question_set.units,
len(question_set.controls),
question_set.sha256,
)
expected = (pin.questions, pin.fasit_entries, pin.controls, pin.sha256)
if measured != expected:
raise GateUsage(
f"{question_set.set_id}: pinned as {pin.questions} question(s), "
f"{pin.fasit_entries} fasit entr(ies), {pin.controls} control(s), "
f"sha256 {pin.sha256[:12]}; measured {measured[0]}, {measured[1]}, "
f"{measured[2]}, sha256 {question_set.sha256[:12]} -- a set of another "
"size or another content is another set wearing this one's name"
)
def bundle_identity(bundle: Path) -> str:
"""What a reader needs to run the same measurement again: the path, the
`bundle_id` the root index declares and the content ref.
SS 3.3's own distinction, both halves printed: a `bundle_id` is the
producer's assertion and a ref is a fact about bytes. Three builds on this
machine carry one `bundle_id` at three refs, so the id alone names a
bundle no better than the set's sha256 names a bundle.
"""
try:
return (
f"{_display(bundle)} | bundle_id {consume.root_bundle_id_of(bundle)} "
f"| ref {consume.bundle_ref(bundle)}"
)
except Exception as error: # an unreadable bundle is a line, never a crash
return f"{_display(bundle)} | identity unreadable: {type(error).__name__}: {error}"
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]}"
)
# THE BUNDLE IS NAMED, not only the set. Until 2026-09-20 the row
# printed the set's digest and nothing about what it was measured
# against, so `44 of 64` could neither be reproduced nor felled by
# anyone reading the output.
used = sorted({question.bundle or question_set.bundle for question, _ in cases})
details.extend(
f" measured against {key or '-'} = {bundle_identity(bundles[key])}"
for key in used
)
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(k2: tuple[QuestionSet, Mapping[str, Path]] | None = None) -> Row:
"""K2: the bundles are on this machine and the gold set is nowhere.
IT TAKES AN INPUT, so it is a measurement and not a placeholder. Until
2026-09-19 this row was a hard-coded RED that could not have gone green on
the day somebody wrote the set; it now reads one through `--k2`, in this
gate's own set shape, and `K2_QUESTIONS` is the denominator whatever the
file carries -- a set of five would be a different set with this one's
name.
WITHOUT A SET IT STAYS RED rather than NOT RUN, and that is this row's own
published rule ("a set that cannot be measured is a red number, never an
absent row"): the denominator is KNOWN -- six questions, recorded -- so
the absence is measured. Row 8 says NOT RUN because ITS denominator is not
known until the sets arrive. Both fail the gate identically.
"""
if k2 is None:
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 on the machine this row was written "
"against; the answer key does not, anywhere",
" 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 shape to write it in is this gate's own set shape, the one "
"`tests/fixtures/retrieval/set-*.json` is written in",
],
)
question_set, bundles = k2
units = [
unit
for question in question_set.questions
for unit in measure_units(bundles[question.bundle or question_set.bundle], question)
]
answered = len({unit.question_id for unit in units if unit.hit})
details = [
f" {question_set.set_id}: {answered} of {K2_QUESTIONS} questions | "
f"{sum(1 for unit in units if unit.hit)} of {len(units)} fasit entries "
f"({'citation' if question_set.quoted else 'concept'} granularity) | "
f"sha256 {question_set.sha256[:12]}"
]
details += [
f" miss {unit.question_id} {unit.named}: class {unit.klass or '-'} ({unit.detail})"
for unit in units
if not unit.hit
]
return _row(
9,
"K2, the sixth set",
answered,
K2_QUESTIONS,
f"the recorded denominator is {K2_QUESTIONS} questions, whatever the file carries",
details,
)
# --- 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.",
"The judge reads a concept through `consume.read_concept` and "
"`consume.delivered_text` -- the same parser it judges, not the same "
"NUMBER. Measured 2026-09-19: the index is warmed BEFORE the first "
"mutation, so a mutation of `delivered_text` moves the payload and not the "
"judge (M14 is felled); an index built UNDER such a mutation would move "
"both sides equally, and the gate never builds one.",
)
#: 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,
k2: tuple[QuestionSet, Mapping[str, Path]] | None = None,
) -> 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(k2),
]
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: list[tuple[QuestionSet, Mapping[str, Path]]] = []
for name, path, sha, bundle in arguments:
question_set = read_real_set(name, Path(path).expanduser(), sha)
# The command line said what the file is; this says what the set is.
check_real_pin(question_set)
bundles = _bundle_map(bundle)
if list(bundles) == [""] and question_set.bundle:
bundles = {question_set.bundle: bundles[""]}
real.append((question_set, bundles))
return real
def _k2_set(
argument: Sequence[str] | None,
) -> tuple[QuestionSet, Mapping[str, Path]] | None:
if not argument:
return None
path, sha, bundle = argument
question_set = load_set(Path(path).expanduser(), sha)
if len(question_set.questions) != K2_QUESTIONS:
raise GateUsage(
f"{path}: K2's denominator is {K2_QUESTIONS} questions and this set "
f"carries {len(question_set.questions)}; a set of another size is "
"another set wearing this one's name"
)
bundles = _bundle_map(bundle)
if list(bundles) == [""]:
bundles = {question_set.bundle: bundles[""]}
return question_set, bundles
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(
"--k2",
nargs=3,
metavar=("SET", "SHA256", "BUNDLE"),
help=(
"run row 9 against a K2 gold set, written in this gate's own set "
"shape; the denominator stays the recorded six questions"
),
)
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)
k2 = _k2_set(args.k2)
with tempfile.TemporaryDirectory(prefix="okf-retrieval-gate-") as scratch:
rows = evaluate(Path(scratch), registration=args.holdout, real=real, k2=k2)
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())
Reference in a new issue View git blame Copy permalink