open/llm-ingestion-okf
1
0
Fork 0
Code Issues Releases 4 Activity
llm-ingestion-okf/tools/okf_accounting_gate.py
Kjell Tore Guttormsen 579cb1c83c
 refactor(assets,accounting): drop two checks no mutant can fell, add the delta cut 
160 passed over the three affected files. Two guards this round introduced are
removed, each because it was measured and found unkillable, and one real cut is
added in their place.

The end-of-line anchor on the judge's clause expression: with the pointer-block
route and the asset-name tie already in place, the only case it covered is a
label forging a clause inside a block that also carries a real one -- which
`assets._inline` now makes unreachable from a document. Removed with `$` gone
the whole suite stays green, which is the measurement: a guard nothing can fell
is not a guard, and it would also have made the mutant below unfaithful.

The length check on a short absolute run in `_bmp_rle8_rows`: unobservable by
construction. A run shorter than its declared count means the stream ran out,
so the loop cannot reach an end-of-bitmap afterwards and the picture is refused
either way -- the partial paint is discarded with the frame.

A FIFTH CUT REPLACES THEM, and it makes an existing bound load-bearing: a
stream cut in the middle of a delta escape, 17 of 32 pixels wrong by this
file's own decoder. Without the two-byte bound in that branch the cut raises
IndexError instead of a coded refusal.

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

1385 lines
54 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]
#: One POINTER BLOCK, as `assets.render_block` writes it: the markdown image
#: line naming a file under `assets/`, and the detail line directly under it.
#: Restated here rather than imported, for the reason `asset_holds` gives about
#: the naming rule -- a judge sharing the judged's own expression agrees with
#: it by construction. It is deliberately the SHAPE and not the readable tail.
_POINTER = re.compile(
r"^!\[[^\]\n]*\]\(/assets/(?P<asset>[^)\s]+)\)\n(?P<detail>Image: [^\n]*)$",
re.MULTILINE,
)
#: The conversion clause `assets.render_block` writes: the source's media type
#: and sha256, and the media type and sha256 of what the run actually carried.
#: One expression, so the judge has one definition of the claim it verifies.
#: It is deliberately NOT anchored to the end of its line: an anchor there
#: would be a third check on a case the other two already close, and a guard
#: no mutant can fell is not a guard -- measured, removing it leaves every
#: test in this suite green.
_CONVERSION = re.compile(
r"converted from \S+ sha256:(?P<before>[0-9a-f]{64}) to \S+ sha256:(?P<after>[0-9a-f]{64})"
)
def _conversions(bundle_text: str) -> dict[str, str]:
"""source digest -> the digest the bundle says it carried instead.
READ ONLY FROM A POINTER BLOCK THIS BUILD WROTE, and only where the clause
names the asset that block points at. The expression used to run over the
whole bundle text, which put an untrusted document inside the judge's own
input: measured by PM 2026-09-19, a BMP refused `asset_too_large` and never
carried was reported as held, both through an image's alt text and through
ordinary body text, because the document simply wrote the sentence. The
gate's first sentence is that the fasit never comes from the reader it
judges, and this route had quietly stopped obeying it.
The other half of the boundary is in the build, not here: a label is
document text written INSIDE a pointer block, so `assets._inline` is what
keeps it from emitting this grammar. Neither half is sufficient alone --
this one cannot tell a label from a field, and that one does not reach
body text or a table cell at all.
"""
found: dict[str, str] = {}
for pointer in _POINTER.finditer(bundle_text):
clause = _CONVERSION.search(pointer.group("detail"))
if clause is None:
continue
after = clause.group("after")
# The claim has to be about the picture the block points at. A clause
# standing in one asset's block while naming another's digest is a
# sentence nothing in this build writes.
if pointer.group("asset").startswith(after[:12]):
found[clause.group("before")] = after
return found
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.
A SECOND ROUTE, for an image the build CONVERTS. Since the viewable-asset
round a source in a format no model can be shown reaches the bundle as a
PNG, so its own bytes are not in `assets/` and never will be -- measured,
the day that landed R761 went from 0 to 19 claimed-and-not-found, which is
exactly its RLE8 BMP count. The bundle states both digests on the pointer
line, and this reads them and then HASHES THE ASSET ITSELF: the claim is
accepted only when a file in `assets/` really holds the bytes the bundle
says were written. A bundle claiming a conversion it did not perform still
fails, which is the difference between reading the bundle and believing
the report.
WHAT NEITHER ROUTE PROVES IS FIDELITY. Both ask whether a file in
`assets/` holds the bytes the bundle names, and neither decodes a PIXEL:
measured 2026-09-19, a mutated converter that writes a BLANK PNG is
accepted here, because the bundle is internally consistent -- the digest
it claims to have written really is the asset's digest -- and nothing in
this gate compares that picture with the source's. The suite fells that
mutant by decoding both sides; the judge cannot. So "claiming a conversion
it did not perform" means claiming one whose FILE is missing, never one
whose pixels are wrong.
"""
digest = _sha256(source)
if any(
found == digest and name.startswith(digest[:12]) for name, found in build.assets.items()
):
return True
written = _conversions(build.bundle_text).get(digest)
return written is not None and any(
found == written and name.startswith(written[: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
)
#: The note `_document_unit` leaves when the run declared no fate at all for a
#: document -- because the build never reached the accounting door.
NO_DECLARED_FATE = "no declared fates"
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_FATE])
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 measures_no_class(units: Sequence[Unit]) -> str | None:
"""Did this corpus fail to measure ANY element class at all? (H6)
Reproduced on N200 Vegbygging:2024, 2026-09-19: `okf build` proposes 0
plans on it, prints `FAILED - no segmentation plans` and exits 2 BEFORE
the accounting door -- no accounting file is written. Every element then
lands as `u` with no declared fate, and 16 549 unaccounted reads like a
finding about the build when it is a finding about the run.
What it would take is a capability and not a threshold: `.json` is read as
generic JSON, and the publisher's STS delivery in that form would have to
reach the same markdown grammar `_extract_xml` writes for the XML one. No
such reader exists in the package -- `standardContent` occurs 0 times in
`src/` and 4 times in this gate's witness (measured 2026-09-19).
"""
documents = [u for u in units if u.kind == "document"]
if not documents:
return None
blank = [u for u in documents if NO_DECLARED_FATE in u.notes]
if len(blank) != len(documents):
return None
return (
f"{len(blank)} of {len(documents)} document(s) have no declared fate: the build "
"did not reach the accounting door, so this corpus measures no element class. "
"`.json` is read as generic JSON; an STS delivery in that form would have to "
"reach the markdown grammar the XML reader writes."
)
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}"
)
blank = measures_no_class(units)
if blank is not None:
details.append(f" gate {label}: {blank}")
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