"""Measure how far Arm D's outline rule reaches into a corpus. Measure, don't build. Order 20260906T213322Z-1044411564-from-.claude, phase 2. `docs/2026-09-04-k3-arm-c.md` measured a SIZE answer to K3 and falsified it: Arm C changed 6 of 12 proposals and moved the four category counts by zero. The operator ruled on 2026-09-04 that the next arm must be content-based. Arm D reads the document's OWN numbered outline; this instrument answers how far that rule reaches, and how much of its reach the shipping orphan check then deletes. It builds nothing. No new rule, no extractor, no bundle, no threshold -- the threshold is the operator's, and only once an arm moves the number. **It imports the shipped functions rather than re-implementing them.** An instrument carrying its own copy of the grammar measures a second definition that can drift from the tool's without a test going red, and every figure it publishes would then be about code nobody ships. `tests/test_outline_measure.py` pins the identity. It also carries the K3 SAMPLE DRAW, for the same reason: `8/12` is a published figure, and until now the draw existed only in an uncommitted script. A number whose script is not committed cannot be reproduced, which is the whole complaint this instrument answers. ## What is an expectation here and what is not **Pre-gate `144` boundaries on `23/39` documents are the declared expectation.** They were measured this session against a replica that first reproduced Arm B's 618 candidates exactly, so a different pre-gate total means the shipped rule is not the measured one. **Post-gate figures are reference values, printed beside the measured ones and never gating anything.** Declaring them would force a later reader to write a number down as true because a plan predicted it. What the orphan check deletes is a property of the shipped code, and this instrument's job is to report it. """ from __future__ import annotations import argparse import hashlib import re import sys import unicodedata from dataclasses import dataclass, field from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) from llm_ingestion_okf.errors import ExtractionError # noqa: E402 from llm_ingestion_okf.extract import extract_text # noqa: E402 from llm_ingestion_okf.materialize import reduce_to_id_grammar # noqa: E402 from llm_ingestion_okf.propose import ( # noqa: E402 RULE_OUTLINE, Candidate, _segment_path, find_candidates, outline_lines, outline_runs, ) #: The gate length Arm D is measured at. Declared, not swept: the brief forbids #: a sweep, and 3 was chosen from the run-length distribution (328/37/18) #: before any reach figure was known. RUN_LENGTH = 3 #: The K3 sample, declared in `docs/2026-09-02-k3-k4-k5-metode.md` and #: reproduced here rather than re-derived: n = 12, stratified 8 pdf / 3 docx / #: 1 xlsx, ordered within each stratum by the hex SHA-256 of the filename. SAMPLE_STRATA = {"pdf": 8, "docx": 3, "xlsx": 1} #: The declared pre-gate expectation. A different value means the shipped rule #: is not the rule that was measured, and no post-gate figure may be read. EXPECTED_PRE_GATE_BOUNDARIES = 144 EXPECTED_PRE_GATE_DOCUMENTS = 23 #: Reference values only -- measured this session against a replica, printed #: beside the shipped code's own numbers and never compared against them. REFERENCE_POST_GATE_BOUNDARIES = 95 REFERENCE_POST_GATE_DOCUMENTS = 21 REFERENCE_TOTAL_ENTRIES = 709 REFERENCE_DELETED_ARM_B = 4 #: "An alphabetic word" -- Unicode letters only, so a title made of digits and #: single letters (`477 3 025`, `D 1 L`) counts as junk. Digits and `_` #: excluded. THIS instrument's definition, not an upstream term. _ALPHA_WORD = re.compile(r"[^\W\d_]{2,}", re.UNICODE) def _draw_key(name: str) -> str: """The draw's ordering key: hex SHA-256 of the NFC-normalised filename. NFC first because macOS hands filenames over decomposed, so the same visual corpus would otherwise draw a different sample depending on which form the names arrived in. """ return hashlib.sha256(unicodedata.normalize("NFC", name).encode("utf-8")).hexdigest() def draw_sample(names: list[str], strata: dict[str, int] | None = None) -> list[str]: """The K3 draw: first k per format stratum, then canonical order over the sample.""" strata = SAMPLE_STRATA if strata is None else strata picked: list[str] = [] for extension, count in strata.items(): pool = sorted( (name for name in names if name.lower().endswith("." + extension)), key=_draw_key, ) picked.extend(pool[:count]) return sorted(picked, key=_draw_key) @dataclass(frozen=True) class DocumentMeasurement: """One document, both sides of the orphan check.""" name: str #: What the outline rule PROPOSES: the length of the last admitted run. pre_gate_boundaries: int #: What survives `find_candidates` -- the shipping orphan check included. post_gate_boundaries: int arm_b_entries: int arm_d_entries: int #: Arm B candidates present without the flag and absent with it, matched on #: start offset. These are coverage the arm COSTS, and they are why a #: corpus-wide entry delta alone would flatter the arm. deleted_arm_b: int unique_paths: int junk_titles: int span_sizes: list[int] = field(default_factory=list) def line(self) -> str: return ( f"| {self.name} | {self.pre_gate_boundaries} | {self.post_gate_boundaries} " f"| {self.arm_b_entries} | {self.arm_d_entries} | {self.deleted_arm_b} " f"| {self.unique_paths} | {self.junk_titles} |" ) @dataclass(frozen=True) class Row: """A corpus file: measured, or a stated reason it was not.""" name: str measurement: DocumentMeasurement | None skip_reason: str | None def line(self) -> str: if self.measurement is None: return f"| {self.name} | -- | -- | -- | -- | -- | -- | -- |" return self.measurement.line() def _paths_for(candidates: list[Candidate], prefix: str) -> list[str]: taken: set[str] = set() return [_segment_path(candidate, taken, prefix) for candidate in candidates] def measure_document(text: str, name: str, run_length: int = RUN_LENGTH) -> DocumentMeasurement: """Both sides of the gate for one document, from the SHIPPED functions.""" runs = outline_runs(outline_lines(text), run_length) pre_gate = len(runs[-1]) if runs else 0 arm_b = find_candidates(text) arm_d = find_candidates(text, outline_run=run_length) outline_kept = [c for c in arm_d if c.rule == RULE_OUTLINE] arm_b_starts = {c.start for c in arm_b} arm_d_starts = {c.start for c in arm_d} deleted = len(arm_b_starts - arm_d_starts) prefix = reduce_to_id_grammar(Path(name).stem) paths = _paths_for(arm_d, prefix) junk = sum(1 for c in outline_kept if not _ALPHA_WORD.search(c.title)) return DocumentMeasurement( name=name, pre_gate_boundaries=pre_gate, post_gate_boundaries=len(outline_kept), arm_b_entries=len(arm_b), arm_d_entries=len(arm_d), deleted_arm_b=deleted, unique_paths=len(set(paths)), junk_titles=junk, span_sizes=[c.end - c.start for c in arm_d], ) def run(corpus: Path, run_length: int = RUN_LENGTH) -> list[Row]: rows: list[Row] = [] for path in sorted(corpus.iterdir(), key=lambda p: p.name): if not path.is_file(): continue try: text = extract_text(path.name, path.read_bytes()) except ExtractionError as exc: rows.append(Row(name=path.name, measurement=None, skip_reason=f"{exc.code}: {exc}")) continue rows.append( Row( name=path.name, measurement=measure_document(text, path.name, run_length), skip_reason=None, ) ) return rows def _percentile(values: list[int], fraction: float) -> int: if not values: return 0 ordered = sorted(values) index = min(len(ordered) - 1, int(fraction * (len(ordered) - 1) + 0.5)) return ordered[index] def render(corpus: Path, rows: list[Row], run_length: int = RUN_LENGTH) -> str: measured = [row.measurement for row in rows if row.measurement is not None] extractable = len(measured) pre_total = sum(m.pre_gate_boundaries for m in measured) post_total = sum(m.post_gate_boundaries for m in measured) pre_docs = sum(1 for m in measured if m.pre_gate_boundaries > 0) post_docs = sum(1 for m in measured if m.post_gate_boundaries > 0) arm_b_total = sum(m.arm_b_entries for m in measured) arm_d_total = sum(m.arm_d_entries for m in measured) deleted_total = sum(m.deleted_arm_b for m in measured) junk_total = sum(m.junk_titles for m in measured) zero_entry = sum(1 for m in measured if m.arm_d_entries == 0) unique_total = sum(m.unique_paths for m in measured) spans = [size for m in measured for size in m.span_sizes] names = [row.name for row in rows] sample = draw_sample(names) by_name = {m.name: m for m in measured} reached = [n for n in sample if n in by_name and by_name[n].post_gate_boundaries > 0] lines = [ f"# Arm D outline reach, {corpus}", "", f"Run-length gate: {run_length}. Declared, not swept.", f"N = {len(rows)} (corpus directory file count). Extractable: {extractable}/{len(rows)}.", "", "## Pre-gate -- what the rule proposes (the DECLARED expectation)", "", f"- boundaries: **{pre_total}** (expected {EXPECTED_PRE_GATE_BOUNDARIES})", f"- documents reached: **{pre_docs}/{extractable}** " f"(expected {EXPECTED_PRE_GATE_DOCUMENTS}/{extractable})", "", "## Post-gate -- what survives the orphan check (MEASURED, not expected)", "", f"- boundaries: **{post_total}** (reference {REFERENCE_POST_GATE_BOUNDARIES})", f"- documents reached: **{post_docs}/{extractable}** " f"(reference {REFERENCE_POST_GATE_DOCUMENTS}/{extractable})", f"- deleted by the gate: **{pre_total - post_total}** of {pre_total}", "", "## Entries, and what the arm costs", "", f"- Arm B entries: **{arm_b_total}**", f"- Arm D entries: **{arm_d_total}** " f"(delta {arm_d_total - arm_b_total:+d}, reference {REFERENCE_TOTAL_ENTRIES})", f"- existing Arm B candidates deleted: **{deleted_total}** " f"(reference {REFERENCE_DELETED_ARM_B})", "", "## The K3 sample", "", f"- sample reach: **{len(reached)}/{len(sample)}**", "", "## Quality counts, each with its denominator", "", f"- unique concept paths: **{unique_total}** of {arm_d_total} entries", f"- documents with zero entries under Arm D: **{zero_entry}**/{extractable}", f"- outline titles with no alphabetic word: **{junk_total}** of {post_total}", "", "## Span-size distribution, Arm D", "", f"- spans: {len(spans)}; min {min(spans) if spans else 0}; " f"p50 {_percentile(spans, 0.5)}; p95 {_percentile(spans, 0.95)}; " f"max {max(spans) if spans else 0}", f"- spans under 200 chars: **{sum(1 for s in spans if s < 200)}** of {len(spans)}", "", "## Per document", "", "| file | pre-gate | post-gate | arm B | arm D | deleted | unique paths | junk titles |", "|---|---|---|---|---|---|---|---|", ] lines.extend(row.line() for row in rows) lines.extend(["", "## The K3 sample, in canonical draw order", ""]) for position, name in enumerate(sample): measurement = by_name.get(name) state = "not extractable" if measurement is None else f"{measurement.post_gate_boundaries}" lines.append(f"- {position}: {name} -- outline boundaries: {state}") return "\n".join(lines) + "\n" def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--corpus", type=Path, required=True, help="the directory to measure") parser.add_argument("--report", type=Path, required=True, help="where to write the report") parser.add_argument( "--run-length", type=int, default=RUN_LENGTH, metavar="N", help="the ascending-run gate length Arm D is measured at (declared, not swept)", ) args = parser.parse_args(argv) if not args.corpus.is_dir(): print( f"okf-outline-measure: FAILED - no corpus directory at {args.corpus}", file=sys.stderr, ) return 1 rows = run(args.corpus, args.run_length) args.report.write_text(render(args.corpus, rows, args.run_length), encoding="utf-8", newline="") print(f"okf-outline-measure: wrote {args.report}") return 0 if __name__ == "__main__": raise SystemExit(main())