llm-ingestion-okf/src/llm_ingestion_okf/corpus.py

"""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 Callable, Mapping
from dataclasses import dataclass, replace
from pathlib import Path

from .errors import IngestError
from .extract import extract_text
from .inbox import (
    GateDecision,
    InboxResult,
    process_inbox,
    relative_source,
    walk_inbox,
)
from .profiles import SEGMENTED_OKF_V0_2, STRUCTURED_V1, BundleProfile
from .segmentation import SegmentationPlan, parse_segmentation_plan

__all__ = [
    "CorpusReport",
    "converter_identity",
    "load_plans",
    "is_degenerate",
    "main",
    "measure",
    "replace",
    "unaccounted_names",
]

HARNESS_ID = "okf-corpus-run"

# The log's name and title in ONE place, because two of them now read it: the
# file's own frontmatter and the root index entry that points at it. Two
# literals would let the link's label drift away from the thing it labels.
LOG_NAME = "log.md"
LOG_TITLE = "Corpus run history"


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 ._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",
            f"title: {LOG_TITLE}",
            "---",
            "",
            f"# {LOG_TITLE}",
            "",
            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,
    pdf_headings: bool = False,
    heading_reserve: Callable[[str], bool] | None = None,
    ocr: bool = False,
) -> 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.
    """
    # ONE walk rule, imported rather than restated: the denominator has to be
    # counted over exactly the set of files the door ingests, or the
    # conservation identity would hold over a different N than the run did.
    walked, _ = walk_inbox(corpus, exclude=bundle)
    dropped = tuple(relative_source(path, corpus) for path in walked)
    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,
        pdf_headings=pdf_headings,
        heading_reserve=heading_reserve,
        ocr=ocr,
    )
    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 "
            "the proposer 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)


# `link_log_in_root_index` LIVED HERE AND WAS REMOVED (2026-09-08).
#
# It appended `- [Corpus run history](log.md)` to the root index (`95eb271`) so
# a reader entering at `index.md` could reach the one file carrying `N`. That
# was a LOCAL choice, said so at the time, and upstream never linked its own
# logs -- measured at `9a15b13`, 0 of 24 shipped `index.md` files name the one
# `log.md` in the set.
#
# The cost was measured on K2 by the first consumer to walk a bundle of ours
# with a live model: consumption contract SS 9.2 forbids a consumer from
# enumerating the bundle directory unless the profile says the index is
# derived, so the index tree IS the entire map a consumer may use, and anything
# it links is a document. Their navigator returned 630 where our own pre-pass
# counts 629, and a corpus run's own log became readable and citable as
# content. `5a0c879` (F2) excluded `log.md` from OUR walk, which fixed the
# count on one side of a disagreement produced on the other.
#
# The log itself is still written to the bundle root, which is where SS 9 puts
# it and all F2 ever needed. Reported in
# `docs/2026-09-08-prisform-og-loggen-k2.md`.


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_NAME).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())