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

"""`okf` — the installed command. One subcommand today: `build`.

## What it replaces

Until this module existed, "run the door over a folder" was a shell loop over
`tools/okf_propose_segments.py` followed by `tools/okf_corpus_run.py`, with
nine flags between them and a `--path-prefix` rule that lived in a code block
in a measurement report. Neither script was packaged, so the whole path was
reachable only from a clone -- a consumer who installed this library could call
`process_inbox` but could not run the thing the reports measured.

`okf build <inbox> --bundle <dir>` is that path, packaged. It is ORCHESTRATION
and nothing else: every rule it applies belongs to `propose` or `corpus`, which
are the same modules `tools/` now calls. There is one implementation of each,
and it is the packaged one.

## The two decisions this layer owns

**The scope of a document's proposed paths is its RELATIVE PATH, minus the
extension.** Section numbering is document-local, so two documents propose the
same path and Door B refuses both; the loop in the reports passed the BASENAME,
which was right while the inbox was flat. It stopped being right when the door
started walking recursively: `a/krav.pdf` and `b/krav.pdf` would both reduce to
`krav` and collide, which is the collision Door B is supposed to make
impossible rather than merely detectable. For a flat inbox the relative path IS
the basename, which is why the published bundles' bytes do not move.

**Omitted timestamps do not come from the clock.** `--ingested-at` and
`--proposed-at` default to `DEFAULT_STAMP`, one constant used for both. A
wall-clock default would put a changing byte into the artifact and break
rebuild-equals-incremental (K6) for every caller who did not pass the flags --
the property the segmented bundle is built on, and the one a convenience
default is most likely to take away silently. The epoch is deliberate and
readable as what it is: a stamp nobody set. A caller who wants a real ingest
time passes one.

## What it does not decide

Arm C (`--max-segment-chars`), Arm D (`--outline-run`) and Arm E
(`--table-grid`) are OFF here and are not exposed: they are measurement arms,
all off by default by operator decision, and a build command is not where an
unadjudicated segmentation heuristic should become one flag away. `tools/`
still reaches them.
"""

from __future__ import annotations

import argparse
import sys
import tempfile
from pathlib import Path

from .corpus import LOG_NAME, CorpusReport, link_log_in_root_index, load_plans, measure
from .errors import IngestError
from .inbox import walk_inbox
from .profiles import SEGMENTED_OKF_V0_2, STRUCTURED_V1, BundleProfile
from .propose import ProposerError
from .propose import run as propose_run

__all__ = ["DEFAULT_STAMP", "build", "main", "measure"]

CLI_ID = "okf build"

#: The timestamp written when the caller passes none, for the ingest stamp and
#: the proposal stamp alike. ONE constant: two independently-defaulted literals
#: drift, and the drift shows up only as two bundles differing in a field
#: nobody set.
DEFAULT_STAMP = "1970-01-01T00:00:00Z"


def _propose_plans(
    inbox: Path, bundle: Path, plans_dir: Path, *, proposed_at: str, okf_type: str
) -> tuple[int, int, int]:
    """Propose a plan per dropped file. Returns (written, nothing, failed).

    The walk is the DOOR's walk, imported rather than restated, so the set of
    documents that get a plan is exactly the set that gets ingested.

    Neither "nothing to propose" nor "cannot read" stops the loop, because
    neither stops the door: the first lands the document as one flat concept
    and the second is a coded rejection that K1b accounts for. The shell loop
    behaved the same way -- `exit 1` for the eleven with no boundary, `exit 2`
    for the four unreadable -- and a build that aborted on either would refuse
    corpora the two-script path completes.
    """
    walked, _ = walk_inbox(inbox, exclude=bundle)
    written = nothing = failed = 0
    for position, source in enumerate(walked, start=1):
        relative = source.relative_to(inbox)
        try:
            outcome = propose_run(
                source,
                plans_dir / f"{position:02d}.json",
                okf_type=okf_type,
                proposed_at=proposed_at,
                path_prefix=relative.with_suffix("").as_posix(),
            )
        except ProposerError as exc:
            print(f"{CLI_ID}: {relative.as_posix()}: {exc}", file=sys.stderr)
            failed += 1
            continue
        if outcome == 0:
            written += 1
        else:
            nothing += 1
    return (written, nothing, failed)


def build(
    inbox: Path,
    bundle: Path,
    *,
    ingested_at: str = DEFAULT_STAMP,
    proposed_at: str = DEFAULT_STAMP,
    bundle_id: str | None = None,
    okf_version: str | None = None,
    segments: bool = True,
    plans_dir: Path | None = None,
    okf_type: str = "reference",
) -> CorpusReport:
    """Folder in, bundle out. The whole command, minus argument parsing.

    Keyword-only with defaults, so a caller who takes this as an API keeps a
    source-compatible call when a flag is added.
    """
    if not segments:
        report = measure(inbox, bundle, ingested_at=ingested_at, profile=STRUCTURED_V1)
        _write_log(bundle, report, profile=STRUCTURED_V1)
        return report

    if bundle_id is None or okf_version is None:
        missing = ", ".join(
            flag
            for flag, value in (("--bundle-id", bundle_id), ("--okf-version", okf_version))
            if value is None
        )
        raise IngestError(
            f"{missing} is required unless --segments off; a profile names a key and the "
            "caller owns its value",
            code="manifest_invalid",
        )

    with tempfile.TemporaryDirectory(prefix="okf-plans-") as scratch:
        target = plans_dir if plans_dir is not None else Path(scratch)
        target.mkdir(parents=True, exist_ok=True)
        written, nothing, failed = _propose_plans(
            inbox, bundle, target, proposed_at=proposed_at, okf_type=okf_type
        )
        print(
            f"{CLI_ID}: proposed {written} plan(s); {nothing} document(s) with no boundary; "
            f"{failed} unreadable",
            file=sys.stderr,
        )
        plans = load_plans(target)
        report = measure(
            inbox,
            bundle,
            ingested_at=ingested_at,
            plans=plans,
            profile=SEGMENTED_OKF_V0_2,
            root_frontmatter_values={"okf_version": okf_version, "bundle_id": bundle_id},
        )
    _write_log(bundle, report, profile=SEGMENTED_OKF_V0_2)
    return report


def _write_log(bundle: Path, report: CorpusReport, *, profile: BundleProfile) -> None:
    """The section 9 log, into the BUNDLE, and the root index link to it.

    Lifted verbatim from the harness's own `main` rather than reimplemented:
    the log carries `N`, which is the one fact about a run the bundle cannot
    otherwise recover, and a build that wrote a bundle without it would ship an
    artifact whose conservation identity is uncheckable.
    """
    bundle.mkdir(parents=True, exist_ok=True)
    (bundle / LOG_NAME).write_text(report.render_log(), encoding="utf-8", newline="")
    link_log_in_root_index(bundle, profile)


def parse_args(argv: list[str] | None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        prog="okf",
        description="OKF bundle tooling. One folder in, one bundle out.",
    )
    subcommands = parser.add_subparsers(dest="command", required=True)
    build_parser = subcommands.add_parser(
        "build",
        help="build an OKF bundle from a folder of documents",
        description=(
            "Walk a folder recursively, propose a segmentation for each document, "
            "replay those proposals through Door B, and write the bundle. Reports "
            "the conservation identity `merged + coded rejections == N` on stdout "
            "and exits non-zero when it does not hold."
        ),
    )
    build_parser.add_argument("inbox", type=Path, help="the folder of documents to ingest")
    build_parser.add_argument(
        "--bundle", type=Path, required=True, help="where to write the OKF bundle"
    )
    build_parser.add_argument(
        "--bundle-id",
        default=None,
        help="required unless --segments off: what a consumer joins the concepts on",
    )
    build_parser.add_argument(
        "--okf-version",
        default=None,
        help=(
            "required unless --segments off: the upstream OKF version this bundle "
            "declares. An argument and never a constant -- the VALUE belongs to the "
            "catalog (decision E1)"
        ),
    )
    build_parser.add_argument(
        "--ingested-at",
        default=DEFAULT_STAMP,
        help=f"stamped verbatim. Default {DEFAULT_STAMP}: deterministic, never the clock",
    )
    build_parser.add_argument(
        "--proposed-at",
        default=DEFAULT_STAMP,
        help=f"written into every proposal. Default {DEFAULT_STAMP}, for the same reason",
    )
    build_parser.add_argument(
        "--segments",
        choices=("on", "off"),
        default="on",
        help=(
            "on (the default) proposes a segmentation per document with the "
            "mechanical rules and replays it; off ingests each document as one "
            "concept and asks for no root values"
        ),
    )
    build_parser.add_argument(
        "--okf-type", default="reference", help="okf_type for every concept and proposal"
    )
    build_parser.add_argument(
        "--plans-dir",
        type=Path,
        default=None,
        help=(
            "keep the proposals here instead of discarding them. Every entry is "
            "PROPOSED, never adjudicated -- this is where an operator reads what "
            "the run replayed"
        ),
    )
    build_parser.add_argument("--report", type=Path, default=None, help="also write the report")
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    if not args.inbox.is_dir():
        print(f"{CLI_ID}: FAILED - no such folder: {args.inbox}", file=sys.stderr)
        return 2
    try:
        report = build(
            args.inbox,
            args.bundle,
            ingested_at=args.ingested_at,
            proposed_at=args.proposed_at,
            bundle_id=args.bundle_id,
            okf_version=args.okf_version,
            segments=args.segments == "on",
            plans_dir=args.plans_dir,
            okf_type=args.okf_type,
        )
    except (IngestError, OSError, ValueError) as exc:
        print(f"{CLI_ID}: FAILED - {exc}", file=sys.stderr)
        return 2

    if args.report is not None:
        args.report.parent.mkdir(parents=True, exist_ok=True)
        args.report.write_text(report.render(), encoding="utf-8", newline="")
    print(report.render())
    if report.unaccounted or report.merged + report.rejected != report.n:
        print(
            f"{CLI_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())