"""`okf quality` -- the per-file-type verdict, and what it refuses to say. `okf check` is a CONTRACT check: it asks whether a payload carries what a claim must rest on. Measured 2026-09-10 by `vegnormal-okf`, it returned 0 findings and exit 0 on three arms over one corpus whose hit@k ranged from 6 of 6 to 0 of 6. This module is the other question -- did the cut find anything worth reading -- and its whole discipline is that it answers it PER FILE TYPE, with the denominator printed, and never answers PASS for a type it has no measurement for. Each rule below is exercised in both directions on purpose: a rule that has only ever been run against the case it fires on has not been measured, it has been asserted. The thresholds themselves, and the four evidence corpora they were read off, live in `docs/2026-09-12-g37-terskler.md`. """ from __future__ import annotations from pathlib import Path from llm_ingestion_okf import quality _FRONTMATTER = """--- type: reference title: {title} source_file: {source_file} ingested_at: 1970-01-01T00:00:00Z --- {body} """ def _bundle(root: Path, concepts: list[tuple[str, str, str]]) -> Path: """A flat bundle: one root index, one file per concept. `concepts` is `(stem, source_file, body)`. `source_file` may be the empty string, which is the shape a bundle from another producer arrives in. """ root.mkdir(parents=True, exist_ok=True) lines = ["---", "okf_version: 0.2", "bundle_id: quality-fixture", "---", ""] for stem, source_file, body in concepts: (root / f"{stem}.md").write_text( _FRONTMATTER.format(title=stem, source_file=source_file, body=body), encoding="utf-8", newline="", ) lines.append(f"- [{stem}]({stem}.md)") (root / "index.md").write_text("\n".join(lines) + "\n", encoding="utf-8", newline="") return root def _pdf_documents(count: int, *, single: int) -> list[tuple[str, str, str]]: """`count` PDF documents, `single` of them yielding one concept only.""" concepts: list[tuple[str, str, str]] = [] for index in range(count): source = f"doc-{index}.pdf" segments = 1 if index < single else 3 for segment in range(segments): concepts.append((f"doc-{index}-{segment}", source, "Body text enough to count.")) return concepts def test_a_type_at_its_measured_reference_passes(tmp_path: Path) -> None: """The known-positive: 8 of 32 is the reference, and 8 of 32 is not worse.""" root = _bundle(tmp_path / "at-reference", _pdf_documents(32, single=8)) report = quality.measure_bundle(root) row = report.row(".pdf") assert row.verdict == "PASS" assert (row.structure_null, row.documents) == (8, 32) def test_a_type_worse_than_its_reference_fails(tmp_path: Path) -> None: """The known-negative for the same rule: 9 of 32 is worse than 8 of 32.""" root = _bundle(tmp_path / "over-reference", _pdf_documents(32, single=9)) report = quality.measure_bundle(root) assert report.row(".pdf").verdict == "FAIL" assert report.exit_code == 1 def test_an_empty_concept_fails_a_type_that_has_no_threshold(tmp_path: Path) -> None: """FAIL is reachable for every type; PASS is not. `.md` has no threshold.""" root = _bundle( tmp_path / "empty-body", [("a", "note.md", "Body text."), ("b", "note.md", " \n")], ) report = quality.measure_bundle(root) row = report.row(".md") assert row.verdict == "FAIL" assert (row.empty, row.concepts) == (1, 2) def test_a_healthy_type_without_a_threshold_is_unmeasured_and_never_pass( tmp_path: Path, ) -> None: """The other direction of the same rule: nothing wrong, and still not PASS.""" root = _bundle( tmp_path / "no-threshold", [("a", "note.md", "Body text."), ("b", "note.md", "More body text.")], ) report = quality.measure_bundle(root) assert report.row(".md").verdict == "UNMEASURED" assert report.exit_code == 3 def test_concepts_without_a_source_file_are_unmeasured(tmp_path: Path) -> None: """The shape three of the four evidence corpora arrive in. `n100-2023`, `n200-2024` and `n500-2024` carry `source_file` on 0 of 446, 0 of 1 133 and 0 of 270 concepts, so a per-file-type gate has no type to speak about. Measured 2026-09-12; the gate says so instead of passing them. """ root = _bundle(tmp_path / "foreign", [("a", "", "Body text."), ("b", "", "More text.")]) report = quality.measure_bundle(root) row = report.row(quality.NO_SOURCE_FILE) assert row.verdict == "UNMEASURED" assert report.exit_code == 3 def test_a_failing_type_outweighs_a_passing_one(tmp_path: Path) -> None: concepts = _pdf_documents(32, single=9) for index in range(5): # five documents: the floor a rate needs to be a rate concepts.append((f"note-{index}-0", f"note-{index}.docx", "Body text.")) concepts.append((f"note-{index}-1", f"note-{index}.docx", "More body text.")) report = quality.measure_bundle(_bundle(tmp_path / "mixed", concepts)) assert report.row(".docx").verdict == "PASS" assert report.row(".pdf").verdict == "FAIL" assert report.exit_code == 1 def test_every_row_prints_its_denominator(tmp_path: Path) -> None: root = _bundle(tmp_path / "printed", _pdf_documents(32, single=8)) rendered = quality.measure_bundle(root).render() assert "8/32" in rendered assert "0/80" in rendered # empty concepts, over the concepts of the type assert "documents" in rendered def test_the_bundle_run_log_is_reported_when_the_bundle_carries_one(tmp_path: Path) -> None: """A rejected document leaves NO row in the bundle: the log is the only trace. On the pinned K2 bundle the log reads `N = 43, merged 39, coded rejections 4`, and the bundle itself shows 39 documents -- so the gate's own denominator is the bundle's, never the corpus's, and it says which one it is using. """ root = _bundle(tmp_path / "logged", _pdf_documents(2, single=0)) (root / "log.md").write_text( "* **Ingested**: /x - N = 3 (the corpus directory's file count, computed at " "run time), merged = 2 (2 substantive, 0 degenerate), coded rejections = 1.\n", encoding="utf-8", newline="", ) report = quality.measure_bundle(root) assert report.run_log == "N = 3, merged = 2, coded rejections = 1" assert "coded rejections = 1" in report.render() def test_a_bundle_without_a_log_says_so_rather_than_reporting_zero(tmp_path: Path) -> None: root = _bundle(tmp_path / "unlogged", _pdf_documents(2, single=0)) report = quality.measure_bundle(root) assert report.run_log is None assert "no run log" in report.render() def test_a_missing_bundle_exits_two(tmp_path: Path, capsys) -> None: # type: ignore[no-untyped-def] assert quality.main([str(tmp_path / "nowhere")]) == 2 def test_the_cli_reaches_the_gate_through_okf_quality(tmp_path: Path, capsys) -> None: # type: ignore[no-untyped-def] from llm_ingestion_okf import cli root = _bundle(tmp_path / "through-cli", _pdf_documents(32, single=9)) assert cli.main(["quality", str(root)]) == 1 assert ".pdf" in capsys.readouterr().out def test_every_threshold_names_the_corpus_and_the_denominator_it_was_read_off() -> None: """A threshold without an N is the thing this whole gate exists to refuse.""" assert quality.THRESHOLDS, "no thresholds at all would make every type UNMEASURED" for extension, threshold in quality.THRESHOLDS.items(): assert extension.startswith("."), extension assert threshold.documents >= quality.MIN_DOCUMENTS_FOR_A_THRESHOLD assert threshold.source, f"{extension} threshold names no source" assert threshold.limit_documents == threshold.documents def test_a_rate_over_too_few_documents_is_not_a_rate(tmp_path: Path) -> None: """FOUND BY RUNNING THE GATE, not by reading it. `~/repos/vegnormal-okf/build/sk2-bundle-default` is one PDF cut into 2 182 concepts. Against the 32-document reference its one-concept share is 0 of 1, which the first version of this rule read as PASS -- a verdict resting on a denominator of one, which is the exact failure `MIN_DOCUMENTS_FOR_A_THRESHOLD` exists to refuse. The floor binds the BUNDLE's denominator too, not only the threshold's. """ root = _bundle(tmp_path / "one-document", _pdf_documents(1, single=0)) assert quality.measure_bundle(root).row(".pdf").verdict == "UNMEASURED" def test_the_floor_admits_a_bundle_that_reaches_it(tmp_path: Path) -> None: """The other direction: five documents is the floor, and five is enough.""" root = _bundle(tmp_path / "five-documents", _pdf_documents(5, single=1)) assert quality.measure_bundle(root).row(".pdf").verdict == "PASS" def test_an_empty_concept_fails_even_below_the_floor(tmp_path: Path) -> None: """The empty-body bar is per concept, not a rate, so the floor does not gate it.""" root = _bundle(tmp_path / "one-document-empty", [("a", "only.pdf", " \n")]) assert quality.measure_bundle(root).row(".pdf").verdict == "FAIL" def test_the_no_source_file_row_reports_no_document_count(tmp_path: Path) -> None: """A row that is not a file type has no documents either. Every concept without the key shares the same empty `source_file`, so a naive grouping reports `documents 1` and `one-concept documents 0/1` for a bundle of 446 concepts -- three numbers that look measured and mean nothing. """ root = _bundle(tmp_path / "foreign-render", [("a", "", "Body."), ("b", "", "More.")]) rendered = quality.measure_bundle(root).render() assert "documents 1" not in rendered assert "one-concept documents" not in rendered assert "empty 0/2" in rendered assert "no source_file on 2 of 2 concepts" in rendered