llm-ingestion-okf/tests/test_cli_build.py

"""`okf build <inbox> --bundle <dir>`: one installed command, the same bytes.

The two-script path this replaces is real and documented (`docs/2026-09-03-k2-
bundle-rebuild.md`, `docs/2026-09-04-k3-arm-c.md`): a shell loop over
`tools/okf_propose_segments.py`, then `tools/okf_corpus_run.py`. Neither script
is packaged, so "run the door over a folder" was reachable only from a clone,
and only by retyping a nine-flag invocation whose `--path-prefix` rule lived in
a code block in a report.

Four properties, each the answer to a way that could go wrong:

- **The command exists in an INSTALLED copy.** Measured against a real install
  into a throwaway venv rather than against the clone, because the clone has
  `tools/` on disk and would pass whatever the wheel contains.
- **The bytes do not move.** The bundle `okf build` produces is compared
  byte-for-byte against the bundle the two scripts produce from the same
  inbox -- the two scripts run as subprocesses, not re-implemented here.
  "Similar" is not the requirement.
- **K1b is on stdout and in the exit status.** The conservation identity
  `merged + coded rejections == N` is the one number a caller must not have to
  take on trust, and a run where it breaks exits non-zero.
- **Omitting the timestamps is deterministic.** A wall-clock default would put
  a changing byte in the artifact and break rebuild-equals-incremental (K6,
  `tests/test_segmented_rebuild.py`) for anyone who did not pass the flags.
"""

from __future__ import annotations

import shutil
import subprocess
import sys
from dataclasses import replace
from pathlib import Path

import pytest

from llm_ingestion_okf import cli
from llm_ingestion_okf.corpus import CorpusReport

PROJECT_ROOT = Path(__file__).resolve().parents[1]
TOOLS = PROJECT_ROOT / "tools"

INGESTED_AT = "2026-09-03T00:00:00Z"
PROPOSED_AT = "2026-09-03T00:00:00Z"
BUNDLE_ID = "cli-build-fixture"
OKF_VERSION = "0.2"

# Headings the mechanical rules match, in a tree with subdirectories -- the
# door walks recursively now, so a flat fixture would leave the interesting
# half of the prefix rule unmeasured.
DOCUMENTS = {
    "alpha.md": (
        "# 1 Innledning\n\nDette dokumentet beskriver krav til seksjonering.\n\n"
        "# 1.1 Omfang\n\nOmfanget er hele anlegget og alle tilhoerende systemer.\n"
    ),
    "sub/beta.md": (
        "# 2 Brannkonsept\n\nBrannkonseptet stiller krav til roemningsveier.\n\n"
        "# 2.1 Roemning\n\nRoemningsveier skal vaere merket og fri for hindringer.\n"
    ),
    "sub/deep/gamma.md": (
        "# 3 Vedlikehold\n\nVedlikeholdet foelger en fast plan gjennom aaret.\n\n"
        "# 3.1 Intervaller\n\nIntervallene er angitt i tabellen under punkt tre.\n"
    ),
}


def inbox_with_subdirectories(root: Path) -> Path:
    inbox = root / "inbox"
    for name, body in DOCUMENTS.items():
        path = inbox / name
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(body, encoding="utf-8", newline="")
    return inbox


def tree(root: Path) -> dict[str, bytes]:
    """Every file under a bundle, keyed by relative path. The comparison unit."""
    return {
        path.relative_to(root).as_posix(): path.read_bytes()
        for path in sorted(root.rglob("*"))
        if path.is_file()
    }


def two_script_bundle(inbox: Path, out: Path) -> Path:
    """The path this command replaces, run as it is documented, as subprocesses.

    Re-implementing the loop here would compare the CLI against a copy of
    itself. Driving the actual scripts is what makes the byte comparison mean
    "the old path and the new path agree".
    """
    plans = out / "plans"
    plans.mkdir(parents=True, exist_ok=True)
    bundle = out / "bundle"
    sources = sorted(
        (path for path in inbox.rglob("*") if path.is_file()),
        key=lambda path: path.relative_to(inbox).as_posix(),
    )
    for position, source in enumerate(sources, start=1):
        relative = source.relative_to(inbox)
        subprocess.run(
            [
                sys.executable,
                str(TOOLS / "okf_propose_segments.py"),
                str(source),
                "--out",
                str(plans / f"{position:02d}.json"),
                "--path-prefix",
                relative.with_suffix("").as_posix(),
                "--proposed-at",
                PROPOSED_AT,
            ],
            capture_output=True,
            check=True,
        )
    subprocess.run(
        [
            sys.executable,
            str(TOOLS / "okf_corpus_run.py"),
            "--corpus",
            str(inbox),
            "--report",
            str(out / "report.md"),
            "--bundle",
            str(bundle),
            "--ingested-at",
            INGESTED_AT,
            "--plans-dir",
            str(plans),
            "--bundle-id",
            BUNDLE_ID,
            "--okf-version",
            OKF_VERSION,
        ],
        capture_output=True,
        check=True,
    )
    return bundle


def build(inbox: Path, bundle: Path, *extra: str) -> int:
    return cli.main(
        [
            "build",
            str(inbox),
            "--bundle",
            str(bundle),
            "--bundle-id",
            BUNDLE_ID,
            "--okf-version",
            OKF_VERSION,
            *extra,
        ]
    )


# --- the command exists, in an installed copy ------------------------------


def test_the_console_script_is_declared_and_resolves() -> None:
    """`[project.scripts]` is what makes `okf` a command rather than a file.

    Read off `pyproject.toml` rather than assumed from a working entry point in
    this venv: an editable install keeps working after the table is deleted.
    """
    tomllib = pytest.importorskip("tomllib")
    pyproject = tomllib.loads((PROJECT_ROOT / "pyproject.toml").read_text(encoding="utf-8"))
    assert pyproject["project"]["scripts"] == {"okf": "llm_ingestion_okf.cli:main"}
    assert callable(cli.main)


@pytest.mark.skipif(shutil.which("uv") is None, reason="uv is the install driver on this machine")
def test_build_help_exits_zero_from_an_installed_copy(tmp_path: Path) -> None:
    """The whole point of packaging it: it runs where `tools/` does not exist.

    `--no-deps` on purpose. The guard is the one runtime dependency and it is
    off-index, so pulling it here would make this test a network test; leaving
    it out also measures something worth knowing -- that the build path does
    not import the security boundary at module load.
    """
    uv = shutil.which("uv")
    assert uv is not None
    venv = tmp_path / "venv"
    subprocess.run([uv, "venv", str(venv), "-q"], check=True, capture_output=True)
    python = venv / "bin" / "python"
    subprocess.run(
        [uv, "pip", "install", "--no-deps", "--python", str(python), str(PROJECT_ROOT), "-q"],
        check=True,
        capture_output=True,
    )
    okf = venv / "bin" / "okf"
    assert okf.is_file(), "the console script was not installed"

    proc = subprocess.run([str(okf), "build", "--help"], capture_output=True, text=True)
    assert proc.returncode == 0, proc.stderr
    assert "--bundle" in proc.stdout

    # It must be the INSTALLED copy answering, not the clone reached through a
    # stray path entry -- otherwise this test passes on a machine where the
    # wheel ships nothing.
    where = subprocess.run(
        [str(python), "-c", "import llm_ingestion_okf.cli as m; print(m.__file__)"],
        capture_output=True,
        text=True,
        check=True,
    )
    assert str(PROJECT_ROOT / "src") not in where.stdout
    assert str(venv) in where.stdout


# --- the bytes do not move -------------------------------------------------


def test_one_command_gives_the_same_bundle_as_the_two_scripts(tmp_path: Path) -> None:
    """Byte-identical, over a tree with subdirectories. The requirement."""
    inbox = inbox_with_subdirectories(tmp_path)
    reference = two_script_bundle(inbox, tmp_path / "reference")

    bundle = tmp_path / "cli-bundle"
    assert build(inbox, bundle, "--ingested-at", INGESTED_AT, "--proposed-at", PROPOSED_AT) == 0

    assert tree(bundle) == tree(reference)
    assert len(tree(bundle)) > 1, "an empty bundle would compare equal to an empty bundle"


def test_the_bundle_carries_the_adjudication_layer_the_plans_produce(tmp_path: Path) -> None:
    """A positive control on the comparison above.

    Two flat bundles would also compare equal, and would prove that the
    segmentation lane never ran on either side.
    """
    inbox = inbox_with_subdirectories(tmp_path)
    bundle = tmp_path / "bundle"
    assert build(inbox, bundle, "--ingested-at", INGESTED_AT, "--proposed-at", PROPOSED_AT) == 0
    bodies = [body for name, body in tree(bundle).items() if name.endswith(".md")]
    assert any(b"adjudication:" in body for body in bodies)
    assert any(b"proposed" in body for body in bodies)


def test_arm_c_and_arm_d_are_off_unless_asked_for(tmp_path: Path) -> None:
    """Every measurement arm stays off by default -- measured on the artifact.

    `rule:size-split`, `rule:outline` and `rule:table-grid` are the markers the
    three arms write into a plan's `derived` list, so their absence is the arms
    being off. Named on the artifact rather than on the flag, because a flag
    `okf build` never passes is not evidence about what it produces.
    """
    inbox = inbox_with_subdirectories(tmp_path)
    plans = tmp_path / "plans"
    bundle = tmp_path / "bundle"
    assert build(inbox, bundle, "--plans-dir", str(plans), "--proposed-at", PROPOSED_AT) == 0
    written = sorted(plans.glob("*.json"))
    assert written, "the fixture must produce plans for this control to mean anything"
    for path in written:
        text = path.read_text(encoding="utf-8")
        assert "rule:size-split" not in text
        assert "rule:outline" not in text
        assert "rule:table-grid" not in text


def test_segments_off_builds_a_flat_bundle_without_a_bundle_id(tmp_path: Path) -> None:
    """`--segments off` is the unsegmented door, and asks for no root values."""
    inbox = inbox_with_subdirectories(tmp_path)
    bundle = tmp_path / "bundle"
    assert (
        cli.main(
            ["build", str(inbox), "--bundle", str(bundle), "--segments", "off"],
        )
        == 0
    )
    bodies = [body for name, body in tree(bundle).items() if name.endswith(".md")]
    assert bodies
    assert not any(b"adjudication:" in body for body in bodies)


def test_segmenting_without_a_bundle_id_is_refused_before_any_write(tmp_path: Path) -> None:
    """A profile names a key and the caller owns its value -- checked up front."""
    inbox = inbox_with_subdirectories(tmp_path)
    bundle = tmp_path / "bundle"
    assert cli.main(["build", str(inbox), "--bundle", str(bundle)]) == 2
    assert not bundle.exists()


# --- K1b on stdout, and in the exit status ---------------------------------


def test_the_conservation_identity_is_printed(
    tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
    inbox = inbox_with_subdirectories(tmp_path)
    bundle = tmp_path / "bundle"
    assert build(inbox, bundle, "--ingested-at", INGESTED_AT, "--proposed-at", PROPOSED_AT) == 0
    out = capsys.readouterr().out
    assert f"merged + coded rejections = {len(DOCUMENTS)}" in out
    assert f"N = {len(DOCUMENTS)}" in out


def test_a_broken_conservation_identity_exits_non_zero(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
    """The negative control. Without it every green run above proves nothing."""
    inbox = inbox_with_subdirectories(tmp_path)
    bundle = tmp_path / "bundle"
    real = cli.measure

    def losing_a_file(*args: object, **kwargs: object) -> CorpusReport:
        report = real(*args, **kwargs)  # type: ignore[arg-type]
        return replace(report, n=report.n + 1, unaccounted=("ghost.md",))

    monkeypatch.setattr(cli, "measure", losing_a_file)
    assert build(inbox, bundle, "--ingested-at", INGESTED_AT, "--proposed-at", PROPOSED_AT) == 1
    assert "K1b FAILED" in capsys.readouterr().err


# --- omitting the timestamps is deterministic ------------------------------


def test_omitted_timestamps_do_not_come_from_the_wall_clock(tmp_path: Path) -> None:
    """Two runs of the same inbox, no timestamp flags, identical bytes.

    A `datetime.now()` default would pass a single run and fail here -- which
    is the whole reason this test compares two builds instead of inspecting the
    default's value.
    """
    inbox = inbox_with_subdirectories(tmp_path)
    first = tmp_path / "first"
    second = tmp_path / "second"
    assert build(inbox, first) == 0
    assert build(inbox, second) == 0
    assert tree(first) == tree(second)


def test_a_rebuild_from_scratch_equals_the_incremental_bundle(tmp_path: Path) -> None:
    """K6 through the command, with the timestamps left to the default.

    `tests/test_segmented_rebuild.py` holds this property for the library call.
    The default stamp is the seam the command adds, so the property is measured
    again HERE rather than assumed to survive the new layer.
    """
    inbox = inbox_with_subdirectories(tmp_path)
    incremental = tmp_path / "incremental"
    assert build(inbox, incremental) == 0
    # A second document arrives, and the same bundle is updated in place.
    (inbox / "sub" / "delta.md").write_text(
        "# 4 Tilsyn\n\nTilsynet gjennomfoeres av en uavhengig part hvert aar.\n",
        encoding="utf-8",
        newline="",
    )
    assert build(inbox, incremental) == 0

    rebuilt = tmp_path / "rebuilt"
    assert build(inbox, rebuilt) == 0
    assert tree(incremental) == tree(rebuilt)


def test_the_default_stamp_is_named_once(tmp_path: Path) -> None:
    """One constant, not two: the ingest stamp and the proposal stamp agree.

    Two independently-defaulted literals would drift, and the drift would only
    show up as two bundles that differ in a field nobody passed.
    """
    assert cli.DEFAULT_STAMP == "1970-01-01T00:00:00Z"
    inbox = inbox_with_subdirectories(tmp_path)
    plans = tmp_path / "plans"
    bundle = tmp_path / "bundle"
    assert build(inbox, bundle, "--plans-dir", str(plans)) == 0
    for path in sorted(plans.glob("*.json")):
        assert cli.DEFAULT_STAMP in path.read_text(encoding="utf-8")
    concepts = [body for name, body in tree(bundle).items() if not name.endswith("index.md")]
    assert any(cli.DEFAULT_STAMP.encode() in body for body in concepts)
    # The log dates its entry from the same stamp, to the day.
    assert cli.DEFAULT_STAMP[:10] in (bundle / "log.md").read_text(encoding="utf-8")


def test_ingested_at_alone_stamps_every_concept_the_same(tmp_path: Path) -> None:
    """`--ingested-at` without `--proposed-at` must stamp every concept alike.

    Measured on K2 (S7 acid test, F1): passing only `--ingested-at` stamped 11
    of 629 concepts -- the unsegmented whole-document concepts, which read the
    call's `ingested_at` directly. The 618 segmented concepts read
    `segment.ingested_at`, which is the PLAN's `proposed_at`, independently
    defaulted to `DEFAULT_STAMP` when the caller never set it. A caller naming
    one clock is naming "when this ran", not asking for two different clocks.
    """
    inbox = inbox_with_subdirectories(tmp_path)
    bundle = tmp_path / "bundle"
    assert build(inbox, bundle, "--ingested-at", INGESTED_AT) == 0
    concepts = {
        name: body
        for name, body in tree(bundle).items()
        if name.endswith(".md") and not name.endswith("index.md") and name != "log.md"
    }
    assert concepts, "the fixture must produce concepts for this control to mean anything"
    for name, body in concepts.items():
        assert f"ingested_at: {INGESTED_AT}".encode() in body, name
        assert cli.DEFAULT_STAMP.encode() not in body, name


# --- the one implementation ------------------------------------------------


def test_the_scripts_are_thin_entries_to_the_packaged_implementation() -> None:
    """No duplicated logic: `tools/` calls the package, and is small enough to see.

    A line budget is a proxy, and a coarse one -- but the failure it catches is
    exactly the one that matters here: logic copied back into `tools/` so the
    two paths can drift apart while both stay green.
    """
    for name in ("okf_propose_segments.py", "okf_corpus_run.py"):
        source = (TOOLS / name).read_text(encoding="utf-8")
        assert "from llm_ingestion_okf." in source
        code = [
            line
            for line in source.splitlines()
            if line.strip() and not line.lstrip().startswith("#")
        ]
        assert len(code) < 40, f"tools/{name} carries logic again ({len(code)} lines)"


def test_the_packaged_modules_do_not_reach_back_into_the_clone() -> None:
    """`sys.path` surgery is what an unpackaged script needs and a package must not.

    Left in place it would half-work from an install: importable, and reading a
    `tools/` directory that is not there.
    """
    for name in ("propose.py", "corpus.py", "cli.py"):
        source = (PROJECT_ROOT / "src" / "llm_ingestion_okf" / name).read_text(encoding="utf-8")
        assert "sys.path" not in source


def test_the_root_index_does_not_link_the_run_log(tmp_path: Path) -> None:
    """Producer and consumer must count the same documents. Measured: they did not.

    `link_log_in_root_index` (`95eb271`) appended a markdown link to `log.md`
    from the root index so a reader entering there could reach the one file
    carrying `N`. The consumption contract SS 9.2 forbids a consumer from
    enumerating the bundle directory unless the named profile says the index is
    derived -- which for this profile it does not -- so the index tree IS the
    whole map a consumer is allowed to use. Anything the index links is a
    document, by that contract.

    The cost was measured on K2 by the first consumer to walk the bundle with a
    live model: their navigator followed the link and returned 630 documents
    where this repository's own pre-pass counts 629, and the corpus run's own
    log was reachable and citable as content. Our pre-pass excluding `log.md`
    (`5a0c879`, F2) fixed the count on OUR side only; the disagreement is
    produced HERE.

    The log still exists at the bundle root, which is where SS 9 puts it and is
    all F2 ever required. Upstream's own bundles show the link was never
    required either: measured at `9a15b13`, 0 of the 24 shipped `index.md`
    files name the single `log.md` in the set.
    """
    inbox = inbox_with_subdirectories(tmp_path)
    bundle = tmp_path / "bundle"
    assert build(inbox, bundle) == 0

    assert (bundle / "log.md").is_file(), "the log itself stays in the bundle"
    index = (bundle / "index.md").read_text(encoding="utf-8")
    assert "log.md" not in index

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

    concepts = {
        name
        for name in tree(bundle)
        if name.endswith(".md") and not name.endswith("index.md") and name != "log.md"
    }
    assert concepts, "the fixture must produce concepts for this control to mean anything"
    assert len(okf_consume.enumerate_concepts(bundle)) == len(concepts)


# --- the arms are reachable, and still off ---------------------------------
#
# Round 2, 2026-09-08. `_propose_plans` called the proposer with no arm flag at
# all, so `okf build` ran Arm B while `tools/okf_propose_segments.py` could run
# Arm D, E and F. Measured on a five-document folder: one tender PDF landed as
# ONE concept from the build path and as NINE from the proposer with
# `--outline-run 3`, and nine is what the operator's unit worksheet asks for.
# The default is unchanged -- that is the operator's decision, not this
# layer's -- but the flags now exist here.

# Two headings the mechanical rules match at two levels, plus a table opening
# directly under a heading: enough for Arm F's clause 2 and D1 to have
# something to do, so "the flag reached the proposer" is measured on the
# ARTIFACT and not on the parser.
ARM_DOCUMENTS = {
    "delta.md": (
        "## 4 Grunnforhold\n\nGrunnen er morene over berg i hele omraadet.\n\n"
        "### 4.1 Loesmasser\n\nLoesmassene er telefarlige og maa skiftes ut.\n"
    ),
    "epsilon.md": ("## 5 Prissammenstilling\n\n| Post | Sum |\n|------|-----|\n| 01 | 100 |\n"),
}


def inbox_for_arms(root: Path) -> Path:
    inbox = root / "arm-inbox"
    inbox.mkdir(parents=True, exist_ok=True)
    for name, body in ARM_DOCUMENTS.items():
        (inbox / name).write_text(body, encoding="utf-8", newline="")
    return inbox


def plan_titles(plans: Path) -> list[str]:
    import json

    titles: list[str] = []
    for path in sorted(plans.glob("*.json")):
        payload = json.loads(path.read_text(encoding="utf-8"))
        titles.extend(str(entry["title"]) for entry in payload["entries"])
    return titles


def test_the_arms_reach_the_proposer_from_the_build_command(tmp_path: Path) -> None:
    """The red test for round 2: the flags exist here and they change the plan.

    Asserted on the PLANS, which is where an arm's effect is visible, and on
    titles rather than a count -- a fold that kept the right number of concepts
    by discarding the wrong ones would pass a count assertion.

    `--unit-fold` folds `4.1 Loesmasser` into `4 Grunnforhold`;
    `--keep-table-heading` keeps `5 Prissammenstilling` as the heading it is
    instead of letting the table block carry the name.
    """
    inbox = inbox_for_arms(tmp_path)
    plans = tmp_path / "plans-armed"
    assert (
        build(
            inbox,
            tmp_path / "bundle-armed",
            "--plans-dir",
            str(plans),
            "--proposed-at",
            PROPOSED_AT,
            "--outline-run",
            "3",
            "--table-grid",
            "--unit-fold",
            "--keep-table-heading",
        )
        == 0
    )
    assert plan_titles(plans) == ["4 Grunnforhold", "5 Prissammenstilling"]


def test_the_build_default_is_unchanged_by_the_flags_existing(tmp_path: Path) -> None:
    """The control the test above rests on: same inbox, no flags, Arm B.

    Both halves over the same fixture. Without this, a change that turned an
    arm ON by default would leave the assertion above green and move every
    bundle every consumer builds.
    """
    inbox = inbox_for_arms(tmp_path)
    plans = tmp_path / "plans-plain"
    assert (
        build(
            inbox,
            tmp_path / "bundle-plain",
            "--plans-dir",
            str(plans),
            "--proposed-at",
            PROPOSED_AT,
        )
        == 0
    )
    assert plan_titles(plans) == ["4 Grunnforhold", "4.1 Loesmasser", "5 Prissammenstilling"]


def test_a_bundle_built_with_no_flags_is_byte_identical_to_the_shipped_one(
    tmp_path: Path,
) -> None:
    """The byte control, on the tree the other build tests use.

    The proposer gained a parameter and the CLI gained four. A default that
    moved by a byte would break rebuild-equals-incremental for every consumer
    who never passes a flag, and a title assertion cannot see that.
    """
    inbox = inbox_with_subdirectories(tmp_path)
    reference = two_script_bundle(inbox, tmp_path / "reference-2")
    bundle = tmp_path / "cli-bundle-2"
    assert build(inbox, bundle, "--ingested-at", INGESTED_AT, "--proposed-at", PROPOSED_AT) == 0
    assert tree(bundle) == tree(reference)