open/llm-ingestion-okf
1
0
Fork 0
Code Issues Releases 4 Activity

test(accounting): row 6 sees a refusal, a second real corpus, and 34 of 34 mutants

Browse source 
Row 6 was GREEN with R761 100 % rejected: every element of a refused document
is booked as a coded rejection, so u = 0 and d = 0 and the bundle is empty.
`refused_whole` asks that question on its own now -- the build order asked for
an honest red there, and PM re-measured the green on 2026-09-18 with
`okf build` exiting 1 unseen.

A skipped row no longer leaves the verdict unqualified (`GATE GREEN (row 6 not
run: ...)`), and the exit code is non-zero locally when a corpus source is on
the machine and its row did not run. m-2.

N200 Vegbygging:2024 joins R761 as a second real corpus. R761 holds 0 `fig`,
0 formulas and 0 references, so the only real corpus could not have found the
hole in the STS role map; N200 carries 194 citations, 49 figures and 135
footnotes. A `.json` whose root holds an STS node tree is counted as STS
rather than as keys and leaves -- the container is not the content.

M-4: the review's 26 mutants, ported to the code as it stands, plus 8 for the
new checks. 34 of 34 killed. `tools/okf_gate_mutants.py` runs on a copy of the
tree, and a pattern that does not match is an ERROR and exit 2 -- a mutant
that could not be applied was never measured. That fired once, on M13, after a
refactor moved the line it edits.

m-3: `APPROVED_EXCEPTIONS` was read by no row, so approving one changed
nothing. Each pair is now checked against the witness's own vocabulary and the
run says why it moves no denominator. The gate also prints its OWN limits
beside the verdict, m-5 among them.

The product's accounting tests state the new truth instead of the old one:
`okf build --accounting` over the fixture corpus exits 1 with SIX unaccounted
elements in its own vocabulary -- its first real finding, reachable only now
that fixtures carry the constructs. Four shared element names disagree with
the witness, each pinned with its cause; one of the four is a double count
this package makes (a text box's paragraph, once inside the box and again in
the paragraph carrying it).

Three fixture defects were found and fixed while building them, each of which
would have reported a loss the build never had: a shared string table not
related to the workbook, a `graphicData` with no `uri`, and an odt
`styles.xml` without `<office:styles/>`.

Report: docs/2026-09-18-regnskapsgaten-herdet.md

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kjell Tore Guttormsen 2026-09-18 02:48:34 +02:00
commit ed8d9d709f
Signed by: ktg
SSH key fingerprint: SHA256:JakMjO6FTBBzN0Bhfj9saOoEjaFxlSdYuZQQpM/lF9Q
11 changed files with 1146 additions and 85 deletions
Download patch file Download diff file Expand all files Collapse all files

331
tools/okf_accounting_gate.py
View file

@ -58,11 +58,12 @@ import io
import json
import os
import re
import shutil
import sys
import tempfile
import warnings
import zipfile
from collections.abc import Iterable, Mapping
from collections.abc import Iterable, Mapping, Sequence
from dataclasses import dataclass, field
from pathlib import Path, PurePosixPath
from typing import Any
@ -85,6 +86,7 @@ PDF_FIXTURE = CORPUS / "prosess-84-tabell.pdf"
README = REPO / "README.md"
R761_DEFAULT = Path.home() / "repos" / "vegnormal-okf" / "data" / "raw" / "860019"
N200_DEFAULT = R761_DEFAULT.parent / "N200-2024-860015.json"
R761_ZIP = "14ce59dc-2150-480b-b661-6ea605fe3b24.zip"
R761_JSON = "R761-2025-860019.json"
R761_PDF = "R761-prosesskoden-2025.pdf"
@ -130,6 +132,60 @@ PROPOSED_EXCEPTIONS: tuple[dict[str, str], ...] = (
)
#: Limits of the instrument itself, printed beside the verdict. A gate that
#: only reports the build's gaps and none of its own invites the reader to
#: take a green row for a guarantee.
LIMITS: tuple[str, ...] = (
"a short element (a label, a one-word title) often stands elsewhere in the "
"same document, so finding it proves it is present and not that THIS one is",
"two pointed files with identical bytes are one content-addressed asset, so "
"one of them losing its pointer is invisible here (m-5)",
"an image embedded in a binary container has no source file to hash, so a "
"carry of it is counted as one the gate cannot check",
"absence is never verified: an element booked as REJECTED is not looked for "
"in the bundle, only one booked as carried",
"the witness is a second implementation of the same definitions, so a "
"definition that is wrong for a format is wrong on both sides at once",
)
def exception_effect(suffix: str, element: str) -> str:
"""What an approved exception actually does to the numbers.
m-3: `APPROVED_EXCEPTIONS` was read by no row at all, so approving one
changed nothing and the list could say anything. It still moves no
denominator -- and now the run SAYS why, per pair, from the witness's own
vocabulary rather than from the sentence next to the list.
"""
vocabulary = FORMAT_VOCABULARY.get(suffix)
if vocabulary is None:
return f"no witness reads {suffix}, so the pair names nothing"
if element in vocabulary:
return (
f"WARNING: the witness DOES count `{element}` for {suffix}, so this "
"approval would lower a denominator"
)
return f"the witness counts no `{element}` for {suffix}: no denominator moves"
#: Every element name a witness can produce, per file type.
FORMAT_VOCABULARY: dict[str, tuple[str, ...]] = {
".csv": witness.CSV,
".docx": witness.DOCX,
".htm": witness.HTML,
".html": witness.HTML,
".json": witness.JSON,
".md": witness.MARKDOWN,
".odt": witness.ODT,
".pdf": witness.PDF,
".pptx": witness.PPTX,
".rtf": witness.RTF,
".txt": witness.TEXT,
".xlsx": witness.XLSX,
".xml": witness.STS_ROLES,
}
@dataclass
class Row:
number: int
@ -305,15 +361,26 @@ def _sha12(path: Path) -> str:
def asset_holds(build: Build, source: Path) -> bool:
"""Did the run carry THESE bytes, under the name the layout gives them?
"""Did the run carry THESE bytes, placed under their own content address?
Both halves are load-bearing. The name alone was the check until an
independent review wrote a zero-byte file called `<sha12>-x.png` and the
gate read it as a carry (m-1); the bytes alone would credit a file the
layout would have named something else.
Both halves are load-bearing and neither is the build's naming rule. The
NAME alone was the check until an independent review wrote a zero-byte
file called `<sha12>-x.png` and the gate read it as a carry (m-1). The
BYTES alone would credit an asset standing under any name at all, which
is the property the layout exists to guarantee.
What the gate deliberately does NOT reproduce is the readable tail: the
build lowercases the source's basename, folds its separator runs, cuts it
to a maximum and takes the suffix from the BYTES rather than from the
name. Re-implementing that here would make the judge agree with the judged
by construction -- and it would be wrong: measured 2026-09-18 on R761,
whose own hrefs carry spaces, capitals and parentheses, a judge checking
the full name reported 50 of 50 carried images as missing.
"""
digest = _sha256(source)
return build.assets.get(f"{digest[:12]}-{source.name}") == digest
return any(
found == digest and name.startswith(digest[:12]) for name, found in build.assets.items()
)
# --- the judge's own reading of the bundle ------------------------------------
@ -526,7 +593,10 @@ def _document_unit(
elif booked > have:
double += booked - have
notes.append(f"{element}: {booked} booked, source has {have}")
want = max(carried, 0) + max(pointer, 0)
# Capped at what the source holds: a booking ABOVE that is already
# reported as double, and counting the excess as "claimed and not
# found" would report one defect under two names.
want = min(max(carried, 0) + max(pointer, 0), have) if have else 0
if want == 0:
continue
if not persisted:
@ -858,60 +928,156 @@ def witness_pairs(r761: Path | None) -> tuple[list[tuple[str, list[str]]], list[
return pairs, notes
#: The two builds row 6 runs: the default gate is what a user gets; `none`
#: persists the document, which is the only way its pictures are carried and
#: the double booking of the files beside it becomes visible.
#: The two builds row 6 runs per corpus: the default gate is what a user gets;
#: `none` persists the document, which is the only way its pictures are
#: carried and the double booking of the files beside it becomes visible.
R761_GATES: tuple[str | None, ...] = (None, "none")
def row6(r761: Path | None, ci: bool) -> Row:
"""R761 through two builds; a unit is clean only if it is clean in both."""
name = "real corpus: R761 Prosesskoden:2025"
if r761 is None or not r761.is_dir():
@dataclass(frozen=True)
class RealCorpus:
"""A corpus of real documents, read only, outside this repository."""
label: str
kind: str # "zip" or "file"
path: Path
@property
def available(self) -> bool:
return self.path.exists()
def real_corpora(r761: Path | None, n200: Path | None) -> list[RealCorpus]:
"""R761 is the gate's original corpus; N200 was added 2026-09-18 because
R761 holds NONE of the STS classes the role map was missing -- 0 `fig`, 0
formulas, 0 references -- so the only real corpus could not have found the
hole. N200 carries 194 citations, 49 figures and 135 footnotes."""
corpora = []
if r761 is not None:
corpora.append(RealCorpus("R761 Prosesskoden:2025", "zip", r761 / R761_ZIP))
if n200 is not None:
corpora.append(RealCorpus("N200 Vegbygging:2024", "file", n200))
return corpora
def _corpus_inbox(corpus: RealCorpus, root: Path) -> Path:
inbox = root / "inbox"
inbox.mkdir(parents=True)
if corpus.kind == "zip":
with zipfile.ZipFile(corpus.path) as archive:
archive.extractall(inbox)
else:
shutil.copy2(corpus.path, inbox / corpus.path.name)
return inbox
def clean_in_every_run(runs: Sequence[Sequence[Unit]]) -> int:
"""Units clean in EVERY build, never in any of them.
The two gates see different things -- the default one refuses, `none`
persists and carries the pictures -- so a unit that is clean in one and
dirty in the other has a fate the run does not agree on, and calling that
clean would let either build cover for the other.
"""
return sum(1 for parts in zip(*runs) if all(unit.clean for unit in parts))
def refused_whole(documents: Mapping[str, Any], build: Build) -> str | None:
"""Did this run persist NOTHING of a corpus that holds documents?
Booking every element of a refused document as a coded rejection gives
u = 0 and d = 0, so the numbers are clean and the bundle is empty. Row 6
read GREEN on R761 with 110 of 110 elements rejected and `okf build`
exiting 1 unseen. The build order asked for an honest red there, so the
row asks this question on its own.
"""
if not documents:
return None
persisted = sum(1 for name in documents if name in build.source_files)
if persisted:
return None
elements = sum(sum(entry["elements"].values()) for entry in documents.values())
codes = sorted(
{str(d.get("code")) for d in (build.accounting or {}).get("documents", []) if d.get("code")}
)
return (
f"the default gate persisted 0 of {len(documents)} document(s) "
f"({', '.join(codes) or 'no code declared'}), {elements} element(s) rejected whole; "
f"okf build exited {build.exit_code}"
)
def row6(r761: Path | None, n200: Path | None, ci: bool) -> Row:
"""Every real corpus through two builds; a unit is clean only in both."""
name = "real corpora"
corpora = [c for c in real_corpora(r761, n200) if c.available]
missing = [c for c in real_corpora(r761, n200) if not c.available]
if not corpora:
# SKIPPED is only free when there is nothing to measure. A source that
# EXISTS and was not measured is a row that did not run, and a row
# that did not run is not a row that passed.
status = SKIPPED if ci else RED
return Row(6, name, 0, 0, status, f"not measured, source missing: {r761}")
names = ", ".join(str(c.path) for c in missing) or "no corpus configured"
return Row(6, name, 0, 0, status, f"not measured, source missing: {names}")
door = door_available()
runs: list[tuple[str, Build, list[Unit]]] = []
with zipfile.ZipFile(r761 / R761_ZIP) as archive, tempfile.TemporaryDirectory() as tmp:
inbox = Path(tmp) / "inbox"
archive.extractall(inbox)
inventory = witness.witness_inbox(inbox)
json_counts = dict(witness.count_sts_json((r761 / R761_JSON).read_bytes()).counts)
for index, gate_name in enumerate(R761_GATES):
build = run_build(inbox, Path(tmp) / f"work{index}", door=door, gate=gate_name)
runs.append((gate_name or "default", build, account(inventory, build, inbox)))
files = inventory["files"]
pointed = sum(1 for f in files.values() if f["pointed_at_by"])
documents = inventory["documents"]
details = [
f"witness (json): {json.dumps(json_counts, sort_keys=True)}",
f"zip: {len(documents)} document(s), {len(files)} other files "
f"({pointed} pointed at, {len(files) - pointed} not)",
]
reasons = []
for label, build, units in runs:
persisted = sum(1 for d in documents if d in build.source_files)
u_total = sum(u.unaccounted for u in units)
d_total = sum(u.double for u in units)
reasons.append(f"gate {label}: u = {u_total}, d = {d_total}")
details: list[str] = [f"{c.path} not measured: source missing" for c in missing]
reasons: list[str] = []
clean = total = 0
refused: list[str] = []
for corpus in corpora:
with tempfile.TemporaryDirectory() as tmp:
inbox = _corpus_inbox(corpus, Path(tmp))
inventory = witness.witness_inbox(inbox)
runs: list[tuple[str, Build, list[Unit]]] = []
for index, gate_name in enumerate(R761_GATES):
build = run_build(inbox, Path(tmp) / f"work{index}", door=door, gate=gate_name)
runs.append((gate_name or "default", build, account(inventory, build, inbox)))
documents = inventory["documents"]
files = inventory["files"]
pointed = sum(1 for f in files.values() if f["pointed_at_by"])
elements = sum(sum(e["elements"].values()) for e in documents.values())
details.append(
f"gate {label}: exit {build.exit_code}, {persisted} of {len(documents)} "
f"document(s) persisted, {len(build.asset_prefixes)} asset file(s)"
f"{corpus.label}: {len(documents)} document(s), {elements} element(s), "
f"{len(files)} other file(s) ({pointed} pointed at, {len(files) - pointed} not)"
)
for unit in units:
if unit.kind == "document":
details.append(
f" {unit.name}: u={unit.unaccounted} d={unit.double} ({'; '.join(unit.notes)})"
)
doubled = [u for u in units if u.kind == "file" and u.double]
if doubled:
details.append(
f" {len(doubled)} file(s) carried through the document AND rejected, "
f"e.g. {doubled[0].name}"
for label, build, units in runs:
persisted = sum(1 for d in documents if d in build.source_files)
u_total = sum(u.unaccounted for u in units)
d_total = sum(u.double for u in units)
unverified = sum(u.unverified for u in units)
reasons.append(
f"{corpus.label} gate {label}: u = {u_total}, d = {d_total}, "
f"{unverified} claimed and not found"
)
clean = sum(1 for parts in zip(*(units for _, _, units in runs)) if all(u.clean for u in parts))
total = len(runs[0][2])
return _row(6, name, clean, total, f"{'; '.join(reasons)} over {total} units", details)
details.append(
f" gate {label}: exit {build.exit_code}, {persisted} of {len(documents)} "
f"document(s) persisted, {len(build.assets)} asset file(s); {_tally(units)}"
)
for unit in units:
if unit.kind == "document" and not unit.clean:
details.append(
f" {unit.name}: u={unit.unaccounted} d={unit.double} "
f"unverified={unit.unverified} invalid={unit.invalid} "
f"({'; '.join(unit.notes)})"
)
doubled = [u for u in units if u.kind == "file" and u.double]
if doubled:
details.append(
f" {len(doubled)} file(s) carried through the document AND rejected, "
f"e.g. {doubled[0].name}"
)
if label == "default":
whole = refused_whole(documents, build)
if whole is not None:
refused.append(f"{corpus.label}: {whole}")
clean += clean_in_every_run([units for _, _, units in runs])
total += len(runs[0][2])
details.extend(refused)
status = GREEN if total > 0 and clean == total and not refused else RED
reason = "; ".join(reasons) + f" over {total} unit(s)"
if refused:
reason = "a real corpus is refused whole under the default gate; " + reason
return Row(6, name, clean, total, status, reason, details)
def row7(workdir: Path) -> Row:
@ -944,7 +1110,7 @@ def row7(workdir: Path) -> Row:
# --- the run -----------------------------------------------------------------
def evaluate(*, r761: Path | None, ci: bool, consume: bool) -> list[Row]:
def evaluate(*, r761: Path | None, n200: Path | None, ci: bool, consume: bool) -> list[Row]:
table = readme_types()
inventory = load_inventory(INVENTORY)
rejected_inventory = load_inventory(REJECTED_INVENTORY)
@ -959,7 +1125,7 @@ def evaluate(*, r761: Path | None, ci: bool, consume: bool) -> list[Row]:
rejected_build = run_build(REJECTED, Path(tmp) / "rejected", door=door)
rows.append(row4(rejected_inventory, rejected_build))
rows.append(row5(*witness_pairs(r761)))
rows.append(row6(r761, ci))
rows.append(row6(r761, n200, ci))
if consume:
rows.append(row7(work))
return rows
@ -979,7 +1145,10 @@ def render(rows: list[Row]) -> str:
)
lines.append(f"exceptions approved ({APPROVED_ON}), and none moves a denominator:")
for suffix, element in sorted(APPROVED_EXCEPTIONS) or [("(none)", "")]:
lines.append(f" - {suffix} {element}")
lines.append(f" - {suffix} {element}: {exception_effect(suffix, element)}")
lines += ["", "what this gate cannot check, whatever the rows say:"] + [
f" - {limit}" for limit in LIMITS
]
lines += [
"",
"not counted by any witness -- what no row here can see (per file type):",
@ -988,11 +1157,17 @@ def render(rows: list[Row]) -> str:
for item in witness.NOT_COUNTED[suffix]:
lines.append(f" - {suffix}: {item}")
failing = [str(r.number) for r in rows if r.fails]
skipped = [r for r in rows if r.status == SKIPPED]
lines.append("")
lines.append(
f"GATE {'RED' if failing else 'GREEN'}"
+ (f": rows {', '.join(failing)}" if failing else "")
)
verdict = f"GATE {'RED' if failing else 'GREEN'}"
if failing:
verdict += f": rows {', '.join(failing)}"
# A row that did not run is not a row that passed, and the one line most
# readers stop at is this one: `GATE GREEN` with a silent SKIPPED behind
# it is the shape the review reproduced with `CI=1` and a missing source.
for row in skipped:
verdict += f" (row {row.number} not run: {row.reason})"
lines.append(verdict)
return "\n".join(lines) + "\n"
@ -1005,14 +1180,29 @@ def main(argv: list[str] | None = None) -> int:
default=R761_DEFAULT,
help="directory holding the R761 zip, JSON and PDF (read only)",
)
parser.add_argument(
"--n200",
type=Path,
default=N200_DEFAULT,
help="the N200 JSON delivery, the second real corpus (read only)",
)
parser.add_argument(
"--consume", action="store_true", help="also run row 7 (diagnostic, never fails)"
)
args = parser.parse_args(argv)
try:
rows = evaluate(r761=args.r761, ci=bool(os.environ.get("CI")), consume=args.consume)
except (OSError, ValueError, KeyError) as exc:
print(f"okf-accounting-gate: did not run: {exc}", file=sys.stderr)
rows = evaluate(
r761=args.r761,
n200=args.n200,
ci=bool(os.environ.get("CI")),
consume=args.consume,
)
# Broad on purpose: a gate that dies with a traceback exits 1, which is
# the code it uses for RED, so a reader cannot tell a finding from a
# crash. `ET.ParseError`, `BadZipFile` and `CalledProcessError` all
# reached that path (m-4).
except Exception as exc:
print(f"okf-accounting-gate: did not run: {type(exc).__name__}: {exc}", file=sys.stderr)
return 2
if args.json:
payload = {
@ -1026,7 +1216,18 @@ def main(argv: list[str] | None = None) -> int:
print(json.dumps(payload, indent=2, ensure_ascii=False))
else:
print(render(rows), end="")
return 1 if any(r.fails for r in rows) else 0
if any(r.fails for r in rows):
return 1
# A row skipped while its source is on this machine did not run, and a
# zero here would report that as a pass.
for row in rows:
if row.status == SKIPPED and any(c.available for c in real_corpora(args.r761, args.n200)):
print(
f"okf-accounting-gate: row {row.number} was skipped while its source exists",
file=sys.stderr,
)
return 1
return 0
if __name__ == "__main__":

318
tools/okf_gate_mutants.py Normal file
View file

@ -0,0 +1,318 @@
"""Mutation run over the content-accounting gate and its witness.
The gate judges `okf build`; this asks what judges the gate. Each mutant is ONE
textual edit that makes the instrument weaker in a way a reader would call a
defect, and the question is whether the suite goes red. A mutant that survives
names a check nothing holds.
The twenty-six mutants are an independent review's, ported to the code as it
stands rather than re-invented: at `0b00de4` twelve of them survived the
forty-two tests, among them `main` always returning 0 and six of the seven
witness mutants -- which had no fixture that could exercise the element they
removed.
python3 tools/okf_gate_mutants.py
Runs on a COPY of the tree in a temporary directory: the working tree is never
edited, so an interrupted run cannot leave a mutant behind. A pattern that does
not match the expected number of times is reported as an ERROR and the run
exits 2 -- a mutant that could not be applied was never measured, and counting
it as killed is the same mistake as reading an empty search as an absence.
"""
from __future__ import annotations
import shutil
import subprocess
import sys
import tempfile
from dataclasses import dataclass
from pathlib import Path
REPO = Path(__file__).resolve().parent.parent
GATE = "tools/okf_accounting_gate.py"
WITNESS = "tools/okf_witness.py"
SUITE = "tests/test_accounting_gate.py"
@dataclass(frozen=True)
class Mutant:
label: str
file: str
old: str
new: str
#: How many times `old` must occur. A role map written twice on purpose
#: (`_sts_role_xml`, `_sts_role_json`) has two, and the mutant edits the
#: first -- the XML witness -- so the JSON one stays honest and row 5 has
#: a chance to see the disagreement.
occurrences: int = 1
first_only: bool = False
MUTANTS: tuple[Mutant, ...] = (
Mutant(
"M01 m=0 counts as green ('not measured' -> GREEN)",
GATE,
"GREEN if m > 0 and k == m else RED",
"GREEN if k == m else RED",
),
Mutant(
"M02 row 6 never fails the gate",
GATE,
"self.status == RED and self.number <= 6",
"self.status == RED and self.number <= 5",
),
Mutant(
"M03 a missing corpus is SKIPPED locally too",
GATE,
"status = SKIPPED if ci else RED",
"status = SKIPPED",
),
Mutant(
"M04 double booking of elements never counted",
GATE,
"double += booked - have",
"double += 0",
),
Mutant(
"M05 'declared carried, bytes absent' never a false claim",
GATE,
'false_claim = fate == "carried" and not carried',
"false_claim = False",
),
Mutant(
"M06 an unpointed duplicate counts as carried",
GATE,
"carried = bool(pointed_by) and asset_holds(build, corpus / name)",
"carried = asset_holds(build, corpus / name)",
),
Mutant(
"M07 row 2 accepts ANY declared inventory",
GATE,
"if got != want:",
"if got is None:",
),
Mutant(
"M08 row 1 ignores a stale committed fasit",
GATE,
'if fresh["documents"].get(name) == entry:',
"if True:",
),
Mutant(
"M09 row 4 ignores the Images 'found' count",
GATE,
"elif int(found.group(2)) != declared_images:",
"elif False:",
),
Mutant(
"M10 row 4 accepts any element total in the log line",
GATE,
'rf"{re.escape(doc)}: {total} elements found in the source, 0 carried: "',
'rf"{re.escape(doc)}: \\d+ elements found in the source, 0 carried: "',
),
Mutant(
"M11 an unavailable witness is agreement",
GATE,
'return ["a witness is unavailable"]',
"return []",
),
Mutant(
"M12 row 5 counts every pair as agreeing",
GATE,
"good = sum(1 for _, problems in pairs if not problems)",
"good = len(pairs)",
),
Mutant(
"M13 row 6: clean in ANY of the two builds is enough",
GATE,
"if all(unit.clean for unit in parts)",
"if any(unit.clean for unit in parts)",
),
Mutant(
"M14 main always exits 0",
GATE,
" if any(r.fails for r in rows):\n return 1",
" if False:\n return 1",
),
Mutant(
"M15 T silently drops .rtf",
GATE,
'return sorted(set(re.findall(r"^\\| `(\\.[a-z0-9]+)` \\|", section, re.MULTILINE)))',
'return sorted(\n set(re.findall(r"^\\| `(\\.[a-z0-9]+)` \\|", section, re.MULTILINE)) - {".rtf"}\n )',
),
Mutant(
"M16 an element with no declared fate is ignored",
GATE,
"for element in sorted(set(elements) | set(fates)):",
"for element in sorted(set(fates)):",
),
Mutant(
"M17 conservation always 'held'",
GATE,
'return build.exit_code == 0 and "K1b FAILED" not in build.log',
"return True",
),
Mutant(
"M18 row 3 counts every unit as clean",
GATE,
"clean = sum(1 for u in units if u.clean)",
"clean = len(units)",
),
Mutant(
"M19 a negative rejection is absorbed into the total",
GATE,
"booked = carried + pointer + sum(rejected.values())",
"booked = carried + pointer + abs(sum(rejected.values()))",
),
# --- the judge's own reading of the bundle, added 2026-09-18 -------------
Mutant(
"M20 a booked carry is never looked for in the bundle",
GATE,
"found = sum(1 for pieces in sayable if all(finder(piece) for piece in pieces))",
"found = len(elements_pieces)",
),
Mutant(
"M21 an unverified booking is clean",
GATE,
"return not (self.unaccounted or self.double or self.unverified or self.invalid)",
"return not (self.unaccounted or self.double)",
),
Mutant(
"M22 any asset file at all proves a carry",
GATE,
" found == digest and name.startswith(digest[:12])"
" for name, found in build.assets.items()",
" True for name, found in build.assets.items()",
),
Mutant(
"M23 a corpus refused whole is not reported",
GATE,
" if persisted:\n return None",
" if True:\n return None",
),
Mutant(
"M24 a rejection code outside the closed list is accepted",
GATE,
" unknown = sorted(c for c in rejected if c not in REJECTION_CODES)",
" unknown = []",
),
Mutant(
"M25 a skipped row leaves the verdict unqualified",
GATE,
' verdict += f" (row {row.number} not run: {row.reason})"',
' verdict += ""',
),
Mutant(
"W01 docx: separator footnotes counted",
WITNESS,
'if role != "comment" and int(note.get(f"{_W}id", "0")) <= 0:',
'if role != "comment" and int(note.get(f"{_W}id", "0")) <= -1:',
),
Mutant(
"W02 pptx: pictures never counted",
WITNESS,
' for _ in root.iter(f"{_P}pic"):\n count.add("image")',
' for _ in ():\n count.add("image")',
),
Mutant(
"W03 sts (xml witness): footnotes have no role",
WITNESS,
' if tag == "fn":\n return "footnote"',
' if tag == "fn":\n return None',
occurrences=2,
first_only=True,
),
Mutant(
"W04 xlsx: images never counted",
WITNESS,
' for _ in root.iter(f"{_XDR}pic"):\n count.add("image")',
' for _ in ():\n count.add("image")',
),
Mutant(
"W05 odt: list items never counted",
WITNESS,
' for item in root.iter(f"{_TEXT}list-item"):',
" for item in ():",
),
Mutant(
"W06 docx: tables never counted",
WITNESS,
' for table in root.iter(f"{_W}tbl"):',
" for table in ():",
),
Mutant(
"W07 sts (xml witness): table cells have no role",
WITNESS,
' if tag in ("td", "th"):\n return "cell"',
' if tag in ("td", "th"):\n return None',
occurrences=2,
first_only=True,
),
Mutant(
"W08 docx: a header or footer is never counted",
WITNESS,
' count.add("header_footer", *lines)',
" pass",
),
Mutant(
"W09 sts json: the twin reuses the XML map",
WITNESS,
" role = _sts_role_json(tag, parent, grandparent)",
" role = _sts_role_xml(tag, parent, grandparent)",
),
)
def _apply(text: str, mutant: Mutant) -> str:
return text.replace(mutant.old, mutant.new, 1 if mutant.first_only else -1)
def main(argv: list[str] | None = None) -> int:
del argv
survived: list[str] = []
killed = 0
errors: list[str] = []
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp) / "tree"
shutil.copytree(
REPO,
root,
ignore=shutil.ignore_patterns(".git", ".venv", "__pycache__", "*.egg-info"),
)
for mutant in MUTANTS:
path = root / mutant.file
original = path.read_text(encoding="utf-8")
seen = original.count(mutant.old)
if seen != mutant.occurrences:
errors.append(
f"{mutant.label}: pattern occurs {seen} times, expected "
f"{mutant.occurrences} -- NOT MEASURED"
)
continue
path.write_text(_apply(original, mutant), encoding="utf-8")
try:
run = subprocess.run(
[sys.executable, "-m", "pytest", SUITE, "-q", "-x", "-p", "no:cacheprovider"],
cwd=root,
capture_output=True,
text=True,
)
finally:
path.write_text(original, encoding="utf-8")
tail = (run.stdout.strip().splitlines() or [run.stderr.strip()[-160:]])[-1]
if run.returncode != 0:
killed += 1
print(f"killed {mutant.label} [{tail}]")
else:
survived.append(mutant.label)
print(f"SURVIVED {mutant.label} [{tail}]")
print(f"\nkilled {killed} of {len(MUTANTS)}")
for label in survived:
print(f" survived: {label}")
for problem in errors:
print(f" ERROR: {problem}")
return 2 if errors else 0
if __name__ == "__main__":
raise SystemExit(main())

27
tools/okf_witness.py
View file

@ -453,13 +453,15 @@ def _sts_role_xml(tag: str, parent: str | None, grandparent: str | None) -> str
number neither of them should have produced (independent review, M-2).
In this delivery a section's label is `sec/label` and a table's label is
`table-wrap/label`.
`table-wrap/label`. Measured on R761 2026-09-18: `sec/label` 7 714,
`sec/title/label` **0** -- the nested placement is a fact about the JSON
delivery and does not belong in this map.
"""
if tag == "sec":
return "section"
if tag == "title" and parent == "sec":
return "title"
if tag == "label" and (parent == "sec" or (parent == "title" and grandparent == "sec")):
if tag == "label" and parent == "sec":
return "section_label"
if tag == "label" and parent == "table-wrap":
return "table_label"
@ -578,6 +580,22 @@ def count_sts_xml(data: bytes) -> tuple[Count, list[str], bool]:
return count, refs, True
def is_sts_json(data: bytes) -> bool:
"""Is this the publisher's JSON delivery of an STS document?
A standard shipped as JSON holds sections, titles, citations and tables.
Counted as generic JSON it holds keys and leaves: the container, not the
content -- and then no row can see that a citation left the bundle."""
if b'"standardContent"' not in data:
return False
try:
document = json.loads(data)
except ValueError:
return False
content = document.get("standardContent") if isinstance(document, dict) else None
return isinstance(content, dict) and isinstance(content.get("c"), list)
def count_sts_json(data: bytes) -> Count:
"""The same roles, read from the publisher's JSON node tree."""
document = json.loads(data)
@ -1226,7 +1244,10 @@ def witness_file(inbox: Path, path: Path) -> Inventory:
elif suffix == ".csv":
count, name = count_csv(data.decode("utf-8-sig")), "csv"
elif suffix == ".json":
count, name = count_json(data.decode("utf-8-sig")), "json"
if is_sts_json(data):
count, name = count_sts_json(data), "sts json node tree"
else:
count, name = count_json(data.decode("utf-8-sig")), "json"
elif suffix in (".html", ".htm"):
parser = _HtmlCounter()
parser.feed(data.decode("utf-8-sig"))