llm-ingestion-okf/tools/okf_table_measure.py

"""Measure how far Arm E's grid-rule join reaches into a corpus. Measure, don't build.

Order 20260907T075834Z-18584396-from-.claude. `docs/2026-09-07-k3-arm-d.md`
moved K3's first-rater row from 8/4/0/0 to 4/5/0/3 and left five documents at
`too fine`. Three of those five are dominated by `rule:table-block`, because the
converter emits pandoc GRID tables whose rows are separated by rule lines the
table grammar cannot match -- so one table becomes one concept per row group.
Arm E stops a rule line closing an open block. This instrument answers how far
that reaches, and what it costs.

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_table_measure.py`
pins the identity with `is` rather than `==`, because `re.compile` returns a
distinct object for an equal pattern and a pasted literal would satisfy `==`.

## What is an expectation here and what is not

**38 grid-rule lines across 3 documents is the declared expectation.** It was
measured on this corpus before the rule was written, and a different value means
the shipped grammar is not the measured one -- no figure below it may be read.

**The `|`-row column is the CEILING, and it is a measurement, not a reference.**
Arm E cannot move a document that carries no table row. Asserting the ceiling
from entry counts alone would assume the orphan check kept every table
candidate, which nobody measured; this column counts rows in the text itself,
for every file including the ones the door refuses.

**Entry counts are reference values**, printed beside the measured ones and
gating nothing.
"""

from __future__ import annotations

import argparse
import sys
from dataclasses import dataclass
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
sys.path.insert(0, str(Path(__file__).resolve().parent))

from okf_outline_measure import draw_sample

from llm_ingestion_okf.errors import ExtractionError
from llm_ingestion_okf.extract import extract_text
from llm_ingestion_okf.propose import (
    _GRID_RULE,
    _TABLE_ROW,
    RULE_TABLE_BLOCK,
    RULE_TABLE_GRID,
    find_candidates,
)

#: The gate Arm D is measured at, and the value Arm E is measured ALONGSIDE.
#: Declared, not swept -- it belongs to Arm D and is reproduced here so the two
#: arms are compared on the same outline.
RUN_LENGTH = 3

#: The declared expectation, measured on this corpus before the rule existed. A
#: different value means the shipped grammar is not the grammar that was
#: measured, and no figure below it may be read.
EXPECTED_GRID_RULE_LINES = 38
EXPECTED_GRID_DOCUMENTS = 3

#: Reference values only -- the Arm D corpus totals, printed beside the measured
#: ones and never compared against them.
REFERENCE_ARM_D_ENTRIES = 709
REFERENCE_TABLE_BLOCK_ENTRIES = 33


@dataclass(frozen=True)
class TableMeasurement:
    """One document, both sides of the join."""

    name: str
    #: Lines matching the SHIPPED table grammar. This is the ceiling column: a
    #: document with none of these cannot be moved by Arm E, whatever the
    #: orphan check later did to its candidates.
    pipe_rows: int
    #: Lines matching the SHIPPED grid-rule grammar.
    grid_rule_lines: int
    #: Table-block candidates surviving the orphan check, before and after.
    blocks_before: int
    blocks_after: int
    #: Blocks Arm E ACTUALLY joined -- not blocks whose span contains a rule
    #: line. A single-row grid table has a rule line and joins nothing.
    joined_blocks: int
    #: All candidates, not only table blocks: what the plan would carry.
    entries_before: int
    entries_after: int

    def line(self) -> str:
        return (
            f"| {self.name} | {self.pipe_rows} | {self.grid_rule_lines} "
            f"| {self.blocks_before} | {self.blocks_after} | {self.joined_blocks} "
            f"| {self.entries_before} | {self.entries_after} |"
        )


@dataclass(frozen=True)
class Row:
    """A corpus file: measured, or a stated reason it was not.

    A refused file stays in the list rather than being dropped, because
    `extractable / len(rows)` is the door count this round reports and a
    silently missing row would make it unmeasurable.
    """

    name: str
    measurement: TableMeasurement | None
    skip_reason: str | None

    def line(self) -> str:
        if self.measurement is None:
            return f"| {self.name} | -- | -- | -- | -- | -- | -- | -- |"
        return self.measurement.line()


def measure_document(text: str, name: str, run_length: int = RUN_LENGTH) -> TableMeasurement:
    """Both sides of the join for one document, from the SHIPPED functions."""
    lines = text.splitlines(keepends=True)
    before = find_candidates(text, outline_run=run_length)
    after = find_candidates(text, outline_run=run_length, table_grid=True)
    return TableMeasurement(
        name=name,
        pipe_rows=sum(1 for line in lines if _TABLE_ROW.match(line)),
        grid_rule_lines=sum(1 for line in lines if _GRID_RULE.match(line)),
        blocks_before=sum(1 for c in before if c.rule == RULE_TABLE_BLOCK),
        blocks_after=sum(1 for c in after if c.rule == RULE_TABLE_BLOCK),
        joined_blocks=sum(1 for c in after if c.grid),
        entries_before=len(before),
        entries_after=len(after),
    )


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 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)
    pipe_docs = sum(1 for m in measured if m.pipe_rows > 0)
    pipe_rows = sum(m.pipe_rows for m in measured)
    grid_docs = sum(1 for m in measured if m.grid_rule_lines > 0)
    grid_lines = sum(m.grid_rule_lines for m in measured)
    joined_docs = sum(1 for m in measured if m.joined_blocks > 0)
    joined = sum(m.joined_blocks for m in measured)
    blocks_before = sum(m.blocks_before for m in measured)
    blocks_after = sum(m.blocks_after for m in measured)
    entries_before = sum(m.entries_before for m in measured)
    entries_after = sum(m.entries_after for m in measured)
    changed = [m for m in measured if m.entries_before != m.entries_after]

    names = [row.name for row in rows]
    sample = draw_sample(names)
    by_name = {m.name: m for m in measured}

    lines = [
        f"# Arm E grid-table reach, {corpus}",
        "",
        f"Outline gate: {run_length} (Arm D's, reproduced so both arms share an outline).",
        f"N = {len(rows)} (corpus directory file count). Extractable: {extractable}/{len(rows)}.",
        "",
        "## The ceiling -- documents Arm E could move at all (MEASURED)",
        "",
        f"- documents with at least one table row: **{pipe_docs}**/{extractable}",
        f"- table rows in total: **{pipe_rows}**",
        f"- documents with at least one grid-rule line: **{grid_docs}**/{extractable} "
        f"(expected {EXPECTED_GRID_DOCUMENTS})",
        f"- grid-rule lines in total: **{grid_lines}** (expected {EXPECTED_GRID_RULE_LINES})",
        "",
        "A document with no table row cannot be moved by this arm, whatever the",
        "orphan check later did to its candidates. That is what makes this the",
        "ceiling rather than the entry counts below.",
        "",
        "## What the join does, and what it costs",
        "",
        f"- documents where any block was joined: **{joined_docs}**/{extractable}",
        f"- blocks joined: **{joined}**",
        f"- table-block candidates: **{blocks_before}** -> **{blocks_after}** "
        f"(reference {REFERENCE_TABLE_BLOCK_ENTRIES})",
        f"- entries, whole corpus: **{entries_before}** -> **{entries_after}** "
        f"(delta {entries_after - entries_before:+d}, reference {REFERENCE_ARM_D_ENTRIES})",
        f"- documents whose entry count changed: **{len(changed)}**/{extractable}",
        "",
        "## Per document",
        "",
        "| file | rows | grid rules | blocks before | blocks after | joined | "
        "entries before | entries after |",
        "|---|---|---|---|---|---|---|---|",
    ]
    lines.extend(row.line() for row in rows)
    lines.extend(["", "## Files the door refused, with their reason", ""])
    refused = [row for row in rows if row.measurement is None]
    if refused:
        lines.extend(f"- {row.name} -- {row.skip_reason}" for row in refused)
    else:
        lines.append("- none")
    lines.extend(["", "## The K3 sample, in canonical draw order", ""])
    for position, name in enumerate(sample):
        measurement = by_name.get(name)
        if measurement is None:
            state = "not extractable"
        else:
            state = (
                f"rows {measurement.pipe_rows}, grid rules {measurement.grid_rule_lines}, "
                f"entries {measurement.entries_before} -> {measurement.entries_after}"
            )
        lines.append(f"- {position}: {name} -- {state}")
    lines.extend(["", f"Rule names in play: `{RULE_TABLE_BLOCK}`, `{RULE_TABLE_GRID}`.", ""])
    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="Arm D's outline gate, reproduced so both arms are measured on one outline",
    )
    args = parser.parse_args(argv)
    if not args.corpus.is_dir():
        print(
            f"okf-table-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-table-measure: wrote {args.report}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())