open/llm-ingestion-okf
1
0
Fork 0
Code Issues Releases 4 Activity
llm-ingestion-okf/tools/okf_retrieval_gate.py
Kjell Tore Guttormsen 88cf67f12e test(fixtures): the STS fixtures and fixture codes are fictitious 
Three STS fixtures still carried the section titles and labels of one real
reference document, and three identifiers were copies of its codes with a
letter or a word swapped. They now describe an invented kitchen counter and
cookbook series: the titles, labels and descriptions of sts-identity.xml,
sts-inherit.xml and sts-empty-label.xml, the P350/P351 document codes, the
99-0001 delivery prefix and chapter 7 of the image and accounting corpora.
Generated fixtures are regenerated and the witness inventory's per-document
totals are identical before and after; only names and text move.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-23 14:52:03 +02:00

2722 lines
106 KiB
Python
Raw Permalink 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 5 and 8 are red on
the shipped code as it stands: no hold-out set has been registered, and the
real set lives outside this repository. Row 7 is red where a mutant survives,
and each survivor is printed with what it moved.
THE PUBLIC ROWS RUN ON INVENTED MATERIAL ONLY (operator decision 2026-09-21).
The earlier test track -- the K2 corpus and two further reference sets -- is
retired: not re-measured, not frozen. Its row (9) and its sets' adapters are gone, and row 8 reads one
local set.
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 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 real set's own `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 bm25, 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. 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
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 sier 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 etter punkt 4.2 og etter "
"noekkelrutine i skjema. Hvert punkt i kontrollen av "
"hytta 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 naar broennen proevetas, hvem som arkiverer "
"analysen, og 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)
),
),
),
)
# --- three fixtures, one per mechanism the default ranking runs ---------------
#
# Until 2026-09-22 row 7 left three mutants standing (M06, M07, M10): every
# synthetic concept was short and opened with its own title, so the passage
# signal, the title/path weight and the fusion constant never DECIDED a
# delivery here, and switching any of them off moved 0 ranks. Each bundle below
# is built so that exactly one of them does. None of them is a tuning of the
# ranker: the corpus is pinned (`SPECS_SHA256`) and `src/` is untouched.
_NOTE = "Styret gjennomgaar notatet og foerer det inn i arkivet. Sekretaeren sender kopi. "
#: THE PASSAGE SIGNAL. One long concept answers in ONE window of its body; ten
#: short concepts carry the question's words in their TITLES and none in their
#: bodies. The field signal reads ten title-weighted matches in short documents
#: above one long body, so only the passage signal -- the best WINDOW, not the
#: whole concept -- puts the gold first. Remove the body windows (M06) and the
#: passage signal reads titles alone, where the gold has none of the words.
PASSAGE = BundleSpec(
"retrieval-passage",
(
DocumentSpec(
"aarbok",
"aarbok.md",
(
ConceptSpec(
"kapittel-tre",
"Kapittel tre i aarboka",
_NOTE * 15
+ "Varmekablene i taket kontrolleres av vaktmesteren hver oktober. "
+ _NOTE * 15,
),
),
),
*(
DocumentSpec(
f"rundskriv-{number:02d}",
f"rundskriv-{number:02d}.md",
(
ConceptSpec(
f"skriv-{number:02d}",
f"Varmekablene i taket, skriv {number:02d}",
"Skrivet er sendt ut og arkivert.",
),
),
)
for number in range(1, 11)
),
),
)
#: THE TITLE/PATH WEIGHT. The gold is named by its PATH alone -- the question's
#: `kanopadling` is its directory and its source file and occurs in no body --
#: while ten decoys carry the question's other word more densely than the gold
#: does, one fewer time each. The path weight puts the gold first in the field
#: signal, which lifts it into k (rank 4 at `k` = 6). Weigh no title and no path
#: (M07) and both signals read bodies only, where the decoys outrank it and it
#: falls out of k.
PATH = BundleSpec(
"retrieval-path",
(
DocumentSpec(
"kanopadling",
"kanopadling.md",
(ConceptSpec("regel", "Regel om vester", "Vester brukes paa vannet."),),
),
*(
DocumentSpec(
f"turnotat-{number:02d}",
f"turnotat-{number:02d}.md",
(
ConceptSpec(
f"notat-{number:02d}",
f"Turnotat {number:02d}",
" ".join(["Vester henger i boden."] * (11 - number))
+ " Notatet er arkivert.",
),
),
)
for number in range(1, 11)
),
),
)
_FUSION_NOTE = "Styret gjennomgaar saken og sekretaeren arkiverer den. "
#: THE FUSION CONSTANT. Reciprocal-rank fusion trades a concept with one very
#: good rank against one with two middling ranks, and `RRF_K` sets the price.
#: The gold is FIRST in the passage signal and 21st in the field signal (a long
#: body, below ten short `isbading-*` path matches, nine longer ones and the
#: decoy); the one decoy is 10th and 11th, a smaller rank SUM (19 against 20).
#: A flattened fusion (M10, `RRF_K` = 10 000, where only the sum counts) puts
#: the decoy first and `k` = 1 no longer reaches the gold. At the shipped
#: `RRF_K` = 60 the gold leads, and measured over K in steps of 10 it keeps the
#: lead through K = 180 and loses it at 190. The eight long log pages hold the
#: decoy's passage rank at 10th and rank below the gold in both signals.
FUSION = BundleSpec(
"retrieval-fusion",
(
DocumentSpec(
"badebok",
"badebok.md",
(
ConceptSpec(
"kapittel",
"Kapittel i badeboka",
_FUSION_NOTE
* 12
+ "Isbading ved brygga, isbading om morgenen, isbading med vakt, "
"isbading hver dag. " + _FUSION_NOTE * 12,
),
),
),
*(
DocumentSpec(
f"logg-{number:02d}",
f"logg-{number:02d}.md",
(
ConceptSpec(
"side",
f"Loggside {number:02d}",
_FUSION_NOTE * 16
+ "Isbading ved brygga, isbading om kvelden, isbading med vakt. "
+ _FUSION_NOTE * 16,
),
),
)
for number in range(8)
),
DocumentSpec(
"isbading-lapp-00",
"isbading-lapp-00.md",
(ConceptSpec("lapp", "Lapp 00", _FUSION_NOTE * 5 + "Isbading er tillatt."),),
),
*(
DocumentSpec(
f"isbading-{number:02d}",
f"isbading-{number:02d}.md",
(ConceptSpec("oppslag", f"Oppslag {number:02d}", "Oppslaget er arkivert."),),
)
for number in range(10)
),
*(
DocumentSpec(
f"isbading-arkiv-{number:02d}",
f"isbading-arkiv-{number:02d}.md",
(ConceptSpec("mappe", f"Mappe {number:02d}", _FUSION_NOTE * 10),),
)
for number in range(9)
),
),
)
_ARCHIVE_LINE = "Styret gjennomgaar notatet og foerer det inn i arkivet. Sekretaeren sender kopi."
#: THE PASSAGE DELIVERY. Every concept above is shorter than
#: `consume.PASSAGE_CHARS`, so no synthetic payload carried a passage and a
#: mutant of one (M15) could not be felled. One concept of about 6 600
#: characters: a heading ten lines in, the answer thirty lines below it, so the
#: delivery is heading + `[...]` + span + `[...]` and the span carries the
#: citation. Rows 1 and 6 read it as a hit only because the judge reads a
#: passage as its exact reconstruction; M15 adds one sentence and fells both.
DELIVERY = BundleSpec(
"retrieval-delivery",
(
DocumentSpec(
"husbok",
"husbok.md",
(
ConceptSpec(
"drift",
"Drift av huset",
"\n".join(
[_ARCHIVE_LINE] * 10
+ ["## Teknisk rom"]
+ [_ARCHIVE_LINE] * 30
+ ["Varmepumpa i kjelleren faar service av roerleggeren hvert aar."]
+ [_ARCHIVE_LINE] * 40
),
),
),
),
),
)
SPECS: Mapping[str, BundleSpec] = {
"positive": POSITIVE,
"miss": MISS,
"single-source": SINGLE_SOURCE,
"budget": BUDGET,
"lookup": LOOKUP,
"quota": QUOTA,
"passage": PASSAGE,
"path": PATH,
"fusion": FUSION,
"delivery": DELIVERY,
}
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 = "089a6a9770c03c59fff2da22430e5f9aeb2b2374c49aed419bae99409a2a5cb9"
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 real
#: sets name different things and a gate that could read only one form would
#: report the others as zero: a source document plus a quote, a requirement
#: number, a 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
#: (`set-classes.json` forces one class per bundle).
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",
"set-mechanisms.json": "435ce620c30c94b0d151335883583520490482be2ade6797524ae928a9f1eab7",
"set-passage.json": "93b6de8ef7012c84516fe9aee560924d4b579669fb046e6eadc70dbf444eb627",
}
# --- 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)
#: What class e says when a passage is not its own reconstruction.
PASSAGE_NOT_RECONSTRUCTED = (
"the delivered passage is not its reconstruction from the bundle's bytes"
)
def passage_span(text: str, passage: object, body: str) -> str | None:
"""The span a passage delivery carries, or None when `text` is not EXACTLY
what the bundle's bytes rebuild.
Since v1.1 C1 a long concept is delivered as `[heading]`, `[...]`, the
span, `[...]` (`consume.as_passage`), with `passage = {start, end, of}`.
Two traps PM measured 2026-09-22, both held here:
- **The offsets count in the DELIVERED body** (`ConceptView.body`), never
in the concept file: read in the file they land a frontmatter's length
off, and 0 of 12 real spans matched.
- **"The span occurs somewhere in the text" is not a check.** It accepts
an invented sentence beside the span. Every character of `text` must be
accounted for: the span is `body[start:end]` byte for byte, a marker
stands exactly where the span leaves text out and nowhere else, and the
one other line allowed is a heading line of the body ABOVE the span (or a
prefix of one -- `as_passage` cuts a long heading). WHICH heading is the
product's choice and is not re-derived here; that it is the bundle's
bytes is the guarantee.
The citation is then read in the span alone -- not in the heading, not in
the markers, and not across the seam between them, which is no sequence
the concept file holds.
"""
if not isinstance(passage, Mapping):
return None
start, end, of = passage.get("start"), passage.get("end"), passage.get("of")
if not all(
isinstance(value, int) and not isinstance(value, bool) for value in (start, end, of)
):
return None
assert isinstance(start, int) and isinstance(end, int)
if of != len(body) or not 0 <= start < end <= len(body):
return None
span = body[start:end]
tail = f"\n{consume.PASSAGE_ELISION}" if end < len(body) else ""
if not text.endswith(span + tail):
return None
head = text[: len(text) - len(span + tail)]
if start == 0:
return span if head == "" else None
marker = f"{consume.PASSAGE_ELISION}\n"
if head == marker:
return span
if not head.endswith(f"\n{marker}"):
return None
heading = head[: -len(f"\n{marker}")]
above = body[:start].split("\n")
if heading.startswith("#") and any(
line.startswith("#") and line.startswith(heading) for line in above
):
return span
return None
def _judged(excerpt: Mapping[str, object], body: str) -> tuple[str, bool, str]:
"""(the text a citation is looked for in, whether it is the bundle's
bytes, the detail when it is not) for one delivered excerpt."""
text = str(excerpt.get("text", ""))
if "passage" in excerpt:
span = passage_span(text, excerpt["passage"], body)
if span is None:
return text, False, PASSAGE_NOT_RECONSTRUCTED
return span, True, ""
return text, _flat(text) == _flat(body), "the delivered text is not the bundle's bytes"
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)
)
# A whole-body delivery is judged whole; a passage is judged as its
# exact reconstruction, and its citation is read in its span alone.
judged = {
concept_id: _judged(delivered[concept_id], index.concepts[concept_id].body)
for concept_id in holding
if concept_id in delivered
}
hit_ids = [
concept_id for concept_id, (text, _, _) in judged.items() if _carries(text, fasit.quote)
]
# The payload SAYS it delivered this; the bundle says what it is.
confirmed: bool | None = None
not_bytes = ""
for concept_id in hit_ids:
_, confirmed, not_bytes = judged[concept_id]
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", not_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.
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 row 8 read then, two of which are retired
#: since 2026-09-21 -- 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 questions of one
#: retired set, 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: one retired set'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 one-document bundle
#: of a few thousand concepts. 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"
if coverage.get("weak") is True:
# v1.1 C4: the payload's own reading, a word the bundle holds in no
# form. Read beside the bar below, never instead of it.
return True
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
#: A registration is committed to a PUBLIC repository and read on more than
#: one machine, so every path it names is written under the home directory
#: and expanded when read. An absolute path names ONE machine -- it is refused
#: with its reason and never followed, so a machine path cannot be registered
#: again without row 5 saying so.
HOME_PREFIX = "~/"
def _home_path(value: object) -> tuple[Path | None, str]:
"""A registered path expanded against the home directory, or None and
the reason it is refused. The text returned beside a path is the
registration's own `~/...`, so a row names what was registered and never
the machine it was read on."""
text = value.strip() if isinstance(value, str) else ""
if not text:
return None, "not named"
if not text.startswith(HOME_PREFIX):
kind = "an absolute path" if Path(text).is_absolute() else "not under the home directory"
return None, f"`{text}` is {kind}; a registration names `{HOME_PREFIX}...` and no machine"
return Path(text).expanduser(), text
def _registered_bundle(spec: Mapping[str, object]) -> tuple[Path | None, str]:
"""The bundle the registration names, IF its tree still measures the
`bundle_ref` the registration pins; otherwise None and why.
A bundle path is a place, and a place can be filled with other bytes: the
ref is what makes the path the registered bundle."""
path, text = _home_path(spec.get("bundle"))
if path is None:
return None, text
registered = spec.get("bundle_ref")
if not isinstance(registered, str) or not registered:
return None, "the registration pins no bundle_ref"
if not path.is_dir():
return None, f"the bundle is absent at {text}"
try:
measured = consume.bundle_ref(path)
except Exception as error: # a bundle that cannot be measured is a NO
return None, f"no ref could be measured at {text}: {type(error).__name__}: {error}"
if measured != registered:
return None, f"{text} measures {measured}, the registration pins {registered}"
return path, f"{text} measures the pinned {registered[:24]}"
def _hold_out_verdict(
set_path: Path | None,
set_text: str,
bundle: Path | None,
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.
A bundle that is not the registered tree is never measured -- a share
read off other bytes is not the registered hold-out's.
"""
if threshold is None:
return False, "no numeric threshold to compare with"
if bundle is None:
return False, "not run: the bundle is not the registered tree"
if set_path is None:
return False, f"not run: the set is {set_text}"
if not set_path.is_file():
return False, f"the set is absent at {set_text}"
try:
question_set = read_hold_out_set(set_path, pinned)
bundles = {question_set.bundle: bundle}
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.
AND IT NAMES NO MACHINE, AND THE BUNDLE IT NAMES IS THE ONE IT PINNED.
Until 2026-09-23 the committed registration carried two absolute paths of
the machine it was written on, in a public repository, and `bundle_ref`
was a field no line read -- other bytes at the registered bundle path
would have been measured as the registered bundle. Its paths are now
`~/...`, expanded when read, and an absolute one is a NO; the bundle's
tree is measured with `consume.bundle_ref` and a ref other than the pinned
one is a NO, with the hold-out NOT run against it.
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",
" both paths written `~/...` -- the file is public and an "
"absolute path names one machine -- and the bundle's tree ref "
"pinned as `bundle_ref`, so other bytes at that path are not measured",
" 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, set_text = _home_path(spec.get("set"))
pinned = str(spec.get("sha256", ""))
readings = spec.get("readings", [])
checks.append(("a set is named", set_path is not None, set_text))
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", "")),
)
)
present = set_path is not None and set_path.is_file()
checks.append(
(
"the pinned bytes are the bytes on disk",
present and set_path is not None and sha256_of(set_path) == pinned,
f"present at {set_text}" if present else f"absent at {set_text}",
)
)
bundle, bundle_note = _registered_bundle(spec)
checks.append(("the bundle is the registered tree", bundle is not None, bundle_note))
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(set_path, set_text, bundle, 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 a mutant that survives, so a survivor is a statement about the
#: mechanism and not a shrug. None does since 2026-09-22: the three that
#: carried a note (M06, M07, M10) each got a fixture that makes their
#: mechanism decide a delivery (`PASSAGE`, `PATH`, `FUSION`).
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)
@contextlib.contextmanager
def _patched_bm25(**attributes: object) -> Iterator[None]:
"""`_patched` for the module the DEFAULT ranking scores in. Since v1.1 C1
`consume` hands the ordering to `bm25.rank`, so a mutant of the fusion's
functions changes code the default no longer runs and can fell nothing."""
original = {name: getattr(bm25, name) for name in attributes}
try:
for name, value in attributes.items():
setattr(bm25, name, value)
yield
finally:
for name, value in original.items():
setattr(bm25, 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 = bm25.rank
def mutant(*args: Any, **kwargs: Any) -> Any:
result = original(*args, **kwargs)
return replace(result, ranked=list(reversed(result.ranked)))
return _patched_bm25(rank=mutant)
def _every_term_everything() -> contextlib.AbstractContextManager[None]:
def mutant(query: Sequence[str], vocabulary: frozenset[str]) -> list[frozenset[str]]:
return [frozenset(vocabulary) for _ in dict.fromkeys(query)]
return _patched_bm25(query_groups=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 _extend_passage() -> contextlib.AbstractContextManager[None]:
"""M14's shape for the delivery form v1.1 C1 added: the passage still
CARRIES the citation in its span, and one sentence at the span's end is not
the concept file's. A judge that asked only whether the span occurs
somewhere in the text would count it."""
original = consume.as_passage
def mutant(excerpt: dict[str, object], window: int) -> dict[str, object]:
out = original(excerpt, window)
if "passage" in out:
text = str(out["text"])
tail = f"\n{consume.PASSAGE_ELISION}"
cut = len(text) - len(tail) if text.endswith(tail) else len(text)
out["text"] = f"{text[:cut]} En setning som ikke staar i konseptfila.{text[cut:]}"
return out
return _patched(as_passage=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 passage signal reads no body", 1, lambda: _patched_bm25(windows=lambda body: [])
),
Mutant(
"M07 the field signal weighs no title and no path",
1,
lambda: _patched_bm25(TITLE_WEIGHT=0, PATH_WEIGHT=0),
),
Mutant("M08 every token matches every token", 4, _every_term_everything),
Mutant("M09 no stem, no prefix: equality only", 1, _equality_only),
Mutant(
"M10 the fusion is flattened (bm25.RRF_K = 10 000)",
1,
lambda: _patched_bm25(RRF_K=10_000),
),
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
),
Mutant("M15 the passage carries a sentence the concept file does not", 6, _extend_passage),
)
#: 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 passage signal reads no body",
"M07 the field signal weighs no title and no path",
"M08 every token matches every token",
"M09 no stem, no prefix: equality only",
"M10 the fusion is flattened (bm25.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",
"M15 the passage carries a sentence the concept file does not",
)
#: The roster's length, written as a number so appending is not one edit.
MUTANT_COUNT = 15
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],
)
# --- row 8: 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",
}
def read_real_set(name: str, path: Path, expected_sha256: str) -> QuestionSet:
"""The real set, in ITS OWN shape, read never written.
The adapter below is the whole of this gate's knowledge of it, and none of
the question text ever reaches a tracked file here. Two more adapters, for
the sets of a retired test track, were removed 2026-09-21 with the track
they belonged to.
"""
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":
return QuestionSet("wiki-20", "wiki", path, measured, _wiki_questions(spec), ())
raise GateUsage(f"unknown real set `{name}`; the one real set is `wiki`")
def _wiki_questions(spec: Mapping[str, Any]) -> tuple[Question, ...]:
"""The wiki set's own `hit_rule`, verbatim: an excerpt whose `source_file`
is the fasit's document AND whose text carries the fasit's quote. One
reading of the schema, shared by row 8 and row 5."""
return 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"]
)
#: The schemas a registered hold-out set may declare in its own `schema`
#: field, and the bundle key its questions are read under. The SET says how
#: it is read, so the registration carries nothing about it; a set with no
#: `schema` is in this gate's own synthetic form (`load_set`), and a schema
#: not named here is a NO with its name -- a set read in a shape it was not
#: written in measures nothing.
HOLD_OUT_SCHEMAS: Mapping[str, str] = {"fase-sporsmaal/1": "wiki"}
def read_hold_out_set(path: Path, expected_sha256: str) -> QuestionSet:
"""The registered hold-out set, read in the schema it declares.
Raises GateUsage for a set that is not the pinned bytes or declares a
schema this gate does not read; `_hold_out_verdict` turns either into a
NO with its reason.
"""
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
schema = spec.get("schema") if isinstance(spec, dict) else None
if schema is None:
return load_set(path, expected_sha256)
if schema not in HOLD_OUT_SCHEMAS:
raise GateUsage(
f"the set's schema `{schema}` is not one this gate reads "
f"({', '.join(HOLD_OUT_SCHEMAS)}, or no `schema` for this gate's own form)"
)
try:
questions = _wiki_questions(spec)
except (KeyError, TypeError, ValueError) as error:
raise GateUsage(f"{path}: not a `{schema}` set this gate can read: {error}") from error
return QuestionSet("hold-out", HOLD_OUT_SCHEMAS[schema], path, measured, questions, ())
#: The sets row 8 is the measurement of, by name: a run that hands over some
#: of them has measured some of them, and the row says so. Left to
#: `len(real)` the row came back `6 of 6 GREEN` on one set of three (PM's J2,
#: 2026-09-19). One set since 2026-09-21, when the two of a retired test track
#: were removed; the rule stays for the day a
#: second set joins.
REQUIRED_REAL_SETS: tuple[str, ...] = ("wiki-20",)
@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. This is the mechanism that refuses a set of
another size, for every set 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 source, read through its own adapter.
#: `questions` is the number of `Question` objects the adapter produces.
REAL_SET_PINS: Mapping[str, RealSetPin] = {
"wiki-20": RealSetPin(
questions=20,
fasit_entries=29,
controls=0,
sha256="972d0f5715d1377b3d89b8ddf391612709b96cd0fe8b96dfe517fe1931a9e333",
),
}
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 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:
sets need not share a unit. One names a citation, another may name a
concept, and adding a citation hit to a concept hit produces a number that
is neither. A question is the one thing every set has, 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 (" + ", ".join(REQUIRED_REAL_SETS) + "), 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 required sets until every one is 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,
)
# --- 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,
) -> 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),
]
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 `Q100:2023=/a,Q200: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 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; "
"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())
Reference in a new issue View git blame Copy permalink