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 .accounting import Accounting, account_run
from .assets import ASSETS_DIR
from .errors import IngestError
from .extract import extract_text
from .inbox import (
    Gate,
    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__ = [
    "GATE_GUARD_TRUSTED_SOURCE",
    "GATE_GUARD_USER_UPLOAD",
    "GATE_NAMES",
    "GATE_NONE",
    "CorpusReport",
    "converter_identity",
    "load_plans",
    "is_degenerate",
    "main",
    "measure",
    "replace",
    "resolve_gate",
    "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"

#: The heading of the summary section that names the documents the gate refused
#: WHOLE, in ONE place because the section, the stderr banner and the tests all
#: have to mean the same section.
REFUSED_HEADING = "## Documents the gate refused WHOLE"

#: How many refused documents the printed summary names before it stops and
#: points at `log.md`. A cap is not a licence to lose the rest: `log.md` carries
#: every name, uncapped, and the section says so on the line where it stops.
REFUSED_NAME_CAP = 10


#: The permissive stub. It approves EVERYTHING, and it is named rather than
#: anonymous because that is the whole lesson of F1: from the day `okf build`
#: was packaged until 2026-09-15 this function was wired unconditionally into
#: every run, so the only path anyone used screened nothing while the guard sat
#: in `pyproject.toml` as a mandatory runtime dependency. A consumer had no way
#: to see it. It survives because the corpus harness reproduces published
#: numbers with it and a caller measuring segmentation alone has a legitimate
#: reason to take the gate out of the picture -- but reaching it is now an act,
#: and the bundle's own log says so.
#: Whether a build carries the images its sources declare (0.10.0).
#:
#: ON, and the reason is the premise the capability was ordered on: what ends
#: up in a bundle after converting N sources has to BE the sources. Measured on
#: R761 Prosesskoden:2025 -- 2 726 of 2 726 processes carry their text and 12
#: `Tabell N-N` and 9 `Figur N-N` captions stand over nothing, because the
#: publisher ships those tables as raster pictures in both the PDF and the
#: NISO-STS delivery. A bundle that reads as complete and answers "toleranse-
#: klasse ... er gitt i tabell 84-2" with an empty space is the defect; it is
#: not a preference about bundle size.
#:
#: WHAT THIS COSTS IS PUBLISHED RATHER THAN IMPLIED. Measured over the K2
#: reference corpus, 2026-09-16: 32 of its 33 PDFs carry image objects and the
#: population is 4 828, of which 2 931 are 512x512 pixels or larger, 1 307 are
#: smaller and 149 declare no size at all. A drawing set contributes 2 524 on
#: its own. So this is a large change for a drawing-heavy corpus and none at
#: all for a prose one, and `--no-assets` reproduces the pre-0.10.0 bytes
#: exactly -- measured by `diff`, not asserted.
#:
#: NO SIZE FLOOR, and that is a measurement too. The obvious filter would be
#: "ignore anything under N pixels", and the distribution above does not offer
#: an N: it is a broad spread with no gap, unlike `OCR_CID_SHARE`'s, which is
#: bimodal with nothing between the modes. A threshold read off no gap is a
#: number this package chose, and it would silently drop somebody's small
#: table. Stated as a limit instead of guessed at.
DEFAULT_ASSETS = True

GATE_NONE = "none"
#: `okf build`'s default, set by the operator 2026-09-15 on a measurement:
#: over the 453 concept bodies of the pinned default bundle, this tier returns
#: the persist disposition on 453 of 453, so the move costs the pinned artifact
#: nothing while the scanner actually runs. Pointing the command at your own
#: folder is the trusted-source case.
GATE_GUARD_TRUSTED_SOURCE = "guard-trusted-source"
#: Door B's own library default, exposed for a folder that IS an untrusted
#: drop. Stricter, and the cost is measured rather than implied: on the same
#: 453 bodies it holds 1, which takes one of the 39 source documents out.
GATE_GUARD_USER_UPLOAD = "guard-user-upload"

#: Every gate a run may declare. A name outside this tuple is refused rather
#: than resolved to the stub: falling back would reproduce F1 with an extra
#: step -- the caller believes they asked for the guard, and the run approves
#: everything.
GATE_NAMES = (GATE_GUARD_TRUSTED_SOURCE, GATE_GUARD_USER_UPLOAD, GATE_NONE)

_GATE_DESCRIPTIONS = {
    GATE_GUARD_TRUSTED_SOURCE: "llm-ingestion-guard, PRESET_TRUSTED_SOURCE",
    GATE_GUARD_USER_UPLOAD: "llm-ingestion-guard, PRESET_USER_UPLOAD",
    GATE_NONE: "permissive stub -- NOTHING WAS SCREENED",
}


def _gate(text: str) -> GateDecision:
    return GateDecision(sanitized_text=text, disposition="warn")


def resolve_gate(name: str) -> Gate:
    """One place maps a gate NAME to the callable, and it is this one.

    The guard is imported lazily, inside the branch that needs it, because
    `guard_adapter` is the only module in this library allowed to import the
    dependency and importing the package must not pull it in.
    """
    if name == GATE_NONE:
        return _gate
    if name == GATE_GUARD_TRUSTED_SOURCE:
        from .guard_adapter import inbox_gate_trusted_source

        return inbox_gate_trusted_source
    if name == GATE_GUARD_USER_UPLOAD:
        from .guard_adapter import inbox_gate

        return inbox_gate
    raise IngestError(
        f"unknown gate {name!r}; one of {', '.join(GATE_NAMES)} -- a gate name that "
        "resolved to the permissive stub would approve everything while the caller "
        "believed they had asked for the guard",
        code="gate_invalid",
    )


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, ...]
    #: The gate that screened this run, by NAME. Defaulted LAST and to the
    #: stub so every existing construction site stays source-compatible; the
    #: two commands that build bundles both pass it explicitly.
    gate: str = GATE_NONE
    #: THE ASSET DENOMINATOR (0.10.0). `assets_carried` is what reached the
    #: bundle, `assets_found` is what the readers saw. Both, always: "51
    #: carried" is a measurement only beside "of 53 found", and a run whose
    #: figures were all refused must not read like a run over documents that had
    #: none. `assets` says whether they were looked for at all.
    assets: bool = False
    assets_carried: int = 0
    assets_found: int = 0
    #: Walked files that are not documents and whose bytes a persisted
    #: document carried as an image. Their own column in the conservation
    #: identity: before this they were ALSO counted as `extractor_unknown`
    #: rejections, so one file had two fates.
    carried_files: int = 0
    #: The content accounting of this run, or None when it was not asked for.
    accounting: Accounting | None = None
    #: Every document the GATE refused whole, as (source file, disposition),
    #: sorted. Its own column because `rejected` sums two different failures
    #: with two different remedies: a document the extractor could not read is
    #: a format this library does not handle, while a document the gate refused
    #: is a screening decision the caller can overrule with `--gate none`.
    #: Measured 2026-09-20 on a 594-file documentation corpus, where 17 sources
    #: were refused whole and the summary named neither the count nor a name.
    gate_refused: tuple[tuple[str, str], ...] = ()

    @property
    def merged(self) -> int:
        return self.substantive + self.degenerate

    @property
    def conservation_failed(self) -> bool:
        """K1b: every walked file is merged, carried or a coded rejection."""
        return bool(self.unaccounted) or (
            self.merged + self.carried_files + self.rejected != self.n
        )

    def identity(self) -> str:
        """The conservation identity with its numbers, in words.

        The carried column is written only when it is non-zero, so a run with
        no image files beside its documents keeps the line it always had.
        """
        if not self.carried_files:
            return (
                f"merged + coded rejections = {self.merged} + {self.rejected} = "
                f"{self.merged + self.rejected}; N = {self.n}"
            )
        total = self.merged + self.carried_files + self.rejected
        return (
            "merged + files carried through a document + coded rejections = "
            f"{self.merged} + {self.carried_files} + {self.rejected} = {total}; N = {self.n}"
        )

    def refusal_section(self) -> list[str]:
        """The four facts a reader needs about a document the gate dropped.

        How many (with the denominator), which ones, under which code, and the
        one command that carries them anyway. Empty when the gate refused
        nothing, so a clean run's summary is the summary it always was.
        """
        if not self.gate_refused:
            return []
        counts: dict[str, int] = {}
        for _, disposition in self.gate_refused:
            counts[disposition] = counts.get(disposition, 0) + 1
        by_code = ", ".join(f"`{code}`: {count}" for code, count in sorted(counts.items()))
        lines = [
            REFUSED_HEADING,
            "",
            f"The gate `{self.gate}` refused {len(self.gate_refused)} of {self.n} "
            "document(s) outright: not one element of them reached the bundle. "
            f"By code: {by_code}.",
            "",
        ]
        lines.extend(
            f"- `{name}` -- `{disposition}`"
            for name, disposition in self.gate_refused[:REFUSED_NAME_CAP]
        )
        remaining = len(self.gate_refused) - REFUSED_NAME_CAP
        if remaining > 0:
            lines.append(
                f"- ... and {remaining} more, every one named in the bundle's `{LOG_NAME}`."
            )
        lines += [
            "",
            "If you trust these sources yourself -- your own folder, your own "
            "documents -- build them with `--gate none`. It screens nothing, and "
            "the bundle records that it screened nothing.",
            "",
        ]
        return lines

    def refusal_banner(self) -> str:
        """The same fact in one line, for a run whose stdout went to a file.

        Built from `refusal_section`'s own data rather than beside it, so the
        loud line and the summary cannot come to disagree about a number.
        """
        if not self.gate_refused:
            return ""
        shown = ", ".join(
            f"{name} (`{disposition}`)" for name, disposition in self.gate_refused[:3]
        )
        remaining = len(self.gate_refused) - 3
        if remaining > 0:
            shown += f", and {remaining} more"
        return (
            f"the gate `{self.gate}` refused {len(self.gate_refused)} of {self.n} "
            f"document(s) WHOLE: {shown}. Trust these sources? build with "
            f"`--gate none`. Full list in the summary above and in {LOG_NAME}"
        )

    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}",
            "",
            *self.refusal_section(),
            "## 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}"
                if not self.carried_files
                else self.identity()
            ),
            "",
            "## 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)**: {self.identity()}. "
            "The run exits non-zero when these differ.",
            f"* **Converter**: {self.converter_path}, version {self.converter_version}.",
            f"* **Gate**: {self.gate} "
            f"({_GATE_DESCRIPTIONS.get(self.gate, 'unrecognised gate name')}). "
            "Every persisted byte of this bundle passed it.",
            self._assets_line(),
        ]
        if self.gate_refused:
            named = ", ".join(
                f"`{name}` (`{disposition}`)" for name, disposition in self.gate_refused
            )
            lines.append(
                f"* **Refused whole by the gate**: {len(self.gate_refused)} of {self.n} "
                f"document(s), so not one element of them is in this bundle: {named}. "
                "Rebuild with `--gate none` to carry sources you vouch for yourself."
            )
        if self.accounting is not None:
            lines.extend(self.accounting.log_lines())
        if self.unaccounted:
            lines.append("* **Unaccounted**: " + ", ".join(self.unaccounted) + " — K1b FAILED.")
        return "\n".join(lines) + "\n"

    def _assets_line(self) -> str:
        """What this run did about the images in its sources, stated either way.

        An absent bullet would make "this corpus had no figures" and "this run
        never looked" the same artifact, which is the trust trap `--gate none`
        already has its own sentence for. So the OFF case is written out in
        capitals rather than left to an inference.

        THE IMAGE BYTES ARE NOT SCREENED, and the log is where a consumer finds
        that out. The guard is text-only, so the pointer block passed the gate
        as body text and the picture beside it did not pass anything.
        """
        if not self.assets:
            return (
                "* **Images**: NOT CARRIED — this run did not look for images, so a "
                "document whose table is a picture reached the bundle as text with a "
                "gap in it. Absence of an image here is not evidence the sources had none."
            )
        # With an account, "found" is what the SOURCES declare, so a refused
        # document's pictures are not reported as never having existed.
        found = self.assets_found if self.accounting is None else self.accounting.images_found
        return (
            f"* **Images**: {self.assets_carried} carried of {found} found, "
            f"written to `{ASSETS_DIR}/` and pointed at from the concepts they stand in. "
            "The image BYTES were not screened: the gate above reads text, and a "
            "picture is not text."
        )


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 _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,
    concept_frontmatter_values: Mapping[str, str] | None = None,
    gate: str = GATE_NONE,
    assets: bool = False,
    account: bool = False,
) -> CorpusReport:
    """Run the corpus through the door and count what happened.

    `account` adds the element-level content accounting (`accounting.py`):
    an inventory of every source and one fate per element, on the report.

    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=resolve_gate(gate),
        profile=profile,
        root_frontmatter_values=root_frontmatter_values,
        segmentations=plans,
        pdf_headings=pdf_headings,
        heading_reserve=heading_reserve,
        ocr=ocr,
        assets=assets,
        concept_frontmatter_values=concept_frontmatter_values,
    )
    elapsed = time.monotonic() - started

    merged_names = tuple(item.source_file for item in result.persisted)
    blocked = result.quarantined + result.rejected
    carried = set(result.carried_files)
    failed = tuple(item for item in result.failed if item.source_file not in carried)
    coded_names = tuple(item.source_file for item in failed) + tuple(
        item.source_file for item in blocked
    )
    counts: dict[str, int] = {}
    for failure in 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 + tuple(sorted(carried)), coded=coded_names
        ),
        gate=gate,
        assets=assets,
        assets_carried=len(result.assets),
        assets_found=len(result.assets) + len(result.assets_rejected),
        carried_files=len(carried),
        accounting=account_run(corpus, walked, result) if account else None,
        gate_refused=tuple(sorted((item.source_file, item.disposition) for item in blocked)),
    )


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(
        "--gate",
        choices=GATE_NAMES,
        default=GATE_GUARD_TRUSTED_SOURCE,
        help=(
            "the persist gate every concept body passes before it is written. The "
            "same name and the same default as `okf build --gate`, deliberately: the "
            "two paths are held byte-equal by a test, and two different defaults "
            "would make that equality depend on which command you ran"
        ),
    )
    parser.add_argument(
        "--assets",
        action="store_true",
        default=DEFAULT_ASSETS,
        help=(
            "carry the images the sources declare into the bundle. The same "
            "name and the same default as `okf build --assets`, for the same "
            "reason the gate carries them: the two paths are held byte-equal "
            "by a test, and two different defaults would make that equality "
            "depend on which command you ran"
        ),
    )
    parser.add_argument(
        "--no-assets",
        action="store_false",
        dest="assets",
        help="the explicit opt-out, reproducing the pre-0.10.0 bytes",
    )
    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,
        assets=args.assets,
        ingested_at=args.ingested_at,
        plans=plans,
        profile=profile,
        root_frontmatter_values=root_values,
        gate=args.gate,
    )
    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.conservation_failed:
        print(
            f"{HARNESS_ID}: K1b FAILED - {report.identity()}. Unaccounted: "
            f"{', '.join(report.unaccounted) or '(none named)'}",
            file=sys.stderr,
        )
        return 1
    return 0


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