feat(tools): the corpus harness replays plans and writes the bundle's log

Two defects a consumer measured on 2026-09-03 against the bundle built by
this harness, both with one cause: `measure` passed `profile=STRUCTURED_V1`
and no plans at all, though Step 17 of the plan says the harness reuses
`process_inbox` "with the per-document plan mapping from Step 15".

- `adjudication` was in 0 of 39 concepts, because the key is written only
  inside the plan-covered branch and no plan was ever passed. `--plans-dir`
  replays proposals produced per document first; the profile follows from
  the flag rather than being something the harness may choose on its own.
  `--bundle-id` and `--okf-version` are arguments, never constants: a
  profile names a key and the caller owns its value (decision E1).
- `log.md` did not exist, so `merged` was countable from the bundle and `N`
  was not -- K1b could only be taken on trust from a report that does not
  travel with the artifact. Written in SPEC section 9 form and dated from
  `ingested_at`, never the wall clock.

Additive: without `--plans-dir` the run stays the flat `STRUCTURED_V1` run
that produced the published K1/K2 numbers.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

tests/test_corpus_run.py

@@ -138,3 +138,186 @@ def test_wall_time_per_file_is_reported(tmp_path: Path) -> None:
     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")

tools/okf_corpus_run.py

@@ -33,8 +33,10 @@ vendored binary is bypassed silently otherwise -- measured three times, wheel
 from __future__ import annotations
 
 import argparse
+import json
 import sys
 import time
+from collections.abc import Mapping
 from dataclasses import dataclass, replace
 from pathlib import Path
 
@@ -43,11 +45,20 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
 from llm_ingestion_okf.errors import IngestError  # noqa: E402
 from llm_ingestion_okf.extract import extract_text  # noqa: E402
 from llm_ingestion_okf.inbox import GateDecision, InboxResult, process_inbox  # noqa: E402
-from llm_ingestion_okf.profiles import STRUCTURED_V1  # noqa: E402
+from llm_ingestion_okf.profiles import (  # noqa: E402
+    SEGMENTED_OKF_V0_2,
+    STRUCTURED_V1,
+    BundleProfile,
+)
+from llm_ingestion_okf.segmentation import (  # noqa: E402
+    SegmentationPlan,
+    parse_segmentation_plan,
+)
 
 __all__ = [
     "CorpusReport",
     "converter_identity",
+    "load_plans",
     "is_degenerate",
     "main",
     "measure",
@@ -100,6 +111,7 @@ 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
@@ -165,6 +177,72 @@ class CorpusReport:
             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",
+            "title: Corpus run history",
+            "---",
+            "",
+            "# Corpus run history",
+            "",
+            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")
@@ -191,8 +269,20 @@ def _split_merges(corpus: Path, result: InboxResult) -> tuple[int, int]:
     return (substantive, degenerate)
 
 
-def measure(corpus: Path, bundle: Path, *, ingested_at: str) -> CorpusReport:
+def measure(
-    """Run the corpus through the door and count what happened."""
+    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,
+) -> 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.
+    """
     dropped = tuple(sorted(path.name for path in corpus.iterdir() if path.is_file()))
     started = time.monotonic()
     result = process_inbox(
@@ -201,7 +291,9 @@ def measure(corpus: Path, bundle: Path, *, ingested_at: str) -> CorpusReport:
         ingested_at,
         okf_type="reference",
         gate=_gate,
-        profile=STRUCTURED_V1,
+        profile=profile,
+        root_frontmatter_values=root_frontmatter_values,
+        segmentations=plans,
     )
     elapsed = time.monotonic() - started
 
@@ -220,6 +312,7 @@ def measure(corpus: Path, bundle: Path, *, ingested_at: str) -> CorpusReport:
     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.
@@ -247,6 +340,32 @@ def parse_args(argv: list[str] | None) -> argparse.Namespace:
     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 "
+            "tools/okf_propose_segments.py 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)
 
 
@@ -256,9 +375,54 @@ def main(argv: list[str] | None = None) -> int:
         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"
-    report = measure(args.corpus, bundle, ingested_at=args.ingested_at)
+
+    # 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.md").write_text(report.render_log(), encoding="utf-8", newline="")
     print(report.render())
     if report.unaccounted or report.merged + report.rejected != report.n:
         print(