llm-ingestion-okf/tests/test_corpus_run.py

"""The corpus harness: it reports numbers with denominators, or it fails.

Three properties, all of them reactions to measured defects rather than good
intentions:

- **K1b is a COMMAND, not a sentence in a report.** The conservation identity
  `merged + Sigma(coded rejections) == N` is checked by the harness, which
  EXITS NON-ZERO when it does not hold. A prose assertion is something a reader
  has to trust; an exit status is something a pipeline cannot ignore.
- **`N` is computed, never typed.** A literal `43` keeps passing after the
  corpus changes, and the number it then reports is 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 reported separately for that reason.

The negative control is the point of this module. A harness that can never
fail proves nothing, so `test_a_file_neither_merged_nor_coded_fails_the_run`
hands the conservation check an inventory it must reject -- if that test ever
passes silently, every green run above it becomes meaningless.
"""

from __future__ import annotations

import sys
from pathlib import Path

import pytest

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

import okf_corpus_run  # noqa: E402

INGESTED_AT = "2026-07-25T12:00:00Z"

SUBSTANTIVE = "Krav til seksjonering av bygget.\n\nEn andre setning som baerer innhold.\n"
DEGENERATE = "   \n\t\n   \n"


def corpus(tmp_path: Path, files: dict[str, str]) -> Path:
    root = tmp_path / "corpus"
    root.mkdir(parents=True, exist_ok=True)
    for name, text in files.items():
        (root / name).write_text(text, encoding="utf-8", newline="")
    return root


def test_the_denominator_is_the_directory_not_a_literal(tmp_path: Path) -> None:
    root = corpus(tmp_path, {"a.md": SUBSTANTIVE, "b.md": SUBSTANTIVE, "c.md": SUBSTANTIVE})
    report = okf_corpus_run.measure(root, tmp_path / "bundle", ingested_at=INGESTED_AT)
    assert report.n == 3
    assert report.n == len(list(root.iterdir()))


def test_the_conservation_identity_holds_and_the_run_exits_zero(tmp_path: Path) -> None:
    root = corpus(
        tmp_path,
        {"a.md": SUBSTANTIVE, "b.md": DEGENERATE, "c.md": SUBSTANTIVE, "d.bin": "x"},
    )
    code = okf_corpus_run.main(
        ["--corpus", str(root), "--report", str(tmp_path / "r.md"), "--ingested-at", INGESTED_AT]
    )
    assert code == 0
    report = okf_corpus_run.measure(root, tmp_path / "bundle2", ingested_at=INGESTED_AT)
    assert report.merged + report.rejected == report.n
    assert not report.unaccounted


def test_a_file_neither_merged_nor_coded_fails_the_run(tmp_path: Path) -> None:
    """The negative control. Without it a green run proves nothing.

    The check is handed an inventory where one name is in neither column --
    exactly what a silently dropped file looks like from the outside -- and it
    must both refuse and NAME the file.
    """
    unaccounted = okf_corpus_run.unaccounted_names(
        dropped=("a.md", "b.md", "vanished.md"),
        merged=("a.md",),
        coded=("b.md",),
    )
    assert unaccounted == ("vanished.md",)
    assert okf_corpus_run.unaccounted_names(dropped=("a.md",), merged=("a.md",), coded=()) == ()


def test_the_harness_exits_non_zero_and_names_the_unaccounted_file(
    tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
) -> None:
    root = corpus(tmp_path, {"a.md": SUBSTANTIVE, "b.md": SUBSTANTIVE})
    real = okf_corpus_run.measure

    def lose_one(*args: object, **kwargs: object):
        report = real(*args, **kwargs)  # type: ignore[arg-type]
        return okf_corpus_run.replace(report, unaccounted=("b.md",))

    monkeypatch.setattr(okf_corpus_run, "measure", lose_one)
    code = okf_corpus_run.main(
        ["--corpus", str(root), "--report", str(tmp_path / "r.md"), "--ingested-at", INGESTED_AT]
    )
    assert code != 0
    assert "b.md" in capsys.readouterr().err


def test_the_degenerate_rule_is_reproducible_from_its_statement(tmp_path: Path) -> None:
    """Zero characters after stripping whitespace. A definition, not a threshold."""
    assert okf_corpus_run.is_degenerate("")
    assert okf_corpus_run.is_degenerate("   \n\t  ")
    assert not okf_corpus_run.is_degenerate("x")
    root = corpus(tmp_path, {"a.md": SUBSTANTIVE, "b.md": DEGENERATE})
    report = okf_corpus_run.measure(root, tmp_path / "bundle", ingested_at=INGESTED_AT)
    assert report.substantive == 1
    assert report.degenerate == 1
    assert report.substantive + report.degenerate == report.merged


def test_the_three_counts_are_reported_separately(tmp_path: Path) -> None:
    """A healthy persisted count can hide a pile of quarantines."""
    root = corpus(tmp_path, {"a.md": SUBSTANTIVE, "b.bin": "x"})
    report = okf_corpus_run.measure(root, tmp_path / "bundle", ingested_at=INGESTED_AT)
    assert (report.extracted, report.gated, report.persisted) == (1, 1, 1)
    assert report.n == 2
    text = report.render()
    for label in ("extracted", "gated", "persisted", "denominator"):
        assert label in text


def test_the_report_names_the_resolved_converter_and_its_version(tmp_path: Path) -> None:
    """The vendored binary is bypassed silently otherwise -- measured three times."""
    root = corpus(tmp_path, {"a.md": SUBSTANTIVE})
    report = okf_corpus_run.measure(root, tmp_path / "bundle", ingested_at=INGESTED_AT)
    text = report.render()
    assert "converter" in text.lower()
    assert okf_corpus_run.converter_identity()[1] in text


def test_wall_time_per_file_is_reported(tmp_path: Path) -> None:
    """The only evidence the scale requirement will ever have."""
    root = corpus(tmp_path, {"a.md": SUBSTANTIVE, "b.md": SUBSTANTIVE})
    report = okf_corpus_run.measure(root, tmp_path / "bundle", ingested_at=INGESTED_AT)
    assert report.seconds_total >= 0.0
    assert "per file" in report.render()


def test_the_bundle_carries_a_log_md_that_makes_k1b_checkable_from_the_bundle(
    tmp_path: Path,
) -> None:
    """K1b was verifiable only from the harness's stdout, which no bundle carries.

    Measured by a consumer 2026-09-03: given the bundle alone, `N` could not be
    recovered, so `merged + Sigma(coded rejections) == N` was not checkable from
    the artifact -- only `merged` was. SPEC section 9 already reserves `log.md`
    for exactly this, so the run path writes the denominator and every rejection
    code into it. Existence is not the property under test: an empty stub would
    pass that and recover nothing.
    """
    root = corpus(tmp_path, {"a.md": SUBSTANTIVE, "b.md": DEGENERATE, "c.bin": "x"})
    bundle = tmp_path / "bundle"
    code = okf_corpus_run.main(
        [
            "--corpus",
            str(root),
            "--report",
            str(tmp_path / "r.md"),
            "--bundle",
            str(bundle),
            "--ingested-at",
            INGESTED_AT,
        ]
    )
    assert code == 0

    log = (bundle / "log.md").read_text(encoding="utf-8")
    # Section 9 form: reserved frontmatter type, a heading, ISO date headings.
    assert log.startswith("---\ntype: Log\n")
    assert f"\n## {INGESTED_AT[:10]}\n" in log
    # The numbers, not merely the file. N is the denominator that was missing.
    assert "N = 3" in log
    assert "merged = 2" in log
    assert "`extractor_unknown`: 1" in log
    # The identity itself, so a reader does not have to re-derive it.
    assert "2 + 1 = 3" in log


def test_the_log_is_dated_from_ingested_at_and_never_the_wall_clock(tmp_path: Path) -> None:
    """Determinism is bit-exact here as everywhere: two runs of the same corpus
    at the same `ingested_at` produce the same `log.md` bytes."""
    root = corpus(tmp_path, {"a.md": SUBSTANTIVE})
    first = tmp_path / "b1"
    second = tmp_path / "b2"
    for bundle in (first, second):
        okf_corpus_run.main(
            [
                "--corpus",
                str(root),
                "--report",
                str(tmp_path / "r.md"),
                "--bundle",
                str(bundle),
                "--ingested-at",
                INGESTED_AT,
            ]
        )
    assert (first / "log.md").read_bytes() == (second / "log.md").read_bytes()


SEGMENTABLE = (
    "# 1 Innledning\n\nBakgrunnen for anskaffelsen er beskrevet her, med nok\n"
    "tekst til at seksjonen baerer innhold.\n\n"
    "# 2 Kravspesifikasjon\n\nKravene til leveransen er listet i dette\n"
    "avsnittet, ogsaa med en andre setning.\n"
)


def _propose(source: Path, out: Path) -> None:
    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools"))
    import okf_propose_segments

    assert okf_propose_segments.run(source, out, okf_type="reference", proposed_at=INGESTED_AT) == 0


def test_a_plan_directory_makes_the_run_segment_and_mark_every_concept_proposed(
    tmp_path: Path,
) -> None:
    """The defect this closes: the harness ran under `STRUCTURED_V1` and passed
    no plans at all, so a corpus arrived as one flat concept per file with no
    `adjudication` key anywhere -- measured by a consumer as 0 of 39.

    Segmentation is not a flag the harness may set on its own: the plans are
    produced per document by the proposer first, and the harness replays them.
    So the wiring under test is `--plans-dir`, and the profile follows from it.
    """
    root = corpus(tmp_path, {"doc.md": SEGMENTABLE})
    plans = tmp_path / "plans"
    plans.mkdir()
    _propose(root / "doc.md", plans / "doc.json")

    bundle = tmp_path / "bundle"
    code = okf_corpus_run.main(
        [
            "--corpus",
            str(root),
            "--report",
            str(tmp_path / "r.md"),
            "--bundle",
            str(bundle),
            "--ingested-at",
            INGESTED_AT,
            "--plans-dir",
            str(plans),
            "--bundle-id",
            "k2-trinn1",
            "--okf-version",
            "0.2",
        ]
    )
    assert code == 0

    concepts = sorted(
        path for path in bundle.rglob("*.md") if path.name not in ("index.md", "log.md")
    )
    # 1-to-N: one dropped file, more than one concept.
    assert len(concepts) > 1
    bodies = [path.read_text(encoding="utf-8") for path in concepts]
    # Every concept carries the marker, and it says PROPOSED -- the proposer
    # never adjudicates, so an `adjudicated` here would be a machine's guess
    # standing where a human's judgement belongs.
    assert all("\nadjudication: proposed\n" in body for body in bodies)
    assert not any("adjudication: adjudicated" in body for body in bodies)


def test_a_plan_directory_without_the_caller_owned_root_values_is_refused(
    tmp_path: Path,
) -> None:
    """`okf_version`'s value belongs to the catalog (decision E1) and `bundle_id`
    to whoever delimits the bundle. A literal for either in this harness would
    claim a decision this repository does not own, so both are arguments and
    their absence is a refusal rather than a default."""
    root = corpus(tmp_path, {"doc.md": SEGMENTABLE})
    plans = tmp_path / "plans"
    plans.mkdir()
    _propose(root / "doc.md", plans / "doc.json")

    code = okf_corpus_run.main(
        [
            "--corpus",
            str(root),
            "--report",
            str(tmp_path / "r.md"),
            "--bundle",
            str(tmp_path / "bundle"),
            "--ingested-at",
            INGESTED_AT,
            "--plans-dir",
            str(plans),
        ]
    )
    assert code == 2
    assert not (tmp_path / "bundle").exists()


def test_without_a_plan_directory_the_run_is_unchanged(tmp_path: Path) -> None:
    """The wiring is additive. A run with no plans stays exactly the flat,
    `STRUCTURED_V1` run it was, so the K1/K2 numbers already published remain
    reproducible from the same command."""
    root = corpus(tmp_path, {"doc.md": SEGMENTABLE})
    bundle = tmp_path / "bundle"
    assert (
        okf_corpus_run.main(
            [
                "--corpus",
                str(root),
                "--report",
                str(tmp_path / "r.md"),
                "--bundle",
                str(bundle),
                "--ingested-at",
                INGESTED_AT,
            ]
        )
        == 0
    )
    concepts = [path for path in bundle.rglob("*.md") if path.name not in ("index.md", "log.md")]
    assert len(concepts) == 1
    assert "adjudication:" not in concepts[0].read_text(encoding="utf-8")


def test_a_second_run_into_the_same_bundle_reproduces_it_byte_for_byte(tmp_path: Path) -> None:
    """`log.md` is a file the harness writes INTO a directory Door B enumerates
    on the next round: it matches the concept glob and is excluded only by
    `index.md`'s name, so a rebuild could see it as pre-existing curated content
    or prune it. Rebuild-equals-incremental is the property the segmented bundle
    is built on, and a log that broke it would be worse than no log.
    """
    root = corpus(tmp_path, {"doc.md": SEGMENTABLE, "flat.md": SUBSTANTIVE})
    plans = tmp_path / "plans"
    plans.mkdir()
    _propose(root / "doc.md", plans / "doc.json")
    bundle = tmp_path / "bundle"
    argv = [
        "--corpus",
        str(root),
        "--report",
        str(tmp_path / "r.md"),
        "--bundle",
        str(bundle),
        "--ingested-at",
        INGESTED_AT,
        "--plans-dir",
        str(plans),
        "--bundle-id",
        "k2",
        "--okf-version",
        "0.2",
    ]

    assert okf_corpus_run.main(argv) == 0
    first = {
        path.relative_to(bundle).as_posix(): path.read_bytes()
        for path in sorted(bundle.rglob("*"))
        if path.is_file()
    }
    assert "log.md" in first

    assert okf_corpus_run.main(argv) == 0
    second = {
        path.relative_to(bundle).as_posix(): path.read_bytes()
        for path in sorted(bundle.rglob("*"))
        if path.is_file()
    }
    assert second == first