open/llm-ingestion-okf
1
0
Fork 0
Code Issues Releases 4 Activity
llm-ingestion-okf/tools/okf_accounting_gate.py
Kjell Tore Guttormsen 8b4f409aa5
 fix(accounting): the skipped-row guard measures the machine, not the argument (H5) 
`real_corpora(R761_DEFAULT, N200_DEFAULT)` instead of the parsed
arguments. Row 6 is SKIPPED exactly when the corpora the arguments
name are absent, so asking the arguments made the branch unreachable.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-19 05:57:02 +02:00

1268 lines
48 KiB
Python
Raw Blame History

"""The content-accounting gate for `okf build` (capability loop, step 3).
One command, one exit code. For every supported file type it asks whether the
bundle accounts for what the SOURCE holds: of M elements the source carries,
how many does the build account for as carried, as a pointer or as a coded
rejection -- and how many does it account for NOT AT ALL (u) or TWICE (d).
Written RED, before any capability. `okf build`'s conservation identity,
`merged + coded rejections == N`, counts FILES: a file can be "merged" while
content inside it is gone, and a file can be "rejected" while its bytes ride
into the bundle through a document that points at it. Neither is visible to
the identity, and both are measured here.
THE FASIT NEVER COMES FROM THE READER IT JUDGES. Element counts come from
`tools/okf_witness.py`, which imports no `llm_ingestion_okf` module (a test
proves it on the live import graph) and is committed as data in
`tests/fixtures/accounting/*inventory.json`. This module imports the package
only to RUN the build it judges.
THE DOOR THE CAPABILITY MUST OPEN (the contract this gate reads). `okf build`
accepts `--accounting PATH` and writes one JSON object there:
{"accounting_version": 1,
"refused": <documents the build read and persisted nothing of>,
"documents": [
{"source_file": "<inbox-relative path>",
"status": "persisted" | "rejected", "code": "<rejection code>" | null,
"inventory": {"<element>": <count>, ...},
"fates": {"<element>": {"carried": n, "pointer": n,
"rejected": {"<code>": n}}}}],
"files": [
{"source_file": "<inbox-relative path>",
"fate": "carried" | "merged" | "rejected", "code": "<code>" | null}]}
`inventory` is taken BEFORE extraction and before the persist gate, in the
witness's element vocabulary (per file type, defined in `okf_witness.py`), so
a document the gate refuses still has one. `files` covers every inbox file
that is not a document the build reads; `fate` is exactly one value, so a file
whose bytes were carried through a document is `carried` and never also
`rejected`. Until the flag exists, rows 2 and 3 say so and stay red.
A rejected document is reported in `log.md` as ONE line, and row 4 reads it:
<source_file>: <M> elements found in the source, 0 carried: document rejected `<code>`
with M the document's inventory total. The `Images` bullet's "found" count is
the SOURCE's (every image the documents declare), never what a reader got to.
EXCEPTIONS to 100 % are listed in the output and are NOT APPROVED: none of
them lowers a denominator until the operator approves it by name.
"""
from __future__ import annotations
import argparse
import contextlib
import hashlib
import io
import json
import os
import re
import shutil
import sys
import tempfile
import warnings
import zipfile
from collections.abc import Iterable, Mapping, Sequence
from dataclasses import dataclass, field
from pathlib import Path, PurePosixPath
from typing import Any
TOOLS = Path(__file__).resolve().parent
REPO = TOOLS.parent
if str(TOOLS) not in sys.path:
sys.path.insert(0, str(TOOLS))
import okf_witness as witness # noqa: E402
FIXTURES = REPO / "tests" / "fixtures" / "accounting"
CORPUS = FIXTURES / "corpus"
REJECTED = FIXTURES / "rejected"
INVENTORY = FIXTURES / "inventory.json"
REJECTED_INVENTORY = FIXTURES / "rejected-inventory.json"
STS_FIXTURE = CORPUS / "prosess-84-sts.xml"
STS_TWIN = FIXTURES / "witness" / "prosess-84-sts.twin.json"
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"
ACCOUNTING_FLAG = "--accounting"
ACCOUNTING_VERSION = 1
BUNDLE_ID = "accounting-gate"
OKF_VERSION = "0.2"
CONSUME_QUESTION = "Hvilken toleranseklasse gjelder for konstruksjoner av betong?"
GREEN = "GREEN"
RED = "RED"
SKIPPED = "SKIPPED"
DIAGNOSTIC = "DIAGNOSTIC"
#: Exceptions the operator has APPROVED, by (suffix, element), with the date.
#: Approving one does not move a number here: the witness counts no such
#: element in the first place, which is the whole reason the exception exists.
#: What approval changes is that the gap is now a stated limit of this
#: instrument rather than an open question about the build.
APPROVED_EXCEPTIONS: frozenset[tuple[str, str]] = frozenset(
{(".pdf", "heading"), (".pdf", "paragraph"), (".pdf", "table")}
)
#: When, and by whom.
APPROVED_ON = "2026-09-17, operator"
#: Exceptions this gate PROPOSES. Listed in every run; none of them is applied.
PROPOSED_EXCEPTIONS: tuple[dict[str, str], ...] = (
{
"suffix": ".xlsx",
"element": "image",
"reason": "the converter writes one pipe table per sheet and a pointer "
"block inside it would break source_rows (README, 0.10.0)",
"carried_instead": "nothing; the image is absent from the bundle",
},
{
"suffix": ".md .txt .csv .json .odt .rtf",
"element": "image",
"reason": "no reader for these types carries image bytes (README, 0.10.0)",
"carried_instead": "the reference text as written, if the format has one",
},
)
#: 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
name: str
k: int
m: int
status: str
reason: str
details: list[str] = field(default_factory=list)
@property
def fails(self) -> bool:
return self.status == RED and self.number <= 6
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:
return Row(number, name, k, m, GREEN if m > 0 and k == m else RED, reason, details)
# --- inputs ------------------------------------------------------------------
def readme_types(readme: Path = README) -> list[str]:
"""T: the rows of README's supported-file-types table."""
text = readme.read_text(encoding="utf-8")
section = text.split("## Supported file types", 1)[1].split("\n## ", 1)[0]
return sorted(set(re.findall(r"^\| `(\.[a-z0-9]+)` \|", section, re.MULTILINE)))
def load_inventory(path: Path) -> dict[str, Any]:
data: dict[str, Any] = json.loads(path.read_text(encoding="utf-8"))
return data
def door_available() -> bool:
"""Does `okf build` accept the accounting flag?"""
from llm_ingestion_okf import cli
argv = ["build", "in", "--bundle", "out", "--bundle-id", "x", "--okf-version", "0.2"]
with contextlib.redirect_stderr(io.StringIO()):
try:
cli.parse_args([*argv, ACCOUNTING_FLAG, "accounting.json"])
except SystemExit:
return False
return True
@dataclass
class Build:
"""What one `okf build` run left behind, read back from the artifacts.
`bundle_text` is every concept BODY the run wrote, joined. It is what
makes this gate a judge rather than a calculator: a booking that says an
element was carried is checked against these bytes, not against the
number beside it.
"""
exit_code: int
log: str
accounting: dict[str, Any] | None
source_files: set[str]
assets: dict[str, str]
bundle_text: str = ""
def _frontmatter_source_file(text: str) -> str | None:
if not text.startswith("---\n"):
return None
head = text[4:].split("\n---\n", 1)[0]
match = re.search(r"^source_file:\s*(.+?)\s*$", head, re.MULTILINE)
if match is None:
return None
value = match.group(1)
if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'":
value = value[1:-1]
return value
def _body(text: str) -> str:
"""A concept file without its frontmatter block."""
if not text.startswith("---\n"):
return text
end = text.find("\n---\n", 4)
return text[end + 5 :] if end >= 0 else text
def read_bundle(bundle: Path, exit_code: int, accounting_path: Path | None) -> Build:
sources: set[str] = set()
bodies: list[str] = []
for path in sorted(bundle.rglob("*.md"), key=lambda p: p.as_posix()):
if "assets" in path.relative_to(bundle).parts:
continue
text = path.read_text(encoding="utf-8")
found = _frontmatter_source_file(text)
if found:
sources.add(found)
bodies.append(_body(text))
assets_dir = bundle / "assets"
assets = (
{p.name: _sha256(p) for p in sorted(assets_dir.iterdir()) if p.is_file()}
if assets_dir.is_dir()
else {}
)
log_path = bundle / "log.md"
accounting = None
if accounting_path is not None and accounting_path.is_file():
accounting = json.loads(accounting_path.read_text(encoding="utf-8"))
return Build(
exit_code=exit_code,
log=log_path.read_text(encoding="utf-8") if log_path.is_file() else "",
accounting=accounting,
source_files=sources,
assets=assets,
bundle_text="\n".join(bodies),
)
def run_build(corpus: Path, workdir: Path, *, door: bool, gate: str | None = None) -> Build:
"""Run the real `okf build` in-process and read back what it wrote."""
from llm_ingestion_okf import cli
bundle = workdir / "bundle"
accounting_path = workdir / "accounting.json" if door else None
argv = [
"build",
str(corpus),
"--bundle",
str(bundle),
"--bundle-id",
BUNDLE_ID,
"--okf-version",
OKF_VERSION,
]
if gate is not None:
argv += ["--gate", gate]
if accounting_path is not None:
argv += [ACCOUNTING_FLAG, str(accounting_path)]
sink = io.StringIO()
with (
contextlib.redirect_stdout(sink),
contextlib.redirect_stderr(sink),
warnings.catch_warnings(),
):
warnings.simplefilter("ignore")
try:
code = cli.main(argv)
except SystemExit as exc:
code = exc.code if isinstance(exc.code, int) else 2
return read_bundle(bundle, code, accounting_path)
def _sha256_bytes(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def _sha256(path: Path) -> str:
return _sha256_bytes(path.read_bytes())
def _sha12(path: Path) -> str:
return _sha256(path)[:12]
def asset_holds(build: Build, source: Path) -> bool:
"""Did the run carry THESE bytes, placed under their own content address?
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 any(
found == digest and name.startswith(digest[:12]) for name, found in build.assets.items()
)
# --- the judge's own reading of the bundle ------------------------------------
#: Every rejection code this build can emit, read off the package's own source
#: 2026-09-18, plus the guard's two dispositions. A code outside this list is
#: RED: a report gets to say WHY it dropped something, not to invent the
#: vocabulary it says it in.
REJECTION_CODES: frozenset[str] = frozenset(
{
"asset_collision",
"asset_pdf_unbounded",
"asset_pdf_unsupported",
"asset_remote",
"asset_samples_invalid",
"asset_size_invalid",
"asset_too_large",
"asset_type_unknown",
"asset_unresolved",
"extractor_binary_missing",
"extractor_binary_version",
"extractor_convert_error",
"extractor_decode_error",
"extractor_empty_conversion",
"extractor_empty_csv",
"extractor_empty_pdf",
"extractor_extra_missing",
"extractor_id",
"extractor_ocr_group_missing",
"extractor_pdf_error",
"extractor_unknown",
"extractor_version",
"extractor_xml_doctype",
"extractor_xml_parse_error",
"fail_secure",
"inbox_gate",
"inbox_slug_collision",
"inbox_slug_empty",
"inbox_slug_too_long",
"inbox_source_file_invalid",
"inbox_source_file_unaddressable",
"inbox_source_title_unaddressable",
"inbox_title_invalid",
"inventory_error",
"inventory_unreadable",
"quarantine_review",
}
)
_NOT_ALNUM = re.compile(r"[\W_]+")
#: A converter attribute block (`{.mark}`, `{#slide-1}`). Its letters stand
#: between words that WERE carried, so it is removed from the bundle text
#: before comparing -- the same allowance the build makes, and no looser.
_CONVERTER_ATTRIBUTE = re.compile(r"\{[#.][^{}\n]*\}")
def _norm(text: str) -> str:
return _NOT_ALNUM.sub("", text.casefold())
class Finder:
"""Is this piece of the SOURCE in the text the bundle holds?
Searched forward from the last hit first, because a reader keeps the
source's order; a miss there falls back to the whole text, so an element
that moved is still found. The gate owns this code: borrowing the build's
own finder would make the judge agree with the judged by construction.
"""
def __init__(self, text: str) -> None:
self.text = _norm(_CONVERTER_ATTRIBUTE.sub("", text))
self.cursor = 0
def __call__(self, piece: str) -> bool:
needle = _norm(piece)
if not needle:
return False
at = self.text.find(needle, self.cursor)
if at < 0:
at = self.text.find(needle)
if at < 0:
return False
self.cursor = at + len(needle)
return True
# --- accounting --------------------------------------------------------------
@dataclass
class Unit:
"""One inventoried thing: a document, or an inbox file that is not one.
`unaccounted` and `double` are about the NUMBERS; `unverified` and
`invalid` are about the bundle and the declaration themselves. A booking
the gate could not verify is never clean, and `verified`/`unverifiable`
carry the denominator behind that word.
`refused` is the fifth and the only one that is not a defect in the
REPORT: the elements of a document the build refused whole. Their fate is
honestly declared and their content is not in the bundle, so the numbers
balance and nothing else here can see the loss (H1).
"""
name: str
kind: str
unaccounted: int
double: int
unverified: int = 0
invalid: int = 0
verified: int = 0
unverifiable: int = 0
refused: int = 0
notes: list[str] = field(default_factory=list)
@property
def clean(self) -> bool:
return not (
self.unaccounted or self.double or self.unverified or self.invalid or self.refused
)
def _conservation_held(build: Build) -> bool:
return build.exit_code == 0 and "K1b FAILED" not in build.log
def _image_proof(
images: Iterable[Mapping[str, Any]], build: Build, corpus: Path, finder: Finder
) -> tuple[int, int, int]:
"""(bytes proved in `assets/`, references found as text, images the gate
cannot check).
An image element carries no text of its own, so the only proof it was
carried is the asset -- and the only proof a POINTER survived is the
reference standing in the bundle. An image embedded in a binary container
has no source file to hash, so the gate says it cannot check it rather
than passing it.
"""
proved = pointed = blind = 0
for image in images:
kind = image.get("kind")
target = image.get("target")
ref = str(image.get("ref") or "")
names = [n for n in (ref, PurePosixPath(ref).name if ref else "") if n]
if target:
names.append(PurePosixPath(target).name)
if asset_holds(build, corpus / target):
proved += 1
elif kind == witness.EMBEDDED:
blind += 1
if any(finder(name) for name in names):
pointed += 1
return proved, pointed, blind
def _document_unit(
name: str,
entry: Mapping[str, Any],
declared: Mapping[str, Any] | None,
build: Build,
corpus: Path,
version_ok: bool,
) -> Unit:
"""One document's account, checked against the bundle the run wrote."""
elements: dict[str, int] = dict(entry["elements"])
texts: Mapping[str, list[str]] = entry.get("texts", {})
images: list[Mapping[str, Any]] = list(entry.get("images", []))
total = sum(elements.values())
if declared is None:
return Unit(name, "document", total, 0, notes=["no declared fates"])
if not version_ok:
return Unit(
name,
"document",
total,
0,
invalid=1,
notes=[
f"accounting_version is not {ACCOUNTING_VERSION}: the declaration is unreadable"
],
)
fates: Mapping[str, Any] = declared.get("fates", {})
persisted = name in build.source_files
unaccounted = double = unverified = invalid = verified = unverifiable = 0
notes: list[str] = []
status = declared.get("status")
code = declared.get("code")
if status == "persisted" and not persisted:
invalid += 1
notes.append("declared persisted; no concept in the bundle names this document")
if status == "rejected" and persisted:
invalid += 1
notes.append("declared rejected; the bundle holds a concept from this document")
if code is not None and code not in REJECTION_CODES:
invalid += 1
notes.append(f"document code `{code}` is not one this gate knows")
finder = Finder(build.bundle_text)
booked_carried = 0
for element in sorted(set(elements) | set(fates)):
fate: Mapping[str, Any] = fates.get(element, {})
carried = int(fate.get("carried", 0))
pointer = int(fate.get("pointer", 0))
rejected = {str(k): int(v) for k, v in (fate.get("rejected") or {}).items()}
have = elements.get(element, 0)
booked_carried += max(carried, 0) + max(pointer, 0)
if min([carried, pointer, *rejected.values()], default=0) < 0:
invalid += 1
notes.append(f"{element}: a negative booking {json.dumps(fate, sort_keys=True)}")
unknown = sorted(c for c in rejected if c not in REJECTION_CODES)
if unknown:
invalid += 1
notes.append(
f"{element}: rejection code(s) {', '.join(unknown)} outside the closed list"
)
booked = carried + pointer + sum(rejected.values())
if booked < have:
unaccounted += have - booked
notes.append(f"{element}: {booked} booked of {have}")
elif booked > have:
double += booked - have
notes.append(f"{element}: {booked} booked, source has {have}")
# 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:
invalid += 1
notes.append(
f"{element}: {want} booked carried, but the bundle holds nothing from here"
)
if element == "image":
proved, pointed, blind = _image_proof(images, build, corpus, finder)
reach = min(carried, proved) + min(pointer, pointed)
verified += reach
short = want - reach
if short > 0:
take = min(short, blind)
unverifiable += take
if short > take:
unverified += short - take
notes.append(
f"image: {want} booked, {proved} asset(s) with the source's bytes and "
f"{pointed} reference(s) found in the bundle"
)
continue
# An element is made of PIECES, and it is carried only if EVERY one of
# them is in the bundle: a reader writes its own markers between the
# parts of a container, so a section is never one contiguous run.
elements_pieces = [
[piece for piece in element_pieces if _norm(piece)]
for element_pieces in texts.get(element, [])
]
sayable = [pieces for pieces in elements_pieces if pieces]
found = sum(1 for pieces in sayable if all(finder(piece) for piece in pieces))
verified += min(found, want)
if want > found:
short = want - found
take = min(short, len(elements_pieces) - len(sayable))
unverifiable += take
if short > take:
unverified += short - take
notes.append(
f"{element}: {want} booked carried, {found} of {len(sayable)} "
"found in the bundle"
)
if persisted and total > 0 and booked_carried == 0:
invalid += 1
notes.append("the build persisted this document and the report carries nothing from it")
# H1: a document refused whole balances by construction -- every element
# is a coded rejection, so u = 0 and d = 0 -- and `refused_whole` below
# asks its question only for a corpus that persisted NOTHING. One refused
# source beside an accepted one is the ordinary case on a heterogeneous
# corpus, and it read CLEAN with the content gone.
refused = 0
if status == "rejected" and not persisted and total > 0:
refused = total
notes.append(
f"refused whole: {total} element(s) declared rejected `{code}`, "
"and the bundle holds nothing from this document"
)
return Unit(
name,
"document",
unaccounted,
double,
unverified=unverified,
invalid=invalid,
verified=verified,
unverifiable=unverifiable,
refused=refused,
notes=notes,
)
def _file_unit(
name: str,
entry: Mapping[str, Any],
declared: Mapping[str, Any] | None,
build: Build,
corpus: Path,
version_ok: bool,
) -> Unit:
"""One inbox file that is not a document the build reads."""
pointed_by = entry["pointed_at_by"]
# Bytes in assets/ prove a carry only for a file a document points at:
# an unpointed file with the same bytes (R761 ships 8 such duplicates)
# was not carried through anything.
carried = bool(pointed_by) and asset_holds(build, corpus / name)
merged = name in build.source_files
notes: list[str] = []
invalid = 0
if declared is not None and not version_ok:
return Unit(
name,
"file",
1,
0,
invalid=1,
notes=[
f"accounting_version is not {ACCOUNTING_VERSION}: the declaration is unreadable"
],
)
false_claim = False
if declared is not None:
fate = declared.get("fate")
code = declared.get("code")
rejected = fate == "rejected"
false_claim = fate == "carried" and not carried
if false_claim:
notes.append("declared carried; no asset holds this file's bytes under its own name")
if code is not None and code not in REJECTION_CODES:
invalid += 1
notes.append(f"file code `{code}` is not one this gate knows")
else:
# K1b: every walked file is merged or a coded rejection, so a file
# that is not merged was booked as a rejection.
rejected = not merged and _conservation_held(build)
fates = sum((carried, merged, rejected))
if carried and rejected:
notes.append(f"carried via {', '.join(pointed_by) or 'a document'} AND rejected")
unaccounted = 1 if fates == 0 or false_claim else 0
return Unit(
name,
"file",
unaccounted,
max(0, fates - 1),
invalid=invalid,
verified=1 if carried else 0,
notes=notes,
)
def account(inventory: Mapping[str, Any], build: Build, corpus: Path) -> list[Unit]:
"""Give every inventoried element and file its fate, and CHECK it."""
declared_docs: dict[str, Any] = {}
declared_files: dict[str, Any] = {}
version_ok = True
if build.accounting is not None:
declared_docs = {d["source_file"]: d for d in build.accounting.get("documents", [])}
declared_files = {f["source_file"]: f for f in build.accounting.get("files", [])}
version_ok = build.accounting.get("accounting_version") == ACCOUNTING_VERSION
units: list[Unit] = []
for name, entry in sorted(inventory["documents"].items()):
units.append(
_document_unit(name, entry, declared_docs.get(name), build, corpus, version_ok)
)
for name, entry in sorted(inventory["files"].items()):
units.append(_file_unit(name, entry, declared_files.get(name), build, corpus, version_ok))
return units
# --- rows --------------------------------------------------------------------
def row1(table: Iterable[str], inventory: Mapping[str, Any], fresh: Mapping[str, Any]) -> Row:
table = sorted(table)
covered: set[str] = set()
stale: list[str] = []
for name, entry in inventory["documents"].items():
if fresh["documents"].get(name) == entry:
covered.add(entry["suffix"])
else:
stale.append(name)
missing = [t for t in table if t not in covered]
k = len(table) - len(missing)
reason = "every README type has a fixture with a reproducible witness count"
if missing:
reason = f"no fixture fasit for {', '.join(missing)}"
details = [f"committed fasit differs from a fresh witness count: {n}" for n in stale]
return _row(1, "file types with a fasit fixture", k, len(table), reason, details)
def row2(table: Iterable[str], inventory: Mapping[str, Any], build: Build, door: bool) -> Row:
table = sorted(table)
name = "source inventory before build"
if not door:
return _row(
2, name, 0, len(table), f"`okf build` has no `{ACCOUNTING_FLAG}` door; no inventory", []
)
declared = {d["source_file"]: d for d in (build.accounting or {}).get("documents", [])}
good: list[str] = []
details: list[str] = []
for suffix in table:
docs = [n for n, e in inventory["documents"].items() if e["suffix"] == suffix]
ok = bool(docs)
for doc in docs:
got = declared.get(doc, {}).get("inventory")
want = inventory["documents"][doc]["elements"]
if got != want:
ok = False
details.append(f"{doc}: declared {got}, witness {want}")
if ok:
good.append(suffix)
missing = [t for t in table if t not in good]
reason = (
"every type's inventory equals the witness"
if not missing
else (f"inventory absent or wrong for {', '.join(missing)}")
)
return _row(2, name, len(good), len(table), reason, details)
def _tally(units: Iterable[Unit]) -> str:
"""What the gate FOUND, with the denominator beside it."""
units = list(units)
verified = sum(u.verified for u in units)
unverified = sum(u.unverified for u in units)
blind = sum(u.unverifiable for u in units)
return (
f"{verified} carried element(s) found in the bundle, {unverified} claimed and not found, "
f"{blind} carrying no text of their own (the gate cannot check those)"
)
def row3(units: list[Unit], door: bool) -> Row:
clean = sum(1 for u in units if u.clean)
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)
invalid = sum(u.invalid for u in units)
refused = sum(u.refused for u in units)
documents = sum(1 for u in units if u.kind == "document")
refused_docs = sum(1 for u in units if u.refused)
reason = (
f"u = {u_total} unaccounted, d = {d_total} double-booked, "
f"{unverified} booked carried and not in the bundle, {invalid} declaration(s) the gate refuses, "
f"{refused} element(s) lost with {refused_docs} of {documents} document(s) refused whole"
)
if not door:
reason += f"; no `{ACCOUNTING_FLAG}` door, so no element has a declared fate"
details = [_tally(units)] + [
f"{u.kind} {u.name}: u={u.unaccounted} d={u.double} "
f"unverified={u.unverified} invalid={u.invalid} refused={u.refused}"
+ (f" ({'; '.join(u.notes)})" if u.notes else "")
for u in units
if not u.clean
]
return _row(
3,
"accounting after build (u = 0, d = 0, and every carried element found)",
clean,
len(units),
reason,
details,
)
def row4(inventory: Mapping[str, Any], build: Build) -> Row:
name = "a rejected document is reported honestly"
rejected = [n for n in sorted(inventory["documents"]) if n not in build.source_files]
if not rejected:
return _row(4, name, 0, 0, "the fixture was not rejected; the row cannot judge", [])
declared_images = sum(len(e["images"]) for e in inventory["documents"].values())
found = re.search(r"\*\*Images\*\*: (\d+) carried of (\d+) found", build.log)
good = 0
details: list[str] = []
for doc in rejected:
total = sum(inventory["documents"][doc]["elements"].values())
line = re.compile(
rf"{re.escape(doc)}: {total} elements found in the source, 0 carried: "
r"document rejected `[a-z_]+`"
)
problems = []
if not line.search(build.log):
problems.append(f"no line '{doc}: {total} elements found in the source, 0 carried'")
if found is None:
problems.append("no Images bullet")
elif int(found.group(2)) != declared_images:
problems.append(
f"log says {found.group(1)} carried of {found.group(2)} found; "
f"the source declares {declared_images}"
)
if problems:
details.extend(f"{doc}: {p}" for p in problems)
else:
good += 1
reason = (
"every rejected document names what it held"
if good == len(rejected)
else (f"{len(rejected) - good} rejected document(s) reported as if they held less")
)
return _row(4, name, good, len(rejected), reason, details)
def compare(left: Mapping[str, int] | None, right: Mapping[str, int] | None) -> list[str]:
"""Disagreements between two witnesses, both numbers kept."""
if left is None or right is None:
return ["a witness is unavailable"]
return [
f"{e}: {left[e]} vs {right[e]}"
for e in sorted(set(left) & set(right))
if left[e] != right[e]
]
def row5(pairs: list[tuple[str, list[str]]], notes: list[str]) -> Row:
good = sum(1 for _, problems in pairs if not problems)
details = [f"{label}: {'; '.join(p)}" for label, p in pairs if p] + notes
reason = (
"both witnesses agree on every element"
if good == len(pairs)
else (f"{len(pairs) - good} pair(s) disagree")
)
return _row(5, "two witnesses agree", good, len(pairs), reason, details)
def _counts(count: witness.Count | None) -> dict[str, int] | None:
return None if count is None else dict(count.counts)
def witness_pairs(r761: Path | None) -> tuple[list[tuple[str, list[str]]], list[str]]:
sts_xml, _, _ = witness.count_sts_xml(STS_FIXTURE.read_bytes())
pairs = [
(
"sts fixture (xml | json)",
compare(_counts(sts_xml), _counts(witness.count_sts_json(STS_TWIN.read_bytes()))),
),
(
"pdf fixture (pdfplumber | poppler)",
compare(
_counts(witness.pdf_objects(PDF_FIXTURE)),
_counts(witness.pdf_poppler(PDF_FIXTURE)),
),
),
]
notes: list[str] = []
if r761 is None or not r761.is_dir():
notes.append("R761 pairs not measured: source missing")
return pairs, notes
with zipfile.ZipFile(r761 / R761_ZIP) as archive:
xml_name = next(n for n in archive.namelist() if n.endswith(".xml"))
r761_xml, _, _ = witness.count_sts_xml(archive.read(xml_name))
pairs.append(
(
"R761 sts (xml | json)",
compare(
_counts(r761_xml),
_counts(witness.count_sts_json((r761 / R761_JSON).read_bytes())),
),
)
)
pdf = r761 / R761_PDF
pairs.append(
(
"R761 pdf (pdfplumber | poppler)",
compare(_counts(witness.pdf_objects(pdf)), _counts(witness.pdf_poppler(pdf))),
)
)
return pairs, notes
#: 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")
@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
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()
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"{corpus.label}: {len(documents)} document(s), {elements} element(s), "
f"{len(files)} other file(s) ({pointed} pointed at, {len(files) - pointed} not)"
)
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"
)
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"refused={unit.refused} "
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:
"""Diagnostic only: a question through `okf consume` on the fixture bundle."""
from llm_ingestion_okf import cli
bundle = workdir / "bundle"
out = workdir / "payload.json"
sink = io.StringIO()
with contextlib.redirect_stdout(sink), contextlib.redirect_stderr(sink):
try:
code = cli.main(
["consume", str(bundle), "--question", CONSUME_QUESTION, "--out", str(out)]
)
except SystemExit as exc:
code = exc.code if isinstance(exc.code, int) else 2
excerpts = 0
if out.is_file():
excerpts = len(json.loads(out.read_text(encoding="utf-8")).get("excerpts", []))
return Row(
7,
"question set via okf consume (diagnostic)",
excerpts,
0,
DIAGNOSTIC,
f"exit {code}, {excerpts} excerpt(s) for {CONSUME_QUESTION!r}",
)
# --- the run -----------------------------------------------------------------
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)
fresh = witness.witness_inbox(CORPUS)
door = door_available()
rows = [row1(table, inventory, fresh)]
with tempfile.TemporaryDirectory() as tmp:
work = Path(tmp) / "corpus"
build = run_build(CORPUS, work, door=door)
rows.append(row2(table, inventory, build, door))
rows.append(row3(account(inventory, build, CORPUS), door))
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, n200, ci))
if consume:
rows.append(row7(work))
return rows
def render(rows: list[Row]) -> str:
lines = ["row | k of M | status | reason"]
for row in rows:
count = "n/a" if row.status == DIAGNOSTIC else f"{row.k} of {row.m}"
lines.append(f"{row.number} {row.name} | {count} | {row.status} | {row.reason}")
lines.extend(f" - {d}" for d in row.details)
lines += ["", "exceptions (PROPOSED, NOT APPROVED -- none lowers a denominator):"]
for item in PROPOSED_EXCEPTIONS:
lines.append(
f" - {item['suffix']} {item['element']}: {item['reason']}; "
f"carried instead: {item['carried_instead']}"
)
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}: {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):",
]
for suffix in sorted(witness.NOT_COUNTED):
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("")
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"
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(
"--r761",
type=Path,
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,
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 = {
"rows": [r.to_json() for r in rows],
"exceptions": {
"approved": sorted(APPROVED_EXCEPTIONS),
"proposed": PROPOSED_EXCEPTIONS,
},
"gate": RED if any(r.fails for r in rows) else GREEN,
}
print(json.dumps(payload, indent=2, ensure_ascii=False))
else:
print(render(rows), end="")
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. Measured against the DEFAULT
# sources, never against the arguments: a row is SKIPPED exactly when the
# corpora the arguments name are absent, so asking the arguments made this
# branch unreachable (H5). Pointing `--r761` at nothing on a machine that
# holds R761 is the case it exists for.
machine = real_corpora(R761_DEFAULT, N200_DEFAULT)
for row in rows:
if row.status == SKIPPED and any(c.available for c in machine):
print(
f"okf-accounting-gate: row {row.number} was skipped while its source exists",
file=sys.stderr,
)
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())
Reference in a new issue View git blame Copy permalink