"""Run a corpus through the whole path and report numbers, never a claim. The instrument behind K1 and K2. It exists because the alternative -- a person running the door by hand and writing down what they saw -- has already produced a number that was right about a directory that no longer existed. Three rules it enforces rather than describes: **K1b is a command.** The conservation identity `merged + Sigma(coded rejections) == N` is CHECKED here, and a run where it does not hold EXITS NON-ZERO. Asserted in prose it would be something a reader has to trust; as an exit status it fails the run that produced it. When it fails, the unaccounted files are NAMED -- "some file went missing" is not actionable. **`N` is computed, never typed.** It is the file count of the corpus directory, read at run time. A literal would keep passing after the corpus changed and would then report a fact about a directory that no longer exists. **Three counts, never one.** The guard sits between extraction and persist, so a healthy persisted count can hide a pile of quarantines. Extracted, gated and persisted are separate numbers for that reason. **The degenerate-merge rule is a DEFINITION, not a threshold: a merge is degenerate when the extracted text is zero characters after stripping whitespace.** A concept with an empty body cannot carry one unit of knowledge, so counting it as a merge would report extraction failure as success. The resolved converter path and version are printed in the output, because the vendored binary is bypassed silently otherwise -- measured three times, wheel 3.9 against host 3.10.2. """ from __future__ import annotations import argparse import json import sys import time from collections.abc import Mapping from dataclasses import dataclass, replace from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) from llm_ingestion_okf.errors import IngestError # noqa: E402 from llm_ingestion_okf.extract import extract_text # noqa: E402 from llm_ingestion_okf.inbox import GateDecision, InboxResult, process_inbox # noqa: E402 from llm_ingestion_okf.profiles import ( # noqa: E402 SEGMENTED_OKF_V0_2, STRUCTURED_V1, BundleProfile, ) from llm_ingestion_okf.segmentation import ( # noqa: E402 SegmentationPlan, parse_segmentation_plan, ) __all__ = [ "CorpusReport", "converter_identity", "load_plans", "is_degenerate", "main", "measure", "replace", "unaccounted_names", ] HARNESS_ID = "okf-corpus-run" def is_degenerate(text: str) -> bool: """Zero characters after stripping whitespace. The whole rule, in one line. A definition rather than a threshold on purpose: a threshold invites a later argument about where it should sit, and every such argument has to be had again the next time the corpus changes. """ return not text.strip() def converter_identity() -> tuple[str, str]: """The converter this run would use, resolved by path, and its version. Reported rather than assumed. `pypandoc` prefers the HIGHEST version it can find over the one this package vendored, so a run that did not say which binary produced its text would be unattributable. """ from llm_ingestion_okf._pandoc import PANDOC_VERSION, resolve_pandoc try: return (str(resolve_pandoc()), PANDOC_VERSION) except IngestError as exc: return (f"unresolved ({exc.code})", PANDOC_VERSION) def unaccounted_names( *, dropped: tuple[str, ...], merged: tuple[str, ...], coded: tuple[str, ...] ) -> tuple[str, ...]: """Every dropped file that is in neither column, in sorted order. The conservation check, isolated so it can be driven with an inventory the door could not produce. A harness whose failure path is unreachable is a harness that proves nothing when it passes. """ return tuple(sorted(set(dropped) - set(merged) - set(coded))) @dataclass(frozen=True) class CorpusReport: """One corpus run's numbers, every one of them with its denominator.""" corpus: str ingested_at: str n: int extracted: int gated: int persisted: int substantive: int degenerate: int rejected: int seconds_total: float converter_path: str converter_version: str codes: tuple[tuple[str, int], ...] unaccounted: tuple[str, ...] @property def merged(self) -> int: return self.substantive + self.degenerate def render(self) -> str: per_file = self.seconds_total / self.n if self.n else 0.0 lines = [ f"# Corpus run: {self.corpus}", "", f"N (denominator, the directory's file count) = {self.n}", "", "## Three counts, never one", "", "The guard sits between extraction and persist, so a healthy persisted", "count can hide a pile of quarantines.", "", f"- extracted: {self.extracted}/{self.n}", f"- gated: {self.gated}/{self.n}", f"- persisted: {self.persisted}/{self.n}", "", "## The numerator, split", "", "A merge is degenerate when the extracted text is zero characters after", "stripping whitespace -- a definition, not a threshold.", "", f"- substantive: {self.substantive}/{self.n}", f"- degenerate: {self.degenerate}/{self.n}", f"- rejected (coded): {self.rejected}/{self.n}", "", f"merged + coded rejections = {self.merged + self.rejected}; N = {self.n}", "", "## Converter", "", f"- resolved converter path: {self.converter_path}", f"- pinned converter version: {self.converter_version}", "", "## Wall time", "", f"- total: {self.seconds_total:.2f} s", f"- per file: {per_file:.3f} s", "", "## Rejection codes", "", ] lines.extend( f"- `{code}`: {count}/{self.n}" for code, count in self.codes or (("(none)", 0),) ) if self.unaccounted: lines += ["", "## UNACCOUNTED", ""] lines.extend(f"- {name}" for name in self.unaccounted) return "\n".join(lines) + "\n" def render_log(self) -> str: """The bundle's own `log.md`, in SPEC section 9 form. Written because a consumer measured that K1b was NOT checkable from the bundle: `merged` is countable from the concepts, `N` is not, so the conservation identity could only be taken on trust from a report that does not travel with the artifact. Section 9 already reserves this file for the history of a scope, and the denominator is the one fact about this run that the bundle cannot otherwise recover. Dated from `ingested_at`, never the wall clock: determinism here is bit-exact, and a date that moved between two replays of the same corpus would put a changing byte in an artifact that must not change. """ codes = self.codes or (("(none)", 0),) rejections = ", ".join(f"`{code}`: {count}" for code, count in codes) lines = [ "---", "type: Log", "title: Corpus run history", "---", "", "# Corpus run history", "", f"## {self.ingested_at[:10]}", "", f"* **Ingested**: {self.corpus} — N = {self.n} " f"(the corpus directory's file count, computed at run time), " f"merged = {self.merged} ({self.substantive} substantive, " f"{self.degenerate} degenerate), coded rejections = {self.rejected}.", f"* **Rejected**: {rejections}.", f"* **Conservation (K1b)**: merged + coded rejections = " f"{self.merged} + {self.rejected} = {self.merged + self.rejected}; " f"N = {self.n}. The run exits non-zero when these differ.", f"* **Converter**: {self.converter_path}, version {self.converter_version}.", ] if self.unaccounted: lines.append("* **Unaccounted**: " + ", ".join(self.unaccounted) + " — K1b FAILED.") return "\n".join(lines) + "\n" def load_plans(plans_dir: Path) -> dict[str, SegmentationPlan]: """Every proposal artifact in a directory, keyed by filename. The key is for the operator, never for selection: `process_inbox` matches a plan to a drop by the source content hash, so a renamed document still finds its plan and a plan filed under the wrong name still cannot be applied to the wrong bytes. A directory with no artifacts raises rather than returning an empty mapping. An empty mapping is indistinguishable from "no plans were asked for", and the run would then report a flat bundle as a success -- the exact silent skip that produced a corpus with zero `adjudication` keys. """ files = sorted(plans_dir.glob("*.json")) if not files: raise IngestError( f"no segmentation plans in {plans_dir} -- a run asked to replay plans and " "given none would build a flat bundle and report it as a success", code="segmentation_plan_invalid", ) return { path.name: parse_segmentation_plan(json.loads(path.read_text(encoding="utf-8"))) for path in files } def _gate(text: str) -> GateDecision: return GateDecision(sanitized_text=text, disposition="warn") def _split_merges(corpus: Path, result: InboxResult) -> tuple[int, int]: """Merged files split into substantive and degenerate, by the stated rule. Re-extracted here rather than read back off the bundle: the rule is about the EXTRACTED text, and a concept body has already been through the gate. """ substantive = 0 degenerate = 0 for item in result.persisted: source = corpus / item.source_file try: text = extract_text(source.name, source.read_bytes()) except (IngestError, OSError): continue if is_degenerate(text): degenerate += 1 else: substantive += 1 return (substantive, degenerate) def measure( corpus: Path, bundle: Path, *, ingested_at: str, plans: Mapping[str, SegmentationPlan] | None = None, profile: BundleProfile = STRUCTURED_V1, root_frontmatter_values: Mapping[str, str] | None = None, ) -> CorpusReport: """Run the corpus through the door and count what happened. Keyword-only with defaults, so the flat call that produced the published K1/K2 numbers stays source-compatible and byte-identical. """ dropped = tuple(sorted(path.name for path in corpus.iterdir() if path.is_file())) started = time.monotonic() result = process_inbox( corpus, bundle, ingested_at, okf_type="reference", gate=_gate, profile=profile, root_frontmatter_values=root_frontmatter_values, segmentations=plans, ) elapsed = time.monotonic() - started merged_names = tuple(item.source_file for item in result.persisted) blocked = result.quarantined + result.rejected coded_names = tuple(item.source_file for item in result.failed) + tuple( item.source_file for item in blocked ) counts: dict[str, int] = {} for failure in result.failed: counts[failure.error.code] = counts.get(failure.error.code, 0) + 1 for item in blocked: counts[item.disposition] = counts.get(item.disposition, 0) + 1 substantive, degenerate = _split_merges(corpus, result) path, version = converter_identity() return CorpusReport( corpus=str(corpus), ingested_at=ingested_at, n=len(dropped), # A file that reached the gate was extracted; the gate here persists # everything it sees, so the two differ only when a gate refuses. extracted=len(merged_names) + len(blocked), gated=len(merged_names) + len(blocked), persisted=len(merged_names), substantive=substantive, degenerate=degenerate, rejected=len(coded_names), seconds_total=elapsed, converter_path=path, converter_version=version, codes=tuple(sorted(counts.items())), unaccounted=unaccounted_names(dropped=dropped, merged=merged_names, coded=coded_names), ) def parse_args(argv: list[str] | None) -> argparse.Namespace: parser = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter ) parser.add_argument("--corpus", type=Path, required=True, help="the directory to run") parser.add_argument("--report", type=Path, required=True, help="where to write the report") parser.add_argument("--bundle", type=Path, default=None, help="where to build the bundle") parser.add_argument( "--ingested-at", default="2026-09-02T00:00:00Z", help="stamped verbatim, as everywhere" ) parser.add_argument( "--plans-dir", type=Path, default=None, help=( "directory of per-document segmentation proposals to REPLAY. Produced by " "tools/okf_propose_segments.py first, one per document; this harness never " "proposes a split of its own, because the split is a judgement and the run " "path is a deterministic replay of one" ), ) parser.add_argument( "--bundle-id", default=None, help="required with --plans-dir: what a consumer joins the bundle's concepts on", ) parser.add_argument( "--okf-version", default=None, help=( "required with --plans-dir: the upstream OKF version this bundle declares. " "An argument and never a constant -- the VALUE belongs to the catalog " "(decision E1), and a literal here would claim a decision this repository " "does not own" ), ) return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: args = parse_args(argv) if not args.corpus.is_dir(): print(f"{HARNESS_ID}: FAILED - no corpus directory at {args.corpus}", file=sys.stderr) return 2 bundle = args.bundle or args.report.parent / f"{args.corpus.name}-bundle" # Both root values or neither, checked BEFORE anything is read or written. # A segmented run that discovered a missing `bundle_id` half way through # would leave a partial bundle behind, and this library refuses half-built # bundles at every other door. plans: dict[str, SegmentationPlan] | None = None profile = STRUCTURED_V1 root_values: dict[str, str] | None = None if args.plans_dir is not None: missing = [ flag for flag, value in ( ("--bundle-id", args.bundle_id), ("--okf-version", args.okf_version), ) if value is None ] if missing: print( f"{HARNESS_ID}: FAILED - {', '.join(missing)} is required with --plans-dir; " "a profile names a key and the caller owns its value", file=sys.stderr, ) return 2 try: plans = load_plans(args.plans_dir) except (IngestError, OSError, ValueError) as exc: print(f"{HARNESS_ID}: FAILED - {exc}", file=sys.stderr) return 2 profile = SEGMENTED_OKF_V0_2 root_values = {"okf_version": args.okf_version, "bundle_id": args.bundle_id} report = measure( args.corpus, bundle, ingested_at=args.ingested_at, plans=plans, profile=profile, root_frontmatter_values=root_values, ) args.report.parent.mkdir(parents=True, exist_ok=True) args.report.write_text(report.render(), encoding="utf-8", newline="") # Into the BUNDLE, not next to the report: section 9's `log.md` is part of # the artifact a consumer receives, and a log that stayed behind in the # harness's output directory would leave the bundle exactly as unverifiable # as it was before. bundle.mkdir(parents=True, exist_ok=True) (bundle / "log.md").write_text(report.render_log(), encoding="utf-8", newline="") print(report.render()) if report.unaccounted or report.merged + report.rejected != report.n: print( f"{HARNESS_ID}: K1b FAILED - merged ({report.merged}) + coded rejections " f"({report.rejected}) != N ({report.n}). Unaccounted: " f"{', '.join(report.unaccounted) or '(none named)'}", file=sys.stderr, ) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())